From 7ce59d2d71193849b27173d1ab7bb708c967757d Mon Sep 17 00:00:00 2001 From: dekun Date: Mon, 29 Jun 2026 10:27:21 +0800 Subject: [PATCH] Detect 10:15-10:30 morning break in trading session clock. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- kline_stream.py | 22 +--------- market_sessions.py | 106 ++++++++++++++++++++++++++++++--------------- 2 files changed, 73 insertions(+), 55 deletions(-) diff --git a/kline_stream.py b/kline_stream.py index 8e192dc..c355a4a 100644 --- a/kline_stream.py +++ b/kline_stream.py @@ -18,6 +18,7 @@ from zoneinfo import ZoneInfo from kline_chart import fetch_market_klines, ths_to_sina_chart_symbol from kline_store import is_cache_fresh, load_meta, ensure_kline_tables +from market_sessions import is_trading_session logger = logging.getLogger(__name__) TZ = ZoneInfo("Asia/Shanghai") @@ -27,27 +28,6 @@ FAST_PERIODS = frozenset({ }) -def is_trading_session() -> bool: - d = datetime.now(TZ) - wd = d.weekday() - if wd == 6: - return False - if wd == 5 and d.hour < 21: - return False - t = d.hour * 60 + d.minute - def in_range(sh: int, sm: int, eh: int, em: int) -> bool: - return t >= sh * 60 + sm and t < eh * 60 + em - if in_range(9, 0, 11, 30): - return True - if in_range(13, 30, 15, 0): - return True - if in_range(21, 0, 24, 0): - return True - if in_range(0, 0, 2, 30): - return True - return False - - def sse_format(event: str, data: dict) -> str: return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" diff --git a/market_sessions.py b/market_sessions.py index 5a013f1..2795f02 100644 --- a/market_sessions.py +++ b/market_sessions.py @@ -15,52 +15,82 @@ 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 is_trading_session(now: Optional[datetime] = None) -> bool: + +def _normalize_dt(now: Optional[datetime] = None) -> datetime: d = now or datetime.now(TZ) if d.tzinfo is None: - d = d.replace(tzinfo=TZ) - else: - d = d.astimezone(TZ) + 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 = d.hour * 60 + d.minute - def in_range(sh: int, sm: int, eh: int, em: int) -> bool: - return t >= sh * 60 + sm and t < eh * 60 + em - if in_range(9, 0, 11, 30): + 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_range(13, 30, 15, 0): - return True - if in_range(21, 0, 24, 0): - return True - if in_range(0, 0, 2, 30): + 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 = now or datetime.now(TZ) - if d.tzinfo is None: - d = d.replace(tzinfo=TZ) - else: - d = d.astimezone(TZ) - t = d.hour * 60 + d.minute + 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) == (9, 0) or (hour, minute) == (13, 30): + if (hour, minute) in ((9, 0), (10, 30), (13, 30)): return wd < 5 if (hour, minute) == (21, 0): if wd < 5: @@ -95,11 +125,7 @@ def iter_session_starts( def minutes_until_next_session(now: Optional[datetime] = None) -> Optional[float]: - d = now or datetime.now(TZ) - if d.tzinfo is None: - d = d.replace(tzinfo=TZ) - else: - d = d.astimezone(TZ) + d = _normalize_dt(now) starts = iter_session_starts(d, hours_ahead=48) if not starts: return None @@ -110,6 +136,8 @@ 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): @@ -117,6 +145,16 @@ def _session_open_label(dt: datetime) -> str: 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) @@ -142,12 +180,16 @@ def _night_close_dt(d: datetime) -> datetime: def _current_break_close(d: datetime) -> tuple[Optional[datetime], Optional[datetime], Optional[str], Optional[str]]: """当前交易段内的休盘/收盘时刻与标签。""" - t = d.hour * 60 + d.minute - if 9 * 60 <= t < 11 * 60 + 30: + 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 13 * 60 + 30 <= t < 15 * 60: + 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: @@ -158,17 +200,13 @@ def _current_break_close(d: datetime) -> tuple[Optional[datetime], Optional[date def trading_session_clock(now: Optional[datetime] = None) -> dict: """顶栏展示：当前时间、交易状态、距开盘/休盘/收盘倒计时。""" - d = now or datetime.now(TZ) - if d.tzinfo is None: - d = d.replace(tzinfo=TZ) - else: - d = d.astimezone(TZ) + 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": "交易时间段" if in_sess else "非交易时间段", + "status_label": _session_status_label(d, in_sess), } if not in_sess: starts = iter_session_starts(d, hours_ahead=72)