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

[verified] Task 2: unified event-driven backtest calendar

Browse Source
This commit is contained in:
Kunthawat Greethong
2026-08-28 10:24:54 +07:00
parent 68f2cc1477
commit 6439e9ce73
2 changed files with 557 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

334
backend/app/backtest_events.py Normal file
View File

@@ -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

223
backend/tests/test_backtest_events.py Normal file
View File

@@ -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()