Python股票大数据分析：行情管道、因子计算与回测避坑

简介：围绕股票大数据分析与Python爬虫实战整理的资料包，面向希望进入金融数据分析、量化研究方向的初学者与进阶学习者，重点解决股票数据采集、清洗、分析与预测流程衔接的问题。内容涉及用BeautifulSoup、Scrapy抓取历史行情与公告资讯，借Pandas处理缺失值、异常值和格式统一，结合NumPy、SciPy完成统计计算，并用Matplotlib、Seaborn绘制K线、成交量与波动性图表；同时延伸到Scikit-learn的线性回归、决策树、随机森林及神经网络预测，以及Tushare、Wind实时数据和Flask、Django结果展示。压缩包为7z格式，约4.47MB，文件总数显示为0且类型明细暂无数据，具体清单以实际下载为准。已有3659人学习，适合按“爬取—预处理—可视化—建模—部署”链路系统实践，也可作为金融数据分析项目查漏补缺的参考。

1. 从一份行情 CSV 到能跑起来的选股流程：这套股票大数据分析实战解决的是哪类麻烦

很多人学 python 爬虫，第一个练手对象就是行情数据：抓下来一堆日线，存成 CSV，画两条均线，然后就没有然后了。真正卡住人的不是 requests 怎么发，而是从「抓得到」到「算得对」中间那一大段没人细讲的东西——字段怎么定、增量怎么去重、复权怎么处理、回测为什么年化高得离谱。这套基于股票大数据分析的 Python 实战资源，拆的正是这一段：抓取、落库、清洗、因子、回测、校验，一条链走完。适合已经会写基础爬虫、但一碰到多标的时序数据就手忙脚乱的人；也适合手里有策略想法、想先搭个能自证的离线环境再谈实盘的从业者。只要你愿意按顺序把每一步跑通，它省下的是反复试错的时间。

2. 数据管道怎么搭：抓取、落库与增量更新的三段式

行情数据管道最容易写成一次性脚本：跑通了，第二天再跑就把昨天的数据覆盖掉，第三天发现缺了几只票，第四天开始手动补。这类资源包里通常会把管道拆成三层——请求层负责拿数据、落库层负责幂等写入、调度层负责决定今天该抓哪些。三层分开写的好处是，任何一层出问题都能单独定位，而不是在一坨两千行的脚本里靠 print 猜。

拆之前先明确一件事：行情数据不是「一张表」，而是「一张主表 + 若干附属表」。主表是日线量价，附属表是复权因子、交易日历、标的清单。主表用(code, trade_date)做联合主键，这个决定一旦定下来，后面所有的增量更新、去重、补数都变得机械而可靠。我见过太多项目在字段设计阶段偷懒，用自增 ID 当主键，结果每次重跑都插入重复行，最后统计收益时同一只票被算了三遍。

2.1 先定字段再写爬虫：一张日线表的字段设计

新手写爬虫的顺序往往是先写请求、拿到什么存什么，字段名跟着接口返回走。接口一改版，整条链路崩。正确顺序是反过来的：先把日线表的字段定死，再让请求层去做字段映射。字段选哪些，取决于你后面要算什么——均线和动量只需要收盘价，波动率需要最高最低，换手率和流动性过滤需要成交额和流通股本，复权需要复权因子。

一份够用的日线表至少包含这些列：

建表语句我一般直接写在初始化脚本里，跟抓取逻辑分开：

-- schema.sql：日线主表，主键保证幂等 CREATE TABLE IF NOT EXISTS daily_bar ( code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, amount REAL, adj_factor REAL DEFAULT 1.0, is_tradable INTEGER DEFAULT 1, PRIMARY KEY (code, trade_date) ); -- 常用查询都走这个索引，按标的取时间序列 CREATE INDEX IF NOT EXISTS idx_bar_code_date ON daily_bar (code, trade_date);

