# Copyright (c) 2025-2026 马建军. All rights reserved. # 专有软件 — 未经授权禁止复制、传播、转售。 # 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。 # 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md """国内期货交易时段与盘前连接窗口。""" from __future__ import annotations from datetime import datetime, timedelta from typing import Optional from zoneinfo import ZoneInfo TZ = ZoneInfo("Asia/Shanghai") # 各交易段开盘时刻 (时, 分) SESSION_OPENS = ( (9, 0), (10, 30), # 上午小节休息后续盘 (13, 30), (21, 0), ) # 日盘各连续交易段 (start_h, start_m, end_h, end_m)，左闭右开 _DAY_SEGMENTS = ( (9, 0, 10, 15), (10, 30, 11, 30), (13, 30, 15, 0), ) def _normalize_dt(now: Optional[datetime] = None) -> datetime: d = now or datetime.now(TZ) if d.tzinfo is None: return d.replace(tzinfo=TZ) return d.astimezone(TZ) def _minutes_of_day(d: datetime) -> int: return d.hour * 60 + d.minute def _in_time_range(t: int, sh: int, sm: int, eh: int, em: int) -> bool: return t >= sh * 60 + sm and t < eh * 60 + em def is_trading_session(now: Optional[datetime] = None) -> bool: d = _normalize_dt(now) wd = d.weekday() if wd == 6: return False if wd == 5 and d.hour < 21: return False t = _minutes_of_day(d) for sh, sm, eh, em in _DAY_SEGMENTS: if _in_time_range(t, sh, sm, eh, em): return True if _in_time_range(t, 21, 0, 24, 0): return True if _in_time_range(t, 0, 0, 2, 30): return True return False def is_morning_break(now: Optional[datetime] = None) -> bool: """10:15–10:30 上午小节休息。""" d = _normalize_dt(now) if d.weekday() >= 5: return False t = _minutes_of_day(d) return _in_time_range(t, 10, 15, 10, 30) def is_lunch_break(now: Optional[datetime] = None) -> bool: """11:30–13:30 午间休盘。""" d = _normalize_dt(now) if d.weekday() >= 5: return False t = _minutes_of_day(d) return _in_time_range(t, 11, 30, 13, 30) def is_night_trading_session(now: Optional[datetime] = None) -> bool: """当前是否处于夜盘时段（21:00–02:30，且整体仍在交易时段内）。""" if not is_trading_session(now): return False d = _normalize_dt(now) t = _minutes_of_day(d) return t >= 21 * 60 or t < 2 * 60 + 30 def _session_open_allowed(day: datetime, hour: int, minute: int) -> bool: wd = day.weekday() if (hour, minute) in ((9, 0), (10, 30), (13, 30)): return wd < 5 if (hour, minute) == (21, 0): if wd < 5: return True return wd == 5 return False def iter_session_starts( start: datetime, *, hours_ahead: int = 36, ) -> list[datetime]: """列出 start 之后若干小时内的各段开盘时刻。""" if start.tzinfo is None: start = start.replace(tzinfo=TZ) else: start = start.astimezone(TZ) end = start + timedelta(hours=hours_ahead) out: list[datetime] = [] day = start.replace(hour=0, minute=0, second=0, microsecond=0) while day <= end: for h, m in SESSION_OPENS: if not _session_open_allowed(day, h, m): continue dt = day.replace(hour=h, minute=m) if dt > start and dt <= end: out.append(dt) day += timedelta(days=1) out.sort() return out def minutes_until_next_session(now: Optional[datetime] = None) -> Optional[float]: d = _normalize_dt(now) starts = iter_session_starts(d, hours_ahead=48) if not starts: return None return (starts[0] - d).total_seconds() / 60.0 def _session_open_label(dt: datetime) -> str: h, m = dt.hour, dt.minute if (h, m) == (9, 0): return "日盘开盘" if (h, m) == (10, 30): return "上午续盘" if (h, m) == (13, 30): return "午盘开盘" if (h, m) == (21, 0): return "夜盘开盘" return "开盘" def _session_status_label(d: datetime, in_sess: bool) -> str: if in_sess: return "交易时间段" if is_morning_break(d): return "上午休盘" if is_lunch_break(d): return "午间休盘" return "非交易时间段" def _fmt_countdown(seconds: int) -> str: s = max(0, int(seconds)) h, rem = divmod(s, 3600) m, sec = divmod(rem, 60) if h > 0: return f"{h}小时{m:02d}分{sec:02d}秒" if m > 0: return f"{m}分{sec:02d}秒" return f"{sec}秒" def _day_close_dt(d: datetime) -> datetime: return d.replace(hour=15, minute=0, second=0, microsecond=0) def _night_close_dt(d: datetime) -> datetime: t = d.hour * 60 + d.minute if t >= 21 * 60: nxt = (d + timedelta(days=1)).replace(hour=2, minute=30, second=0, microsecond=0) return nxt return d.replace(hour=2, minute=30, second=0, microsecond=0) def _current_break_close(d: datetime) -> tuple[Optional[datetime], Optional[datetime], Optional[str], Optional[str]]: """当前交易段内的休盘/收盘时刻与标签。""" t = _minutes_of_day(d) if _in_time_range(t, 9, 0, 10, 15): br = d.replace(hour=10, minute=15, second=0, microsecond=0) cl = _day_close_dt(d) return br, cl, "上午休盘", "日盘收盘" if _in_time_range(t, 10, 30, 11, 30): br = d.replace(hour=11, minute=30, second=0, microsecond=0) cl = _day_close_dt(d) return br, cl, "午间休盘", "日盘收盘" if _in_time_range(t, 13, 30, 15, 0): cl = _day_close_dt(d) return None, cl, None, "日盘收盘" if t >= 21 * 60 or t < 2 * 60 + 30: cl = _night_close_dt(d) return None, cl, None, "夜盘收盘" return None, None, None, None def trading_session_clock(now: Optional[datetime] = None) -> dict: """顶栏展示：当前时间、交易状态、距开盘/休盘/收盘倒计时。""" d = _normalize_dt(now) in_sess = is_trading_session(d) out = { "now": d.strftime("%Y-%m-%d %H:%M:%S"), "now_time": d.strftime("%m-%d %H:%M:%S"), "in_session": in_sess, "status_label": _session_status_label(d, in_sess), } if not in_sess: starts = iter_session_starts(d, hours_ahead=72) if starts: nxt = starts[0] secs = int(max(0, (nxt - d).total_seconds())) out["next_open_at"] = nxt.strftime("%m-%d %H:%M") out["next_open_label"] = _session_open_label(nxt) out["secs_to_open"] = secs out["countdown_open"] = _fmt_countdown(secs) return out br, cl, br_label, cl_label = _current_break_close(d) if br and br > d: secs = int((br - d).total_seconds()) out["break_at"] = br.strftime("%H:%M") out["break_label"] = br_label or "休盘" out["secs_to_break"] = secs out["countdown_break"] = _fmt_countdown(secs) if cl and cl > d: secs = int((cl - d).total_seconds()) out["close_at"] = cl.strftime("%H:%M") out["close_label"] = cl_label or "收盘" out["secs_to_close"] = secs out["countdown_close"] = _fmt_countdown(secs) return out def in_premarket_connect_window( now: Optional[datetime] = None, *, minutes_before: int = 30, ) -> bool: """距下一段开盘 <= minutes_before 分钟，且当前尚未进入交易时段。""" if is_trading_session(now): return False mins = minutes_until_next_session(now) if mins is None: return False return 0 < mins <= float(minutes_before)