100 lines
4.9 KiB
Python
100 lines
4.9 KiB
Python
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|