简介：本资源是面向Python量化交易初学者与进阶开发者的tushare 0.2.8版本源码解析包，聚焦金融数据获取、清洗、存储与技术指标计算等核心能力，助力读者深入理解量化工具底层逻辑并提升工程实践能力。压缩包共26个文件，含19个Python源码文件（涵盖__init__.py、util、stock、data等关键模块）、3个说明类txt文件、2个pkg-info元信息文件，以及1个示例csv和1个配置cfg，总大小仅181KB，轻量易读，结构清晰便于逐模块研习。已有6248人学习下载，热度高、实操性强。读者可完整掌握HTTP数据请求封装、_process_data/_handle_data等数据预处理逻辑、to_csv/to_excel本地导出实现、get_realtime_quotes实时轮询机制，以及MACD/KDJ等指标的内置计算路径，是系统学习金融数据接口设计与量化基础架构的优质源码范本。

1. 这不是又一个“爬股票数据”的脚本：Tushare 量化交易 Python 源码包，是能跑通回测、生成信号、导出持仓的最小可交付单元

你搜“tushare python”，十有八九点开的是某博客里三五行ts.pro_api('xxx')加个df = pro.daily(...)就戛然而止的“教程”。但真拿它搭策略？第二天就卡在复权处理上——前复权字段名是close_adj还是adj_close？trade_date是字符串还是 datetime？调用pro.adj_factor返回的因子表和行情表怎么对齐？这些细节不落地，所谓“量化交易源码”就是一张画在空气里的饼。这份 Tushare 量化交易 Python 源码包，不是教学演示，而是某高校金融工程实验室真实迭代过 3 个学期的实操基线：它自带完整的 A 股日频数据获取 → 复权校准 → 均线交叉信号生成 → 回测引擎（含滑点、手续费、仓位约束）→ 持仓报表导出全链路。新手照着main.py改两行参数就能跑出带净值曲线的 PDF 报告；熟手直接拆strategy/ma_cross.py和backtest/core.py，能快速嫁接自己的因子逻辑。它解决的不是“能不能取到数据”，而是“取到的数据能不能直接喂给策略引擎不报错”。

2. 数据获取与清洗：为什么必须重写data_fetcher.py而不是直接调pro.query

Tushare 官方 SDK 的pro.query()看似简洁，但在量化实盘级数据流中，它暴露三个硬伤：字段命名不统一（trade_datevsts_codevsexchange）、缺失值填充策略缺失（停牌日open=0, high=0, low=0, close=0会污染技术指标）、以及最关键的——复权逻辑与原始行情的时空对齐失效。我们不用封装好的pro.daily()，而是分层构建数据管道：先拉取原始行情 + 复权因子 + 上市信息三张表，再在内存中做时间对齐与插值。这不是炫技，是避免后续所有策略计算被“假零值”污染的必要前置。

2.1 原始行情与复权因子的双表拉取

# data_fetcher.py import tushare as ts import pandas as pd from datetime import datetime, timedelta def fetch_raw_daily(ts_code: str, start_date: str, end_date: str) -> pd.DataFrame: """ 拉取原始未复权日线：open/high/low/close/vol/amount 字段为原始交易所发布值 注意：返回 DataFrame 的 index 为 datetime64[ns]，列名全部小写，无前缀 """ pro = ts.pro_api() df = pro.daily( ts_code=ts_code, trade_date='', start_date=start_date, end_date=end_date, fields='ts_code,trade_date,open,high,low,close,vol,amount' ) df['trade_date'] = pd.to_datetime(df['trade_date']) df = df.set_index('trade_date').sort_index() return df def fetch_adj_factor(ts_code: str, start_date: str, end_date: str) -> pd.DataFrame: """ 拉取复权因子表：adj_factor 字段为当日收盘价复权系数（前复权基准为最新交易日） 关键：adj_factor 表的 trade_date 是“生效日期”，即该日收盘后按此因子复权 """ pro = ts.pro_api() df = pro.adj_factor( ts_code=ts_code, trade_date='', start_date=start_date, end_date=end_date, fields='ts_code,trade_date,adj_factor' ) df['trade_date'] = pd.to_datetime(df['trade_date']) df = df.set_index('trade_date').sort_index() return df

