import unittest from app.event_study import EventStudyError, assess_backtest_readiness, run_event_study class EventStudyTests(unittest.TestCase): def test_readiness_blocks_when_independent_events_are_insufficient(self): result = assess_backtest_readiness([{"vintage_id": "v1"}], min_events=12) self.assertEqual(result["status"], "blocked") self.assertEqual(result["reason"], "insufficient_vintages") self.assertEqual(result["available_events"], 1) self.assertEqual(result["required_events"], 12) def test_readiness_rejects_boolean_min_events(self): with self.assertRaisesRegex(EventStudyError, "min_events"): assess_backtest_readiness([], min_events=True) def test_readiness_does_not_count_revisions_as_independent_events(self): entries = [ {"vintage_id": "v1", "source_id": "bot", "published_at": "2026-01-01T08:00:00+07:00"}, {"vintage_id": "v1-revised", "source_id": "bot", "published_at": "2026-01-01T08:00:00+07:00"}, {"vintage_id": "v2", "source_id": "bot", "published_at": "2026-02-01T08:00:00+07:00"}, ] result = assess_backtest_readiness(entries, min_events=3) self.assertEqual(result["status"], "blocked") self.assertEqual(result["available_events"], 2) def test_readiness_canonicalizes_release_timezone_before_deduplication(self): entries = [ {"vintage_id": "v1", "source_id": "bot", "published_at": "2026-01-01T08:00:00+07:00"}, {"vintage_id": "v1-revised", "source_id": "bot", "published_at": "2026-01-01T01:00:00Z"}, ] result = assess_backtest_readiness(entries, min_events=2) self.assertEqual(result["available_events"], 1) self.assertEqual(result["status"], "blocked") def test_event_study_applies_windowed_returns_and_costs(self): events = [ { "event_id": "v1", "published_at": "2026-01-01T08:00:00+07:00", "signals": [ {"symbol": "AOT", "target_weight": 0.5}, {"symbol": "PTT", "target_weight": -0.5}, ], }, { "event_id": "v2", "published_at": "2026-01-01T08:00:00+07:00", "signals": [ {"symbol": "AOT", "target_weight": 0.5}, {"symbol": "PTT", "target_weight": -0.5}, ], }, ] prices = { "AOT": [{"date": "2026-01-01", "close": 100}, {"date": "2026-01-02", "close": 102}, {"date": "2026-01-05", "close": 104}], "PTT": [{"date": "2026-01-01", "close": 100}, {"date": "2026-01-02", "close": 99}, {"date": "2026-01-05", "close": 98}], } result = run_event_study(events, prices, benchmark_prices=[{"date": "2026-01-01", "close": 100}, {"date": "2026-01-02", "close": 101}, {"date": "2026-01-05", "close": 102}], windows=(1,), cost_bps=100, min_events=2) window = result["windows"]["1"] self.assertEqual(result["status"], "ready") self.assertEqual(result["event_count"], 2) self.assertGreater(window["gross_return"], window["net_return"]) self.assertEqual(window["cost_bps"], 100) self.assertIsNotNone(window["benchmark_return"]) def test_event_study_rejects_missing_price_series(self): events = [{"event_id": "v1", "published_at": "2026-01-01T08:00:00+07:00", "signals": [{"symbol": "AOT", "target_weight": 1.0}]}] with self.assertRaisesRegex(EventStudyError, "missing prices"): run_event_study(events, {}, windows=(1,), min_events=1) def test_event_study_uses_next_trading_session_as_default_anchor(self): events = [{"event_id": "v1", "published_at": "2026-01-01T08:00:00+07:00", "signals": [{"symbol": "AOT", "target_weight": 1.0}]}] prices = { "AOT": [ {"date": "2026-01-01", "close": 100}, {"date": "2026-01-02", "close": 105}, {"date": "2026-01-05", "close": 110}, ] } result = run_event_study(events, prices, windows=(1,), min_events=1) self.assertAlmostEqual(result["windows"]["1"]["gross_return"], 110 / 105 - 1, places=8) def test_event_study_does_not_skip_first_session_after_non_trading_event_date(self): events = [{"event_id": "weekend", "published_at": "2026-01-03T08:00:00+07:00", "signals": [{"symbol": "AOT", "target_weight": 1.0}]}] prices = { "AOT": [ {"date": "2026-01-05", "close": 100}, {"date": "2026-01-06", "close": 101}, {"date": "2026-01-07", "close": 102}, ] } result = run_event_study(events, prices, windows=(1,), min_events=1) self.assertAlmostEqual(result["windows"]["1"]["gross_return"], 101 / 100 - 1, places=8) def test_event_study_requires_known_at_for_point_in_time_prices(self): events = [{"event_id": "pit", "published_at": "2026-01-01T08:00:00+07:00", "signals": [{"symbol": "AOT", "target_weight": 1.0}]}] prices = { "AOT": [ {"session_date": "2026-01-01", "close": 100}, {"session_date": "2026-01-02", "close": 105, "known_at": "2026-01-02T17:00:00+07:00"}, ] } with self.assertRaisesRegex(EventStudyError, "known_at"): run_event_study(events, prices, windows=(1,), min_events=1, require_price_known_at=True) def test_event_study_rejects_price_known_after_session_as_lookahead(self): events = [{"event_id": "lookahead", "published_at": "2026-01-01T08:00:00+07:00", "signals": [{"symbol": "AOT", "target_weight": 1.0}]}] prices = { "AOT": [ {"session_date": "2026-01-01", "close": 100, "known_at": "2026-01-01T17:00:00+07:00", "market_timezone": "Asia/Bangkok"}, {"session_date": "2026-01-02", "close": 105, "known_at": "2026-01-03T09:00:00+07:00", "market_timezone": "Asia/Bangkok"}, ] } with self.assertRaisesRegex(EventStudyError, "known_at"): run_event_study(events, prices, windows=(1,), min_events=1, require_price_known_at=True) def test_event_study_checks_known_at_in_market_timezone(self): events = [{"event_id": "timezone-lookahead", "published_at": "2026-01-01T08:00:00+07:00", "signals": [{"symbol": "AOT", "target_weight": 1.0}]}] prices = { "AOT": [ {"session_date": "2026-01-01", "close": 100, "known_at": "2026-01-01T18:30:00Z", "market_timezone": "Asia/Bangkok"}, {"session_date": "2026-01-02", "close": 105, "known_at": "2026-01-02T17:00:00+07:00", "market_timezone": "Asia/Bangkok"}, ] } with self.assertRaisesRegex(EventStudyError, "known_at"): run_event_study(events, prices, windows=(1,), min_events=1, require_price_known_at=True) if __name__ == "__main__": unittest.main()