简介这是一套面向A股个人投资者与量化初学者的Python自动选股工具包内置10种常见技术选股策略覆盖250日均线突破、平台突破、回踩均线、停机坪形态、低ATR波动筛选及持续上涨趋势识别等场景支持灵活组合与快速替换。整个压缩包共27个文件以Python脚本为主17个py脚本另含策略示意图png/jpg、依赖清单、配置文件与使用说明解压后约1.96MB模块划分清晰便于直接运行和替换策略。目前已有110人学习下载适合快速验证选股思路并作为二次开发基础。工具包附带数据获取、本地SQLite存储、策略回测统计、执行流程控制及买卖信号触发模块可独立完成从行情获取到信号输出的完整链路同时配有说明文档和策略逻辑图能帮助用户快速上手实践。1. 把选股从“盘中盯盘”改成“收盘后跑脚本”这个工具包解决的不只是时间先说一个可能反直觉的结论每天都盯盘四个小时的人往往不如每天收盘后用十分钟跑一次量化选股脚本的人更清楚第二天该看什么。原因很简单——盘中盯盘看到的是价格跳动和情绪噪音而量化选股工具包解决的是「把选股标准固化成代码」这件事。这个包含 10 种 Python 选股策略的工具包底层用 TuShare 拉取 A 股行情与财务数据通过统一框架批量扫描全市场最后输出一份带因子分数的候选股票 CSV。它不预测涨跌只帮你把「什么股票值得关注」这件事从拍脑袋变成可复用、可回测、可复盘的程序逻辑。适合两类人一类是会用 Python 但还没系统做过量化选股的投资者另一类是已经有主观交易体系、想把自己的选股规则固化下来的从业者。2. 工具包骨架与前置准备TuShare 接口、目录结构和策略基类设计2.1 调通 TuShare 接口token、pro 接口与积分限制所有策略的数据源都建立在 TuShare 之上。使用前需要先注册一个账号在个人主页拿到 token。这个 token 就是你调用数据接口的钥匙配置方式有两种直接写在代码里或者通过环境变量传入。我一般用后者因为脚本会分享给同事或换机器跑硬编码 token 迟早会泄露。import os import tushare as ts # 从环境变量读取 token避免硬编码 token os.environ.get(TUSHARE_TOKEN, 你的token放在这里) ts.set_token(token) # 初始化 pro 接口后续所有数据调用都走 pro pro ts.pro_api() # 验证接口是否通拉取最近一个交易日 df pro.trade_cal(exchangeSSE, start_date20240101, end_date20241231, is_open1) print(df.tail(3))这段代码先设置 token然后调用trade_cal获取交易日历。注意exchange参数传SSE即可因为沪深两市日历基本一致没必要分别拉取。is_open1表示只要开盘日这在你后面判断「今天该不该跑策略」时非常有用。TuShare 的积分是很多人第一次跑脚本时翻车的根源。新注册账号默认积分比较低能调用的接口和数据频率都受限。比如daily接口完整调用需要更高的积分daily_basic里的市净率、市销率字段同样有权限门槛。我遇到的情况是当天注册后拉取单只股票日线没问题但全市场批量取数时直接被限流。解决办法是在 TuShare 社区完善资料、通过实名认证提升积分或者在代码里对每次请求做限速主动 sleep减少被限流的概率。2.2 读懂包里的目录策略基类与 Runner 的分工工具包不是十个脚本堆在一起而是有一个清晰的骨架。每个策略文件都继承同一个基类基类定义好「要做什么」具体策略只负责「怎么做」。这样做的直接收益是你想加第 11 个策略只需要新建一个文件并实现两个方法Runner 会自动加载它。# base_strategy.py from abc import ABC, abstractmethod from dataclasses import dataclass, field import pandas as pd dataclass class StockCandidate: ts_code: str # 股票代码如 600000.SH name: str # 股票名称 score: float # 策略打分越高越靠前 factors: dict field(default_factorydict) # 因子明细方便复盘 class BaseStrategy(ABC): strategy_name base def __init__(self, params: dict None): self.params params or {} self.result [] abstractmethod def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: 输入全市场股票池输出过滤后的股票列表 abstractmethod def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: 对过滤后的股票打分排序返回带 score 列的数据框 def run(self, stock_pool: pd.DataFrame) - pd.DataFrame: filtered self.filter_stocks(stock_pool) scored self.score_stocks(filtered) scored scored.sort_values(score, ascendingFalse) return scored基类里我用dataclass定义了候选股票的实体结构StockCandidate里除了代码和名称还有个factors字典专门用来保存每个策略算出来的因子明细。这样输出的 CSV 不只是「推荐了哪只股票」还能看到「为什么推荐它」。比如说某只股票是因为低市净率被选中的还是因为动量分值高被选中的复盘时一眼就能看清楚。run()方法是模板方法先filter_stocks做粗筛再score_stocks做排序。之所以把过滤和打分拆成两步是因为 A 股全市场五千多只股票很多策略第一步可以用市值、ST 状态、停牌状态快速过滤掉大量明显不合规的标的减少后续计算量。Runner 的角色是调度者。它读取某个策略类传入当天的股票池然后触发run()。策略之间互相独立即使某个策略因为数据缺失挂了不影响其他策略继续跑。# runner.py import importlib import pandas as pd from base_strategy import BaseStrategy def load_strategy(strategy_name: str, params: dict) - BaseStrategy: module importlib.import_module(fstrategies.{strategy_name}) clazz getattr(module, strategy_name.split(_)[0].capitalize() Strategy) return clazz(params) def run_single_strategy(strategy_name: str, stock_pool: pd.DataFrame, params: dict): strategy load_strategy(strategy_name, params) result strategy.run(stock_pool) return result这里加载策略用的是importlib动态导入模块名和策略名一一对应。比如strategies.small_cap对应SmallCapStrategy。注意命名要规范否则getattr找不到类名时会直接抛异常。我会在 Runner 里加一层异常捕获单独记录哪个策略失败了而不是让整个脚本崩掉——全市场扫描跑二十分钟因为最后一个策略的字段缺失全部重跑这体验太糟糕了。3. 十种策略拆解与参数改动多因子、均线、动量、小市值的取舍3.1 多因子打分PE、PB、ROE 合成截面分数多因子策略是这个工具包里的「万金油」它的思路不是找一个独门指标而是把几个有效因子标准化后加权合成一个综合分数。比如市盈率、市净率越低越好ROE 越高越好那么先对每个因子做截面标准化再按权重相加。# strategies/multi_factor.py import pandas as pd import numpy as np from base_strategy import BaseStrategy class MultiFactorStrategy(BaseStrategy): strategy_name multi_factor def __init__(self, params: dict None): super().__init__(params) self.weight_pe params.get(weight_pe, -0.3) # 负权重PE 越低分越高 self.weight_pb params.get(weight_pb, -0.2) self.weight_roe params.get(weight_roe, 0.5) def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: # 去掉 ST、停牌、上市不满一年的股票 df stock_pool.copy() df df[~df[name].str.contains(ST, naFalse)] df df[df[list_date] 20250101] return df def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: # 拉取每日基本面指标 import tushare as ts pro ts.pro_api() all_daily_basic [] for code in filtered[ts_code].tolist(): db pro.daily_basic(ts_codecode, fieldsts_code,trade_date,pe,pb,roe) if db is not None and len(db): last db.iloc[0] # 默认按日期倒序取最新 all_daily_basic.append(last) basic_df pd.DataFrame(all_daily_basic) merged filtered.merge(basic_df, onts_code, howleft) # 对每个因子做 zscore 标准化 merged[pe_z] (merged[pe] - merged[pe].mean()) / merged[pe].std() merged[pb_z] (merged[pb] - merged[pb].mean()) / merged[pb].std() merged[roe_z] (merged[roe] - merged[roe].mean()) / merged[roe].std() # 合成综合分数 merged[score] (self.weight_pe * merged[pe_z] self.weight_pb * merged[pb_z] self.weight_roe * merged[roe_z]) return merged[[ts_code, name, score, pe, pb, roe]]这个策略跑起来最大的瓶颈在于for循环调用daily_basic。虽然代码直观但全市场逐只拉取要请求几千次速度很慢。更高效的做法是调用 TuShare 的daily_basic(trade_date某一天)一次性获取全市场当天的基本面数据但这对积分的门槛更高。工具包默认提供的是逐只拉取的兼容写法实际使用中我通常会改成按日期批量拉取的模式。weight_pe、weight_pb、weight_roe是三个核心参数。负权重表示指标值越低贡献越高ROE 是正向的。这套默认权重来自经典的价值选股思路但你可以按自己的偏好改比如更偏好成长的话可以把weight_roe调到 0.7。改参数时注意每个权重的量级因为 zscore 标准化后的数值已经消除了量纲差异权重可以直接比较大小。3.2 均线多头排列与动量过滤趋势型策略的信号写法趋势型策略的核心假设是「强者恒强」。均线多头排列是最经典的形态短期均线在中期均线之上中期均线在长期均线之上。但只看排列还不够我会额外加一个动量过滤条件要求近 20 日涨幅大于一定阈值这样能过滤掉那些刚刚从下跌转为粘合的模糊信号。# strategies/ma_trend.py import pandas as pd import tushare as ts from base_strategy import BaseStrategy class MaTrendStrategy(BaseStrategy): strategy_name ma_trend def __init__(self, params: dict None): super().__init__(params) self.short_win params.get(short_win, 5) self.mid_win params.get(mid_win, 10) self.long_win params.get(long_win, 20) self.momentum_threshold params.get(momentum_threshold, 5.0) def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: # 做趋势策略尽量选流动性好的过滤掉日均成交额太低的 df stock_pool.copy() return df def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: pro ts.pro_api() result_rows [] for code in filtered[ts_code].tolist(): df pro.daily(ts_codecode, start_date20250115, end_date20250415) if df is None or len(df) 40: continue df df.sort_values(trade_date) df[fma_{self.short_win}] df[close].rolling(self.short_win).mean() df[fma_{self.mid_win}] df[close].rolling(self.mid_win).mean() df[fma_{self.long_win}] df[close].rolling(self.long_win).mean() latest df.iloc[-1] ma_short, ma_mid, ma_long latest[fma_{self.short_win}], latest[fma_{self.mid_win}], latest[fma_{self.long_win}] if not (ma_short ma_mid ma_long): continue # 动量近 20 日累计涨跌幅 close_20d_ago df[close].iloc[-min(20, len(df))] momentum (latest[close] / close_20d_ago - 1) * 100 if momentum self.momentum_threshold: continue row filtered[filtered[ts_code] code].iloc[0] result_rows.append({ ts_code: code, name: row[name], score: momentum, # 动量越高越靠前 factors: {momentum_20d: round(momentum, 2), ma_long: round(ma_long, 2)} }) return pd.DataFrame(result_rows)这段代码有几个细节需要留意。一是sorted_values(trade_date)必须做升序排列因为pro.daily返回的数据默认按日期倒序如果不排序rolling算出来的均线就是「未来数据」会导致买到未来的信号——这在回测里叫未来函数是最严重的逻辑漏洞。二是动量的计算我取了倒数第 20 个交易日的数据并用min(20, len(df))保护避免新股数据不足时越界。momentum_threshold参数建议设成 310 之间。设得太低会混入横盘整理的股票设得太高容易追在情绪高点。我一般分市场环境调整强势市场用 8震荡市用 3。3.3 小市值策略我为什么保留它但加了价格下限小市值策略在过去很长一段时间里是 A 股超额收益最稳定的来源之一但它也是回撤最猛烈的策略。每天收盘后按总市值排序买入市值最小的几十只这个逻辑简单到不需要任何财务分析但工具包还是保留了它因为它是验证「策略逻辑—代码实现—数据链路」整条流程是否通畅的最佳样例。# strategies/small_cap.py import pandas as pd import tushare as ts from base_strategy import BaseStrategy class SmallCapStrategy(BaseStrategy): strategy_name small_cap def __init__(self, params: dict None): super().__init__(params) self.min_price params.get(min_price, 2.0) # 最低股价过滤低价股 self.max_price params.get(max_price, 20.0) # 最高股价过滤高价股 self.max_mv params.get(max_mv, 50) # 最大流通市值亿元 self.top_n params.get(top_n, 30) # 最终输出前 30 只 def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: df stock_pool.copy() df df[~df[name].str.contains(ST, naFalse)] df df[df[trade_status] 交易中] return df def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: pro ts.pro_api() # 拉取全市场最新市值与股价 daily_basic pro.daily_basic(trade_date20250415, fieldsts_code,close,total_mv,circ_mv) merged filtered.merge(daily_basic, onts_code, howinner) # 股价过滤低价股流动性差且容易受借壳炒作影响 merged merged[(merged[close] self.min_price) (merged[close] self.max_price)] # 流通市值过滤单位是万元转成亿元 merged[circ_mv_yi] merged[circ_mv] / 10000 merged merged[merged[circ_mv_yi] self.max_mv] # 市值越小分数越高 merged[score] -merged[circ_mv_yi] merged merged.sort_values(score).head(self.top_n) return merged[[ts_code, name, score, close, circ_mv_yi]]我特意加了min_price和max_price两个参数这是从小市值策略翻车里总结出来的。纯粹按市值排序会把大量股价两块钱以下的股票选进来这些股票看着便宜但多数是基本面出了问题或者流通盘被操控的标的普通投资者根本拿不住。加上价格区间过滤后候选列表的质地会明显改善。circ_mv字段单位是万元所以除以 10000 换算成亿元才便于人读。这里有个隐蔽的坑TuShare 的total_mv和circ_mv字段返回单位是万元但不同接口文档表述可能不同有些第三方封装库会直接换算成亿元。如果你在运行工具包时发现市值数值大得离谱先确认是不是做了单位换算。3.4 策略全景十个策略怎么构成一个体系除上面三个重点策略外工具包还包含另外七种策略。它们共用同一套基类和 Runner 调度逻辑我在这里用一张表把策略核心逻辑和关键参数列清楚策略模块名核心因子关键入参典型适用场景rps_momentum相对价格强度RPSrps_win120, top_pct10强势股二阶段启动pe_percentilePE 历史分位lookback5年, max_pct30低估值洼地识别dividend_yield股息率min_div_rate3.0高分红防守型选股volume_ratio量比与换手率min_vol_ratio2.0, max_turnover10放量异动挖掘rsi_reversalRSI 超跌反转rsi_low30, rsi_win14超跌反弹左侧布局roe_qualityROE 与毛利率min_roe15, min_gross_margin30白马成长筛选earnings_growth净利润同比增速min_growth30, max_growth300业绩共振策略每个策略都是独立的.py文件文件头部的__init__决定了参数字典。使用的时候不用改策略内部的逻辑只需在配置文件或命令行参数里覆盖默认值。比如你想把股息率门槛从 3% 提到 5%直接改params里的min_div_rate即可。4. 自动选股脚本的完整运行流程从一条命令行到一张候选清单4.1 交易日判断与股票池过滤别在非交易日白跑工具包的入口是scan.py它做的事情分三步判断今天是否交易日、获取基础股票池、依次运行十个策略。非交易日跑脚本纯属浪费积分和时间所以我把交易日历判断放在最前面宁可多一次接口调用也不想让脚本在周末还去扫描全市场。# scan.py import os import argparse import tushare as ts import pandas as pd from datetime import datetime def is_trade_date(pro: ts.pro_api, date_str: str) - bool: df pro.trade_cal(exchangeSSE, start_datedate_str, end_datedate_str) return (df[is_open] 1).any() def load_stock_pool(pro: ts.pro_api) - pd.DataFrame: # 获取全部 A 股基础信息 stock_basic pro.stock_basic(exchange, list_statusL, fieldsts_code,name,industry,list_date) # 过滤北交所避免流动性陷阱 stock_basic stock_basic[~stock_basic[ts_code].str.endswith(.BJ)] return stock_basic def main(): parser argparse.ArgumentParser() parser.add_argument(--date, typestr, defaultdatetime.now().strftime(%Y%m%d)) parser.add_argument(--strategies, typestr, defaultall, help策略列表逗号分隔或 all 表示全部策略) parser.add_argument(--topn, typeint, default20, help每个策略输出前 N 只) args parser.parse_args() pro ts.pro_api() if not is_trade_date(pro, args.date): print(f{args.date} 不是交易日脚本退出) return stock_pool load_stock_pool(pro) print(f股票池数量{len(stock_pool)}) if __name__ __main__: main()股票池加载用的stock_basic接口list_statusL表示只在上市状态为 LListed的股票过滤掉退市整理期和暂停上市的。末端endswith(.BJ)的过滤是我自己加的北交所股票的流动性和涨跌幅规则跟沪深两市差异较大混在一起做统一扫描容易失真所以我默认剔除。如果你确实想看北交所把这行去掉就行。4.2 批量获取行情与并发取数的节奏控制工具包里最耗时的是策略内部循环调daily接口。要提高吞吐量团队一般会引入 Python 的ThreadPoolExecutor做有限并发——注意是「有限」不是无限。TuShare 接口有频率限制把并发数控制在 5 左右既能明显缩短运行时间又不会触发限流。# utils/fetch_pool.py from concurrent.futures import ThreadPoolExecutor, as_completed import tushare as ts import pandas as pd def fetch_kline_batch(pro, ts_codes: list, start_date: str, end_date: str, concurrency: int 5): def fetch_one(code): try: df pro.daily(ts_codecode, start_datestart_date, end_dateend_date) return code, df except Exception as e: print(f拉取 {code} 失败{e}) return code, None results {} with ThreadPoolExecutor(max_workersconcurrency) as executor: futures {executor.submit(fetch_one, code): code for code in ts_codes} for future in as_completed(futures): code, df future.result() results[code] df return results这套并发取数的封装放在utils目录下所有策略都在用它。每次请求之间没有显式sleep而是靠线程池排队自然错开。如果你发现仍然被限流调小concurrency到一个保守值或者在fetch_one里加一句time.sleep(0.2)。4.3 输出报告CSV、JSON 和日志每个策略跑完后最终结果会落到output/目录里。输出格式设计成「一个策略一个 CSV 文件 一个全量合并文件」。CSV 文件的列要包含股票代码、名称、综合分数、因子明细因子明细不像普通列那样扁平化而是用下划线拼接键值对便于直接查看。# utils/reporter.py import json import pandas as pd from pathlib import Path def save_strategy_result(result: pd.DataFrame, strategy_name: str, topn: int): out_dir Path(output) out_dir.mkdir(exist_okTrue) # 只保留前 topn 只 top result.head(topn).copy() # 把 factors 字段从 dict 转成可读字符串 if factors in top.columns: top[factor_detail] top[factors].apply( lambda x: json.dumps(x, ensure_asciiFalse) if isinstance(x, dict) else ) top top.drop(columns[factors]) path out_dir / f{strategy_name}_{datetime.now().strftime(%Y%m%d)}.csv top.to_csv(path, indexFalse, encodingutf-8-sig) print(f策略 {strategy_name} 结果已保存{path})这里有个细节值得注意CSV 编码用了utf-8-sig而不是utf-8。A 股股票名称包含中文用utf-8写入的文件在 Windows 上直接用 Excel 打开会乱码utf-8-sig会自动加 BOM 头Excel 就能正确识别。这是我第一次跑脚本输出报告后踩过的坑如果你用的是 macOS 的 Numbers 或者直接用pandas读取两者差别不大但做量化的人大部分还是习惯 Excel 复核候选列表。输出文件按日期命名同一策略不同日期的结果不会相互覆盖。这也是为后面要讲的「复盘数据资产」做的铺垫没有历史快照策略的好与坏根本无从评估。5. 常见问题与避坑排查回测漂亮但实盘买不进问题出在这四步5.1 选了买不进的票一字涨停板为什么也在候选列表里现象策略输出里出现一只昨天一字涨停的股票盘口封单巨大第二天开盘大概率还是一字板普通投资者根本成交不了。原因策略逻辑只看了日线收盘价和均线没有把涨跌停状态纳入过滤条件。经过连续一字板的股票移动平均线与动量指标天然会触发趋势策略的买入信号这是策略逻辑与交易执行脱节的典型表现。解决在获取日线数据时加上pct_chg字段过滤掉当日涨幅大于 9.8% 的标的。需要注意沪深主板涨停是 10%创业板和科创板是 20%不能用单一阈值。min_price我在代码里用一个参数limit_up_threshold区分板块创业板和科创板单独放宽到 19.5% 以上才过滤。def filter_limit_up(df: pd.DataFrame) - pd.DataFrame: # df 必须含 trade_date, pct_chg, 以及板块判断字段 cond ~( (df[pct_chg] 9.8) | (df[is_star] (df[pct_chg] 19.5)) ) return df[cond]5.2 均线算出来的「历史信号」在当下根本不成立前复权价格的使用现象同样一只股票用不同时间段拉取日线数据算出来的均线多头排列结论不一样。原因A 股有分红送转除权除息后价格会出现跳空。如果直接用 TuShare 返回的原始价格历史价格没有调整会导致除权日的均线突然断裂信号失真。工具包默认用的是pro.daily它返回的是不复权价格。解决统一用前复权价格计算所有技术指标。TuShare 提供ts.pro_bar接口支持adjqfq参数在获取日线数据时指定前复权即可。注意pro_bar和pro.daily的字段顺序略有差异用前复权数据做回测时整体信号会往右偏移一天左右建议统一用同一套接口获取数据。5.3 个别股票数据缺失导致策略中断接口返回空 DataFrame现象跑全市场扫描时某一只股票的daily返回空数据策略循环里没有做空值判断直接对空DataFrame取iloc[-1]抛IndexError导致整个策略中断。原因次新股或者刚复牌的股票在某个时间窗口可能确实没有行情数据。这个错误在开发环境很难发现因为单测通常用数据完善的股票做样例但全市场一跑就暴露了。解决所有取数逻辑都加「如果df为空或长度小于滚动窗口直接continue」的兜底逻辑。我在基类里放了safe_tail()方法专门处理这种情况。另一个隐蔽的细节是要检查数据里是否有全为NaN的列rolling计算遇到全空列会持续报警但不中断最终算出一堆脏结果。5.4 跑了四十分钟才跑完十个策略限流和数据量失衡现象脚本从 15:10 开始跑跑完全部策略已经 15:50中间还偶发超时错误。原因积分不足导致 TuShare 接口被限频每次请求之间实际等待时间远超正常水平另外每个策略都独立请求了一遍日线数据十个策略就是十份重复的网络开销。解决加一层缓存。工具包里的utils/cache.py提供了一个简单的本地缓存机制同一交易日的数据请求结果先写进内存再落到 SQLite后续策略取数时优先读缓存只有缓存不存在时才发请求。我试过的实测效果是第一个策略跑完后其余九个策略的取数耗时几乎可以忽略不计。实际跑一遍后10 个策略的总时间能从 40 分钟压到 15 分钟以内。6. 进阶把每日选股结果变成可复盘的数据资产6.1 把候选落库每天追加一条记录做横截面比对输出 CSV 是给人看的但如果想系统评估策略质量就必须把每日选股结果存进结构化数据库里。SQLite 是零配置、单文件、够用的方案。我每天收市后跑完脚本会把每个策略前 20 名的结果统一写入pick_result.db。import sqlite3 from datetime import datetime def save_picks_to_db(strategy_name: str, result: pd.DataFrame, trade_date: str): conn sqlite3.connect(pick_result.db) result result.copy() result[strategy] strategy_name result[trade_date] trade_date result[save_time] datetime.now().strftime(%Y-%m-%d %H:%M:%S) result.to_sql(daily_picks, conn, if_existsappend, indexFalse) conn.close()if_existsappend表示每天追加不会覆盖历史记录。有了这张daily_picks表你就能回答「某个策略上周选的股票这周表现到底怎么样」这类问题。表里的factors字段保存因子明细方便逐日漂移分析。6.2 结合次日行情给信号打分反馈脚本与质量追踪落库只是第一步真正有价值的是建立「信号—次日表现」的反馈链路。工具包里附带了一个feedback.py它的逻辑是读取daily_picks表里昨天的记录用今天的实际行情计算每只候选股票相对于当天开盘的涨跌幅再按策略聚合出平均涨幅。def calculate_feedback(picks: pd.DataFrame, pro): feedback_rows [] for row in picks.itertuples(): daily pro.daily(ts_coderow.ts_code, start_daterow.trade_date, end_daterow.trade_date) if daily is None or daily.empty: continue open_price daily[open].iloc[0] close_price daily[close].iloc[0] feedback_rows.append({ ts_code: row.ts_code, strategy: row.strategy, overnight_return: (close_price / open_price - 1) * 100 }) return pd.DataFrame(feedback_rows)这里统计的是「次日开盘买入到次日收盘」的收益本质上是一个简化的持有期收益。你可以把窗口延长到 3 日、5 日只要你自己逻辑一致就行。关键不是收益率数字本身的绝对高低而是策略之间的相对排序是否稳定——连续两周排名垫底的策略数据再好看也要先停掉。6.3 加一道冒烟测试跑策略前先自检工具包配了一个用pytest写的冒烟测试脚本逻辑是用极小的股票池比如 10 只跑一遍全部十个策略要求不报错、不返回空结果。这样做能提前暴露接口权限变化、字段缺失或数据格式异常等问题。cd quant_picker pytest tests/smoke_test.py -v跑这条命令大约需要半分钟但能极大减少全量扫描跑到一半才发现问题的尴尬。冒烟测试里除了校验返回非空还会校验score列是否全是合法数值、输出 CSV 是否能正常写入。这样每次改动策略参数后第一步是跑冒烟测试第二步才是全量扫描两条命令各司其职。从那以后我每天收市后的固定动作是先跑冒烟测试再跑十个策略最后把候选列表和因子明细存进 SQLite每周抽一天把累计结果拉出来看看哪些策略连续跑赢均值。中间也出现过策略因子数据字段改名导致冒烟测试红了半天、结果排重时把同一天相同代码重复入库的事情现在我一律把检查和入库的步骤固定成脚本流程不再手动操作。数据链路越自动化你才越敢把资金决策交给它。希望这些实操细节对你有用也帮你在自己的机器上尽快把这套选股流程跑顺。本文还有配套的精品资源点击获取 
长河君 · 2026/10/18 16:02:52 · 阅读 1.2k Python+TuShare构建A股量化选股工具包，10大策略实战解析
简介这是一套面向A股个人投资者与量化初学者的Python自动选股工具包内置10种常见技术选股策略覆盖250日均线突破、平台突破、回踩均线、停机坪形态、低ATR波动筛选及持续上涨趋势识别等场景支持灵活组合与快速替换。整个压缩包共27个文件以Python脚本为主17个py脚本另含策略示意图png/jpg、依赖清单、配置文件与使用说明解压后约1.96MB模块划分清晰便于直接运行和替换策略。目前已有110人学习下载适合快速验证选股思路并作为二次开发基础。工具包附带数据获取、本地SQLite存储、策略回测统计、执行流程控制及买卖信号触发模块可独立完成从行情获取到信号输出的完整链路同时配有说明文档和策略逻辑图能帮助用户快速上手实践。1. 把选股从“盘中盯盘”改成“收盘后跑脚本”这个工具包解决的不只是时间先说一个可能反直觉的结论每天都盯盘四个小时的人往往不如每天收盘后用十分钟跑一次量化选股脚本的人更清楚第二天该看什么。原因很简单——盘中盯盘看到的是价格跳动和情绪噪音而量化选股工具包解决的是「把选股标准固化成代码」这件事。这个包含 10 种 Python 选股策略的工具包底层用 TuShare 拉取 A 股行情与财务数据通过统一框架批量扫描全市场最后输出一份带因子分数的候选股票 CSV。它不预测涨跌只帮你把「什么股票值得关注」这件事从拍脑袋变成可复用、可回测、可复盘的程序逻辑。适合两类人一类是会用 Python 但还没系统做过量化选股的投资者另一类是已经有主观交易体系、想把自己的选股规则固化下来的从业者。2. 工具包骨架与前置准备TuShare 接口、目录结构和策略基类设计2.1 调通 TuShare 接口token、pro 接口与积分限制所有策略的数据源都建立在 TuShare 之上。使用前需要先注册一个账号在个人主页拿到 token。这个 token 就是你调用数据接口的钥匙配置方式有两种直接写在代码里或者通过环境变量传入。我一般用后者因为脚本会分享给同事或换机器跑硬编码 token 迟早会泄露。import os import tushare as ts # 从环境变量读取 token避免硬编码 token os.environ.get(TUSHARE_TOKEN, 你的token放在这里) ts.set_token(token) # 初始化 pro 接口后续所有数据调用都走 pro pro ts.pro_api() # 验证接口是否通拉取最近一个交易日 df pro.trade_cal(exchangeSSE, start_date20240101, end_date20241231, is_open1) print(df.tail(3))这段代码先设置 token然后调用trade_cal获取交易日历。注意exchange参数传SSE即可因为沪深两市日历基本一致没必要分别拉取。is_open1表示只要开盘日这在你后面判断「今天该不该跑策略」时非常有用。TuShare 的积分是很多人第一次跑脚本时翻车的根源。新注册账号默认积分比较低能调用的接口和数据频率都受限。比如daily接口完整调用需要更高的积分daily_basic里的市净率、市销率字段同样有权限门槛。我遇到的情况是当天注册后拉取单只股票日线没问题但全市场批量取数时直接被限流。解决办法是在 TuShare 社区完善资料、通过实名认证提升积分或者在代码里对每次请求做限速主动 sleep减少被限流的概率。2.2 读懂包里的目录策略基类与 Runner 的分工工具包不是十个脚本堆在一起而是有一个清晰的骨架。每个策略文件都继承同一个基类基类定义好「要做什么」具体策略只负责「怎么做」。这样做的直接收益是你想加第 11 个策略只需要新建一个文件并实现两个方法Runner 会自动加载它。# base_strategy.py from abc import ABC, abstractmethod from dataclasses import dataclass, field import pandas as pd dataclass class StockCandidate: ts_code: str # 股票代码如 600000.SH name: str # 股票名称 score: float # 策略打分越高越靠前 factors: dict field(default_factorydict) # 因子明细方便复盘 class BaseStrategy(ABC): strategy_name base def __init__(self, params: dict None): self.params params or {} self.result [] abstractmethod def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: 输入全市场股票池输出过滤后的股票列表 abstractmethod def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: 对过滤后的股票打分排序返回带 score 列的数据框 def run(self, stock_pool: pd.DataFrame) - pd.DataFrame: filtered self.filter_stocks(stock_pool) scored self.score_stocks(filtered) scored scored.sort_values(score, ascendingFalse) return scored基类里我用dataclass定义了候选股票的实体结构StockCandidate里除了代码和名称还有个factors字典专门用来保存每个策略算出来的因子明细。这样输出的 CSV 不只是「推荐了哪只股票」还能看到「为什么推荐它」。比如说某只股票是因为低市净率被选中的还是因为动量分值高被选中的复盘时一眼就能看清楚。run()方法是模板方法先filter_stocks做粗筛再score_stocks做排序。之所以把过滤和打分拆成两步是因为 A 股全市场五千多只股票很多策略第一步可以用市值、ST 状态、停牌状态快速过滤掉大量明显不合规的标的减少后续计算量。Runner 的角色是调度者。它读取某个策略类传入当天的股票池然后触发run()。策略之间互相独立即使某个策略因为数据缺失挂了不影响其他策略继续跑。# runner.py import importlib import pandas as pd from base_strategy import BaseStrategy def load_strategy(strategy_name: str, params: dict) - BaseStrategy: module importlib.import_module(fstrategies.{strategy_name}) clazz getattr(module, strategy_name.split(_)[0].capitalize() Strategy) return clazz(params) def run_single_strategy(strategy_name: str, stock_pool: pd.DataFrame, params: dict): strategy load_strategy(strategy_name, params) result strategy.run(stock_pool) return result这里加载策略用的是importlib动态导入模块名和策略名一一对应。比如strategies.small_cap对应SmallCapStrategy。注意命名要规范否则getattr找不到类名时会直接抛异常。我会在 Runner 里加一层异常捕获单独记录哪个策略失败了而不是让整个脚本崩掉——全市场扫描跑二十分钟因为最后一个策略的字段缺失全部重跑这体验太糟糕了。3. 十种策略拆解与参数改动多因子、均线、动量、小市值的取舍3.1 多因子打分PE、PB、ROE 合成截面分数多因子策略是这个工具包里的「万金油」它的思路不是找一个独门指标而是把几个有效因子标准化后加权合成一个综合分数。比如市盈率、市净率越低越好ROE 越高越好那么先对每个因子做截面标准化再按权重相加。# strategies/multi_factor.py import pandas as pd import numpy as np from base_strategy import BaseStrategy class MultiFactorStrategy(BaseStrategy): strategy_name multi_factor def __init__(self, params: dict None): super().__init__(params) self.weight_pe params.get(weight_pe, -0.3) # 负权重PE 越低分越高 self.weight_pb params.get(weight_pb, -0.2) self.weight_roe params.get(weight_roe, 0.5) def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: # 去掉 ST、停牌、上市不满一年的股票 df stock_pool.copy() df df[~df[name].str.contains(ST, naFalse)] df df[df[list_date] 20250101] return df def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: # 拉取每日基本面指标 import tushare as ts pro ts.pro_api() all_daily_basic [] for code in filtered[ts_code].tolist(): db pro.daily_basic(ts_codecode, fieldsts_code,trade_date,pe,pb,roe) if db is not None and len(db): last db.iloc[0] # 默认按日期倒序取最新 all_daily_basic.append(last) basic_df pd.DataFrame(all_daily_basic) merged filtered.merge(basic_df, onts_code, howleft) # 对每个因子做 zscore 标准化 merged[pe_z] (merged[pe] - merged[pe].mean()) / merged[pe].std() merged[pb_z] (merged[pb] - merged[pb].mean()) / merged[pb].std() merged[roe_z] (merged[roe] - merged[roe].mean()) / merged[roe].std() # 合成综合分数 merged[score] (self.weight_pe * merged[pe_z] self.weight_pb * merged[pb_z] self.weight_roe * merged[roe_z]) return merged[[ts_code, name, score, pe, pb, roe]]这个策略跑起来最大的瓶颈在于for循环调用daily_basic。虽然代码直观但全市场逐只拉取要请求几千次速度很慢。更高效的做法是调用 TuShare 的daily_basic(trade_date某一天)一次性获取全市场当天的基本面数据但这对积分的门槛更高。工具包默认提供的是逐只拉取的兼容写法实际使用中我通常会改成按日期批量拉取的模式。weight_pe、weight_pb、weight_roe是三个核心参数。负权重表示指标值越低贡献越高ROE 是正向的。这套默认权重来自经典的价值选股思路但你可以按自己的偏好改比如更偏好成长的话可以把weight_roe调到 0.7。改参数时注意每个权重的量级因为 zscore 标准化后的数值已经消除了量纲差异权重可以直接比较大小。3.2 均线多头排列与动量过滤趋势型策略的信号写法趋势型策略的核心假设是「强者恒强」。均线多头排列是最经典的形态短期均线在中期均线之上中期均线在长期均线之上。但只看排列还不够我会额外加一个动量过滤条件要求近 20 日涨幅大于一定阈值这样能过滤掉那些刚刚从下跌转为粘合的模糊信号。# strategies/ma_trend.py import pandas as pd import tushare as ts from base_strategy import BaseStrategy class MaTrendStrategy(BaseStrategy): strategy_name ma_trend def __init__(self, params: dict None): super().__init__(params) self.short_win params.get(short_win, 5) self.mid_win params.get(mid_win, 10) self.long_win params.get(long_win, 20) self.momentum_threshold params.get(momentum_threshold, 5.0) def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: # 做趋势策略尽量选流动性好的过滤掉日均成交额太低的 df stock_pool.copy() return df def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: pro ts.pro_api() result_rows [] for code in filtered[ts_code].tolist(): df pro.daily(ts_codecode, start_date20250115, end_date20250415) if df is None or len(df) 40: continue df df.sort_values(trade_date) df[fma_{self.short_win}] df[close].rolling(self.short_win).mean() df[fma_{self.mid_win}] df[close].rolling(self.mid_win).mean() df[fma_{self.long_win}] df[close].rolling(self.long_win).mean() latest df.iloc[-1] ma_short, ma_mid, ma_long latest[fma_{self.short_win}], latest[fma_{self.mid_win}], latest[fma_{self.long_win}] if not (ma_short ma_mid ma_long): continue # 动量近 20 日累计涨跌幅 close_20d_ago df[close].iloc[-min(20, len(df))] momentum (latest[close] / close_20d_ago - 1) * 100 if momentum self.momentum_threshold: continue row filtered[filtered[ts_code] code].iloc[0] result_rows.append({ ts_code: code, name: row[name], score: momentum, # 动量越高越靠前 factors: {momentum_20d: round(momentum, 2), ma_long: round(ma_long, 2)} }) return pd.DataFrame(result_rows)这段代码有几个细节需要留意。一是sorted_values(trade_date)必须做升序排列因为pro.daily返回的数据默认按日期倒序如果不排序rolling算出来的均线就是「未来数据」会导致买到未来的信号——这在回测里叫未来函数是最严重的逻辑漏洞。二是动量的计算我取了倒数第 20 个交易日的数据并用min(20, len(df))保护避免新股数据不足时越界。momentum_threshold参数建议设成 310 之间。设得太低会混入横盘整理的股票设得太高容易追在情绪高点。我一般分市场环境调整强势市场用 8震荡市用 3。3.3 小市值策略我为什么保留它但加了价格下限小市值策略在过去很长一段时间里是 A 股超额收益最稳定的来源之一但它也是回撤最猛烈的策略。每天收盘后按总市值排序买入市值最小的几十只这个逻辑简单到不需要任何财务分析但工具包还是保留了它因为它是验证「策略逻辑—代码实现—数据链路」整条流程是否通畅的最佳样例。# strategies/small_cap.py import pandas as pd import tushare as ts from base_strategy import BaseStrategy class SmallCapStrategy(BaseStrategy): strategy_name small_cap def __init__(self, params: dict None): super().__init__(params) self.min_price params.get(min_price, 2.0) # 最低股价过滤低价股 self.max_price params.get(max_price, 20.0) # 最高股价过滤高价股 self.max_mv params.get(max_mv, 50) # 最大流通市值亿元 self.top_n params.get(top_n, 30) # 最终输出前 30 只 def filter_stocks(self, stock_pool: pd.DataFrame) - pd.DataFrame: df stock_pool.copy() df df[~df[name].str.contains(ST, naFalse)] df df[df[trade_status] 交易中] return df def score_stocks(self, filtered: pd.DataFrame) - pd.DataFrame: pro ts.pro_api() # 拉取全市场最新市值与股价 daily_basic pro.daily_basic(trade_date20250415, fieldsts_code,close,total_mv,circ_mv) merged filtered.merge(daily_basic, onts_code, howinner) # 股价过滤低价股流动性差且容易受借壳炒作影响 merged merged[(merged[close] self.min_price) (merged[close] self.max_price)] # 流通市值过滤单位是万元转成亿元 merged[circ_mv_yi] merged[circ_mv] / 10000 merged merged[merged[circ_mv_yi] self.max_mv] # 市值越小分数越高 merged[score] -merged[circ_mv_yi] merged merged.sort_values(score).head(self.top_n) return merged[[ts_code, name, score, close, circ_mv_yi]]我特意加了min_price和max_price两个参数这是从小市值策略翻车里总结出来的。纯粹按市值排序会把大量股价两块钱以下的股票选进来这些股票看着便宜但多数是基本面出了问题或者流通盘被操控的标的普通投资者根本拿不住。加上价格区间过滤后候选列表的质地会明显改善。circ_mv字段单位是万元所以除以 10000 换算成亿元才便于人读。这里有个隐蔽的坑TuShare 的total_mv和circ_mv字段返回单位是万元但不同接口文档表述可能不同有些第三方封装库会直接换算成亿元。如果你在运行工具包时发现市值数值大得离谱先确认是不是做了单位换算。3.4 策略全景十个策略怎么构成一个体系除上面三个重点策略外工具包还包含另外七种策略。它们共用同一套基类和 Runner 调度逻辑我在这里用一张表把策略核心逻辑和关键参数列清楚策略模块名核心因子关键入参典型适用场景rps_momentum相对价格强度RPSrps_win120, top_pct10强势股二阶段启动pe_percentilePE 历史分位lookback5年, max_pct30低估值洼地识别dividend_yield股息率min_div_rate3.0高分红防守型选股volume_ratio量比与换手率min_vol_ratio2.0, max_turnover10放量异动挖掘rsi_reversalRSI 超跌反转rsi_low30, rsi_win14超跌反弹左侧布局roe_qualityROE 与毛利率min_roe15, min_gross_margin30白马成长筛选earnings_growth净利润同比增速min_growth30, max_growth300业绩共振策略每个策略都是独立的.py文件文件头部的__init__决定了参数字典。使用的时候不用改策略内部的逻辑只需在配置文件或命令行参数里覆盖默认值。比如你想把股息率门槛从 3% 提到 5%直接改params里的min_div_rate即可。4. 自动选股脚本的完整运行流程从一条命令行到一张候选清单4.1 交易日判断与股票池过滤别在非交易日白跑工具包的入口是scan.py它做的事情分三步判断今天是否交易日、获取基础股票池、依次运行十个策略。非交易日跑脚本纯属浪费积分和时间所以我把交易日历判断放在最前面宁可多一次接口调用也不想让脚本在周末还去扫描全市场。# scan.py import os import argparse import tushare as ts import pandas as pd from datetime import datetime def is_trade_date(pro: ts.pro_api, date_str: str) - bool: df pro.trade_cal(exchangeSSE, start_datedate_str, end_datedate_str) return (df[is_open] 1).any() def load_stock_pool(pro: ts.pro_api) - pd.DataFrame: # 获取全部 A 股基础信息 stock_basic pro.stock_basic(exchange, list_statusL, fieldsts_code,name,industry,list_date) # 过滤北交所避免流动性陷阱 stock_basic stock_basic[~stock_basic[ts_code].str.endswith(.BJ)] return stock_basic def main(): parser argparse.ArgumentParser() parser.add_argument(--date, typestr, defaultdatetime.now().strftime(%Y%m%d)) parser.add_argument(--strategies, typestr, defaultall, help策略列表逗号分隔或 all 表示全部策略) parser.add_argument(--topn, typeint, default20, help每个策略输出前 N 只) args parser.parse_args() pro ts.pro_api() if not is_trade_date(pro, args.date): print(f{args.date} 不是交易日脚本退出) return stock_pool load_stock_pool(pro) print(f股票池数量{len(stock_pool)}) if __name__ __main__: main()股票池加载用的stock_basic接口list_statusL表示只在上市状态为 LListed的股票过滤掉退市整理期和暂停上市的。末端endswith(.BJ)的过滤是我自己加的北交所股票的流动性和涨跌幅规则跟沪深两市差异较大混在一起做统一扫描容易失真所以我默认剔除。如果你确实想看北交所把这行去掉就行。4.2 批量获取行情与并发取数的节奏控制工具包里最耗时的是策略内部循环调daily接口。要提高吞吐量团队一般会引入 Python 的ThreadPoolExecutor做有限并发——注意是「有限」不是无限。TuShare 接口有频率限制把并发数控制在 5 左右既能明显缩短运行时间又不会触发限流。# utils/fetch_pool.py from concurrent.futures import ThreadPoolExecutor, as_completed import tushare as ts import pandas as pd def fetch_kline_batch(pro, ts_codes: list, start_date: str, end_date: str, concurrency: int 5): def fetch_one(code): try: df pro.daily(ts_codecode, start_datestart_date, end_dateend_date) return code, df except Exception as e: print(f拉取 {code} 失败{e}) return code, None results {} with ThreadPoolExecutor(max_workersconcurrency) as executor: futures {executor.submit(fetch_one, code): code for code in ts_codes} for future in as_completed(futures): code, df future.result() results[code] df return results这套并发取数的封装放在utils目录下所有策略都在用它。每次请求之间没有显式sleep而是靠线程池排队自然错开。如果你发现仍然被限流调小concurrency到一个保守值或者在fetch_one里加一句time.sleep(0.2)。4.3 输出报告CSV、JSON 和日志每个策略跑完后最终结果会落到output/目录里。输出格式设计成「一个策略一个 CSV 文件 一个全量合并文件」。CSV 文件的列要包含股票代码、名称、综合分数、因子明细因子明细不像普通列那样扁平化而是用下划线拼接键值对便于直接查看。# utils/reporter.py import json import pandas as pd from pathlib import Path def save_strategy_result(result: pd.DataFrame, strategy_name: str, topn: int): out_dir Path(output) out_dir.mkdir(exist_okTrue) # 只保留前 topn 只 top result.head(topn).copy() # 把 factors 字段从 dict 转成可读字符串 if factors in top.columns: top[factor_detail] top[factors].apply( lambda x: json.dumps(x, ensure_asciiFalse) if isinstance(x, dict) else ) top top.drop(columns[factors]) path out_dir / f{strategy_name}_{datetime.now().strftime(%Y%m%d)}.csv top.to_csv(path, indexFalse, encodingutf-8-sig) print(f策略 {strategy_name} 结果已保存{path})这里有个细节值得注意CSV 编码用了utf-8-sig而不是utf-8。A 股股票名称包含中文用utf-8写入的文件在 Windows 上直接用 Excel 打开会乱码utf-8-sig会自动加 BOM 头Excel 就能正确识别。这是我第一次跑脚本输出报告后踩过的坑如果你用的是 macOS 的 Numbers 或者直接用pandas读取两者差别不大但做量化的人大部分还是习惯 Excel 复核候选列表。输出文件按日期命名同一策略不同日期的结果不会相互覆盖。这也是为后面要讲的「复盘数据资产」做的铺垫没有历史快照策略的好与坏根本无从评估。5. 常见问题与避坑排查回测漂亮但实盘买不进问题出在这四步5.1 选了买不进的票一字涨停板为什么也在候选列表里现象策略输出里出现一只昨天一字涨停的股票盘口封单巨大第二天开盘大概率还是一字板普通投资者根本成交不了。原因策略逻辑只看了日线收盘价和均线没有把涨跌停状态纳入过滤条件。经过连续一字板的股票移动平均线与动量指标天然会触发趋势策略的买入信号这是策略逻辑与交易执行脱节的典型表现。解决在获取日线数据时加上pct_chg字段过滤掉当日涨幅大于 9.8% 的标的。需要注意沪深主板涨停是 10%创业板和科创板是 20%不能用单一阈值。min_price我在代码里用一个参数limit_up_threshold区分板块创业板和科创板单独放宽到 19.5% 以上才过滤。def filter_limit_up(df: pd.DataFrame) - pd.DataFrame: # df 必须含 trade_date, pct_chg, 以及板块判断字段 cond ~( (df[pct_chg] 9.8) | (df[is_star] (df[pct_chg] 19.5)) ) return df[cond]5.2 均线算出来的「历史信号」在当下根本不成立前复权价格的使用现象同样一只股票用不同时间段拉取日线数据算出来的均线多头排列结论不一样。原因A 股有分红送转除权除息后价格会出现跳空。如果直接用 TuShare 返回的原始价格历史价格没有调整会导致除权日的均线突然断裂信号失真。工具包默认用的是pro.daily它返回的是不复权价格。解决统一用前复权价格计算所有技术指标。TuShare 提供ts.pro_bar接口支持adjqfq参数在获取日线数据时指定前复权即可。注意pro_bar和pro.daily的字段顺序略有差异用前复权数据做回测时整体信号会往右偏移一天左右建议统一用同一套接口获取数据。5.3 个别股票数据缺失导致策略中断接口返回空 DataFrame现象跑全市场扫描时某一只股票的daily返回空数据策略循环里没有做空值判断直接对空DataFrame取iloc[-1]抛IndexError导致整个策略中断。原因次新股或者刚复牌的股票在某个时间窗口可能确实没有行情数据。这个错误在开发环境很难发现因为单测通常用数据完善的股票做样例但全市场一跑就暴露了。解决所有取数逻辑都加「如果df为空或长度小于滚动窗口直接continue」的兜底逻辑。我在基类里放了safe_tail()方法专门处理这种情况。另一个隐蔽的细节是要检查数据里是否有全为NaN的列rolling计算遇到全空列会持续报警但不中断最终算出一堆脏结果。5.4 跑了四十分钟才跑完十个策略限流和数据量失衡现象脚本从 15:10 开始跑跑完全部策略已经 15:50中间还偶发超时错误。原因积分不足导致 TuShare 接口被限频每次请求之间实际等待时间远超正常水平另外每个策略都独立请求了一遍日线数据十个策略就是十份重复的网络开销。解决加一层缓存。工具包里的utils/cache.py提供了一个简单的本地缓存机制同一交易日的数据请求结果先写进内存再落到 SQLite后续策略取数时优先读缓存只有缓存不存在时才发请求。我试过的实测效果是第一个策略跑完后其余九个策略的取数耗时几乎可以忽略不计。实际跑一遍后10 个策略的总时间能从 40 分钟压到 15 分钟以内。6. 进阶把每日选股结果变成可复盘的数据资产6.1 把候选落库每天追加一条记录做横截面比对输出 CSV 是给人看的但如果想系统评估策略质量就必须把每日选股结果存进结构化数据库里。SQLite 是零配置、单文件、够用的方案。我每天收市后跑完脚本会把每个策略前 20 名的结果统一写入pick_result.db。import sqlite3 from datetime import datetime def save_picks_to_db(strategy_name: str, result: pd.DataFrame, trade_date: str): conn sqlite3.connect(pick_result.db) result result.copy() result[strategy] strategy_name result[trade_date] trade_date result[save_time] datetime.now().strftime(%Y-%m-%d %H:%M:%S) result.to_sql(daily_picks, conn, if_existsappend, indexFalse) conn.close()if_existsappend表示每天追加不会覆盖历史记录。有了这张daily_picks表你就能回答「某个策略上周选的股票这周表现到底怎么样」这类问题。表里的factors字段保存因子明细方便逐日漂移分析。6.2 结合次日行情给信号打分反馈脚本与质量追踪落库只是第一步真正有价值的是建立「信号—次日表现」的反馈链路。工具包里附带了一个feedback.py它的逻辑是读取daily_picks表里昨天的记录用今天的实际行情计算每只候选股票相对于当天开盘的涨跌幅再按策略聚合出平均涨幅。def calculate_feedback(picks: pd.DataFrame, pro): feedback_rows [] for row in picks.itertuples(): daily pro.daily(ts_coderow.ts_code, start_daterow.trade_date, end_daterow.trade_date) if daily is None or daily.empty: continue open_price daily[open].iloc[0] close_price daily[close].iloc[0] feedback_rows.append({ ts_code: row.ts_code, strategy: row.strategy, overnight_return: (close_price / open_price - 1) * 100 }) return pd.DataFrame(feedback_rows)这里统计的是「次日开盘买入到次日收盘」的收益本质上是一个简化的持有期收益。你可以把窗口延长到 3 日、5 日只要你自己逻辑一致就行。关键不是收益率数字本身的绝对高低而是策略之间的相对排序是否稳定——连续两周排名垫底的策略数据再好看也要先停掉。6.3 加一道冒烟测试跑策略前先自检工具包配了一个用pytest写的冒烟测试脚本逻辑是用极小的股票池比如 10 只跑一遍全部十个策略要求不报错、不返回空结果。这样做能提前暴露接口权限变化、字段缺失或数据格式异常等问题。cd quant_picker pytest tests/smoke_test.py -v跑这条命令大约需要半分钟但能极大减少全量扫描跑到一半才发现问题的尴尬。冒烟测试里除了校验返回非空还会校验score列是否全是合法数值、输出 CSV 是否能正常写入。这样每次改动策略参数后第一步是跑冒烟测试第二步才是全量扫描两条命令各司其职。从那以后我每天收市后的固定动作是先跑冒烟测试再跑十个策略最后把候选列表和因子明细存进 SQLite每周抽一天把累计结果拉出来看看哪些策略连续跑赢均值。中间也出现过策略因子数据字段改名导致冒烟测试红了半天、结果排重时把同一天相同代码重复入库的事情现在我一律把检查和入库的步骤固定成脚本流程不再手动操作。数据链路越自动化你才越敢把资金决策交给它。希望这些实操细节对你有用也帮你在自己的机器上尽快把这套选股流程跑顺。本文还有配套的精品资源点击获取