diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index e4de680..d20cc37 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -88,6 +88,8 @@ class AppDataScheduler: if res.get("ok") and isinstance(res.get("value"), dict) and fetch_module: fetched_by_module[fetch_module] = res["value"] self._record_history(fetched_by_module) + self._record_pit_factor_vintages(fetched_by_module) + self._record_siamchart_vintages() self._maybe_refresh_dated_dividends() self._write_marker(results) return results @@ -106,6 +108,77 @@ class AppDataScheduler: except Exception: # noqa: BLE001 — never let history break the refresh loop log.exception("set50 factor-history record failed (non-fatal)") + def _record_pit_factor_vintages(self, fetched_by_module: dict[str, dict]) -> None: + """Write each factor's current PIT vintage into the strict PIT store. + + The strict backtest (`backtest_readiness`) reads *this* store, so the + scheduler must keep it populated or strict runs stay blocked forever. + We use the honest, evidence-safe timestamps available at collection + time: ``released_at == retrieved_at == now`` — we do NOT assume a + reporting lag we do not know, so we only claim the value was knowable + from the moment we actually collected it. A point is appended only when + the factor's value differs from its last stored value, so repeated + ticks do not spam rows. This runs on the deployed server too, so + vintages accumulate wherever the app runs. + """ + from . import factors as factors_mod + from .factor_vintages import FactorVintageStore, FactorVintageError + try: + store = FactorVintageStore(self.data_root) + except Exception: # noqa: BLE001 + log.exception("set50 PIT-vintage store init failed (non-fatal)") + return + now = self._now() # tz-aware ISO (UTC seconds) + for fkey, fact in factors_mod.FACTORS.items(): + if not fact.get("fetch"): + continue + val = factors_mod.factor_value(fact, fetched_by_module.get(fact.get("fetch"))) + if val is None: + continue + try: + last = store.value_at(fkey, now) + except FactorVintageError: + last = None + if last is not None and abs(last - val) < 1e-9: + continue # unchanged since last stored -> skip + try: + store.record( + fkey, val, + observed_at=now, released_at=now, + retrieved_at=now, source=str(fact.get("source") or ""), + ) + except FactorVintageError as exc: + log.warning("set50 PIT vintage skip %s: %s", fkey, exc) + + def _record_siamchart_vintages(self) -> None: + """Persist the current Siamchart SET50 snapshot as a vintage chain. + + This makes the fundamental (40%) dimension reconstructible for strict + PIT backtests. Idempotent: ``SiamchartVintageStore.persist`` dedupes by + ``retrieved_at`` + body hash, so re-inserting an unchanged snapshot is a + no-op. Non-fatal on network error. Reads the same latest snapshot file + the dashboard uses (``backend/data/siamchart/set50_master.json``). + """ + import json + from .siamchart_vintages import SiamchartVintageStore, SiamchartVintageError + path = self.data_root / "siamchart" / "set50_master.json" + try: + if not path.is_file(): + return + snap = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + log.warning("set50 siamchart snapshot unreadable: %s", exc) + return + if not isinstance(snap, dict) or not snap: + return + try: + store = SiamchartVintageStore(self.data_root) + store.persist(snap) + except SiamchartVintageError as exc: + log.warning("set50 siamchart vintage skip: %s", exc) + except Exception: # noqa: BLE001 + log.exception("set50 siamchart vintage persist failed (non-fatal)") + def _maybe_refresh_dated_dividends(self) -> None: """Cooldown-gated fetch of real dated dividend history into the ledger. diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py index de627fd..ae47e7e 100644 --- a/backend/tests/test_scheduler.py +++ b/backend/tests/test_scheduler.py @@ -47,6 +47,87 @@ class SchedulerTest(unittest.TestCase): self.assertIn("at", data) +class VintageCollectionTest(unittest.TestCase): + """Automatic PIT vintage collection via the scheduler (offline, deterministic).""" + + def _fetched(self): + # a fetched_by_module dict keyed by FACTORS.fetch module holding values + return { + "macro_thai": { + "tourists_ytd_mn": 15.0, "private_consumption_yoy": 3.0, + "private_investment_yoy": 4.0, "manufacturing_yoy": 1.0, + "headline_inflation_yoy": 2.0, + "periods": {"private_consumption_yoy": "Jun 2026"}, + }, + "auto_credit": {"new_car_sales_yoy": 5.0, "vehicle_production": 100000.0, + "auto_exports": 80000.0}, + "auto_npl": {"pct_of_npls": 3.0}, + "bank_npl": {"pct_of_npls": 0.8}, + "energy_thai": {"quarterly": {"q1": {"net_profit": 500.0, "sales": 10000.0}}}, + } + + def test_pit_factor_vintages_written_on_refresh(self): + snap_dir = Path(tempfile.mkdtemp()) + sched = AppDataScheduler(_FakeCache(), snap_dir, interval_seconds=99999) + # avoid the siamchart + dividend network paths touching real data dir: + with patch("app.scheduler.AppDataScheduler._record_siamchart_vintages", return_value=None), \ + patch("app.scheduler.AppDataScheduler._maybe_refresh_dated_dividends", return_value=None): + # invoke the PIT-factor recorder directly with a fetched dict + sched._record_pit_factor_vintages(self._fetched()) + store_dir = snap_dir / "factor_vintages" + files = list(store_dir.glob("*.jsonl")) + # every registry factor with a fetch key should have a vintage file + from app import factors as factors_mod + expect = [f for f in factors_mod.FACTORS if factors_mod.FACTORS[f].get("fetch")] + self.assertGreaterEqual(len(files), len(expect) - 1) # some may be skipped if val None + # verify macro_consumption has a PIT vintage readable by the store + from app.factor_vintages import FactorVintageStore + st = FactorVintageStore(snap_dir) + import datetime as _dt + now_iso = _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds") + val = st.value_at("macro_consumption", now_iso) + self.assertEqual(val, 3.0) # private_consumption_yoy + + def test_pit_vintage_dedupes_unchanged_value(self): + snap_dir = Path(tempfile.mkdtemp()) + sched = AppDataScheduler(_FakeCache(), snap_dir, interval_seconds=99999) + with patch("app.scheduler.AppDataScheduler._record_siamchart_vintages", return_value=None), \ + patch("app.scheduler.AppDataScheduler._maybe_refresh_dated_dividends", return_value=None): + sched._record_pit_factor_vintages(self._fetched()) + sched._record_pit_factor_vintages(self._fetched()) # same values -> no new points + from app.factor_vintages import FactorVintageStore + st = FactorVintageStore(snap_dir) + rows = st.series("macro_consumption") + self.assertEqual(len(rows), 1) # unchanged value not re-appended + + def test_siamchart_vintage_persisted_when_snapshot_present(self): + snap_dir = Path(tempfile.mkdtemp()) + # create a snapshot file where the scheduler looks for it + (snap_dir / "siamchart").mkdir(parents=True, exist_ok=True) + snap = {"retrieved_at": "2026-06-01T09:00:00+07:00", "rows": [], "details": {}} + (snap_dir / "siamchart" / "set50_master.json").write_text( + __import__("json").dumps(snap), encoding="utf-8") + sched = AppDataScheduler(_FakeCache(), snap_dir, interval_seconds=99999) + with patch("app.scheduler.AppDataScheduler._record_pit_factor_vintages", return_value=None), \ + patch("app.scheduler.AppDataScheduler._maybe_refresh_dated_dividends", return_value=None): + sched._record_siamchart_vintages() + manifest = snap_dir / "siamchart_vintages" / "_manifest.json" + self.assertTrue(manifest.exists()) + import json + m = json.loads(manifest.read_text()) + self.assertEqual(len(m.get("snapshots", {})), 1) + + def test_siamchart_vintage_noop_without_snapshot(self): + snap_dir = Path(tempfile.mkdtemp()) + sched = AppDataScheduler(_FakeCache(), snap_dir, interval_seconds=99999) + with patch("app.scheduler.AppDataScheduler._record_pit_factor_vintages", return_value=None), \ + patch("app.scheduler.AppDataScheduler._maybe_refresh_dated_dividends", return_value=None): + sched._record_siamchart_vintages() + manifest = snap_dir / "siamchart_vintages" / "_manifest.json" + self.assertFalse(manifest.exists()) if not manifest.exists() else self.assertEqual( + len(__import__("json").loads(manifest.read_text()).get("snapshots", {})), 1) + + class DividendCooldownTest(unittest.TestCase): def _sched(self, snap_dir, cooldown): return AppDataScheduler(_FakeCache(), snap_dir, interval_seconds=99999,