dekun
59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""业务窗时钟：16:00 开 → 08:00 停开；轮次与休息。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
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 parse_hhmm(s: str) -> tuple[int, int]:
|
|
parts = (s or "16:00").strip().split(":")
|
|
return int(parts[0]), int(parts[1]) if len(parts) > 1 else 0
|
|
|
|
|
|
def window_key(now: datetime | None = None) -> str:
|
|
"""
|
|
业务窗键：若当前 >= 当日 16:00，窗从今日 16:00 起，键=今日日期；
|
|
若 < 16:00，仍可能属于「昨日起的窗」（到今日 08:00），键=昨日。
|
|
"""
|
|
n = now_sh(now)
|
|
open_h, open_m = 16, 0
|
|
stop_h, stop_m = 8, 0
|
|
today_open = n.replace(hour=open_h, minute=open_m, second=0, microsecond=0)
|
|
today_stop = n.replace(hour=stop_h, minute=stop_m, second=0, microsecond=0)
|
|
if n >= today_open:
|
|
return n.strftime("%Y%m%d")
|
|
if n < today_stop:
|
|
# 仍在昨 16:00 开启的窗内
|
|
return (n.date() - timedelta(days=1)).strftime("%Y%m%d")
|
|
# 08:00～16:00：不在开仓窗，键用「即将开始」的今日窗
|
|
return n.strftime("%Y%m%d")
|
|
|
|
|
|
def can_open_new(
|
|
now: datetime | None = None,
|
|
*,
|
|
open_hhmm: str = "16:00",
|
|
stop_hhmm: str = "08:00",
|
|
) -> bool:
|
|
n = now_sh(now)
|
|
oh, om = parse_hhmm(open_hhmm)
|
|
sh, sm = parse_hhmm(stop_hhmm)
|
|
today_open = n.replace(hour=oh, minute=om, second=0, microsecond=0)
|
|
today_stop = n.replace(hour=sh, minute=sm, second=0, microsecond=0)
|
|
if n >= today_open:
|
|
return True
|
|
if n < today_stop:
|
|
return True
|
|
return False
|
|
|
|
|
|
def group_date_ymd(now: datetime | None = None) -> str:
|
|
return window_key(now)
|