自动交易程序增加节日过滤规则非交易日跳过行情检测一、实际应用场景描述在 A 股自动交易系统的实际运行中交易日历Trading Calendar 管理是最基础却最容易被忽视的环节。一个没有节日过滤的交易程序会在非交易日周末、法定节假日、调休补班日 不断尝试获取行情、计算信号、甚至下单导致一系列严重问题。典型场景场景 问题表现国庆长假后第一个交易日 程序在假期期间持续运行积累了 7 天的待处理信号开盘瞬间集中下单造成价格冲击周末运行策略 周六早上定时任务触发API 返回空数据或上周五的缓存数据策略基于过时信息计算信号春节假期 9 天假期中程序每天尝试连接交易接口触发风控告警账户被标记为异常登录调休工作日如春节前补班 程序误判为假期跳过实际是交易日错失当天交易机会境外市场节假日不同步 港股通标的在 A 股休市时仍可交易过滤规则需区分市场二、引入痛点痛点 具体表现 无效 API 调用 非交易日调用行情接口消耗配额、产生空结果部分 API 返回错误导致程序崩溃 信号失真 基于非交易日数据如前一日收盘价计算的信号在下一个交易日开盘时已经失效 集中下单冲击 假期积累的待处理信号在开盘瞬间释放大资金造成显著价格冲击 风控告警 券商系统检测到非交易时段频繁登录/下单尝试可能冻结账户 调休日误判 简单用weekday() 判断无法处理周末调休上班等特殊情况 多市场日历差异 A 股、港股、美股交易日历不同需分别处理 夜盘/盘前竞价 部分策略需要在盘前竞价阶段运行交易日历需精确到时段三、核心逻辑讲解3.1 A 股交易日历规则A 股非交易日类型┌─────────────────────────────────────────────────────┐│ 类型 │ 示例 │├───────────────────┼───────────────────────────────┤│ 周末 │ 周六、周日 ││ 法定节假日 │ 元旦、春节、清明、五一、端午、 ││ │ 中秋、国庆 ││ 调休补班日 │ 周末上班但股市不开市 ││ 临时休市 │ 极端天气、重大事件等 ││ 日内非交易时段 │ 9:30 前 / 11:30~13:00 / 15:00后│└─────────────────────────────────────────────────────┘3.2 节日过滤核心设计┌──────────────────────────────────────────────────────────┐│ 自动交易程序交易日历过滤模块 │├──────────────────────────────────────────────────────────┤│ ││ 程序启动 / 定时触发 ││ │ ││ ▼ ││ ┌──────────────────────────────────┐ ││ │ ★ 第一步获取交易日历 │ ││ │ 调用 exchange_cal.get(A股) │ ││ │ 返回今日是否为交易日 │ ││ └──────────────────────────────────┘ ││ │ ││ ▼ 是交易日 ││ ┌────┴────┐ ││ │ │ ││ 是┘ └否 ││ │ │ ││ ▼ ▼ ││ 执行完整 记录日志: ││ 行情检测 YYYY-MM-DD 非交易日跳过 ││ 信号计算 计算下次交易日 ││ 下单逻辑 休眠至下一交易日开盘前 ││ │└──────────────────────────────────────────────────────────┘3.3 交易日历数据来源来源 优点 缺点交易所官方公告 最权威 需爬取/解析 PDF第三方金融数据库 接口友好、数据完整 依赖外部服务本地静态文件 零延迟、无配额限制 需每年更新在线 API推荐 自动更新、支持多市场 需处理网络异常3.4 核心判断逻辑# 伪代码def should_trade_today(date, marketA股):返回 (是否交易, 原因)# 1. 周末检查快速过滤 5/7 的非交易日if date.weekday() in (5, 6): # 周六、周日return False, weekend# 2. 查询交易日历cal get_trading_calendar(market, yeardate.year)if not cal.is_trading_day(date):return False, holiday# 3. 检查日内交易时段如需要if not is_within_trading_hours(now()):return False, after_hoursreturn True, trading_day四、项目结构trading_calendar_filter/├── README.md├── requirements.txt├── config.yaml # 全局配置含交易日历来源├── data/│ └── trading_calendar.csv # 交易日历数据静态备份├── src/│ ├── calendar_provider.py # ★ 交易日历数据提供者│ ├── trading_day_checker.py # ★ 交易日判断器│ ├── trading_engine.py # 交易引擎集成日历过滤│ ├── backtester.py # 回测引擎含日历感知│ └── visualizer.py # 可视化工具├── main.py # 主入口└── update_calendar.py # 更新交易日历脚本五、完整代码模块化 清晰注释requirements.txtpandas1.5numpy1.21pyyaml6.0matplotlib3.5requests2.28config.yaml# 交易日历与交易时段配置# ★ 交易日历数据源calendar:# 数据来源local_file / online_api / hybrid推荐source: hybridlocal_file: data/trading_calendar.csv# 在线 API 地址示例实际替换为真实接口api_url: https://api.example.com/trading_calendarapi_key: # 支持的市场markets:- name: A股code: CNtimezone: Asia/Shanghai- name: 港股code: HKtimezone: Asia/Hong_Kong- name: 美股code: UStimezone: America/New_York# ★ 交易时段配置trading_hours:A股:pre_market: 09:15-09:25 # 竞价morning: 09:30-11:30 # 上午afternoon: 13:00-15:00 # 下午# 是否包含盘前竞价时段include_pre_market: false# 日内检查粒度秒check_interval: 60# 策略参数strategy:initial_capital: 1000000max_positions: 5take_profit_pct: 0.08stop_loss_pct: -0.05# 日志logging:level: INFOfile: logs/trading.logsrc/calendar_provider.py★ 核心模块calendar_provider.py★ 交易日历数据提供者职责1. 从本地文件/在线 API 获取交易日历2. 缓存到本地减少 API 调用3. 支持多市场A 股、港股、美股import pandas as pdimport numpy as npfrom pathlib import Pathfrom typing import Optional, Set, Dictimport jsonimport loggingfrom datetime import datelogging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s)logger logging.getLogger(__name__)class TradingCalendarProvider:交易日历数据提供者支持三种模式- local_file: 从本地 CSV 读取离线可用- online_api: 从在线 API 获取自动更新- hybrid: 优先本地缺失时 fallback 到 API推荐def __init__(self,source: str hybrid,local_file: str data/trading_calendar.csv,api_url: str ,cache_dir: str data/.cache):参数:source: 数据源模式 (local_file / online_api / hybrid)local_file: 本地交易日历文件路径api_url: 在线 API 地址cache_dir: 缓存目录self.source sourceself.local_file Path(local_file)self.api_url api_urlself.cache_dir Path(cache_dir)self.cache_dir.mkdir(parentsTrue, exist_okTrue)# 内存缓存: {market: {date_str: is_trading}}self._memory_cache: Dict[str, Dict[str, bool]] {}logger.info(f交易日历提供者初始化: 模式{source})def get_calendar(self,market: str A股,year: int None) - pd.DataFrame:获取指定市场、指定年份的交易日历返回 DataFrame:date is_trading holiday_name2024-01-01 False 元旦2024-01-02 True NaNyear year or date.today().yearcache_key f{market}_{year}# 尝试从本地加载if self.source in (local_file, hybrid):df self._load_local(market, year)if df is not None:logger.debug(f从本地加载: {cache_key})return df# Fallback 到 APIif self.source in (online_api, hybrid):df self._fetch_online(market, year)if df is not None:self._save_local(df, market, year)logger.info(f从 API 获取并缓存: {cache_key})return df# 都失败 → 用简易规则生成仅周末 主要节假日logger.warning(f无法获取交易日历使用简易规则生成)return self._generate_simple_calendar(year)def is_trading_day(self,date: date,market: str A股) - bool:★ 核心方法判断某日期是否为交易日参数:date: 要查询的日期market: 市场名称返回:bool: True 交易日date_str date.isoformat()cache_key f{market}# 检查内存缓存if cache_key in self._memory_cache:if date_str in self._memory_cache[cache_key]:return self._memory_cache[cache_key][date_str]# 获取日历cal self.get_calendar(market, date.year)# 查询if date_str in cal.index:result bool(cal.loc[date_str, is_trading])else:# 不在日历中 → 用简易规则result self._simple_is_trading(date)# 写入缓存if cache_key not in self._memory_cache:self._memory_cache[cache_key] {}self._memory_cache[cache_key][date_str] resultreturn resultdef get_next_trading_day(self,date: date,market: str A股) - date:获取下一个交易日from datetime import timedeltacandidate date timedelta(days1)max_search 30 # 最多往前找 30 天for i in range(max_search):if self.is_trading_day(candidate, market):return candidatecandidate timedelta(days1)# 找不到 → 返回 30 天后的日期logger.warning(f在 30 天内未找到 {market} 的下一个交易日)return date timedelta(days30)def get_trading_days_between(self,start: date,end: date,market: str A股) - pd.DatetimeIndex:获取两个日期之间的所有交易日# 确保日历覆盖区间cal self.get_calendar(market, start.year)if end.year ! start.year:cal_next self.get_calendar(market, end.year)cal pd.concat([cal, cal_next])mask (cal.index start.isoformat()) (cal.index end.isoformat())trading cal[mask cal[is_trading]]return pd.DatetimeIndex(pd.to_datetime(trading.index))def _load_local(self,market: str,year: int) - Optional[pd.DataFrame]:从本地文件加载path self.cache_dir / f{market}_{year}.csvif not path.exists() and self.local_file.exists():path self.local_fileif not path.exists():return Nonetry:df pd.read_csv(path, parse_dates[date]).set_index(date)return dfexcept Exception as e:logger.error(f加载本地交易日历失败: {e})return Nonedef _fetch_online(self,market: str,year: int) - Optional[pd.DataFrame]:从在线 API 获取if not self.api_url:return Nonetry:import requestsresp requests.get(f{self.api_url}/calendar,params{market: market, year: year},timeout10)if resp.status_code 200:data resp.json()records data.get(data, [])df pd.DataFrame([{date: r[date],is_trading: r[is_trading],holiday_name: r.get(holiday, )} for r in records]).set_index(date)return dfexcept Exception as e:logger.error(f在线获取交易日历失败: {e})return Nonedef _save_local(self, df: pd.DataFrame, market: str, year: int):保存到本地缓存path self.cache_dir / f{market}_{year}.csvdf.to_csv(path)def _generate_simple_calendar(self, year: int) - pd.DataFrame:简易交易日历生成器仅处理- 周末周六日非交易- 主要法定节假日元旦、春节、清明、五一、端午、中秋、国庆注意不处理调休补班精度有限仅作降级方案import holidaysfrom datetime import date, timedeltastart date(year, 1, 1)end date(year, 12, 31)dates []current startwhile current end:dates.append(current)current timedelta(days1)records []for d in dates:# 周末if d.weekday() in (5, 6):is_td Falsehname weekendelse:# 法定节假日使用 holidays 库is_td Truehname try:cn_holidays holidays.China(yearsyear)if d in cn_holidays:is_td Falsehname str(cn_holidays[d])except:pass# 手动补充主要节假日降级方案if is_td and hname :is_td, hname self._check_major_holidays(d)records.append({date: d.isoformat(),is_trading: is_td,holiday_name: hname})return pd.DataFrame(records).set_index(date)def _check_major_holidays(self, d: date) - tuple[bool, str]:检查是否为主要法定节假日降级方案# 元旦if d.month 1 and d.day 3:return False, 元旦# 五一if d.month 5 and 1 d.day 5:return False, 劳动节# 国庆if d.month 10 and 1 d.day 7:return False, 国庆节# 中秋简化农历八月十五附近实际需要农历转换# 此处省略农历计算实际项目中应使用 lunardate 库return True, def _simple_is_trading(self, d: date) - bool:简易判断内存缓存未命中时if d.weekday() in (5, 6):return Falsereturn True # 简化非周末即视为交易日def print_calendar_summary(self, market: str A股, year: int None):打印交易日历摘要cal self.get_calendar(market, year)total len(cal)trading cal[is_trading].sum()non_trading total - tradingprint(f\n{*60})print(f 交易日历摘要: {market} {year or 最新})print(f{*60})print(f 总天数: {total})print(f 交易日: {trading} ({trading/total*100:.1f}%))print(f 非交易日: {non_trading} ({non_trading/total*100:.1f}%))# 列出节假日holidays cal[~cal[is_trading] (cal[holiday_name] ! )]if len(holidays) 0:print(f\n 节假日明细前 10 条:)for idx, row in holidays.head(10).iterrows():print(f {idx}: {row[holiday_name]})if len(holidays) 10:print(f ... 共 {len(holidays)} 个非交易日)print(f{*60}\n)src/trading_day_checker.py★ 核心模块trading_day_checker.py★ 交易日判断器集成日内时段检查在交易日历基础上增加1. 日内交易时段判断开盘前/交易中/收盘后2. 竞价时段处理3. 连续交易时段如需要import pandas as pdimport numpy as npfrom datetime import datetime, time, date, timedeltafrom typing import Optional, Tuplefrom enum import Enumimport logginglogging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s)logger logging.getLogger(__name__)class MarketSession(Enum):市场交易时段PRE_MARKET pre_market # 盘前竞价MORNING morning # 上午交易MIDDAY_BREAK midday_break # 午间休市AFTERNOON afternoon # 下午交易AFTER_HOURS after_hours # 盘后CLOSED closed # 休市class TradingDayChecker:★ 交易日 交易时段检查器集成功能1. 交易日历查询委托给 CalendarProvider2. 日内交易时段判断3. 距离下次开盘倒计时def __init__(self,calendar_provider: TradingCalendarProvider,market: str A股,trading_hours: Optional[Dict] None):参数:calendar_provider: 交易日历提供者market: 市场名称trading_hours: 交易时段配置如:{pre_market: 09:15-09:25,morning: 09:30-11:30,afternoon: 13:00-15:00,include_pre_market: False}self.provider calendar_providerself.market market# 解析交易时段self.sessions self._parse_trading_hours(trading_hours or {})self.include_pre_market trading_hours.get(include_pre_market, False)logger.info(f交易日检查器初始化: 市场{market})def _parse_trading_hours(self, config: Dict) - list:解析交易时段配置为时间区间列表sessions []if config.get(include_pre_market) and pre_market in config:start, end config[pre_market].split(-)sessions.append((MarketSession.PRE_MARKET, self._parse_time(start), self._parse_time(end)))if morning in config:start, end config[morning].split(-)sessions.append((MarketSession.MORNING, self._parse_time(start), self._parse_time(end)))if afternoon in config:start, end config[afternoon].split(-)sessions.append((MarketSession.AFTERNOON, self._parse_time(start), self._parse_time(end)))return sessionsdef _parse_time(self, t_str: str) - time:解析 HH:MM 为 time 对象h, m map(int, t_str.split(:))return time(h, m)def check_now(self, now: Optional[datetime] None) - Tuple[bool, MarketSession, str]:★ 核心方法检查当前时刻是否应该运行交易逻辑参数:now: 当前时间默认取系统时间返回:Tuple[是否交易日, 当前时段, 说明信息]now now or datetime.now()today now.date()# 第一步检查是否为交易日 if not self.provider.is_trading_day(today, self.market):# 非交易日 → 计算距离下一个交易日还有多久next_td self.provider.get_next_trading_day(today, self.market)days_until (next_td - today).daysreason f非交易日距离下次交易还有 {days_until} 天 ({next_td})logger.debug(reason)return False, MarketSession.CLOSED, reason# 第二步检查日内交易时段 current_time now.time()for session, start, end in self.sessions:if start current_time end:reason f交易时段: {session.value}return True, session, reason# 不在任何交易时段内# 判断是盘前还是盘后morning_start self.sessions[0][1] if self.sessions else time(9, 30)if current_time morning_start:reason 盘前等待开盘else:reason 盘后/午间休市return False, MarketSession.AFTER_HOURS, reasondef should_run_strategy(self, now: Optional[datetime] None) - Tuple[bool, str]:★ 策略层调用入口判断当前是否应该执行行情检测和交易逻辑返回:(should_run, reason)is_trading, session, reason self.check_now(now)if not is_trading:return False, reason# 如果策略不包含盘前竞价需要额外检查if not self.include_pre_market and session MarketSession.PRE_MARKET:return False, 竞价时段策略未启用竞价return True, reasondef get_seconds_to_next_open(self, now: Optional[datetime] None) - int:计算距离下次开盘还有多少秒用于非交易时段让程序休眠避免空轮询now now or datetime.now()today now.date()current_time now.time()# 情况 1今天就是交易日但还没开盘if self.provider.is_trading_day(today, self.market):morning_start self.sessions[0][1] if self.sessions else time(9, 30)if current_time morning_start:next_open datetime.combine(today, morning_start)return max(0, int((next_open - now).total_seconds()))# 情况 2今天不是交易日或已经收盘next_td self.provider.get_next_trading_day(today, self.market)morning_start self.sessions[0][1] if self.sessions else time(9, 30)next_open datetime.combine(next_td, morning_start)return max(0, int((next_open - now).total_seconds()))def print_today_status(self):打印今日交易状态today date.today()is_td self.provider.is_trading_day(today, self.market)print(f\n{*50})print(f 今日交易状态: {today})print(f{*50})print(f 是否交易日: {✅ 是 if is_td else ❌ 否})if is_本文代码仅供学习与技术交流不构成任何投资建议股市有风险入市需谨慎利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