From 887e9c920810bc55d45730ccb18e05a93a5b3e88 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Thu, 27 Aug 2026 11:55:36 +0700 Subject: [PATCH] [verified] PIT siamchart vintage store un-partials the fundamental dimension MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add an append-only, hash-chained store of every collected Siamchart fundamental snapshot so the 40% fundamental dimension can be reconstructed at a historical date instead of always reading the latest snapshot: - backend/app/siamchart_vintages.py: SiamchartVintageStore persists each snapshot under its retrieved_at with a SHA-256 canonical hash chain (tamper/reorder detectable); snapshot_at(as_of) returns the newest snapshot whose retrieved_at <= as_of (anti-look-ahead), and fails closed (returns {}) when none is knowable yet. Deduplicates identical retrieved_at+body persists. - backend/app/pit_scorer.py: PitScoreProvider accepts siamchart_store; when wired, siamchart_factor_view reads the snapshot knowable at as_of (pit_grade='pit') instead of the current snapshot (pit_grade='current'). score_board no longer forces partial_pit when a store is present — the fundamental dimension is PIT; the theme dimension still fails closed (pit=false) unless every theme factor has a released PIT value by as_of. - backend/app/__init__.py: /api/v1/backtest use_pit seeds the first vintage from the current snapshot (idempotent) and wires the store. - tests: store (6) + scorer-with-store anti-look-ahead (1) — full backend suite 273 passed. Honest scope: snapshots are stored whole and reconstructible forward; EPS year-keys inside a snapshot are not tied to calendar years, so EPS growth stays latest-vs-prior (not fiscal-year-pinned). No history before the first collected snapshot exists. --- backend/app/__init__.py | 22 ++- backend/app/pit_scorer.py | 40 +++-- backend/app/siamchart_vintages.py | 180 +++++++++++++++++++++++ backend/tests/test_pit_scorer.py | 26 +++- backend/tests/test_siamchart_vintages.py | 74 ++++++++++ 5 files changed, 330 insertions(+), 12 deletions(-) create mode 100644 backend/app/siamchart_vintages.py create mode 100644 backend/tests/test_siamchart_vintages.py diff --git a/backend/app/__init__.py b/backend/app/__init__.py index fc87449..3282918 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -54,6 +54,20 @@ def _load_siamchart_snapshot() -> dict[str, Any]: return data +def _seed_siamchart_vintage(store) -> None: + """Persist the current on-disk siamchart snapshot as the first vintage, if + the store is empty. Intentionally idempotent (store.deduplicates by + retrieved_at + body).""" + try: + from .siamchart_vintages import SiamchartVintageStore + if isinstance(store, SiamchartVintageStore) and not store.list_ids(): + snap = _load_siamchart_snapshot() + if snap: + store.persist(snap) + except Exception: # noqa: BLE001 — seeding must never break the route + pass + + def _signal_summary(result: dict[str, Any]) -> dict[str, int]: signals = result["signals"] return { @@ -735,10 +749,16 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: if use_pit: from pathlib import Path as _Path from .factor_vintages import FactorVintageStore + from .siamchart_vintages import SiamchartVintageStore from .pit_scorer import PitScoreProvider, make_pit_score_fn froot = _Path(__file__).resolve().parents[1] / "data" store = FactorVintageStore(froot) - provider = PitScoreProvider(store, _load_siamchart_snapshot()) + sstore = SiamchartVintageStore(froot) + # seed the first vintage from the current snapshot so the store + # has a known baseline (no-op if already stored). + _seed_siamchart_vintage(sstore) + provider = PitScoreProvider(store, _load_siamchart_snapshot(), + siamchart_store=sstore) score_fn = make_pit_score_fn(provider) res = run_backtest(start, end, capital=capital, rebalance_freq=freq, score_fn=score_fn, diff --git a/backend/app/pit_scorer.py b/backend/app/pit_scorer.py index 84c2f15..382b0ef 100644 --- a/backend/app/pit_scorer.py +++ b/backend/app/pit_scorer.py @@ -65,9 +65,11 @@ class PitScoreProvider: siamchart_snapshot: Optional[dict] = None, *, theme_factor_map: Optional[dict[str, list[dict]]] = None, + siamchart_store=None, ) -> None: self.factor_store = factor_store self.siamchart_snapshot = siamchart_snapshot + self.siamchart_store = siamchart_store # theme_key -> list of {key, weight}; defaults to the registry THEMES. self.theme_factor_map = theme_factor_map or _default_theme_factor_map() @@ -77,19 +79,32 @@ class PitScoreProvider: return self.factor_store.value_at(factor_key, as_of) # -- fundamental PIT (partial) --------------------------------------- + def _snapshot_at(self, as_of: str) -> dict: + """Newest siamchart snapshot knowable at ``as_of``. + + When a ``siamchart_store`` is provided it is the PIT source (snapshots + retrieved <= as_of). Otherwise we fall back to the single current + snapshot (``self.siamchart_snapshot``) which is **not** point-in-time — + callers must treat the fundamental dimension as partial in that case. + """ + if self.siamchart_store is not None: + return self.siamchart_store.snapshot_at(as_of) or {} + return self.siamchart_snapshot or {} + def siamchart_factor_view(self, as_of: str) -> dict[str, dict]: - """Per-symbol fundamental dict at ``as_of`` (partial PIT). + """Per-symbol fundamental dict at ``as_of``. Returns {symbol: {eps_growth_yoy, dividend_yield, is_dividend, - pit_grade}}. eps_growth_yoy is PIT-grade (derived from the 5-yr series); - dividend_yield / is_dividend are current snapshot values and are marked - ``pit_grade='current'`` so the caller knows the fundamental dimension is - not fully point-in-time yet. + pit_grade}}. eps_growth_yoy is PIT-grade (derived from the 5-yr series). + ``pit_grade`` is ``'pit'`` when read from the PIT snapshot store (the + snapshot was knowable at as_of), ``'current'`` otherwise (no store -> + current snapshot, not point-in-time). """ out: dict[str, dict] = {} - snap = self.siamchart_snapshot + snap = self._snapshot_at(as_of) if not snap: return out + pit_grade = "pit" if self.siamchart_store is not None else "current" details = snap.get("details", {}) for row in snap.get("rows", []): symbol = row.get("symbol") @@ -101,7 +116,7 @@ class PitScoreProvider: "eps_growth_yoy": _eps_growth_from_series(row.get("eps", {})), "dividend_yield": yield_, "is_dividend": bool(yield_ and yield_ > 0), - "pit_grade": "partial", # share-level growth from series, ratios current + "pit_grade": pit_grade, } out[symbol] = essential return out @@ -198,6 +213,9 @@ class PitScoreProvider: combined = themes_mod.combine_score(list(theme_scores.values()), siamchart_score) is_full_pit = not blocked_themes + # fundamental is PIT only when a siamchart vintage store is wired; + # otherwise the current snapshot makes the overall result partial. + fundamental_pit = self.siamchart_store is not None out: dict[str, dict] = {} for sym, meta in combined.items(): fm = fv.get(sym, {}) @@ -208,10 +226,12 @@ class PitScoreProvider: "is_dividend": bool(fm.get("is_dividend", False)), "dividend_yield": fm.get("dividend_yield") or 0.0, "pit_meta": { - "pit": is_full_pit, - "partial_pit": True, # siamchart ratios are current, not PIT + "pit": is_full_pit and fundamental_pit, + "partial_pit": not fundamental_pit, "blocked_theme": blocked_themes, - "note": "theme surprises PIT; siamchart fundamental partial (current ratios)", + "note": ("theme surprises PIT + siamchart fundamental PIT (store)" + if fundamental_pit else + "theme surprises PIT; siamchart fundamental partial (current snapshot)"), }, } # fall back to the current board for any symbol the PIT path could not diff --git a/backend/app/siamchart_vintages.py b/backend/app/siamchart_vintages.py new file mode 100644 index 0000000..fbe7db9 --- /dev/null +++ b/backend/app/siamchart_vintages.py @@ -0,0 +1,180 @@ +"""Point-in-time (PIT) Siamchart fundamental snapshot store. + +The dashboard currently reads a single latest snapshot +(``data/siamchart/set50_master.json``); older snapshots are overwritten, so the +fundamental dimension (40% of the combined score) is only ever "as of today" — +not reconstructible at a historical date. That is why the PIT score provider +flags the fundamental dimension as ``partial``. + +This module adds an append-only, hash-chained store of **every collected +Siamchart snapshot**, keyed by its ``retrieved_at``. Future-ward it lets a PIT +scorer answer "what fundamental ratios were actually published as of date T" +by returning the newest snapshot whose ``retrieved_at <= T`` (anti-look-ahead). + +Integrity mirrors ``factor_vintages`` and the price observation store: each +snapshot is stored with a canonical-JSON SHA-256 of its payload, chained to the +previous snapshot's hash, so tampering or reordering is detectable. + +Honest scope: this store only makes the fundamental dimension *reconstructible +from collected snapshots forward*. It does not backfill history that was never +collected, and a snapshot provides ratios + 5-year EPS but the EPS year-keys +are not tied to calendar years inside the snapshot (so EPS growth is derived +latest-vs-prior, not pinned to a fiscal year). Snapshots are stored whole; the +per-symbol interpretation lives in ``siamchart_factors`` / ``pit_scorer``. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Mapping, Optional + +_NORMALIZED_HASH_ALGORITHM = "sha256-json-canonical-v1" + + +class SiamchartVintageError(ValueError): + """Raised when a snapshot cannot be safely stored or loaded.""" + + +def _parse_timestamp(value: Any) -> datetime: + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except (TypeError, ValueError) as exc: + raise SiamchartVintageError("timestamp must be ISO-8601 with timezone") from exc + if parsed.tzinfo is None: + raise SiamchartVintageError("timestamp must include a timezone (was naive)") + return parsed + + +def _canonical_json(payload: Mapping[str, Any]) -> str: + return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +class SiamchartVintageStore: + """Append-only, hash-chained store of collected Siamchart snapshots.""" + + def __init__(self, root: Path | str) -> None: + self.root = Path(root).resolve() + self.dir = self.root / "siamchart_vintages" + self.dir.mkdir(parents=True, exist_ok=True) + self.manifest_path = self.dir / "_manifest.json" + + # -- manifest --------------------------------------------------------- + def _load_manifest(self) -> dict[str, Any]: + if not self.manifest_path.is_file(): + return {"schema_version": 1, "snapshots": {}} + try: + payload = json.loads(self.manifest_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise SiamchartVintageError("siamchart vintage manifest unreadable") from exc + if not isinstance(payload, dict) or payload.get("schema_version") != 1: + raise SiamchartVintageError("unsupported siamchart vintage manifest schema") + if not isinstance(payload.get("snapshots"), dict): + raise SiamchartVintageError("siamchart vintage manifest has invalid snapshots map") + return payload + + def _save_manifest(self, manifest: dict[str, Any]) -> None: + tmp = self.manifest_path.with_suffix(".tmp") + tmp.write_text(_canonical_json(manifest) + "\n", encoding="utf-8") + tmp.replace(self.manifest_path) + + # -- write ------------------------------------------------------------ + def persist(self, snapshot: Mapping[str, Any]) -> dict[str, Any]: + """Store one collected snapshot under a stable snapshot id. + + ``snapshot["retrieved_at"]`` is the PIT timestamp. Duplicate + ``retrieved_at`` from an equal payload is a no-op (returns existing). + """ + if not isinstance(snapshot, dict): + raise SiamchartVintageError("snapshot must be a dict") + retrieved = snapshot.get("retrieved_at") + if not retrieved: + raise SiamchartVintageError("snapshot requires retrieved_at") + _parse_timestamp(retrieved) + + payload = dict(snapshot) + # strip any prior store-injected fields so hashes are canonical + payload.pop("_store_id", None) + payload.pop("_prev_hash", None) + body_hash = hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() + # preserve readability of the source snapshot on disk in a dated file + filename = f"{retrieved[:10]}_{body_hash[:8]}.json" + + manifest = self._load_manifest() + # idempotent: identical retrieved_at + body already stored -> return existing + for sid, entry in manifest["snapshots"].items(): + if entry.get("retrieved_at") == retrieved and entry.get("body_hash") == body_hash: + return {"_store_id": sid, "existing": True} + + snap_dir = self.dir / "snapshots" + snap_dir.mkdir(parents=True, exist_ok=True) + (snap_dir / filename).write_text(_canonical_json(payload) + "\n", encoding="utf-8") + + # next sequential id + next_id = 1 + for sid in manifest["snapshots"]: + if sid.isdigit(): + next_id = max(next_id, int(sid) + 1) + sid = str(next_id) + entries = list(manifest["snapshots"].values()) + prev_hash = entries[-1]["_hash"] if entries else "" + entry = { + "id": sid, + "retrieved_at": retrieved, + "filename": filename, + "body_hash": body_hash, + "_prev_hash": prev_hash, + } + entry["_hash"] = hashlib.sha256( + _canonical_json({k: v for k, v in entry.items() if k != "_hash"}).encode("utf-8") + ).hexdigest() + manifest["snapshots"][sid] = entry + self._save_manifest(manifest) + return {"_store_id": sid, "existing": False} + + # -- reads ------------------------------------------------------------ + def _chain_ok(self) -> None: + """Verify the manifest chain is intact (no reordering/tampering).""" + manifest = self._load_manifest() + entries = [manifest["snapshots"][sid] for sid in sorted(manifest["snapshots"], key=lambda s: int(s) if s.isdigit() else 9999)] + prev = "" + for e in entries: + if e.get("_prev_hash") != prev: + raise SiamchartVintageError("siamchart vintage chain broken") + check = hashlib.sha256( + _canonical_json({k: v for k, v in e.items() if k != "_hash"}).encode("utf-8") + ).hexdigest() + if check != e.get("_hash"): + raise SiamchartVintageError("siamchart vintage manifest entry hash mismatch") + prev = e["_hash"] + + def snapshot_at(self, as_of: str) -> dict[str, Any]: + """Newest stored snapshot whose ``retrieved_at <= as_of``, or {} if none. + + This is the anti-look-ahead read for a PIT scorer: any snapshot + retrieved after ``as_of`` is invisible. + """ + cutoff = _parse_timestamp(as_of) + self._chain_ok() + manifest = self._load_manifest() + chosen: Optional[dict[str, Any]] = None + chosen_ts: Optional[datetime] = None + for e in manifest["snapshots"].values(): + ts = _parse_timestamp(e["retrieved_at"]) + if ts <= cutoff and (chosen_ts is None or ts > chosen_ts): + chosen = e + chosen_ts = ts + if chosen is None: + return {} + try: + payload = json.loads((self.dir / "snapshots" / chosen["filename"]).read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise SiamchartVintageError("siamchart vintage snapshot unreadable") from exc + payload["_store_id"] = chosen["id"] + payload["_retrieved_at"] = chosen["retrieved_at"] + return payload + + def list_ids(self) -> list[str]: + return sorted(self._load_manifest()["snapshots"].keys(), key=lambda s: int(s) if s.isdigit() else 9999) diff --git a/backend/tests/test_pit_scorer.py b/backend/tests/test_pit_scorer.py index 256d794..6e87e85 100644 --- a/backend/tests/test_pit_scorer.py +++ b/backend/tests/test_pit_scorer.py @@ -77,11 +77,35 @@ class PitScoreProviderTest(unittest.TestCase): provider = PitScoreProvider(self.store, siamchart_snapshot=snap) view = provider.siamchart_factor_view("2026-08-25T00:00:00+07:00") aot = view["AOT"] - self.assertEqual(aot["pit_grade"], "partial") + self.assertEqual(aot["pit_grade"], "current") self.assertAlmostEqual(aot["eps_growth_yoy"], round((1.4 - 1.3) / 1.3 * 100, 2)) self.assertEqual(aot["dividend_yield"], 1.21) self.assertTrue(aot["is_dividend"]) + def test_siamchart_store_reads_snapshot_at_as_of(self): + # With a vintage store wired, the provider reads the snapshot knowable + # at as_of (anti-look-ahead), not the current snapshot. + from app.siamchart_vintages import SiamchartVintageStore + import tempfile + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + sstore = SiamchartVintageStore(tmp.name) + sstore.persist({"retrieved_at": "2026-01-10T00:00:00+07:00", + "rows": [{"symbol": "AOT", "eps": {"1": 1.0, "2": 1.1, "3": 1.2, "4": 1.3, "5": 1.4}}], + "details": {"AOT": {"ratios": {"Yield %": 1.0}}}}) + sstore.persist({"retrieved_at": "2026-06-10T00:00:00+07:00", + "rows": [{"symbol": "AOT", "eps": {"1": 1.0, "2": 1.1, "3": 1.2, "4": 1.3, "5": 1.4}}], + "details": {"AOT": {"ratios": {"Yield %": 9.0}}}}) + provider = PitScoreProvider(self.store, siamchart_snapshot=None, + siamchart_store=sstore) + # at March only the January snapshot is visible -> yield 1.0, pit grade + march = provider.siamchart_factor_view("2026-03-01T00:00:00+07:00") + self.assertEqual(march["AOT"]["dividend_yield"], 1.0) + self.assertEqual(march["AOT"]["pit_grade"], "pit") + # at July the June snapshot is visible -> yield 9.0 + july = provider.siamchart_factor_view("2026-07-01T00:00:00+07:00") + self.assertEqual(july["AOT"]["dividend_yield"], 9.0) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_siamchart_vintages.py b/backend/tests/test_siamchart_vintages.py new file mode 100644 index 0000000..64762cb --- /dev/null +++ b/backend/tests/test_siamchart_vintages.py @@ -0,0 +1,74 @@ +"""Tests for the point-in-time Siamchart snapshot store (PIT enabler).""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from app.siamchart_vintages import SiamchartVintageError, SiamchartVintageStore + + +class SiamchartVintageStoreTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.store = SiamchartVintageStore(Path(self._tmp.name)) + + def tearDown(self): + self._tmp.cleanup() + + def _snap(self, retrieved, symbol="PTT", yield_pct=5.64): + return { + "retrieved_at": retrieved, + "source": "siamchart", + "group": "SET50", + "rows": [{"symbol": symbol, "eps": {"1": 4.0, "2": 4.1, "3": 4.2, "4": 4.3, "5": 4.4}}], + "details": {symbol: {"ratios": {"Yield %": yield_pct, "DPS": 4.0}}}, + } + + def test_persist_and_snapshot_at_roundtrip(self): + self.store.persist(self._snap("2026-01-10T00:00:00+07:00")) + s = self.store.snapshot_at("2026-02-01T00:00:00+07:00") + self.assertEqual(s.get("group"), "SET50") + self.assertEqual(s["_store_id"], "1") + + def test_anti_lookahead_filters_later_snapshots(self): + self.store.persist(self._snap("2026-01-10T00:00:00+07:00", yield_pct=5.0)) + self.store.persist(self._snap("2026-06-15T00:00:00+07:00", yield_pct=9.0)) + # at March, only the January snapshot is visible -> yield 5.0 + s = self.store.snapshot_at("2026-03-01T00:00:00+07:00") + self.assertEqual(s["details"]["PTT"]["ratios"]["Yield %"], 5.0) + # at July, the June snapshot is the newest -> yield 9.0 + s2 = self.store.snapshot_at("2026-07-01T00:00:00+07:00") + self.assertEqual(s2["details"]["PTT"]["ratios"]["Yield %"], 9.0) + + def test_fail_closed_before_first_snapshot(self): + self.store.persist(self._snap("2026-02-10T00:00:00+07:00")) + self.assertEqual(self.store.snapshot_at("2026-01-01T00:00:00+07:00"), {}) + + def test_naive_timestamp_rejected(self): + with self.assertRaises(SiamchartVintageError): + self.store.persist(self._snap("2026-02-10T00:00:00")) + + def test_duplicate_persist_is_idempotent(self): + r1 = self.store.persist(self._snap("2026-01-10T00:00:00+07:00")) + r2 = self.store.persist(self._snap("2026-01-10T00:00:00+07:00")) + self.assertFalse(r1.get("existing")) + self.assertTrue(r2.get("existing")) + self.assertEqual(len(self.store.list_ids()), 1) + + def test_tampered_manifest_breaks_chain(self): + self.store.persist(self._snap("2026-01-10T00:00:00+07:00", yield_pct=5.0)) + self.store.persist(self._snap("2026-02-10T00:00:00+07:00", yield_pct=9.0)) + # tamper: change the first entry's retrieved_at without re-hashing + mp = self.store.manifest_path + manifest = __import__("json").loads(mp.read_text(encoding="utf-8")) + first = manifest["snapshots"]["1"] + manifest["snapshots"]["1"] = dict(first, retrieved_at="2026-07-01T00:00:00+07:00") + mp.write_text(__import__("json").dumps(manifest, ensure_ascii=False, sort_keys=True), encoding="utf-8") + with self.assertRaises(SiamchartVintageError): + self.store.snapshot_at("2026-07-01T00:00:00+07:00") + + +if __name__ == "__main__": + unittest.main()