dekun
73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
"""日历日分组键；可选周末跳过开仓（上海时区）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import Any
|
||
from zoneinfo import ZoneInfo
|
||
|
||
_SH = ZoneInfo("Asia/Shanghai")
|
||
|
||
|
||
def now_sh(now: datetime | None = None) -> datetime:
|
||
return (now or datetime.now(tz=_SH)).astimezone(_SH)
|
||
|
||
|
||
def window_key(now: datetime | None = None) -> str:
|
||
"""组号日期键：日历日 YYYYMMDD。"""
|
||
return now_sh(now).strftime("%Y%m%d")
|
||
|
||
|
||
def is_weekend(now: datetime | None = None) -> bool:
|
||
"""上海时区：周六=5、周日=6。"""
|
||
return now_sh(now).weekday() >= 5
|
||
|
||
|
||
def can_open_new(
|
||
now: datetime | None = None,
|
||
*,
|
||
skip_weekends: bool = True,
|
||
open_hhmm: str = "16:00",
|
||
stop_hhmm: str = "08:00",
|
||
) -> bool:
|
||
"""是否允许新开仓。开仓窗已取消；可选跳过周六日。持仓平仓不受此限制。"""
|
||
_ = open_hhmm, stop_hhmm
|
||
if skip_weekends and is_weekend(now):
|
||
return False
|
||
return True
|
||
|
||
|
||
def group_date_ymd(now: datetime | None = None) -> str:
|
||
return window_key(now)
|
||
|
||
|
||
def used_expiry_ymds_for_day(db: Any, now: datetime | None = None) -> set[str]:
|
||
"""
|
||
上海日历日已开过的期权到期日（groups.expiry_ymd，YYMMDD）。
|
||
按当日组号 G-{YYYYMMDD}-% 统计；含已平仓，用于「同到期一天只开一次」。
|
||
"""
|
||
wkey = window_key(now)
|
||
rows = db.fetchall(
|
||
"SELECT DISTINCT expiry_ymd FROM groups WHERE group_id LIKE ?",
|
||
(f"G-{wkey}-%",),
|
||
)
|
||
out: set[str] = set()
|
||
for r in rows or []:
|
||
y = str(r["expiry_ymd"] or "").strip()
|
||
if y:
|
||
out.add(y)
|
||
return out
|
||
|
||
|
||
def expiry_blocked_by_one_per_day(
|
||
expiry_ymd: str | None,
|
||
used: set[str],
|
||
*,
|
||
enabled: bool = True,
|
||
) -> bool:
|
||
"""开启时：候选到期已在当日用过则拦截。"""
|
||
if not enabled:
|
||
return False
|
||
y = str(expiry_ymd or "").strip()
|
||
return bool(y and y in used)
|