Source code for trader.replay.bootstrap

"""A date-clustered block bootstrap for the edge's confidence interval.

Rows on one decision date share a market factor — nine symbols all traded on
the same news day are not nine independent observations — so this resamples
*dates*, not rows. A per-row t-test would treat correlated rows as
independent and understate the interval.
"""

import random
from collections import defaultdict
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from datetime import date

from trader.replay.scoring import ScoredRow

__all__ = ["Interval", "date_clustered_interval"]




@dataclass(frozen=True, slots=True)
class Interval:
    """A percentile interval from a block bootstrap, with its own provenance."""

    low: float | None
    high: float | None
    point: float | None
    draws: int
    seed: int
    resamples_used: int





def date_clustered_interval(
    rows: Sequence[ScoredRow],
    statistic: Callable[[Sequence[ScoredRow]], float | None],
    *,
    draws: int = 10_000,
    seed: int = 7,
    confidence: float = 0.95,
) -> Interval:
    """Resample decision dates with replacement and interval `statistic`.

    Each draw picks as many dates (with replacement) as the row set has
    distinct dates, and takes every row on each picked date — so a draw with
    two dates of three rows each always has six rows, never four: the unit
    being resampled is the date, not the row.

    A draw whose `statistic` returns `None` (e.g. no acting rows in that
    resample) is skipped rather than counted as zero, which would drag the
    interval toward zero and understate the uncertainty. `resamples_used`
    reports how many draws actually contributed, so a degenerate sample is
    visible rather than silently averaged away.
    """
    by_date: dict[date, list[ScoredRow]] = defaultdict(list)
    for row in rows:
        by_date[row.decision_at.date()].append(row)
    dates = list(by_date.keys())
    point = statistic(rows)

    rng = random.Random(seed)  # noqa: S311 - a statistical bootstrap, not a security context
    samples: list[float] = []
    for _ in range(draws):
        if not dates:
            break
        resample_rows: list[ScoredRow] = []
        for _ in range(len(dates)):
            picked = dates[rng.randrange(len(dates))]
            resample_rows.extend(by_date[picked])
        value = statistic(resample_rows)
        if value is not None:
            samples.append(value)

    samples.sort()
    resamples_used = len(samples)
    if resamples_used == 0:
        return Interval(None, None, point, draws, seed, 0)

    alpha = (1 - confidence) / 2
    low_index = min(resamples_used - 1, max(0, int(alpha * resamples_used)))
    high_index = min(
        resamples_used - 1, max(low_index, int((1 - alpha) * resamples_used) - 1)
    )
    return Interval(
        low=samples[low_index],
        high=samples[high_index],
        point=point,
        draws=draws,
        seed=seed,
        resamples_used=resamples_used,
    )