简介：这份资料面向希望用Python打通股票数据采集、清洗、分析与建模全流程的金融数据分析学习者和量化初学者，适合具备Python基础、想积累金融数据项目经验的人群参考。内容围绕爬虫抓取、预处理、可视化、机器学习预测与实时数据应用展开，涉及BeautifulSoup、Scrapy抓取行情与公告，Pandas处理缺失值和异常值，NumPy、SciPy做统计计算，Matplotlib、Seaborn绘制K线与波动图，并用Scikit-learn尝试价格走势预测，最后结合Tushare、Wind接口及Flask、Django展示分析结果。压缩包约4.47MB，文件总数与类型明细上游暂未提供，需以下载后实际目录为准；已有3659人学习，说明该主题具备较高关注度。其价值在于把爬虫、数据清洗、特征分析、模型训练和Web呈现串成一条可复用的实战路线，便于读者按模块理解股票大数据项目的关键环节，并据此搭建自己的分析脚本与实验流程。

1. 一份没人愿意手动整理的日线表

做过 Python 股票大数据分析的人，大多经历过同一个阶段：收盘后开着一堆行情页面，把开盘、最高、最低、收盘、成交量一个个抄进表格，抄到第三十只标的的时候已经开始怀疑人生。这个实战项目就是把这段重复劳动彻底交出去——用 Python 爬虫按交易日把公开行情数据拉下来，用 pandas 做清洗和复权校正，算出均线、MACD、RSI 这些常用指标，最后用 matplotlib 和 plotly 出图，再按条件批量筛出符合形态的标的。

它适合两类人：一类是刚学完 requests 和 pandas、想找个有真实结构的数据源练手的开发者；另一类是想把盯盘这件事变成可回放、可验证流程的量化爱好者。整套代码是单机可跑的分析管线，不依赖付费接口和云端服务，装好依赖就能在本地跑出一份自己的日报。

2. 技术选型与整体架构：一条数据从抓取到出图的完整链路

2.1 为什么是 requests + pandas，而不是 Scrapy 全家桶

这个项目的数据源是公开行情接口返回的 JSON 与 HTML 页面，拿到的本来就是结构化数据。Scrapy 那套调度器、去重队列、下载中间件是为海量页面爬取设计的，用在这里属于杀鸡用牛刀，配置成本反而高过收益。我一般的选择是requests.Session做连接复用和频率控制，配合pandas做字段规整，两百来行就能覆盖采集层的全部需求。

真正需要框架的地方在数据侧。日线数据一旦拉下来，后续全是列式运算：rolling 均线、ewm 指数平滑、跨表 merge、分组筛选。这些用 pandas 写十几行，用纯 Python 循环写两百行还容易出错。所以分工很清楚：requests 负责「把字节拿回来」，pandas 负责「把字节变成能算的表」，SQLite 负责「把表存住」。

依赖清单不长，下面这份是我实际跑通的组合：

# requirements.txt 核心依赖 requests==2.31.0 # 采集层：会话复用 + 超时控制 pandas==2.1.4 # 清洗与指标计算的主力 numpy==1.26.2 # pandas 底层依赖，偶尔直接用 lxml==4.9.3 # HTML 页面解析，接口返回 JSON 时用不上 matplotlib==3.8.2 # 静态日报图 plotly==5.18.0 # 交互式 K 线，方便放大看局部 APScheduler==3.10.4 # 收盘后定时触发整条管线

装依赖这一步没什么玄学，真正容易翻车的是 pandas 2.x 和 1.x 的行为差异：df.append被移除、inplace参数在链式调用里返回 None、时间频率字符串从M改成ME。如果手上的旧代码是 pandas 1.x 写的，建议直接把 pandas 钉在 1.5.3，别一边改业务逻辑一边改 API 兼容，血泪经验。

2.2 目录结构：五个包各管一段

项目按数据流向切包，而不是按技术分层切。每个包只对上游的产物负责，这样任何一个环节出问题，看一眼目录就知道该去哪找。结构大致是这样：

stock_analysis/ ├── config/ │ └── settings.py # 接口地址、字段映射、并发阈值 ├── collector/ # 采集层 │ ├── session.py # 会话、重试、UA │ ├── fetcher.py # 单个标的的单日/区间抓取 │ └── calendar.py # 交易日历，增量抓取的基准 ├── storage/ # 存储层 │ ├── schema.sql # 建表语句 │ └── dao.py # 幂等写入与查询封装 ├── analysis/ # 计算层 │ ├── clean.py # 缺失值、异常值、复权校正 │ ├── indicator.py # MA / MACD / RSI / 布林带 │ └── screener.py # 条件选股 ├── report/ # 展示层 │ ├── plot_static.py # matplotlib 日报 │ └── plot_interactive.py # plotly 交互图 ├── main.py # 管线入口 └── data/ └── market.db # SQLite 单文件库

