dekun
54f1857fa2
Mirror perpetual archive flow into archive_options_trade_cache for offline calendar and review. Co-authored-by: Cursor <cursoragent@cursor.com>
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""期权档案缓存 upsert / 列表 / 日历."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from lib.hub.hub_options_archive_lib import (
|
|
init_options_archive_db,
|
|
list_archive_options_calendar,
|
|
list_daily_options_trades,
|
|
upsert_options_trades_cache,
|
|
)
|
|
|
|
|
|
class TestHubOptionsArchive(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._td = tempfile.TemporaryDirectory()
|
|
self.db = Path(self._td.name) / "hub_symbol_archive.db"
|
|
init_options_archive_db(self.db)
|
|
|
|
def tearDown(self) -> None:
|
|
self._td.cleanup()
|
|
|
|
def test_upsert_and_list_daily(self) -> None:
|
|
trades = [
|
|
{
|
|
"history_key": "local_opt:1",
|
|
"source_type": "option_spot",
|
|
"source_label": "纯期权",
|
|
"underlying": "ETH",
|
|
"inst_id": "ETH-USD-250725-3200-C",
|
|
"opt_type": "C",
|
|
"opened_at": "2026-07-20 10:00:00",
|
|
"closed_at": "2026-07-20 16:00:00",
|
|
"hold_seconds": 21600,
|
|
"realized_pnl_total": 12.5,
|
|
"premium_paid": 8.0,
|
|
"reviewed": True,
|
|
"strategy_tag": "假突破",
|
|
},
|
|
{
|
|
"history_key": "local_opt:2",
|
|
"source_type": "option_spot",
|
|
"underlying": "ETH",
|
|
"opened_at": "2026-07-19 10:00:00",
|
|
"closed_at": "2026-07-19 12:00:00",
|
|
"realized_pnl_total": -3.0,
|
|
"excluded_as_hedge_leg": 1,
|
|
},
|
|
]
|
|
r = upsert_options_trades_cache("okx", trades, db_path=self.db)
|
|
self.assertEqual(r["upserted"], 1)
|
|
payload = list_daily_options_trades(
|
|
"2026-07-20",
|
|
period="today",
|
|
db_path=self.db,
|
|
)
|
|
self.assertEqual(len(payload["trades"]), 1)
|
|
self.assertEqual(payload["trades"][0]["history_key"], "local_opt:1")
|
|
self.assertAlmostEqual(payload["stats"]["pnl_total"], 12.5)
|
|
cal = list_archive_options_calendar(2026, 7, db_path=self.db)
|
|
self.assertIn("2026-07-20", cal["days"])
|
|
self.assertEqual(cal["days"]["2026-07-20"]["open_count"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|