Python+TuShare构建A股自动选股程序：数据、规则与避坑指南-北京尧图网络

简介面向量化投资入门者与Python开发者的A股自动选股实战项目基于TuShare金融数据接口覆盖行情获取、数据清洗、策略制定与回测分析全流程。压缩包共25个文件以17个Python脚本为主体包含数据抓取、策略模块、工作流调度、工具函数等核心代码另有说明文档、配置文件及多张策略效果图整体仅1.95MB目录结构清晰便于按模块研读与二次开发。目前已有6000余人学习浏览适合想掌握金融数据接口调用、量化选股思路的初中级开发者。包内源码涉及移动平均线、RSI、MACD等技术指标以及基于财务数据的基本面判断逻辑并附带回测模块用于验证策略在不同市场环境下的表现。通过阅读与运行这套代码读者可快速搭建个人量化研究框架理解从数据获取到买卖信号生成的完整决策链路是一份实战价值较高的Python金融编程学习资料。1. 这到底是个什么程序TuShare选股不是写个指标那么简单“Python-基于TuShare的A股自动选股程序”这句话听起来像个脚本但真正落地是一条数据流水线用 TuShare 把行情和基础数据拉到本地按你定义的规则筛选出股票池再把结果落成每日信号。很多新手把它当成“写个 MACD 交叉跑一遍”结果数据不全、复权不对、停牌没处理、信号重复推送每一步都变成玄学。它能解决的问题很具体每天手工翻股票太耗时、选股标准不统一、复盘时说不清为什么选它。适合有点 Python 基础、想做量化入门或者要给内部小组提供每日股票池的人。注意它自动化的是选股流程不是“躺赚”能力选股逻辑的质量决定程序上限代码只是把上限稳定地兑现出来。2. 先搭数据底座Token、接口限频与日线复权数据落库选股程序最怕的不是策略对不对而是数据能不能回答“这个信号到底来自哪一天”。我一般先建一个本地数据库把所有原始行情落进去之后所有指标和信号都从本地算而不是每天现场拉一遍全网数据。这一步做扎实后面调策略、改参数、回看信号都会轻松很多。2.1 为什么我选 TuShare 而不是自己抓网页自己写爬虫抓行情网页第一眼看上去简单一个请求拿 JSON解析一下就有价格。但后面的坑是连续的页面结构改一次崩一次盘中接口会限流字段对不上还要自己处理复权因子和交易日历。TuShare 至少把字段标准化了返回 pandas DataFrame历史接口也相对稳定。它的限制同样要早知道需要 token有积分和频次限制部分接口不是所有用户都能调用。所以我只把 TuShare 当数据源不在主程序里频繁实时调用而是先落库。存储上我一般先用 SQLite而不是立刻上 MySQL。单机选股、每天一次全市场扫描SQLite 足够备份就把文件拷走。等股票池和频率增长到需要并发读取再考虑换 MySQL 或 PostgreSQL。先跑通流程再谈集群。2.2 最小可用代码取日线数据并入库下面这段是日常扫描的基础版本。注意这里拿的是不复权日线适合做原始底表算指标时再叠加复权逻辑。import tushare as ts import pandas as pd import sqlite3 from datetime import datetime # token 在 TuShare 个人页面申请这里用占位符 ts.set_token(your_token_here) pro ts.pro_api() def init_db(conn): conn.execute( CREATE TABLE IF NOT EXISTS daily ( ts_code TEXT, trade_date TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL, change REAL, pct_chg REAL, vol REAL, amount REAL, PRIMARY KEY (ts_code, trade_date) ) ) conn.commit() def fetch_daily_and_save(conn, ts_code, start_date, end_date): df pro.daily(ts_codets_code, start_datestart_date, end_dateend_date) if df is None or df.empty: return 0 df df.sort_values(trade_date) rows df[[ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount]].values.tolist() conn.executemany( INSERT OR REPLACE INTO daily (ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , rows) conn.commit() return len(rows)pro.daily的字段含义pre_close是昨收pct_chg是涨跌幅百分比vol单位是手amount单位是千元。日期参数都要求YYYYMMDD字符串不要传 datetime 对象。INSERT OR REPLACE保证同一只股票同一天重复拉取时不会生成重复行如果某天数据被修订过重跑一遍就能覆盖成新值。股票代码格式是带交易所后缀的比如000XXX.SZ、600XXX.SH。真实代码列表可以从pro.stock_basic()拉出来list_status可以筛上市、退市状态。入库时直接用这个带后缀的字符串做主键的一部分避免沪深代码重复。2.3 增量更新别每次都全市场全量拉第一次可以全量之后每天只补最近缺口。增量更新的核心是记住每只股票已经入库的最新日期然后用交易日历跳过周末和节假日。def get_latest_date(conn, ts_code): cur conn.execute( SELECT MAX(trade_date) FROM daily WHERE ts_code ?, (ts_code,) ) row cur.fetchone() return row[0] if row and row[0] else None def update_stock(conn, ts_code, end_dateNone): end_date end_date or datetime.now().strftime(%Y%m%d) latest get_latest_date(conn, ts_code) if latest is None: start_date 20150101 # 第一次全量按你的回测区间调整 else: start_date (pd.to_datetime(latest) pd.Timedelta(days1)).strftime(%Y%m%d) return fetch_daily_and_save(conn, ts_code, start_date, end_date)不要直接把latest 1 天当成请求起始日去调用接口因为如果最新数据是周五那么周六日没有行情得先过滤交易日历。常见做法是再查一遍pro.trade_cal(exchangeSSE, start_datestart_date, end_dateend_date)只取is_open 1的日期再逐日补齐。这样增量更新就不会把“没有行情”误判成“数据缺失”。3. 把选股策略拆成规则引擎基本面、动量与流动性的组合打分有了数据底表下一步就是把选股逻辑从“人脑执行”变成“代码执行”。常见做法是拆成规则引擎先用硬条件过滤再用软因子打分排序。3.1 为什么用规则引擎而不是一堆 if 嵌套如果你只有五六个条件if 嵌套也能写。但实际选股会不断加条件剔除 ST、剔除上市不足 60 天、成交额下限、20 日涨幅区间、均线多头排列、PE 区间……用 if 写到最后要么程序很难测试要么改一个阈值牵一发动全身。规则引擎的思路是过滤阶段去掉明显不合格的股票打分阶段对剩余股票排序。这样每天输出的是一个带分数的候选列表而不是一堆“为什么选它”说不清的信号。调参时也只需要改配置不需要动主流程代码。我把所有阈值都放到一个params字典里方便做参数实验。3.2 实现一个打分器从日线特征到分数排序先给每只股票的日线算基础特征。这里用 groupby而不是循环每只股票速度会快很多。import pandas as pd def make_features(daily_df): df daily_df.sort_values([ts_code, trade_date]) df[ma5] df.groupby(ts_code)[close].transform( lambda x: x.rolling(5).mean()) df[ma20] df.groupby(ts_code)[close].transform( lambda x: x.rolling(20).mean()) # 20日动量今天相对20天前的涨跌幅 df[mom_20] df.groupby(ts_code)[close].pct_change(20) # 量比当天成交量 / 过去5日均量 df[vol_ma5] df.groupby(ts_code)[vol].transform( lambda x: x.rolling(5).mean()) df[vol_ratio] df[vol] / df[vol_ma5] return df注意pct_change(20)产生的前 20 行是 NaN计算指标时要在后续过滤里去掉。vol_ma5如果为 0量比会变成 inf所以真实环境里要先看数据有没有异常值。过滤和打分可以放在一起import numpy as np def filter_and_score(feature_df, date, params): today feature_df[feature_df[trade_date] date].copy() # 硬过滤成交额下限单位是千元 today today[today[amount] params[min_amount]] # 硬过滤站上20日线避免接下跌趋势 today today[today[close] today[ma20]] # 硬过滤20日动量区间防过冷和过热 today today[(today[mom_20] params[min_mom]) (today[mom_20] params[max_mom])] # 软打分不同因子量纲不同这里先用原始值做演示 today[score_trend] today[close] / today[ma20] - 1 today[score_mom] today[mom_20] today[score_liq] np.log(today[amount] 1) # 组合分 today[score] ( today[score_trend] * params[w_trend] today[score_mom] * params[w_mom] today[score_liq] * params[w_liq] ) return today.sort_values(score, ascendingFalse)min_amount要结合你的资金量调。因为amount单位是千元50000代表 5000 万成交额适合资金量大、需要流动性的场景资金量小可以把门槛降到10000但要注意过滤掉更多流动性差的票。mom_20是原始比例比如0.05代表 5%一般取值范围放在-0.1到0.4之间短线策略可以放宽趋势策略要防追高。3.3 基础参数该怎么定下面是这套规则引擎的默认参数你可以把它放到配置文件里。参数默认值作用min_amount50000过滤成交额过低的票min_mom-0.0520日跌幅超过5%的剔除max_mom0.3020日涨幅超过30%的剔除w_trend1.0趋势因子权重w_mom1.0动量因子权重w_liq0.5流动性与可交易性权重这套默认值的问题很明显score_trend和score_mom都是小数值score_liq是 log 量级如果不做标准化流动性会主导总分。我一般会先按历史数据把每个因子做 z-score 或 min-max 归一化再乘权重。调参时不要贪心一次调很多参数建议固定其他参数只动一个用分组对照看结果。4. 自动交易闭环怎么接先做信号推送再谈人工复核自动选股程序的终点不一定必须自动下单。我见过很多项目最后死在自动下单上原因不是接口难而是选股信号和可执行订单之间隔着很多条件当天是否停牌、是否一字涨停、是否 ST、账户是否有足够资金、下单接口是否有权限。所以我的习惯是选股程序只负责出候选清单订单环节由人复核或者由另一个独立交易模块经过风控后再执行。4.1 信号与执行之间的边界选股程序跑出来的是一张信号表包含股票代码、信号生成日期、分数、收盘价等字段。它不等于“明天买这些”。真正的执行需要再判断明天开盘是否高开太多、当天是否出现涨跌停、账户是否有仓位限制。常见做法是每日收盘后跑选股生成一个 CSV 和一条消息通知人看一眼再决定下一步。我把信号和交易放在两个目录里信号表只追加不覆盖订单状态单独记录。这样就算自动下单出问题也能从信号表倒查当时的选股依据。4.2 一个可用的结果输出与消息推送输出文件命名带上日期同一天重复跑覆盖同一份文件不要每次追加否则复盘时看到的全是重复信号。如果你有内部群机器人可以把 Top 信号推过去。import datetime import requests def publish_daily(signal_df, output_dir, webhook_urlNone): today datetime.datetime.now().strftime(%Y-%m-%d) path f{output_dir}/signals_{today}.csv signal_df.head(30).to_csv(path, indexFalse) if webhook_url and len(signal_df) 0: content \n.join( f{row.ts_code} score{row.score:.2f} close{row.close:.2f} for row in signal_df.head(10).itertuples() ) payload {msgtype: text, text: {content: f今日选股Top10\n{content}}} requests.post(webhook_url, jsonpayload, timeout5) return pathwebhook_url是你所在团队内部群机器人提供的地址我用它做每日信号推送。timeout5是防止推送服务响应慢把选股主流程阻塞住。更稳妥的写法是 catch 异常并记日志推送失败不能影响信号文件落地。把分数一起推出来还有一个好处如果某天分数分布突然异常能尽早发现因子量纲或数据更新出了问题。4.3 接模拟交易前先做这三件事第一把历史信号和次日行情交叉验证排除停牌、一字板这类实际买不进的票。做法很简单用信号日 T 选出的股票去查 T1 是否有成交、是否openhighlowclose且量很小。第二建一张订单状态表记录每次信号的处理进度。第三先用模拟账户跑至少两周经历一次下跌后再决定是否提高资金门槛。订单状态表可以长成这样CREATE TABLE order_status ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts_code TEXT, signal_date TEXT, score REAL, status TEXT DEFAULT pending, filled_price REAL, note TEXT, created_at TEXT );status至少要有pending、filled、canceled三种。pending表示信号已产生但还未下单filled表示已成交canceled表示人工驳回或漏单。一张信号进入订单表后选股程序不要再去改它交易执行是另一个独立模块的事不要和选股线程共享变量。5. 避坑指南TuShare限频、复权因子、停牌股与回测陷阱这个方向跑通不难但坑很密集。我把最常见的四个问题按“现象、原因、解决”拆开讲每个都值得记录在你自己项目的排错表里。5.1 接口限频程序跑到一半被锁现象某天扫描全市场跑到第 300 只股票时突然连续报错接口返回类似“请求频率超限”的信息。原因TuShare 不是无限拉每个接口按积分绑定分钟或日频率单线程 for 循环里频繁请求很容易触达上限。解决一是把全量数据落库后离线计算不要在选股过程中高频调接口二是如果确实需要临时补数据用一个限速器每次请求后休眠一小段时间。import time class TuShareThrottle: def __init__(self, min_interval0.2): self.min_interval min_interval self.last_ts 0.0 def wait(self): now time.time() gap now - self.last_ts if gap self.min_interval: time.sleep(self.min_interval - gap) self.last_ts time.time()min_interval0.2表示每次请求至少间隔 200ms。如果接口从未被锁可以调到0.1如果已经被锁过调到0.5甚至1.0。不要设成 0网络波动和接口重试会叠加很快又把频率打满。5.2 复权因子同一个指标结果差一大截现象某只股票曾高比例送转如果不复权20 日涨跌幅可能显示为暴跌换成前复权后趋势信号完全反过来。原因pro.daily返回的是不复权价前复权、后复权会调整历史价格所有均线、涨跌幅、量比指标都会随之变化。解决做趋势和动量指标时用前复权价格做估值和涨跌停判断时用不复权价格。常见做法是用ts.pro_bar(ts_codecode, adjqfq, start_date..., end_date...)单独取前复权日线和原始不复权底表分开保存。df_qfq ts.pro_bar(ts_code000XXX.SZ, adjqfq, start_date20240101, end_date20241231)如果发现pro_bar拿到的前复权价和其他数据源对不上优先检查最近是否有分红除权。复权因子会随新除权事件更新所以历史指标需要定期重算不能把指标快照当永久结论。5.3 停牌股和一子板别把买不进的票选出来现象选股结果里出现一只当天停牌的股票或者当天一字涨停但成交量极低人眼一看就是买不进的票。原因如果用全市场历史数据 join 出每日截面停牌股当天没有成交记录可能会沿用前一天的旧行情一字涨停虽然当天有行情但排队买入成功率很低选股程序不考虑可交易性会高估收益。解决在过滤阶段只保留trade_date 目标日的行确保用的是当天真实数据一字板判断可以用open high low close且vol极低来识别要么剔除要么单独标记成“可观察但不可追”。还要注意新股上市初期波动大常见做法是剔除上市不足 60 天的票用pro.stock_basic()里的list_date做过滤。这一条不处理好回测曲线通常会异常漂亮实盘却跟不住。5.4 回测中的未来函数和偷价现象回测里某一天信号收益特别漂亮但模拟盘怎么都追不上。原因指标计算时用到了当日收盘后才确定的数据却假设能在当日收盘价成交。例如用当日close算 MA再以同一根close买入等于你在知道收盘价的同时立刻成交这是最经典的未来函数。解决信号日 T 只负责产生候选池交易从 T1 开盘价或 T1 收盘价开始评估回测时对比“T 日选出、T1 买入”和“T 日选出、当日偷价买入”的收益差。另一个技巧是按时间把数据切两段前半段调参后半段验证不要在全样本上反复改参数。6. 最后把自动化跑起来定时任务、日志与一个验证技巧如果只写一遍脚本那不叫自动选股叫手动验数据。把程序挂到定时任务上我建议选在每日收盘后的固定时间比如 16:00。用 cron 的话大概是这样0 16 * * 1-5 cd /path/to/our_selector /usr/bin/python3 run_daily.py logs/run_$(date \%Y\%m\%d).log 21cron 表达式里%需要转义成\%否则会被当成换行符如果你用其他调度器只要保证工作目录和环境变量一致就行。日志不是用来炫耀跑成功了而是用来失败时定位。我一般至少记三行开始拉数、拉数完成、选股完成。异常堆栈要单独写到 traceback 文件别和正常日志混在一起。import logging from datetime import datetime logging.basicConfig( levellogging.INFO, filenameflogs/selector_{datetime.now():%Y%m%d}.log, format%(asctime)s %(levelname)s %(message)s ) logger logging.getLogger(selector)最后一个技巧是我每天最愿意花五分钟做的信号回放。每天选完股票后把最近 60 天的信号全部重算一遍统计每天选出的数量、平均分数、次日跑赢或跑输大盘的比例。不用做复杂绩效归因只看两个数信号数量是否稳定次日正收益比例是否稳住。如果今天和三周前选出的股票完全不一样参数很可能过拟合如果每天选出来总是同一批大票说明因子区分度不够。我吃过这个亏后来把所有每日信号追加进一张daily_picks表每周回放一次顺着名单看逻辑是否还能解释。没有这个校验程序跑得再顺也只是把错误重复得更快。希望帮到你。本文还有配套的精品资源点击获取

Python+TuShare构建A股自动选股程序：数据、规则与避坑指南

相关推荐

数学建模种植策略可视化：从决策变量到论文级图表

VSCode 插件工具链配置：把 Base URL 改到 TaoToken 的完整实践

DeepLabv3+实战：PyTorch从零构建可部署语义分割流水线

更多资讯

Mangos服务端改端利器：物品、任务、BOSS编辑器实战详解

VC中集成SQLite：从编译到增删改查的避坑指南

模拟传输课件精讲：从AM到FM的调制主线与教学技巧

关闭不必要的网络服务：提升安全访问与网络稳定性的实操指南

Python抓取A股个股与指数历史数据：复权处理与增量更新实战

5G信令分析实战：从抓包到注册失败定位的排查指南

今日更新

Susi AI API实战指南：susi_alexa_skill如何用Node.js调用chat.json获取智能回答

换新电脑了？KeyStats 恢复码数据找回完全指南，端到端加密统计一键重建

重建二叉树：由前序与中序遍历序列还原二叉树的原理与 Java 实现（YCBlogs 剑指 Offer 精讲）

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？