各模块的职责和依赖关系列成表更直观：

这套结构的好处是采集层可以被整体替换。哪天接口换了协议，只需要重写collector/fetcher.py里的解析函数，analysis和report一行都不用动。很多人写这类项目会把请求和计算糊在一个脚本里，跑通没问题，但半个月后想换个数据源就得推倒重来。

2.3 数据流与主入口：一次 run 走过四个阶段

整条管线是单向的，入口函数只做编排，不写业务。这样做的直接收益是你可以单独跑任何一个阶段——调试指标时不用每次都重新抓数据。

# main.py 管线编排 from collector.fetcher import fetch_range from storage.dao import save_bars, load_bars from analysis.clean import clean_bars, apply_adjust from analysis.indicator import add_indicators from analysis.screener import screen from report.plot_static import draw_daily_report def run(codes, start, end): # 阶段一：采集，只抓本地库里没有的交易日 raw = fetch_range(codes, start, end) # 阶段二：存储，主键冲突时覆盖写，保证可重复执行 save_bars(raw) # 阶段三：计算，先清洗再复权，最后算指标 df = load_bars(codes, start, end) df = clean_bars(df) df = apply_adjust(df) df = add_indicators(df) # 阶段四：筛选与出图 picked = screen(df) draw_daily_report(df, picked, out_dir="output") return picked

四个阶段的参数含义值得说清楚：codes传标的代码列表，形如["600XXX", "000XXX"]，脱敏后用XXX占位；start和end是YYYY-MM-DD字符串，内部统一转成datetime再比较，避免字符串比较在跨年时出错；save_bars内部用INSERT OR REPLACE，所以整条管线是幂等的——同一天跑三遍不会产生重复行，这点在调试时省了大量清理工作。

一个常见的编排错误是把clean放在apply_adjust之后。复权因子本身也可能有缺失，正确的顺序是先丢掉缺失严重的行、再做复权乘法，否则空值乘任何数还是空值，最后指标全是 NaN，排查半天才发现是顺序问题。

3. 爬虫采集层拆解：把公开行情接口稳定地拉下来

3.1 请求封装：会话复用、超时、重试三件套

采集层最容易写错的地方不是解析，而是异常处理。新手写爬虫习惯一个大for循环里直接requests.get，遇到超时就整个崩掉，重跑一遍又要从头开始。合格的封装至少要解决三件事：连接复用、超时、有限次重试。

# collector/session.py import time import random import requests def build_session(retries=3, backoff=1.5): """构造带重试策略的会话，全局复用一个实例""" sess = requests.Session() # 固定 UA 容易被识别，这里从几个常见 UA 里随机取 sess.headers.update({ "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36"), "Accept": "application/json, text/plain, */*", }) # 挂载重试：只对连接错误和 5xx 重试，4xx 直接抛 adapter = requests.adapters.HTTPAdapter( max_retries=retries, pool_connections=10, pool_maxsize=10, ) sess.mount("https://", adapter) sess.backoff = backoff return sess def fetch_json(sess, url, params, timeout=8, pause=(0.3, 0.8)): """取单个 JSON 响应，失败时退避重试""" for attempt in range(3): try: resp = sess.get(url, params=params, timeout=timeout) resp.raise_for_status() # 4xx/5xx 走异常分支 time.sleep(random.uniform(*pause)) # 请求间隔随机化 return resp.json() except (requests.Timeout, requests.ConnectionError) as exc: if attempt == 2: raise # 指数退避：1.5 -> 2.25 -> 3.375 秒 time.sleep(sess.backoff ** attempt)

max_retries交给HTTPAdapter处理连接级失败，fetch_json里的循环处理超时这类应用级失败，两层互不冲突。pause参数是区间随机化，固定 0.5 秒的间隔在服务端看来是一条完美的等距直线，加一点抖动更像人。timeout=8是我反复调出来的值：设 3 秒会在网络抖动时大面积误判超时，设 30 秒会让一个卡住的请求拖垮整批任务。

