自动交易程序：增加节日过滤规则，非交易日跳过行情检测
一、实际应用场景描述
在 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, market='A股'):
"""返回 (是否交易, 原因)"""
# 1. 周末检查（快速过滤 5/7 的非交易日）
if date.weekday() in (5, 6): # 周六、周日
return False, "weekend"
# 2. 查询交易日历
cal = get_trading_calendar(market, year=date.year)
if not cal.is_trading_day(date):
return False, "holiday"
# 3. 检查日内交易时段（如需要）
if not is_within_trading_hours(now()):
return False, "after_hours"
return 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.txt"
pandas>=1.5
numpy>=1.21
pyyaml>=6.0
matplotlib>=3.5
requests>=2.28
"config.yaml"
# 交易日历与交易时段配置
# ★ 交易日历数据源
calendar:
# 数据来源：local_file / online_api / hybrid（推荐）
source: "hybrid"
local_file: "data/trading_calendar.csv"
# 在线 API 地址（示例，实际替换为真实接口）
api_url: "https://api.example.com/trading_calendar"
api_key: ""
# 支持的市场
markets:
- name: "A股"
code: "CN"
timezone: "Asia/Shanghai"
- name: "港股"
code: "HK"
timezone: "Asia/Hong_Kong"
- name: "美股"
code: "US"
timezone: "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: 1000000
max_positions: 5
take_profit_pct: 0.08
stop_loss_pct: -0.05
# 日志
logging:
level: "INFO"
file: "logs/trading.log"
"src/calendar_provider.py"（★ 核心模块）
"""
calendar_provider.py
★ 交易日历数据提供者
职责：
1. 从本地文件/在线 API 获取交易日历
2. 缓存到本地，减少 API 调用
3. 支持多市场（A 股、港股、美股）
"""
import pandas as pd
import numpy as np
from pathlib import Path
from typing import Optional, Set, Dict
import json
import logging
from datetime import date
logging.basicConfig(level=logging.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 = source
self.local_file = Path(local_file)
self.api_url = api_url
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
# 内存缓存: {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_name
2024-01-01 False 元旦
2024-01-02 True NaN
"""
year = year or date.today().year
cache_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 到 API
if 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] = result
return result
def get_next_trading_day(
self,
date: date,
market: str = "A股"
) -> date:
"""获取下一个交易日"""
from datetime import timedelta
candidate = date + timedelta(days=1)
max_search = 30 # 最多往前找 30 天
for i in range(max_search):
if self.is_trading_day(candidate, market):
return candidate
candidate += timedelta(days=1)
# 找不到 → 返回 30 天后的日期
logger.warning(f"在 30 天内未找到 {market} 的下一个交易日")
return date + timedelta(days=30)
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}.csv"
if not path.exists() and self.local_file.exists():
path = self.local_file
if not path.exists():
return None
try:
df = pd.read_csv(path, parse_dates=['date']).set_index('date')
return df
except Exception as e:
logger.error(f"加载本地交易日历失败: {e}")
return None
def _fetch_online(
self,
market: str,
year: int
) -> Optional[pd.DataFrame]:
"""从在线 API 获取"""
if not self.api_url:
return None
try:
import requests
resp = requests.get(
f"{self.api_url}/calendar",
params={'market': market, 'year': year},
timeout=10
)
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 df
except Exception as e:
logger.error(f"在线获取交易日历失败: {e}")
return None
def _save_local(self, df: pd.DataFrame, market: str, year: int):
"""保存到本地缓存"""
path = self.cache_dir / f"{market}_{year}.csv"
df.to_csv(path)
def _generate_simple_calendar(self, year: int) -> pd.DataFrame:
"""
简易交易日历生成器
仅处理：
- 周末（周六日非交易）
- 主要法定节假日（元旦、春节、清明、五一、端午、中秋、国庆）
注意：不处理调休补班，精度有限，仅作降级方案
"""
import holidays
from datetime import date, timedelta
start = date(year, 1, 1)
end = date(year, 12, 31)
dates = []
current = start
while current <= end:
dates.append(current)
current += timedelta(days=1)
records = []
for d in dates:
# 周末
if d.weekday() in (5, 6):
is_td = False
hname = 'weekend'
else:
# 法定节假日（使用 holidays 库）
is_td = True
hname = ''
try:
cn_holidays = holidays.China(years=year)
if d in cn_holidays:
is_td = False
hname = 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 False
return 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 - trading
print(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 pd
import numpy as np
from datetime import datetime, time, date, timedelta
from typing import Optional, Tuple
from enum import Enum
import logging
logging.basicConfig(level=logging.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. 交易日历查询（委托给 CalendarProvider）
2. 日内交易时段判断
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_provider
self.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 sessions
def _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).days
reason = 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, reason
def 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, reason
def 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解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！