224 lines
8.3 KiB
Python
224 lines
8.3 KiB
Python
"""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()
|