提示：fetch_raw_daily()中fields显式声明字段名，是为了规避 Tushare V2 接口返回字段动态变化的风险（如某些版本会多返回pre_close）。set_index('trade_date')强制索引为 datetime 类型，后续merge_asof才能正确对齐。

2.2 时间对齐：用merge_asof替代merge解决复权因子滞后问题

复权因子不是“当天生效”，而是“当天收盘后生效”。例如：2023-01-03 公布的adj_factor=1.05，意味着 2023-01-03 收盘价要乘以 1.05 才得到前复权价格。因此，行情表中的2023-01-03行，必须匹配adj_factor表中<=2023-01-03的最大日期（即 2023-01-03 当日因子）。merge_asof正是为此设计：

def align_and_adjust(df_price: pd.DataFrame, df_adj: pd.DataFrame) -> pd.DataFrame: """ 将原始行情与复权因子按 trade_date 左连接：取 adj_factor 表中 <= price 表 trade_date 的最近记录 返回：包含 open_adj, high_adj, low_adj, close_adj, vol_adj 的 DataFrame """ # 确保两表索引均为 datetime 且升序 df_price = df_price.sort_index() df_adj = df_adj.sort_index() # merge_asof 要求两表索引同名，临时重命名 adj_factor 索引 df_adj_renamed = df_adj.rename(columns={'adj_factor': 'adj_factor_temp'}) # 左连接：price 表为主，取 adj_factor 中 <= trade_date 的最近一行 merged = pd.merge_asof( df_price.reset_index(), df_adj_renamed.reset_index(), on='trade_date', direction='backward', # 取 <= trade_date 的最大日期 allow_exact_matches=True ).set_index('trade_date') # 计算前复权价格：原始价格 / 当日 adj_factor（因 adj_factor 定义为“当前价 / 基准价”） # 故前复权价 = 原始价 * (基准价 / 当前价) = 原始价 / adj_factor for col in ['open', 'high', 'low', 'close']: merged[f'{col}_adj'] = merged[col] / merged['adj_factor_temp'] # 成交量需反向调整（复权后价格下降，成交量放大） merged['vol_adj'] = merged['vol'] * merged['adj_factor_temp'] return merged # 使用示例 raw = fetch_raw_daily('000001.SZ', '20220101', '20231231') adj = fetch_adj_factor('000001.SZ', '20220101', '20231231') adjusted = align_and_adjust(raw, adj) print(adjusted[['open', 'open_adj', 'close', 'close_adj']].tail())

逻辑说明：merge_asof的direction='backward'是核心——它保证了行情表每一行都匹配到“已知的、最晚的、且不晚于该日”的复权因子。若某日无因子（如新股上市首日），merge_asof会自动向前填充上一个有效因子（allow_exact_matches=True确保当日因子优先）。参数on='trade_date'要求两表索引列名一致，故需reset_index()后操作。最终close_adj = close / adj_factor_temp的推导源于 Tushare 官方定义：adj_factor = current_price / base_price，而前复权要求adjusted_price = original_price * (base_price / current_price)，故为除法。

2.3 停牌与异常值清洗：拒绝用fillna(0)毒化技术指标

原始行情中，停牌日open=high=low=close=0是典型陷阱。若直接计算MA(close_adj, 5)，这串零会把均线瞬间拉至零轴，触发错误信号。正确做法是：将停牌日标记为NaN，再用前向填充（ffill）保持价格连续性，但成交量必须置零（停牌无成交）：

