qihuo/market_sessions.py

# 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),
    (13, 30),
    (21, 0),
)


def is_trading_session(now: Optional[datetime] = None) -> bool:
    d = now or datetime.now(TZ)
    if d.tzinfo is None:
        d = d.replace(tzinfo=TZ)
    else:
        d = d.astimezone(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 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
    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):
        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 = now or datetime.now(TZ)
    if d.tzinfo is None:
        d = d.replace(tzinfo=TZ)
    else:
        d = d.astimezone(TZ)
    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) == (13, 30):
        return "午盘开盘"
    if (h, m) == (21, 0):
        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 = d.hour * 60 + d.minute
    if 9 * 60 <= t < 11 * 60 + 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:
        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 = now or datetime.now(TZ)
    if d.tzinfo is None:
        d = d.replace(tzinfo=TZ)
    else:
        d = d.astimezone(TZ)
    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 "非交易时间段",
    }
    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)