2026/9/1 1:24:18

用Python量化验证“可交易反弹”:从数据到回测的完整实现

用Python量化验证“可交易反弹”:从数据到回测的完整实现 洪灝三周前提出过一个判断市场会出现“可交易反弹”。到了8月13日他在英文访谈中被再度问到这个话题很多人关心的核心问题几乎一样反弹到底“到哪里了”是兑现了、结束了还是根本没有出现这类预测在财经讨论中非常常见但争议也很大因为“可交易反弹”本身不是一个精确的量化定义。有人看5日反弹有人看20日反弹有人要求放量突破均线有人只看RSI背离标准不同结论自然不同。所以这篇文章不打算争论预测本身的对错而是把“可交易反弹”当成一个可验证的技术任务来处理用Python从行情数据获取、技术指标计算、信号识别、简单回测到批量跟踪搭一套最小可用的“反弹观点跟踪系统”。跑完之后你不仅能复盘这次观点的时间窗口还能把未来任何类似的宏观判断转成代码规则用数据而不是情绪来判断“反弹到底在哪里”。整套流程不需要GPU不需要高配服务器普通笔记本安装Python环境就能跑数据源使用公开行情接口。先说明一个边界本文所有代码只是技术分析和教学演示不构成投资建议。凡是涉及行情判断、卖出买入信号的内容请把它当作量化研究方法而不是操作指令。市场有风险入市需要自己做决策。1. 核心能力速览能力项说明项目类型市场观点跟踪与量化验证脚本核心目标把“可交易反弹”这类主观观点拆成可验证的技术信号主要功能行情数据获取、均线/RSI/MACD计算、反弹信号识别、简单回测、批量跟踪数据来源公开行情接口如 akshare、yfinance需要按接口实际可用性确认硬件要求普通CPU电脑内存8G以上即可无GPU要求运行环境Python 3.9建议使用 Anaconda 或 venv 隔离环境启动方式Jupyter Notebook 分步执行或 Python 脚本直接运行接口能力可封装为本地HTTP API供外部程序调用批量任务支持多指数、多股票、多周期批量扫描输出形式信号标记K线图、回测收益曲线、信号结果CSV适合人群技术分析学习者、量化入门者、关注宏观观点验证的投资者这套东西的核心价值不是预测未来而是提供一套“事后可复盘、事前可跟踪”的规则化验证流程。把一个模糊的反弹观点落到代码里剩下的就是看市场走势是否符合规则。2. 适用场景与使用边界先回答谁适合用。第一类是想复盘宏观观点的人比如今天要验证洪灝此前提到的“可交易反弹”你可以用代码圈定时间区间看看从观点提出到8月13日指数是否满足了反弹条件反弹幅度有多少持续时间是多少。第二类是技术分析初学者通过修改参数、观察信号变化能快速理解均线、RSI、MACD这些指标在真实行情中长什么样。第三类是量化研究入门者把手工看图变成可回测的规则是走向系统化交易的第一步。这套方法不适合什么地方不适合做高频交易因为日线数据颗粒度不够不适合做基本面研究因为它完全不看财报和估值更不适合作为“荐股系统”因为它只是技术面信号并不能保证未来上涨。还需要特别注意技术指标存在大量“事后看起来很美、事前很难执行”的问题比如信号延迟、假突破、震荡行情反复打脸。所以不要把一个简单的技术信号当成完整策略。从合规和版权角度也要说清楚行情数据要使用合法公开的接口引用财经观点时要标明出处不要篡改原意如果使用他人文章、访谈内容需要遵守转载规范任何涉及实际资金操作的决策责任在操作者本人。涉及历史行情、指数数据、公开访谈观点只要不歪曲、不荐股、不承诺收益技术分析演示是在安全范围内的。3. 环境准备与数据接口3.1 安装依赖建议先建一个干净的Python环境避免和系统环境冲突。使用Anaconda的话可以这样初始化conda create -n bounce_research python3.9 -y conda activate bounce_research然后安装本次需要的基础库pip install pandas numpy matplotlib akshare yfinance如果只需要国内行情yfinance可以不用装。akshare是开源财经数据接口库不需要额外token但部分接口可能因为数据源限制在特定网络环境下访问不稳定。yfinance用于拉取海外指数比如恒生指数、标普500实际是否可用要看你所在网络到数据源的通畅程度。3.2 数据接口选择本文演示以日线数据为主。技术分析需要开高低收和成交量五个字段大部分指数接口都能返回。常见数据源如下数据源适合场景是否需要token说明akshareA股指数、个股、部分港股指数一般不需要接口丰富字段随版本变化使用前查看文档yfinance美股、港股等海外指数不需要网络依赖较高字段相对稳定tushareA股专业研究需要token数据质量好但有积分门槛需要注意akshare的接口名经常随版本调整不能只看一篇教程就闭眼抄。建议在Python里先确认当前版本支持的函数名import akshare as ak print(ak.__version__) # 查看所有包含 index 的函数名 names [name for name in dir(ak) if index in name.lower()] for n in names: print(n)这样能避免使用过时接口。下面所有代码均使用“示意接口名”实际运行时以你本机akshare版本为准。4. 数据获取与预处理以沪深300指数为例先把日线数据拉下来统一列名和索引。这里用A股接口做演示字段命名以实际返回为准import pandas as pd import akshare as ak def load_index_data(symbol: str sh000300): 获取指数日线数据并清洗。 注意akshare 接口名会随版本变化实际运行时需要根据 当前版本和返回字段调整列名。 df ak.stock_zh_index_daily(symbolsymbol) df df.rename(columns{ date: date, open: open, high: high, low: low, close: close, volume: volume }) df[date] pd.to_datetime(df[date]) df df.set_index(date) df df[[open, high, low, close, volume]].dropna() df df.sort_index() return df df load_index_data(sh000300) print(df.tail())如果使用yfinance拉取海外指数代码更简单import yfinance as yf df yf.download(^HSI, start2025-01-01, auto_adjustTrue) df df[[Open, High, Low, Close, Volume]] df.columns [open, high, low, close, volume] df.index pd.to_datetime(df.index) df df.dropna() print(df.tail())预处理阶段最关键的是三个问题第一日期索引必须有序排序后才能计算滚动指标第二字段必须齐全没有成交量的反弹信号缺少说服力第三要注意停牌日和缺失值直接用dropna会丢失一部分数据如果担心影响连续性可以用前向填充但默认情况下指数数据比较干净直接剔除缺失值问题不大。拿到干净数据后先做一遍可视化看看这一阶段市场整体走势再进入指标计算。5. 技术指标计算与“可交易反弹”信号定义“可交易反弹”在洪灝的语境里是一种对市场状态的判断但要在代码里验证必须把它翻译成明确的规则。下面这套规则是我自定义的常见技术分析组合不代表任何人的原话只是为了给“可交易反弹”一个可计算的代理标准。实际使用时你完全可以调整阈值。一个可交易反弹信号我定义为四个条件的叠加价格处于阶段低点区域最新收盘价相对过去20根K线最低点的回撤达到一定幅度。动量转强RSI从超卖区域回升比如从35以下重新站上40。均线修复价格向上突破短期均线说明短期趋势开始转多。量价配合当日成交量大于过去5天平均成交量说明反弹有资金参与。先计算RSI和均线。RSI的经典周期是14天均线用10日和30日作为快慢线def add_indicators(df: pd.DataFrame, rsi_period14, ma_fast10, ma_slow30): df df.copy() # 计算RSI delta df[close].diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.rolling(rsi_period, min_periodsrsi_period).mean() avg_loss loss.rolling(rsi_period, min_periodsrsi_period).mean() rs avg_gain / avg_loss df[rsi] 100 - (100 / (1 rs)) # 计算均线 df[ma_fast] df[close].rolling(ma_fast).mean() df[ma_slow] df[close].rolling(ma_slow).mean() # 成交量均线 df[vol_ma5] df[volume].rolling(5).mean() return df再定义信号识别函数。这里把“从近期低点反弹”和“站上短期均线”作为核心条件def detect_bounce_signal(df: pd.DataFrame, lookback20, oversold_level35, min_rebound-0.08): df df.copy() # 阶段低点 rolling_low df[close].rolling(lookback).min() df[drawdown] df[close] / rolling_low - 1 # 信号条件 df[signal] ( (df[drawdown] min_rebound) (df[rsi] oversold_level) (df[close] df[ma_fast]) (df[volume] df[vol_ma5]) ) return df这段代码里min_rebound-0.08表示从阶段低点回撤超过8%oversold_level35表示RSI必须高于35close ma_fast表示价格站上10日均线volume vol_ma5表示放量。如果四个条件同时满足就认为当前K线发出了可交易反弹信号。需要特别注意一点这里没有引入任何未来数据。滚动窗口默认只使用当前及过去的数据不会偷看未来。如果你后续自己扩展指标不要使用shift(-1)这类未来偏移来构造信号否则回测结果会严重失真。6. 反弹信号可视化验证信号计算完之后不能只看数字必须画图验证。把K线、短期均线、RSI、信号点放在一张图上直观判断信号是否出现在合理位置。import matplotlib.pyplot as plt def plot_signals(df: pd.DataFrame, titleBounce Signal): fig, axes plt.subplots(3, 1, figsize(12, 10), sharexTrue) # 第一张图收盘价 均线 信号点 axes[0].plot(df.index, df[close], labelclose, colorblack, linewidth1.2) axes[0].plot(df.index, df[ma_fast], labelMA10, colororange, linewidth1) axes[0].plot(df.index, df[ma_slow], labelMA30, colorblue, linewidth1) signal_dates df.index[df[signal]] signal_prices df.loc[signal_dates, close] axes[0].scatter(signal_dates, signal_prices, marker^, colorred, s120, labelbounce signal) axes[0].set_ylabel(Price) axes[0].legend() # 第二张图RSI axes[1].plot(df.index, df[rsi], labelRSI, colorpurple) axes[1].axhline(70, colorgray, linestyle--, linewidth0.8) axes[1].axhline(30, colorgray, linestyle--, linewidth0.8) axes[1].set_ylabel(RSI) axes[1].legend() # 第三张图回撤 axes[2].plot(df.index, df[drawdown], labeldrawdown, colorgreen) axes[2].axhline(-0.08, colorgray, linestyle--, linewidth0.8) axes[2].set_ylabel(Drawdown) axes[2].legend() plt.suptitle(title) plt.tight_layout() plt.show() # 示例拉一段2023年以来的数据做演示 df load_index_data(sh000300) df add_indicators(df) df detect_bounce_signal(df) plot_signals(df, titleCSI300 Bounce Signal)运行后你应该能看到三类信息红色三角标记的日期是否出现在阶段低点附近RSI是否处于超卖区域结束的位置回撤是否已经达到阈值。如果信号出现在一波下跌末端、RSI刚从低位回升的位置说明规则基本符合“反弹启动”的直觉如果信号频繁出现在下跌途中说明阈值太宽松需要调整lookback或者oversold_level。判断信号是否有效的标准不是信号当天涨了多少而是信号之后5到10个交易日是否真的出现了一段可交易涨幅。这就要进入回测环节。7. 简单回测与效果评估手工看图只能解决“信号合不合理”不能解决“信号赚不赚钱”。做一个最简单的策略回测有信号时全仓买入信号消失时清仓。这个回测不考虑手续费和滑点只用来评估信号逻辑是否有效。def simple_backtest(df: pd.DataFrame, initial_cash100000): cash initial_cash position 0 trade_log [] last_signal False for date, row in df.iterrows(): current_signal bool(row[signal]) # 信号出现且未持仓买入 if current_signal and not last_signal: position cash / row[close] cash 0 trade_log.append({ date: date, action: buy, price: row[close] }) # 信号结束且有持仓卖出 elif not current_signal and last_signal and position 0: cash position * row[close] position 0 trade_log.append({ date: date, action: sell, price: row[close] }) last_signal current_signal # 如果回测结束时仍持仓按最后收盘价强制平仓 if position 0: cash position * df.iloc[-1][close] position 0 trade_log.append({ date: df.index[-1], action: force_sell, price: df.iloc[-1][close] }) return cash, trade_log cash, trades simple_backtest(df) print(最终资产:, round(cash, 2)) print(交易次数:, len(trades)) print(trades[-10:])运行完成后需要把策略结果和“简单持有”对比。可以先计算买入持有收益buy_hold_return df.iloc[-1][close] / df.iloc[0][close] - 1再用交易记录计算策略收益。更完整的评估维度包括指标计算方式说明策略累计收益最终资产 / 初始资金 - 1回测期总收益买入持有收益期末收盘价 / 期初收盘价 - 1基准参考交易次数买卖记录数量次数过少说明信号太严格过多说明反复打脸最大回撤策略资产曲线高点到低点的最大跌幅衡量波动风险胜率盈利交易次数 / 总交易次数需要配合盈亏比看这里没有给你一段编造的收益数字因为结果完全依赖数据区间、指数标的和参数设置。你可以用同一个函数跑不同时间段比如2024年至今、2023年全年对比策略表现。如果在一个区间有效、另一个区间无效是正常的说明规则存在过拟合风险。最怕的是任何一个区间都产生几十笔交易且大部分是亏损那说明阈值设置不合理需要调参。回测只是验证工具不是赚钱机器。任何规则都可能在未来失效。8. 批量任务与接口API封装洪灝的观点不止针对沪深300市场判断通常是宏观层面的可能同时覆盖A股、港股、美股。手动一个个跑太慢所以要写批量扫描。下面是一个多指数批量信号扫描的框架symbols { 沪深300: sh000300, 上证指数: sh000001, 恒生指数: hsi_hk, } def batch_scan(symbol_map: dict): results [] for name, symbol in symbol_map.items(): try: tmp load_index_data(symbol) tmp add_indicators(tmp) tmp detect_bounce_signal(tmp) latest tmp.iloc[-1] results.append({ name: name, symbol: symbol, last_close: round(latest[close], 2), rsi: round(latest[rsi], 2), signal: bool(latest[signal]), }) except Exception as e: print(f{name} 处理失败: {e}) return pd.DataFrame(results) result_df batch_scan(symbols) result_df.to_csv(bounce_signals.csv, indexFalse) print(result_df)注意不同市场的指数在akshare中可能使用完全不同的接口函数上面的symbol只是示意值实际批量运行时需要为每个市场写对应的数据获取逻辑。更稳妥的做法是先保证单个指数跑通再复制成多个函数最后统一封装。批量任务工程化还需要考虑几个点数据缓存同一个指数一天之内重复拉取会浪费请求可以把原始数据缓存到本地parquet或csv文件。失败重试某个数据源接口可能临时失败加一个3次重试机制。日志输出批量扫描要记录每个标的成功还是失败方便排查。如果希望把信号服务提供给外部程序使用可以封装成HTTP接口。下面是一个Flask示例注意这是演示结构需要把前面的load_index_data、add_indicators、detect_bounce_signal串进来from flask import Flask, request, jsonify app Flask(__name__) app.route(/api/bounce_signal, methods[POST]) def bounce_signal(): data request.get_json(forceTrue) symbol data.get(symbol, sh000300) try: df load_index_data(symbol) df add_indicators(df) df detect_bounce_signal(df) latest df.tail(1).to_dict(records)[0] latest[signal] bool(latest[signal]) return jsonify({code: 0, data: latest}) except Exception as e: return jsonify({code: 500, msg: str(e)}), 500 if __name__ __main__: app.run(host127.0.0.1, port5000)启动后可以用curl做一次简单调用curl -X POST http://127.0.0.1:5000/api/bounce_signal \ -H Content-Type: application/json \ -d {symbol: sh000300}接口返回的是最新一天的指标和信号后续可以封装成定时任务每天收盘后自动扫描把结果推送到企业微信机器人、邮件或者存储到数据库。这样才算真正把“跟踪反弹观点”变成自动化流程。9. 资源占用与性能观察这套系统属于轻量级数据分析不涉及深度学习资源占用非常有限。单个指数10年日线数据大约2500条在pandas里计算RSI、均线、信号一次运行通常只需要几十毫秒到几百毫秒内存占用不会超过几百MB基本不存在硬件瓶颈。真正需要考虑性能的是批量任务。如果你把范围扩大到全市场5000只A股每只股票都跑一遍指标单线程运行可能需要几分钟到十几分钟。优化方向有三个用concurrent.futures线程池做并发但注意数据接口可能有频率限制并发过高会被封禁。数据只拉取一次存到本地避免重复请求。用numba加速循环计算场景但对pandas向量化操作帮助不大。观察性能的方法很简单在代码前后记录时间import time start time.time() result_df batch_scan(symbols) print(耗时:, round(time.time() - start, 2), 秒)如果发现某个接口拉数据特别慢可以先缓存数据再计算不要每次都调接口。Flask API服务启动后占用的内存也很小普通云服务器1G内存都够跑但需要限制访问范围不要把接口暴露到公网完全无防护。10. 常见问题与排查方法问题现象可能原因排查方式解决方案akshare 提示找不到接口版本过旧或接口名已更新打印dir(ak)搜索index相关函数升级akshare或更换为当前版本支持的接口获取数据全是NaN数据源返回字段名不一致打印df.head()查看原始列名修改rename映射RSI计算结果为NaN未设置min_periods前几行没有足够数据查看计算函数设置min_periodsrsi_period或直接丢弃前N行信号数量过多阈值太宽松查看信号在K线图上的位置提高min_rebound绝对值或抬高RSI阈值信号数量为零阈值太严格减少min_rebound绝对值放宽RSI阈值“为或降低为回测收益异常高可能存在未来函数检查是否使用了shift(-1)回测信号只用当前及历史数据禁用未来偏移批量扫描部分标的失败数据接口不支持该标的查看打印的异常信息单独处理该标的的数据获取逻辑Flask接口调用超时数据拉取或指标计算阻塞查看服务日志增加数据缓存缩短请求耗时不同数据源同一指数结果不一致复权方式或数据供应商不同对比首尾日期和关键价格统一数据源不要混合使用不同复权设置如果遇到“启动后页面打不开”这类问题在这套系统里对应的是Flask服务无法访问。先检查服务进程是否启动再检查端口是否被占用lsof -i :5000端口冲突时换成其他端口比如5001。11. 最佳实践与扩展方向这套系统虽然简单但很多细节决定它是否可靠。第一次运行建议只用最小参数比如单指数、近一年数据确认数据能拉下来、指标能算出来再扩大到多指数。参数调整要小步走不要同时改五个阈值否则无法判断是哪个修改导致结果变化。数据文件、输出图片、回测报告要分目录管理不要全部堆在一个目录里。工程化之后可以继续扩展的方向有四个。第一加入更多市场维度。例如同时跟踪A股、港股、美股的主要指数对比不同市场在同一宏观预期下的表现能从侧面验证“全球流动性驱动的反弹”是否在所有市场同时出现。第二引入更多技术指标。布林带、MACD、成交量OBV、ATR波动率等都可以作为反弹信号的辅助条件。注意指标不是越多越好指标之间高度相关时增加指标并不会带来信息增量反而容易过拟合。第三接入宏观事件日历。把重要经济数据发布、政策会议时间标记在K线图上可以观察反弹信号是否集中在某些事件窗口附近。这里不讨论具体政策内容只做数据对齐分析。第四把信号服务接入投研流程。每天收盘后自动扫描一批标的生成信号报告推送到即时通讯工具。这样就不需要每天手动打开行情软件逐一确认这是一个很实用的自动化场景。每次跑完数据都要保留一份数据和代码快照。如果后续想回看某个历史信号没有快照就非常难复现。12. 总结回到最初的问题三周前预测的“可交易反弹”到哪里了这个问题不该靠记忆和争论回答而应该靠数据回答。用本文的方法你可以把“可交易反弹”定义成一组明确的规则把时间区间设置到观点提出之后然后跑一遍信号检测和回测看看市场是否真的出现过符合规则的可交易窗口。规则定义不同结论可能不同这正是量化的价值它让观点可被检验。这套方法最值得尝试的点是它把“宏观观点验证”变成了一道可以写代码解决的工程题。最先需要验证的不是信号准不准而是数据接口能不能稳定拉到数据、指标计算能不能跑通。最容易踩的坑是接口版本过时和未来函数污染回测这两个问题排查思路都在前面的表格里。跑通单指数之后建议立刻扩展到多指数批量扫描再封装成HTTP接口或定时任务。这时候你会发现技术分析的核心不是预测而是纪律用统一的规则去观察市场并用回测反复修正规则。把这套流程收藏备用下一次再看到任何“反弹”“反转”“突破”类观点你都可以用代码快速检验。