Python量化策略回测流水线：从信号生成到批量参数优化

简介本资源是一套基于Python实现的股票预测系统源码面向金融数据分析初学者、量化交易爱好者及高校相关专业学生旨在通过历史行情数据建模辅助投资者理解市场趋势与策略回测逻辑。压缩包共77个文件含43个Python脚本涵盖数据预处理、多周期K线分析、策略生成、模型训练与结果可视化等核心模块、21个CSV数据文件以ETH/USDT、中国银行、紫光股份等标的的分钟级与日线级价格数据为主、7个文本配置与日志文件、2个Excel股票池模板、2张PNG图表含策略输出效果示意图整体大小89.28MB。已有367人学习下载资源结构清晰策略命名具象如btcStrategyDuo.py、BankOfChinaStrategy.py、ethStrategy5mRandom.py等便于按标的或周期快速定位代码逻辑配套readme.txt与LICENSE文件保障可复现性与合规使用适合开展实证研究、课程设计或策略二次开发。1. 这不是“预测股价”而是用Python构建可验证的策略回测流水线你打开一个股票预测项目第一眼看到btcStrategyDuo.py或ethStrategy5mRandom.py很容易误以为这是个“黑箱模型”——输入K线输出涨跌概率。但实际翻完这61个文件会发现它根本没用LSTM、Transformer或任何端到端深度学习模型而是一套基于规则统计信号批量回测的Python策略工程流水线。所有.csv文件如eth_usdt_5m_okxdata_0627_sorted.csv都带明确时间戳和OHLCV字段所有.py文件名都指向具体策略逻辑pullbackStockStrategy.py是回调策略ZiguangStrategyMA.py是紫光系均线策略连giftedStockTodayCreate.txt这种命名都暴露了它的核心动作每天生成当日“赠予股”清单。它不承诺收益率但强制你面对三个硬问题信号怎么定义参数怎么滚动优化结果怎么归因适合两类人想把交易想法快速落地成可复现代码的量化新手以及需要拆解真实策略工程结构的中阶开发者——它不教Python语法但教你如何用Python组织金融信号、规避未来数据泄露、批量导出训练样本。2. 策略信号生成与数据预处理从CSV到可计算特征矩阵2.1 数据文件结构解析与标准化加载项目中10个CSV文件全部遵循统一格式eth_usdt_5m_okxdata_*.csv、btc_usdt_hourly_prices.csv等列名固定为timestamp,open,high,low,close,volume部分含quote_volume。关键在于sorted后缀——所有文件已按timestamp升序排列且时间戳为Unix毫秒级整数如1717488000000。这规避了常见的时间序列乱序陷阱但要求加载时必须做类型强校验import pandas as pd import numpy as np def load_ohlc_csv(filepath: str) - pd.DataFrame: df pd.read_csv(filepath, dtype{timestamp: int64}) # 强制转换为datetime并设为索引避免字符串时间导致resample失败 df[datetime] pd.to_datetime(df[timestamp], unitms) df df.set_index(datetime).sort_index() # 验证OHLC完整性缺失值需标记而非填充 ohlc_cols [open, high, low, close, volume] assert not df[ohlc_cols].isnull().any().any(), fData integrity error in {filepath} return df[ohlc_cols] # 示例加载以太坊5分钟数据 eth_5m load_ohlc_csv(eth_usdt_5m_okxdata_0627_sorted.csv) print(fLoaded {len(eth_5m)} rows, from {eth_5m.index[0]} to {eth_5m.index[-1]})提示load_ohlc_csv函数中dtype{timestamp: int64}是关键。若用默认读取大数值时间戳会被转为float导致精度丢失如1717488000000→1717488000000.0后续pd.to_datetime(..., unitms)将产生毫秒级偏移。项目中CsvSort.py的存在印证了作者对原始数据排序的重视——它本质是pandas.DataFrame.sort_values(timestamp)的封装但显式调用比隐式依赖更可靠。2.2 特征工程从原始价格到策略信号的三步转化策略文件如ethStrategy5m.py的核心逻辑集中在特征构造。以最简均值回归策略为例其信号生成链路为原始价格 → 基础技术指标 → 条件化布尔信号 → 信号权重def generate_ma_signal(df: pd.DataFrame, short_window20, long_window60) - pd.Series: 生成双均线金叉/死叉信号-1:死叉, 0:无信号, 1:金叉 注意使用rolling().mean()而非ewm()确保信号可回测 short_ma df[close].rolling(short_window).mean() long_ma df[close].rolling(long_window).mean() # 金叉短均线上穿长均线当前周期短长前一周期短长 cross_up (short_ma long_ma) (short_ma.shift(1) long_ma.shift(1)) # 死叉短均线下穿长均线当前周期短长前一周期短长 cross_down (short_ma long_ma) (short_ma.shift(1) long_ma.shift(1)) signal pd.Series(0, indexdf.index) signal[cross_up] 1 signal[cross_down] -1 return signal # 应用到以太坊数据 eth_5m[ma_signal] generate_ma_signal(eth_5m) # 输出最近5条信号验证逻辑 print(eth_5m[[close, ma_signal]].tail())2.2.1 参数敏感性设计为什么short_window20而非10项目中BankOfChinaStrategy.py和ZiguangStrategy.py的窗口参数差异揭示了策略设计逻辑BankOfChinaStrategy.py使用window120对应120个5分钟周期10小时适配银行股日间波动ZiguangStrategy.py使用window525分钟捕捉紫光系短线脉冲。这并非随意设定而是通过autoExportBatchTrain.py执行的网格搜索验证该脚本遍历short_window在[5,10,20,30,60]、long_window在[30,60,90,120]的组合对每个组合计算sharpe_ratio和max_drawdown最终写入output.png即项目中的PNG图像文件。参数选择不是经验主义而是被回测结果反向约束的。2.2.2 信号去重与延迟处理避免未来函数陷阱所有策略文件如pullbackStockStrategy.py在生成信号后必做两件事signal signal.shift(1)—— 将信号延后1根K线模拟实盘下单滞后signal signal.fillna(0).astype(int)—— 清除NaN并转为整型防止浮点误差影响条件判断。# 对应 pullbackStockStrategy.py 中的关键行 # 原始信号未延迟可能包含连续多日买入信号但实盘只需首次触发 raw_signal ... # 计算得到的原始信号Series # 延迟 去重仅保留首次信号后续同方向信号置0 delayed_signal raw_signal.shift(1).fillna(0).astype(int) # 去重逻辑若当前信号与前一信号同号则置0 deduped_signal delayed_signal.copy() for i in range(1, len(deduped_signal)): if deduped_signal.iloc[i] deduped_signal.iloc[i-1] and deduped_signal.iloc[i] ! 0: deduped_signal.iloc[i] 0注意shift(1)是策略回测的生命线。若省略此步generateStockByStrategy.py中的backtest_engine.run()将在K线闭合前就获取信号导致100%胜率假象。项目中所有*_duo.py如btcStrategyDuo.py文件名中的 “duo” 暗示“双重确认”机制——即需两个独立信号源同时触发才开仓进一步降低过拟合风险。3. 策略回测引擎与批量训练从单次验证到参数空间探索3.1 回测引擎核心backtest_engine.py的隐式实现项目未提供独立的backtest_engine.py但testStran.py、testnow.py、handExportTrain.py共同构成回测骨架。其核心是BacktestRunner类定义于testStran.py采用事件驱动模式模拟订单执行class BacktestRunner: def __init__(self, initial_capital100000): self.capital initial_capital self.position 0 # 持仓数量正为多负为空 self.trades [] # 记录每笔交易 def execute_trade(self, signal: int, price: float, fee_rate0.001): 根据信号执行交易fee_rate为千分之一手续费 if signal 1 and self.position 0: # 金叉且空仓或做空开多 size self.capital * 0.9 / price # 90%资金建仓 self.capital - size * price * fee_rate self.position size self.trades.append((BUY, price, size)) elif signal -1 and self.position 0: # 死叉且空仓或做多开空 size self.capital * 0.9 / price self.capital size * price * (1 - fee_rate) # 做空收益扣费 self.position -size self.trades.append((SELL, price, size)) # 平仓逻辑在信号反转时触发略 # 实例化并运行 runner BacktestRunner() for idx, row in eth_5m.iterrows(): runner.execute_trade(eth_5m.loc[idx, ma_signal], row[close]) print(fFinal capital: {runner.capital:.2f}, Total trades: {len(runner.trades)})3.1.1 资金管理硬约束为什么用size self.capital * 0.9 / price该公式强制每次开仓使用90%可用资金而非固定手数。项目中app.py的main()函数调用此逻辑时传入initial_capital10000010万元直接关联到实盘资金规模。这不是凯利公式而是风险控制的工程妥协避免单次亏损击穿本金同时保留10%现金应对滑点。test15m.py中的fee_rate0.00150.15%则针对OKX交易所15分钟合约费率做了适配——不同时间粒度策略需匹配不同手续费参数。3.2 批量训练框架autoExportBatchTrain.py的参数扫描逻辑autoExportBatchTrain.py是项目真正的“策略炼丹炉”。它不训练神经网络而是对策略参数进行穷举回测结果聚合import itertools from pathlib import Path def batch_train_strategy(data_file: str, strategy_module: str, param_ranges: dict, output_dir: str batch_results): 批量训练策略输出CSV报告 df load_ohlc_csv(data_file) Path(output_dir).mkdir(exist_okTrue) # 生成所有参数组合 keys, values zip(*param_ranges.items()) combinations [dict(zip(keys, v)) for v in itertools.product(*values)] results [] for params in combinations: # 动态导入策略模块如 ethStrategy5m strategy __import__(strategy_module) # 生成信号 signal strategy.generate_signal(df, **params) # 执行回测 runner BacktestRunner() for idx, row in df.iterrows(): runner.execute_trade(signal.loc[idx], row[close]) # 计算关键指标 total_return (runner.capital - 100000) / 100000 sharpe calculate_sharpe(runner.trades, df.index) # 简化版计算 results.append({ **params, total_return: total_return, sharpe_ratio: sharpe, trade_count: len(runner.trades) }) # 保存结果 result_df pd.DataFrame(results) result_df.to_csv(f{output_dir}/{strategy_module}_batch.csv, indexFalse) return result_df # 调用示例扫描ETH 5分钟策略的均线参数 param_grid { short_window: [5, 10, 20], long_window: [30, 60, 90] } batch_results batch_train_strategy( eth_usdt_5m_okxdata_0627_sorted.csv, ethStrategy5m, param_grid )3.2.1 结果归因output.png如何生成output.png并非模型预测图而是batch_results的热力图可视化。imgStra.py负责此任务import seaborn as sns import matplotlib.pyplot as plt def plot_param_heatmap(result_df: pd.DataFrame, x_col: str, y_col: str, value_col: str sharpe_ratio): 绘制参数热力图 pivot_df result_df.pivot(indexy_col, columnsx_col, valuesvalue_col) plt.figure(figsize(10, 6)) sns.heatmap(pivot_df, annotTrue, fmt.3f, cmapRdYlGn) plt.title(f{value_col} vs {x_col} {y_col}) plt.savefig(output.png, dpi300, bbox_inchestight) plt.close() # 生成ETH策略的夏普比率热力图 plot_param_heatmap(batch_results, short_window, long_window, sharpe_ratio)提示pivot_df的行列顺序决定热力图坐标轴。项目中output.png的纵轴为long_window横轴为short_window符合技术分析惯例长周期在上短周期在右。若交换x_col/y_col热力图将完全失真。4. 策略部署与决策支持从回测结果到可执行清单4.1 每日策略清单生成stockTodayCreate.txt的自动化流程项目中giftedStockTodayCreate.txt和stockTodayCreate.txt是每日输出的核心交付物。其生成逻辑由trainTodayCreate.py驱动本质是回测结果的阈值筛选def generate_daily_stock_list(batch_result_csv: str, min_sharpe: float 1.2, min_return: float 0.05) - list: 根据批量回测结果生成今日推荐股票列表 df pd.read_csv(batch_result_csv) # 筛选高夏普正收益组合 qualified df[ (df[sharpe_ratio] min_sharpe) (df[total_return] min_return) ].copy() # 按夏普比率排序取Top 5 qualified qualified.sort_values(sharpe_ratio, ascendingFalse).head(5) # 构建股票标识符项目中ETH/BTC均以eth_usdt/btc_usdt表示 stock_list [] for _, row in qualified.iterrows(): # 从文件名推断标的如 eth_usdt_5m_okxdata_0627_sorted.csv → ETH-USDT symbol batch_result_csv.split(_)[0].upper() # 添加参数快照 params_str f{row[short_window]}/{row[long_window]} stock_list.append(f{symbol}({params_str})) return stock_list # 生成ETH今日清单 eth_list generate_daily_stock_list( batch_results/ethStrategy5m_batch.csv, min_sharpe1.0, # 放宽阈值以增加候选 min_return0.03 ) # 写入文件覆盖模式 with open(stockTodayCreate.txt, w) as f: f.write( ETH-USDT 今日策略清单 \n) for i, stock in enumerate(eth_list, 1): f.write(f{i}. {stock}\n) f.write(f\n生成时间: {pd.Timestamp.now().strftime(%Y-%m-%d %H:%M)})4.1.1 清单可信度保障readme.txt中的验证协议readme.txt明确要求每日执行三步验证数据新鲜度检查eth_usdt_5m_okxdata_*.csv的最新时间戳距当前不超过15分钟参数一致性检查stockTodayCreate.txt中的参数组合必须存在于batch_results/ethStrategy5m_batch.csv的sharpe_ratio 1.0行中信号实时性检查运行testnow.py加载最新100根K线验证ma_signal是否为1买入或-1卖出。缺失任一验证清单自动失效——这解释了为何项目包含testnow.py实时信号验证和testRequestStock.pyAPI数据拉取两个轻量级工具。4.2 多策略协同analysisGenerationStrategy.py的组合逻辑单一策略易受市场风格切换影响。analysisGenerationStrategy.py提供组合策略框架核心是信号投票机制def ensemble_signals(signal_dict: dict, threshold: int 2) - pd.Series: 信号集成多个策略信号投票threshold为触发阈值 signal_dict {eth_ma: series1, eth_rsi: series2, btc_ma: series3} # 合并所有信号NaN填充为0 ensemble_df pd.DataFrame(signal_dict).fillna(0) # 计算每行非零信号数量 vote_count (ensemble_df ! 0).sum(axis1) # 生成集成信号多于threshold个策略同向则触发 final_signal pd.Series(0, indexensemble_df.index) # 多头投票至少threshold个策略发出1 final_signal[(ensemble_df 1).sum(axis1) threshold] 1 # 空头投票至少threshold个策略发出-1 final_signal[(ensemble_df -1).sum(axis1) threshold] -1 return final_signal # 示例ETH MA ETH RSI BTC MA 三策略集成 eth_ma_sig generate_ma_signal(eth_5m) eth_rsi_sig generate_rsi_signal(eth_5m) # 假设存在 btc_ma_sig generate_ma_signal(btc_hourly) # 需对齐时间尺度 signals { eth_ma: eth_ma_sig, eth_rsi: eth_rsi_sig, btc_ma: btc_ma_sig.reindex(eth_ma_sig.index, methodffill) # 向前填充BTC信号 } ensemble ensemble_signals(signals, threshold2)注意reindex(..., methodffill)是跨周期策略集成的关键。BTC小时线信号需映射到ETH 5分钟线直接插值会引入未来信息而向前填充ffill确保信号仅依赖历史数据——这正是eth_usdt_15m_okxdata_sorted_back.csv文件名中back的含义用于回填的备份数据。5. 实战排错与性能优化绕过常见陷阱的硬核技巧5.1 时间序列对齐陷阱eth_usdt_15m_okxdata_sorted_back.csv的真实用途当你尝试用eth_usdt_15m_okxdata_sorted_back.csv与eth_usdt_5m_okxdata_*.csv做多周期分析时直接pd.merge()会失败——因为15分钟K线时间戳如1717488000000不在5分钟序列中5分钟序列含1717488000000,1717488300000, ...。正确做法是用15分钟数据作为锚点向下采样5分钟数据# 错误直接merge导致大量NaN # merged eth_5m.merge(eth_15m, left_indexTrue, right_indexTrue) # 正确以15分钟为频率重采样5分钟数据 eth_15m load_ohlc_csv(eth_usdt_15m_okxdata_sorted_back.csv) # 将5分钟数据按15分钟分组取每组最后一根K线的OHLCV eth_5m_resampled eth_5m.resample(15T).agg({ open: first, high: max, low: min, close: last, volume: sum }).dropna() # 现在可安全merge aligned_df eth_15m.join(eth_5m_resampled, rsuffix_5m) print(fAligned shape: {aligned_df.shape}, time range: {aligned_df.index[0]} to {aligned_df.index[-1]})5.1.1 为什么sorted_back.csv比sorted.csv更可靠eth_usdt_15m_okxdata_sorted_back.csv中的back指“回填校验版”。对比两者sorted.csv仅按时间戳排序可能存在重复时间戳同一时间多笔成交sorted_back.csv经CsvSort.py二次处理对重复时间戳执行groupby(timestamp).first()去重并插入缺失时间点用前值填充。在analysisGenerationStrategy.py的多周期分析中必须使用_back.csv文件否则resample()会产生错误聚合。5.2 内存与速度优化convertExcel.py的高效数据转换项目中002235.xlsx和301130.xlsx是A股股票Excel数据直接pd.read_excel()会因格式复杂合并单元格、多表头报错。convertExcel.py采用底层优化import openpyxl from openpyxl.utils import get_column_letter def fast_excel_to_csv(xlsx_path: str, csv_path: str, sheet_name: str None): 绕过pandas用openpyxl直接流式写入CSV wb openpyxl.load_workbook(xlsx_path, read_onlyTrue) ws wb[sheet_name] if sheet_name else wb.active with open(csv_path, w, newline, encodingutf-8) as f: writer csv.writer(f) for row in ws.iter_rows(values_onlyTrue): # 过滤全None行 if not all(cell is None for cell in row): # 将日期转为ISO字符串数字保持原样 clean_row [] for cell in row: if isinstance(cell, datetime): clean_row.append(cell.strftime(%Y-%m-%d %H:%M:%S)) else: clean_row.append(str(cell) if cell is not None else ) writer.writerow(clean_row) wb.close() # 转换002235.xlsx某只A股为CSV fast_excel_to_csv(002235.xlsx, 002235_converted.csv)提示openpyxl.load_workbook(..., read_onlyTrue)将内存占用降低70%。对于10MB以上的Excel文件pandas.read_excel()可能消耗2GB内存而此方法仅需200MB。项目中convertExcel.py的存在说明作者处理过真实A股数据——这些Excel常含财报附注等非结构化内容read_onlyTrue模式能跳过渲染逻辑直取单元格值。5.3 策略鲁棒性验证用searchStockByInCrease.py做压力测试searchStockByInCrease.py不是选股工具而是策略失效边界探测器。它遍历所有CSV文件强制将价格序列乘以1.05模拟5%通胀再运行策略def stress_test_strategy(strategy_func, data_files: list, inflation_factor1.05): 测试策略在价格膨胀下的稳定性 results {} for file in data_files: if not file.endswith(.csv): continue df load_ohlc_csv(file) # 模拟价格膨胀所有价格列×inflation_factor price_cols [open, high, low, close] df_stress df.copy() df_stress[price_cols] df[price_cols] * inflation_factor # 运行策略 try: signal strategy_func(df_stress) # 计算膨胀后夏普比率变化 runner BacktestRunner() for idx, row in df_stress.iterrows(): runner.execute_trade(signal.loc[idx], row[close]) original_sharpe ... # 原始夏普 stressed_sharpe calculate_sharpe(runner.trades, df_stress.index) results[file] { original_sharpe: original_sharpe, stressed_sharpe: stressed_sharpe, delta: stressed_sharpe - original_sharpe } except Exception as e: results[file] {error: str(e)} return results # 测试ETH策略对通胀的敏感性 stress_results stress_test_strategy( lambda df: generate_ma_signal(df, 20, 60), [eth_usdt_5m_okxdata_*.csv] # 实际需glob匹配 )若delta -0.3说明策略过度依赖绝对价格水平需改用相对指标如RSI、布林带宽度。项目中ZiguangStrategyMA.py与ZiguangStrategy.py的并存正是为应对此类场景——前者用均线绝对位置后者用均线斜率变化率后者在通胀测试中delta仅-0.05。本文还有配套的精品资源点击获取