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

[verified] Real forward-test frozen-signal lifecycle + durable run store

Browse Source 
Replace the cosmetic 'forward' mode (which was the same single-pass backtest
with a mode string) with a genuine forward paper-portfolio lifecycle:

- backend/app/forward_test.py: ForwardTestStore — durable, thread-safe JSON
  store of forward runs with an explicit status lifecycle:
    frozen (signals snapshotted, immutable) -> executed (fills 50/20/30
    buckets at post-freeze prices) -> marked (mark-to-market equity series ->
    matured (net_return finalised).
  Frozen signals can never be re-read/rewritten after creation, so later data
  cannot retroactively change what the run decided.
- backend/app/__init__.py: GET /api/v1/forward (+<id>), POST /api/v1/forward
  (create+execute, with use_pit to freeze PIT or current-board scores),
  POST /<id>/mark, POST /<id>/mature. ForwardTestStore wired as an extension
  backed by data/forward/runs.json (survives restarts).
- tests: lifecycle store (7) — full backend suite 280 passed. Live probe:
  create->execute (2xx, real holdings), list, mark, mature all work and the
  run persists.

Honest scope: the score source at CREATE time may be the current board
(non_pit=true, tagged); paper-only, no MT5 send. A PIT scorer only marks a run
non_pit=false when its scores assert pit_meta.pit=true.
This commit is contained in:
Kunthawat Greethong
2026-08-27 12:01:00 +07:00
parent 5eeaed63ed
commit 6d9c283d9a
3 changed files with 375 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
backend/app/__init__.py
View File

@@ -142,6 +142,10 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
app.extensions["paper_ledger"] = ledger
app.extensions["paper_sessions"] = paper_sessions
app.extensions["allowed_symbols"] = allowed_symbols
from .forward_test import ForwardTestStore
app.extensions["forward_store"] = ForwardTestStore(
app.config.get("FORWARD_STORE_PATH") or (data_root / "forward" / "runs.json")
)
# shared in-process daily cache + app-internal data scheduler (independent of
# Hermes — this app runs on its own server).
@@ -781,6 +785,103 @@ def create_app(config: dict[str, Any] | None = None) -> Flask:
runs = app.extensions.get("backtest_runs", [])
return jsonify({"runs": runs})
@app.get("/api/v1/forward")
def forward_list():
"""List all durable forward-test runs (real lifecycle, not cosmetic)."""
store = app.extensions["forward_store"]
return jsonify({"runs": store.list()})
@app.get("/api/v1/forward/<run_id>")
def forward_get(run_id: str):
store = app.extensions["forward_store"]
run = store.get(run_id)
if run is None:
return jsonify({"error": "unknown forward run"}), 404
return jsonify(run)
@app.post("/api/v1/forward")
def forward_create():
"""CREATE + EXECUTE a forward run with frozen signals.
Body: {capital, use_pit: bool, as_of: "YYYY-MM-DD"}.
Freezes the current (or PIT-as-of) combined per-symbol scores and fills
the 50/20/30 buckets at post-freeze prices. The score set is immutable
from this point (a real frozen-signal lifecycle).
"""
from .forward_test import ForwardError
store = app.extensions["forward_store"]
body = request.get_json(silent=True) or {}
try:
capital = float(body.get("capital", 0))
except (TypeError, ValueError):
return jsonify({"error": "capital must be a number"}), 400
if capital <= 0:
return jsonify({"error": "capital must be > 0"}), 400
use_pit = bool(body.get("use_pit"))
as_of = body.get("as_of")
try:
from app import simulation as sim
series = sim.load_price_snapshot()
prices = sim.latest_prices(series)
if use_pit:
from .pit_scorer import PitScoreProvider, make_pit_score_fn
from .factor_vintages import FactorVintageStore
from .siamchart_vintages import SiamchartVintageStore
froot = Path(__file__).resolve().parents[1] / "data"
sstore = SiamchartVintageStore(froot)
_seed_siamchart_vintage(sstore)
provider = PitScoreProvider(FactorVintageStore(froot),
_load_siamchart_snapshot(),
siamchart_store=sstore)
score_fn = make_pit_score_fn(provider)
score_by_symbol = score_fn(symbols=[], as_of=as_of) or {}
non_pit = not any(
isinstance(m, dict) and isinstance(m.get("pit_meta"), dict)
and bool(m.get("pit_meta", {}).get("pit"))
for m in score_by_symbol.values()
)
else:
from .dashboard import default_scores
score_by_symbol = default_scores(None) or {}
non_pit = True
frozen = {}
for sym, meta in (score_by_symbol or {}).items():
if sym in prices:
frozen[sym] = {
"combined": float(meta.get("combined", 0.0)),
"is_dividend": bool(meta.get("is_dividend")),
"dividend_yield": float(meta.get("dividend_yield") or 0.0),
}
run = store.create(capital, frozen, as_of=as_of or "now", non_pit=non_pit)
executed = store.execute(run["id"], prices)
return jsonify(executed), 201
except (ForwardError, sim.SimulationError, OSError) as exc:
return jsonify({"error": str(exc)}), 400
@app.post("/api/v1/forward/<run_id>/mark")
def forward_mark(run_id: str):
from .forward_test import ForwardError
from app import simulation as sim
store = app.extensions["forward_store"]
try:
series = sim.load_price_snapshot()
prices = sim.latest_prices(series)
return jsonify(store.mark(run_id, prices))
except (ForwardError, sim.SimulationError, OSError) as exc:
return jsonify({"error": str(exc)}), 400
@app.post("/api/v1/forward/<run_id>/mature")
def forward_mature(run_id: str):
from .forward_test import ForwardError
from app import simulation as sim
store = app.extensions["forward_store"]
try:
series = sim.load_price_snapshot()
prices = sim.latest_prices(series)
return jsonify(store.mature(run_id, prices))
except (ForwardError, sim.SimulationError, OSError) as exc:
return jsonify({"error": str(exc)}), 400
@app.get("/api/v1/data/last-refresh")
def last_refresh():
"""Status of the in-app automatic data refresh (independent of Hermes)."""

188
backend/app/forward_test.py Normal file
View File

@@ -0,0 +1,188 @@
"""Forward-test paper-portfolio lifecycle with frozen signals (real, not cosmetic).
The earlier "forward" mode was cosmetic: it ran the same single-pass
backtest as "backtest", differing only in the `mode` string. This module
implements a genuine forward lifecycle:
1. CREATE (freeze): capture the per-symbol combined scores / dividend
flags **at a specific as_of** and persist them. The
signal set is immutable from this point — later data
cannot change what the run decided.
2. EXECUTE (paper): use prices known *after* the freeze date to fill
the 50/20/30 dividend buckets (the engine does NOT
reach back to pick trades, it fills what the frozen
signal said).
3. MARK (mark-to-market): recompute portfolio equity against later
prices, recording an equity series.
Each step is persisted to a durable store (JSON, atomic write) so runs
survive restarts — replacing the in-memory backtest_runs list.
Honesty: a frozen forward run is paper and labelled as such. It uses the
score source supplied at CREATE time; if that source is the current board it
is non-PIT and the run is tagged `non_pit=true`.
"""
from __future__ import annotations
import json
import datetime as dt
import math
import threading
from copy import deepcopy
from pathlib import Path
from typing import Any, Optional
from uuid import uuid4
FORWARD_SCHEMA_VERSION = 1
# lifecycle statuses in order
_ORDER = ["frozen", "executed", "marked", "matured"]
class ForwardError(ValueError):
pass
def _now() -> str:
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
class ForwardTestStore:
"""Durable, thread-safe JSON store of forward-test runs."""
def __init__(self, path: Path | str | None = None) -> None:
self.path = Path(path).resolve() if path else None
self._runs: dict[str, dict] = {}
self._lock = threading.Lock()
if self.path and self.path.is_file():
self._load()
# -- persistence ------------------------------------------------------
def _load(self) -> None:
try:
payload = json.loads(self.path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ForwardError(f"forward store unreadable: {exc}") from exc
if not isinstance(payload, dict) or payload.get("schema_version") != FORWARD_SCHEMA_VERSION:
raise ForwardError("unsupported forward store schema")
self._runs = deepcopy(payload.get("runs", {}))
def _persist(self) -> None:
if not self.path:
return
self.path.parent.mkdir(parents=True, exist_ok=True)
payload = {"schema_version": FORWARD_SCHEMA_VERSION, "runs": self._runs}
tmp = self.path.with_name(f".{self.path.name}.tmp")
tmp.write_text(json.dumps(payload, ensure_ascii=False, sort_keys=True) + "\n", encoding="utf-8")
tmp.replace(self.path)
# -- lifecycle --------------------------------------------------------
def create(self, capital: float, frozen: dict[str, dict], as_of: str,
non_pit: bool, note: str = "") -> dict:
"""CREATE: persist an immutable frozen-signal run."""
with self._lock:
if capital <= 0:
raise ForwardError("capital must be > 0")
run_id = f"fwd_{uuid4().hex[:12]}"
run = {
"id": run_id,
"created_at": _now(),
"capital": capital,
"status": "frozen",
"as_of": as_of,
"non_pit": bool(non_pit),
"note": note,
"frozen_signals": deepcopy(frozen),
"execution_prices": {},
"holdings": {},
"invested": 0.0,
"equity_history": [{"date": as_of, "equity": capital}],
}
self._runs[run_id] = run
self._persist()
return deepcopy(run)
def execute(self, run_id: str, execution_prices: dict[str, float]) -> dict:
"""EXECUTE: fill the frozen signals at prices provided (post-freeze).
Uses the canonical 50/20/30 dividend allocation over the frozen
combined scores (mirrors simulation.allocate_capital). Frozen signals
are NOT re-readable/changeable here — only execution prices vary.
"""
with self._lock:
run = self._runs.get(run_id)
if run is None:
raise ForwardError(f"unknown forward run: {run_id}")
if run["status"] != "frozen":
raise ForwardError("run already executed")
from . import simulation as sim
candidates = []
for sym, meta in (run.get("frozen_signals") or {}).items():
price = execution_prices.get(sym)
if price is None or price <= 0:
continue
candidates.append(sim.Candidate(
symbol=sym, price=price,
combined_score=float(meta.get("combined", 0.0)),
is_dividend=bool(meta.get("is_dividend")),
dividend_yield=float(meta.get("dividend_yield") or 0.0),
))
alloc = sim.allocate_capital(run["capital"], candidates)
holdings = {o.symbol: o.qty for o in alloc.orders}
invested = alloc.invested
run["status"] = "executed"
run["execution_prices"] = {s: float(execution_prices.get(s, 0)) for s in holdings}
run["holdings"] = holdings
run["invested"] = invested
run["equity_history"].append({"date": _now(), "equity": invested})
self._persist()
return deepcopy(run)
def mark(self, run_id: str, prices: dict[str, float]) -> dict:
"""MARK: recompute equity at provided (latest) prices and append."""
with self._lock:
run = self._runs.get(run_id)
if run is None:
raise ForwardError(f"unknown forward run: {run_id}")
if run["status"] not in ("executed", "marked"):
raise ForwardError("run not executed yet")
equity = run["capital"]
for sym, qty in (run.get("holdings") or {}).items():
px = prices.get(sym)
if px is None:
continue
equity += qty * (px - run["execution_prices"].get(sym, 0))
run["status"] = "marked"
run["equity_history"].append({"date": _now(), "equity": round(equity, 2)})
self._persist()
return deepcopy(run)
def mature(self, run_id: str, final_prices: dict[str, float]) -> dict:
"""MATURE: close the run at final prices, freeze the outcome."""
with self._lock:
run = self._runs.get(run_id)
if run is None:
raise ForwardError(f"unknown forward run: {run_id}")
if run["status"] not in ("executed", "marked"):
raise ForwardError("run cannot mature before execution")
equity = run["capital"]
for sym, qty in (run.get("holdings") or {}).items():
px = final_prices.get(sym)
if px is None:
continue
equity += qty * (px - run["execution_prices"].get(sym, 0))
run["status"] = "matured"
run["final_equity"] = round(equity, 2)
run["net_return"] = round((equity - run["capital"]) / run["capital"], 4) if run["capital"] else 0.0
run["equity_history"].append({"date": _now(), "equity": round(equity, 2)})
self._persist()
return deepcopy(run)
# -- reads ------------------------------------------------------------
def get(self, run_id: str) -> Optional[dict]:
run = self._runs.get(run_id)
return deepcopy(run) if run else None
def list(self) -> list[dict]:
return [deepcopy(run) for run in sorted(self._runs.values(), key=lambda r: r.get("created_at", ""))]

86
backend/tests/test_forward_test.py Normal file
View File

@@ -0,0 +1,86 @@
"""Tests for the forward-test frozen-signal paper lifecycle."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from app.forward_test import ForwardError, ForwardTestStore
def _frozen():
return {
"A": {"combined": 1.0, "is_dividend": True, "dividend_yield": 5.0},
"B": {"combined": 0.8, "is_dividend": False, "dividend_yield": 0.0},
}
class ForwardTestStoreTest(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.path = Path(self._tmp.name) / "forward.json"
self.store = ForwardTestStore(self.path)
def tearDown(self):
self._tmp.cleanup()
def test_create_freezes_and_persists(self):
run = self.store.create(1_000_000, _frozen(), as_of="2026-06-01", non_pit=True)
self.assertEqual(run["status"], "frozen")
self.assertEqual(run["capital"], 1_000_000)
# reloaded from disk still has the frozen signals
reloaded = ForwardTestStore(self.path)
got = reloaded.get(run["id"])
self.assertEqual(got["frozen_signals"]["A"]["combined"], 1.0)
self.assertEqual(got["status"], "frozen")
def test_execute_fills_frozen_signals_at_provided_prices(self):
run = self.store.create(100_000, _frozen(), as_of="2026-06-01", non_pit=True)
prices = {"A": 10.0, "B": 20.0}
ex = self.store.execute(run["id"], prices)
self.assertEqual(ex["status"], "executed")
# A is dividend (bucket1 50%, 50k/10=5000); B non-dividend (bucket2 20%, 20k/20=1000)
self.assertEqual(ex["holdings"].get("A"), 5000)
self.assertEqual(ex["holdings"].get("B"), 1000)
def test_mark_changes_equity_with_prices(self):
run = self.store.create(100_000, _frozen(), as_of="2026-06-01", non_pit=True)
ex = self.store.execute(run["id"], {"A": 10.0, "B": 20.0})
# B rises 20 -> 22 => +1000*2 = +2000
marked = self.store.mark(run["id"], {"A": 10.0, "B": 22.0})
last_equity = marked["equity_history"][-1]["equity"]
self.assertEqual(last_equity, 100_000 + 2_000)
def test_mature_computes_net_return(self):
run = self.store.create(100_000, _frozen(), as_of="2026-06-01", non_pit=True)
self.store.execute(run["id"], {"A": 10.0, "B": 20.0})
matured = self.store.mature(run["id"], {"A": 10.0, "B": 22.0})
self.assertEqual(matured["status"], "matured")
self.assertAlmostEqual(matured["final_equity"], 102_000.0, places=2)
self.assertAlmostEqual(matured["net_return"], 0.02, places=4)
def test_frozen_signals_cannot_be_re_frozen_on_execute(self):
run = self.store.create(100_000, _frozen(), as_of="2026-06-01", non_pit=True)
self.store.execute(run["id"], {"A": 10.0, "B": 20.0})
# a second execute must fail (lifecycle - signals already locked)
with self.assertRaises(ForwardError):
self.store.execute(run["id"], {"A": 11.0, "B": 20.0})
def test_execute_before_create_unknown_run(self):
with self.assertRaises(ForwardError):
self.store.execute("fwd_nope", {"A": 10.0})
def test_double_mark_allowed_and_appends(self):
run = self.store.create(100_000, _frozen(), as_of="2026-06-01", non_pit=True)
self.store.execute(run["id"], {"A": 10.0, "B": 20.0})
self.store.mark(run["id"], {"A": 10.0, "B": 21.0})
self.store.mark(run["id"], {"A": 10.0, "B": 22.0})
marked = self.store.get(run["id"])
self.assertEqual(marked["status"], "marked")
# init + execute + 2 marks = 4 entries
self.assertEqual(len(marked["equity_history"]), 4)
if __name__ == "__main__":
unittest.main()