dekun
93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
"""期权业务时区与 UTC 墙钟迁移单测."""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from lib.options.options_db import init_options_tables
|
|
from lib.options.options_review_db import init_options_review_tables
|
|
from lib.options.options_review_lib import get_sync_state, upsert_option_history_row
|
|
from lib.options.options_time_lib import (
|
|
migrate_options_utc_wall_clock_to_app,
|
|
ms_to_options_app_str,
|
|
options_wall_clock_to_ms,
|
|
utc_wall_clock_to_app_str,
|
|
)
|
|
|
|
|
|
class OptionsTimeLibTests(unittest.TestCase):
|
|
def test_ms_to_shanghai_wall_clock(self):
|
|
# 2026-09-13 21:41:57 UTC → 2026-09-14 05:41:57 Asia/Shanghai
|
|
ms = int(
|
|
datetime(2026, 9, 13, 21, 41, 57, tzinfo=timezone.utc).timestamp() * 1000
|
|
)
|
|
self.assertEqual(ms_to_options_app_str(ms), "2026-09-14 05:41:57")
|
|
|
|
def test_utc_wall_to_app(self):
|
|
self.assertEqual(
|
|
utc_wall_clock_to_app_str("2026-09-13 21:41:57"),
|
|
"2026-09-14 05:41:57",
|
|
)
|
|
|
|
def test_wall_clock_to_ms_roundtrip(self):
|
|
raw = "2026-09-14 05:41:57"
|
|
ms = options_wall_clock_to_ms(raw)
|
|
self.assertIsNotNone(ms)
|
|
dt = datetime.fromtimestamp(ms / 1000.0, tz=ZoneInfo("Asia/Shanghai"))
|
|
self.assertEqual(dt.strftime("%Y-%m-%d %H:%M:%S"), raw)
|
|
|
|
def test_migrate_once(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
init_options_tables(conn)
|
|
init_options_review_tables(conn)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO options_trades(
|
|
inst_id, underlying, opt_type, sheets, eth_amount, premium_paid, status,
|
|
created_at, closed_at
|
|
) VALUES ('ETH-USD-260917-2525-P', 'ETH', 'P', 1, 0.01, 1, 'closed',
|
|
'2026-09-13 21:41:57', '2026-09-14 02:31:00')
|
|
"""
|
|
)
|
|
upsert_option_history_row(
|
|
conn,
|
|
{
|
|
"history_key": "local_opt:1",
|
|
"inst_id": "ETH-USD-260917-2525-P",
|
|
"underlying": "ETH",
|
|
"opt_type": "P",
|
|
"realized_pnl": 0.5,
|
|
"created_at": "2026-09-13 21:41:57",
|
|
"closed_at": "2026-09-14 02:31:00",
|
|
},
|
|
)
|
|
conn.commit()
|
|
|
|
first = migrate_options_utc_wall_clock_to_app(conn)
|
|
self.assertEqual(first["skipped"], 0)
|
|
self.assertGreaterEqual(first["trades"], 1)
|
|
self.assertGreaterEqual(first["review"], 1)
|
|
|
|
row = conn.execute(
|
|
"SELECT created_at, closed_at FROM options_trades WHERE id=1"
|
|
).fetchone()
|
|
self.assertEqual(row["created_at"], "2026-09-14 05:41:57")
|
|
self.assertEqual(row["closed_at"], "2026-09-14 10:31:00")
|
|
|
|
rev = conn.execute(
|
|
"SELECT opened_at, closed_at FROM options_review_trades WHERE history_key='local_opt:1'"
|
|
).fetchone()
|
|
self.assertEqual(rev["opened_at"], "2026-09-14 05:41:57")
|
|
self.assertEqual(rev["closed_at"], "2026-09-14 10:31:00")
|
|
self.assertEqual(get_sync_state(conn, "options_ts_utc_wall_to_app_done"), "1")
|
|
|
|
second = migrate_options_utc_wall_clock_to_app(conn)
|
|
self.assertEqual(second["skipped"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|