字段设计里有三个点值得多说两句。第一，价格存未复权值、单独存复权因子，而不是直接存前复权价。前复权价会随着每次除权除息而整体变化，今天存进去的前复权序列，明天再看就不是同一组数字了，任何跨期比较都会失真。第二，is_tradable这个字段一定要留，它记录停牌、一字板这类「价格存在但成交不了」的状态，回测阶段全靠它过滤，事后再补就得重新拉一遍历史。第三，日期统一用YYYY-MM-DD字符串而不是时间戳，好处是排序即时间序、肉眼可读、跨库迁移不踩时区坑，代价是查询时别指望用日期函数做区间运算，得靠字符串比较——而这个恰好是字典序正确的前提。

2.2 请求层：限速、重试与显式失败

请求层最容易出的问题不是抓不到，而是「抓不到却不报错」。接口限流时返回一个空 JSON、网络抖动时返回半截数据，脚本默默把空结果写进库，等到回测阶段发现某只票中间断了两个月，你已经无从判断是源头缺数据还是自己漏抓。我的做法是：任何一次请求，只要解析出来的行数为零或者关键字段缺失，直接抛异常中断，让调度层去决定重试还是跳过。

限速是另一件事。单机串行抓几百只票的历史日线，如果每个请求间隔几十毫秒，很容易触发对端频控；间隔放到一秒，又慢得让人想放弃。折中方案是固定间隔加随机抖动，配合连接复用，把单只票的抓取压到一次请求拿完一个区间。

