From 1f630be2b5f896fb5ae3c29fe22ec437b58855ea Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Thu, 27 Aug 2026 09:26:12 +0700 Subject: [PATCH] [verified] PIT factor store + partial PIT score provider (PIT enabler) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a point-in-time (PIT) factor/data store and a score provider so the backtest engine can rebuild per-symbol scores from data actually knowable at a given date, instead of silently reusing the live board: - backend/app/factor_vintages.py: append-only, provenance-complete store (observed_at/released_at/retrieved_at) with a SHA-256 canonical hash chain. value_at(as_of) only ever returns rows whose released_at <= as_of (real, testable anti-look-ahead); no value by as_of fails closed (returns None). - backend/app/pit_scorer.py: PitScoreProvider computes theme surprises from PIT factor values only, and a partial siamchart fundamental view (EPS growth from the 5-year series; current ratios marked partial). score_board attaches pit_meta so callers can tell PIT from fallback. - backend/app/backtest.py: _resolve_scores now sets leakage_guard ONLY when the supplied score_fn's meta asserts pit_meta.pit=true; an arbitrary callable with no PIT proof is no longer treated as PIT (closes the 'supplied fn => PIT' hole). - backend/app/__init__.py: /api/v1/backtest accepts use_pit, wiring the PIT provider; _load_siamchart_snapshot loads the SET50 fundamental snapshot. - tests: factor store (9), pit scorer (5), backtest leakage-guard gating (2 new + 1 corrected) — full backend suite 255 passed. Empty store fail-closes (leakage_guard=false) as proven by a live route probe. Honest scope: theme dimension is PIT from this store forward; siamchart fundamental remains partial (current ratios) and is flagged as such. No historical factor data before today exists, so pre-today backtests remain non-PIT by construction. --- backend/app/__init__.py | 33 ++- backend/app/backtest.py | 26 ++- backend/app/factor_vintages.py | 197 +++++++++++++++++ backend/app/pit_scorer.py | 291 ++++++++++++++++++++++++++ backend/tests/test_backtest.py | 34 ++- backend/tests/test_factor_vintages.py | 98 +++++++++ backend/tests/test_pit_scorer.py | 87 ++++++++ 7 files changed, 759 insertions(+), 7 deletions(-) create mode 100644 backend/app/factor_vintages.py create mode 100644 backend/app/pit_scorer.py create mode 100644 backend/tests/test_factor_vintages.py create mode 100644 backend/tests/test_pit_scorer.py diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 0290e7f..3c9e6f0 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -35,6 +35,25 @@ def _load_default_snapshot() -> dict[str, Any]: return json.loads(fixture_path.read_text(encoding="utf-8")) +def _load_siamchart_snapshot() -> dict[str, Any]: + """Load the latest Siamchart SET50 fundamental snapshot (raw dict with + ``rows`` + ``details`` + ``retrieved_at``) or {} if absent/malformed. + + Used by the PIT scorer: its EPS 5-year series and per-symbol ratios feed + the fundamental (40%) dimension. Absent snapshot -> {} (the PIT provider + then falls back to the current board for fundamental, flagged partial).""" + path = Path(__file__).resolve().parents[1] / "data" / "siamchart" / "set50_master.json" + if not path.is_file(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + if not isinstance(data, dict): + return {} + return data + + def _signal_summary(result: dict[str, Any]) -> dict[str, int]: signals = result["signals"] return { @@ -706,8 +725,20 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: end = body.get("end") or "2026-06-01" capital = float(body.get("capital") or 1_000_000) freq = body.get("freq") or "monthly" + use_pit = bool(body.get("use_pit")) try: - res = run_backtest(start, end, capital=capital, rebalance_freq=freq) + if use_pit: + from pathlib import Path as _Path + from .factor_vintages import FactorVintageStore + 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()) + score_fn = make_pit_score_fn(provider) + res = run_backtest(start, end, capital=capital, + rebalance_freq=freq, score_fn=score_fn) + else: + res = run_backtest(start, end, capital=capital, rebalance_freq=freq) except BacktestError as exc: return jsonify({"error": str(exc)}), 400 runs = app.extensions.setdefault("backtest_runs", []) diff --git a/backend/app/backtest.py b/backend/app/backtest.py index 161b36f..2e213ca 100644 --- a/backend/app/backtest.py +++ b/backend/app/backtest.py @@ -125,13 +125,31 @@ def _rebalance_dates(start: str, end: str, freq: str = "monthly") -> list[str]: def _resolve_scores(score_fn, syms: list[str], as_of: Optional[str]) -> tuple[dict, bool]: - """Return (score_by_symbol, is_pit). A supplied score_fn marks leakage_guard; - the default (None) uses the current board -> non-PIT.""" + """Return (score_by_symbol, is_pit). + + A supplied score_fn marks ``leakage_guard`` ONLY when the returned scores + carry a ``pit_meta`` proving they were built point-in-time: + - ``pit_meta = {"pit": true, ...}`` -> leakage_guard = True + - ``pit_meta`` present but ``pit=false`` (blocked/partial/fallback) -> False + - no ``pit_meta`` at all (an arbitrary caller-provided fn) -> False + + This replaces the old behaviour that set leakage_guard=True for ANY supplied + callable, which could not distinguish a genuine PIT scorer from one that + silently reused the current board. + """ if score_fn is None: from .dashboard import default_scores return default_scores(syms) or {}, False - out = score_fn(syms, as_of) - return out or {}, True + out = score_fn(syms, as_of) or {} + if not out: + return out, False + # is_pit: the scores themselves assert PIT integrity via pit_meta. + any_pit = any( + isinstance(m, dict) and isinstance(m.get("pit_meta"), dict) + and bool(m.get("pit_meta", {}).get("pit")) + for m in out.values() + ) + return out, any_pit def _candidates_at(series: dict, syms: list[str], date: dt.date, diff --git a/backend/app/factor_vintages.py b/backend/app/factor_vintages.py new file mode 100644 index 0000000..25a49bb --- /dev/null +++ b/backend/app/factor_vintages.py @@ -0,0 +1,197 @@ +"""Point-in-time (PIT) factor value store with provenance (PIT enabler). + +Background (honest scope): the existing ``factor_history`` store records each +factor's *current* value once per scheduler run as ``{ts, value, as_of}``. It +records from "now on" only — it cannot reconstruct the past it never captured. +This module adds a stricter, provenance-complete store so that *from this point +forward* every recorded factor value carries the timestamps needed to answer +"what was known as of date T" without look-ahead: + + observed_at when the underlying measure actually occurred (e.g. the + end of the reported month / quarter). + released_at when the value first became knowable to the public (the + reporting lag; this is the field a PIT scorer must gate on). + retrieved_at when this backend collected it (what a point-in-time audit + can verify). + +A PIT scorer must only ever read values whose ``released_at <= as_of``. The +only way to guarantee that is to persist these timestamps at collection time. +No historical backfill is fabricated: rows only exist for what this store has +actually recorded, so any date before the first recording simply has no PIT +value for that factor (a scorer must then fail closed, not assume). + +Integrity: each per-factor JSONL row is a canonical-JSON SHA-256 hash of the +previous row's hash, so tampering or reordering is detectable (same approach +as the price observation store). +""" + +from __future__ import annotations + +import hashlib +import json +import math +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Optional + +_NORMALIZED_HASH_ALGORITHM = "sha256-json-canonical-v1" + + +class FactorVintageError(ValueError): + """Raised when a PIT factor value 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 FactorVintageError("timestamp must be ISO-8601 with timezone") from exc + if parsed.tzinfo is None: + raise FactorVintageError("timestamp must include a timezone (was naive)") + return parsed + + +def _canonical_row(row: dict[str, Any]) -> str: + """Stable canonical JSON encoding for hashing / chaining.""" + return json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _row_hash(row: dict[str, Any]) -> str: + """Canonical SHA-256 of a row EXCLUDING the self-referential ``row_hash`` + field. Both writer and verifier must use the same excludes, otherwise the + stored hash (computed without the field) never matches the re-computed hash + (with the field present).""" + payload = {k: v for k, v in row.items() if k != "row_hash"} + return hashlib.sha256(_canonical_row(payload).encode("utf-8")).hexdigest() + + +def _safe_factor_path(factor_key: str) -> str: + safe = "".join(c if (c.isalnum() or c in "._-") else "_" for c in factor_key) + if not safe: + raise FactorVintageError("invalid factor_key") + return f"{safe}.jsonl" + + +class FactorVintageStore: + """Append-only, provenance-complete, hash-chained PIT factor store.""" + + def __init__(self, root: Path | str) -> None: + self.root = Path(root).resolve() + self.dir = self.root / "factor_vintages" + self.dir.mkdir(parents=True, exist_ok=True) + + # -- internal -------------------------------------------------------- + def _path(self, factor_key: str) -> Path: + return self.dir / _safe_factor_path(factor_key) + + def _rows(self, factor_key: str) -> list[dict[str, Any]]: + path = self._path(factor_key) + if not path.is_file(): + return [] + rows: list[dict[str, Any]] = [] + prev_hash = "" + try: + with path.open(encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + rec = json.loads(line) + except json.JSONDecodeError as exc: + raise FactorVintageError("factor vintage unreadable") from exc + if not isinstance(rec, dict): + raise FactorVintageError("factor vintage row is not an object") + if rec.get("prev_hash") != prev_hash: + raise FactorVintageError("factor vintage hash chain broken") + if _row_hash(rec) != rec.get("row_hash"): + raise FactorVintageError("factor vintage row hash mismatch") + rows.append(rec) + prev_hash = rec.get("row_hash", "") + except OSError as exc: + raise FactorVintageError(f"cannot read factor vintage: {exc}") from exc + return rows + + def _last(self, factor_key: str) -> Optional[dict[str, Any]]: + rows = self._rows(factor_key) + return rows[-1] if rows else None + + # -- write ----------------------------------------------------------- + def record( + self, + factor_key: str, + value: Optional[float], + *, + observed_at: str, + released_at: str, + retrieved_at: Optional[str] = None, + source: str = "", + ) -> bool: + """Append one PIT observation. Returns True if a new row was written. + + Raised (never silently dropped) when the timestamps are invalid or the + value is non-finite, so a caller cannot accidentally persist something + that would later be interpreted as a valid PIT observation. + """ + if value is None: + return False + try: + value = float(value) + except (TypeError, ValueError) as exc: + raise FactorVintageError("factor value must be numeric") from exc + if not math.isfinite(value): + raise FactorVintageError("factor value must be finite") + observed = _parse_timestamp(observed_at) + released = _parse_timestamp(released_at) + if released < observed: + raise FactorVintageError("released_at cannot precede observed_at") + retrieved = _parse_timestamp(retrieved_at) if retrieved_at else datetime.now(timezone.utc) + if retrieved < released: + raise FactorVintageError("retrieved_at cannot precede released_at") + + last = self._last(factor_key) + prev_hash = last.get("row_hash", "") if last else "" + row: dict[str, Any] = { + "factor_key": factor_key, + "value": value, + "observed_at": observed.isoformat(timespec="seconds"), + "released_at": released.isoformat(timespec="seconds"), + "retrieved_at": retrieved.isoformat(timespec="seconds"), + "source": source, + "hash_algorithm": _NORMALIZED_HASH_ALGORITHM, + "prev_hash": prev_hash, + } + row["row_hash"] = _row_hash(row) + path = self._path(factor_key) + try: + # single append line is atomic on local FS for a small write + with path.open("a", encoding="utf-8") as fh: + fh.write(_canonical_row(row) + "\n") + except OSError as exc: + raise FactorVintageError(f"cannot write factor vintage: {exc}") from exc + return True + + # -- PIT reads -------------------------------------------------------- + def value_at(self, factor_key: str, as_of: str) -> Optional[float]: + """Most recent value whose ``released_at <= as_of``, else None. + + This is the only read a PIT scorer should use. Any row released after + ``as_of`` is invisible — that is the anti-look-ahead guarantee. + """ + cutoff = _parse_timestamp(as_of) + best: Optional[float] = None + best_released: Optional[datetime] = None + for row in self._rows(factor_key): + released = _parse_timestamp(row["released_at"]) + if released <= cutoff and (best_released is None or released > best_released): + best = row["value"] + best_released = released + return best + + def have_pit_value(self, factor_key: str, as_of: str) -> bool: + """True if the store has a released value for ``factor_key`` by ``as_of``.""" + return self.value_at(factor_key, as_of) is not None + + def series(self, factor_key: str) -> list[dict[str, Any]]: + """Full provenance row series for a factor (for audit / learning).""" + return list(self._rows(factor_key)) diff --git a/backend/app/pit_scorer.py b/backend/app/pit_scorer.py new file mode 100644 index 0000000..84c2f15 --- /dev/null +++ b/backend/app/pit_scorer.py @@ -0,0 +1,291 @@ +"""Point-in-time (PIT) factor + fundamental score provider. + +Goal: give the backtest/simulation engines a ``score_at(as_of)`` that rebuilds +per-symbol scores from data that was actually knowable at ``as_of`` — instead of +silently reusing the live current board (which leaks the future backward). + +Honest scope (this is a *partial* PIT provider, not full PIT): + - **Theme/factor dimension (60%)** is read through ``FactorVintageStore``. + Each factor value is recorded with ``observed_at`` / ``released_at`` / + ``retrieved_at`` and only a row whose ``released_at <= as_of`` is visible. + This gives a real, testable anti-look-ahead guarantee. + - **Siamchart fundamental (40%)**: the snapshot carries a per-symbol EPS + series across ~5 years ("1" = oldest .. "5" = latest). EPS *growth* for a + past period can be derived from that series without peeking at later + revisions, but the current snapshot does NOT carry a dated vintage chain and + the ratios (Yield %, PE, P/BV, ROE, DPS) are current-only. So the + fundamental dimension is reported as ``partial_pit=True`` until a dated + siamchart vintage store exists — the provider never claims fully-PIT. + +A caller that demands full PIT must not treat ``partial_pit=True`` results as +validated PIT backtests. ``leakage_guard`` is only honoured when every factor a +symbol depends on had a released value by ``as_of``; otherwise that factor +contributes nothing and the provider reports ``blocked``. +""" + +from __future__ import annotations + +import datetime as dt +from typing import Any, Optional + +from .factor_vintages import FactorVintageStore + + +class PitScoreError(ValueError): + """Raised on invalid as_of / inconsistent inputs.""" + + +def _parse_date(value: str) -> dt.date: + try: + return dt.date.fromisoformat(value[:10]) + except (ValueError, TypeError) as exc: + raise PitScoreError(f"invalid as_of: {value!r}") from exc + + +def _eps_growth_from_series(eps_series: dict) -> Optional[float]: + """EPS YoY growth from a 5-year EPS series, latest vs prior. + + ``eps_series`` maps year-key ("1" oldest .. "5" latest) to a numeric EPS. + Growth uses the two most recent non-None periods, mirroring + ``siamchart_factors.build_factor_view`` so the PIT variant is consistent + with the live board. + """ + vals = [eps_series[k] for k in sorted(eps_series) if eps_series.get(k) is not None] + if len(vals) < 2 or not vals[-2]: + return None + return round((vals[-1] - vals[-2]) / abs(vals[-2]) * 100.0, 2) + + +class PitScoreProvider: + """Rebuild per-symbol scores as known at ``as_of`` from PIT stores.""" + + def __init__( + self, + factor_store: FactorVintageStore, + siamchart_snapshot: Optional[dict] = None, + *, + theme_factor_map: Optional[dict[str, list[dict]]] = None, + ) -> None: + self.factor_store = factor_store + self.siamchart_snapshot = siamchart_snapshot + # theme_key -> list of {key, weight}; defaults to the registry THEMES. + self.theme_factor_map = theme_factor_map or _default_theme_factor_map() + + # -- per-factor PIT value -------------------------------------------- + def factor_at(self, factor_key: str, as_of: str) -> Optional[float]: + """PIT value of a factor as released by ``as_of`` (or None = blocked).""" + return self.factor_store.value_at(factor_key, as_of) + + # -- fundamental PIT (partial) --------------------------------------- + def siamchart_factor_view(self, as_of: str) -> dict[str, dict]: + """Per-symbol fundamental dict at ``as_of`` (partial PIT). + + 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. + """ + out: dict[str, dict] = {} + snap = self.siamchart_snapshot + if not snap: + return out + details = snap.get("details", {}) + for row in snap.get("rows", []): + symbol = row.get("symbol") + if not symbol: + continue + ratios = (details.get(symbol) or {}).get("ratios", {}) + yield_ = _as_float(ratios.get("Yield %") or ratios.get("Yield")) + essential = { + "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 + } + out[symbol] = essential + return out + + # -- theme PIT score ------------------------------------------------- + def theme_surprise_report(self, theme_key: str, as_of: str) -> dict[str, Any]: + """Weighted-average theme surprise from PIT factor values only. + + Returns {theme, surprise, blocked, partial_pit}. ``blocked`` is True + when at least one of the theme's factors had no value released by + ``as_of`` — a caller must never treat a blocked theme as PIT. + """ + factors = self.theme_factor_map.get(theme_key, []) + if not factors: + return {"theme": theme_key, "surprise": None, "blocked": True, "partial_pit": True} + from . import factors as factors_mod + weighted = 0.0 + w_sum = 0.0 + blocked = False + for spec in factors: + fkey = spec.get("key") + if not isinstance(fkey, str) or not fkey: + continue + fact = factors_mod.FACTORS.get(fkey) + if not fact: + continue + value = self.factor_at(fkey, as_of) + if value is None: + blocked = True + continue + norm = factors_mod.normalize( + value, sign=fact.get("sign", 1), + center=fact.get("center", 0.0), span=fact.get("span", 10.0), + ) + if norm is None: + continue + weighted += spec.get("weight", 1.0) * norm + w_sum += abs(spec.get("weight", 1.0)) + surprise: Optional[float] + if w_sum == 0: + surprise = None + blocked = True + else: + surprise = round(min(1.0, max(-1.0, weighted / w_sum)), 3) + return { + "theme": theme_key, + "surprise": surprise, + "blocked": blocked, + "partial_pit": blocked, + } + + # -- per-symbol combined board as of a date ---------------------------- + def score_board(self, as_of: str, momentum_series: Optional[dict] = None) -> dict: + """Per-symbol {combined, is_dividend, dividend_yield, pit_meta} as of + ``as_of``, rebuilt from PIT stores. + + ``pit_meta`` = {"pit": bool, "partial_pit": bool, "blocked_theme": [...], + "note": "..."}. ``pit`` is True only when NO theme this provider serves + was blocked by missing PIT releases (all requested factors were known by + ``as_of``). Because the siamchart fundamental dimension is only partial + (current ratios), ``partial_pit`` is kept True to be honest — a caller + should not label this a fully-PIT backtest. + + Reuses the canonical ``combine_score`` / ``THEME_SYMBOLS`` / + ``quality_within_theme`` from themes.py so the PIT board is consistent + with the live dashboard (same 0.6/0.4 weighting). + """ + from . import themes as themes_mod + from .dashboard import default_scores + + theme_scores: dict[str, dict[str, float]] = {} + blocked_themes: list[str] = [] + fv = self.siamchart_factor_view(as_of) + # reverse-map symbol -> fundamental for quality_within_theme's factor view + fv_for_quality = { + "available": bool(fv), "as_of": as_of, "source": "siamchart", + "factors": [dict(f) | {"symbol": s} for s, f in fv.items()], + } + momentum = momentum_series or {} + siamchart_score = themes_mod.build_siamchart_score(fv_for_quality, momentum=momentum) + for tid, spec in self.theme_factor_map.items(): + rep = self.theme_surprise_report(tid, as_of) + if rep["blocked"] or rep["surprise"] is None: + blocked_themes.append(tid) + theme_scores[tid] = {} + continue + q = {} + for sym in themes_mod.THEME_SYMBOLS.get(tid, set()): + if sym not in fv: + continue + quality = themes_mod.quality_within_theme(sym, tid, fv_for_quality) + q[sym] = float(rep["surprise"]) * quality + theme_scores[tid] = q + + combined = themes_mod.combine_score(list(theme_scores.values()), siamchart_score) + is_full_pit = not blocked_themes + out: dict[str, dict] = {} + for sym, meta in combined.items(): + fm = fv.get(sym, {}) + out[sym] = { + "combined": round(meta.get("combined", 0.0), 3), + "theme_score": round(meta.get("theme_score", 0.0), 3), + "siamchart_score": round(meta.get("siamchart_score", 0.0), 3), + "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 + "blocked_theme": blocked_themes, + "note": "theme surprises PIT; siamchart fundamental partial (current ratios)", + }, + } + # fall back to the current board for any symbol the PIT path could not + # cover, but flag it so it is never mistaken for PIT. + default = default_scores() or {} + for sym, meta in default.items(): + if sym not in out: + out[sym] = { + "combined": meta.get("combined", 0.0), + "is_dividend": meta.get("is_dividend", False), + "dividend_yield": meta.get("dividend_yield") or 0.0, + "pit_meta": {"pit": False, "partial_pit": True, + "blocked_theme": blocked_themes, + "note": "current-board fallback (non-PIT)"}, + } + return out + + +def _as_float(value: Any) -> Optional[float]: + if value is None or value == "": + return None + try: + return float(str(value).replace(",", "")) + except (ValueError, TypeError): + return None + + +def _normalize_as_of(as_of: Optional[str]) -> str: + """Coerce an ``as_of`` into a tz-aware ISO timestamp. + + The backtest engine passes date-only strings (e.g. ``"2026-01-01"``) as the + rebalance window; a date-only as_of means "end of that day" here. The PIT + store rejects naive timestamps (an integrity invariant), so we attach the + platform timezone (+07:00, Bangkok) at midnight rather than dropping the + guard. Full tz-aware ISO timestamps are passed through untouched. + """ + if as_of is None: + return _now_iso() + s = str(as_of).strip() + # tz-aware already? (offset +HH:MM / -HH:MM, or trailing Z) + has_tz = "Z" in s or ("T" in s and ("+" in s or "-" in s[10:])) + if has_tz: + return s + # bare YYYY-MM-DD or naive datetime -> midnight +07:00 (Bangkok) + return s[:10] + "T00:00:00+07:00" + + +def make_pit_score_fn(provider: PitScoreProvider, momentum_series: Optional[dict] = None): + """Return a ``ScoreFn(symbols, as_of)`` bound to `provider`. + + This is the adapter that lets ``run_backtest`` consume the PIT provider as a + drop-in ``score_fn``. Each call re-resolves the board as of ``as_of`` (a + date-only as_of is normalized to midnight +07:00, so a rebalance date means + "end of that day"), so a multi-rebalance backtest gets fresh point-in-time + scores at every window. The returned per-symbol dicts include ``pit_meta`` + so ``_resolve_scores`` can truthfully set ``leakage_guard``. + """ + def pit_score_fn(symbols, as_of=None) -> dict: + board = provider.score_board( + as_of=_normalize_as_of(as_of), momentum_series=momentum_series) + if not symbols: + return board + return {s: board.get(s, {}) for s in symbols if s in board} + return pit_score_fn + + +def _now_iso() -> str: + import datetime as _dt + return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds") + + +def _default_theme_factor_map() -> dict[str, list[dict]]: + from .themes import THEMES + return { + tid: [dict(spec) for spec in (tdef.get("factors", []) or [])] + for tid, tdef in THEMES.items() + } diff --git a/backend/tests/test_backtest.py b/backend/tests/test_backtest.py index 743c7a6..6bb93d9 100644 --- a/backend/tests/test_backtest.py +++ b/backend/tests/test_backtest.py @@ -79,8 +79,10 @@ class RunBacktestTest(unittest.TestCase): # Real re-allocations happened (price data exists for every window). self.assertEqual(res.rebalances, 6) self.assertGreater(res.trades, 0) - # Supplying a score_fn -> leakage_guard True (PIT contract). - self.assertTrue(res.leakage_guard) + # Supplying a score_fn WITHOUT pit_meta is NOT PIT: the engine can no + # longer trust an arbitrary callable. Only scores that assert + # pit_meta.pit=True set leakage_guard (see the two tests below). + self.assertFalse(res.leakage_guard) self.assertAlmostEqual( res.final_value, res.capital + res.price_pnl + res.dividend_income, @@ -159,6 +161,34 @@ class RunBacktestTest(unittest.TestCase): res.capital + res.price_pnl + res.dividend_income, ) + @patch("app.backtest.load_price_snapshot", return_value=_fake_series()) + def test_supplied_fn_without_pit_meta_is_not_pit(self, _load): + # A supplied score_fn that does NOT assert PIT integrity via pit_meta + # must NOT set leakage_guard (the old behaviour trusted any callable). + def naive(syms, as_of=None): + return {s: {"combined": 0.5, "is_dividend": True, "dividend_yield": 5.0} + for s in syms} + res = backtest.run_backtest( + "2026-01-01", "2026-03-01", capital=100_000, + rebalance_freq="monthly", score_fn=naive, symbols=["A", "B"], + ) + self.assertIs(res.leakage_guard, False) + + @patch("app.backtest.load_price_snapshot", return_value=_fake_series()) + def test_supplied_fn_with_pit_meta_sets_leakage_guard(self, _load): + # Only a score_fn whose scores assert pit_meta.pit=True may set guard. + def pit(syms, as_of=None): + return { + s: {"combined": 0.5, "is_dividend": True, "dividend_yield": 5.0, + "pit_meta": {"pit": True, "partial_pit": True, "note": "pit"}} + for s in syms + } + res = backtest.run_backtest( + "2026-01-01", "2026-03-01", capital=100_000, + rebalance_freq="monthly", score_fn=pit, symbols=["A", "B"], + ) + self.assertIs(res.leakage_guard, True) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_factor_vintages.py b/backend/tests/test_factor_vintages.py new file mode 100644 index 0000000..ed53a93 --- /dev/null +++ b/backend/tests/test_factor_vintages.py @@ -0,0 +1,98 @@ +"""Tests for the point-in-time factor vintage store (PIT enabler).""" + +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path + +from app.factor_vintages import FactorVintageError, FactorVintageStore + + +class FactorVintageStoreTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + self.store = FactorVintageStore(self.dir) + + def tearDown(self): + self._tmp.cleanup() + + # -- anti-look-ahead: the core PIT guarantee -------------------------- + def test_value_at_does_not_see_future_releases(self): + # a value released in June must be invisible to any as_of before June + self.assertTrue(self.store.record( + "macro_consumption", 4.9, + observed_at="2026-01-31T00:00:00+07:00", + released_at="2026-02-15T09:00:00+07:00", + )) + # the leak/decoy value released AFTER the as_of we will query + self.assertTrue(self.store.record( + "macro_consumption", 99.0, # decoy — must NOT be seen at 2026-03-01 + observed_at="2026-05-31T00:00:00+07:00", + released_at="2026-06-15T09:00:00+07:00", + )) + self.assertEqual(self.store.value_at("macro_consumption", "2026-03-01T00:00:00+07:00"), 4.9) + # the store DOES see it once as_of passes its release + self.assertEqual(self.store.value_at("macro_consumption", "2026-07-01T00:00:00+07:00"), 99.0) + + def test_value_at_returns_latest_released_before_as_of(self): + self.store.record("f", 1.0, observed_at="2026-01-01T00:00:00+07:00", released_at="2026-01-15T00:00:00+07:00") + self.store.record("f", 2.0, observed_at="2026-02-01T00:00:00+07:00", released_at="2026-02-15T00:00:00+07:00") + self.assertEqual(self.store.value_at("f", "2026-01-31T00:00:00+07:00"), 1.0) + self.assertEqual(self.store.value_at("f", "2026-03-01T00:00:00+07:00"), 2.0) + + # -- fail closed when nothing released by as_of ----------------------- + def test_value_at_none_and_fail_closed_before_first_release(self): + self.store.record("f", 5.0, observed_at="2026-01-01T00:00:00+07:00", released_at="2026-02-01T00:00:00+07:00") + self.assertIsNone(self.store.value_at("f", "2026-01-15T00:00:00+07:00")) + self.assertFalse(self.store.have_pit_value("f", "2026-01-15T00:00:00+07:00")) + # after release it is available + self.assertTrue(self.store.have_pit_value("f", "2026-02-15T00:00:00+07:00")) + + # -- provenance invariants -------------------------------------------- + def test_released_cannot_precede_observed(self): + with self.assertRaises(FactorVintageError): + self.store.record("f", 1.0, observed_at="2026-02-01T00:00:00+07:00", released_at="2026-01-15T00:00:00+07:00") + + def test_retrieved_cannot_precede_released(self): + with self.assertRaises(FactorVintageError): + self.store.record( + "f", 1.0, + observed_at="2026-01-01T00:00:00+07:00", + released_at="2026-01-15T00:00:00+07:00", + retrieved_at="2026-01-10T00:00:00+07:00", + ) + + def test_non_finite_value_raises_not_silently_dropped(self): + with self.assertRaises(FactorVintageError): + self.store.record("f", float("nan"), observed_at="2026-01-01T00:00:00+07:00", released_at="2026-01-15T00:00:00+07:00") + + def test_naive_timestamp_rejected(self): + with self.assertRaises(FactorVintageError): + self.store.record("f", 1.0, observed_at="2026-01-01T00:00:00", released_at="2026-01-15T00:00:00+07:00") + + # -- integrity: hash chain -------------------------------------------- + def test_tampering_with_prior_row_breaks_chain(self): + self.store.record("f", 1.0, observed_at="2026-01-01T00:00:00+07:00", released_at="2026-01-15T00:00:00+07:00") + self.store.record("f", 2.0, observed_at="2026-02-01T00:00:00+07:00", released_at="2026-02-15T00:00:00+07:00") + path = self.store.dir / "f.jsonl" + # rewrite the first row's value without re-hashing (tamper) + lines = path.read_text(encoding="utf-8").splitlines() + first = json.loads(lines[0]) + first["value"] = 999.0 + lines[0] = json.dumps(first, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + with self.assertRaises(FactorVintageError): + self.store.series("f") + + def test_append_keeps_chain_intact_and_series_ordered(self): + self.store.record("f", 1.0, observed_at="2026-01-01T00:00:00+07:00", released_at="2026-01-15T00:00:00+07:00") + self.store.record("f", 2.0, observed_at="2026-02-01T00:00:00+07:00", released_at="2026-02-15T00:00:00+07:00") + series = self.store.series("f") + self.assertEqual([r["value"] for r in series], [1.0, 2.0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_pit_scorer.py b/backend/tests/test_pit_scorer.py new file mode 100644 index 0000000..256d794 --- /dev/null +++ b/backend/tests/test_pit_scorer.py @@ -0,0 +1,87 @@ +"""Tests for the partial point-in-time score provider (PIT enabler).""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from app.factor_vintages import FactorVintageStore +from app.pit_scorer import PitScoreProvider, _eps_growth_from_series + + +def _make_store(): + tmp = tempfile.TemporaryDirectory() + return FactorVintageStore(Path(tmp.name)), tmp + + +class EpsGrowthTest(unittest.TestCase): + def test_growth_from_five_year_series(self): + # 1.0 -> 1.1 -> 1.2 -> 1.3 -> 1.4 : latest growth = (1.4-1.3)/1.3 + self.assertAlmostEqual( + _eps_growth_from_series({"1": 1.0, "2": 1.1, "3": 1.2, "4": 1.3, "5": 1.4}), + round((1.4 - 1.3) / 1.3 * 100, 2), + ) + + def test_growth_none_when_fewer_than_two(self): + self.assertIsNone(_eps_growth_from_series({"1": 1.0})) + self.assertIsNone(_eps_growth_from_series({})) + + +class PitScoreProviderTest(unittest.TestCase): + def setUp(self): + self.store, self._tmp = _make_store() + self.addCleanup(self._tmp.cleanup) + # tourism theme uses tourism_arrivals_ytd (BOT, monthly) in the real + # registry; build a tiny explicit map so the test is self-contained. + self.provider = PitScoreProvider( + self.store, + siamchart_snapshot=None, + theme_factor_map={"tourism": [{"key": "tourism_arrivals_ytd", "weight": 1.0}]}, + ) + + def test_theme_surprise_uses_only_released_before_as_of(self): + # real value released Feb + self.store.record( + "tourism_arrivals_ytd", 16.2, + observed_at="2026-01-31T00:00:00+07:00", + released_at="2026-02-15T09:00:00+07:00", + ) + # decoy released June — must be invisible at 2026-03-01 + self.store.record( + "tourism_arrivals_ytd", 999.0, + observed_at="2026-05-31T00:00:00+07:00", + released_at="2026-06-15T09:00:00+07:00", + ) + at_march = self.provider.theme_surprise_report("tourism", "2026-03-01T00:00:00+07:00") + self.assertFalse(at_march["blocked"]) + # normalize(16.2, center=20, span=15) = (16.2-20)/15 = -0.2533 (sign +1) + self.assertAlmostEqual(at_march["surprise"], round((16.2 - 20.0) / 15.0, 4), places=3) + + def test_theme_blocked_when_no_value_released_by_as_of(self): + self.store.record( + "tourism_arrivals_ytd", 16.2, + observed_at="2026-01-31T00:00:00+07:00", + released_at="2026-02-15T09:00:00+07:00", + ) + report = self.provider.theme_surprise_report("tourism", "2026-01-15T00:00:00+07:00") + self.assertTrue(report["blocked"]) + self.assertIsNone(report["surprise"]) + + def test_siamchart_partial_grade_and_eps_growth(self): + snap = { + "retrieved_at": "2026-08-25T01:41:43Z", + "rows": [{"symbol": "AOT", "eps": {"1": 1.0, "2": 1.1, "3": 1.2, "4": 1.3, "5": 1.4}}], + "details": {"AOT": {"ratios": {"Yield %": 1.21, "PE": 51.25}}}, + } + 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.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"]) + + +if __name__ == "__main__": + unittest.main()