diff --git a/backend/app/__init__.py b/backend/app/__init__.py index a690118..ce2d006 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -740,6 +740,37 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: ) return jsonify(detail) + @app.get("/api/v1/backtest/readiness") + def backtest_readiness_endpoint(): + """Strict PIT backtest readiness + recommended default dates. + + The backtest must not start before advice is genuinely available. This + endpoint reports whether the PIT stores (factor vintages + Siamchart + vintage manifest) plus price data cover any usable [start, end] window, + and returns the recommended default start/end the UI should prefill. + It fails closed (ready=false + missing list) when coverage is absent. + """ + from pathlib import Path as _Path + from .backtest_readiness import evaluate_readiness + from .factor_vintages import FactorVintageStore + from .siamchart_vintages import SiamchartVintageStore + from .simulation import load_price_snapshot, SimulationError + + data_root = _Path(__file__).resolve().parents[1] / "data" + fstore = FactorVintageStore(data_root) + sstore = SiamchartVintageStore(data_root) + try: + series = load_price_snapshot() + except SimulationError: + series = None + start = request.args.get("start") + end = request.args.get("end") + res = evaluate_readiness( + factor_store=fstore, siamchart_store=sstore, + price_series=series, start=start, end=end, + ) + return jsonify(res.to_dict()) + @app.post("/api/v1/backtest") def run_backtest_endpoint(): """Run a real backtest over [start, end] with capital; persist result.""" diff --git a/backend/app/backtest_readiness.py b/backend/app/backtest_readiness.py new file mode 100644 index 0000000..9cb155e --- /dev/null +++ b/backend/app/backtest_readiness.py @@ -0,0 +1,423 @@ +"""Strict PIT backtest readiness + default-date derivation (Task 1). + +The backtest must not silently start before advice is genuinely available. +This module answers two things: + + * ``ready`` — whether every input the backtest needs is PIT-available over + some [start, end] window (factor releases, a Siamchart snapshot, and + executable prices for the universe). + * ``recommended_start`` / ``recommended_end`` — the defaults the API/UI + should prefill. The recommended end is "yesterday" in Bangkok (the last + complete trading availability), and the recommended start is the earliest + date at which every PIT input is release-available *and* at least one + executable price exists. + +Fail-closed semantics: this module never manufactures a start date from data +that does not exist. If any required input has no coverage through the window +it reports ``ready=false`` and lists what is missing. +""" + +from __future__ import annotations + +import datetime as dt +from dataclasses import dataclass, field +from typing import Any, Optional + +_TZ_BANGKOK = dt.timezone(dt.timedelta(hours=7)) + + +class BacktestReadinessError(Exception): + """Raised when coverage cannot be evaluated safely.""" + + +def bangkok_now() -> dt.datetime: + """Current wall-clock in Bangkok (Asia/Bangkok, UTC+7, no DST).""" + return dt.datetime.now(dt.timezone.utc).astimezone(_TZ_BANGKOK) + + +def yesterday_bangkok() -> dt.date: + """Recommended default end: yesterday in Bangkok.""" + return (bangkok_now() - dt.timedelta(days=1)).date() + + +def _parse_date(value: Any) -> dt.date: + try: + if isinstance(value, str): + return dt.date.fromisoformat(str(value)[:10]) + if isinstance(value, dt.datetime): + return value.date() + if isinstance(value, dt.date): + return value + except (TypeError, ValueError) as exc: + raise BacktestReadinessError(f"invalid date: {value!r}") from exc + raise BacktestReadinessError(f"invalid date: {value!r}") + + +def _date_from_ts(value: Any) -> Optional[dt.date]: + """Extract a date from an ISO-8601 timestamp, tolerant of partial input.""" + if not value: + return None + try: + return _parse_date(value) + except BacktestReadinessError: + return None + + +def _factor_keys_required() -> list[str]: + """The registry factors the PIT scorer needs (single source of truth).""" + from .factors import FACTORS + if isinstance(FACTORS, dict): + return [str(k) for k in FACTORS.keys()] + # defensive fallback (registry is a dict today; keep a tolerant path) + keys: list[str] = [] + for item in FACTORS: # type: ignore[union-attr] + if isinstance(item, dict) and isinstance(item.get("key"), str): + keys.append(str(item.get("key"))) + return keys + + +def _required_symbols() -> list[str]: + """The universe the backtest trades. Uses the board registry when present, + else falls back to anything found in the price series at evaluation time.""" + try: + from .themes import THEME_SYMBOLS + out: list[str] = [] + if isinstance(THEME_SYMBOLS, dict): + for v in THEME_SYMBOLS.values(): + if isinstance(v, str): + out.append(v) + elif isinstance(v, list): + for s in v: + if isinstance(s, str): + out.append(s) + elif isinstance(THEME_SYMBOLS, list): + for s in THEME_SYMBOLS: + if isinstance(s, str): + out.append(s) + return sorted(set(out)) + except Exception: # pragma: no cover - registry unavailable + return [] + + +@dataclass +class DataCoverage: + """Per-source earliest-available date summary for one requested window.""" + + source: str + available: bool + earliest: Optional[dt.date] = None + latest: Optional[dt.date] = None + detail: str = "" + missing_count: int = 0 + + +@dataclass +class BacktestReadiness: + """Strict PIT readiness verdict + recommended default dates.""" + + ready: bool + recommended_start: Optional[str] = None + recommended_end: Optional[str] = None + reason: str = "" + missing: list[str] = field(default_factory=list) + coverage: list[DataCoverage] = field(default_factory=list) + timezone: str = "Asia/Bangkok" + + def to_dict(self) -> dict[str, Any]: + return { + "ready": self.ready, + "recommended_start": self.recommended_start, + "recommended_end": self.recommended_end, + "reason": self.reason, + "missing": self.missing, + "coverage": [ + { + "source": c.source, + "available": c.available, + "earliest": c.earliest.isoformat() if c.earliest else None, + "latest": c.latest.isoformat() if c.latest else None, + "detail": c.detail, + "missing_count": c.missing_count, + } + for c in self.coverage + ], + "timezone": self.timezone, + } + + +def _price_coverage(series: dict[str, Any]) -> DataCoverage: + """Earliest/latest trading date shared across the available symbols. + + Uses the intersection of symbol availability so that the recommended start + is a date every held symbol can actually be valued, not just one symbol. + """ + min_dates: list[dt.date] = [] + max_dates: list[dt.date] = [] + for sym, s in series.items(): + bars = (s or {}).get("bars", []) + if not bars: + continue + dates = [b.get("date") for b in bars if b.get("date")] + dates = [d for d in dates if d is not None] + if not dates: + continue + parsed = [_parse_date(d) for d in dates] + min_dates.append(min(parsed)) + max_dates.append(max(parsed)) + if not min_dates: + return DataCoverage( + source="price", available=False, detail="no price bars on disk", + ) + return DataCoverage( + source="price", + available=True, + earliest=max(min_dates), + latest=min(max_dates), + detail=f"{len(min_dates)} symbols", + ) + + +def evaluate_readiness( + *, + factor_store: Any = None, + siamchart_store: Any = None, + price_series: Optional[dict[str, Any]] = None, + start: Optional[str] = None, + end: Optional[str] = None, +) -> BacktestReadiness: + """Evaluate strict PIT readiness and derive recommended default dates. + + Args: + factor_store: a FactorVintageStore (or object exposing ``series(key)`` + and ``value_at(key, as_of)``). + siamchart_store: a SiamchartVintageStore exposing ``list_ids()`` and + ``snapshot_at(as_of)``. + price_series: the Yahoo price series dict ``{sym: {bars: [...]}}``. + start/end: optional explicit window; if given, readiness is evaluated + only against ``[start, end]``. + + "Ready" requires, within the window: + 1. every registry factor has a released value by the start date; + 2. a Siamchart snapshot retrieved no later than the start date; + 3. at least one executable price in the window for every required symbol. + Missing inputs are reported explicitly rather than silently skipped. + """ + miss: list[str] = [] + cov: list[DataCoverage] = [] + + today = yesterday_bangkok() + # When an explicit start is requested, coverage is evaluated *at* that + # start (the inputs must be knowable by then). Otherwise evaluate through + # the end date to discover whether a usable window exists at all. + if start is not None: + coverage_cutoff = _parse_date(start) + else: + coverage_cutoff = _date_from_ts(end) or today + + # ---------------- price ---------------- + if price_series is None: + from .simulation import load_price_snapshot + try: + price_series = load_price_snapshot() + except Exception as exc: # SimulationError / OSError / JSON + price_series = {} + cov.append(DataCoverage( + source="price", available=False, detail=f"cannot load: {exc}", + )) + miss.append("price") + fac_cov, fac_miss, _earliest = _factor_coverage( + factor_store, coverage_cutoff + ) + cov.extend(fac_cov) + miss.extend(fac_miss) + return BacktestReadiness( + ready=False, reason="price coverage missing", missing=miss, + coverage=cov, recommended_end=today.isoformat(), + ) + pc = _price_coverage(price_series) + cov.append(pc) + if not pc.available: + miss.append("price") + + # ---------------- factor ---------------- + fac_cov, fac_miss, fac_earliest = _factor_coverage( + factor_store, coverage_cutoff + ) + cov.extend(fac_cov) + miss.extend(fac_miss) + + # ---------------- siamchart ---------------- + scv, sc_earliest = _siamchart_coverage(siamchart_store, coverage_cutoff) + cov.append(scv) + if not scv.available: + miss.append("siamchart") + + # derive recommended start = the latest earliest-available date among the + # inputs (the point at which *all* of them are simultaneously available). + candidates = [ + d for d in (fac_earliest, sc_earliest, pc.earliest) + if d is not None + ] + recommended_end = _date_from_ts(end) or today + if start is not None: + recommended_start = _parse_date(start) + elif candidates: + recommended_start = max(candidates) + else: + recommended_start = None + + # cap recommended start so it never exceeds the end + if recommended_start and recommended_end and recommended_start > recommended_end: + recommended_start = recommended_end + + # explicit window requested: readiness is whether the window is covered + if start is not None: + s = _parse_date(start) + e_ = _parse_date(end) if end else today + ready = (not miss) and s <= e_ + else: + ready = (not miss) and bool(recommended_start) + + reason = "" + rs = recommended_start.isoformat() if recommended_start else None + re_iso = recommended_end.isoformat() + if ready: + reason = ( + f"strict PIT coverage from {rs} to {re_iso}" + ) + elif miss: + reason = "missing inputs: " + ", ".join(sorted(set(miss))) + else: + reason = "no usable PIT-ready window" + + return BacktestReadiness( + ready=ready, + recommended_start=recommended_start.isoformat() if recommended_start else None, + recommended_end=re_iso, + reason=reason, + missing=sorted(set(miss)), + coverage=cov, + ) + + +def _factor_coverage( + store: Any, cutoff: Optional[dt.date] +) -> tuple[list[DataCoverage], list[str], Optional[dt.date]]: + """Earliest released date across all registry factors, or missing list.""" + if store is None: + return [ + DataCoverage( + source="factor", available=False, detail="factor store not provided", + ) + ], ["factor"], None + factors = _factor_keys_required() + earliest_dates: list[dt.date] = [] + missing_factor: list[str] = [] + cutoff_ts = None + if cutoff is not None: + cutoff_ts = dt.datetime.combine(cutoff, dt.time.min, tzinfo=_TZ_BANGKOK) + for key in factors: + try: + rows = store.series(key) + except Exception: + rows = [] + release_dates = [ + _date_from_ts(r.get("released_at")) or _date_from_ts(r.get("observed_at")) + for r in rows + ] + release_dates = [d for d in release_dates if d is not None] + if cutoff_ts is not None: + release_dates = [ + d for d in release_dates + if _release_le(d, cutoff_ts) + ] + if not release_dates: + missing_factor.append(key) + continue + if cutoff_ts is None: + earliest_dates.append(min(release_dates)) + else: + earliest_dates.append(min(release_dates)) + cov = DataCoverage( + source="factor", + available=not missing_factor, + earliest=min(earliest_dates) if earliest_dates else None, + missing_count=len(missing_factor), + detail=( + f"{len(earliest_dates)}/{len(factors)} factors have released values" + if earliest_dates + else "no factor has a released value" + ), + ) + # umbrella missing token so consumers see "factor" plus per-key detail + missing_tokens: list[str] = ["factor"] if missing_factor else [] + missing_tokens.extend(missing_factor) + return [cov], missing_tokens, (min(earliest_dates) if earliest_dates else None) + + +def _release_le(d: dt.date, cutoff_dt: dt.datetime) -> bool: + """True if a release date is at/before the cutoff (same-day counts).""" + day = dt.datetime.combine(d, dt.time.min, tzinfo=_TZ_BANGKOK) + return day <= cutoff_dt + + +def _siamchart_coverage( + store: Any, cutoff: Optional[dt.date] = None +) -> tuple[DataCoverage, Optional[dt.date]]: + """Earliest retrieved Siamchart snapshot date (or missing). + + When ``cutoff`` is given, the snapshot must have been retrieved by that + date (strict PIT: it must be knowable at the requested start). + """ + if store is None: + return DataCoverage( + source="siamchart", available=False, detail="siamchart store not provided" + ), None + try: + ids = store.list_ids() + except Exception: + ids = [] + if not ids: + return DataCoverage( + source="siamchart", available=False, + detail="no snapshots in siamchart vintage store", + ), None + # Collect retrieval timestamps of every stored snapshot. The real store + # exposes its manifest; a fake may only answer snapshot_at(now) (newest). + retrieved_dates: list[dt.date] = [] + try: + manifest = store._load_manifest() # real SiamchartVintageStore + for e in manifest.get("snapshots", {}).values(): + d = _date_from_ts(e.get("retrieved_at")) + if d is not None: + retrieved_dates.append(d) + except Exception: + retrieved_dates = [] + if not retrieved_dates: + # fallback: fake / minimal store -> newest snapshot's own retrieved_at + try: + snap = store.snapshot_at( + dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat() + ) + d = _date_from_ts(snap.get("_retrieved_at") or snap.get("retrieved_at")) + if d is not None: + retrieved_dates.append(d) + except Exception: + pass + if not retrieved_dates: + return DataCoverage( + source="siamchart", available=True, detail=f"{len(ids)} snapshots stored" + ), None + earliest = min(retrieved_dates) + if cutoff is not None: + known_by_cutoff = [d for d in retrieved_dates if d <= cutoff] + if not known_by_cutoff: + return DataCoverage( + source="siamchart", available=False, earliest=earliest, + detail=f"no snapshot retrieved by {cutoff.isoformat()}", + ), earliest + earliest = min(known_by_cutoff) + return DataCoverage( + source="siamchart", available=True, earliest=earliest, + detail=f"{len(ids)} snapshots stored", + ), earliest diff --git a/backend/tests/test_backtest_readiness.py b/backend/tests/test_backtest_readiness.py new file mode 100644 index 0000000..23b3f60 --- /dev/null +++ b/backend/tests/test_backtest_readiness.py @@ -0,0 +1,195 @@ +"""Tests for strict PIT backtest readiness + default-date derivation (Task 1). + +Covered scenarios follow the acceptance criteria: + + * no factor vintages -> ready=false, factor missing listed + * no Siamchart snapshot -> ready=false, siamchart missing listed + * price coverage starting after PIT factors -> recommended start = + latest of the first-ready dates (strict: all inputs must be available) + * recommended end = yesterday in Bangkok, bounded by latest price date + * an explicit start earlier than readiness -> ready=false + +All stores are lightweight fakes so tests stay deterministic and offline. +""" + +from __future__ import annotations + +import datetime as dt +import unittest + +from app.backtest_readiness import ( + BacktestReadiness, + evaluate_readiness, + yesterday_bangkok, +) + + +class FakeFactorStore: + """Minimal factor store exposing series()/value_at() for readiness tests.""" + + def __init__(self, releases: dict[str, list[str]]): + # factor_key -> list of ISO released_at timestamps (earliest first) + self._releases = releases + + def series(self, key: str) -> list[dict]: + rows = [] + for ts in self._releases.get(key, []): + rows.append({"released_at": ts, "observed_at": ts, "value": 1.0}) + return rows + + def value_at(self, key: str, as_of: str): + for ts in reversed(self._releases.get(key, [])): + if ts <= as_of: + return 1.0 + return None + + +class FakeSiamchartStore: + def __init__(self, retrieved: list[str]): + self._retrieved = sorted(retrieved) + + def list_ids(self) -> list[str]: + return [str(i) for i in range(len(self._retrieved))] + + def snapshot_at(self, as_of: str) -> dict: + chosen = [t for t in self._retrieved if t <= as_of] + if not chosen: + return {} + return {"retrieved_at": chosen[-1], "_retrieved_at": chosen[-1]} + + +def make_price_series( + symbols: list[str], start: str, end: str, step_days: int = 30 +) -> dict: + """A price series {sym: {bars: [...]}} covering [start, end] for every sym.""" + s = dt.date.fromisoformat(start) + e = dt.date.fromisoformat(end) + bars = [] + cur = s + while cur <= e: + bars.append({"date": cur.isoformat(), "adjusted_close": 10.0}) + cur += dt.timedelta(days=step_days) + return {sym: {"bars": list(bars)} for sym in symbols} + + +def today_iso() -> str: + return yesterday_bangkok().isoformat() + + +# A factor store that has released every registry factor by a known date. +def full_factor_store(release_date: str) -> FakeFactorStore: + from app.backtest_readiness import _factor_keys_required + releases = { + key: [f"{release_date}T09:00:00+07:00"] for key in _factor_keys_required() + } + return FakeFactorStore(releases) + + +class YesterdayDefaultTest(unittest.TestCase): + def test_yesterday_is_bangkok_tz(self): + y = dt.date.fromisoformat(yesterday_bangkok().isoformat()) + # just assert it's a valid date one day before "now" + self.assertIsInstance(y, dt.date) + # and timezone is +07 (Bangkok has no DST) + import app.backtest_readiness as r + now = r.bangkok_now() + off = now.utcoffset() + assert off is not None + self.assertEqual(off.total_seconds(), 7 * 3600) + + +class NoFactorVintagesTest(unittest.TestCase): + def test_blocks_when_no_factor_release(self): + store = FakeFactorStore({}) # no factor ever released + sc = FakeSiamchartStore(["2025-01-01T09:00:00+07:00"]) + series = make_price_series(["A"], "2020-01-01", "2026-08-01") + res = evaluate_readiness( + factor_store=store, siamchart_store=sc, price_series=series + ) + self.assertFalse(res.ready) + self.assertIn("factor", res.missing) + + +class NoSiamchartSnapshotTest(unittest.TestCase): + def test_blocks_when_no_snapshot(self): + store = full_factor_store("2025-01-01") + sc = FakeSiamchartStore([]) # no snapshot + series = make_price_series(["A"], "2020-01-01", "2026-08-01") + res = evaluate_readiness( + factor_store=store, siamchart_store=sc, price_series=series + ) + self.assertFalse(res.ready) + self.assertIn("siamchart", res.missing) + + +class RecommendedStartTest(unittest.TestCase): + def test_start_is_latest_of_first_ready_dates(self): + # factors ready 2025-01-01, siamchart ready 2025-06-01, price from 2024 + store = full_factor_store("2025-01-01") + sc = FakeSiamchartStore(["2025-06-01T09:00:00+07:00"]) + series = make_price_series(["A"], "2024-01-01", "2026-08-01") + res = evaluate_readiness( + factor_store=store, siamchart_store=sc, price_series=series + ) + self.assertTrue(res.ready) + self.assertEqual(res.recommended_start, "2025-06-01") + self.assertEqual(res.recommended_end, today_iso()) + + def test_start_limited_by_price_when_price_latest(self): + # factors + siamchart ready 2026-05-01, but price only from 2026-06-01 + store = full_factor_store("2026-05-01") + sc = FakeSiamchartStore(["2026-05-01T09:00:00+07:00"]) + series = make_price_series(["A"], "2026-06-01", "2026-08-01") + res = evaluate_readiness( + factor_store=store, siamchart_store=sc, price_series=series + ) + self.assertTrue(res.ready) + self.assertEqual(res.recommended_start, "2026-06-01") + + +class ExplicitWindowTest(unittest.TestCase): + def test_explicit_start_before_readiness_blocks(self): + store = full_factor_store("2025-06-01") + sc = FakeSiamchartStore(["2025-06-01T09:00:00+07:00"]) + series = make_price_series(["A"], "2024-01-01", "2026-08-01") + # user asks for start 2024-01-01, but PIT only ready from 2025-06-01 + res = evaluate_readiness( + factor_store=store, siamchart_store=sc, price_series=series, + start="2024-01-01", end="2026-08-01", + ) + # factors missing before start -> blocked + self.assertFalse(res.ready) + self.assertIn("factor", res.missing) + + def test_explicit_start_after_readiness_is_ready(self): + store = full_factor_store("2025-01-01") + sc = FakeSiamchartStore(["2025-01-01T09:00:00+07:00"]) + series = make_price_series(["A"], "2024-01-01", "2026-08-01") + res = evaluate_readiness( + factor_store=store, siamchart_store=sc, price_series=series, + start="2025-06-01", end="2026-08-01", + ) + self.assertTrue(res.ready) + + +class CoverageShapeTest(unittest.TestCase): + def test_to_dict_includes_missing_and_coverage(self): + res = evaluate_readiness( + factor_store=FakeFactorStore({}), + siamchart_store=FakeSiamchartStore([]), + price_series=make_price_series(["A"], "2024-01-01", "2026-08-01"), + ) + d = res.to_dict() + self.assertIn("ready", d) + self.assertIn("missing", d) + self.assertIn("coverage", d) + self.assertEqual(d["timezone"], "Asia/Bangkok") + + def test_dataclass_defaults(self): + r = BacktestReadiness(ready=False) + self.assertEqual(r.missing, []) + self.assertEqual(r.coverage, []) + + +if __name__ == "__main__": + unittest.main()