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

[verified] PIT factor store + partial PIT score provider (PIT enabler)

Browse Source 
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.
This commit is contained in:
Kunthawat Greethong
2026-08-27 09:26:12 +07:00
parent 1b971ac72d
commit 1f630be2b5
7 changed files with 759 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
backend/app/__init__.py
View File

@@ -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", [])

26
backend/app/backtest.py
View File

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

197
backend/app/factor_vintages.py Normal file
View File

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

291
backend/app/pit_scorer.py Normal file
View File

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

34
backend/tests/test_backtest.py
View File

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

98
backend/tests/test_factor_vintages.py Normal file
View File

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

87
backend/tests/test_pit_scorer.py Normal file
View File

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