"""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_before_price_history_blocks(self): # Regression (cycle-1 review): explicit start that predates ALL usable # price history must not manufacture a false strict-PIT window. Factor + # Siamchart are ready at the start, but price is not -> block. store = full_factor_store("2024-01-01") sc = FakeSiamchartStore(["2024-01-01T09:00:00+07:00"]) # price only starts 2026-06-01 series = make_price_series(["A"], "2026-06-01", "2026-08-01") res = evaluate_readiness( factor_store=store, siamchart_store=sc, price_series=series, start="2025-01-01", end="2026-08-01", ) self.assertFalse(res.ready) self.assertIn("price", 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()