r"""Score replayed decisions against an always-BUY null.
Pure functions over a row list — no I/O, no model calls — so every scoring
rule is testable with no Ollama and no database. `score_all` is the one
function that reaches into `outcomes.py`, turning one replayed decision plus
its symbol's `ForwardPrices` into a `ScoredRow`; everything else here
operates on `ScoredRow`\s alone.
"""
from collections.abc import Sequence
from dataclasses import dataclass, replace
from datetime import date, datetime
from decimal import Decimal
from typing import Literal
from trader.replay.outcomes import (
HORIZONS,
ForwardPrices,
forward_return,
max_adverse_excursion,
)
from trader.strategies.base import Action
__all__ = [
"EdgeResult",
"ScoredRow",
"abstention_value",
"confidence_tertiles",
"counts",
"edge",
"edge_a_is_degenerate",
"hit_rate",
"market_adjusted",
"score_all",
"signed_return",
"tertile_means",
]
Split = Literal["train", "holdout"]
[docs]
@dataclass(frozen=True, slots=True)
class ScoredRow:
"""One replayed decision, with its forward returns and MAE at every horizon."""
symbol: str
decision_at: datetime
split: Split
entry_day: date
action: Action
confidence: float | None
returns: dict[int, Decimal | None]
mae: dict[int, Decimal | None]
dense: bool
[docs]
def score_all(
*,
symbol: str,
decision_at: datetime,
split: Split,
entry_day: date,
action: Action,
confidence: float | None,
prices: ForwardPrices | None,
dense: bool,
) -> ScoredRow:
"""Build a `ScoredRow` from a decision and the symbol's forward prices.
`prices=None` (forward outcomes not yet fetched, e.g. a `--dry-run` pass)
scores every horizon as `None` rather than raising: a dry run exercises
the prompt and the point-in-time guards without pricing a single row.
"""
if prices is None:
returns: dict[int, Decimal | None] = dict.fromkeys(HORIZONS)
mae: dict[int, Decimal | None] = dict.fromkeys(HORIZONS)
else:
returns = {n: forward_return(prices, entry_day, n) for n in HORIZONS}
mae = {n: max_adverse_excursion(prices, entry_day, n) for n in HORIZONS}
return ScoredRow(
symbol=symbol,
decision_at=decision_at,
split=split,
entry_day=entry_day,
action=action,
confidence=confidence,
returns=returns,
mae=mae,
dense=dense,
)
[docs]
def signed_return(action: Action, r: Decimal | None) -> Decimal | None:
"""`+r` for BUY, `-r` for SELL, `None` for HOLD (not an acting row) or a
missing return."""
if r is None:
return None
if action is Action.BUY:
return r
if action is Action.SELL:
return -r
return None
[docs]
@dataclass(frozen=True, slots=True)
class EdgeResult:
"""The model's edge over a null, at one horizon."""
model_return: float | None
null_return: float | None
edge: float | None
n_acting: int
n_rows: int
def _mean(values: Sequence[Decimal]) -> float | None:
if not values:
return None
return float(sum(values) / len(values))
[docs]
def edge(rows: Sequence[ScoredRow], n: int, *, include_holds_in_null: bool) -> EdgeResult:
"""`edge_A` (`include_holds_in_null=False`) or `edge_B` (`True`).
`edge_A`'s null is always-BUY over the *same acting rows* the model
traded — never over every row, which would let an unrelated HOLD's raw
return leak into the comparison and swing the edge on rows the model
never acted on. `edge_B`'s null spans every row with a defined horizon,
which is what makes it answer "does abstaining have value" instead.
"""
acting_signed: list[Decimal] = []
null_values: list[Decimal] = []
for row in rows:
r = row.returns.get(n)
if r is None:
continue
signed = signed_return(row.action, r)
acting = signed is not None
if acting:
acting_signed.append(signed)
if include_holds_in_null or acting:
null_values.append(r)
model_return = _mean(acting_signed)
null_return = _mean(null_values)
edge_value = (
None
if model_return is None or null_return is None
else model_return - null_return
)
return EdgeResult(
model_return=model_return,
null_return=null_return,
edge=edge_value,
n_acting=len(acting_signed),
n_rows=len(null_values),
)
[docs]
def edge_a_is_degenerate(rows: Sequence[ScoredRow], n: int) -> bool:
"""True when edge_A's null cannot possibly differ from its model side.
edge_A's null is always-BUY over the *same acting rows* the model
traded (see `edge`'s docstring): for each acting row it takes the raw
return `r`, exactly as `signed_return` does for a BUY. If there is no
SELL among the acting rows at horizon `n`, `signed_return` returns `+r`
for every one of them, so the model-side list and the null-side list are
the identical sequence of values — `edge_A == 0.0` by construction at
every draw of a bootstrap over this row set, never by measurement
(issue #29). Built the same way `edge` builds its own two lists, so a
change to one without the other cannot silently drift them apart.
The vacuous case (zero acting rows at this horizon) also returns `True`:
both lists are empty and therefore identical, and `edge` already reports
`None` rather than a misleading `0.0` in that case — flagging it
degenerate too costs nothing and is the conservative side to be wrong on.
"""
signed: list[Decimal] = []
raw: list[Decimal] = []
for row in rows:
r = row.returns.get(n)
if r is None:
continue
s = signed_return(row.action, r)
if s is None:
continue
signed.append(s)
raw.append(r)
return signed == raw
[docs]
def market_adjusted(
rows: Sequence[ScoredRow], benchmark: dict[tuple[date, int], Decimal]
) -> list[ScoredRow]:
"""Subtract SPY's same-date, same-horizon return, dropping SPY's own rows.
`benchmark` maps `(entry_day, horizon) -> SPY's forward_return`, built by
the caller from SPY's own `ForwardPrices` over the sample's entry days.
"""
adjusted: list[ScoredRow] = []
for row in rows:
if row.symbol == "SPY":
continue
new_returns: dict[int, Decimal | None] = {}
for n, r in row.returns.items():
bench = benchmark.get((row.entry_day, n))
new_returns[n] = None if (r is None or bench is None) else r - bench
adjusted.append(replace(row, returns=new_returns))
return adjusted
[docs]
def hit_rate(rows: Sequence[ScoredRow], n: int) -> tuple[float, float]:
"""(acting hit rate, base rate) — fraction of rows with `signed_return > 0`.
The acting rate is over BUY/SELL rows scored by their signed return; the
base rate is `r > 0` across every row with a defined horizon, acting or
not, so the two are comparable as "the model" versus "the market".
"""
acting_hits = acting_total = 0
base_hits = base_total = 0
for row in rows:
r = row.returns.get(n)
if r is None:
continue
base_total += 1
if r > 0:
base_hits += 1
signed = signed_return(row.action, r)
if signed is not None:
acting_total += 1
if signed > 0:
acting_hits += 1
acting_rate = acting_hits / acting_total if acting_total else 0.0
base_rate = base_hits / base_total if base_total else 0.0
return acting_rate, base_rate
[docs]
def abstention_value(rows: Sequence[ScoredRow], n: int) -> Decimal | None:
"""`mean(r | HOLD) - mean(r | BUY)`. Negative means the HOLDs carry information.
`None` when either side has no rows with a defined horizon — there is no
abstention value to report from an empty comparison.
"""
holds = [
row.returns[n]
for row in rows
if row.action is Action.HOLD and row.returns.get(n) is not None
]
buys = [
row.returns[n]
for row in rows
if row.action is Action.BUY and row.returns.get(n) is not None
]
if not holds or not buys:
return None
return (sum(holds) / len(holds)) - (sum(buys) / len(buys)) # type: ignore[operator]
[docs]
def confidence_tertiles(train_rows: Sequence[ScoredRow]) -> tuple[float, float]:
"""Two cut points splitting `train_rows`' confidences into thirds.
Computed on train and applied to holdout unchanged by the caller —
recomputing on holdout is how a calibration result becomes circular.
"""
values = sorted(row.confidence for row in train_rows if row.confidence is not None)
if not values:
raise ValueError("no confidence values to compute tertiles from")
def _percentile(p: float) -> float:
index = min(len(values) - 1, max(0, round(p * (len(values) - 1))))
return values[index]
return _percentile(1 / 3), _percentile(2 / 3)
[docs]
def tertile_means(
rows: Sequence[ScoredRow], cuts: tuple[float, float], n: int
) -> dict[str, Decimal | None]:
"""Mean `signed_return` at horizon `n`, bucketed by `cuts` (from `confidence_tertiles`)."""
low_cut, high_cut = cuts
buckets: dict[str, list[Decimal]] = {"low": [], "mid": [], "high": []}
for row in rows:
if row.confidence is None:
continue
signed = signed_return(row.action, row.returns.get(n))
if signed is None:
continue
bucket = (
"low"
if row.confidence <= low_cut
else "mid"
if row.confidence <= high_cut
else "high"
)
buckets[bucket].append(signed)
return {k: (sum(v) / len(v) if v else None) for k, v in buckets.items()}
[docs]
def counts(rows: Sequence[ScoredRow]) -> dict[str, dict[str, int]]:
"""Row counts by action, symbol, year, density (dense/thin), and split."""
tally: dict[str, dict[str, int]] = {
"action": {},
"symbol": {},
"year": {},
"density": {},
"split": {},
}
for row in rows:
tally["action"][row.action.value] = tally["action"].get(row.action.value, 0) + 1
tally["symbol"][row.symbol] = tally["symbol"].get(row.symbol, 0) + 1
year = str(row.decision_at.year)
tally["year"][year] = tally["year"].get(year, 0) + 1
density = "dense" if row.dense else "thin"
tally["density"][density] = tally["density"].get(density, 0) + 1
tally["split"][row.split] = tally["split"].get(row.split, 0) + 1
return tally