dekun
e470c5952f
Manual FOMC/CPI/employment entries in settings drive ±1h monitor banners without touching exchange instances. Co-authored-by: Cursor <cursoragent@cursor.com>
74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from hub_macro_calendar_lib import (
|
|
build_banner_message,
|
|
create_event,
|
|
delete_event,
|
|
enrich_alert,
|
|
init_db,
|
|
list_active_alerts,
|
|
list_events,
|
|
update_event,
|
|
)
|
|
|
|
|
|
class HubMacroCalendarLibTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = Path(self.tmp.name) / "macro.db"
|
|
init_db(self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_create_and_list(self):
|
|
row = create_event("cpi", "2026-06-18 20:30", note="核心CPI", db_path=self.db_path)
|
|
self.assertEqual(row["event_type"], "cpi")
|
|
self.assertEqual(row["event_at"], "2026-06-18 20:30")
|
|
rows = list_events(now_ms=row["event_at_ms"] - 86400000, db_path=self.db_path)
|
|
self.assertEqual(len(rows), 1)
|
|
|
|
def test_duplicate_rejected(self):
|
|
create_event("fomc", "2026-07-01 02:00", db_path=self.db_path)
|
|
with self.assertRaises(ValueError):
|
|
create_event("fomc", "2026-07-01 02:00", db_path=self.db_path)
|
|
|
|
def test_active_window_and_messages(self):
|
|
row = create_event("employment", "2026-06-18 20:30", db_path=self.db_path)
|
|
t0 = int(row["event_at_ms"])
|
|
inside = enrich_alert(row, now_ms=t0 - 30 * 60 * 1000)
|
|
self.assertIsNotNone(inside)
|
|
self.assertEqual(inside["phase"], "imminent")
|
|
outside = enrich_alert(row, now_ms=t0 - 2 * 3600 * 1000)
|
|
self.assertIsNone(outside)
|
|
alerts = list_active_alerts(now_ms=t0 + 15 * 60 * 1000, db_path=self.db_path)
|
|
self.assertEqual(len(alerts), 1)
|
|
msg_pos = build_banner_message(alerts[0], has_positions=True)
|
|
msg_flat = build_banner_message(alerts[0], has_positions=False)
|
|
self.assertIn("注意仓位风险", msg_pos)
|
|
self.assertIn("建议等待", msg_flat)
|
|
|
|
def test_update_and_delete(self):
|
|
row = create_event("cpi", "2026-06-18 20:30", db_path=self.db_path)
|
|
updated = update_event(
|
|
row["id"],
|
|
event_at="2026-06-18 21:00",
|
|
note="修正时间",
|
|
db_path=self.db_path,
|
|
)
|
|
self.assertEqual(updated["event_at"], "2026-06-18 21:00")
|
|
self.assertTrue(delete_event(row["id"], db_path=self.db_path))
|
|
self.assertEqual(len(list_events(now_ms=updated["event_at_ms"], db_path=self.db_path)), 0)
|
|
|
|
def test_invalid_type(self):
|
|
with self.assertRaises(ValueError):
|
|
create_event("nfp", "2026-06-18 20:30", db_path=self.db_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|