import random import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(total=3, backoff=0.5): """构造带重试的会话：连接复用 + 指数退避""" session = requests.Session() retry = Retry( total=total, # 最多重试 3 次 backoff_factor=backoff, # 退避基数，等待 0.5/1/2 秒 status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"], ) adapter = HTTPAdapter(max_retries=retry, pool_connections=8, pool_maxsize=8) session.mount("https://", adapter) session.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; research-bot)", "Accept": "application/json", }) return session def fetch_daily(session, code, start, end, min_interval=0.8): """抓单只标的一段日线，空结果直接抛错""" url = "https://example.com/quote/daily" # 替换成你实际使用的行情源 params = {"code": code, "start": start, "end": end} # 固定间隔 + 随机抖动，避免整点齐射 time.sleep(min_interval + random.uniform(0, 0.4)) resp = session.get(url, params=params, timeout=8) resp.raise_for_status() payload = resp.json() rows = payload.get("data") or [] if not rows: raise ValueError(f"empty payload: {code} {start}~{end}") return rows

Retry的total控制总重试次数，backoff_factor是指数退避的基数，实际等待时间约为backoff_factor × 2^(n-1)，第一次重试 0.5 秒、第二次 1 秒、第三次 2 秒。status_forcelist里放的是值得重试的状态码，429 是限流、5xx 是服务端抖动，而 404 这类不该重试的不要放进去，否则会白等三轮。pool_connections和pool_maxsize设成同样的值，是为了让连接池大小和并发度匹配，太小会频繁建连，太大对单机脚本没意义。

min_interval是我习惯留的手动旋钮。抓历史数据时可以压到 0.3 秒，因为总量有限、跑一次就完；日更时可以放到 1.5 秒，因为每天只抓增量、稳定性优先。这个值没有标准答案，取决于对端频控的松紧，改之前先小批量试二十只票，看有没有出现 429。

2.3 落库与增量：用主键去重而不是每天重抓全量

全量重抓是最省事也最浪费的做法。几百只票十年的日线，每次跑都重新拉一遍，几十分钟下去，网络稍抖一下还得从头再来。正确姿势是先问库「你已经有到哪天了」，再只抓缺失的那一段。

import sqlite3 import pandas as pd UPSERT_SQL = """ INSERT INTO daily_bar (code, trade_date, open, high, low, close, volume, amount, adj_factor, is_tradable) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(code, 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, is_tradable=excluded.is_tradable; """ def last_date(conn, code): """查某只票已入库的最后一天，没有则返回 None""" row = conn.execute( "SELECT MAX(trade_date) FROM daily_bar WHERE code = ?", (code,) ).fetchone() return row[0] if row and row[0] else None def upsert_bars(conn, df: pd.DataFrame): """按主键幂等写入，重复区间自动覆盖""" cols = ["code", "trade_date", "open", "high", "low", "close", "volume", "amount", "adj_factor", "is_tradable"] conn.executemany(UPSERT_SQL, df[cols].itertuples(index=False, name=None)) conn.commit()

ON CONFLICT ... DO UPDATE是这套写法的核心：同一只票同一天重复抓，不会插出两行，而是把旧值覆盖成新值，正好用来修复历史上抓错的字段。last_date返回的日期直接作为下一次请求的起点（可以往前多退几天做重叠校验），第一次抓则用配置里的起始日期。

增量更新还有个隐藏细节：交易日历。如果某只票昨天停牌，last_date停在停牌前一天，第二天增量抓取会从那天开始，把停牌期间的空洞当成「还没抓」而反复请求。解决办法是维护一张交易日历表，用「日历上最后一个交易日」而不是「该票最后一条记录」来推进水位线，同时对返回的空区间做标记，避免下次重复请求。

3. 清洗与因子计算：把原始行情变成能算的宽表

数据进库只是原料，能直接喂给策略的是清洗后的宽表：行是交易日，列是标的，值是价格或因子。这一步的返工率最高，因为问题往往不是报错，而是「算出来的数看着对、其实是错的」。复权用错方式，均线会穿越；缺失值一路前填，停牌股被当成价格不动；rolling不设min_periods，前二十天的均线是用不足窗口的数据算出来的。这三类问题都不会抛异常，只会在回测结果里悄悄放大收益。

3.1 复权、停牌与缺失值：处理顺序不能反

正确顺序是：先复权，再标记可交易状态，最后才谈填充。很多人反过来，先ffill把缺失价填上，再做复权乘法，结果复权因子把填充出来的假价格也一起放大了。

复权本身有两种口径。后复权是「以最早价格为基准，把之后的价格按因子往上调」，历史值一旦生成就不再变化，适合做回测和因子计算；前复权是「以最新价格为基准往回调」，看着直观，但每次除权都要全量重算，历史序列会变。做研究我一律用后复权，落库时只存因子，用的时候现乘：

import pandas as pd def to_hfq(df: pd.DataFrame) -> pd.DataFrame: """原始价 × 复权因子 = 后复权价，按标的分组、按日期排序""" df = df.sort_values(["code", "trade_date"]).copy() for col in ["open", "high", "low", "close"]: df[f"{col}_hfq"] = df[col] * df["adj_factor"] # 成交额不随复权变化，但成交量要按因子反向调整才能和价格口径一致 df["volume_adj"] = df["volume"] / df["adj_factor"] return df def mark_tradable(df: pd.DataFrame) -> pd.DataFrame: """停牌（成交量为 0）或价格缺失，标记为不可交易""" df = df.copy() df["is_tradable"] = ( df["volume"].fillna(0).gt(0) & df["close_hfq"].notna() ).astype(int) return df

volume / adj_factor这一步常被忽略。送转之后股本变多、股价变低，如果不调整成交量口径，用「成交额 / 成交量」倒推出来的均价在除权日附近会出现明显跳变，任何基于量价的因子都会被打断。

缺失值处理要分场景。价格缺失超过三个交易日，基本可以判断是长期停牌，这种直接从样本里剔除，不要填；缺失一两天且前后价格连续，用前值填充可以接受，但要记一个filled标记，回测时把这段的信号置空。我一般设两个阈值：连续缺失 ≤ 2 天用前填，> 2 天置为 NaN 并在调仓日剔除。这样既保住了短停牌造成的序列断裂，也不会让一只停牌半年的票在因子表里显示「价格纹丝不动、波动率为零」，从而被低波动策略优先选中。

3.2 用滚动窗口算均线、波动率和动量因子

因子计算的主体是分组滚动。groupby(...).transform(...)和groupby(...).apply(...)的区别在于前者返回与原表等长的结果、索引对齐，后者可能改变形状，容易在后续 merge 时错位。做时序因子我基本只用transform。

import numpy as np import pandas as pd def add_factors(df: pd.DataFrame, windows=(5, 20, 60)) -> pd.DataFrame: """按标的分组计算均线、动量、波动率，窗口不足的位置留 NaN""" df = df.sort_values(["code", "trade_date"]).copy() grp = df.groupby("code", group_keys=False) for w in windows: # 均线：不足 w 个样本时不出值，避免用短窗口冒充长窗口 df[f"ma{w}"] = grp["close_hfq"].transform( lambda s, w=w: s.rolling(w, min_periods=w).mean() ) # 动量：w 日收益率 df[f"mom{w}"] = grp["close_hfq"].transform( lambda s, w=w: s.pct_change(w) ) # 波动率：日收益的 w 日滚动标准差，年化 df[f"vol{w}"] = grp["close_hfq"].transform( lambda s, w=w: s.pct_change().rolling(w, min_periods=w).std() * np.sqrt(252) ) return df

三个参数值得逐个说。min_periods=w是关键，不设它rolling会从第一个样本就开始出值，前 19 天的 20 日均线其实只用了 1 到 19 个点，这类数值在横截面上排名会异常靠前，直接把选股结果带偏。pct_change(w)里的w是滞后阶数，算的是close[t] / close[t-w] - 1，注意它内部同样受分组影响，transform会保证每组独立计算，不会跨标的串行。年化系数sqrt(252)假设一年 252 个交易日，如果你的数据里含港股或美股，这个常数要换成对应市场的交易日数，混用会让不同市场的波动率不可比。

另外，lambda s, w=w里的默认参数绑定不能省。循环里直接用闭包引用w，所有 lambda 会共享同一个变量，最后全部按最后一次循环的值计算——这是新手写因子脚本最常见的静默错误之一，结果就是所有窗口算出来完全一样，而你还在怀疑数据。

3.3 长表转宽表：pivot 与内存的关系

因子算完还是长表（一行一个标的-日期），但很多操作天然适合宽表：横截面排名、行业中性化、组合权重分配。转换用pivot_table而不是pivot，因为前者对重复键有聚合行为，不会直接抛错。

# 长转宽：行是日期，列是标的 wide = df.pivot_table( index="trade_date", columns="code", values="mom20", aggfunc="last" ) # 宽转长：melt 回长表，方便和原始表 merge long = wide.reset_index().melt( id_vars="trade_date", var_name="code", value_name="mom20" )

aggfunc="last"的意思是同一标的同一天如果有多行（比如数据源重复推送），取最后一条。这一步在数据干净时是冗余的，在不干净时能救命。宽表的代价是内存：5000 只票 × 2500 个交易日的浮点矩阵大约 100 MB，再做几个因子叠加到 GB 级很正常，所以我的习惯是宽表只保留当前策略真正用到的少数几列，算完立刻melt回去或者落盘成 parquet，不要长期驻留。

4. 策略回测框架：从一个均线交叉例子看穿未来函数

回测最容易骗自己的地方，是让「今天收盘才知道的信息」参与了「今天开盘的决策」。这类错误在代码里往往只差一个shift，在结果上却能把年化从 8% 抬到 40%。向量化回测的最小骨架其实很短：算信号、延迟一天、乘收益、扣成本、算净值。短归短，每一行都有讲究。

4.1 向量化回测的最小骨架

import numpy as np import pandas as pd def backtest_ma(df: pd.DataFrame, fast=5, slow=20, fee=0.0003, slippage=0.0005) -> pd.DataFrame: """双均线向量化回测：t 日信号，t+1 日成交""" df = df.sort_values(["code", "trade_date"]).copy() grp = df.groupby("code", group_keys=False) df["fast"] = grp["close_hfq"].transform( lambda s: s.rolling(fast, min_periods=fast).mean()) df["slow"] = grp["close_hfq"].transform( lambda s: s.rolling(slow, min_periods=slow).mean()) # t 日收盘后才能确定的信号 df["signal"] = (df["fast"] > df["slow"]).astype(int) # 关键一行：仓位延迟一天，杜绝未来函数 df["pos"] = grp["signal"].shift(1).fillna(0) # t 日收益 = t 日收盘 / t-1 日收盘 - 1 df["ret"] = grp["close_hfq"].transform(lambda s: s.pct_change()) # 换手发生在仓位变化的当天，成本 = |Δpos| × (费率 + 滑点) df["turnover"] = (df["pos"] - grp["pos"].shift(1).fillna(0)).abs() df["cost"] = df["turnover"] * (fee + slippage) df["strategy_ret"] = df["pos"] * df["ret"] - df["cost"] return df

shift(1)是整段代码里唯一不能省的一行。没有它，pos和ret用的是同一天的收盘价，等效于「收盘算信号、收盘成交」，这就是最典型的未来函数，回测曲线会漂亮得不像话。

fee和slippage分开传，是因为它们性质不同：费率是显性成本，按成交额比例收取，A 股还有印花税只在卖出时收，实盘要单独处理；滑点是隐性成本，取决于你的单子相对盘口的冲击，小资金可以设 5 个基点，资金量大就得按日均成交额的占比动态估。

turnover用|Δpos|而不是pos，是因为只有仓位变化的那一刻才产生交易成本。如果写成pos * cost，会变成持仓期间每天都在扣费，长期持有的策略会被过度惩罚。这个错误藏得深，因为它只影响幅度、不影响方向，很容易被当成「成本设高了」。

4.2 手续费、滑点与成交价：把乐观值压下去

向量化回测的默认假设是「按收盘价成交、想买多少买多少」。这两个假设在实盘里都不成立。收盘价成交意味着你在 14:57 集合竞价结束时才下决定，实际上信号是用收盘价算的，下单只能等到次日开盘；想买多少买多少意味着忽略涨跌停和流动性约束。

常见的修正有三条。第一，成交价改成次日开盘价，代码上就是让ret用open_hfq的隔夜加日内收益：

# 用次日开盘买入、持有到再下一次开盘卖出 df["ret_next_open"] = grp["open_hfq"].transform(lambda s: s.pct_change().shift(-1))

注意这里的shift(-1)是往前取一行，配合已经shift(1)的仓位，整体仍然是「t 日信号 → t+1 日开盘成交」。两个 shift 方向相反，很容易把人绕晕，建议在纸上画三行日期对齐一遍再写。

第二，涨停买不进、跌停卖不出。这就要用到前面留的is_tradable字段：如果开仓日is_tradable == 0，把当天仓位强制设为前一日仓位，并把信号顺延。第三，流动性约束，单只票单日买入金额不超过其近 20 日日均成交额的 1%，超出部分按比例缩减仓位。这三条加进去，回测曲线的年化通常会掉三分之一到一半，掉得越多说明策略越依赖不现实的成交假设。

成本参数没有统一标准，我给一组常用的起步值：佣金双边万三、印花税卖出千一、滑点双边各五到十个基点。小资金做中低频，这套参数下的结果和实盘的偏差通常在可接受范围内；高频或者大资金，必须按自己的实际费率和冲击模型重设。

4.3 绩效指标怎么算才不糊弄自己

净值曲线好看不代表策略成立。至少要同时看五个数：年化收益、最大回撤、夏普、胜率、换手率。这五个数互相牵制，单看任何一个都能被优化出花来。

def performance(ret: pd.Series, periods=252) -> dict: """输入日频策略收益序列，输出常用绩效指标""" ret = ret.fillna(0.0) nav = (1 + ret).cumprod() total_return = nav.iloc[-1] - 1 years = len(nav) / periods cagr = (1 + total_return) ** (1 / years) - 1 if years > 0 else np.nan max_dd = (nav / nav.cummax() - 1).min() sharpe = ret.mean() / ret.std(ddof=0) * np.sqrt(periods) if ret.std() else np.nan return { "年化收益": round(cagr, 4), "最大回撤": round(max_dd, 4), "夏普": round(sharpe, 3), "卡玛": round(cagr / abs(max_dd), 3) if max_dd else np.nan, "累计净值": round(nav.iloc[-1], 4), }

ddof=0用总体标准差而不是样本标准差，日频几千个样本下差别很小，但保持口径一致比取哪个更重要。年化用几何平均而不是ret.mean() * 252，后者在波动大的策略上会明显高估。卡玛比率等于年化收益除以最大回撤的绝对值，比夏普更能反映「亏起来疼不疼」，中低频策略我一般要求卡玛不低于 1。

换手率单独提一句：它的计算要基于turnover的累加，日频策略年换手 20 倍和 200 倍是两种完全不同的生意。换手高到一定程度，费率参数的微小变化就能让策略从盈转亏。看回测结果时，先看换手，再看收益，顺序反了容易被漂亮曲线骗。

5. 避坑与排查：这套流程里最容易翻车的五个地方

跑通一遍和跑对一遍是两回事。下面五条是我在这类项目里反复遇到的，每条都按现象、原因、解决写，出现频率从高到低排。

5.1 回测年化高得离谱：先查未来函数

现象：双均线策略在几百只票上跑出年化 60% 以上，夏普超过 3，净值曲线几乎不回头。

原因：仓位和收益用了同一天的价格。要么signal之后忘了shift(1)，要么用收盘价算信号又用收盘价成交，要么在因子计算里用了shift(-1)取未来数据做标准化。还有一种隐蔽的：用全样本的均值方差做因子标准化，等于把未来的分布信息泄露到了历史每一天。

解决：在所有涉及「信号 → 收益」的地方强制加一天延迟，因子标准化一律改成滚动窗口（如过去 252 个交易日的均值方差）或横截面当期标准化。改完后先跑一次随机信号做对照，随机信号的年化应该在 0 附近小幅波动，如果随机信号都能跑出 15% 年化，说明框架里还有泄漏。

5.2 回测里成交了，实盘却买不进：停牌和涨跌停没过滤

现象：回测中某些票在停牌期间仓位不变、收益为零，看起来「稳稳扛过下跌」，实盘根本卖不掉。

原因：只用价格序列做回测，没有判断当日是否可交易。停牌股价格是前值，涨跌停股价格存在但成交量为极小值。

解决：落库阶段就写is_tradable，回测撮合时如果is_tradable == 0，仓位强制等于前一日，信号顺延到下一个可交易日。涨跌停的判定可以简化成「收盘价等于涨停价」或「当日振幅小于 0.5% 且成交量低于近 20 日均量的 10%」，不必追求精确，能挡住大部分虚假成交就够了。

5.3 单只票收益贡献 80%：横截面样本被污染

现象：组合整体收益不错，但拆开看，绝大部分利润来自一只票的某一段。

原因：缺失值填充过度。长期停牌的票被ffill一路填成价格不变、波动率为零，在低波动或低估值因子上排名靠前，被反复选中。

解决：给填充设上限，连续缺失超过 2 到 3 个交易日就置为 NaN 并在调仓日剔除；同时在绩效归因里强制输出「单标的收益贡献占比」，超过 50% 就回去查数据而不是庆祝。

5.4 换了个时间区间，结论完全反过来：复权口径不一致

现象：同一套参数，在 2018 到 2022 的数据上夏普 1.2，换成 2020 到 2024 变成 0.3，而重叠年份的走势本该一致。

原因：混用了前复权和后复权。前复权序列每次除权都会整体变化，不同时间抓下来的数据拼在一起，除权日附近出现价格跳变，均线被硬生生掰弯。

解决：落库只存原始价和复权因子，所有计算统一用后复权价，因子一次落库不再改。历史数据的复权因子要定期回补，因为除权后因子会更新，增量更新时把最近 60 个交易日的因子重刷一遍。

5.5 脚本跑完显示成功，库里却是空的：静默失败

现象：调度日志一片绿色，第二天打开库发现昨天一条数据都没进来。

原因：接口限流返回空结果，代码用payload.get("data", [])拿到空列表后直接for循环零次，写入零行，退出码 0。或者异常被一个宽泛的except Exception: pass吃掉了。

解决：请求函数里对空结果显式抛异常；写入前校验行数和日期区间，行数为零直接让任务失败；调度层检查「本次写入行数 / 预期标的数」这个比值，低于阈值就告警。日志里必须打印每次抓取的标的数、起始日期、返回行数这三个数，出问题时能第一时间定位是漏抓还是源头缺数据。

6. 进阶：把脚本变成每日可校验的定时任务

单机脚本跑通之后，下一个问题是「怎么知道它今天跑得对不对」。我的做法是给管道加一层轻量校验，用一张检查表把每次运行的几个关键数字落下来，长期看趋势，短期看异常。

def daily_check(conn, trade_date: str) -> dict: """每日数据质量快照，写入 check_log 后可用于趋势告警""" q = "SELECT COUNT(*), SUM(is_tradable) FROM daily_bar WHERE trade_date = ?" total, tradable = conn.execute(q, (trade_date,)).fetchone() # 与前一日对比，标的数突变往往意味着抓取中断 prev = conn.execute( "SELECT COUNT(*) FROM daily_bar WHERE trade_date < ? " "ORDER BY trade_date DESC LIMIT 1", (trade_date,) ).fetchone()[0] return { "trade_date": trade_date, "rows": total or 0, "tradable": tradable or 0, "tradable_ratio": round((tradable or 0) / total, 4) if total else 0, "delta_vs_prev": (total or 0) - (prev or 0), }

三个指标各有阈值：rows应当和标的池规模基本一致，波动超过 5% 就要查；tradable_ratio正常在 0.95 以上，低于 0.9 说明当天大面积停牌或者数据源出问题；delta_vs_prev绝对值超过 20 就要人工看一眼，因为正常情况下每天新增的标的数应该接近零。把这些值按日写进一张check_log表，跑一个月之后你就有了一条基线，后面任何偏离都能立刻看出来。

定时任务本身用系统自带的调度就够了，关键是两件事：一是任务在收盘后足够晚的时间启动（比如 18:30），避开数据源当天的更新窗口；二是任务结束必须写一条带状态码的日志，失败时能重跑，而不是靠人记得。我一般把抓取、清洗、回测拆成三个独立任务串起来，中间用文件或表做交接，任何一环失败后面不启动，避免用半成品数据算出一堆看着正常的错误结果。

下面这张表是我常用的启动时间与检查项对照，可以直接照抄改：

有一次我在调一个动量因子的窗口参数，为了让曲线好看，顺手把回测区间改成了「剔除所有下跌月份」。跑出来的夏普 4.2，我当时还真兴奋了半分钟，直到发现代码里那段区间过滤是硬编码的。从那以后我每次改完参数，都强制走一遍「随机信号对照 + 单标的贡献占比 + 换手率检查」这三步，任何一步不过就直接回滚，不看收益曲线。踩过的坑记在本子上，比记在脑子里可靠得多。希望帮到你。

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