[verified] Task 1: strict PIT backtest readiness + default-date derivation
This commit is contained in:
Kunthawat Greethong
@@ -740,6 +740,37 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
|
||||
)
|
||||
return jsonify(detail)
|
||||
|
||||
@app.get("/api/v1/backtest/readiness")
|
||||
def backtest_readiness_endpoint():
|
||||
"""Strict PIT backtest readiness + recommended default dates.
|
||||
|
||||
The backtest must not start before advice is genuinely available. This
|
||||
endpoint reports whether the PIT stores (factor vintages + Siamchart
|
||||
vintage manifest) plus price data cover any usable [start, end] window,
|
||||
and returns the recommended default start/end the UI should prefill.
|
||||
It fails closed (ready=false + missing list) when coverage is absent.
|
||||
"""
|
||||
from pathlib import Path as _Path
|
||||
from .backtest_readiness import evaluate_readiness
|
||||
from .factor_vintages import FactorVintageStore
|
||||
from .siamchart_vintages import SiamchartVintageStore
|
||||
from .simulation import load_price_snapshot, SimulationError
|
||||
|
||||
data_root = _Path(__file__).resolve().parents[1] / "data"
|
||||
fstore = FactorVintageStore(data_root)
|
||||
sstore = SiamchartVintageStore(data_root)
|
||||
try:
|
||||
series = load_price_snapshot()
|
||||
except SimulationError:
|
||||
series = None
|
||||
start = request.args.get("start")
|
||||
end = request.args.get("end")
|
||||
res = evaluate_readiness(
|
||||
factor_store=fstore, siamchart_store=sstore,
|
||||
price_series=series, start=start, end=end,
|
||||
)
|
||||
return jsonify(res.to_dict())
|
||||
|
||||
@app.post("/api/v1/backtest")
|
||||
def run_backtest_endpoint():
|
||||
"""Run a real backtest over [start, end] with capital; persist result."""
|
||||
|
||||
423
backend/app/backtest_readiness.py
Normal file
423
backend/app/backtest_readiness.py
Normal file
@@ -0,0 +1,423 @@
|
||||
"""Strict PIT backtest readiness + default-date derivation (Task 1).
|
||||
|
||||
The backtest must not silently start before advice is genuinely available.
|
||||
This module answers two things:
|
||||
|
||||
* ``ready`` — whether every input the backtest needs is PIT-available over
|
||||
some [start, end] window (factor releases, a Siamchart snapshot, and
|
||||
executable prices for the universe).
|
||||
* ``recommended_start`` / ``recommended_end`` — the defaults the API/UI
|
||||
should prefill. The recommended end is "yesterday" in Bangkok (the last
|
||||
complete trading availability), and the recommended start is the earliest
|
||||
date at which every PIT input is release-available *and* at least one
|
||||
executable price exists.
|
||||
|
||||
Fail-closed semantics: this module never manufactures a start date from data
|
||||
that does not exist. If any required input has no coverage through the window
|
||||
it reports ``ready=false`` and lists what is missing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Optional
|
||||
|
||||
_TZ_BANGKOK = dt.timezone(dt.timedelta(hours=7))
|
||||
|
||||
|
||||
class BacktestReadinessError(Exception):
|
||||
"""Raised when coverage cannot be evaluated safely."""
|
||||
|
||||
|
||||
def bangkok_now() -> dt.datetime:
|
||||
"""Current wall-clock in Bangkok (Asia/Bangkok, UTC+7, no DST)."""
|
||||
return dt.datetime.now(dt.timezone.utc).astimezone(_TZ_BANGKOK)
|
||||
|
||||
|
||||
def yesterday_bangkok() -> dt.date:
|
||||
"""Recommended default end: yesterday in Bangkok."""
|
||||
return (bangkok_now() - dt.timedelta(days=1)).date()
|
||||
|
||||
|
||||
def _parse_date(value: Any) -> dt.date:
|
||||
try:
|
||||
if isinstance(value, str):
|
||||
return dt.date.fromisoformat(str(value)[:10])
|
||||
if isinstance(value, dt.datetime):
|
||||
return value.date()
|
||||
if isinstance(value, dt.date):
|
||||
return value
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise BacktestReadinessError(f"invalid date: {value!r}") from exc
|
||||
raise BacktestReadinessError(f"invalid date: {value!r}")
|
||||
|
||||
|
||||
def _date_from_ts(value: Any) -> Optional[dt.date]:
|
||||
"""Extract a date from an ISO-8601 timestamp, tolerant of partial input."""
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
return _parse_date(value)
|
||||
except BacktestReadinessError:
|
||||
return None
|
||||
|
||||
|
||||
def _factor_keys_required() -> list[str]:
|
||||
"""The registry factors the PIT scorer needs (single source of truth)."""
|
||||
from .factors import FACTORS
|
||||
if isinstance(FACTORS, dict):
|
||||
return [str(k) for k in FACTORS.keys()]
|
||||
# defensive fallback (registry is a dict today; keep a tolerant path)
|
||||
keys: list[str] = []
|
||||
for item in FACTORS: # type: ignore[union-attr]
|
||||
if isinstance(item, dict) and isinstance(item.get("key"), str):
|
||||
keys.append(str(item.get("key")))
|
||||
return keys
|
||||
|
||||
|
||||
def _required_symbols() -> list[str]:
|
||||
"""The universe the backtest trades. Uses the board registry when present,
|
||||
else falls back to anything found in the price series at evaluation time."""
|
||||
try:
|
||||
from .themes import THEME_SYMBOLS
|
||||
out: list[str] = []
|
||||
if isinstance(THEME_SYMBOLS, dict):
|
||||
for v in THEME_SYMBOLS.values():
|
||||
if isinstance(v, str):
|
||||
out.append(v)
|
||||
elif isinstance(v, list):
|
||||
for s in v:
|
||||
if isinstance(s, str):
|
||||
out.append(s)
|
||||
elif isinstance(THEME_SYMBOLS, list):
|
||||
for s in THEME_SYMBOLS:
|
||||
if isinstance(s, str):
|
||||
out.append(s)
|
||||
return sorted(set(out))
|
||||
except Exception: # pragma: no cover - registry unavailable
|
||||
return []
|
||||
|
||||
|
||||
@dataclass
|
||||
class DataCoverage:
|
||||
"""Per-source earliest-available date summary for one requested window."""
|
||||
|
||||
source: str
|
||||
available: bool
|
||||
earliest: Optional[dt.date] = None
|
||||
latest: Optional[dt.date] = None
|
||||
detail: str = ""
|
||||
missing_count: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class BacktestReadiness:
|
||||
"""Strict PIT readiness verdict + recommended default dates."""
|
||||
|
||||
ready: bool
|
||||
recommended_start: Optional[str] = None
|
||||
recommended_end: Optional[str] = None
|
||||
reason: str = ""
|
||||
missing: list[str] = field(default_factory=list)
|
||||
coverage: list[DataCoverage] = field(default_factory=list)
|
||||
timezone: str = "Asia/Bangkok"
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"ready": self.ready,
|
||||
"recommended_start": self.recommended_start,
|
||||
"recommended_end": self.recommended_end,
|
||||
"reason": self.reason,
|
||||
"missing": self.missing,
|
||||
"coverage": [
|
||||
{
|
||||
"source": c.source,
|
||||
"available": c.available,
|
||||
"earliest": c.earliest.isoformat() if c.earliest else None,
|
||||
"latest": c.latest.isoformat() if c.latest else None,
|
||||
"detail": c.detail,
|
||||
"missing_count": c.missing_count,
|
||||
}
|
||||
for c in self.coverage
|
||||
],
|
||||
"timezone": self.timezone,
|
||||
}
|
||||
|
||||
|
||||
def _price_coverage(series: dict[str, Any]) -> DataCoverage:
|
||||
"""Earliest/latest trading date shared across the available symbols.
|
||||
|
||||
Uses the intersection of symbol availability so that the recommended start
|
||||
is a date every held symbol can actually be valued, not just one symbol.
|
||||
"""
|
||||
min_dates: list[dt.date] = []
|
||||
max_dates: list[dt.date] = []
|
||||
for sym, s in series.items():
|
||||
bars = (s or {}).get("bars", [])
|
||||
if not bars:
|
||||
continue
|
||||
dates = [b.get("date") for b in bars if b.get("date")]
|
||||
dates = [d for d in dates if d is not None]
|
||||
if not dates:
|
||||
continue
|
||||
parsed = [_parse_date(d) for d in dates]
|
||||
min_dates.append(min(parsed))
|
||||
max_dates.append(max(parsed))
|
||||
if not min_dates:
|
||||
return DataCoverage(
|
||||
source="price", available=False, detail="no price bars on disk",
|
||||
)
|
||||
return DataCoverage(
|
||||
source="price",
|
||||
available=True,
|
||||
earliest=max(min_dates),
|
||||
latest=min(max_dates),
|
||||
detail=f"{len(min_dates)} symbols",
|
||||
)
|
||||
|
||||
|
||||
def evaluate_readiness(
|
||||
*,
|
||||
factor_store: Any = None,
|
||||
siamchart_store: Any = None,
|
||||
price_series: Optional[dict[str, Any]] = None,
|
||||
start: Optional[str] = None,
|
||||
end: Optional[str] = None,
|
||||
) -> BacktestReadiness:
|
||||
"""Evaluate strict PIT readiness and derive recommended default dates.
|
||||
|
||||
Args:
|
||||
factor_store: a FactorVintageStore (or object exposing ``series(key)``
|
||||
and ``value_at(key, as_of)``).
|
||||
siamchart_store: a SiamchartVintageStore exposing ``list_ids()`` and
|
||||
``snapshot_at(as_of)``.
|
||||
price_series: the Yahoo price series dict ``{sym: {bars: [...]}}``.
|
||||
start/end: optional explicit window; if given, readiness is evaluated
|
||||
only against ``[start, end]``.
|
||||
|
||||
"Ready" requires, within the window:
|
||||
1. every registry factor has a released value by the start date;
|
||||
2. a Siamchart snapshot retrieved no later than the start date;
|
||||
3. at least one executable price in the window for every required symbol.
|
||||
Missing inputs are reported explicitly rather than silently skipped.
|
||||
"""
|
||||
miss: list[str] = []
|
||||
cov: list[DataCoverage] = []
|
||||
|
||||
today = yesterday_bangkok()
|
||||
# When an explicit start is requested, coverage is evaluated *at* that
|
||||
# start (the inputs must be knowable by then). Otherwise evaluate through
|
||||
# the end date to discover whether a usable window exists at all.
|
||||
if start is not None:
|
||||
coverage_cutoff = _parse_date(start)
|
||||
else:
|
||||
coverage_cutoff = _date_from_ts(end) or today
|
||||
|
||||
# ---------------- price ----------------
|
||||
if price_series is None:
|
||||
from .simulation import load_price_snapshot
|
||||
try:
|
||||
price_series = load_price_snapshot()
|
||||
except Exception as exc: # SimulationError / OSError / JSON
|
||||
price_series = {}
|
||||
cov.append(DataCoverage(
|
||||
source="price", available=False, detail=f"cannot load: {exc}",
|
||||
))
|
||||
miss.append("price")
|
||||
fac_cov, fac_miss, _earliest = _factor_coverage(
|
||||
factor_store, coverage_cutoff
|
||||
)
|
||||
cov.extend(fac_cov)
|
||||
miss.extend(fac_miss)
|
||||
return BacktestReadiness(
|
||||
ready=False, reason="price coverage missing", missing=miss,
|
||||
coverage=cov, recommended_end=today.isoformat(),
|
||||
)
|
||||
pc = _price_coverage(price_series)
|
||||
cov.append(pc)
|
||||
if not pc.available:
|
||||
miss.append("price")
|
||||
|
||||
# ---------------- factor ----------------
|
||||
fac_cov, fac_miss, fac_earliest = _factor_coverage(
|
||||
factor_store, coverage_cutoff
|
||||
)
|
||||
cov.extend(fac_cov)
|
||||
miss.extend(fac_miss)
|
||||
|
||||
# ---------------- siamchart ----------------
|
||||
scv, sc_earliest = _siamchart_coverage(siamchart_store, coverage_cutoff)
|
||||
cov.append(scv)
|
||||
if not scv.available:
|
||||
miss.append("siamchart")
|
||||
|
||||
# derive recommended start = the latest earliest-available date among the
|
||||
# inputs (the point at which *all* of them are simultaneously available).
|
||||
candidates = [
|
||||
d for d in (fac_earliest, sc_earliest, pc.earliest)
|
||||
if d is not None
|
||||
]
|
||||
recommended_end = _date_from_ts(end) or today
|
||||
if start is not None:
|
||||
recommended_start = _parse_date(start)
|
||||
elif candidates:
|
||||
recommended_start = max(candidates)
|
||||
else:
|
||||
recommended_start = None
|
||||
|
||||
# cap recommended start so it never exceeds the end
|
||||
if recommended_start and recommended_end and recommended_start > recommended_end:
|
||||
recommended_start = recommended_end
|
||||
|
||||
# explicit window requested: readiness is whether the window is covered
|
||||
if start is not None:
|
||||
s = _parse_date(start)
|
||||
e_ = _parse_date(end) if end else today
|
||||
ready = (not miss) and s <= e_
|
||||
else:
|
||||
ready = (not miss) and bool(recommended_start)
|
||||
|
||||
reason = ""
|
||||
rs = recommended_start.isoformat() if recommended_start else None
|
||||
re_iso = recommended_end.isoformat()
|
||||
if ready:
|
||||
reason = (
|
||||
f"strict PIT coverage from {rs} to {re_iso}"
|
||||
)
|
||||
elif miss:
|
||||
reason = "missing inputs: " + ", ".join(sorted(set(miss)))
|
||||
else:
|
||||
reason = "no usable PIT-ready window"
|
||||
|
||||
return BacktestReadiness(
|
||||
ready=ready,
|
||||
recommended_start=recommended_start.isoformat() if recommended_start else None,
|
||||
recommended_end=re_iso,
|
||||
reason=reason,
|
||||
missing=sorted(set(miss)),
|
||||
coverage=cov,
|
||||
)
|
||||
|
||||
|
||||
def _factor_coverage(
|
||||
store: Any, cutoff: Optional[dt.date]
|
||||
) -> tuple[list[DataCoverage], list[str], Optional[dt.date]]:
|
||||
"""Earliest released date across all registry factors, or missing list."""
|
||||
if store is None:
|
||||
return [
|
||||
DataCoverage(
|
||||
source="factor", available=False, detail="factor store not provided",
|
||||
)
|
||||
], ["factor"], None
|
||||
factors = _factor_keys_required()
|
||||
earliest_dates: list[dt.date] = []
|
||||
missing_factor: list[str] = []
|
||||
cutoff_ts = None
|
||||
if cutoff is not None:
|
||||
cutoff_ts = dt.datetime.combine(cutoff, dt.time.min, tzinfo=_TZ_BANGKOK)
|
||||
for key in factors:
|
||||
try:
|
||||
rows = store.series(key)
|
||||
except Exception:
|
||||
rows = []
|
||||
release_dates = [
|
||||
_date_from_ts(r.get("released_at")) or _date_from_ts(r.get("observed_at"))
|
||||
for r in rows
|
||||
]
|
||||
release_dates = [d for d in release_dates if d is not None]
|
||||
if cutoff_ts is not None:
|
||||
release_dates = [
|
||||
d for d in release_dates
|
||||
if _release_le(d, cutoff_ts)
|
||||
]
|
||||
if not release_dates:
|
||||
missing_factor.append(key)
|
||||
continue
|
||||
if cutoff_ts is None:
|
||||
earliest_dates.append(min(release_dates))
|
||||
else:
|
||||
earliest_dates.append(min(release_dates))
|
||||
cov = DataCoverage(
|
||||
source="factor",
|
||||
available=not missing_factor,
|
||||
earliest=min(earliest_dates) if earliest_dates else None,
|
||||
missing_count=len(missing_factor),
|
||||
detail=(
|
||||
f"{len(earliest_dates)}/{len(factors)} factors have released values"
|
||||
if earliest_dates
|
||||
else "no factor has a released value"
|
||||
),
|
||||
)
|
||||
# umbrella missing token so consumers see "factor" plus per-key detail
|
||||
missing_tokens: list[str] = ["factor"] if missing_factor else []
|
||||
missing_tokens.extend(missing_factor)
|
||||
return [cov], missing_tokens, (min(earliest_dates) if earliest_dates else None)
|
||||
|
||||
|
||||
def _release_le(d: dt.date, cutoff_dt: dt.datetime) -> bool:
|
||||
"""True if a release date is at/before the cutoff (same-day counts)."""
|
||||
day = dt.datetime.combine(d, dt.time.min, tzinfo=_TZ_BANGKOK)
|
||||
return day <= cutoff_dt
|
||||
|
||||
|
||||
def _siamchart_coverage(
|
||||
store: Any, cutoff: Optional[dt.date] = None
|
||||
) -> tuple[DataCoverage, Optional[dt.date]]:
|
||||
"""Earliest retrieved Siamchart snapshot date (or missing).
|
||||
|
||||
When ``cutoff`` is given, the snapshot must have been retrieved by that
|
||||
date (strict PIT: it must be knowable at the requested start).
|
||||
"""
|
||||
if store is None:
|
||||
return DataCoverage(
|
||||
source="siamchart", available=False, detail="siamchart store not provided"
|
||||
), None
|
||||
try:
|
||||
ids = store.list_ids()
|
||||
except Exception:
|
||||
ids = []
|
||||
if not ids:
|
||||
return DataCoverage(
|
||||
source="siamchart", available=False,
|
||||
detail="no snapshots in siamchart vintage store",
|
||||
), None
|
||||
# Collect retrieval timestamps of every stored snapshot. The real store
|
||||
# exposes its manifest; a fake may only answer snapshot_at(now) (newest).
|
||||
retrieved_dates: list[dt.date] = []
|
||||
try:
|
||||
manifest = store._load_manifest() # real SiamchartVintageStore
|
||||
for e in manifest.get("snapshots", {}).values():
|
||||
d = _date_from_ts(e.get("retrieved_at"))
|
||||
if d is not None:
|
||||
retrieved_dates.append(d)
|
||||
except Exception:
|
||||
retrieved_dates = []
|
||||
if not retrieved_dates:
|
||||
# fallback: fake / minimal store -> newest snapshot's own retrieved_at
|
||||
try:
|
||||
snap = store.snapshot_at(
|
||||
dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
|
||||
)
|
||||
d = _date_from_ts(snap.get("_retrieved_at") or snap.get("retrieved_at"))
|
||||
if d is not None:
|
||||
retrieved_dates.append(d)
|
||||
except Exception:
|
||||
pass
|
||||
if not retrieved_dates:
|
||||
return DataCoverage(
|
||||
source="siamchart", available=True, detail=f"{len(ids)} snapshots stored"
|
||||
), None
|
||||
earliest = min(retrieved_dates)
|
||||
if cutoff is not None:
|
||||
known_by_cutoff = [d for d in retrieved_dates if d <= cutoff]
|
||||
if not known_by_cutoff:
|
||||
return DataCoverage(
|
||||
source="siamchart", available=False, earliest=earliest,
|
||||
detail=f"no snapshot retrieved by {cutoff.isoformat()}",
|
||||
), earliest
|
||||
earliest = min(known_by_cutoff)
|
||||
return DataCoverage(
|
||||
source="siamchart", available=True, earliest=earliest,
|
||||
detail=f"{len(ids)} snapshots stored",
|
||||
), earliest
|
||||
195
backend/tests/test_backtest_readiness.py
Normal file
195
backend/tests/test_backtest_readiness.py
Normal file
@@ -0,0 +1,195 @@
|
||||
"""Tests for strict PIT backtest readiness + default-date derivation (Task 1).
|
||||
|
||||
Covered scenarios follow the acceptance criteria:
|
||||
|
||||
* no factor vintages -> ready=false, factor missing listed
|
||||
* no Siamchart snapshot -> ready=false, siamchart missing listed
|
||||
* price coverage starting after PIT factors -> recommended start =
|
||||
latest of the first-ready dates (strict: all inputs must be available)
|
||||
* recommended end = yesterday in Bangkok, bounded by latest price date
|
||||
* an explicit start earlier than readiness -> ready=false
|
||||
|
||||
All stores are lightweight fakes so tests stay deterministic and offline.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import unittest
|
||||
|
||||
from app.backtest_readiness import (
|
||||
BacktestReadiness,
|
||||
evaluate_readiness,
|
||||
yesterday_bangkok,
|
||||
)
|
||||
|
||||
|
||||
class FakeFactorStore:
|
||||
"""Minimal factor store exposing series()/value_at() for readiness tests."""
|
||||
|
||||
def __init__(self, releases: dict[str, list[str]]):
|
||||
# factor_key -> list of ISO released_at timestamps (earliest first)
|
||||
self._releases = releases
|
||||
|
||||
def series(self, key: str) -> list[dict]:
|
||||
rows = []
|
||||
for ts in self._releases.get(key, []):
|
||||
rows.append({"released_at": ts, "observed_at": ts, "value": 1.0})
|
||||
return rows
|
||||
|
||||
def value_at(self, key: str, as_of: str):
|
||||
for ts in reversed(self._releases.get(key, [])):
|
||||
if ts <= as_of:
|
||||
return 1.0
|
||||
return None
|
||||
|
||||
|
||||
class FakeSiamchartStore:
|
||||
def __init__(self, retrieved: list[str]):
|
||||
self._retrieved = sorted(retrieved)
|
||||
|
||||
def list_ids(self) -> list[str]:
|
||||
return [str(i) for i in range(len(self._retrieved))]
|
||||
|
||||
def snapshot_at(self, as_of: str) -> dict:
|
||||
chosen = [t for t in self._retrieved if t <= as_of]
|
||||
if not chosen:
|
||||
return {}
|
||||
return {"retrieved_at": chosen[-1], "_retrieved_at": chosen[-1]}
|
||||
|
||||
|
||||
def make_price_series(
|
||||
symbols: list[str], start: str, end: str, step_days: int = 30
|
||||
) -> dict:
|
||||
"""A price series {sym: {bars: [...]}} covering [start, end] for every sym."""
|
||||
s = dt.date.fromisoformat(start)
|
||||
e = dt.date.fromisoformat(end)
|
||||
bars = []
|
||||
cur = s
|
||||
while cur <= e:
|
||||
bars.append({"date": cur.isoformat(), "adjusted_close": 10.0})
|
||||
cur += dt.timedelta(days=step_days)
|
||||
return {sym: {"bars": list(bars)} for sym in symbols}
|
||||
|
||||
|
||||
def today_iso() -> str:
|
||||
return yesterday_bangkok().isoformat()
|
||||
|
||||
|
||||
# A factor store that has released every registry factor by a known date.
|
||||
def full_factor_store(release_date: str) -> FakeFactorStore:
|
||||
from app.backtest_readiness import _factor_keys_required
|
||||
releases = {
|
||||
key: [f"{release_date}T09:00:00+07:00"] for key in _factor_keys_required()
|
||||
}
|
||||
return FakeFactorStore(releases)
|
||||
|
||||
|
||||
class YesterdayDefaultTest(unittest.TestCase):
|
||||
def test_yesterday_is_bangkok_tz(self):
|
||||
y = dt.date.fromisoformat(yesterday_bangkok().isoformat())
|
||||
# just assert it's a valid date one day before "now"
|
||||
self.assertIsInstance(y, dt.date)
|
||||
# and timezone is +07 (Bangkok has no DST)
|
||||
import app.backtest_readiness as r
|
||||
now = r.bangkok_now()
|
||||
off = now.utcoffset()
|
||||
assert off is not None
|
||||
self.assertEqual(off.total_seconds(), 7 * 3600)
|
||||
|
||||
|
||||
class NoFactorVintagesTest(unittest.TestCase):
|
||||
def test_blocks_when_no_factor_release(self):
|
||||
store = FakeFactorStore({}) # no factor ever released
|
||||
sc = FakeSiamchartStore(["2025-01-01T09:00:00+07:00"])
|
||||
series = make_price_series(["A"], "2020-01-01", "2026-08-01")
|
||||
res = evaluate_readiness(
|
||||
factor_store=store, siamchart_store=sc, price_series=series
|
||||
)
|
||||
self.assertFalse(res.ready)
|
||||
self.assertIn("factor", res.missing)
|
||||
|
||||
|
||||
class NoSiamchartSnapshotTest(unittest.TestCase):
|
||||
def test_blocks_when_no_snapshot(self):
|
||||
store = full_factor_store("2025-01-01")
|
||||
sc = FakeSiamchartStore([]) # no snapshot
|
||||
series = make_price_series(["A"], "2020-01-01", "2026-08-01")
|
||||
res = evaluate_readiness(
|
||||
factor_store=store, siamchart_store=sc, price_series=series
|
||||
)
|
||||
self.assertFalse(res.ready)
|
||||
self.assertIn("siamchart", res.missing)
|
||||
|
||||
|
||||
class RecommendedStartTest(unittest.TestCase):
|
||||
def test_start_is_latest_of_first_ready_dates(self):
|
||||
# factors ready 2025-01-01, siamchart ready 2025-06-01, price from 2024
|
||||
store = full_factor_store("2025-01-01")
|
||||
sc = FakeSiamchartStore(["2025-06-01T09:00:00+07:00"])
|
||||
series = make_price_series(["A"], "2024-01-01", "2026-08-01")
|
||||
res = evaluate_readiness(
|
||||
factor_store=store, siamchart_store=sc, price_series=series
|
||||
)
|
||||
self.assertTrue(res.ready)
|
||||
self.assertEqual(res.recommended_start, "2025-06-01")
|
||||
self.assertEqual(res.recommended_end, today_iso())
|
||||
|
||||
def test_start_limited_by_price_when_price_latest(self):
|
||||
# factors + siamchart ready 2026-05-01, but price only from 2026-06-01
|
||||
store = full_factor_store("2026-05-01")
|
||||
sc = FakeSiamchartStore(["2026-05-01T09:00:00+07:00"])
|
||||
series = make_price_series(["A"], "2026-06-01", "2026-08-01")
|
||||
res = evaluate_readiness(
|
||||
factor_store=store, siamchart_store=sc, price_series=series
|
||||
)
|
||||
self.assertTrue(res.ready)
|
||||
self.assertEqual(res.recommended_start, "2026-06-01")
|
||||
|
||||
|
||||
class ExplicitWindowTest(unittest.TestCase):
|
||||
def test_explicit_start_before_readiness_blocks(self):
|
||||
store = full_factor_store("2025-06-01")
|
||||
sc = FakeSiamchartStore(["2025-06-01T09:00:00+07:00"])
|
||||
series = make_price_series(["A"], "2024-01-01", "2026-08-01")
|
||||
# user asks for start 2024-01-01, but PIT only ready from 2025-06-01
|
||||
res = evaluate_readiness(
|
||||
factor_store=store, siamchart_store=sc, price_series=series,
|
||||
start="2024-01-01", end="2026-08-01",
|
||||
)
|
||||
# factors missing before start -> blocked
|
||||
self.assertFalse(res.ready)
|
||||
self.assertIn("factor", res.missing)
|
||||
|
||||
def test_explicit_start_after_readiness_is_ready(self):
|
||||
store = full_factor_store("2025-01-01")
|
||||
sc = FakeSiamchartStore(["2025-01-01T09:00:00+07:00"])
|
||||
series = make_price_series(["A"], "2024-01-01", "2026-08-01")
|
||||
res = evaluate_readiness(
|
||||
factor_store=store, siamchart_store=sc, price_series=series,
|
||||
start="2025-06-01", end="2026-08-01",
|
||||
)
|
||||
self.assertTrue(res.ready)
|
||||
|
||||
|
||||
class CoverageShapeTest(unittest.TestCase):
|
||||
def test_to_dict_includes_missing_and_coverage(self):
|
||||
res = evaluate_readiness(
|
||||
factor_store=FakeFactorStore({}),
|
||||
siamchart_store=FakeSiamchartStore([]),
|
||||
price_series=make_price_series(["A"], "2024-01-01", "2026-08-01"),
|
||||
)
|
||||
d = res.to_dict()
|
||||
self.assertIn("ready", d)
|
||||
self.assertIn("missing", d)
|
||||
self.assertIn("coverage", d)
|
||||
self.assertEqual(d["timezone"], "Asia/Bangkok")
|
||||
|
||||
def test_dataclass_defaults(self):
|
||||
r = BacktestReadiness(ready=False)
|
||||
self.assertEqual(r.missing, [])
|
||||
self.assertEqual(r.coverage, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user