887e9c9208
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.
312 lines
14 KiB
Python
312 lines
14 KiB
Python
"""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,
|
|
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()
|
|
|
|
# -- 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 _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``.
|
|
|
|
Returns {symbol: {eps_growth_yoy, dividend_yield, is_dividend,
|
|
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._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")
|
|
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": pit_grade,
|
|
}
|
|
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
|
|
# 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, {})
|
|
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 and fundamental_pit,
|
|
"partial_pit": not fundamental_pit,
|
|
"blocked_theme": blocked_themes,
|
|
"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
|
|
# 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()
|
|
}
|