简介：这是一套基于Qt框架的股票行情分析程序源码，面向希望学习Qt界面开发与股票数据可视化的开发者。项目完整实现了文章所展示的股票查询与图表交互效果，包含行情列表、分时走势、K线图、五日图及资金视图等功能模块。图表部分采用QChart绘制，并借助自定义表格模型与委托实现数据展示与点击联动。
压缩包共82个文件，以34个头文件与32个C++源文件为主体，另有8个Qt界面文件、2个工程文件及1个SQLite数据库文件，整体体积仅232KB，结构紧凑。工程按功能拆分为多个类与界面，便于阅读和二次开发。
目前已有728人学习下载。通过研读源码，可掌握QChart的多种图表用法、Model/View框架下的表格定制、拼音首字母搜索、股票数据文件解析等实用技巧，适合具备基础C++和Qt知识、希望深入项目实践的开发人员。
1. shares 和 stockAnalyse：先把数据层和分析层拆开
很多人第一次看到 shares 和 stockAnalyse 这两个模块摆在一起，会误以为它们是同一个工具的重复实现。其实分工很明确：shares 负责把原始行情数据拿回来、清洗干净、按规则缓存好，stockAnalyse 负责在干净数据上计算指标、生成交易信号、跑回测。前者管“数据对不对”，后者管“信号准不准”，两者拆开之后，你换数据源不用改分析逻辑，换分析策略也不用重写数据层。这篇文章适合正在做股票量化验证、又不想被复杂框架绑架的开发者——用最朴素的 Python 代码，把这两个模块从零搭起来，并说清楚参数怎么设、坑在哪里。
2. shares 数据层：用最小代码把日线数据管起来
2.1 选型：为什么选择“文件缓存 + 增量拉取”而不是直接上数据库
很多新手一上来就考虑 MySQL、PostgreSQL，或者引入专业的时序数据库，结果数据还没拉多少天，先花了两周在环境搭建上。我一般的做法是：分析阶段的数据量根本到不了数据库的规模。单只股票十多年日线也不过三千行，就算每秒级 tick 数据，本地用 Parquet 也能扛住。shares 模块在早期验证阶段，用“文件缓存 + 增量拉取”是最稳定的组合。
文件缓存的意思是：把每次从公开行情接口拿到的数据，落到本地 CSV 或 Parquet 文件里，下次启动时先读本地文件，只有本地文件缺失或过期时才重新请求接口。增量拉取的意思更简单：本地已经有 2020 年到昨天的数据，就只请求昨天到今天这一小段，而不是每次全量重拉。这样既绕开了接口频率限制，也避免每次启动都花几十秒做无用功。
实现上，我建议用 pandas 的 DataFrame 作为内存中的统一数据格式，落盘格式用 Parquet。Parquet 比 CSV 好在三点：读取速度快一个数量级、自动保留数据类型、支持按行组压缩减小体积。但如果你的 Python 环境没有 pyarrow，先用 CSV 也不要紧，shares 的接口设计可以保证底层存储换成什么都只需要改一个方法。
# shares/storage.py from pathlib import Path import pandas as pd DATA_DIR = Path("./data") def load_local(symbol: str, start: str, end: str) -> pd.DataFrame | None: """从本地缓存读取区间数据，缺失返回 None""" f = DATA_DIR / f"{symbol}.parquet" if not f.exists(): return None df = pd.read_parquet(f) mask = (df["date"] >= start) & (df["date"] <= end) return df.loc[mask].reset_index(drop=True) if mask.any() else None def save_local(symbol: str, df: pd.DataFrame) -> None: """全量落盘：先写临时文件再替换，避免写一半损坏""" DATA_DIR.mkdir(exist_ok=True) tmp = DATA_DIR / f"{symbol}.tmp.parquet" df.to_parquet(tmp, index=False) tmp.replace(DATA_DIR / f"{symbol}.parquet")逻辑说明：load_local 先按文件存在性判断，避免每次请求都打接口；save_local 采用“临时文件 + 原子替换”的方式，防止程序中途崩溃把主文件写坏，这是很多自写缓存工具翻车的地方。
参数说明：start 和 end 统一用"YYYY-MM-DD"字符串传入，在比较时依赖字符串排序与日期排序一致这个特性，省去一次日期解析。如果你的本地文件记录的是带时区的 datetime，这里要先把日期列统一成YYYY-MM-DD字符串再比较，否则会遇到“明明有数据却查不到”的怪问题。
2.2 拉取与落盘：shares 模块的最小实现
有了存储层，接下来写拉取逻辑。公开行情接口通常返回 JSON，里面包含开高低收、成交量、成交额等字段。shares 的职责不是做分析，而是把这些字段标准化成统一的列名：date、open、high、low、close、volume、amount。标准化这一步特别重要，因为不同数据源的字段名差异很大，有的叫trade_date，有的叫day，统一之后 stockAnalyse 才不用关心上游是谁。
# shares/fetcher.py import requests import pandas as pd from pathlib import Path def fetch_daily(symbol: str, start: str = "20000101", end: str = "") -> pd.DataFrame: """从公开接口拉取日线数据并标准化列名""" if not end: end = pd.Timestamp.today().strftime("%Y%m%d") url = f"https://api.example.market/daily?code={symbol}&start={start}&end={end}" resp = requests.get(url, timeout=10) resp.raise_for_status() raw = resp.json()["data"]["items"] df = pd.DataFrame(raw) df = df.rename(columns={ "trade_date": "date", "open_price": "open", "high_price": "high", "low_price": "low", "close_price": "close", "volume": "volume", "amount": "amount", }) df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d") df = df.sort_values("date").drop_duplicates(subset="date", keep="last") return df[["date", "open", "high", "low", "close", "volume", "amount"]]逻辑说明：请求的 URL 和字段名是示意——真实接口需要按你自己的数据源替换。关键是最后两行：排序去重。同一交易日可能因为接口重试而返回重复行，保留最后一条通常是服务端最新修正的数据。这在分红除权日尤其重要，因为部分接口会返回修正前后的两条记录。
参数说明：timeout=10 是必写的，股票接口在开盘时段经常变慢，不设超时会导致线程卡死。symbol 建议直接传交易所标准代码（如SH.600000或HK.00700这类带交易所前缀的格式），避免后续不同市场的代码歧义。start 默认给 20000101，是因为 A 股早期数据对策略验证价值有限，反而容易把退市整理期的异常波动带进来。
2.3 把一份原始数据变成干净的数据集
拉下来的数据不能直接用。真实行情数据有几个常见脏点：停牌日返回空数据或零成交；部分接口在复权因子变更后返回的历史价格会有跳变；还有极少数情况会混入前复权价和后复权价不一致的行。shares 的清洗逻辑要解决三个问题：空值怎么补、重复怎么去、停牌日怎么标记。
def clean_daily(df: pd.DataFrame) -> pd.DataFrame: """清洗日线数据：去空、去重、标记停牌""" df = df.copy() df["date"] = pd.to_datetime(df["date"]) df = df.sort_values("date").drop_duplicates(subset="date", keep="last") # 成交量或收盘价为空的整行删除 df = df.dropna(subset=["close", "volume"]) # 成交量极低且高低价相同的，标记为疑似停牌 df["suspend"] = (df["volume"] <= 0) | (df["high"] == df["low"]) return df.reset_index(drop=True) def update_cache(symbol: str) -> pd.DataFrame: """增量更新入口：先读本地，再拉新增段""" local = load_local(symbol, "1970-01-01", "2099-12-31") if local is None or local.empty: df = fetch_daily(symbol) df = clean_daily(df) save_local(symbol, df) return df last_date = local["date"].max() start = (pd.Timestamp(last_date) + pd.Timedelta(days=1)).strftime("%Y%m%d") new_df = fetch_daily(symbol, start=start) if new_df.empty: return local merged = pd.concat([local, clean_daily(new_df)], ignore_index=True) merged = merged.drop_duplicates(subset="date", keep="last").sort_values("date") save_local(symbol, merged) return merged逻辑说明：update_cache 是 shares 对外的核心入口。它先从本地缓存取全量数据，取不到就全量拉取；取得到就按“本地最后日期 + 1 天”作为新起点做增量拉取，最后合并去重。suspend列是被分析层使用的标记，而不是直接删掉停牌日——因为某些动量策略需要知道“这只股票停牌了多久”，直接删行会丢失这个信息。
参数说明：注意pd.Timedelta(days=1)的用法——它跳过的是自然日而不是交易日，所以增量拉取会请求到周末和节假日，但接口对无交易日期通常会返回空列表，不会产生额外数据行。如果你用的是严格按交易日返回的接口，这里可以不做任何改动。
3. stockAnalyse 分析层：指标计算和信号生成的落地方式
3.1 复权和价格序列的处理顺序
stockAnalyse 接收的是 shares 清洗后的 DataFrame，但第一步不是直接算 MACD，而是先检查复权状态。未复权数据在除权除息日会出现价格跳空，比如一只股票 10 派 5，除息日股价直接从 20 元跳到 19.5 元，这在未复权序列里看起来像“跌了 2.5%”，实际上没有任何交易发生。如果你直接在未复权序列上算均线或突破信号，每年都会产生几次假突破。
常见的做法是前复权：把历史价格统一按最新除权因子做调整，让历史价格整体下移，保证价格序列连续。但它有一个副作用——历史价格是动态变的。每次除权发生，过去所有的价格都会跟着变一次，所以 stockAnalyse 里每次计算指标前都应该重新复权，而不是缓存复权后的数据。后复权则相反，历史价格保持真实成交价，最新价格被调整到未来值，更适合做长周期收益计算。我通常在信号生成阶段用前复权序列，在绩效评估阶段再切换到后复权序列验证一次，避免单一系列掩盖问题。
如果你的行情源没有直接提供复权数据，可以用“复权因子表”自己算。shares 的缓存文件里可以多存两列factor和prev_close，分析层用close * factor / factor.iloc[-1]得到前复权收盘价。这里最容易被忽略的是：复权应该对open、high、low、close四列同时做，只调整close会导致高低价关系错乱，进而让 AT R 这类波动率指标失真。
def adjust_price(df: pd.DataFrame, mode: str = "forward") -> pd.DataFrame: """按复权因子调整价格序列，支持前复权和后复权""" out = df.copy() if "factor" not in out.columns: return out f = out["factor"] if mode == "forward": ratio = f / f.iloc[-1] elif mode == "backward": ratio = f / f.iloc[0] else: ratio = 1.0 for col in ["open", "high", "low", "close"]: out[col] = out[col] * ratio return out逻辑说明：前复权把最新因子作为基准，历史价格反推回去；后复权以最早因子为基准，最新价格被放大。两种模式都不改变相对涨跌幅。注意这个函数假设因子列已经按日期排序且单调可乘——如果 shares 清洗阶段做过去重，这一点通常成立。
参数说明：mode参数决定基准点。实际使用中，我会在信号生成前调用adjust_price(df, "forward")，在回测结果评估时调用adjust_price(df, "backward")再重跑一次回测，两次收益偏差如果超过 5%，基本可以断定策略对复权敏感，需要回到数据层排查。
3.2 指标计算模块：EMA、MACD 与 RSI 的稳定实现
指标计算是 stockAnalyse 的核心，牵一发动全身。pandas 自带的ewm、rolling很好用，但细节上有很多坑。比如ewm(span=12)里的 span 对应的是多少根 K 线的衰减周期，很多人直接照搬传统软件上的 “12”，其实差异不大，可一旦指标被用在分钟线上就需要重新校准。我习惯把所有指标封装成纯函数，输入 DataFrame，输出追加了指标列的新 DataFrame，不修改原始数据，这样复现结果时可追溯。
def add_ema(df: pd.DataFrame, span: int = 12, col: str = "close") -> pd.DataFrame: out = df.copy() out[f"ema_{span}"] = out[col].ewm(span=span, adjust=False).mean() return out def add_macd(df: pd.DataFrame, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame: out = df.copy() out["ema_fast"] = out["close"].ewm(span=fast, adjust=False).mean() out["ema_slow"] = out["close"].ewm(span=slow, adjust=False).mean() out["macd"] = out["ema_fast"] - out["ema_slow"] out["macd_signal"] = out["macd"].ewm(span=signal, adjust=False).mean() out["macd_hist"] = out["macd"] - out["macd_signal"] # 柱状图的值 return out def add_rsi(df: pd.DataFrame, period: int = 14) -> pd.DataFrame: out = df.copy() delta = out["close"].diff() gain = delta.clip(lower=0).rolling(window=period).mean() loss = (-delta.clip(upper=0)).rolling(window=period).mean() rs = gain / loss.replace(0, pd.NA) out["rsi"] = 100 - (100 / (1 + rs)) out["rsi"] = out["rsi"].fillna(50) # 早期无足够数据时给中性值 return out逻辑说明：EMA 用adjust=False表示从第一个有效值开始递归计算，这和大多数行情软件的结果一致。MACD 的三个参数 fast、slow、signal 沿用传统值 12/26/9，如果你想做参数敏感性测试，直接传不同值进来，不需要改动函数内部。RSI 里loss.replace(0, pd.NA)是为了处理连续上涨导致平均下跌为 0 的情况——此时 RSI 在数学上是 100，但直接除零会产生 inf，后面画图时会把整条曲线拉坏。
参数说明：period和span指的是 K 线根数而非自然日。如果你用的是周线数据，14 根周线的 RSI 和 14 根日线的 RSI 含义完全不同，策略参数不能跨周期照搬。另外，所有指标函数都返回out.copy()后的新对象，避免多个策略共享同一个 DataFrame 时互相污染列名——这是很多人在多策略并跑时踩过的坑，一个策略往 DataFrame 里加了一列，另一个策略拿到的 DataFrame 里多了一根无用列，导致最终信号完全错位。
3.3 信号生成与导出：输出信号字典而不是直接下单
指标算完之后，stockAnalyse 面临一个选择：直接生成买卖单，还是先生成信号标记？我强烈建议先生成信号标记（signal 列），再在回测引擎里根据信号和持仓状态生成交易记录。因为直接输出买卖单，会过早地把仓位管理逻辑耦合进信号层，后面想调止损、调仓位都得改信号函数，得不偿失。
def generate_signals(df: pd.DataFrame, buy_threshold: float = 20, sell_threshold: float = 80) -> pd.DataFrame: """基于 RSI 的简单信号：超卖买入、超买卖出""" out = df.copy() if "rsi" not in out.columns: out = add_rsi(out) out["signal"] = 0 out.loc[out["rsi"] < buy_threshold, "signal"] = 1 # 买入信号 out.loc[out["rsi"] > sell_threshold, "signal"] = -1 # 卖出信号 return out def to_signal_events(sig_df: pd.DataFrame) -> list[dict]: """把 signal 列压缩成事件序列，剔除连续重复信号""" events = [] prev = 0 for row in sig_df.itertuples(): if row.signal != 0 and row.signal != prev: events.append({"date": row.date, "action": "buy" if row.signal == 1 else "sell", "price": row.close, "rsi": row.rsi}) prev = row.signal return events逻辑说明：to_signal_events 做了一件容易被忽略的事——把连续多个相同的买入信号压缩成一个。如果不做压缩，RSI 连续多日低于 20 时，策略会在每一天都记一次买入信号，回测时会重复买入，资金直接买光。压缩后每个方向只有一个事件，后续回测引擎再根据自己的持仓状态决定是否真正执行。
参数说明：buy_threshold 和 sell_threshold 是策略级参数，这里给了 20/80 的经典值。注意 RSI 的阈值不对称性：很多实证分析里，RSI 低于 30 才适合买，高于 70 就卖，因为 A 股散户占比高，超卖持续时间长，20/80 会更保守。如果你换到指数或 ETF 上，阈值需要重新校准，不能沿用。信号事件里同时记录当时的 close 价格，目的是让回测阶段可以对信号价格和实际成交价格做比较，判断滑点影响有多大。
4. 把 shares 和 stockAnalyse 串起来：一个可复现的回测流程
4.1 主流程：数据 → 信号 → 回测执行
数据层和分析层都有了，下一步是把它们串成一个完整的回测流程。这个环节最容易出现的问题是“模块之间数据格式不一致”，比如 shares 返回的 date 是字符串，stockAnalyse 生成事件时用的是 Timestamp，回测引擎再按 Timestamp 排序时字符串和 Timestamp 混在一起直接报错。所以我建议在回测引擎入口统一做一次类型转换，之后所有模块都按转换后的格式处理。
# backtest/engine.py import pandas as pd from shares.fetcher import update_cache from stockAnalysis.signals import generate_signals, to_signal_events def run_backtest(symbol: str, initial_cash: float = 100_000, fee_rate: float = 0.0003, slippage: float = 0.001) -> dict: df = update_cache(symbol) df = adjust_price(df, mode="forward") df = generate_signals(df) events = to_signal_events(df) cash = initial_cash position = 0 # 0 空仓，1 满仓 trade_log = [] for ev in events: px = ev["price"] * (1 + slippage if ev["action"] == "buy" else 1 - slippage) if ev["action"] == "buy" and position == 0: shares = int(cash * (1 - fee_rate) // px) cost = shares * px * (1 + fee_rate) cash -= cost position = shares trade_log.append({"date": ev["date"], "action": "buy", "price": px, "shares": shares}) elif ev["action"] == "sell" and position > 0: income = position * px * (1 - fee_rate) cash += income trade_log.append({"date": ev["date"], "action": "sell", "price": px, "shares": position}) position = 0 final_value = cash + position * df["close"].iloc[-1] return {"symbol": symbol, "final_value": final_value, "return": final_value / initial_cash - 1, "trade_count": len(trade_log), "log": trade_log}逻辑说明：回测引擎的核心逻辑是“只有信号还不够，还要判断当前持仓状态”。买入时计算能买多少手/股：先用现金扣除手续费，再按成交价（加上滑点）取整；卖出时把持仓全部卖出，扣除手续费。这里没有实现分批建仓或止损，因为单次全进全出最容易验证策略本身的 alpha。
参数说明：fee_rate 默认 0.0003（万三）是常见的佣金水平，但实际账户还有最低五元佣金、印花税、过户费等。slippage 是滑点率，这里设置 0.1% 用于模拟“信号价格到实际成交价格之间的偏差”。这两个参数直接决定回测结果的真实性——把 fee_rate 设成 0，回测收益可能虚高 20% 以上，这是很多“回测很赚钱实盘就亏损”的主要原因之一。
4.2 参数配置和结果评估：不只看总收益，还要看交易次数
回测结果不能只看最终收益率。我习惯把结果拆成四个维度：总收益、交易次数、最大回撤、胜率。交易次数尤其重要——如果一只股票跑一年只交易了两次，收益再高也不具备统计意义；如果交易了 50 次但胜率只有 30%，说明信号阈值过敏感。stockAnalyse 可以加一个简单的回撤计算函数：
def calc_max_drawdown(equity_curve: pd.Series) -> float: """计算最大回撤，用于评估策略风险""" peak = equity_curve.cummax() dd = (equity_curve - peak) / peak return dd.min() def evaluate(log: list[dict], closes: pd.Series) -> dict: """基于交易日志生成评估指标""" if not log: return {"trade_count": 0, "win_rate": 0.0, "max_drawdown": 0.0} # 净收益序列从每笔交易的价格变化累积 returns = [] buy_price = None for trade in log: if trade["action"] == "buy": buy_price = trade["price"] else: ret = trade["price"] / buy_price - 1 returns.append(ret) buy_price = None wins = [r for r in returns if r > 0] return {"trade_count": len(log), "win_rate": len(wins) / len(returns) if returns else 0.0}逻辑说明：calc_max_drawdown 用累计峰值的回撤来度量策略在最差时点的亏损幅度。evaluate 里的胜率按“每一轮买-卖周期”计算，而不是按单笔信号计算，这样更接近实际账户的盈亏体验。如果你发现胜率不错但总收益还是负的，问题大概率出在亏损单的平均跌幅大于盈利单的平均涨幅，这时要调整卖出阈值，而不是继续调买入。
参数说明：evaluate 只拿了交易日志和收盘价序列，没拿完整的资金曲线。要算最大回撤，最好在 run_backtest 里记录每个交易日的账户总资产（现金 + 持仓市值），再传给 calc_max_drawdown。我为了方便演示省略了这部分，实际项目里至少每根 K 线记录一次账户净值，否则回撤会被低估很多。
4.3 常见误用：把分析模块和数据层耦合在一起会怎样
把 shares 和 stockAnalyse 拆开，最大的收益不是代码复用，而是“排查问题时有明确的边界”。如果耦合在一起，当出现“信号看起来不对”的时候，你分不清楚是数据错了还是指标错了。举个例子：某开发者把拉数据的代码直接写在指标计算函数里，每次调用指标都重新拉一次全量行情，结果接口被限流，数据返回不完整，MACD 柱状图直接断了一截。他调了三天指标参数，最后才发现是数据源的问题——这就是耦合的代价。
保持独立的另一个好处是：你可以分别给两侧写单元测试。shares 的测试可以用一段人工构造的行情数据验证清洗逻辑；stockAnalyse 的测试可以用固定输入验证指标公式是否算得对。两边的测试数据可以完全分离，互不依赖。我在实际项目中，会把 shares 的测试数据固定成 20 行手工计算过的序列，stockAnalyse 的测试输入直接引用这个序列的close列，任何一边改动导致结果变化，测试立刻报红。
5. 避坑：shares 和 stockAnalyse 联调时最容易翻车的 5 个点
5.1 除权跳空被当成突破信号
现象：一只股票在除权日“价格跌了 5%”，突破策略当天触发买入信号，买入后被套。
原因：shares 返回的是未复权价格，除权日开盘价相对于前收盘价有一个天然的向下跳空缺口，这是分红送股造成的，不是市场价格下跌。stockAnalyse 在未复权数据上计算突破指标，把这个缺口误判为价格突破。
解决：在 generate_signals 之前强制调用 adjust_price 做前复权，并检查复权后数据在除权日是否还有跳空。如果仍然存在超过 5% 的跳空，说明行情源给的复权因子不完整，需要到数据层排查。
5.2 停牌日和零成交数据混入回测
现象：回测结果里出现“跌停买入”的神操作，或者某天价格没动但策略交易了，成交记录里的价格明显不合理。
原因：shares 清洗阶段保留了停牌日（volume 为 0），stockAnalyse 没检查 suspend 列，指标计算把停牌日当成正常交易日参与 rolling 窗口计算，导致均线和波动率指标失真。
解决：在 generate_signals 开头加一行df = df[df["suspend"] == False]。但注意，如果策略要处理停牌期间的信息（比如复牌后的跳空风险），不要删除停牌行，而是把停牌日单独标记并在回测引擎里禁止交易。
5.3 缓存数据过期导致分析结论失真
现象：昨天回测结果很漂亮，今天重跑一次，结果完全变了，而且数据只到上周五。
原因：update_cache 的增量拉取逻辑只在本地文件存在且非空时生效，如果上次拉取失败写入了一个空 DataFrame，下次读取时local非空但里面没有数据，last_date取到 NaT，增量起点计算崩溃或者退回全量拉取。
解决：在 update_cache 里检查local.empty和last_date是否为 NaT。我的习惯是：本地文件存在但行数少于 100 时直接删除文件重新全量拉取，宁多拉一次，不能拿脏数据跑分析。
5.4 指标计算用了未来数据
现象：回测收益极高，曲线光滑一路向上，实盘却完全跑不出这个收益。
原因：stockAnalyse 在计算指标时误用了shift(-1)或者把所有数据一次性传入指标函数，导致某些指标在 T 日的值里包含了 T+1 日的信息。最典型的错误是：计算某日均线时没排除当天收盘后的数据，而回测引擎却在当天开盘就用这个值做决策。
解决：信号生成时保证每个信号点的指标值只由“到今天为止”的数据计算出来。可以在指标函数里增加一个lookahead_check：计算完成后，对比最后一行指标值与倒数第二行同指标值，如果差异超过阈值说明可能在边缘位置用了不该用的数据。更稳妥的办法是把回测的决策点放在 T 日收盘后，用 T 日及之前的数据计算信号，T+1 日开盘再执行。
5.5 涨跌停价格限制导致回测滑点失真
现象：回测里一只股票连续涨停时策略买入成功，但实际交易中涨停板上根本买不进；或者连续跌停时卖出单完全成交不了，回测却以跌停价成交了。
原因：回测引擎没有检查价格是否涨跌停。涨停时最高价就是涨停价，买单无法成交；跌停时最低价就是跌停价，卖单无法成交。stockAnalyse 生成的信号只是“想买”，回测引擎需要判断“能不能买”。
解决：在回测引擎里增加涨跌停过滤——如果买入信号当天的close等于当日high且涨幅超过 9.8%（A 股主板），跳过该信号；卖出信号当天close等于当日low且跌幅超过 9.8%，跳过。这个近似策略不能代替真正的盘口判断，但至少避免回测结果里出现大量虚假成交。
6. 最后再进一步：让 stockAnalyse 批量跑几十只股票
6.1 批量调度与结果汇总
单只股票跑通之后，你会很快遇到新问题：想同时验证 30 只股票，怎么办？shares 的缓存机制在这里发挥作用——只有第一次需要全量拉取，之后增量更新非常快。批量跑的时候，建议用简单的顺序循环加进度打印，不要一上来就开多线程。因为公开行情接口普遍有限流，并发一高就会被限频，出现大量超时重试，反而比串行更慢。
# batch_runner.py from backtest.engine import run_backtest from shares.fetcher import update_cache symbols = ["SH.600000", "SH.600519", "SZ.000001", "SZ.300750"] results = [] for sym in symbols: try: r = run_backtest(sym, initial_cash=100_000) results.append({"symbol": sym, **{k: v for k, v in r.items() if k != "log"}}) print(f"{sym}: return={r['return']:.2%}, trades={r['trade_count']}") except Exception as exc: print(f"{sym}: failed -> {exc}") df_out = pd.DataFrame(results) df_out.to_parquet("./results/batch_results.parquet")逻辑说明：每个 symbol 独立跑完整流程，结果汇总成 DataFrame 落盘。用 try/except 包住每一个 symbol 的原因是：一只股票的数据异常不应该影响整个批次的验证。比如某只股票退市整理期数据缺失，回测引擎直接抛异常，你要能看到它失败了，而不是让整个脚本崩溃。
参数说明：symbols 列表里既有大盘股也有创业板，它们的涨跌停幅度不同（主板 10%，创业板 20%）。如果你的回测引擎按 9.8% 判断涨停，创业板股票的涨停信号就会被漏掉。批量跑之前，先确认回测引擎能识别代码前缀，按不同板块应用不同的涨跌停阈值。
6.2 验证方法：用已知样本做回归测试
批量跑的结果不要直接信。我习惯做一个回归测试：手工构造一段 30 天的数据，其中包含一次明显的超卖（RSI 低于 20）和一次超买（RSI 高于 80），然后验证 stockAnalyse 生成的信号日期是否和预期吻合。这个方法成本很低，但能捕获 80% 的指标计算错误。另一件事是：把一个策略在 30 只股票上的平均收益和单只股票的收益对比，如果平均收益远低于单只，说明策略可能只是在个别股票上表现好，不具备普适性——这时要回头调整信号阈值，而不是增加股票数量。
我这几年跑下来最大的教训是：数据层和分析层任何一边改了一个小参数，另一边往往要跟着改，但代码不会提示你。所以我现在每个参数都会单独复制一份配置出来，跑完对比，再合入主流程。shares 和 stockAnalyse 这个组合本身不难，难的是每一条数据都经得起检查。希望这篇笔记能帮你在搭自己的验证流程时少走几步弯路。
本文还有配套的精品资源，点击获取
