"""Drives the pre-registered sample through the real `LlmStrategy`, and scores it.
Instantiates `LlmStrategy` directly — never through `registry.build_strategy`,
which exposes no `clock` — with a fresh `ArchiveNewsAt` and a fresh instance
per row: both are bound to that row's decision instant, and reusing one
`LlmStrategy` across rows would leave `last_inputs` from the previous row
readable if a call raised. `decision_memory=None` **and**
`max_reuse_age_minutes=0` together switch the reuse gate off completely —
either alone would too, but a carried-forward HOLD from a neighbouring
replayed row must never be possible, so both are set so it cannot be
half-configured.
Reads `data/trader.db` and Alpaca's news history through the production
repositories, for reads only. Writes no `decisions`, `bars`, or
`news_archive` row, and never constructs a broker — this module cannot place
an order.
"""
from __future__ import annotations
import csv
import json
import logging
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Protocol
from trader.domain import Bar
from trader.llm.base import LlmProvider
from trader.news.base import NewsItem
from trader.replay.bars_at import bars_before
from trader.replay.bootstrap import date_clustered_interval
from trader.replay.news_at import ArchiveNewsAt
from trader.replay.outcomes import (
HORIZONS,
ForwardPrices,
fetch_forward_prices,
forward_return,
)
from trader.replay.sample import ReplayRow, SampleShortfall
from trader.replay.scoring import (
EdgeResult,
ScoredRow,
abstention_value,
confidence_tertiles,
counts,
edge,
edge_a_is_degenerate,
hit_rate,
market_adjusted,
score_all,
tertile_means,
)
from trader.strategies.base import Action
from trader.strategies.llm_strategy import LlmStrategy
__all__ = ["ReplayRunner", "RunReport", "rescore_from_csv"]
_logger = logging.getLogger("trader.replay")
#: The symbols the note calls "dense". Everything else in the pre-registered
#: 9-symbol sample (ANNX, ITRG, UUUU) is "thin".
DENSE_SYMBOLS = frozenset({"AAPL", "GOOG", "SPY", "PLTR", "NET", "TOST"})
#: How far past a row's latest entry day to fetch forward prices. N=21
#: trading days is roughly a calendar month; 45 calendar days comfortably
#: covers it including weekends and holidays.
_FORWARD_FETCH_PAD_DAYS = 45
_CSV_FIELDNAMES = [
"symbol",
"decision_at",
"split",
"entry_day",
"action",
"confidence",
"model_called",
*[f"return_{n}" for n in HORIZONS],
*[f"mae_{n}" for n in HORIZONS],
]
class _NewsRepo(Protocol):
def load(self, symbol: str, start: datetime, end: datetime) -> list[NewsItem]: ...
class _BarRepo(Protocol):
def load_bars(
self, symbol: str, interval: str, start: datetime, end: datetime
) -> list[Bar]: ...
class _MarketDataProvider(Protocol):
def get_history(
self, symbol: str, start: date, end: date, interval: str = "1d"
) -> list[Bar]: ...
class _NullLlmProvider:
"""A stand-in that never reaches a real model — `--dry-run` only.
Returns a fixed, schema-valid HOLD payload so `LlmStrategy.evaluate()`
runs its **entire** pipeline unmodified — including `build_messages`,
which is where a leaked future headline or a leaked future bar would
surface — without a network call or an Ollama server. That is what lets
`--dry-run` exercise both point-in-time guards over the whole sample for
free before a single real model call is spent.
"""
model = "dry-run"
def generate_json(
self, system: str, user: str, schema: dict[str, object]
) -> dict[str, object]:
return {"action": "hold", "confidence": 0.0, "reason": "dry-run: no model called"}
def health(self) -> str:
return self.model
[docs]
@dataclass(frozen=True, slots=True)
class RunReport:
"""What a run produced, and the provenance needed to reproduce it."""
seed: int
generated_at: str
fetch_dates: dict[str, str]
n_rows: int
n_scored: int
csv_path: str
json_path: str
def _fixed_clock(instant: datetime) -> Callable[[], datetime]:
return lambda: instant
[docs]
class ReplayRunner:
"""Replays a pre-registered `ReplayRow` sample through the real `LlmStrategy`."""
def __init__(
self,
*,
news_repository: _NewsRepo,
bar_repository: _BarRepo,
market_data: _MarketDataProvider,
llm_provider_factory: Callable[[], LlmProvider],
window_hours: float = 72.0,
lookback_bars: int = 30,
seed: int = 7,
) -> None:
self._news_repository = news_repository
self._bar_repository = bar_repository
self._market_data = market_data
self._llm_provider_factory = llm_provider_factory
self._window_hours = window_hours
self._lookback_bars = lookback_bars
self._seed = seed
[docs]
def run(
self,
rows: Sequence[ReplayRow],
out_dir: Path,
*,
resume: bool = False,
dry_run: bool = False,
shortfalls: Sequence[SampleShortfall] = (),
) -> RunReport:
"""Replay every row not already present, append as it goes, then score.
`--resume` reads the CSV back and skips any `(symbol, decision_at)`
already present — a run that loses everything to a crash at row 500
of 540 is a design defect, not bad luck. Every row is flushed to disk
the moment it completes.
"""
out_dir.mkdir(parents=True, exist_ok=True)
csv_path = out_dir / "replay_rows.csv"
json_path = out_dir / "replay_summary.json"
append = resume and csv_path.exists()
already_done = _read_done_keys(csv_path) if append else set()
scored_rows: list[ScoredRow] = list(_read_scored_rows(csv_path)) if append else []
forward_prices: dict[str, ForwardPrices] = {}
fetch_dates: dict[str, str] = {}
if not dry_run:
for symbol in sorted({row.symbol for row in rows}):
symbol_days = [row.entry_day for row in rows if row.symbol == symbol]
start = min(symbol_days)
end = max(symbol_days) + timedelta(days=_FORWARD_FETCH_PAD_DAYS)
prices = fetch_forward_prices(self._market_data, symbol, start, end)
forward_prices[symbol] = prices
fetch_dates[symbol] = prices.fetched_at.isoformat()
with csv_path.open("a" if append else "w", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=_CSV_FIELDNAMES)
if not append:
writer.writeheader()
for row in rows:
key = (row.symbol, row.decision_at.isoformat())
if key in already_done:
continue
scored, model_called = self._run_one(
row, forward_prices.get(row.symbol), dry_run=dry_run
)
writer.writerow(_to_csv_record(scored, model_called=model_called))
handle.flush()
scored_rows.append(scored)
summary = _summarize(
scored_rows,
forward_prices.get("SPY"),
shortfalls=shortfalls,
seed=self._seed,
fetch_dates=fetch_dates,
dry_run=dry_run,
)
json_path.write_text(json.dumps(summary, indent=2, default=str))
return RunReport(
seed=self._seed,
generated_at=datetime.now(UTC).isoformat(),
fetch_dates=fetch_dates,
n_rows=len(rows),
n_scored=len(scored_rows),
csv_path=str(csv_path),
json_path=str(json_path),
)
def _run_one(
self, row: ReplayRow, prices: ForwardPrices | None, *, dry_run: bool
) -> tuple[ScoredRow, bool]:
news_at = ArchiveNewsAt(
self._news_repository, row.decision_at, self._window_hours
)
llm: LlmProvider = _NullLlmProvider() if dry_run else self._llm_provider_factory()
strategy = LlmStrategy(
id="replay",
llm=llm,
news_provider=news_at,
lookback_bars=self._lookback_bars,
max_news_age_hours=self._window_hours,
decision_memory=None,
max_reuse_age_minutes=0,
clock=_fixed_clock(row.decision_at),
)
# Built once, outside `evaluate`, so the bar guard fires here on every
# row regardless of whether the strategy goes on to skip the model.
bars = bars_before(
self._bar_repository, row.symbol, row.decision_at, self._lookback_bars
)
signal = strategy.evaluate(bars, None)
last_inputs = dict(
strategy.last_inputs
) # copied immediately; see module docstring
confidence_value = signal.indicators.get("confidence")
if not isinstance(confidence_value, int | float) or isinstance(
confidence_value, bool
):
# Absent, never zero: `confidence` lives in `indicators["confidence"]`
# and is missing whenever the model was not called at all.
confidence_value = None
model_called = (not dry_run) and "skipped_model_call" not in last_inputs
scored = score_all(
symbol=row.symbol,
decision_at=row.decision_at,
split=row.split,
entry_day=row.entry_day,
action=signal.action,
confidence=confidence_value,
prices=None if dry_run else prices,
dense=row.symbol in DENSE_SYMBOLS,
)
return scored, model_called
[docs]
def rescore_from_csv(
csv_path: Path,
out_dir: Path,
*,
seed: int = 7,
fetch_dates: dict[str, str] | None = None,
shortfalls: Sequence[SampleShortfall] = (),
dry_run: bool = False,
spy_prices: ForwardPrices | None = None,
) -> RunReport:
"""Re-run `_summarize` over an already-replayed CSV — no model, no fetch.
Every replayed decision (action, confidence, and every horizon's forward
return) is already on disk in `csv_path`; a change to the *scoring*
formula (issue #29's edge_A -> edge_B headline swap is the first case)
should not cost another 1.7-1.9 hours of Ollama calls to re-judge. This
is the capability `run()`'s own `--resume` path already leans on
(`_read_scored_rows` reads exactly this CSV shape back) turned into a
standalone entry point, because a metric amendment is not a one-off.
`spy_prices` is optional and, when omitted, `edge_grid_market_adjusted`
is **not recomputed** — SPY's forward-price grid needed to market-adjust
an arbitrary other symbol's entry day cannot be reconstructed from the
CSV alone (each symbol's sampled entry days are drawn independently, so
the CSV's own SPY rows do not cover every day another symbol needs), and
fetching it fresh means a market-data call this function deliberately
does not make on your behalf. Every other section — the headline, the
non-market-adjusted `edge_grid`, hit rate, abstention, calibration,
counts — comes entirely from the CSV and is fully reprocessed. Pass a
freshly fetched `ForwardPrices` for `"SPY"` to get the market-adjusted
grid recomputed too.
"""
rows = _read_scored_rows(csv_path)
out_dir.mkdir(parents=True, exist_ok=True)
json_path = out_dir / "replay_summary.json"
summary = _summarize(
rows,
spy_prices,
shortfalls=shortfalls,
seed=seed,
fetch_dates=fetch_dates or {},
dry_run=dry_run,
)
json_path.write_text(json.dumps(summary, indent=2, default=str))
return RunReport(
seed=seed,
generated_at=datetime.now(UTC).isoformat(),
fetch_dates=fetch_dates or {},
n_rows=len(rows),
n_scored=len(rows),
csv_path=str(csv_path),
json_path=str(json_path),
)
def _to_csv_record(scored: ScoredRow, *, model_called: bool) -> dict[str, object]:
record: dict[str, object] = {
"symbol": scored.symbol,
"decision_at": scored.decision_at.isoformat(),
"split": scored.split,
"entry_day": scored.entry_day.isoformat(),
"action": scored.action.value,
"confidence": "" if scored.confidence is None else scored.confidence,
"model_called": model_called,
}
for n in HORIZONS:
r = scored.returns.get(n)
m = scored.mae.get(n)
record[f"return_{n}"] = "" if r is None else str(r)
record[f"mae_{n}"] = "" if m is None else str(m)
return record
def _read_done_keys(csv_path: Path) -> set[tuple[str, str]]:
with csv_path.open(newline="") as handle:
return {(r["symbol"], r["decision_at"]) for r in csv.DictReader(handle)}
def _read_scored_rows(csv_path: Path) -> list[ScoredRow]:
rows: list[ScoredRow] = []
with csv_path.open(newline="") as handle:
for record in csv.DictReader(handle):
rows.append(_from_csv_record(record))
return rows
def _from_csv_record(record: dict[str, str]) -> ScoredRow:
def _decimal(text: str) -> Decimal | None:
if not text:
return None
try:
return Decimal(text)
except InvalidOperation:
return None
confidence = float(record["confidence"]) if record["confidence"] else None
return ScoredRow(
symbol=record["symbol"],
decision_at=datetime.fromisoformat(record["decision_at"]),
split=record["split"], # type: ignore[arg-type]
entry_day=date.fromisoformat(record["entry_day"]),
action=Action(record["action"]),
confidence=confidence,
returns={n: _decimal(record[f"return_{n}"]) for n in HORIZONS},
mae={n: _decimal(record[f"mae_{n}"]) for n in HORIZONS},
dense=record["symbol"] in DENSE_SYMBOLS,
)
def _benchmark_map(spy_prices: ForwardPrices | None) -> dict[tuple[date, int], Decimal]:
if spy_prices is None:
return {}
benchmark: dict[tuple[date, int], Decimal] = {}
for entry_day in spy_prices.grid:
for n in HORIZONS:
r = forward_return(spy_prices, entry_day, n)
if r is not None:
benchmark[(entry_day, n)] = r
return benchmark
def _summarize(
rows: Sequence[ScoredRow],
spy_prices: ForwardPrices | None,
*,
shortfalls: Sequence[SampleShortfall],
seed: int,
fetch_dates: dict[str, str],
dry_run: bool,
) -> dict[str, object]:
"""The full pre-registered metric suite, computed once over the row set.
Train and holdout are scored separately throughout; only the holdout
number is the one the pre-committed null sentence is judged against.
"""
train_rows = [r for r in rows if r.split == "train"]
holdout_rows = [r for r in rows if r.split == "holdout"]
benchmark = _benchmark_map(spy_prices)
train_adjusted = market_adjusted(train_rows, benchmark)
holdout_adjusted = market_adjusted(holdout_rows, benchmark)
def _edge_grid(row_set: Sequence[ScoredRow]) -> dict[str, dict[str, object]]:
return {
str(n): {
"edge_a": _edge_dict(edge(row_set, n, include_holds_in_null=False)),
"edge_a_degenerate": edge_a_is_degenerate(row_set, n),
"edge_b": _edge_dict(edge(row_set, n, include_holds_in_null=True)),
}
for n in HORIZONS
}
# issue #29: edge_A's null is always-BUY over the *same acting rows* the
# model traded, so with zero SELL actions among them (the case for every
# row in this sample) `edge_A == 0.0` by construction, not measurement —
# its holdout 95% CI is [0.0, 0.0] regardless of what the model actually
# did. edge_B (null spans every row, acting or not) is not degenerate and
# is promoted to headline; edge_A is kept alongside as a diagnostic only.
headline_degenerate = edge_a_is_degenerate(holdout_rows, 5)
headline = edge(holdout_rows, 5, include_holds_in_null=True)
headline_interval = date_clustered_interval(
holdout_rows,
lambda sample: edge(sample, 5, include_holds_in_null=True).edge,
seed=seed,
)
edge_a_diagnostic = edge(holdout_rows, 5, include_holds_in_null=False)
calibration: dict[str, object] = {}
try:
cuts = confidence_tertiles(train_rows)
calibration = {
"cuts": list(cuts),
"holdout_means_at_n5": {
k: (str(v) if v is not None else None)
for k, v in tertile_means(holdout_rows, cuts, 5).items()
},
}
except ValueError:
calibration = {
"error": "no confidence values in train rows to compute tertiles from"
}
hit_holdout = hit_rate(holdout_rows, 5)
abstain_holdout = abstention_value(holdout_rows, 5)
return {
"seed": seed,
"generated_at": datetime.now(UTC).isoformat(),
"dry_run": dry_run,
"fetch_dates": fetch_dates,
"n_rows": len(rows),
"n_train": len(train_rows),
"n_holdout": len(holdout_rows),
"shortfalls": [
{"symbol": s.symbol, "split": s.split, "wanted": s.wanted, "got": s.got}
for s in shortfalls
],
# issue #29 (2026-08-19): headline is edge_B, not edge_A — see the
# comment above `headline_degenerate`. `headline_metric` names which
# formula produced `headline_edge_n5_holdout` so the two can never be
# read as interchangeable.
"headline_metric": "edge_b",
"headline_edge_n5_holdout": _edge_dict(headline),
"headline_interval_n5_holdout": {
"low": headline_interval.low,
"high": headline_interval.high,
"point": headline_interval.point,
"draws": headline_interval.draws,
"seed": headline_interval.seed,
"resamples_used": headline_interval.resamples_used,
},
"edge_a_entry_only_diagnostic_n5_holdout": {
**_edge_dict(edge_a_diagnostic),
"degenerate": headline_degenerate,
},
"edge_grid": {
"train": _edge_grid(train_rows),
"holdout": _edge_grid(holdout_rows),
},
"edge_grid_market_adjusted": {
"train": _edge_grid(train_adjusted),
"holdout": _edge_grid(holdout_adjusted),
},
"hit_rate_n5_holdout": {"acting": hit_holdout[0], "base": hit_holdout[1]},
"abstention_value_n5_holdout": (
str(abstain_holdout) if abstain_holdout is not None else None
),
"confidence_calibration": calibration,
"counts": counts(rows),
"null_sentence": (
"AMENDED 2026-08-19 (issue #29), replacing edge_A as headline: "
"edge_A's null is always-BUY over the same acting rows the model "
"traded, and this sample has zero SELL actions across all 540 "
"replayed rows, so edge_A == 0.0 by construction, not by "
"measurement, and its holdout 95% CI is [0.0, 0.0] regardless of "
"what the model did. edge_B (null spans every row, acting or not) "
"is not degenerate and is promoted to headline; its formula did "
"not change, it was already computed for every horizon/split "
"before this amendment. edge_A's point estimate is kept as "
'"edge_a_entry_only_diagnostic_n5_holdout" but no longer decides '
"the null sentence. If the holdout 95% interval on edge_B at N=5 "
'includes zero, the conclusion is "no measured edge over '
'blind-buy-everything on 540 decisions". The prompt is not '
"changed on this evidence, #6's relevance filter is not adopted "
"on this evidence, and the run is reported as-is."
),
}
def _edge_dict(result: EdgeResult) -> dict[str, object]:
return {
"model_return": result.model_return,
"null_return": result.null_return,
"edge": result.edge,
"n_acting": result.n_acting,
"n_rows": result.n_rows,
}