Python股票行情数据采集清洗复权与回测可视化实战

简介：面向希望用Python打通股票数据采集、清洗、分析与建模流程的金融数据分析学习者，这份实战资料围绕爬虫获取行情、新闻与公司公告，Pandas完成缺失值异常值处理与格式统一，NumPy/SciPy承担统计计算，Matplotlib/Seaborn绘制K线图、成交量与波动性图表，并用Scikit-learn尝试线性回归、决策树、随机森林等预测模型；同时涉及Tushare、Wind等实时接口和Flask/Django结果展示，适合具备Python基础、想进入量化与金融分析场景的读者系统查漏补缺。资源包约4.47MB，文件总数与类型明细暂未提供，内容组织以页面展示为准。页面显示已有3659人学习，说明该主题具有一定关注度。读者可据此梳理从数据抓取到模型评估的完整链路，理解反爬策略、数据清洗、可视化解读、模型评估与Web部署的衔接方式，并积累实时数据接入与事件驱动分析的实践线索。

1. 从一份行情数据包说起：它解决的是“数据从哪来、怎么用”这两件事

做股票数据分析的人，十个里有八个卡在第一步：数据。用 python 爬虫从公开行情页抓下来的表，字段名是中文的，日期是字符串，成交量带千分位逗号，停牌那天干脆没这一行；再从别处补一份 CSV，复权口径又对不上。后面写指标、跑回测，全在给这些脏数据擦屁股。这套「基于股票大数据分析 Python 实战」的资源，价值就在于把采集、清洗、复权、指标、回测、可视化串成一条能跑通的链路，而不是丢给你一堆互相不认识的脚本。它适合想自己搭一条日频行情分析流水线的开发者、做课程设计需要完整闭环的学生，以及被第三方数据接口限流折腾过的量化入门者。不适合追求毫秒级 tick 和实盘下单的团队——这套东西的定位是日频研究，别拿它当交易系统用。

2. 行情数据采集：从公开页面到本地表的完整链路

2.1 三类抓取对象的取舍：静态表格、JSON 接口还是批量文件

动手写 python 爬虫之前，先想清楚抓什么形态的数据，这决定了后面解析代码长什么样。常见的目标有三种：返回完整 HTML 的行情列表页、返回 JSON 的异步接口、以及按交易日打包的压缩文件。三者的差别不只是“好不好抓”，还在于字段稳定性、增量成本和失败恢复的难度。

我一般这么组合：初始化用批量文件把几年历史一次性灌进库，日常增量走 JSON 接口按交易日拉当天数据，HTML 表格只在前两者都拿不到某些字段（比如换手率、流通股本）时作为补充。这样做的好处是增量路径很短，每天只处理几十行，出问题容易定位；而全量重建的频率很低，就算接口改版也有时间从容修。

选源阶段有一条硬约束必须提前定下来：只抓公开的、允许访问的数据，控制请求频率，不去碰需要登录态或者明确禁止采集的页面。这不是道德说教，是工程现实——高频请求换来的往往是 IP 被封、返回空壳页面，你后面所有排错时间都花在跟风控斗智上，得不偿失。

2.2 请求层封装：会话复用、重试与限速

直接写requests.get的脚本，跑几十个标的就开始出问题：连接不复用导致握手开销大，偶发 502 直接抛异常中断整个循环，固定 sleep 又让整批任务慢得离谱。合理的做法是把请求层封成一个带重试和抖动的会话对象，所有抓取函数共用它。

