kunthawat/set50-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[verified] PIT siamchart vintage store un-partials the fundamental dimension

Browse Source 
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.
This commit is contained in:
Kunthawat Greethong
2026-08-27 11:55:36 +07:00
parent 377771be65
commit 887e9c9208
5 changed files with 330 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
backend/app/__init__.py
View File

@@ -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,

40
backend/app/pit_scorer.py
View File

@@ -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

180
backend/app/siamchart_vintages.py Normal file
View File

@@ -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)

26
backend/tests/test_pit_scorer.py
View File

@@ -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()

74
backend/tests/test_siamchart_vintages.py Normal file
View File

@@ -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()