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

[verified] Task 5: event-driven PIT backtest engine

Browse Source
This commit is contained in:
Kunthawat Greethong
2026-08-28 10:28:35 +07:00
parent 71b893ee73
commit 64590156f1
2 changed files with 430 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

265
backend/app/backtest_engine.py Normal file
View File

@@ -0,0 +1,265 @@
"""Event-driven PIT backtest engine (Task 5).
Replaces the calendar-rebalance loop with a strict point-in-time event
processor. It consumes the unified event calendar (Task 2), the portfolio
ledger (Task 3), and the lot/cash rebalancer (Task 4) to simulate exactly the
user-described lifecycle:
1. An initial signal is frozen at the run start (advice knows only what was
released by then), and executed at the next trading day.
2. Each new data release freezes a new signal; if the target changes, the
portfolio is rebalanced on the next trading day after the release.
3. Dividends are entitled on ex-date (shares held before it) and their cash
becomes available exactly 30 calendar days later, funding later buys.
4. Sales realize P&L (average cost, minus a 0.3% all-in fee) and free up cash
before subsequent buys.
5. At the end date the portfolio is marked to market and every P&L component
is reported and reconciled.
Every score is computed with only data knowable at the release timestamp
(``score_fn(symbols, as_of)``): if the supplied scorer is PIT-aware and returns
``pit_meta`` proving it, ``leakage_guard`` is True; otherwise it is False.
"""
from __future__ import annotations
import datetime as dt
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from .backtest_events import (
DIVIDEND_PAYMENT_LAG_DAYS,
DividendEntitlementEvent,
DividendPaymentEvent,
EndValuationEvent,
SignalReleaseEvent,
_date_from_ts,
build_event_calendar,
next_trading_day,
)
from .portfolio_ledger import PortfolioLedger
from .portfolio_rebalancer import PortfolioRebalancer, build_candidates
@dataclass
class EventBacktestResult:
start: str
end: str
capital: float
ledger: Optional[PortfolioLedger] = None
leakage_guard: bool = False
rebalances: int = 0
events: list = field(default_factory=list)
# summary fields (filled by to_dict)
ending_cash: float = 0.0
ending_market_value: float = 0.0
final_equity: float = 0.0
realized_trading_pnl: float = 0.0
unrealized_trading_pnl: float = 0.0
price_pnl: float = 0.0
dividend_cash_received: float = 0.0
dividend_receivable: float = 0.0
transaction_costs: float = 0.0
net_return: float = 0.0
account_reconciled: bool = False
_final_prices: dict = field(default_factory=dict, repr=False)
def to_dict(self) -> dict:
prices = self._final_prices or {}
rec = self.ledger.reconcile(prices) if self.ledger else {}
holdings = [
{
"symbol": p.symbol, "qty": p.qty,
"average_cost": round(p.average_cost, 4),
"last_price": round(prices.get(p.symbol, 0.0), 4),
"market_value": round(p.qty * prices.get(p.symbol, 0.0), 2),
"unrealized_pnl": round(p.qty * (prices.get(p.symbol, 0.0) - p.average_cost), 2),
}
for p in (self.ledger.positions() if self.ledger else [])
]
return {
"start": self.start, "end": self.end,
"capital": self.capital,
"ending_cash": round(self.ending_cash, 2),
"ending_market_value": round(self.ending_market_value, 2),
"dividend_receivable": round(self.dividend_receivable, 2),
"final_equity": round(self.final_equity, 2),
"net_change": round(self.final_equity - self.capital, 2),
"net_return": round(self.net_return, 4),
"realized_trading_pnl": round(self.realized_trading_pnl, 2),
"unrealized_trading_pnl": round(self.unrealized_trading_pnl, 2),
"price_pnl": round(self.price_pnl, 2),
"dividend_cash_received": round(self.dividend_cash_received, 2),
"transaction_costs": round(self.transaction_costs, 2),
"fee_rate": 0.003,
"dividend_timing": "ex_date_plus_30d",
"holdings": holdings,
"trades": [t.to_dict() for t in (self.ledger.state.trades if self.ledger else [])],
"dividends": (self.ledger.state.dividends if self.ledger else []),
"rebalances": self.rebalances,
"leakage_guard": self.leakage_guard,
"accounting_reconciled": self.account_reconciled,
# compatibility aliases for the existing UI/API
"final_value": round(self.final_equity, 2),
"dividend_income": round(self.dividend_cash_received, 2),
"dividend_method": "dated_ledger",
"holdings_map": {p.symbol: p.qty for p in (self.ledger.positions() if self.ledger else [])},
}
def _latest_close(series: dict, sym: str, date: dt.date) -> Optional[float]:
bars = (series.get(sym) or {}).get("bars", [])
chosen = None
for b in bars:
d = str(b.get("date") or "")[:10]
try:
bd = dt.date.fromisoformat(d)
except ValueError:
continue
if bd <= date:
chosen = b.get("adjusted_close")
return float(chosen) if chosen is not None else None
def run_event_backtest(
*,
start: str,
end: str,
capital: float = 1_000_000,
factor_store: Any = None,
siamchart_store: Any = None,
dividend_ledger: Any = None,
price_series: dict | None = None,
score_fn: Optional[Callable] = None,
symbols: Optional[list[str]] = None,
) -> EventBacktestResult:
"""Run the event-driven PIT backtest over [start, end]."""
from .simulation import load_price_snapshot
if price_series is None:
price_series = load_price_snapshot()
if not price_series:
raise ValueError("no price data for backtest")
ledger = PortfolioLedger(capital)
result = EventBacktestResult(start=start, end=end, capital=float(capital))
result.ledger = ledger
s_date = dt.date.fromisoformat(start)
e_date = dt.date.fromisoformat(end)
# Build the event timeline.
calendar = build_event_calendar(
factor_store=factor_store,
siamchart_store=siamchart_store,
dividend_ledger=dividend_ledger,
start=start, end=end,
)
# Collect signal releases (chronological) and pair each to an execution.
signal_events = [ev for ev in calendar if isinstance(ev, SignalReleaseEvent)]
# ensure an initial signal at the start so the first allocation happens
init_signal = SignalReleaseEvent(released_at=start + "T00:00:00+07:00")
all_signals = [init_signal] + signal_events
# Map each signal to its next trading day.
exec_map: dict[dt.date, SignalReleaseEvent] = {}
for sig in all_signals:
nxt = next_trading_day(price_series, sig.date)
if nxt is None:
continue
if nxt > e_date:
continue
# if multiple signals map to the same execution day, keep the latest
exec_map[nxt] = sig
# Also collect dividend events from the calendar with their own ordering.
dividend_events = [
ev for ev in calendar
if isinstance(ev, (DividendEntitlementEvent, DividendPaymentEvent))
]
# Merge all dated actions into one timeline.
# Each day holds a list of (action, payload) where action is one of
# "rebalance", "dividend_entitlement", "dividend_payment".
timeline: dict[dt.date, list[tuple[str, Any]]] = {}
for exec_day, sig in exec_map.items():
timeline.setdefault(exec_day, []).append(("rebalance", sig))
for ev in dividend_events:
timeline.setdefault(ev.date, []).append((ev.kind, ev))
last_scores: dict = {}
leakage_guard = False
def freeze_scores(sig: SignalReleaseEvent) -> None:
nonlocal last_scores, leakage_guard
as_of = sig.released_at
syms = symbols or list(price_series.keys())
if score_fn is not None:
sc = score_fn(syms, as_of) or {}
else:
from .dashboard import default_scores
sc = default_scores(syms) or {}
# leakage_guard only when the scores attest PIT provenance
any_pit = any(
isinstance(m, dict) and isinstance(m.get("pit_meta"), dict)
and bool(m.get("pit_meta", {}).get("pit"))
for m in sc.values()
)
if any_pit:
leakage_guard = True
last_scores = sc
def execute_rebalance(exec_day: dt.date, sig: SignalReleaseEvent) -> None:
nonlocal result
cands = build_candidates(last_scores, price_series, exec_day)
if not cands:
return
rb = PortfolioRebalancer(ledger, price_series, cands)
r = rb.rebalance(date=exec_day, signal_date=sig.date)
result.rebalances += 1
result.events.append({
"type": "rebalance", "signal_date": sig.date.isoformat(),
"execution_date": exec_day.isoformat(),
"trades": len(r.trades), "notes": r.notes,
})
# ---------- process timeline ----------
for day in sorted(timeline.keys()):
for action, payload in timeline[day]:
if action == "rebalance":
sig = payload # SignalReleaseEvent
freeze_scores(sig)
execute_rebalance(day, sig)
elif action == "dividend_entitlement":
ledger.record_dividend_entitlement(
payload.symbol, payload.ex_date, payload.per_share
)
elif action == "dividend_payment":
ledger.pay_due_dividends(payload.payment_date)
# payout any dividend whose assumed payment date is reached by today
ledger.pay_due_dividends(day.isoformat())
# ---------- final mark-to-market ----------
# value every held symbol at the end date (latest close on-or-before end)
final_prices: dict[str, float] = {}
for sym in list(price_series.keys()):
px = _latest_close(price_series, sym, e_date)
if px is not None:
final_prices[sym] = px
ledger.pay_due_dividends(end)
rec = ledger.reconcile(final_prices)
result._final_prices = final_prices
result.ending_cash = rec["cash"]
result.ending_market_value = rec["market_value"]
result.final_equity = rec["equity"]
result.realized_trading_pnl = rec["realized_trading_pnl"]
result.unrealized_trading_pnl = rec["unrealized_trading_pnl"]
result.price_pnl = rec["realized_trading_pnl"] + rec["unrealized_trading_pnl"]
result.dividend_cash_received = rec["dividend_cash_received"]
result.dividend_receivable = rec["dividend_receivable"]
result.transaction_costs = rec["transaction_costs"]
result.net_return = (rec["equity"] - capital) / capital if capital else 0.0
result.account_reconciled = rec["balanced"]
result.leakage_guard = leakage_guard
return result

