dekun
7ce59d2d71
Split day session into 9:00-10:15 and 10:30-11:30; show 上午休盘 status and countdown to 10:30 reopen. Co-authored-by: Cursor <cursoragent@cursor.com>
249 lines
7.6 KiB
Python
249 lines
7.6 KiB
Python
# 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)
|