105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
|
# -*- 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 # 防呆: 连续找不到交易日时的扫描上限
|
||
|
|
|
||
|
|
|
||
|
|
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:
|
||
|
|
return bool(_is_workday(dd))
|
||
|
|
except Exception:
|
||
|
|
return True # 年份超范围 → 按工作日处理 (degraded)
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def calendar_degraded() -> bool:
|
||
|
|
"""True = 未装 chinesecalendar, 节假日不可辨 (页面与日报应提示)。"""
|
||
|
|
return not _HAS_CAL
|
||
|
|
|
||
|
|
|
||
|
|
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
|