165
backend/tests/test_backtest_engine.py Normal file
View File

@@ -0,0 +1,165 @@
"""Integration tests for the event-driven backtest engine (Task 5)."""
from __future__ import annotations
import datetime as dt
import unittest
from app.backtest_engine import (
run_event_backtest,
)
from app.backtest_events import DIVIDEND_PAYMENT_LAG_DAYS
def make_series(symbols, start, days, price_map=None):
"""Price series; price_map maps date->{sym:price} overrides, else flat."""
s = dt.date.fromisoformat(start)
out = {}
for sym in symbols:
bars = []
for i in range(days):
d = (s + dt.timedelta(days=i)).isoformat()
base = (price_map or {}).get(d, {}).get(sym, 10.0)
bars.append({"date": d, "adjusted_close": float(base)})
out[sym] = {"bars": bars}
return out
def price_jump_series(symbols, start, days, jump_date, new_price=20.0):
"""Price series that jumps on/after jump_date for all symbols."""
s = dt.date.fromisoformat(start)
jd = dt.date.fromisoformat(jump_date)
out = {}
for sym in symbols:
bars = []
for i in range(days):
d = (s + dt.timedelta(days=i))
px = new_price if d >= jd else 10.0
bars.append({"date": d.isoformat(), "adjusted_close": px})
out[sym] = {"bars": bars}
return out
class FakeFactorStore:
"""factor_key -> release timestamps; used to trigger releases on a date."""
def __init__(self, releases=None):
self._rels = releases or {}
def series(self, key):
return [
{"released_at": ts, "observed_at": ts, "value": 1.0}
for ts in self._rels.get(key, [])
]
class FakeLedger:
def __init__(self, entries):
self._by = {}
for e in entries:
self._by.setdefault(e["symbol"], []).append(e)
def symbols(self):
return list(self._by.keys())
def entries(self, sym):
return self._by.get(sym, [])
def noop_scorer(symbols, as_of=None):
# non-PIT scorer: all dividend, equal score -> allocation picks top by list
out = {}
for i, sym in enumerate(symbols):
out[sym] = {
"combined": 1.0 / (i + 1),
"is_dividend": True,
"dividend_yield": 3.0,
}
return out
class NoLookAheadTest(unittest.TestCase):
def test_signal_release_three_months_in_triggers_one_rebalance(self):
series = make_series(["A", "B"], "2026-01-01", 200)
# a factor release 3 months later (daily series, days ~90)
rel = "2026-04-01T09:00:00+07:00"
fstore = FakeFactorStore({"energy_net_margin": [rel]})
res = run_event_backtest(
start="2026-01-01", end="2026-06-30",
capital=100_000,
factor_store=fstore,
price_series=series,
score_fn=noop_scorer,
symbols=["A", "B"],
)
# event-driven: releases were 1 (the release day) + initial; the initial
# signal always triggers the first buy; the release triggers one more
# rebalance (>=1 because portfolio changes when scores re-rank).
self.assertGreaterEqual(res.rebalances, 1)
self.assertTrue(res.account_reconciled)
class LifecycleTest(unittest.TestCase):
def test_full_lifecycle_reconciles_with_trade_pnl(self):
# price jumps mid-run so a sell realizes real P&L
series = price_jump_series(["A", "B"], "2026-01-01", 90, "2026-03-01", 20.0)
# dividend on A before the jump
ledger = FakeLedger([{
"symbol": "A", "ex_date": "2026-02-15",
"per_share": 1.0, "estimate": False,
}])
res = run_event_backtest(
start="2026-01-01", end="2026-03-30",
capital=100_000,
dividend_ledger=ledger,
price_series=series,
score_fn=noop_scorer,
symbols=["A", "B"],
)
self.assertTrue(res.account_reconciled)
# final equity must be consistent with reported components
self.assertAlmostEqual(
res.final_equity,
res.capital + res.realized_trading_pnl + res.unrealized_trading_pnl
+ res.dividend_cash_received - res.transaction_costs,
places=1,
)
def test_dividend_cash_becomes_available_after_ex_date_plus_30(self):
series = make_series(["A"], "2026-01-01", 200)
ex = "2026-01-15"
ledger = FakeLedger([{
"symbol": "A", "ex_date": ex, "per_share": 2.0, "estimate": False,
}])
res = run_event_backtest(
start="2026-01-01", end="2026-04-01",
capital=200_000, dividend_ledger=ledger,
price_series=series, score_fn=noop_scorer, symbols=["A"],
)
# by end (after ex+30) the dividend has been paid into cash
self.assertGreater(res.dividend_cash_received, 0)
self.assertEqual(res.dividend_receivable, 0.0)
class ResultContractTest(unittest.TestCase):
def test_to_dict_has_expected_shape(self):
series = make_series(["A"], "2026-01-01", 120)
res = run_event_backtest(
start="2026-01-01", end="2026-04-30",
capital=100_000, price_series=series,
score_fn=noop_scorer, symbols=["A"],
)
d = res.to_dict()
for key in [
"final_equity", "net_return", "realized_trading_pnl",
"unrealized_trading_pnl", "price_pnl", "dividend_cash_received",
"transaction_costs", "holdings", "trades", "leakage_guard",
"accounting_reconciled", "fee_rate", "dividend_timing",
]:
self.assertIn(key, d)
self.assertEqual(d["fee_rate"], 0.003)
self.assertEqual(d["dividend_timing"], "ex_date_plus_30d")
if __name__ == "__main__":
unittest.main()