dekun
45 lines
1.7 KiB
Python
45 lines
1.7 KiB
Python
"""Verify CTP scheduled connect/disconnect windows."""
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from market_sessions import (
|
|
in_premarket_connect_window,
|
|
in_postmarket_grace_window,
|
|
should_keep_ctp_connected,
|
|
)
|
|
|
|
TZ = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def chk(label, dt, exp_keep, exp_pre=None, exp_post=None):
|
|
keep = should_keep_ctp_connected(dt)
|
|
pre = in_premarket_connect_window(dt)
|
|
post = in_postmarket_grace_window(dt)
|
|
auto = keep
|
|
ok = keep == exp_keep
|
|
if exp_pre is not None:
|
|
ok = ok and pre == exp_pre
|
|
if exp_post is not None:
|
|
ok = ok and post == exp_post
|
|
print(
|
|
f"{'OK' if ok else 'FAIL'} {label} {dt.strftime('%a %H:%M')} "
|
|
f"keep={keep} pre={pre} post={post} auto={auto}"
|
|
)
|
|
return ok
|
|
|
|
|
|
cases = [
|
|
("日盘交易中", datetime(2026, 6, 30, 10, 0, tzinfo=TZ), True, False, False),
|
|
("上午小节休盘前10分", datetime(2026, 6, 30, 10, 20, tzinfo=TZ), True, True, False),
|
|
("日盘盘前28分", datetime(2026, 6, 30, 8, 32, tzinfo=TZ), True, True, False),
|
|
("日盘盘前31分", datetime(2026, 6, 30, 8, 29, tzinfo=TZ), False, False, False),
|
|
("日盘收盘后15分", datetime(2026, 6, 30, 15, 15, tzinfo=TZ), True, False, True),
|
|
("日盘收盘后35分", datetime(2026, 6, 30, 15, 35, tzinfo=TZ), False, False, False),
|
|
("夜盘盘前28分", datetime(2026, 6, 30, 20, 32, tzinfo=TZ), True, True, False),
|
|
("夜盘交易中", datetime(2026, 6, 30, 22, 0, tzinfo=TZ), True, False, False),
|
|
("夜盘收盘后15分", datetime(2026, 7, 1, 2, 45, tzinfo=TZ), True, False, True),
|
|
("夜盘收盘后35分", datetime(2026, 7, 1, 3, 5, tzinfo=TZ), False, False, False),
|
|
]
|
|
failed = sum(0 if chk(*c) else 1 for c in cases)
|
|
print("failed", failed)
|