raise_for_status()这行别省。少了它，接口返回 404 或者 403 的错误页时，resp.json()会抛一个跟业务毫无关系的解析异常，日志里根本看不出是权限问题还是数据问题。

3.2 分页拉取与增量更新：只抓新交易日

全量抓取的代价是线性的：标的数量乘以交易日数量。两百只标的、五年日线就是二十多万条记录，每次都重抓既慢又容易触发限流。增量策略是这套代码里最值钱的部分——先从本地库查每个标的的最新交易日，再从那天往后补齐。

# collector/fetcher.py import datetime as dt from storage.dao import get_last_date, save_bars from collector.session import build_session, fetch_json def trading_days(start, end, holidays=()): """生成区间内的交易日，简单剔除周末和给定假期""" cur = start while cur <= end: if cur.weekday() < 5 and cur.strftime("%Y-%m-%d") not in holidays: yield cur cur += dt.timedelta(days=1) def fetch_range(codes, start, end, sess=None): sess = sess or build_session() rows = [] for code in codes: # 本地已有数据的最后一天，没有则从头开始 last = get_last_date(code) begin = start if last is None else last + dt.timedelta(days=1) if begin > end: continue # 已是最新，跳过 for day in trading_days(begin, end): data = fetch_json(sess, "https://example-market/api/daily", { "code": code, "date": day.strftime("%Y%m%d"), }) rows.extend(parse_daily(code, day, data)) return rows def parse_daily(code, day, payload): """把接口返回的嵌套结构摊平成一行行记录""" out = [] for item in payload.get("items", []): out.append({ "ts_code": code, "trade_date": day.strftime("%Y-%m-%d"), "open": float(item["o"]), "high": float(item["h"]), "low": float(item["l"]), "close": float(item["c"]), "pre_close": float(item.get("pc", 0)), "volume": float(item["v"]), "amount": float(item.get("a", 0)), }) return out

get_last_date返回None时走全量，返回具体日期时从次日开始，这是增量的关键判断。begin > end那个continue不能少，否则对已经是最新的标的还会白跑一轮空循环，标的多了就是几分钟的浪费。

parse_daily里全部字段做float转换，别偷懒。接口返回的价格有时候是字符串，有时候是null，直接进 pandas 会变成object列，后面rolling().mean()直接报TypeError，而且报错位置在指标模块，排查方向会跑偏。

3.3 落库：SQLite 表结构与幂等写入

存储层用 SQLite，理由很实际：单文件、零部署、拷走就能发给别人复现。数据量上到千万级再考虑换 MySQL，换的时候只改dao.py里的连接串和几条方言语句。

-- storage/schema.sql CREATE TABLE IF NOT EXISTS daily_bar ( ts_code TEXT NOT NULL, -- 标的代码，如 600XXX trade_date TEXT NOT NULL, -- 交易日，YYYY-MM-DD open REAL, high REAL, low REAL, close REAL, pre_close REAL, -- 前收盘价，用于算涨跌幅 volume REAL, -- 成交量，单位：手 amount REAL, -- 成交额，单位：元 adj_factor REAL DEFAULT 1.0, -- 复权因子，初始化填 1 PRIMARY KEY (ts_code, trade_date) ); CREATE INDEX IF NOT EXISTS idx_date ON daily_bar(trade_date);

复合主键(ts_code, trade_date)是不重复的保证，INSERT OR REPLACE依赖它做冲突覆盖。日期上单独建索引，是因为筛选和分组几乎都是按时间范围来的，没索引的表在几十万行时查询会从毫秒级掉到秒级。

# storage/dao.py import sqlite3 import pandas as pd DB = "data/market.db" def save_bars(rows): if not rows: return 0 with sqlite3.connect(DB) as conn: conn.executemany( """INSERT OR REPLACE INTO daily_bar (ts_code, trade_date, open, high, low, close, pre_close, volume, amount, adj_factor) VALUES (:ts_code, :trade_date, :open, :high, :low, :close, :pre_close, :volume, :amount, :adj_factor)""", [{**r, "adj_factor": r.get("adj_factor", 1.0)} for r in rows], ) return len(rows) def load_bars(codes, start, end): sql = """SELECT * FROM daily_bar WHERE ts_code IN ({}) AND trade_date BETWEEN ? AND ? ORDER BY ts_code, trade_date""".format( ",".join("?" * len(codes))) with sqlite3.connect(DB) as conn: return pd.read_sql(sql, conn, params=[*codes, str(start), str(end)])

