"""业务窗时钟: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)