tradingSystem/app/core/tradedays.py

# -*- coding: utf-8 -*-
"""
交易日历 (A股) —— 调度守卫与执行窗口计算共用
==============================================
设计 §10 全部调度任务带交易日守卫; §3.1 任务命令的执行窗口以「交易日」计数。

口径: 工作日且非法定节假日 = 交易日 (chinesecalendar 提供节假日表)。
chinesecalendar 未安装或年份超出其数据范围时, 退化为「周一至周五」并置 degraded 标记
—— 宁可多跑一次带守卫的任务, 也不静默跳过 (调度侧另有幂等)。
"""
from __future__ import annotations

from datetime import date, datetime, timedelta

try:  # pragma: no cover - 依赖可用性分支
    from chinese_calendar import is_workday as _is_workday
    _HAS_CAL = True
except Exception:  # pragma: no cover
    _is_workday = None
    _HAS_CAL = False

MAX_SCAN_DAYS = 400  # 防呆: 连续找不到交易日时的扫描上限

# 按年份缓存「chinesecalendar 有没有这一年的节假日数据」的探测结果。
# 库装了但版本旧 (没有当年数据) 时, is_workday 对当年任何日期都抛异常 —— 这是
# 最常见的降级场景 (每年年初库没升级), 原来 degraded 标记只看"装没装", 这种情况
# 全年法定节假日都被当交易日而页面显示一切正常 (2026-08-28 审查修)。
_YEAR_OK: dict = {}


def _year_supported(year: int) -> bool:
    if not _HAS_CAL:
        return False
    hit = _YEAR_OK.get(year)
    if hit is not None:
        return hit
    try:
        _is_workday(date(year, 1, 1))
        ok = True
    except Exception:
        ok = False
    _YEAR_OK[year] = ok
    return ok


def _as_date(d) -> date:
    if d is None:
        return datetime.now().date()
    if isinstance(d, datetime):
        return d.date()
    if isinstance(d, date):
        return d
    s = str(d).strip()
    if len(s) == 8 and s.isdigit():
        return date(int(s[:4]), int(s[4:6]), int(s[6:]))
    return datetime.strptime(s[:10], "%Y-%m-%d").date()


def is_trade_day(d=None) -> bool:
    dd = _as_date(d)
    if dd.weekday() >= 5:
        return False
    if _HAS_CAL:
        try:
            ok = bool(_is_workday(dd))
            _YEAR_OK.setdefault(dd.year, True)
            return ok
        except Exception:
            _YEAR_OK[dd.year] = False
            return True   # 年份超范围 → 按工作日处理 (degraded, 页面会亮标记)
    return True


def calendar_degraded(d=None) -> bool:
    """True = 节假日不可辨: 未装 chinesecalendar, **或它没有当年的数据** (库版本旧)。
    页面与日报据此提示。默认探测今天所在年份。"""
    if not _HAS_CAL:
        return True
    return not _year_supported(_as_date(d).year)


def next_trade_day(d=None, n: int = 1) -> date:
    """d 之后的第 n 个交易日 (n≥1); n=0 返回 d 当天 (不判是否交易日)。"""
    dd = _as_date(d)
    if n <= 0:
        return dd
    cnt, cur, guard = 0, dd, 0
    while cnt < n and guard < MAX_SCAN_DAYS:
        cur += timedelta(days=1)
        guard += 1
        if is_trade_day(cur):
            cnt += 1
    return cur


def window_deadline(start=None, window_tdays: int = 3) -> date:
    """执行窗口截止日 = 起始日(含, 若为交易日) 起的第 window_tdays 个交易日。

    例: 周一下达、窗口 3 → 周三 (周一/二/三 计 3 个交易日)。
    """
    n = max(1, int(window_tdays or 1))
    cur = _as_date(start)
    cnt = 1 if is_trade_day(cur) else 0
    guard = 0
    while cnt < n and guard < MAX_SCAN_DAYS:
        cur += timedelta(days=1)
        guard += 1
        if is_trade_day(cur):
            cnt += 1
    if cnt == 0:  # 起始日非交易日且窗口未推进 → 取下一个交易日
        return next_trade_day(cur, 1)
    return cur


def trade_days_left(deadline, today=None) -> int:
    """距截止日剩余交易日数 (含今日, 已过期返回 0) —— 择时执行器分配每日配额用。"""
    dl, cur = _as_date(deadline), _as_date(today)
    if cur > dl:
        return 0
    cnt, guard = 0, 0
    while cur <= dl and guard < MAX_SCAN_DAYS:
        if is_trade_day(cur):
            cnt += 1
        cur += timedelta(days=1)
        guard += 1
    return cnt


def ymd(d=None) -> int:
    dd = _as_date(d)
    return dd.year * 10000 + dd.month * 100 + dd.day