executemany比循环execute快一个数量级，这是 SQLite 的写入优化点。load_bars里 SQL 的占位符用join拼出来，参数仍然走params传入，没有拼接字符串，所以不存在注入问题；ORDER BY ts_code, trade_date是给后面的groupby和shift铺路，顺序乱了均线会算错。

4. 指标计算与可视化：从原始 K 线到能看懂的图

4.1 指标计算：均线、MACD、RSI 三件套

指标层是纯计算，输入输出都是 DataFrame，不碰 IO，所以最好写单元测试。核心原则是按标的groupby之后再算，否则 A 标的的收盘价会和 B 标的的接在一起，均线直接失真。

# analysis/indicator.py import pandas as pd import numpy as np def add_indicators(df, fast=12, slow=26, signal=9, rsi_n=14): """输入长表，按标的分别计算，返回追加指标列的长表""" df = df.sort_values(["ts_code", "trade_date"]).copy() out = [] for code, g in df.groupby("ts_code", sort=False): g = g.reset_index(drop=True) close = g["close"] # 均线：窗口不足时是 NaN，属正常 g["ma5"] = close.rolling(5).mean() g["ma20"] = close.rolling(20).mean() # MACD：EMA 用 adjust=False 才是递推定义 ema_fast = close.ewm(span=fast, adjust=False).mean() ema_slow = close.ewm(span=slow, adjust=False).mean() g["dif"] = ema_fast - ema_slow g["dea"] = g["dif"].ewm(span=signal, adjust=False).mean() g["macd"] = (g["dif"] - g["dea"]) * 2 # 国内习惯乘 2 # RSI：用 Wilder 平滑，alpha = 1/n delta = close.diff() gain = delta.clip(lower=0) loss = -delta.clip(upper=0) avg_gain = gain.ewm(alpha=1 / rsi_n, adjust=False).mean() avg_loss = loss.ewm(alpha=1 / rsi_n, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) g["rsi"] = 100 - 100 / (1 + rs) out.append(g) return pd.concat(out, ignore_index=True)

参数里fast/slow/signal是 MACD 的三个标准周期，改成别的值也行，但改完要和行情软件对齐后再确认，否则会以为代码算错了。adjust=False这个参数必须显式写，默认是True，两种算法的结果在序列前期差异明显。macd乘 2 是国内软件的惯例，画柱状图时视觉高度差一倍，不乘会看着别扭。

avg_loss.replace(0, np.nan)处理的是连续上涨导致平均跌幅为零的情况。不替换的话rs变成无穷大，rsi算出NaN，绘图时那一截直接断线。

4.2 出图：matplotlib 画静态日报，plotly 画交互

两套绘图各管一段场景。收盘后生成一张 PNG 发给自己看，用 matplotlib 就够；需要缩放、悬浮看某一天的具体数值，用 plotly 导出 HTML。

# report/plot_static.py import matplotlib matplotlib.use("Agg") # 无界面环境必须指定后端 import matplotlib.pyplot as plt import matplotlib.dates as mdates import pandas as pd plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"] plt.rcParams["axes.unicode_minus"] = False def draw_one(g, title, out_path): g = g.copy() g["trade_date"] = pd.to_datetime(g["trade_date"]) fig, (ax1, ax2) = plt.subplots( 2, 1, figsize=(12, 7), sharex=True, gridspec_kw={"height_ratios": [3, 1]}) ax1.plot(g["trade_date"], g["close"], lw=1.2, label="close") ax1.plot(g["trade_date"], g["ma5"], lw=0.9, label="MA5") ax1.plot(g["trade_date"], g["ma20"], lw=0.9, label="MA20") ax1.set_title(title) ax1.legend(loc="upper left", fontsize=9) ax2.bar(g["trade_date"], g["macd"], width=1.0) ax2.axhline(0, color="gray", lw=0.6) ax2.xaxis.set_major_formatter(mdates.DateFormatter("%m-%d")) fig.autofmt_xdate() fig.tight_layout() fig.savefig(out_path, dpi=120) plt.close(fig) # 不关会累积内存

matplotlib.use("Agg")是跑在服务器或无显示器环境的关键，不加这一行会在pyplot初始化时直接抛后端异常。font.sans-serif那段是中文字体兜底，Windows 上用SimHei，Linux 上得换成Noto Sans CJK SC之类，配错了图上全是方块。plt.close(fig)千万别漏，批量生成几十张图不关句柄，内存会一路涨到被系统杀掉。

