From 6439e9ce73698ab3b19f2560433de99e4afe2b48 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Fri, 28 Aug 2026 10:24:54 +0700 Subject: [PATCH] [verified] Task 2: unified event-driven backtest calendar --- backend/app/backtest_events.py | 334 ++++++++++++++++++++++++++ backend/tests/test_backtest_events.py | 223 +++++++++++++++++ 2 files changed, 557 insertions(+) create mode 100644 backend/app/backtest_events.py create mode 100644 backend/tests/test_backtest_events.py diff --git a/backend/app/backtest_events.py b/backend/app/backtest_events.py new file mode 100644 index 0000000..78abaea --- /dev/null +++ b/backend/app/backtest_events.py @@ -0,0 +1,334 @@ +"""Unified event-driven backtest calendar (Task 2). + +A backtest must rebalance when *information changes*, not on an arbitrary +monthly/quarterly grid. This module derives a chronological event calendar from +the actual PIT sources: + + * factor releases (``FactorVintageStore`` rows carry ``released_at``) + * Siamchart snapshot retrievals (the vintage manifest) + * dividend entitlements (ex-date) and payments (ex-date + 30 calendar days + per the confirmed timing assumption ``ex_date_plus_30d``) + * the end-of-run valuation date + +Each *signal* event (a release that may change the target) is paired with the +next available execution date after it — the trade must not see a price that +did not exist yet. Events are coalesced by day so a day with several releases +produces exactly one frozen signal and at most one rebalance. + +All events are plain dataclasses so the engine (Task 5) can consume them +without coupling to the stores. +""" + +from __future__ import annotations + +import datetime as dt +from dataclasses import dataclass, field +from typing import Any, Iterable, Optional + +# Timezone-aware default for timestamps the calendar normalises to dates. +_TZ_BANGKOK = dt.timezone(dt.timedelta(hours=7)) + +# Dividend cash becomes available this many calendar days after ex-date. +# Confirmed user decision (timing assumption, not an observed pay date). +DIVIDEND_PAYMENT_LAG_DAYS = 30 + + +class BacktestEventError(ValueError): + pass + + +def _parse_date(value: Any) -> dt.date: + if isinstance(value, dt.date): + return value + try: + return dt.date.fromisoformat(str(value)[:10]) + except (TypeError, ValueError) as exc: + raise BacktestEventError(f"invalid date: {value!r}") from exc + + +def _date_from_ts(value: Any) -> Optional[dt.date]: + if not value: + return None + try: + return _parse_date(value) + except BacktestEventError: + return None + + +# --------------------------------------------------------------------------- +# Event types +# --------------------------------------------------------------------------- +@dataclass(frozen=True) +class SignalReleaseEvent: + """One or more data sources produced new information on ``released_at``. + + ``sources`` names which changed (e.g. ``["factor:energy_net_margin"]`` or + ``["siamchart"]``). The engine freezes the target using only data knowable + by ``released_at`` (anti-look-ahead). + """ + + released_at: str + sources: list[str] = field(default_factory=list) + kind: str = "signal" + + @property + def date(self) -> dt.date: + return _parse_date(self.released_at) + + +@dataclass(frozen=True) +class ExecutionEvent: + """The next available trading date to execute the signal frozen on + ``signal_date``. Never before the signal's release date.""" + + signal_date: str + execution_date: str + kind: str = "execution" + + @property + def date(self) -> dt.date: + return _parse_date(self.execution_date) + + +@dataclass(frozen=True) +class DividendEntitlementEvent: + """Shares held *before* ``ex_date`` qualify for this dividend.""" + + symbol: str + ex_date: str + per_share: float + kind: str = "dividend_entitlement" + + @property + def date(self) -> dt.date: + return _parse_date(self.ex_date) + + +@dataclass(frozen=True) +class DividendPaymentEvent: + """Cash becomes available on ``payment_date``. + + ``timing_method = ex_date_plus_30d`` because the Siamchart source supplies + ex-date + DPS but not an observed payment date; the confirmed assumption is + that cash settles exactly ``DIVIDEND_PAYMENT_LAG_DAYS`` after ex-date. + """ + + symbol: str + ex_date: str + payment_date: str + per_share: float + timing_method: str = "ex_date_plus_30d" + kind: str = "dividend_payment" + + @property + def date(self) -> dt.date: + return _parse_date(self.payment_date) + + +@dataclass(frozen=True) +class EndValuationEvent: + """Mark holdings to market at/after ``end_date`` (final valuation).""" + + end_date: str + kind: str = "end_valuation" + + @property + def date(self) -> dt.date: + return _parse_date(self.end_date) + + +# --------------------------------------------------------------------------- +# Calendar builders +# --------------------------------------------------------------------------- +def _yield_signal_dates(factor_store: Any, start: dt.date, end: dt.date) -> list[SignalReleaseEvent]: + """Factor releases within [start, end], one event per released_at day.""" + from .backtest_readiness import _factor_keys_required + + factors = _factor_keys_required() + by_day: dict[dt.date, list[str]] = {} + for key in factors: + try: + rows = factor_store.series(key) + except Exception: + rows = [] + for r in rows: + d = _date_from_ts(r.get("released_at")) or _date_from_ts(r.get("observed_at")) + if d is None or d < start or d > end: + continue + by_day.setdefault(d, []).append(f"factor:{key}") + return [ + SignalReleaseEvent( + released_at=d.isoformat(), + sources=sorted(set(srcs)), + ) + for d, srcs in sorted(by_day.items()) + ] + + +def _yield_siamchart_signals( + siamchart_store: Any, start: dt.date, end: dt.date +) -> list[SignalReleaseEvent]: + """Siamchart snapshot retrievals within [start, end], coalesced by day.""" + by_day: dict[dt.date, int] = {} + try: + ids = siamchart_store.list_ids() + except Exception: + ids = [] + for sid in ids: + # read each snapshot's retrieved_at from the manifest if available + d = None + try: + manifest = siamchart_store._load_manifest() + entry = manifest.get("snapshots", {}).get(str(sid), {}) + d = _date_from_ts(entry.get("retrieved_at")) + except Exception: + d = None + if d is None: + # fallback: probe snapshot_at at a far-future cutoff for the newest + # (only usable when the store is single-snapshot). + try: + snap = siamchart_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")) + except Exception: + d = None + if d is not None and start <= d <= end: + by_day[d] = by_day.get(d, 0) + 1 + return [ + SignalReleaseEvent(released_at=d.isoformat(), sources=["siamchart"]) + for d in sorted(by_day) + ] + + +def _yield_dividend_events(ledger: Any, start: dt.date, end: dt.date) -> list: + """Dividend entitlement + payment events for dated (non-estimate) entries.""" + events: list = [] + if ledger is None: + return events + try: + symbols = ledger.symbols() + except Exception: + return events + for sym in symbols: + try: + entries = ledger.entries(sym) + except Exception: + entries = [] + for e in entries: + if e.get("estimate"): + continue # estimates are not dated cash flows + ex_date = e.get("ex_date") + if not ex_date: + continue + try: + ex = _parse_date(ex_date) + except BacktestEventError: + continue + per_share = float(e.get("per_share") or 0.0) + if per_share <= 0: + continue + if start <= ex <= end: + events.append(DividendEntitlementEvent( + symbol=sym, ex_date=ex.isoformat(), per_share=per_share, + )) + pay = ex + dt.timedelta(days=DIVIDEND_PAYMENT_LAG_DAYS) + if start <= pay <= end: + events.append(DividendPaymentEvent( + symbol=sym, ex_date=ex.isoformat(), + payment_date=pay.isoformat(), per_share=per_share, + )) + return events + + +def build_event_calendar( + *, + factor_store: Any = None, + siamchart_store: Any = None, + dividend_ledger: Any = None, + start: str, + end: str, +) -> list: + """Build the full chronological event list for [start, end]. + + Events are returned **sorted by date**, with signal events (factor/Siamchart + releases) interleaved with dividend and end events. Same-day releases are + coalesced so a single day yields one signal. + + Execution pairing is the engine's job (Task 5), but this builder guarantees + only events within the window are produced and that the end valuation event + is present. + """ + s = _parse_date(start) + e = _parse_date(end) + if s > e: + raise BacktestEventError("end must be on/after start") + + events: list = [] + events.extend(_yield_signal_dates(factor_store, s, e)) + events.extend(_yield_siamchart_signals(siamchart_store, s, e)) + events.extend(_yield_dividend_events(dividend_ledger, s, e)) + events.append(EndValuationEvent(end_date=e.isoformat())) + + # stable sort by event date + events.sort(key=_event_sort_key) + return events + + +def _event_sort_key(ev: Any): + if hasattr(ev, "date"): + return ev.date.isoformat() + return "" + + +# --------------------------------------------------------------------------- +# Execution-date mapping +# --------------------------------------------------------------------------- +def _trading_days(price_series: dict[str, Any]) -> set[dt.date]: + """Union of all distinct bar dates across symbols (the tradable calendar).""" + days: set[dt.date] = set() + for sym, s in (price_series or {}).items(): + for b in (s or {}).get("bars", []): + d = _date_from_ts(b.get("date")) + if d is not None: + days.add(d) + return days + + +def next_trading_day(price_series: dict[str, Any], after: dt.date) -> Optional[dt.date]: + """Strictly-first trading day after ``after`` in the price calendar. + + Returns None if no trading day exists strictly after ``after``. This is the + anti-look-ahead execution rule: the trade cannot see a price that was not + yet known on the release day. + """ + trading = _trading_days(price_series) + future = [d for d in trading if d > after] + return min(future) if future else None + + +def pair_signal_executions( + signals: Iterable[SignalReleaseEvent], + price_series: dict[str, Any], + end: dt.date, +) -> list[ExecutionEvent]: + """Map each signal to the next trading day after its release, capped at end. + + A signal whose next trading day falls beyond the run end cannot execute and + is dropped (the engine simply values the final holdings at end instead). + """ + trading = _trading_days(price_series) + out: list[ExecutionEvent] = [] + for sig in signals: + sdate = sig.date + future = [d for d in trading if d > sdate] + if not future: + continue + exec_day = min(future) + if exec_day > end: + continue # cannot execute within the backtest horizon + out.append(ExecutionEvent( + signal_date=sdate.isoformat(), execution_date=exec_day.isoformat(), + )) + out.sort(key=lambda e: e.date) + return out diff --git a/backend/tests/test_backtest_events.py b/backend/tests/test_backtest_events.py new file mode 100644 index 0000000..5e8bb5a --- /dev/null +++ b/backend/tests/test_backtest_events.py @@ -0,0 +1,223 @@ +"""Tests for the unified event-driven backtest calendar (Task 2).""" + +from __future__ import annotations + +import datetime as dt +import unittest + +from app.backtest_events import ( + DIVIDEND_PAYMENT_LAG_DAYS, + DividendEntitlementEvent, + DividendPaymentEvent, + EndValuationEvent, + ExecutionEvent, + SignalReleaseEvent, + build_event_calendar, +) + + +class FakeFactorStore: + """factor_key -> rows with released_at (chronological).""" + + def __init__(self, releases: dict[str, list[str]]): + self._releases = releases + + def series(self, key: str) -> list[dict]: + return [ + {"released_at": ts, "observed_at": ts, "value": 1.0} + for ts in self._releases.get(key, []) + ] + + +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 _load_manifest(self) -> dict: + return { + "snapshots": { + str(i): {"retrieved_at": ts} for i, ts in enumerate(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]} + + +class FakeLedger: + def __init__(self, entries): + # entries: list of {symbol, ex_date, per_share, estimate} + self._by = {} + for e in entries: + self._by.setdefault(e["symbol"], []).append(e) + + def symbols(self) -> list[str]: + return list(self._by.keys()) + + def entries(self, sym: str) -> list[dict]: + return self._by.get(sym, []) + + +class SignalReleaseTest(unittest.TestCase): + def test_factor_release_produces_signal_event(self): + store = FakeFactorStore({ + "energy_net_margin": ["2026-03-15T09:00:00+07:00"], + }) + events = build_event_calendar( + factor_store=store, start="2026-03-01", end="2026-04-01" + ) + sigs = [e for e in events if isinstance(e, SignalReleaseEvent)] + self.assertEqual(len(sigs), 1) + self.assertEqual(sigs[0].released_at[:10], "2026-03-15") + self.assertIn("factor:energy_net_margin", sigs[0].sources) + + def test_same_day_factor_releases_coalesce_into_one_signal(self): + store = FakeFactorStore({ + "energy_net_margin": ["2026-03-15T09:00:00+07:00"], + "macro_consumption": ["2026-03-15T14:00:00+07:00"], + }) + events = build_event_calendar( + factor_store=store, start="2026-03-01", end="2026-04-01" + ) + sigs = [e for e in events if isinstance(e, SignalReleaseEvent)] + self.assertEqual(len(sigs), 1) # coalesced + self.assertEqual(len(sigs[0].sources), 2) + + def test_release_outside_window_is_excluded(self): + store = FakeFactorStore({ + "energy_net_margin": ["2026-03-15T09:00:00+07:00"], + }) + events = build_event_calendar( + factor_store=store, start="2026-04-01", end="2026-05-01" + ) + sigs = [e for e in events if isinstance(e, SignalReleaseEvent)] + self.assertEqual(len(sigs), 0) + + +class SiamchartSignalTest(unittest.TestCase): + def test_snapshot_retrieval_produces_signal(self): + store = FakeSiamchartStore(["2026-03-20T09:00:00+07:00"]) + events = build_event_calendar( + siamchart_store=store, start="2026-03-01", end="2026-04-01" + ) + sigs = [e for e in events if isinstance(e, SignalReleaseEvent)] + sc = [s for s in sigs if "siamchart" in s.sources] + self.assertEqual(len(sc), 1) + self.assertEqual(sc[0].released_at[:10], "2026-03-20") + + +class DividendEventTest(unittest.TestCase): + def test_entitlement_and_payment_events(self): + ledger = FakeLedger([{ + "symbol": "A", "ex_date": "2026-03-10", + "per_share": 1.5, "estimate": False, + }]) + events = build_event_calendar( + dividend_ledger=ledger, start="2026-03-01", end="2026-05-01" + ) + ents = [e for e in events if isinstance(e, DividendEntitlementEvent)] + pays = [e for e in events if isinstance(e, DividendPaymentEvent)] + self.assertEqual(len(ents), 1) + self.assertEqual(ents[0].ex_date, "2026-03-10") + self.assertEqual(ents[0].per_share, 1.5) + self.assertEqual(len(pays), 1) + expected_pay = dt.date(2026, 3, 10) + dt.timedelta(days=DIVIDEND_PAYMENT_LAG_DAYS) + self.assertEqual(pays[0].payment_date, expected_pay.isoformat()) + self.assertEqual(pays[0].timing_method, "ex_date_plus_30d") + + def test_estimate_entry_produces_no_dated_events(self): + ledger = FakeLedger([{ + "symbol": "A", "ex_date": "2026-03-10", + "per_share": 1.5, "estimate": True, + }]) + events = build_event_calendar( + dividend_ledger=ledger, start="2026-03-01", end="2026-05-01" + ) + ents = [e for e in events if isinstance(e, DividendEntitlementEvent)] + pays = [e for e in events if isinstance(e, DividendPaymentEvent)] + self.assertEqual(len(ents), 0) + self.assertEqual(len(pays), 0) + + +class CalendarShapeTest(unittest.TestCase): + def test_end_valuation_event_always_present(self): + events = build_event_calendar(start="2026-03-01", end="2026-04-01") + ends = [e for e in events if isinstance(e, EndValuationEvent)] + self.assertEqual(len(ends), 1) + self.assertEqual(ends[0].end_date, "2026-04-01") + + def test_events_sorted_chronologically(self): + factor_store = FakeFactorStore({ + "energy_net_margin": ["2026-03-15T09:00:00+07:00"], + }) + sc = FakeSiamchartStore(["2026-03-20T09:00:00+07:00"]) + ledger = FakeLedger([{ + "symbol": "A", "ex_date": "2026-03-25", + "per_share": 1.0, "estimate": False, + }]) + events = build_event_calendar( + factor_store=factor_store, siamchart_store=sc, + dividend_ledger=ledger, start="2026-03-01", end="2026-05-20", + ) + dates = [e.date for e in events] + self.assertEqual(dates, sorted(dates)) + + def test_invalid_window_raises(self): + with self.assertRaises(Exception): + build_event_calendar(start="2026-05-01", end="2026-03-01") + + +class ExecutionEventTest(unittest.TestCase): + def test_execution_event_shape(self): + ev = ExecutionEvent(signal_date="2026-03-15", execution_date="2026-03-16") + self.assertEqual(ev.kind, "execution") + self.assertEqual(ev.date.isoformat(), "2026-03-16") + + +def make_daily_series(start: str, days: int) -> dict: + """Random-free daily price series with one bar per calendar day.""" + bars = [] + s = dt.date.fromisoformat(start) + for i in range(days): + bars.append({"date": (s + dt.timedelta(days=i)).isoformat(), + "adjusted_close": 10.0}) + return {"A": {"bars": bars}} + + +class ExecutionMappingTest(unittest.TestCase): + def test_next_trading_day_is_strictly_after_signal(self): + from app.backtest_events import next_trading_day + series = make_daily_series("2026-03-10", 10) # 03-10..03-19 + nxt = next_trading_day(series, dt.date(2026, 3, 15)) + self.assertEqual(nxt, dt.date(2026, 3, 16)) + + def test_no_trading_day_after_returns_none(self): + from app.backtest_events import next_trading_day + series = make_daily_series("2026-03-10", 10) + nxt = next_trading_day(series, dt.date(2026, 3, 25)) + self.assertIsNone(nxt) + + def test_pair_signal_executions(self): + from app.backtest_events import pair_signal_executions + series = make_daily_series("2026-03-10", 10) + sigs = [SignalReleaseEvent(released_at="2026-03-15T09:00:00+07:00")] + pairs = pair_signal_executions(sigs, series, dt.date(2026, 3, 19)) + self.assertEqual(len(pairs), 1) + self.assertEqual(pairs[0].execution_date, "2026-03-16") + + def test_signal_past_end_not_executed(self): + from app.backtest_events import pair_signal_executions + series = make_daily_series("2026-03-10", 10) + sigs = [SignalReleaseEvent(released_at="2026-03-15T09:00:00+07:00")] + pairs = pair_signal_executions(sigs, series, dt.date(2026, 3, 14)) + self.assertEqual(len(pairs), 0) + + +if __name__ == "__main__": + unittest.main()