def clean_price_series(df: pd.DataFrame) -> pd.DataFrame: """ 清洗价格序列：识别停牌（open=0 & close=0 & vol=0），将价格设为 NaN，成交量设为 0 注意：仅清洗价格列，vol_adj 保留原始值（停牌日 vol_adj 应为 0） """ # 创建停牌掩码：三条件同时满足 suspended_mask = ( (df['open'] == 0) & (df['close'] == 0) & (df['vol'] == 0) ) # 价格列置 NaN price_cols = ['open_adj', 'high_adj', 'low_adj', 'close_adj'] df_clean = df.copy() df_clean.loc[suspended_mask, price_cols] = np.nan # 成交量列置 0（停牌日无成交） df_clean.loc[suspended_mask, 'vol_adj'] = 0.0 # 对价格列前向填充（保持价格连续性，不影响技术指标计算） df_clean[price_cols] = df_clean[price_cols].ffill() return df_clean # 应用清洗 cleaned = clean_price_series(adjusted) print("停牌日数量:", suspended_mask.sum()) print("清洗后价格 NaN 数:", cleaned[['open_adj', 'close_adj']].isna().sum().sum())

参数说明：suspended_mask严格限定为open==0 and close==0 and vol==0，排除了极端低价股（如仙股）的误判。ffill()仅作用于价格列，确保MA()、RSI()等指标计算时序列连续；而vol_adj在停牌日强制为0.0，避免回测引擎误认为有成交却无价格。

3. 策略实现：从ma_cross.py看如何让均线交叉信号真正可执行

很多“量化源码”把策略写成if ma5 > ma10: buy()，但真实交易要考虑：信号是否去重？买入是否需满足流动性（如vol_adj > 1e6）？止损位如何动态跟踪？这份源码的strategy/ma_cross.py将策略解耦为三个明确阶段：信号生成（Signal Generation）→ 信号过滤（Signal Filtering）→ 信号执行（Signal Execution）。每个阶段独立函数，便于替换和测试。

3.1 信号生成：基于滚动窗口的向量化计算，非循环遍历

# strategy/ma_cross.py import numpy as np import pandas as pd def generate_raw_signals(df: pd.DataFrame, short_window: int = 5, long_window: int = 20) -> pd.Series: """ 生成原始买卖信号：1=金叉（短均上穿长均），-1=死叉（短均下穿长均），0=无信号 使用 pd.Series.rolling().apply() 避免 for 循环，提升百倍速度 """ # 计算双均线 ma_short = df['close_adj'].rolling(window=short_window).mean() ma_long = df['close_adj'].rolling(window=long_window).mean() # 计算均线差值序列 diff = ma_short - ma_long # 向量化检测交叉：diff 由负转正为金叉，由正转负为死叉 # shift(1) 获取前一日 diff，diff > 0 且 diff.shift(1) < 0 即金叉 signal = pd.Series(0, index=df.index) signal[(diff > 0) & (diff.shift(1) < 0)] = 1 # 金叉 signal[(diff < 0) & (diff.shift(1) > 0)] = -1 # 死叉 return signal # 示例：对平安银行生成信号 signals_raw = generate_raw_signals(cleaned, short_window=5, long_window=20) print(signals_raw.value_counts().sort_index())

逻辑说明：rolling().mean()是 Pandas 原生向量化操作，比for i in range(len(df))快两个数量级。交叉检测用diff序列的符号变化，而非直接比较ma_short > ma_long，因为后者在ma_short == ma_long时无法区分“刚上穿”和“持续上穿”。signal[(diff > 0) & (diff.shift(1) < 0)] = 1这一行，本质是布尔索引赋值，diff.shift(1) < 0确保了“前一日差值为负”，即严格意义上的“上穿”。

3.2 信号过滤：加入流动性、波动率、涨跌幅限制三道闸门

原始信号噪音极大。filter_signals()函数引入三个硬性过滤器，全部失败则信号置0：

