当前位置：首页 > news >正文

【Python量化实战 #13】想有自己的选股器？从零搭一套 Python 多条件筛选+回测系统

news 2026/8/3 16:38:43

连载系列「Python量化实战」第 13 篇（模块六 · 量化策略实战）。上一篇《Python多因子选股模型搭建》用打分排序解决"谁更好"，本篇换一种更直观的思路——多条件筛选，解决"谁合格"，并把入选组合放回历史里验证效果。

搭建一个 Python 量化选股系统，核心就四步：批量拉取股票池数据 → 按多个条件逐层筛选 → 对入选组合做历史回测 → 结果导出复盘。多数教程只讲到"筛出名单"就结束了，但名单不回测等于没有闭环——你不知道这套条件历史上到底管不管用。本文用 mairui SDK 把四步完整走通：条件写成可插拔的规则列表，筛选过程打印"漏斗"，入选组合与全池基准对比净值，全部代码可直接运行。

本文你将得到什么

一套可插拔的选股系统：规则列表式多条件筛选（漏斗可视化）+ 入选组合 vs 全池基准回测 + CSV 导出复盘
真实运行结果（20 只池筛出 3 只，入选组合 +21.65% vs 全池 -4.79%）及"回测意义是证伪"的讨论
缺失值、前复权、幸存者偏差等 5 个避坑点，及"改配置不改代码"的进阶方向

一、在线体验

想先看看筛选用的行情快照接口返回什么字段？打开 API Playground 在线调用即可：
https://mairuiapi.com/playground

建议先在 Playground 里测试"实时交易数据"接口，能直接看到pe（市盈率）、sjl（市净率）、zdf60（60日涨跌幅）等字段——这些就是下文筛选条件的数据来源。

二、环境准备

本文代码使用 mairui SDK 获取股票数据，安装方法如下：

pipinstallmairui

SDK 的完整接口文档与使用说明请查阅 GitHub 仓库：https://github.com/MaiRuiApi/mairui

接口的详细参数说明请查阅官网 API 文档：https://mairuiapi.com/hsdata

运行环境：

Python 3.9+
mairui SDK 1.0.0
pandas 2.2+ / numpy

importosimportmairui# 证书从环境变量读取，注册后可在官网获取api=mairui.Client(os.environ["MAIRUI_LICENCE"])

本文数据截至 2026-07-28（行情为实时快照，回测区间 2025-07-28 至 2026-07-27），读者复现时数值会随行情更新而变化。

三、第一步：批量拉取股票池数据

选股系统的地基是一张"股票池指标快照表"。本文用 20 只跨行业沪深大盘股做演示池（固定池便于复现），实际使用时换成指数成分股或全市场列表即可，代码结构不变。

每只股票取 5 个筛选用指标，分别来自两个接口：stock_ssjy（实时快照）提供现价、PE、PB、60日涨跌幅；stock_pershareindex（历季财务指标）提供 ROE。

importtimeimportnumpyasnpimportpandasaspd# 示例股票池：沪深大盘股样本（跨行业；实际可换成指数成分股/全市场列表）STOCK_POOL={"600519":"贵州茅台","000858":"五粮液","601318":"中国平安","600036":"招商银行","601166":"兴业银行","000333":"美的集团","600900":"长江电力","601088":"中国神华","600030":"中信证券","000651":"格力电器","600276":"恒瑞医药","300750":"宁德时代","601899":"紫金矿业","600887":"伊利股份","601668":"中国建筑","002594":"比亚迪","601012":"隆基绿能","600809":"山西汾酒","000002":"万科A","601398":"工商银行",}defto_num(value):"""接口缺失值以 '-' 返回，统一转成 NaN。"""try:returnfloat(value)except(TypeError,ValueError):returnnp.nandeffetch_stock_snapshot(api,code,name):"""取单只股票的筛选用指标。"""snap=api.stock_ssjy(code)# 实时交易快照snap=snap[0]ifisinstance(snap,list)elsesnap fin_rows=api.stock_pershareindex(code)# 历季财务指标，[-1] 为最新一期fin=fin_rows[-1]iffin_rowselse{}return{"code":code,"name":name,"price":to_num(snap.get("p")),"pe":to_num(snap.get("pe")),# 市盈率（动）"pb":to_num(snap.get("sjl")),# 市净率"momentum_60d":to_num(snap.get("zdf60")),# 60日涨跌幅(%)"roe":to_num(fin.get("jzcsyl")),# 净资产收益率(%)}defbuild_universe(api):records=[]forcode,nameinSTOCK_POOL.items():records.append(fetch_stock_snapshot(api,code,name))time.sleep(0.2)# 控制请求频率returnpd.DataFrame(records)

关键解释：财务接口的缺失值以字符串"-"返回，to_num统一转成NaN，后续筛选时NaN会被当作"不通过"处理，避免脏数据混进名单。

四、第二步：多条件筛选组合器

选股系统最怕把条件写死在一堆if里——每次调整策略都要改逻辑代码。更好的做法是把条件抽象成(名称, 判断函数)的规则列表，增删条件只改配置：

# 条件写成 (名称, 判断函数) 列表，增删条件只需改这里SCREEN_RULES=[("PE介于0~30",lambdadf:(df["pe"]>0)&(df["pe"]<30)),("PB低于5",lambdadf:df["pb"]<5),("ROE不低于10%",lambdadf:df["roe"]>=10),]defrun_screener(universe_df,rules):"""逐条应用筛选规则，打印每一步剩余数量（筛选漏斗）。"""mask=pd.Series(True,index=universe_df.index)print(f"初始股票池:{len(universe_df)}只")forrule_name,rule_fninrules:mask&=rule_fn(universe_df).fillna(False)print(f"通过[{rule_name}]后剩余:{int(mask.sum())}只")returnuniverse_df[mask].reset_index(drop=True)

