Python量化实战：市场温度计算与暴跌短线信号回测

常士杉8.13市场温度里有一句被反复讨论的话暴跌战法完全就是短线不是长线。这句话放在交易里很好理解放在程序里怎么落地这次我们围绕这个观点直接拆一个可运行的量化分析工具用 Python 读取行情数据计算市场温度识别暴跌后的短线信号再做批量回测最后还能通过 HTTP 接口对外提供服务。整个过程不依赖任何商业化交易软件普通笔记本就可以跑重点是把“短线”变成一套可验证的规则。先说结论这套思路不需要显卡不需要高配置服务器。核心依赖只有 pandas、numpy、matplotlib、flask数据源可以先用本地 CSV后续再换实时行情接口。支持整目录批量扫描也支持 API 调用。工具本身不是预测涨跌而是把短线的观察条件指标化、可视化、可回测。这样做的好处是任何一版规则改动后都能立刻看到历史信号和回测结果不用凭感觉判断一个战法到底有没有效。这篇文章适合三类人想验证超跌反弹策略的量化入门者需要把财经观点转成策略代码的开发者以及做周度复盘工具的技术人员。全文只聊技术实现不构成投资建议。下面所有阈值、窗口、参数都是示例你可以按自己的交易习惯调整。1. 核心能力速览能力项说明项目类型市场温度量化分析与短线策略验证工具数据来源本地 CSV 行情数据可扩展实时行情接口核心功能市场温度计算、暴跌信号识别、短线回测、信号导出、HTTP 接口运行平台Windows / Linux / macOSPython 3.9 以上硬件门槛CPU 即可建议 4 核以上内存 8GB 以上显存占用0不需要 GPU启动方式命令行脚本 / Jupyter Notebook / Flask API是否支持 API支持可启动 HTTP 服务是否支持批量支持按目录批量读取 CSV 并输出信号适合人群量化学习者、短线策略研究者、技术开发人员从表里可以看出来这个工具的定位不是实盘喊单而是辅助验证。尤其是“暴跌战法完全就是短线”这个观点如果不做回测很难判断它只在特定行情下有效还是长期都有效。用代码验证之后至少能知道过去一年里这类信号出现过多少次胜率和盈亏比大概是什么水平。2. 适用场景与使用边界这种工具最适用的场景是策略研究。比如你关注某个股票池想在每次市场大跌之后观察哪些标的出现短期企稳迹象就可以每天跑一遍信号扫描把结果存成 CSV 慢慢看。也可以把工具当成复盘助手通过 K 线图自动标注出暴跌后的买入信号位置方便快速浏览。它不适合用来自动下单。原因很简单技术模型不能处理所有突发情况尤其是个股停牌、涨跌停限制、消息面冲击、流动性不足这些都不是简单指标能覆盖的。把工具输出的信号直接接进实盘交易风险非常高。比较稳妥的用法是人工复核信号再决定是否进一步分析。合规方面也要注意。行情数据来源必须合法不要使用未经授权的数据接口。如果分析报告中涉及具体个股名称建议做脱敏处理。文中提到的常士杉市场温度观点仅作为策略思路来源不代表本文推荐任何个股或交易行为。任何策略都要先经过模拟和充分测试再考虑小资金验证不能把过去的表现直接等同于未来收益。3. 环境准备与前置条件工具开发不复杂依赖也很少。建议用虚拟环境隔离避免把全局 Python 环境搞乱。下面是在命令行创建虚拟环境并安装依赖的步骤。# 创建并激活虚拟环境 python -m venv venv source venv/bin/activate # Windows 下运行 venv\Scripts\activate激活虚拟环境后再安装依赖。核心库只需要 pandas、numpy、matplotlib、flask另外可以加一个 schedule 用来做定时任务但这里不是必须的。pip install pandas numpy matplotlib flask schedule如果需要长期维护可以把依赖写进 requirements.txt。pandas1.5.0 numpy1.23.0 matplotlib3.6.0 flask2.2.0 schedule1.2.0安装完成后可以检查一下 pandas 是否能正常读取 CSV。这个工具不涉及 GPU也不需要 CUDA、PyTorch 之类的东西整链路非常轻量。如果你的机器上已经装了 Anaconda也可以直接用 conda 环境跑不需要额外安装 Python。4. 项目结构与数据准备建议用下面的目录结构管理工程文件输入数据、输出结果、源码分开后面批量跑任务会方便很多。market_temperature/ ├── data/ │ └── daily/ │ ├── 000001.csv │ └── 600000.csv ├── output/ ├── src/ │ ├── temperature.py │ ├── signal.py │ ├── backtest.py │ └── api.py ├── main.py └── requirements.txt行情数据这里统一使用 CSV 格式每一列分别表示日期、开盘价、最高价、最低价、收盘价、成交量和成交额。具体列名可以自定义但建议保持一致。date,open,high,low,close,volume,amount 2024-01-02,10.0,10.5,9.9,10.4,100000,1000000 2024-01-03,10.4,10.8,10.2,10.6,120000,1200000如果你手上没有现成行情数据可以先做一套模拟数据来验证流程。模拟数据不等于真实行情但可以帮你确认代码有没有问题。下面这段代码会生成一段走势包含明显下跌和反弹的数据用来测试暴跌战法信号。import pandas as pd import numpy as np np.random.seed(42) date_range pd.date_range(start2024-01-01, periods250, freqD) close 100 closes [] for i in range(250): close close * (1 np.random.normal(0, 0.02)) if 80 i 85: close close * 0.92 # 模拟一次快速下跌 closes.append(close) df pd.DataFrame({ date: date_range, close: closes, open: closes, high: closes, low: closes, volume: np.random.randint(80000, 200000, size250) }) df.to_csv(data/daily/000001.csv, indexFalse) print(df.tail())这个模拟数据生成了 250 个交易日的价格序列并在第 80 到第 85 个交易日内加入了一段快速下跌刚好能用来观察暴跌战法信号识别。实际使用时把 CSV 替换成真实行情数据即可代码逻辑不用改。5. 市场温度与暴跌战法策略实现市场温度没有统一标准不同人定义不同。这里用三个可计算的指标来综合衡量市场宽度、量能温度和均线偏离度。三个指标都归一化到 0 到 100 之间最后组合成一个温度值。温度值越低说明市场整体越冷越接近超跌区域。首先是市场宽度。最简单的计算方式是统计上涨股票数量在全市场中的占比。如果只有单只股票数据可以用该股过去 N 日的涨跌分布来近似观察个股强弱。下面代码实现的是经典市场宽度版本需要输入一个包含多只股票收益率的 DataFrame。def market_width(returns: pd.DataFrame, window: int 20): up_count (returns 0).rolling(window).sum() width up_count / returns.shape[1] * 100 return width然后是量能温度。量能温度用来判断下跌过程中是否出现放量。放量暴跌和缩量阴跌的含义完全不同。这里用当前成交量除以过去 20 日平均成交量再做归一化。def volume_temperature(volume: pd.Series, window: int 20): avg_volume volume.rolling(window).mean() ratio volume / avg_volume temp np.clip(ratio * 50, 0, 100) return temp均线偏离度用来衡量价格离短期均线有多远。当价格大幅低于 5 日均线时短线超卖的概率会上升但超卖不等于马上反转需要等信号确认。def deviation_temperature(close: pd.Series, window: int 5): ma close.rolling(window).mean() deviation (close - ma) / ma * 100 # 偏离 -10% 时给 0 分偏离 0% 时给 50 分偏离 10% 时给 100 分 temp np.clip(50 - deviation * 5, 0, 100) return temp三个指标计算完成后可以按一定权重合成市场温度。这里采用 0.5、0.2、0.3 的权重分别对应市场宽度、量能温度和均线偏离度。def market_temperature(returns: pd.DataFrame, volume: pd.DataFrame, close: pd.DataFrame): width market_width(returns) vol_temp volume_temperature(volume) dev_temp deviation_temperature(close) temperature 0.5 * width 0.2 * vol_temp 0.3 * dev_temp return temperature暴跌战法的信号识别逻辑本质上是“超跌后出现短线企稳”。这里定义一组可解释的条件第一当日收盘价从近 20 日最高点回撤超过 10%第二RSI 指标低于 30第三收盘价重新站上 3 日均线。三者同时满足时输出一个短线信号。def rs_index(close: pd.Series, window: int 14): delta close.diff() gain delta.clip(lower0).rolling(window).mean() loss -delta.clip(upper0).rolling(window).mean() rs gain / (loss 1e-10) return 100 - (100 / (1 rs)) def crash_short_signal(df: pd.DataFrame, crash_threshold: float -0.10): df df.copy() df[high_20] df[close].rolling(20).max() df[drawdown] df[close] / df[high_20] - 1 df[ma3] df[close].rolling(3).mean() df[rsi] rs_index(df[close]) df[rebound] df[close] df[ma3] df[signal] ( (df[drawdown] crash_threshold) (df[rsi] 30) (df[rebound]) ).astype(int) return df这里有两点需要说明。一是 RSI 的计算用了简化方式实际使用可以换成更严谨的 Wilder 平滑算法。二是signal为 1 只代表满足技术条件不构成买入建议。信号质量需要通过回测来评估不能只看单个信号是否出现。6. 功能测试与效果验证数据文件和策略函数准备好之后先用命令行脚本跑通整个流程。main.py负责读取指定目录下所有 CSV对每只股票计算信号并把结果输出到 output 目录。import pandas as pd from pathlib import Path from src.signal import crash_short_signal def process_one(file_path: Path, output_dir: Path): df pd.read_csv(file_path, parse_dates[date]) df crash_short_signal(df) output_file output_dir / f{file_path.stem}_signal.csv df.to_csv(output_file, indexFalse) return output_file if __name__ __main__: input_dir Path(data/daily) output_dir Path(output) output_dir.mkdir(exist_okTrue) for f in sorted(input_dir.glob(*.csv)): out process_one(f, output_dir) print(f处理完成: {f.name} - {out})运行方式如下python main.py如果模拟数据生成正确最终可以在 output 目录里看到000001_signal.csv里面多出drawdown、ma3、rsi、signal等列。用文本编辑器或 pandas 打开筛选signal 1的行就能看到具体触发日期。import pandas as pd result pd.read_csv(output/000001_signal.csv, parse_dates[date]) signals result[result[signal] 1] print(f信号数量: {len(signals)}) print(signals[[date, close, drawdown, rsi, signal]])下一步是简单回测。这里采用一个最粗糙的验证方式信号出现后分别计算未来 1 日、3 日、5 日的收益率然后统计平均收益和胜率。注意这种回测没有考虑交易成本、滑点和涨跌停无法买入的情况结果只能用来做横向对比。def quick_backtest(df: pd.DataFrame, forward_days: int 5): df df.copy() df[future_close] df[close].shift(-forward_days) df[future_return] df[future_close] / df[close] - 1 signal_df df[df[signal] 1] if len(signal_df) 0: return {} return { signal_count: len(signal_df), avg_return: signal_df[future_return].mean(), win_rate: (signal_df[future_return] 0).mean(), } stat quick_backtest(result, forward_days5) print(stat)从回测输出可以直观看到两个问题一是信号数量够不够多如果过去一年只有 2 次信号统计意义很弱二是平均收益和胜率是否稳定如果某一次极端行情把均值拉高就需要去看中位数。建议后续补充最大回撤、盈亏比和逐笔明细。7. 接口 API 与批量任务工具不能只停留在脚本层面最好能提供一个 HTTP 接口方便其他系统调用。这里用 Flask 实现一个最简单的接口接收日期范围和 CSV 文件路径返回最近一次信号信息。生产环境加入鉴权和限流。from flask import Flask, request, jsonify import pandas as pd from src.signal import crash_short_signal app Flask(__name__) app.route(/analyze, methods[POST]) def analyze(): payload request.get_json(forceTrue) file_path payload.get(file_path) if not file_path: return jsonify({error: file_path is required}), 400 try: df pd.read_csv(file_path, parse_dates[date]) df crash_short_signal(df) latest df[df[signal] 1].tail(1) if len(latest) 0: return jsonify({signal: False}) row latest.iloc[0] return jsonify({ signal: True, date: str(row[date]), close: row[close], drawdown: row[drawdown], rsi: row[rsi] }) except Exception as e: return jsonify({error: str(e)}), 500 if __name__ __main__: app.run(host127.0.0.1, port8000)启动接口服务python src/api.py接口服务启动后可以用 curl 或 Python 发一个 POST 请求验证。curl -X POST http://127.0.0.1:8000/analyze \ -H Content-Type: application/json \ -d {file_path: data/daily/000001.csv}正常返回结果类似下面这样{ signal: true, date: 2024-06-20, close: 95.2, drawdown: -0.12, rsi: 25.3 }批量任务方面核心不是循环读取文件而是加日志和失败重试。因为数据量变大后总有某只股票的 CSV 格式不对比如日期列缺失、价格字符串带单位、除零错误等。建议用一个批量任务表记录每个文件的处理状态。import logging from pathlib import Path from time import sleep logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) def run_batch(input_dir: Path, output_dir: Path, retry: int 2): output_dir.mkdir(exist_okTrue) for file_path in sorted(input_dir.glob(*.csv)): for attempt in range(retry): try: out process_one(file_path, output_dir) logging.info(f成功: {file_path.name} - {out}) break except Exception as e: logging.warning(f失败: {file_path.name}, 第{attempt 1}次: {e}) sleep(1) else: logging.error(f持续失败: {file_path.name})这种设计可以避免单个文件异常导致整个批量任务中断。如果数据源是实时行情接口还要考虑请求频率限制建议把每次请求间隔设为随机 0.5 到 1.5 秒避免短时间高频请求触发限流。8. 资源占用与性能观察这个工具不需要 GPU所以显存占用是 0。性能瓶颈主要出现在全市场股票池处理上。假设有 5000 只股票的 CSV每只 250 个交易日数据量并不大pandas 完全可以吃下。但如果把 5000 只股票的日线全部读入一个 DataFrame内存占用会明显上升建议按目录分批处理。观察资源占用可以用系统自带的任务管理器或top命令。处理单只股票时内存占用通常很低但如果同时进行多进程并行处理内存会成倍增长。这里建议第一版不要做并行先跑完单线程批量任务确认逻辑正确后再考虑用 multiprocessing 或 ThreadPoolExecutor 加速。from concurrent.futures import ProcessPoolExecutor def process_wrapper(args): file_path, output_dir args return process_one(file_path, output_dir) if __name__ __main__: input_dir Path(data/daily) output_dir Path(output) files list(input_dir.glob(*.csv)) with ProcessPoolExecutor(max_workers4) as executor: results list(executor.map(process_wrapper, [(f, output_dir) for f in files]))如果行情数据源采用 HTTP 接口实时拉取还要关注网络延迟和接口超时。建议把每次请求的外层超时时间设置为 10 秒避免单次请求卡死批量任务。如果端口 8000 被占用可以在启动时换一个端口比如--port 8001或者通过环境变量指定。# 使用自定义端口启动 API 服务 python src/api.py --port 8001如果 API 服务需要长时间运行建议用 nohup 或后台进程方式启动同时把日志写到独立文件。nohup python src/api.py --port 8001 api.log 21 9. 常见问题与排查方法问题现象可能原因排查方式解决方案启动后 CSV 读取报错列名不匹配或编码错误打印文件头检查列名和原始编码统一列名读取时指定encodinggbk或encodingutf-8日期解析失败日期格式不统一定位报错行观察日期字符串用pd.to_datetime(..., format...)明确格式RSI 计算结果为空数据长度不够检查 DataFrame 行数过滤掉少于 30 条记录的文件均线或回撤计算大量 NaN前 N 根K线没有足够窗口查看前 20 行输出回测时对 NaN 信号做填充或剔除行情数据出现除零错误成交量为 0 或平均价为 0检查异常行在计算分母处加极小值1e-10批量任务卡在某个文件文件读取有异常导致死循环看日志卡在哪个文件名增加异常捕获和超时控制API 请求超时全市场扫描耗时长检查后端日志改为异步任务或分批提交端口被占用之前服务未关闭用lsof -i:8000查看进程杀掉旧进程或更换端口信号数量过度密集阈值设得过于宽松打印信号触发日提高回撤阈值或加入 RSI 过滤条件回测结果波动太大样本数太少统计信号次数扩大股票池或延长数据时间范围排查时要记住一条原则先看日志再看数据最后改参数。大多数问题都不是代码逻辑问题而是数据格式不统一。10. 最佳实践与使用建议第一版代码建议先跑模拟数据不急着接真实行情。模拟数据能快速暴露逻辑问题比如信号触发位置是否合理、回测收益曲线是否异常。真实行情数据接入后先用一个不包含近期数据的测试集做验证避免用到未来函数。第二次迭代时可以把参数抽成配置文件不要每次改代码。比如回撤阈值、RSI 阈值、均线窗口都放到config.yaml里这样在调整战法时只需要改配置不需要重新跑代码。crash_threshold: -0.10 rsi_threshold: 30 rsi_window: 14 ma_window: 3 volume_window: 20策略验证时不要把参数调到历史表现最好。过度拟合的参数在实盘中很容易失效。更稳妥的做法是把数据集分成训练期和验证期先用训练期调参再用验证期检查稳定性。批量任务必须保留日志。日志里至少要记录文件路径、处理状态、耗时、失败原因。没有日志的批量任务一旦跑挂很难定位。输出结果建议按日期和股票代码命名例如output/20240813/000001_signal.csv方便后续做历史回溯。如果要把工具分享给团队建议在 README 里写清楚数据格式、依赖版本、运行命令和免责声明。使用实时行情数据时必须确认数据服务商的授权范围。涉及个股分析时尽量只做技术特征的客观展示不输出任何带推荐性质的表述。总结与下一步常士杉市场温度里强调“暴跌战法完全就是短线不是长线”这句话真正落地到代码后会变成一个条件明确的信号系统。你可以用它验证历史信号也可以每天跑一遍全市场扫描但要记住工具只负责把规则执行出来不负责预测未来。对任何策略来说回测统计、参数验证、风险控制都比单次信号重要得多。建议先收藏这篇文章然后按顺序跑三件事生成模拟数据、运行main.py批量扫描、调用/analyze接口验证结果。跑通之后再把真实行情数据和自己的交易规则填进去。这个项目的下一步可以扩展的方向有很多接入更多技术指标、增加持仓周期优化、输出图形化日报或者对接推送服务。但所有扩展的前提都是先保证基础信号识别和批量回测是稳定的。