买疫情坑底和把仓位再配置到 AI 主题这两类操作在过去几年里被反复讨论。真正有价值的问题不是“当时该不该买”而是把决策放到一年后的时间尺度上重新检查回撤触发点是否真的买在了便宜区间AI 再配置的切换信号是否带来了超额收益还是只是行情上涨时顺势加仓。本文会用一次“买入疫情坑底 AI 再配置”的一年期复盘作为主线把这条决策链拆成可验证的数据流程先用回撤阈值定义买入触发点再用均线或动量信号定义 AI 再配置窗口最后用 Python 计算累计收益、年化收益、最大回撤、恢复天数和相对基准的超额收益。文章适合对量化回测、金融数据分析和 AI 工程化复盘感兴趣的开发者。读完可以搭出一套从原始行情清洗、规则回测、指标计算到自动生成复盘摘要的最小工具链。先说明边界这是量化复盘方法演示不是投资建议文中的规则、参数和示例数据都要结合自己的数据重新验证。1. 为什么要做“一年后”复盘而不是只看买入时的逻辑1.1 坑底买入和 AI 再配置是两种不同性质的决策“疫情坑底买入”的底层假设是均值回归市场因为恐慌出现超跌价格偏离长期价值买入后价格会回到合理区间。这个假设成立的前提是下跌确实由情绪驱动而不是基本面永久恶化。如果下跌背后是企业盈利塌方、行业逻辑被颠覆那么“便宜”会变成“更便宜”回撤触发点之后还有更大的回撤。“AI 再配置”的底层假设则是趋势延续和产业迁移AI 主题的景气度能持续向上资金会从旧仓位移到新方向所以要在右侧确认后加大曝光。这里的问题在于右侧信号天然滞后等价格站上均线时行情可能已经走完一大半如果产业趋势是伪命题再配置反而会把前面积累的收益在高位换仓。把两种假设放在一起本质上是一个混合策略左侧用纪律性分批买入降低估值风险右侧用趋势信号切仓位。一年后复盘就是要回答三个问题左侧触发点是否选对了右侧切换信号是早了还是晚了两类决策叠加后的风险是否比单独持有某个方向更差。1.2 复盘重点收益之外还要看回撤、恢复时间和错失成本只比较最终收益是最容易误导人的复盘方式。一年内策略 A 收益 20%、策略 B 收益 15%看起来 A 更好但如果 A 的最大回撤是 35%而 B 只有 10%两者的实际持有体验完全不同。尤其对“坑底买入”这类策略买入后可能还要经历相当长的浮亏期最大回撤、从回撤到前高恢复所需天数才是判断策略是否可执行的关键。建议在一年后的复盘中至少覆盖以下维度。复盘维度核心指标想暴露的问题收益累计收益、年化收益规则整体是否创造了正收益风险最大回撤、波动率买入后是否扛得住浮亏恢复从最大回撤到前高天数资金占用时间是否过长执行买卖次数、换手率规则是否真的可执行、成本是否可控超额相对基准的年化超额赚的是市场上涨的 beta 还是规则本身的 alpha归因分阶段收益是坑底买入贡献多还是 AI 再配置贡献多这组指标不需要一次全部算出来但至少要包含累计收益和最大回撤。后续章节会按这个框架逐步落地代码。2. 用最小 Python 工具链搭出可复现的复盘框架2.1 依赖清单与安装复盘框架不需要复杂组件核心就是数据处理、数值计算和结果持久化三部分。建议在独立的虚拟环境里安装依赖避免污染已有的 Python 环境。python -m venv .reviewenv source .reviewenv/bin/activate pip install --upgrade pip需要安装的核心库如下pandas2.0 numpy1.24 matplotlib3.7 requests2.31其中 pandas 负责行情数据的读取、对齐和信号计算numpy 用于向量化计算matplotlib 用于画出回撤和净值曲线requests 用于在最后一节调用本地 AI 摘要服务。如果还要跑自动化任务可以再加schedule或直接使用系统 crontab。注意版本号只是常见环境下的参考组合。落地前要确认自己的 Python 版本例如 pandas 2.x 要求 Python 3.9 以上避免装完版本冲突。2.2 行情数据准备先让 CSV 结构稳定再谈数据源很多复盘脚本一开始纠结用哪个数据源结果数据源接口一变代码就全废。更稳妥的做法是先约定一个统一的本地 CSV 结构把不同的数据源都清洗到这个结构里后续回测只依赖这个中间格式。建议的 CSV 字段如下。字段类型说明datestr/date交易日期格式统一为 YYYY-MM-DDopenfloat开盘价highfloat最高价lowfloat最低价closefloat收盘价volumefloat/int成交量adjust_factorfloat复权因子可选用于价格复权无论数据来自行情接口还是券商软件导出都要先转换成这个结构再进入复盘流程。这样可以避免后面每次换数据源都改回测代码。2.3 项目目录结构一个适合复盘的极简目录结构如下pandemic_dip_review/ ├── data/ │ ├── sp500_proxy.csv │ ├── ai_basket_proxy.csv │ └── benchmark.csv ├── src/ │ ├── data_loader.py │ ├── signals.py │ ├── simulate.py │ └── metrics.py ├── output/ │ └── review_report.md └── requirements.txtdata目录放原始行情src目录放清洗、信号、模拟和指标函数output目录放每次复盘生成的报告。这样每次跑完都能留档方便对比不同参数的结果。3. 数据预处理复权、对齐和缺失处理决定复盘是否失真3.1 复权方式前复权还是后复权回测中必须处理分红、拆股和送股。如果直接用不复权价格卖出信号可能在除权日被误触发收益计算也会失真。常见选择是前复权或后复权。前复权以当前价格为基准调整历史价格优点是图形连续、容易和实时价格对比缺点是历史上每次复权后过去的价格会变化。后复权以最早价格为基准历史价格不变适合计算真实收益但界面上的价格和实际成交价差异很大。对回测来说后复权更贴近“真实收益”口径但大多数行情软件导出的是前复权数据。建议在 CSV 里保留原始价格和复权因子清洗时再统一计算对应复权价。3.2 用 pandas 完成加载、去重、对齐先把 CSV 加载成统一的 DataFrame并处理重复日期和排序。import pandas as pd def load_price(path: str) - pd.DataFrame: df pd.read_csv(path, parse_dates[date]) df df.drop_duplicates(subset[date], keeplast) df df.sort_values(date).reset_index(dropTrue) df df.set_index(date) return df如果复盘要同时用到普通持仓和 AI 主题持仓需要把两个资产的价格按交易日对齐。最稳妥的做法是取两个序列的交集日期避免某一天只有一个资产有数据导致净值计算错误。def align_assets(prices: dict[str, pd.DataFrame]) - pd.DataFrame: close_dict {name: df[close] for name, df in prices.items()} aligned pd.DataFrame(close_dict) aligned aligned.dropna(howany) return aligned这里用dropna(howany)会删除任一资产缺失的日期。对于停牌较多的个股这种处理会把停牌期间也删掉导致回测区间不连续需要结合具体数据决定是删除还是前向填充。3.3 缺失值、停牌和极端值处理行情数据常见的坑有三个假期缺失、临时停牌和盘中异常价。节假日缺失是可以接受的停牌则需要警惕极端值往往来自数据源错误。def check_data_quality(df: pd.DataFrame) - dict: null_counts df.isnull().sum().to_dict() pct_change df[close].pct_change().abs() extreme_days (pct_change 0.2).sum() return { nulls: null_counts, extreme_move_days: int(extreme_days), }如果发现单日涨跌幅超过 20% 的交易日先确认是不是真实事件比如重大重组或指数调样如果不是就要回看源数据而不是默默删除。缺失值优先用前向填充但连续缺失超过 5 个交易日的建议标记出来人工确认避免掩盖长期停牌。4. 把“坑底买入 AI 再配置”写成可以回测的规则4.1 规则拆解触发、执行、再平衡三阶段把策略写成回测规则前先拆成三个阶段等待期用收盘价计算从阶段高点的回撤当回撤达到阈值例如 -25%触发坑底买入信号。买入期触发后按固定比例分批买入避免一次性抄底抄在半山腰。再配置期价格重新站上长期均线说明趋势恢复此时把一部分仓位切换到 AI 主题持仓完成再配置。这套规则的关键是要明确每个动作的具体条件否则回测无法复现。例如“站上均线”需要明确是当日收盘价大于 120 日均线还是连续多日大于均线分批买入需要明确每批比例和间隔条件。4.2 用向量化方式算回撤和信号回撤的计算公式是当日收盘价 / 历史最高收盘价 - 1。用 pandas 的cummax可以一次性算出完整回撤序列。def calc_drawdown(close: pd.Series) - pd.Series: return close / close.cummax() - 1.0 def gen_signal(close: pd.Series, trigger: float -0.25, ma_window: int 120) - pd.DataFrame: dd calc_drawdown(close) ma close.rolling(ma_window).mean() signal pd.DataFrame({ close: close, drawdown: dd, ma: ma, hit_trigger: dd trigger, above_ma: close ma, }) return signal这里的trigger表示回撤阈值ma_window是判断右侧行情的均线窗口。回撤触发和站上均线是两种完全不同的信号前者是左侧防御信号后者是右侧趋势信号不要混在一起判断。4.3 简化版资金模拟器下面是一个教学用的简化资金模拟器。它把每次交易都按当日收盘价成交忽略手续费和滑点目的只是把规则跑通。真实回测还需要补充下单价格偏移、税费模型和持仓数量约束。def simulate(signal: pd.DataFrame, init_cash: float 100_000.0, lots: tuple (0.2, 0.3, 0.5), ai_share: float 0.3) - pd.Series: close signal[close] cash init_cash shares 0.0 # 普通持仓数量 ai_shares 0.0 # AI 主题持仓数量 lot_idx 0 rebalanced False nav [] for date, row in signal.iterrows(): price float(row[close]) # 触发回撤买入按初始资金比例分批投入 if row[hit_trigger] and lot_idx len(lots): budget init_cash * lots[lot_idx] shares budget / price cash - budget lot_idx 1 # 价格站上均线后把对应比例市值从普通持仓切换到 AI 持仓 if row[above_ma] and not rebalanced and (shares 0 or cash 0): total_value cash shares * price ai_shares * price target_ai_value total_value * ai_share current_ai_value ai_shares * price diff_value target_ai_value - current_ai_value if diff_value 0: buy_shares diff_value / price sell_shares min(buy_shares, shares) shares - sell_shares ai_shares sell_shares remaining buy_shares - sell_shares if remaining 0 and cash remaining * price: ai_shares remaining cash - remaining * price rebalanced True nav.append(cash shares * price ai_shares * price) return pd.Series(nav, indexsignal.index)这个实现里分批买入用的是“初始资金的比例”而不是“剩余现金的比例”所以三批合计就是初始资金的 100%。rebalanced只允许再配置一次避免在震荡行情里反复切换仓位。教学版本最大的问题是每次触发后第二天继续触发会把三批快速买完可以在真实版本里加一个“触发冷却期”例如触发后 10 个交易日内不再合计买入。4.4 关键参数速查参数的选择直接影响回测结果需要理解每个参数改大改小的含义。参数含义示例值调大的影响调小的影响trigger回撤触发阈值-0.25买入更晚更安全但也更贵买入更早可能抄在半山腰lots分批买入比例(0.2, 0.3, 0.5)前期仓位重风险高前期仓位轻可能买不够ma_window再配置均线窗口120信号更平滑切换更晚信号更敏感切换更早ai_shareAI 再配置比例0.3暴露更集中波动更大暴露不足超额收益有限这些参数不应该通过反复试错来“调出最好看的结果”而应该基于对策略逻辑的理解预先设定然后用样本外数据验证。5. 一年后的结果怎么算收益、回撤、与基准对比5.1 核心指标计算函数净值序列是模拟器的最终输出后续指标都从它派生。下面是几个最常用的指标函数。def annualized_return(nav: pd.Series, periods: int 252) - float: total_return nav.iloc[-1] / nav.iloc[0] - 1.0 years len(nav) / periods return (1.0 total_return) ** (1.0 / years) - 1.0 def max_drawdown(nav: pd.Series) - float: return float((nav / nav.cummax() - 1.0).min()) def calmar_ratio(nav: pd.Series, periods: int 252) - float: ar annualized_return(nav, periods) mdd abs(max_drawdown(nav)) return ar / mdd if mdd 0 else 0.0年化收益默认按每年 252 个交易日换算。最大回撤是净值从阶段高点回落到低点的最大幅度没有正负号歧义直接取最小值即可。Calmar 比率是年化收益与最大回撤绝对值的比值代表“每承担一单位回撤能换来多少年化收益”适合用来比较不同规则。5.2 和基准对比先确定“不买”或“一直拿着”是什么结果一个策略好不好不能只看自身收益还要和基准比。对这套规则至少要有两个基准一个是“始终不参与、全部持有现金”的结果另一个是“从最开始就买入持有”的结果。前者代表错失成本后者代表单纯暴露在市场里的收益。对比结果可以用如下代码输出。def compare_with_benchmark(strategy_nav: pd.Series, benchmark_nav: pd.Series) - pd.DataFrame: rows { strategy: annualized_return(strategy_nav), benchmark: annualized_return(benchmark_nav), } rows[excess] rows[strategy] - rows[benchmark] return pd.DataFrame([rows], index[annual_return])输出结果类似下面的形式。注意以下数值是示例格式用于演示输出结构真实结果取决于你的行情数据、时间区间和参数。strategy benchmark excess annual_return 0.1875 0.1211 0.0664如果策略的年化收益低于买入持有基准说明规则没有创造 alpha只是在和 beta 同步波动。此时需要回到信号层检查触发和再配置时机而不是继续优化参数。5.3 用归因区分 beta 和规则 alpha更精细的做法是把回测区间分成两个阶段坑底买入阶段和 AI 再配置阶段分别计算每段贡献。def attribution(strategy_nav: pd.Series, benchmark_nav: pd.Series) - dict: def seg_return(nav: pd.Series, start: str, end: str) - float: seg nav.loc[start:end] return seg.iloc[-1] / seg.iloc[0] - 1.0 return { dip_buy_stage: round(seg_return(strategy_nav, 2020-03-01, 2020-08-01), 4), ai_realloc_stage: round(seg_return(strategy_nav, 2021-01-01, 2021-12-31), 4), }这里的分段日期只是示例实际要根据信号的触发时间和再配置时间来确定。归因的核心意义是如果收益全部来自坑底买入后的反弹AI 再配置反而拉低了收益那么“切换 AI 主题”这个假设就要重新审视反过来如果收益主要来自再配置后的 AI 行情说明右侧信号确实抓住了产业趋势。6. 这个复盘最容易踩的五个坑6.1 幸存者偏差让“AI 篮子”看起来比实际更美复盘 AI 再配置时如果直接用当前还活着的 AI 龙头股构建持仓池就会引入幸存者偏差。那些当年被当成 AI 概念、后来业绩暴雷或退市的公司因为已经不在当前指数里会被自动排除回测结果自然好看。正确做法是回到再配置发生的时点使用当时可买入的完整股票池包括后来下跌和退市的公司。如果拿不到完整成分股至少要在报告里注明“持仓池存在幸存者偏差结果可能高估策略表现”。6.2 前视偏差信号和成交用到了未来信息前视偏差是量化回测里最隐蔽的问题。典型场景包括用当天的收盘价计算信号却假设当天能以开盘价成交或者用整个区间的数据做标准化导致历史信号包含了未来统计量。检查方法是把信号和成交日逐笔打印出来确认每个动作只使用了 T 日及之前的数据。建议在代码里统一约定所有指标计算用 T 日收盘价成交也按 T 日收盘价禁止跨日使用数据。6.3 交易成本、滑点和冲击成本被忽略教学模拟器可以忽略成本但一年期复盘如果换手不少成本影响不可忽略。A 股双边交易成本通常包含佣金、印花税和滑点港股和美股规则不同要按实际市场设置。建议至少按单边 0.1% 到 0.3% 的区间测试策略敏感性。如果加入成本后策略从盈利变成亏损说明规则依赖过高频的切换不适合实际执行。6.4 用指数替代个股掩盖分化用 AI 主题指数代替 AI 个股池做回测优点是数据干净缺点是指数会平滑掉个股分化。实际买卖时你买的是具体公司不是指数。指数里可能有亏损股也可能有你根本买不到的一篮子成分。如果数据允许尽量用自定义等权股票池至少覆盖 10 到 20 只相关标的。这样能看出再配置规则在个股层面是否依然成立。6.5 用事后最优参数冒充事前规则反复调整 trigger、ma_window 和 ai_share直到回测曲线最漂亮这是典型的过拟合。看起来是规则在赚钱其实是参数在匹配历史行情。正确做法是先写下一套事前规则用一段不参与调参的历史数据做样本外验证。只要样本外表现和样本内差异过大就说明参数过拟合需要退回更保守的参数。坑现象检查方式处理建议幸存者偏差持仓池全是活下来的股票收益高得异常核查选股日是否包含退市和暴跌公司使用“当时可买入”的全样本前视偏差信号和成交存在时间错位逐笔打印信号与成交日统一只用 T 日及之前数据成本被忽略换手频繁但收益异常好统计换手率加成本重跑至少按单边 0.1%-0.3% 测试指数替代个股分化被指数抹平对比个股与指数走势用自定义股票池回测事后参数参数反复调优后曲线完美做样本外测试固定事前规则再验证7. 把复盘工程化定时跑批、结果入库、AI 自动摘要7.1 单次脚本的局限上面这套代码跑一次复盘没问题但一年后真正要用的场景是每个月或每个季度自动更新行情重新计算一次指标叠加新的净值数据然后生成一份可读的对比报告。单次脚本的问题在于没有状态上次跑的结果、参数和结论都散落在终端里无法追溯。工程化第一步是引入结果存储至少把每次复盘的关键指标存下来方便后面对比不同时间点的结论变化。7.2 定时任务和结果入库用 SQLite 存储每次复盘的指标快照轻量且无需额外服务。import sqlite3 import json conn sqlite3.connect(backtest.db) conn.execute( CREATE TABLE IF NOT EXISTS backtest_result ( id INTEGER PRIMARY KEY AUTOINCREMENT, snapshot TEXT NOT NULL, strategy_name TEXT NOT NULL, metrics_json TEXT NOT NULL, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) metrics { annual_return: round(ar, 4), max_drawdown: round(mdd, 4), calmar: round(calmar, 4), } conn.execute( INSERT INTO backtest_result(snapshot, strategy_name, metrics_json) VALUES (?, ?, ?), (2024-03-01, dip_buy_ai_realloc, json.dumps(metrics, ensure_asciiFalse)), ) conn.commit()存储在数据表里的metrics_json就是一次复盘的完整快照。后续可以做区间对比也能在生成报告时直接读取。定时执行最简单的方式是系统 crontab例如每个月首个交易日早上 8 点跑一次0 8 1 * * cd /path/to/pandemic_dip_review .reviewenv/bin/python run_review.py output/review.log 21如果团队已经有 Airflow、DolphinScheduler 或 Jenkins可以把run_review.py包装成一个任务节点输入是行情 CSV 路径输出是 SQLite 快照和报告文件。7.3 用 AI/LLM 生成复盘摘要并防幻觉指标算完后可以调用本地部署的 LLM 服务生成自然语言摘要让非量化背景的读者也能看懂。这里强调本地部署是为了避免把行情和持仓数据发送到外部服务降低数据外泄风险。import requests import json def build_summary(metrics: dict, endpoint: str http://127.0.0.1:8000/v1/chat/completions) - str: prompt ( 你是量化复盘助手。只根据下面 JSON 里的指标做中文解读 不要编造指标中没有的数值不要给出买卖建议。\n json.dumps(metrics, ensure_asciiFalse) ) resp requests.post( endpoint, headers{Authorization: Bearer sk-local}, json{ model: local-model, messages: [{role: user, content: prompt}], temperature: 0.2, }, timeout30, ) resp.raise_for_status() return resp.json()[choices][0][message][content]这段代码假设本地服务兼容 OpenAI 的/v1/chat/completions接口。如果团队使用 Java 技术栈可以换成 Spring AI 的 ChatClient把指标 JSON 作为 user message 传入效果等价。这里要特别处理 LLM 的“AI 幻觉”问题。模型很可能在解读时补充出指标之外的结论比如擅自断言“该策略未来仍然有效”。缓解办法有两个一是在 system prompt 里明确禁止补充数据外的结论二是在生成摘要后做一次字段校验把原文里的关键指标回填到固定模板中再让 LLM 只负责润色。审计日志也应该保留模型输出原文便于追溯。在这个环节复盘工具已经不再是“一次性脚本”而是一个完整的数据管道行情清洗、规则模拟、指标计算、结果存储、自动摘要。后续可以继续扩展的方向包括接入更多 AI 主题标的池、增加滚动回测窗口、把指标快照展示到简单的 Web 页面或者用 AI Agent 监听指标异常并主动告警。对新手来说最有价值的练习不是追求更复杂的模型而是先把这套最小闭环跑通拿到一份真实行情写清规则算出指标承认偏差最后用一段可审计的文字把结论记录下来。买疫情坑底和 AI 再配置的讨论在未来还会反复出现但只要你手里有一套能复现、能检查、能更新的复盘流程每一次讨论都会变成一次更干净的数据验证而不是情绪判断。