企业官网建设流程全解析

简介：面向量化投资入门者与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_code=ts_code, start_date=start_date, end_date=end_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_date=None): 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(days=1)).strftime('%Y%m%d') return fetch_daily_and_save(conn, ts_code, start_date, end_date)

不要直接把latest + 1 天当成请求起始日去调用接口，因为如果最新数据是周五，那么周六日没有行情，得先过滤交易日历。常见做法是再查一遍pro.trade_cal(exchange='SSE', start_date=start_date, end_date=end_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', ascending=False)

min_amount要结合你的资金量调。因为amount单位是千元，50000代表 5000 万成交额，适合资金量大、需要流动性的场景；资金量小可以把门槛降到10000，但要注意过滤掉更多流动性差的票。mom_20是原始比例，比如0.05代表 5%，一般取值范围放在-0.1到0.4之间，短线策略可以放宽，趋势策略要防追高。

3.3 基础参数该怎么定

下面是这套规则引擎的默认参数，你可以把它放到配置文件里。

这套默认值的问题很明显：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_url=None): today = datetime.datetime.now().strftime('%Y-%m-%d') path = f"{output_dir}/signals_{today}.csv" signal_df.head(30).to_csv(path, index=False) 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, json=payload, timeout=5) return path

webhook_url是你所在团队内部群机器人提供的地址；我用它做每日信号推送。timeout=5是防止推送服务响应慢，把选股主流程阻塞住。更稳妥的写法是 catch 异常并记日志，推送失败不能影响信号文件落地。把分数一起推出来还有一个好处：如果某天分数分布突然异常，能尽早发现因子量纲或数据更新出了问题。

4.3 接模拟交易前先做这三件事

第一，把历史信号和次日行情交叉验证，排除停牌、一字板这类实际买不进的票。做法很简单：用信号日 T 选出的股票，去查 T+1 是否有成交、是否open==high==low==close且量很小。第二，建一张订单状态表，记录每次信号的处理进度。第三，先用模拟账户跑至少两周，经历一次下跌后再决定是否提高资金门槛。

订单状态表可以长成这样：

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_interval=0.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_interval=0.2表示每次请求至少间隔 200ms。如果接口从未被锁，可以调到0.1；如果已经被锁过，调到0.5甚至1.0。不要设成 0，网络波动和接口重试会叠加，很快又把频率打满。

5.2 复权因子：同一个指标结果差一大截

现象：某只股票曾高比例送转，如果不复权，20 日涨跌幅可能显示为暴跌；换成前复权后，趋势信号完全反过来。原因：pro.daily返回的是不复权价，前复权、后复权会调整历史价格，所有均线、涨跌幅、量比指标都会随之变化。解决：做趋势和动量指标时用前复权价格，做估值和涨跌停判断时用不复权价格。常见做法是用ts.pro_bar(ts_code=code, adj='qfq', start_date=..., end_date=...)单独取前复权日线，和原始不复权底表分开保存。

df_qfq = ts.pro_bar(ts_code='000XXX.SZ', adj='qfq', start_date='20240101', end_date='20241231')

如果发现pro_bar拿到的前复权价和其他数据源对不上，优先检查最近是否有分红除权。复权因子会随新除权事件更新，所以历史指标需要定期重算，不能把指标快照当永久结论。

5.3 停牌股和一子板：别把买不进的票选出来

现象：选股结果里出现一只当天停牌的股票，或者当天一字涨停但成交量极低，人眼一看就是买不进的票。原因：如果用全市场历史数据 join 出每日截面，停牌股当天没有成交记录，可能会沿用前一天的旧行情；一字涨停虽然当天有行情，但排队买入成功率很低，选股程序不考虑可交易性会高估收益。解决：在过滤阶段只保留trade_date == 目标日的行，确保用的是当天真实数据；一字板判断可以用open == high == low == close且vol极低来识别，要么剔除，要么单独标记成“可观察但不可追”。

还要注意新股上市初期波动大，常见做法是剔除上市不足 60 天的票，用pro.stock_basic()里的list_date做过滤。这一条不处理好，回测曲线通常会异常漂亮，实盘却跟不住。

5.4 回测中的未来函数和偷价

现象：回测里某一天信号收益特别漂亮，但模拟盘怎么都追不上。原因：指标计算时用到了当日收盘后才确定的数据，却假设能在当日收盘价成交。例如用当日close算 MA，再以同一根close买入，等于你在知道收盘价的同时立刻成交，这是最经典的未来函数。解决：信号日 T 只负责产生候选池，交易从 T+1 开盘价或 T+1 收盘价开始评估；回测时对比“T 日选出、T+1 买入”和“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 2>&1

cron 表达式里%需要转义成\%，否则会被当成换行符；如果你用其他调度器，只要保证工作目录和环境变量一致就行。日志不是用来炫耀跑成功了，而是用来失败时定位。我一般至少记三行：开始拉数、拉数完成、选股完成。异常堆栈要单独写到 traceback 文件，别和正常日志混在一起。

import logging from datetime import datetime logging.basicConfig( level=logging.INFO, filename=f"logs/selector_{datetime.now():%Y%m%d}.log", format="%(asctime)s %(levelname)s %(message)s" ) logger = logging.getLogger("selector")

最后一个技巧，是我每天最愿意花五分钟做的：信号回放。每天选完股票后，把最近 60 天的信号全部重算一遍，统计每天选出的数量、平均分数、次日跑赢或跑输大盘的比例。不用做复杂绩效归因，只看两个数：信号数量是否稳定，次日正收益比例是否稳住。如果今天和三周前选出的股票完全不一样，参数很可能过拟合；如果每天选出来总是同一批大票，说明因子区分度不够。我吃过这个亏，后来把所有每日信号追加进一张daily_picks表，每周回放一次，顺着名单看逻辑是否还能解释。没有这个校验，程序跑得再顺，也只是把错误重复得更快。希望帮到你。

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