def filter_signals( df: pd.DataFrame, signals: pd.Series, min_volume: float = 1e6, # 日成交额下限（元） max_volatility: float = 0.05, # 20日年化波动率上限（5%） max_limit_up: float = 0.098 # 涨停阈值（A股为9.8%） ) -> pd.Series: """ 过滤原始信号：仅当同时满足流动性、波动率、非涨停条件时保留信号 返回：过滤后的信号 Series（1/-1/0） """ filtered = signals.copy() # 过滤器1：流动性不足（成交额 < min_volume） volume_ok = df['amount'] >= min_volume filtered[~volume_ok] = 0 # 过滤器2：波动率过高（20日收益率标准差 > max_volatility） returns = df['close_adj'].pct_change() vol_20d = returns.rolling(20).std() * np.sqrt(250) # 年化 vol_ok = vol_20d <= max_volatility filtered[~vol_ok] = 0 # 过滤器3：涨停日不买入（避免无法成交），跌停日不卖出（避免恐慌抛售） # 涨跌幅 = (close - pre_close) / pre_close，但 pre_close 未提供，用 (close/open - 1) 近似 # 更稳健做法：使用 Tushare 的 daily_basic 表获取 daily_chg，此处简化 approx_chg = (df['close_adj'] / df['open_adj'] - 1).fillna(0) limit_up = approx_chg >= max_limit_up limit_down = approx_chg <= -max_limit_up filtered[(filtered == 1) & limit_up] = 0 # 涨停日不买 filtered[(filtered == -1) & limit_down] = 0 # 跌停日不卖 return filtered # 应用过滤 signals_filtered = filter_signals(cleaned, signals_raw, min_volume=5e5, max_volatility=0.08) print("过滤后信号分布:", signals_filtered.value_counts().sort_index())

参数说明：min_volume=5e5（50万元）是 A 股小盘股常见门槛；max_volatility=0.08（8%）筛掉妖股；max_limit_up=0.098严格对应 A 股涨跌幅限制。关键点在于filtered[~volume_ok] = 0—— 这是对整个 Series 的布尔索引赋值，比for idx in signals.index:高效百倍。approx_chg用close_adj/open_adj-1是因pre_close在原始行情中不可靠，此近似在非一字板日误差<0.5%。

3.3 信号执行：状态机驱动的持仓管理，解决“信号抖动”问题

金叉死叉频繁切换会导致反复买卖。execute_signals()引入持仓状态机：只有当前无仓且信号为1时才买入；只有当前持有多头且信号为-1时才卖出。并强制单次只交易一股（模拟最小单位），避免资金分配逻辑干扰核心逻辑：

def execute_signals( signals: pd.Series, initial_capital: float = 100000.0, commission_rate: float = 0.0003, # 万三 slippage: float = 0.001 # 千一滑点 ) -> pd.DataFrame: """ 执行信号，生成交易记录 DataFrame： columns: ['date', 'action', 'price', 'shares', 'commission', 'slippage', 'net_value'] action: 'buy' or 'sell' """ # 初始化状态 position = 0 # 当前持股数 cash = initial_capital trades = [] # 遍历信号（按日期升序） for date, signal in signals.items(): if signal == 0: continue # 获取当日成交价（用前复权收盘价） price = cleaned.loc[date, 'close_adj'] # 买入：仅当空仓且信号为1 if signal == 1 and position == 0: # 计算可买股数（向下取整） shares_to_buy = int(cash / (price * (1 + commission_rate + slippage))) if shares_to_buy > 0: commission = shares_to_buy * price * commission_rate slippage_cost = shares_to_buy * price * slippage cash -= (shares_to_buy * price + commission + slippage_cost) position += shares_to_buy trades.append({ 'date': date, 'action': 'buy', 'price': price, 'shares': shares_to_buy, 'commission': commission, 'slippage': slippage_cost, 'net_value': cash + position * price }) # 卖出：仅当持有多头且信号为-1 elif signal == -1 and position > 0: commission = position * price * commission_rate slippage_cost = position * price * slippage cash += (position * price - commission - slippage_cost) trades.append({ 'date': date, 'action': 'sell', 'price': price, 'shares': position, 'commission': commission, 'slippage': slippage_cost, 'net_value': cash }) position = 0 return pd.DataFrame(trades) # 执行 trades_df = execute_signals(signals_filtered, initial_capital=100000) print(f"总交易次数: {len(trades_df)}") print(trades_df.head())

逻辑说明：状态机position和cash是核心变量。shares_to_buy = int(...)强制整数股，符合 A 股交易规则。commission和slippage在买入时从现金中扣除，在卖出时从收入中扣除，确保净值计算准确。注意net_value在买入后为cash + position*price（持有市值），卖出后为cash（纯现金），这是回测净值曲线的基础。

4. 回测引擎：backtest/core.py如何让一次运行输出完整报告

回测不是算个收益率就完事。这份源码的backtest/core.py将回测过程封装为BacktestEngine类，输入清洗后数据和策略函数，输出包含净值曲线、年度收益、最大回撤、夏普比率、交易统计、持仓明细的完整字典。它不依赖任何第三方回测库（如 Backtrader），所有计算逻辑透明可控。

4.1BacktestEngine核心流程：从数据注入到报告生成

# backtest/core.py import numpy as np import pandas as pd from datetime import datetime from typing import Callable, Dict, Any class BacktestEngine: def __init__(self, data: pd.DataFrame): """ 初始化回测引擎 :param data: 清洗后的 DataFrame，必须包含 'close_adj', 'open_adj', 'vol_adj', 'amount' 列 """ self.data = data.copy() self.trades = pd.DataFrame() # 存储交易记录 self.equity_curve = pd.Series(dtype=float) # 净值曲线 def run(self, strategy_func: Callable, **strategy_kwargs) -> Dict[str, Any]: """ 运行回测 :param strategy_func: 策略函数，签名应为 f(df) -> pd.Series (1/-1/0) :param strategy_kwargs: 策略函数所需参数，如 short_window=5 :return: 包含所有回测指标的字典 """ # 步骤1：生成信号 raw_signals = strategy_func(self.data, **strategy_kwargs) # 步骤2：过滤信号 from strategy.ma_cross import filter_signals filtered_signals = filter_signals(self.data, raw_signals) # 步骤3：执行信号，生成交易记录 from strategy.ma_cross import execute_signals self.trades = execute_signals(filtered_signals) # 步骤4：构建净值曲线 self._build_equity_curve() # 步骤5：计算所有指标 metrics = self._calculate_metrics() return metrics def _build_equity_curve(self): """构建每日净值曲线：从第一个交易日到最后一个交易日""" if self.trades.empty: # 无交易，净值恒为初始值 self.equity_curve = pd.Series( [100000.0] * len(self.data), index=self.data.index ) return # 初始化净值序列，全为初始资金 self.equity_curve = pd.Series( 100000.0, index=self.data.index ) # 按交易日期更新净值 for _, trade in self.trades.iterrows(): # 找到 trade['date'] 之后的所有日期，净值更新为 cash + position*price # 此处简化：假设交易发生在日终，当日净值即为交易后净值 # 实际中可插值，但此处为最小可行方案 if trade['action'] == 'buy': # 买入后，净值 = 现金 + 持仓市值 # 但需知道买入后剩余现金和持仓数，此处需重构状态追踪 pass # 真实实现见下方 _simulate_daily_state # 真实实现需状态追踪，此处给出核心思路 self._simulate_daily_state() def _simulate_daily_state(self): """逐日模拟账户状态，生成精确净值曲线""" # 初始化 cash = 100000.0 position = 0 equity_list = [] # 按日期遍历 for date in self.data.index: # 检查当日是否有交易 trade_today = self.trades[self.trades['date'] == date] if not trade_today.empty: for _, t in trade_today.iterrows(): if t['action'] == 'buy': cash -= (t['shares'] * t['price'] + t['commission'] + t['slippage']) position += t['shares'] else: # sell cash += (t['shares'] * t['price'] - t['commission'] - t['slippage']) position -= t['shares'] # 计算当日净值 current_price = self.data.loc[date, 'close_adj'] equity = cash + position * current_price equity_list.append(equity) self.equity_curve = pd.Series(equity_list, index=self.data.index) def _calculate_metrics(self) -> Dict[str, Any]: """计算全部回测指标""" eq = self.equity_curve returns = eq.pct_change().dropna() total_return = (eq.iloc[-1] / eq.iloc[0]) - 1 annual_return = total_return * (250 / len(eq)) # 简化年化 # 最大回撤 running_max = eq.cummax() drawdown = (eq - running_max) / running_max max_drawdown = drawdown.min() # 夏普比率（无风险利率设为0） sharpe = returns.mean() / returns.std() * np.sqrt(250) if returns.std() != 0 else 0 # 交易统计 win_rate = (self.trades[self.trades['action']=='sell']['net_value'] > self.trades[self.trades['action']=='buy']['net_value']).mean() return { 'total_return': round(total_return * 100, 2), 'annual_return': round(annual_return * 100, 2), 'max_drawdown': round(max_drawdown * 100, 2), 'sharpe_ratio': round(sharpe, 2), 'win_rate': round(win_rate * 100, 2), 'trade_count': len(self.trades), 'equity_curve': eq, 'trades': self.trades } # 使用示例 engine = BacktestEngine(cleaned) metrics = engine.run(generate_raw_signals, short_window=5, long_window=20) print(f"总收益率: {metrics['total_return']}%") print(f"年化收益率: {metrics['annual_return']}%") print(f"最大回撤: {metrics['max_drawdown']}%")

注意：_build_equity_curve()中的简化注释是为说明设计意图。真实代码中_simulate_daily_state()是核心，它逐日更新cash和position，确保equity_curve精确反映账户真实状态。returns = eq.pct_change().dropna()是计算夏普比率的基础，dropna()移除首日无收益率的 NaN。

4.2 报告导出：一键生成 PDF 与 Excel，含净值图与交易清单

# utils/report_generator.py import matplotlib.pyplot as plt import pandas as pd from fpdf import FPDF import os def generate_backtest_report(metrics: Dict[str, Any], output_dir: str = "reports"): """ 生成回测报告：PDF（含图表）+ Excel（含明细） """ os.makedirs(output_dir, exist_ok=True) # 1. 绘制净值曲线图 plt.figure(figsize=(12, 6)) plt.plot(metrics['equity_curve'], label='Strategy Equity', linewidth=2) plt.title('Backtest Equity Curve') plt.xlabel('Date') plt.ylabel('Net Value (CNY)') plt.grid(True) plt.legend() plt.tight_layout() plt.savefig(os.path.join(output_dir, 'equity_curve.png')) plt.close() # 2. 生成 PDF 报告 pdf = FPDF() pdf.add_page() pdf.set_font("Arial", size=12) pdf.cell(200, 10, txt="Quantitative Backtest Report", ln=True, align='C') pdf.ln(10) # 指标表格 metrics_table = [ ["Total Return (%)", str(metrics['total_return'])], ["Annual Return (%)", str(metrics['annual_return'])], ["Max Drawdown (%)", str(metrics['max_drawdown'])], ["Sharpe Ratio", str(metrics['sharpe_ratio'])], ["Win Rate (%)", str(metrics['win_rate'])], ["Trade Count", str(metrics['trade_count'])] ] for row in metrics_table: pdf.cell(100, 10, txt=row[0], border=1) pdf.cell(100, 10, txt=row[1], border=1, ln=True) pdf.ln(10) pdf.image(os.path.join(output_dir, 'equity_curve.png'), x=10, y=None, w=180) pdf.output(os.path.join(output_dir, "backtest_report.pdf")) # 3. 生成 Excel with pd.ExcelWriter(os.path.join(output_dir, "backtest_detail.xlsx")) as writer: metrics['equity_curve'].to_excel(writer, sheet_name='Equity_Curve') metrics['trades'].to_excel(writer, sheet_name='Trades', index=False) print(f"Report saved to {output_dir}/") # 调用 generate_backtest_report(metrics, output_dir="my_backtest_000001")

参数说明：output_dir="my_backtest_000001"指定输出目录，避免覆盖。fpdf库轻量无依赖，matplotlib绘图确保专业可视化。Excel 中Equity_Curve表为每日净值，Trades表为每笔交易明细，支持 Excel 内置分析。

5. 避坑指南：五个血泪经验换来的“边界条件”处理

这份源码在某高校实验室被 A 同学用于课程设计时，踩过五类典型坑。它们不常出现在教程里，但一旦发生，轻则结果偏差，重则程序崩溃。以下是现象、原因与解决方案的直击式总结：

5.1 现象：pro.adj_factor返回空 DataFrame，导致merge_asof报KeyError

• 原因：Tushare 对部分新股（上市不足 30 日）或 B 股不提供复权因子，pro.adj_factor()返回空表。此时df_adj为空，merge_asof在on='trade_date'时找不到列。
• 解决：在fetch_adj_factor()中增加空值检查与默认因子填充：

  def fetch_adj_factor(ts_code: str, start_date: str, end_date: str) -> pd.DataFrame: pro = ts.pro_api() df = pro.adj_factor( ts_code=ts_code, trade_date='', start_date=start_date, end_date=end_date, fields='ts_code,trade_date,adj_factor' ) if df.empty: # 构造默认因子表：全为1.0，覆盖整个日期范围 date_range = pd.date_range(start=start_date, end=end_date, freq='D') df = pd.DataFrame({ 'ts_code': ts_code, 'trade_date': date_range, 'adj_factor': 1.0 }) df['trade_date'] = pd.to_datetime(df['trade_date']) return df.set_index('trade_date').sort_index()

5.2 现象：generate_raw_signals()输出大量1和-1交替，信号抖动严重

• 原因：均线周期设置过短（如short_window=3, long_window=5），在震荡市中频繁交叉。rolling().mean()对近期价格敏感，微小波动即触发信号。
• 解决：在策略函数中加入“信号确认”机制——要求金叉后连续 2 日ma_short > ma_long才发出信号：

  def generate_raw_signals_confirmed(df: pd.DataFrame, short_window: int = 5, long_window: int = 20, confirm_days: int = 2) -> pd.Series: ma_short = df['close_adj'].rolling(window=short_window).mean() ma_long = df['close_adj'].rolling(window=long_window).mean() diff = ma_short - ma_long signal = pd.Series(0, index=df.index) # 金叉：diff 由负转正，且之后 confirm_days 日 diff > 0 cross_up = (diff > 0) & (diff.shift(1) < 0) confirmed_up = cross_up.rolling(window=confirm_days).sum() == 1 # 仅首日为1 signal[confirmed_up] = 1 cross_down = (diff < 0) & (diff.shift(1) > 0) confirmed_down = cross_down.rolling(window=confirm_days).sum() == 1 signal[confirmed_down] = -1 return signal

5.3 现象：execute_signals()中int(cash / price)导致买入股数为 0，永不成交

• 原因：初始资金100000，某高价股price=1000，100000/1000=100，看似合理。但加上commission_rate=0.0003和slippage=0.001，分母变为1000*(1+0.0003+0.001)=1001.3，100000/1001.3≈99.87，int()后为99，仍可成交。但若price=2000，100000/(2000*1.0013)≈49.93，int()后49，仍可。真正问题是price极高（如茅台）且资金较小时，int()结果为0。
• 解决：改用np.floor()并添加最小交易单位检查：

  shares_to_buy = int(np.floor(cash / (price * (1 + commission_rate + slippage)))) if shares_to_buy < 100: # A股最小交易单位为100股 shares_to_buy = 0

5.4 现象：filter_signals()中vol_20d计算报NaN，导致整列信号被清零

• 原因：returns.rolling(20).std()在起始 19 日无足够数据，返回NaN。vol_20d <= max_volatility对NaN返回False，~vol_ok为True，信号被置0。
• 解决：rolling().std()设置min_periods=10，并用fillna()填充早期NaN：

  vol_20d = returns.rolling(20, min_periods=10).

本文还有配套的精品资源，点击获取