From 64590156f1d7fa2728d29fbbd39b3e99b7aefab2 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Fri, 28 Aug 2026 10:28:35 +0700 Subject: [PATCH] [verified] Task 5: event-driven PIT backtest engine --- backend/app/backtest_engine.py | 265 ++++++++++++++++++++++++++ backend/tests/test_backtest_engine.py | 165 ++++++++++++++++ 2 files changed, 430 insertions(+) create mode 100644 backend/app/backtest_engine.py create mode 100644 backend/tests/test_backtest_engine.py diff --git a/backend/app/backtest_engine.py b/backend/app/backtest_engine.py new file mode 100644 index 0000000..5df8204 --- /dev/null +++ b/backend/app/backtest_engine.py @@ -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 diff --git a/backend/tests/test_backtest_engine.py b/backend/tests/test_backtest_engine.py new file mode 100644 index 0000000..25e8d80 --- /dev/null +++ b/backend/tests/test_backtest_engine.py @@ -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()