sharex=True让上下两张图共用一个时间轴，缩放时联动，比手动同步xlim省事。height_ratios把主图压到 3/4 高度，MACD 给 1/4，比例看着最舒服。

4.3 批量选股：把筛选条件写成配置

筛选逻辑硬编码在循环里是最难维护的写法。把条件抽成字典，加一条规则只改配置，不动代码。

# analysis/screener.py import pandas as pd # 条件全部写成 lambda，方便单测和热调 RULES = { "站上均线": lambda d: d["close"] > d["ma20"], "短期多头": lambda d: d["ma5"] > d["ma20"], "MACD 金叉": lambda d: (d["dif"] > d["dea"]) & (d["dif"].shift(1) <= d["dea"].shift(1)), "量能放大": lambda d: d["volume"] > d["volume"].rolling(5).mean() * 1.5, } def screen(df, rules=None, top_n=20): """对每只标的取最后一个交易日，逐条规则判定""" rules = rules or RULES last = df.groupby("ts_code").tail(1).set_index("ts_code") mask = pd.Series(True, index=last.index) for name, fn in rules.items(): hit = fn(last).fillna(False) mask &= hit result = last[mask].sort_values("amount", ascending=False) return result.head(top_n)

tail(1)取每只标的最后一行，假设数据已经是最新的。如果库里数据只到上周，这个筛选结果就是过期的，所以我在main.py里会先检查max(trade_date)和当前日期差了多少天，超过 3 个自然日就打印警告。

fillna(False)处理的是新股或数据不足导致指标为 NaN 的情况。不填的话mask &= NaN会让整个结果变成空，看着像「今天一只都没选中」，其实是数据类型在作怪。sort_values("amount")按成交额排序，比按涨幅排序稳定，不会把几只异常跳动的标的顶到最前面。

5. 踩坑排查：爬虫与数据分析里最容易翻车的五个地方

5.1 现象：抓下来的收盘价和行情软件差了几分钱

原因几乎都是复权口径不一致。行情软件默认展示前复权价，而原始接口给的是不复权价，遇到除权除息日两者就会岔开。这个差异在指标上会被放大：一根假的跳空会把 MA20 拉歪，MACD 也会在错误的位置金叉。

解决办法是显式做复权转换，别指望数据源自带。

# analysis/clean.py def apply_adjust(df): """按复权因子折算，factor 为 1 时等价于不复权""" df = df.sort_values(["ts_code", "trade_date"]).copy() # 后复权：价格 * 累计因子 df["close_adj"] = df["close"] * df["adj_factor"] df["open_adj"] = df["open"] * df["adj_factor"] # 成交量反向调整，保持成交额守恒 df["volume_adj"] = df["volume"] / df["adj_factor"] return df

改完之后和软件对数，误差应该在一分钱以内。如果还是对不上，先确认是不是接口本身做了四舍五入，再检查是不是把pre_close当成了前复权价用。

5.2 现象：跑了半小时，后半段全是空数据

原因是被限流了。接口没返回 403，而是返回了一个结构相同但items为空的 JSON，raise_for_status()完全不认为这是错误，代码就当成「当天没有数据」继续往下跑，最后库里多了一堆空行。

解决办法有两层。第一层是识别空响应并抛出明确异常，让它进重试分支；第二层是控制并发和节奏。

def fetch_json(sess, url, params, timeout=8, pause=(0.3, 0.8)): resp = sess.get(url, params=params, timeout=timeout) resp.raise_for_status() data = resp.json() if not data.get("items"): raise ValueError(f"空响应: {params}") # 触发上层重试 time.sleep(random.uniform(*pause)) return data

并发上我一般把线程数压在 3 到 5 之间，再配一个全局限速器。真遇到连续空响应，先停下来歇十分钟，硬刚只会让限流窗口越来越长。

5.3 现象：控制台刷一屏 SettingWithCopyWarning，结果还时对时错

原因是链式赋值。df[df["close"] > 0]["ma5"] = ...这种写法，前半段返回的是视图还是副本不确定，pandas 只能发警告，实际行为依赖底层内存布局，同一份代码换个环境结果就变。

解决办法是养成两步走的习惯：先筛出子集并显式.copy()，再做赋值。

# 错误写法：可能在副本上改，原表没变 # df[df["close"] > 0]["ma5"] = df["close"].rolling(5).mean() sub = df[df["close"] > 0].copy() # 显式复制 sub["ma5"] = sub["close"].rolling(5).mean() # 再赋值