真实运行的筛选漏斗输出如下（2026-07-28 实时数据）：

=== 筛选漏斗 === 初始股票池: 20 只 通过[PE介于0~30]后剩余: 16 只 通过[PB低于5]后剩余: 15 只 通过[ROE不低于10%]后剩余: 3 只 === 入选名单 === code name price pe pb roe momentum_60d 300750 宁德时代 398.37 21.29 4.86 12.08 -10.48 601899 紫金矿业 31.47 10.42 4.42 10.40 -6.78 600809 山西汾酒 122.80 6.96 4.05 12.72 -14.21

关键解释：漏斗式打印能直观看到每个条件"杀掉"了多少股票——如果某个条件一下子筛掉 90%，通常说明阈值过严或数据口径有问题，这是调试选股条件最实用的手段。PE > 0同时排除了亏损股（亏损股 PE 为负，不具可比性）。

五、第三步：入选组合历史回测

名单筛出来只是一半，另一半是回答"这套条件历史上表现如何"。做法：拉取全池近一年前复权日 K 线，入选组合等权持有，与"全池等权"基准对比净值。

deffetch_close_series(api,code,start,end):"""拉前复权日K，返回以日期为索引的收盘价序列。"""rows=api.stock_history(code,"d","f",st=start,et=end)# d=日线 f=前复权kline_df=pd.DataFrame(rows)kline_df["t"]=pd.to_datetime(kline_df["t"])returnkline_df.set_index("t")["c"].astype(float).sort_index()defbacktest_portfolio(api,picked,benchmark_pool,start,end):"""等权持有入选组合，与全池等权基准对比（教学用，不含手续费/调仓）。"""closes={}forcodeinbenchmark_pool:closes[code]=fetch_close_series(api,code,start,end)time.sleep(0.2)close_df=pd.DataFrame(closes).dropna()ret_df=close_df.pct_change().fillna(0)# 日收益率picked_codes=[cforcinpicked["code"]ifcinclose_df.columns]nav=pd.DataFrame({"入选组合":(1+ret_df[picked_codes].mean(axis=1)).cumprod(),"全池基准":(1+ret_df.mean(axis=1)).cumprod(),})returnnav

真实回测输出（回测区间 2025-07-28 至 2026-07-27，共 242 个交易日）：

=== 回测结果（2025-07-28 ~ 2026-07-27, 共242个交易日）=== 入选组合累计收益: +21.65% | 全池等权基准: -4.79%

关键解释：stock_history的st/et参数只认YYYYMMDD格式；第三个参数"f"表示前复权，回测必须用复权价，否则除权缺口会被当成暴跌。这次筛选条件在过去一年跑赢了全池基准约 26 个百分点，但务必清醒：单次、单区间、小样本池的回测结果不代表条件本身长期有效，换一个区间结论完全可能反转——回测的意义是证伪明显糟糕的条件，而不是证明未来能赚钱。

六、完整实战示例

把四步串成完整可运行脚本（含结果导出）：

importosimportmairui api=mairui.Client(os.environ["MAIRUI_LICENCE"])# 1) 批量取数universe_df=build_universe(api)# 2) 多条件筛选picked_df=run_screener(universe_df,SCREEN_RULES)print(picked_df[["code","name","price","pe","pb","roe"]].round(2))# 3) 历史回测：入选组合 vs 全池等权nav_df=backtest_portfolio(api,picked_df,STOCK_POOL,"20250728","20260727")total=(nav_df.iloc[-1]-1)*100print(f"入选组合:{total['入选组合']:+.2f}% | 全池基准:{total['全池基准']:+.2f}%")# 4) 结果导出，便于 Excel 复盘universe_df.to_csv("universe_snapshot.csv",index=False,encoding="utf-8-sig")picked_df.to_csv("picked_stocks.csv",index=False,encoding="utf-8-sig")nav_df.to_csv("backtest_nav.csv",encoding="utf-8-sig")

导出三张表：universe_snapshot.csv（全池指标快照）、picked_stocks.csv（入选名单）、backtest_nav.csv（净值曲线，可直接在 Excel 里画图）。CSV 用utf-8-sig编码，Excel 打开不乱码。

七、避坑与进阶

缺失值必须显式处理：筛选条件里NaN参与比较结果是False，看似"自动排除"，但mask &= rule_fn(df)时若不fillna(False)，NaN会在取反、组合等场景产生意外结果，务必显式处理。
回测必须用前复权价：stock_history第三个参数传"f"。用不复权价回测，分红除权日会被算成大跌，结论完全失真。
警惕幸存者偏差与过拟合：固定池演示没有包含已退市股票；条件调到"历史收益最好看"就是过拟合。实用做法是把条件的经济含义讲清楚（低估值+高质量），而非无限调参。
频率控制：批量拉取时加time.sleep(0.2)，20 只以内很快；全市场级别的批量拉取与并发方案，本系列工程化篇（#16）会专门展开。
进阶方向：把SCREEN_RULES改成从 JSON/YAML 配置文件加载，即可做成"改配置不改代码"的选股服务；再配合定时任务每日收盘后自动跑一遍并推送名单。

八、总结与延伸

本文从零搭建了一个可扩展的 Python 量化选股系统：stock_ssjy+stock_pershareindex批量取数、规则列表式多条件筛选（漏斗可视化）、入选组合 vs 全池基准的历史回测、CSV 导出复盘。系统的价值不在"筛出了哪 3 只"，而在这套可插拔、可回测、可复盘的结构——条件随你换，验证框架不变。

下一篇预告：【Python量化实战 #14】将接入专业回测框架 backtrader，跑通第一个带手续费、夏普比率与最大回撤分析的策略回测。

延伸阅读：