"""The pre-registered row set: which (symbol, date) pairs get replayed.
Fixed in advance by issue #7 `note_3689814772`, and deliberately not
parameterised beyond what that note names. A sampler with knobs is a sampler
that can be turned until the result flatters the model, which is the
selection-on-outcome error this project already made once with
`oversold: 45`.
"""
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from typing import Literal, Protocol
__all__ = [
"DECISION_HOUR_UTC",
"ReplayRow",
"SampleShortfall",
"build_sample",
"decision_instant",
"eligible_dates",
"stride_sample",
]
#: 14:00 UTC. Inside regular trading hours (13:30-20:00 UTC) and when the app
#: actually traded — the live GOOG entry filled at 2026-08-14 14:00:09Z, and 5
#: of 9 live decision dates opened between 14:21 and 15:15 UTC. Fixing the
#: time of day removes an otherwise free parameter.
DECISION_HOUR_UTC = 14
Split = Literal["train", "holdout"]
[docs]
@dataclass(frozen=True, slots=True)
class ReplayRow:
"""One decision to replay."""
symbol: str
decision_at: datetime
split: Split
entry_day: date
[docs]
@dataclass(frozen=True, slots=True)
class SampleShortfall:
"""A symbol that could not fill its quota in a split."""
symbol: str
split: str
wanted: int
got: int
@property
def missing(self) -> int:
return self.wanted - self.got
class _ArchiveRepo(Protocol):
def load_items(
self, symbol: str, start: datetime, end: datetime
) -> Sequence[object]: ...
[docs]
def decision_instant(day: date) -> datetime:
"""`day` at the pre-registered decision hour, tz-aware UTC."""
return datetime(day.year, day.month, day.day, DECISION_HOUR_UTC, tzinfo=UTC)
[docs]
def eligible_dates(
archive: _ArchiveRepo,
symbol: str,
days: Sequence[date],
window_hours: float,
) -> list[date]:
"""The trading days on which `symbol` had at least one item in the window.
Deliberately the *same* gate as production's `max_news_age_hours`: with
no fresh news and no position the live strategy does not call the model
at all, so an ineligible date is one on which no live decision would have
existed. The replay set is therefore the set of decisions the strategy
would actually have made, not an arbitrary calendar.
"""
picked: list[date] = []
for day in days:
moment = decision_instant(day)
if archive.load_items(symbol, moment - timedelta(hours=window_hours), moment):
picked.append(day)
return picked
[docs]
def stride_sample(dates: Sequence[date], quota: int) -> list[date]:
"""`quota` dates, uniformly spaced through `dates`, both ends kept.
A stride, not a random draw and not "the most newsworthy dates": it is
reproducible, it cannot be reshuffled until it works, and it keeps the
panel balanced across the ~30x coverage difference between the dense and
thin names. Taking the first `quota` instead would bias every symbol to
the start of its window.
"""
if quota <= 0 or not dates:
return []
if quota >= len(dates):
return list(dates)
if quota == 1:
return [dates[0]]
step = (len(dates) - 1) / (quota - 1)
return [dates[round(i * step)] for i in range(quota)]
[docs]
def build_sample(
archive: _ArchiveRepo,
symbols: Sequence[str],
grid: Sequence[date],
*,
train: tuple[date, date],
holdout: tuple[date, date],
quotas: tuple[int, int] = (40, 20),
window_hours: float = 72.0,
) -> tuple[list[ReplayRow], list[SampleShortfall]]:
"""The pre-registered sample: 40 train + 20 holdout dates per symbol.
Rows come back grouped by symbol then chronological within each symbol's
split. Shortfalls are returned rather than raised — a thin symbol with 6
eligible holdout dates is a finding to report, not a failure — and are
**never** backfilled from the other split, which would leak holdout dates
into train.
"""
train_start, train_end = train
holdout_start, holdout_end = holdout
train_quota, holdout_quota = quotas
rows: list[ReplayRow] = []
shortfalls: list[SampleShortfall] = []
for symbol in symbols:
for split, (span_start, span_end), quota in (
("train", (train_start, train_end), train_quota),
("holdout", (holdout_start, holdout_end), holdout_quota),
):
days_in_span = [d for d in grid if span_start <= d <= span_end]
candidates = eligible_dates(archive, symbol, days_in_span, window_hours)
picked = stride_sample(candidates, quota)
if len(picked) < quota:
shortfalls.append(
SampleShortfall(
symbol=symbol, split=split, wanted=quota, got=len(picked)
)
)
for day in picked:
rows.append(
ReplayRow(
symbol=symbol,
decision_at=decision_instant(day),
split=split, # type: ignore[arg-type]
entry_day=day,
)
)
return rows, shortfalls