import time import random import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(total_retry=3, backoff=0.5, pool_size=4): """构造一个带重试、连接池复用的会话""" session = requests.Session() retry = Retry( total=total_retry, # 最多重试 3 次 backoff_factor=backoff, # 退避基数，第 n 次等待 backoff * 2^n 秒 status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"], # 只对 GET 重试，避免重复提交 ) adapter = HTTPAdapter(max_retries=retry, pool_connections=pool_size, pool_maxsize=pool_size) session.mount("http://", adapter) session.mount("https://", adapter) session.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; research-collector/1.0)", "Accept": "application/json,text/html;q=0.9,*/*;q=0.8", }) return session def polite_get(session, url, params=None, min_interval=1.2, timeout=10): """限速 + 抖动 + 状态码校验，返回响应对象""" time.sleep(min_interval + random.uniform(0.0, 0.6)) resp = session.get(url, params=params, timeout=timeout) resp.raise_for_status() return resp

这段代码有三个参数值得掰开说。total_retry控制重试次数，设太大在对方服务真挂了的时候只会拖长整体耗时，3 次是经验值。backoff_factor是退避基数，实际等待时间是0.5 * 2^n秒，配合min_interval形成的总间隔大概在 2 到 3 秒之间，既不会把对方打疼，也不会慢到无法接受。min_interval上加的random.uniform(0, 0.6)不是玄学，固定节奏的请求特征太明显，加抖动让访问曲线更接近正常用户。

raise_for_status()必须在解析之前调用。很多新手直接resp.json()，遇到 403 返回的错误页就抛 JSON 解析异常，报错信息完全指不到真实原因，白白浪费半小时。想更稳一点，可以在外层再包一层，把失败请求的 URL、状态码、响应前 200 字符写进日志文件，出问题时直接翻日志比重跑一遍快得多。

2.3 解析与入库：字段对齐和幂等写入

抓下来只是半成品，真正的数据结构要在入库这一步定死。我习惯在本地建一张 SQLite 表，主键设成(symbol, trade_date)，这样同一只票同一天无论重抓多少次都只会更新不会重复。字段统一用英文短名，避免中文列名在后续 pandas 操作里到处加引号。

import sqlite3 import pandas as pd DDL = """ CREATE TABLE IF NOT EXISTS daily_bar ( symbol TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, amount REAL, adj_factor REAL, src TEXT, PRIMARY KEY (symbol, trade_date) ); CREATE INDEX IF NOT EXISTS idx_bar_date ON daily_bar(trade_date); """ UPSERT = """ INSERT INTO daily_bar (symbol, trade_date, open, high, low, close, volume, amount, adj_factor, src) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(symbol, trade_date) DO UPDATE SET open=excluded.open, high=excluded.high, low=excluded.low, close=excluded.close, volume=excluded.volume, amount=excluded.amount, adj_factor=excluded.adj_factor, src=excluded.src WHERE excluded.close IS NOT daily_bar.close OR excluded.adj_factor IS NOT daily_bar.adj_factor; """ COLUMN_MAP = { "日期": "trade_date", "开盘": "open", "最高": "high", "最低": "low", "收盘": "close", "成交量": "volume", "成交额": "amount", } def parse_html_table(html: str, symbol: str) -> pd.DataFrame: """从页面表格里抽出标准字段，缺失直接丢弃该行""" tables = pd.read_html(html) # 依赖 lxml raw = max(tables, key=len) # 取行数最多的那张表，通常是主表 df = raw.rename(columns=COLUMN_MAP) need = ["trade_date", "open", "high", "low", "close", "volume", "amount"] df = df[[c for c in need if c in df.columns]].copy() df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d") for c in ["open", "high", "low", "close", "volume", "amount"]: if c in df.columns: df[c] = pd.to_numeric(df[c].astype(str).str.replace(",", ""), errors="coerce") df["symbol"] = symbol df["adj_factor"] = None df["src"] = "html" return df.dropna(subset=["trade_date", "close"]) def save_bars(conn: sqlite3.Connection, df: pd.DataFrame) -> int: """幂等写入，返回受影响行数""" cols = ["symbol", "trade_date", "open", "high", "low", "close", "volume", "amount", "adj_factor", "src"] rows = df.reindex(columns=cols).itertuples(index=False, name=None) cur = conn.executemany(UPSERT, rows) conn.commit() return cur.rowcount

几个细节解释一下。max(tables, key=len)是为了应付页面里同时存在汇总表和明细表的情况，按行数挑最长的通常不会错。str.replace(",", "")处理成交量、成交额里的千分位分隔符，不处理的话pd.to_numeric会整列变成 NaN，而且不报错，属于典型的静默失败。UPSERT里的WHERE excluded.close IS NOT daily_bar.close是幂等性的关键：值没变的行不会触发实际写入，重跑一万次也不会让数据库膨胀。

adj_factor字段先留空是有意的。复权因子往往来自另一个接口或文件，采集阶段拿不到就留 NULL，后面统一在清洗层回填，不要让采集脚本承担太多职责。

2.4 增量更新与断点续爬：让交易日历驱动任务

全量跑一次之后，日常任务就变成“今天有哪些票缺数据”。最省事的做法不是遍历所有标的去判断，而是先维护一张交易日历表，再用它跟已有数据做差集。

def missing_dates(conn, symbol, start, end): """返回该标的在区间内缺失的交易日列表""" cal = [r[0] for r in conn.execute( "SELECT trade_date FROM trade_calendar " "WHERE trade_date BETWEEN ? AND ? AND is_open = 1 ORDER BY trade_date", (start, end))] have = {r[0] for r in conn.execute( "SELECT trade_date FROM daily_bar WHERE symbol = ? AND trade_date BETWEEN ? AND ?", (symbol, start, end))} return [d for d in cal if d not in have] def run_incremental(conn, symbols, start, end, fetch_one): """fetch_one(symbol, date) -> DataFrame，单日单票抓取函数""" for sym in symbols: todo = missing_dates(conn, sym, start, end) if not todo: continue for d in todo: try: df = fetch_one(sym, d) if df is not None and len(df): save_bars(conn, df) except Exception as e: # 单点失败只记录，不中断整批 conn.execute( "INSERT INTO fetch_fail(symbol, trade_date, err) VALUES (?, ?, ?)", (sym, d, repr(e)[:300])) conn.commit()

这套写法的核心思路是“把状态放在数据库里，而不是放在内存或文件里”。任务中断后重新运行，missing_dates会自动跳过已经入库的日期，天然支持断点续爬。失败记录单独落一张fetch_fail表，第二天可以按表里没成功的记录定向重试，不用整批重跑。

这里有个容易被忽略的点：交易日历必须包含is_open标记，而不能只存“周一到周五”。遇到节假日调休，用星期几判断会多算出好几天，导致爬虫反复请求根本没有数据的日期，返回空结果后你还以为是解析逻辑写错了。日历表本身也可以定期从公开渠道更新一次，属于低频维护的成本。

3. 数据清洗与特征计算：把原始 K 线变成能算的信号

3.1 复权口径：前复权、后复权与不复权怎么选

爬下来的价格是不复权的，遇到除权除息那天会出现一个巨大的跳空缺口，直接拿去算收益率会得到一个莫名其妙的 -30%，然后你以为发现了什么黑天鹅。复权这件事没有“正确答案”，只有“跟你的用途匹配”的答案，三种口径的差别必须心里有数。

我的一般习惯是：分析和回测统一用后复权，因为它不会因为数据更新（最新因子变化）而导致历史序列整体漂移——这算是回测里的一条“后悔药”。前复权留给画图，读者看到的图形跟行情软件一致，不会一脸疑惑。

import numpy as np def apply_adjust(df: pd.DataFrame, mode="hfq") -> pd.DataFrame: """按因子生成复权价，factor 缺失时用 1.0 兜底""" df = df.sort_values("trade_date").reset_index(drop=True).copy() factor = df["adj_factor"].fillna(1.0).astype(float) if mode == "hfq": ratio = factor elif mode == "qfq": ratio = factor / factor.iloc[-1] else: ratio = pd.Series(1.0, index=df.index) for c in ["open", "high", "low", "close"]: df[f"{c}_adj"] = df[c] * ratio # 成交量做反向调整，保证成交额不变 df["volume_adj"] = df["volume"] / ratio return df

factor.iloc[-1]取的是区间内最后一个交易日的因子，所以前复权价会随数据截取区间变化——这就是为什么同一只票在不同时间点导出的前复权价格会对不上，不是数据错了，是基准变了。成交量用除法调整是很多人漏掉的细节：价格乘以因子后，为了让价格 × 成交量保持守恒，成交量必须除以同一个因子，否则换手率、量比这类指标会整体失真。

3.2 停牌、缺失值与异常价格的处理

真实数据永远比示例脏。放进 pandas 之前要先把几类问题处理掉，否则它们会以 NaN 的形式渗透到指标里，最后表现为“某些日期指标是空的”，排查起来极其费时。

第一类是停牌。停牌当天通常没有成交记录，表里直接缺这一行。如果直接按行做rolling(20)，滚动窗口会跨越一个非交易日，算出来的 20 日均线其实覆盖了更长的时间跨度。规范做法是以交易日历为主表左连接行情数据，停牌日成交量填 0、价格前向填充，并在旁边加一列is_suspend标记，让下游逻辑自己决定要不要剔除。

第二类是异常价格。偶尔会抓到close = 0或者close = 999999这种明显错误的值，多半是页面改版或者字段错位。用一条简单的规则兜住：当日最高价低于最低价、收盘价不在[low, high]区间内、单日涨跌幅绝对值超过 30%（非新股），一律标记为可疑行并落日志，不直接删除，方便回头核实。

def clean_bars(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df["close"] = df["close"].replace(0, np.nan) df["high"] = df[["high", "open", "close"]].max(axis=1) df["low"] = df[["low", "open", "close"]].min(axis=1) df["ret_raw"] = df["close"].pct_change() df["suspect"] = ( (df["high"] < df["low"]) | (df["close"] > df["high"] * 1.001) | (df["ret_raw"].abs() > 0.3) ) df.loc[df["suspect"], ["open", "high", "low", "close"]] = np.nan df[["open", "high", "low", "close"]] = df[["open", "high", "low", "close"]].ffill() return df

把可疑值置为 NaN 再ffill，比直接删行安全：删行会让时间索引错位，而ffill至少保持了日期的连续性，同时suspect列留了追踪线索。注意high和low的修正放在异常检测之前，否则close越界的情况会被误判成整行可疑。

3.3 用 pandas 算均线、波动率与换手率

数据干净了，指标计算本身很短，难点在于口径。均线用简单移动平均还是指数移动平均，波动率用日频标准差年化还是用真实波幅，换手率的分母用流通股本还是总股本——这些选择会直接改变信号频率，必须在写代码前定好并写进注释，不然三个月后自己都看不懂。

def add_features(df: pd.DataFrame, window_fast=5, window_slow=20, vol_window=20) -> pd.DataFrame: df = df.copy() c = df["close_adj"] df["ma_fast"] = c.rolling(window_fast, min_periods=window_fast).mean() df["ma_slow"] = c.rolling(window_slow, min_periods=window_slow).mean() df["ret"] = c.pct_change() # 年化波动率：日收益标准差 * sqrt(252) df["vol"] = df["ret"].rolling(vol_window, min_periods=vol_window).std() * np.sqrt(252) # 真实波幅的均值，衡量日内波动 prev_close = c.shift(1) tr = pd.concat([ df["high_adj"] - df["low_adj"], (df["high_adj"] - prev_close).abs(), (df["low_adj"] - prev_close).abs(), ], axis=1).max(axis=1) df["atr"] = tr.rolling(vol_window, min_periods=vol_window).mean() if "float_share" in df.columns: df["turnover"] = df["volume_adj"] / df["float_share"] return df

min_periods这个参数容易被忽略。默认情况下rolling(20)在凑满 20 个点之前返回 NaN，行为是对的；但如果你的数据有缺口又被ffill填过，实际参与计算的可能包含填充值，出来的均线会偏离。所以我在清洗阶段把is_suspend标出来，计算指标时可以选择性把这些点设为 NaN 让窗口自然延后，宁可少几个信号，也不要一个假的均线值。

年化因子np.sqrt(252)是日频的常用约定，用它算出来的波动率可以直接跟同类标的横向比较。如果你的数据是周频，得换成sqrt(52)，这种地方写错不会报错，只会让所有标的的波动率整体偏大或偏小一个常数倍，属于典型的静默错误。

3.4 因子面板的组织：长表还是宽表

单只票分析用长表（每行一个日期）最顺手，但一旦要做横截面排序、多标的对比，就必须转成宽表（索引是日期，列是标的）。这两种形态在 pandas 里各有用武之地，来回转换的成本也很低，关键是别混着用。

# 长表 -> 宽表：每个字段一张面板 panel_close = df.pivot_table(index="trade_date", columns="symbol", values="close_adj") panel_ret = panel_close.pct_change() # 横截面标准化，用于多因子比较 def cs_zscore(panel: pd.DataFrame) -> pd.DataFrame: return (panel - panel.mean(axis=1, skipna=True).to_frame() ).div(panel.std(axis=1, skipna=True).to_frame()) z = cs_zscore(panel_ret.rolling(20).mean())

pivot_table而不是pivot，是因为真实数据里同一(trade_date, symbol)可能重复（尤其在多源合并之后），pivot遇到重复会直接抛异常，而pivot_table默认取均值，至少能跑通并让你有机会事后查重。横截面标准化一定要指定axis=1，按行（即按日期）计算均值和标准差，如果写成axis=0就变成时间序列标准化，含义完全不同，而且不会报错。

4. 避坑与常见问题排查：五个真实翻车场景

4.1 数据层的三个坑

现象一：抓下来的价格跟行情软件对不上，差了整整一个涨停。原因几乎总是复权口径不一致。行情软件默认展示前复权，而你库里存的是不复权原始价，除权日之后两者会差出一大截。解决方式是统一口径，分析层一律用后复权价计算收益率，展示层再转成前复权，并且在字段名上就写清楚close_adj，别留一个模糊的close让人猜。

现象二：重跑一次脚本，数据库里的行数翻倍。原因是建表时没设主键，或者用了INSERT而不是INSERT ... ON CONFLICT。修法很直接，加上PRIMARY KEY (symbol, trade_date)，写入语句换成幂等的 UPSERT，重跑多少次都不会膨胀。已经脏了的库可以用一句DELETE FROM daily_bar WHERE rowid NOT IN (SELECT MIN(rowid) FROM daily_bar GROUP BY symbol, trade_date)清掉重复行再补主键。

现象三：接口返回 200，但解析出来全是空的。状态码正常不代表内容正常，很多页面在触发频率限制时会返回一个空列表或者一段提示文案，resp.json()依然成功。解决方式是在解析函数里加一道断言，比如data里必须有trade_date和close两个 key，缺了就直接抛异常并记录原始响应片段。这一步加上之后，排错时间能砍掉一大半。

4.2 计算和回测层的两个坑

现象四：回测年化收益 300%，夏普 8，兴奋了十分钟。十有八九是未来函数。最常见的就是用当日收盘价生成的信号在当日成交——你的策略在开盘那一刻就知道了收盘价，这是作弊。修正方法是把仓位信号整体shift(1)，用昨天的信号决定今天的持仓，代价是收益会大幅缩水，但那才是真实的。

df["signal"] = (df["ma_fast"] > df["ma_slow"]).astype(int) df["pos"] = df["signal"].shift(1).fillna(0) # 信号延后一天执行 df["strat_ret"] = df["pos"] * df["ret"]

另一个隐蔽的未来函数是rolling之后忘了shift，或者用了以未来数据计算的分位数做阈值。检查办法很简单：把回测区间从中间截断，只用前半段数据重新跑，如果前半段的信号和完整数据跑出来的不一致，说明有地方偷看了后面。

现象五：指标在某个日期突然变成 NaN，后面全是空的。这是ffill失效的典型表现——如果某一天的值是 NaN，而前面恰好也是 NaN，ffill填不出东西。根源往往是清洗阶段把连续多天的异常值都置空了，或者pct_change遇到价格为 0 产生了 Inf。解决方式是清洗后加一步完整性检查，统计每个标的的 NaN 比例，超过阈值（比如 5%）就整只票标记为数据不足，不要带着窟窿往下算。

5. 回测与可视化：验证你的分析到底有没有信息量

5.1 向量化回测的边界与写法

工程上没必要一上来就搭事件驱动引擎。日频、单标的、只做多这一类的策略，向量化写法几十行就能跑，迭代速度快得多。真正需要事件驱动的场景是日内撮合、多腿对冲、复杂订单类型，那些才值得上框架。

def backtest_ma(df: pd.DataFrame, fast=5, slow=20, cost=0.0005) -> pd.DataFrame: """双均线向量化回测，cost 为单边费率""" d = df.copy() d["ma_fast"] = d["close_adj"].rolling(fast).mean() d["ma_slow"] = d["close_adj"].rolling(slow).mean() d["signal"] = (d["ma_fast"] > d["ma_slow"]).astype(int) d["pos"] = d["signal"].shift(1).fillna(0) # 次日生效 d["ret"] = d["close_adj"].pct_change().fillna(0) d["turn"] = d["pos"].diff().abs().fillna(0) # 仓位变化量 d["cost"] = d["turn"] * cost # 双边实际各收一次 d["strat_ret"] = d["pos"] * d["ret"] - d["cost"] d["nav"] = (1 + d["strat_ret"]).cumprod() d["bench"] = (1 + d["ret"]).cumprod() return d

cost按单边费率算、乘上仓位变化量，是因为一次买卖各收一次，diff().abs()在开仓和平仓时各产生 1，正好对应两笔。如果直接用pos * cost只在持仓时扣费，就漏掉了空仓转持仓那一下的成本，长期下来会高估收益。这个参数别嫌小，日频策略一年换手几十次，万五的费用累计起来能吃掉相当一部分超额收益。

5.2 绩效指标：别只看总收益

一条净值曲线能算出的东西远不止收益率。最大回撤告诉你最难受的时候有多难受，夏普告诉你单位风险换来多少收益，卡玛比率把年化收益和最大回撤放在一起看，更适合评估“能不能拿得住”。

def perf(nav: pd.Series, periods=252) -> dict: ret = nav.pct_change().dropna() total = nav.iloc[-1] / nav.iloc[0] - 1 years = max(len(ret) / periods, 1e-9) cagr = (1 + total) ** (1 / years) - 1 mdd = (nav / nav.cummax() - 1).min() sharpe = ret.mean() / ret.std() * np.sqrt(periods) if ret.std() > 0 else float("nan") calmar = cagr / abs(mdd) if mdd < 0 else float("nan") win = (ret > 0).mean() return {"total": round(total, 4), "cagr": round(cagr, 4), "mdd": round(mdd, 4), "sharpe": round(sharpe, 3), "calmar": round(calmar, 3), "win_rate": round(win, 4)}

years用max(..., 1e-9)兜底，是为了防止回测区间太短导致除零。sharpe里乘sqrt(periods)是把日频夏普年化，跟业界口径一致；如果你的净值是周频，periods要改成 52，否则夏普会偏小一个常数倍，横向比较时得出错误结论。win_rate看着直观，但对盈亏比大的策略有误导性——胜率 40% 但盈亏比 3:1 的策略远比胜率 60% 盈亏比 0.5:1 的好，别单独拿胜率下判断。

5.3 可视化：一眼看出问题的图比十张报表有用

图的作用不是好看，是快速暴露异常。三张图基本够用：净值与基准的对比曲线、回撤填充图、以及信号与价格叠加的散点。matplotlib 写十来行就能出，不必上复杂的可视化库。

import matplotlib.pyplot as plt def plot_result(d: pd.DataFrame, title="strategy vs benchmark"): fig, axes = plt.subplots(2, 1, figsize=(11, 7), sharex=True, gridspec_kw={"height_ratios": [3, 1]}) axes[0].plot(d["trade_date"], d["nav"], label="strategy") axes[0].plot(d["trade_date"], d["bench"], label="benchmark", alpha=0.7) axes[0].set_ylabel("nav") axes[0].legend(loc="upper left") axes[0].set_title(title) dd = d["nav"] / d["nav"].cummax() - 1 axes[1].fill_between(d["trade_date"], dd, 0, color="tab:red", alpha=0.35) axes[1].set_ylabel("drawdown") plt.tight_layout() plt.savefig("backtest.png", dpi=130)

回撤子图用fill_between而不是折线，是为了让“水下时间”一眼可见——一段长达半年的水下期，在填充图里是一大片红色，在折线图里可能只是一条不起眼的低平线。sharex=True保证上下两图的时间轴对齐，否则你会发现两边的信号总是差那么几天，白排查半天。

6. 进阶收尾：把日频数据做成一个可复用的本地数据仓库

跑到这一步，脚本已经能出结果了，但离“能用”还差一层：目录结构。我见过的翻车大多不是算法写错，而是三个月后想复现某个结论，发现原始 CSV 被覆盖了、复权因子换过一版、脚本里硬编码了一个日期区间，三者对不上，只能重头再来。

比较稳的做法是按data/raw/{symbol}/{year}.parquet分区存储原始数据，data/curated/放清洗和复权后的宽表，data/factor/放计算好的特征面板，每一层都带一个_meta.json记录生成时间、行数、字段版本和源数据指纹。这样任何一个中间结论都能追溯到具体哪一版数据，出问题也能定位到是哪一层污染了。

import json, hashlib, os import pandas as pd def save_layer(df: pd.DataFrame, path: str, note: str = "") -> None: os.makedirs(os.path.dirname(path), exist_ok=True) df.to_parquet(path, index=False) meta = { "path": path, "rows": int(len(df)), "cols": list(df.columns), "sha1": hashlib.sha1(pd.util.hash_pandas_object(df).values.tobytes()).hexdigest()[:16], "note": note, "built_at": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"), } with open(path.replace(".parquet", "_meta.json"), "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2)

hash_pandas_object算出来的指纹是整表级别的，只要有一行变了指纹就变，用来做“这份数据是不是上次那份”的判断非常省事。日常更新完数据后跑一遍指纹比对，如果发现跟昨天算出来的完全一样，说明增量任务其实什么都没抓到，这时候就该去查接口是不是挂了，而不是想当然地以为“今天没行情”。

还有一个习惯值得强制养成：任何一次回测输出，都必须同时保存三样东西——策略参数、数据指纹、绩效字典。我在早期吃过一次亏，某次跑出来一个很漂亮的曲线，一周后想复现，参数记不清了，数据也被增量更新覆盖了，只能从头再调一遍，白白浪费一个下午。从那以后我每次跑回测，脚本入口第一件事就是把params + data_sha1 + perf写进runs/下的一个 json，跑完顺手看一眼文件有没有生成。看着麻烦，但它救过我很多次。希望帮到你。

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