批量处理时我通常在函数入口统一.copy()一次，后面随便改，比逐行加copy()省心。代价是多占一份内存，十万行级别的数据完全扛得住。

5.4 现象：除权除息日附近，均线突然出现一个巨大的缺口

原因是不复权序列在除权日天然跳空。10 派 5 元的标的，除权当天价格直接下移，工程上这就是一条真实存在的价格断层，均线跟着断很正常。但对形态识别来说它是噪声，会把「回踩均线」误判成「破位」。

解决办法是把指标计算统一放到复权价上，原始价只用于展示。

def add_indicators(df, price_col="close_adj"): """所有指标基于复权价计算，避免除权跳空干扰""" df = df.sort_values(["ts_code", "trade_date"]).copy() for code, g in df.groupby("ts_code", sort=False): close = g[price_col] df.loc[g.index, "ma20"] = close.rolling(20).mean().values return df

price_col做成参数，切换复权口径时不用改函数体。df.loc[g.index, ...]是按原索引回写，比concat再排序快，也不容易打乱行序。

5.5 现象：图上中文全是方块，负号也显示成方框

原因是 matplotlib 默认字体不含中文字形，找不到就退化成豆腐块。这个问题在 Windows 上不明显，一上 Linux 服务器就全暴露。

解决分两步：先确认系统里装了中文字体，再在绘图模块顶部设置。

# 先看系统里有哪些中文字体可用 fc-list :lang=zh | head -5

import matplotlib matplotlib.use("Agg") # 无界面环境必加，且必须在 pyplot 之前 import matplotlib.pyplot as plt # 按优先级回退，第一个找不到就用第二个 plt.rcParams["font.sans-serif"] = [ "Noto Sans CJK SC", "SimHei", "Microsoft YaHei", "WenQuanYi Zen Hei" ] plt.rcParams["axes.unicode_minus"] = False # 负号用 ASCII 连字符

axes.unicode_minus = False这行解决的是负号问题，中文字体里通常没有 U+2212 这个减号字形。设成False后 matplotlib 会用普通连字符，MACD 柱状图的负值就不会变成方框。

6. 二次开发：把它改成自己的盯盘与预警工具

代码跑通只是起点，这套管线真正好用的地方在于每一层都能单独换掉。我在自己那份上做过几轮改造，按投入产出排了个序，可以直接照着改：

最值得先做的是定时任务，因为它把「想起来了才跑」变成「每天自动出结果」，习惯一旦建立，这套代码才算真正长在你身上。

# scheduler.py 收盘后 15 分钟触发，避开数据源的结算窗口 from apscheduler.schedulers.blocking import BlockingScheduler import datetime as dt from main import run sched = BlockingScheduler(timezone="Asia/Shanghai") @sched.scheduled_job("cron", day_of_week="mon-fri", hour=15, minute=30) def daily_job(): end = dt.date.today() start = end - dt.timedelta(days=30) # 回溯 30 天补缺口 picked = run(["600XXX", "000XXX"], start, end) print(f"{end} 命中 {len(picked)} 只") sched.start()

注意day_of_week只排除了周末，节假日得靠交易日历判断，不然放假期间也会触发，抓回来一堆空数据，还会误报预警。

回测那块我建议先别急着上框架。用现有的screen()加上「命中后第 N 天收益」这一列，几十行就能看出规则有没有用：

def simple_backtest(df, hold_days=5): """统计命中的标的后 N 个交易日的收益分布""" df = df.sort_values(["ts_code", "trade_date"]).copy() df["fwd_ret"] = ( df.groupby("ts_code")["close_adj"].shift(-hold_days) / df["close_adj"] - 1 ) hits = df[df["ma5"] > df["ma20"]].dropna(subset=["fwd_ret"]) return hits["fwd_ret"].describe()

shift(-hold_days)是向未来取数，只能用在历史数据上验证，实盘里拿不到。这个函数最有价值的是describe()出来的那几个分位数，如果 25% 分位数是 -8%，说明这套规则在震荡市里会频繁打到止损，光看均值会亏得很难受。

改造时最容易犯的错是一次动太多。我吃过这个亏：一口气把存储换成 MySQL、把指标重算一遍、又改了复权口径，结果回测收益从正变负，花了两个晚上才定位到是复权因子在换库时默认成了 1，跟指标毫无关系。从那以后我每次改完一个模块，都会先单独跑一遍该模块的对照输出，确认和改动前的结果差异能解释，再往下走。希望帮到你。

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