用Python构建A股量化数据管道：从自动化抓取到信号生成

简介InStock股票系统是专为量化投资者设计的自动化行情工具聚焦A股与ETF的每日数据抓取覆盖个股行情、资金流向、基本面指标、龙虎榜、大宗交易及行业概念板块可有效解决投研中数据获取分散、更新不及时的痛点适合个人量化研究者与数据爱好者部署使用。资源包共178个文件、约4.01MB以82个Python脚本作为采集与自动化核心配合前端展示用的HTML/JS/CSS、定时运行脚本、配置与说明文档构成一套较完整的本地数据平台。当前已有148人学习。通过部署该系统可快速建立个人A股数据仓库跟踪资金动向与机构交易痕迹理解板块轮动特征目录结构清晰、模块划分明确也便于二次开发接入自己的选股策略是入门A股量化数据处理的实用参考。1. 把“找数据”这件事自动化InStock 到底解决什么问题每天收盘后我见过太多个人投资者把大量时间花在“找数据”上打开行情软件看个股涨跌再去翻资金流向最后到晚上才想起来龙虎榜和大宗交易还没看等资料凑齐已经过九点根本没精力想第二天的操作。InStock股票系统这个标题里最关键的词不是“量化”而是“自动化”——它把每日抓取A股市场股票与ETF的关键数据做成了流水线个股行情、资金流向、基本面指标、龙虎榜、大宗交易、行业与概念板块全部定时抓取、落库、清洗再在统一的数据表上跑信号。它适合两类人一类是有 Python 基础、不想花几万块买终端、想自己搭数据管道的开发者另一类是已经厌倦手工整理数据、希望用规则生成候选池的投资者。它不预测涨跌只负责把“数据获取”这个最脏最累的环节固定下来。2. 数据抓取层怎么搭日频行情、资金流与龙虎榜的采集架构这一章先讲数据源选型再给爬虫最小实现最后落到存储。搭这套抓取层时我最关心的三件事数据源稳不稳定、抓取代码断网后能不能自我恢复、存储结构在数据量涨到几百万行之后还能不能查得快。后面所有信号计算都建立在这一层上所以这一层不能图省事。2.1 数据源选型为什么优先走免费公开接口而不是买终端个人做量化最现实的门槛是数据成本。Wind 终端一年几万Choice 也要账号费而且桌面终端的数据导出接口基本是为人工操作设计的长时间无人值守抓取很容易被限制。InStock 这类项目把数据源放在财经门户的公开 HTTP 接口上是合理选择零费用、响应快、字段覆盖日线行情与资金流足够支撑日频量化。缺点是接口文档不公开、字段名随时可能变、对访问频率有隐性限制。所以我搭数据源时通常做两手准备主源抓行情和资金流备源只抓三个字段——close、volume、pct_change。万一主源改版备源还能保证当日数据不至于断档。不要一上来就接入多家付费数据商先把一条免费链路跑通再在它上面做增量补偿。备源的价值在自动化任务里会被放大主源凌晨 1 点维护备用源可能在凌晨 2 点还能正常返回。数据管道里“最后一条可用路径”远比“字段最全的路径”重要。2.2 采集器最小实现用 requests.Session 保证每天稳定拉数据抓取层最常见的翻车点不是接口参数错了而是网络抖动和频率限制。为此我给采集函数套一个带退避的重试装饰器把“请求失败→等待→重试”的逻辑抽出来避免每个业务函数重复写 try except。import requests import time import random from functools import wraps UA Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 def retry_on_failure(max_retries3, base_delay2): 给抓取函数加上指数退避重试。 def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(1, max_retries 1): try: return func(*args, **kwargs) except (requests.Timeout, requests.ConnectionError, ValueError): if attempt max_retries: raise sleep_sec base_delay * attempt random.random() time.sleep(sleep_sec) return None return wrapper return decorator retry_on_failure(max_retries3, base_delay2) def fetch_json(session, url, paramsNone): resp session.get(url, paramsparams, timeout10) resp.raise_for_status() return resp.json()这段代码里有两个关键设置。第一个是timeout10超过 10 秒直接判定失败避免某个接口长时间卡住拖垮整批任务。第二个是退避公式base_delay * attempt random.random()第一次失败等 2 秒第二次等 4 秒第三次等 6 秒左右随机数防止多只标的的重试请求在同一个时间点砸向服务器。调用时用同一个 Session 对象连接复用能明显降低被风控盯上的概率def daily_job(all_codes): session requests.Session() # 带 Referer 是为了模拟从页面发起的正常请求降低被识别为脚本的概率。 session.headers.update({User-Agent: UA, Referer: https://quote.example.com/}) for code in all_codes: params {code: code, fields: open,high,low,close,volume,amount} data fetch_json(session, https://quote.example.com/api/stock, paramsparams) # 每只股票间隔 0.2~0.5 秒全市场 5000 只标的约半小时跑完。 time.sleep(random.uniform(0.2, 0.5))间隔的设定逻辑是如果按 5 个请求每秒跑5000 只标的约 17 分钟跑完全市场加上 0.2~0.5 秒随机间隔时间翻倍但在可接受范围。这个节奏对大多数免费接口来说算安全。如果目标只有沪深 300 成分股间隔可以放宽到 0.5 秒以上没必要抢时间。2.3 存储与更新策略先用 CSV 落地再平滑迁移到 MySQL数据抓下来之后存储结构直接影响后续开发效率。A 股加 ETF 大约 5500 个标的日频数据一年约 140 万行五年约 700 万行。这个量级用 CSV 完全能撑住一开始不用急着上数据库。from pathlib import Path import pandas as pd base_dir Path(./data) date_dir base_dir / 2025-01-15 date_dir.mkdir(parentsTrue, exist_okTrue) # quote_df 是当天清洗后的行情表按日期分目录回看某一天直接读对应文件。 quote_df.to_csv(date_dir / daily_quote.csv, indexFalse)按交易日分目录的好处是增量更新天然是“只写当天文件”不会误碰历史数据排查问题时也能直接看某个日期目录是否存在。缺点是跨日期查询需要拼文件所以在数据量上来之后建议迁移到 MySQL。CREATE TABLE IF NOT EXISTS daily_quote ( trade_date DATE NOT NULL, code VARCHAR(10) NOT NULL, name VARCHAR(20) DEFAULT NULL, open DECIMAL(10,2) DEFAULT NULL, high DECIMAL(10,2) DEFAULT NULL, low DECIMAL(10,2) DEFAULT NULL, close DECIMAL(10,2) DEFAULT NULL, volume BIGINT DEFAULT NULL, amount DECIMAL(20,2) DEFAULT NULL, pct_change DECIMAL(6,2) DEFAULT NULL, PRIMARY KEY (trade_date, code), KEY idx_code_trade (code, trade_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;这张表的设计有一个容易被忽略的点主键用(trade_date, code)而不是自增 id。复合主键保证同一个交易日同一只股票只会存在一行重复抓取时执行INSERT ... ON DUPLICATE KEY UPDATE就能幂等更新不会产生脏数据。索引(code, trade_date)则是为“取某只股票最近 N 天行情”这类高频查询设计的。从 CSV 迁到 MySQL 时我一般写一个同步脚本按日期目录逐批读入用LOAD DATA LOCAL INFILE导入导入后立刻跑一行 COUNT 对比两边行数。不要把整个历史数据一次性塞进去一旦中间断掉很难定位断点。3. 核心数据模块拆解从原始接口到可用的量化底表数据抓到本地只是第一步。接口返回的原始字段往往不是我们想要的样子代码格式不统一、资金流是嵌套结构、龙虎榜按榜单而不是按单只股票组织。这一章要把这些原始数据归一成能直接参与计算的底表。底表设计得好不好决定后面写信号时是几行代码还是一大堆 if else。3.1 股票与 ETF 的统一代码模型先解决撞码再去谈数据上证和深证都存在 6 位数字代码不做处理会撞车。比如 5 开头的沪市 ETF 和 1 开头的深市基金如果只存 6 位数字分不清归属交易所600 和 000 开头的代码倒是能从首位数区分但依赖人眼判断不能写进程序。抓回来的数据源有的给.SH后缀有的只给纯数字第一步必须做标准化。def normalize_code(raw) - str: 把不同来源的证券代码统一为 6 位代码.交易所 格式。 code str(raw).strip().upper() if . in code: return code if len(code) ! 6 or not code.isdigit(): raise ValueError(f无法识别代码: {raw}) if code.startswith(6): return f{code}.SH # 沪市主板、科创板均以 6 开头 if code.startswith((0, 3)): return f{code}.SZ # 深市主板 0 开头创业板 3 开头 if code.startswith(5): return f{code}.SH # 沪市 ETF/LOF if code.startswith(1): return f{code}.SZ # 深市基金 if code.startswith((4, 8)): return f{code}.BJ # 北交所 raise ValueError(f未知代码段: {raw})这段代码的判断顺序不能乱先处理带后缀的字符串再判断纯数字。startswith((0, 3))用元组参数一次完成两个前缀匹配比连续两个 if 干净。注意北交所在旧数据源里可能是 8 开头纳入后要单独判断不要让它落到默认分支。统一代码后所有下游代码都按600000.SH这种格式关联避免在 join 时出现“一边是 600000另一边是 600000.SH”的尴尬。3.2 资金流向与龙虎榜的清洗把嵌套 JSON 拍平成行资金流接口返回的往往是树状结构外层是按日期组织的数据内层又分主力、超大单、大单、中单、小单多个子对象。如果不拍平pandas 读进来会得到一堆 dict 嵌套列后续筛选没法写。def flatten_flow(raw_item): 把资金流接口返回的嵌套结构转成一行底表记录。 main raw_item.get(main, {}) or {} super_ raw_item.get(super, {}) or {} row { code: normalize_code(raw_item[code]), trade_date: raw_item[date], main_net_inflow: main.get(net_inflow), main_net_inflow_pct: main.get(net_inflow_pct), super_net_inflow: super_.get(net_inflow), huge_net_inflow: raw_item.get(huge, {}).get(net_inflow), retail_net_inflow: raw_item.get(retail, {}).get(net_inflow), } return row这里使用了dict.get而不是dict[net_inflow]访问因为资金流数据经常出现某个子结构缺失——新股上市第一天没有历史资金流部分北交所标的主力字段一直为空。get方法在字段缺失时返回 None不会中断整批清洗。龙虎榜是另一个典型清洗场景。交易所每天收盘后才公布当日上榜名单返回的 JSON 往往是一个榜单包含多个股票列表。要把它拆成“日期 股票 买入卖出金额”的明细行def clean_lhb(raw_rows): 把龙虎榜按榜单组织的原始数据拆成按股票维度组织的明细。 records [] for r in raw_rows: for stock in r.get(stock_list, []): records.append({ trade_date: r[trade_date], code: normalize_code(stock[code]), buy_amount: stock.get(buy_amount), sell_amount: stock.get(sell_amount), net_buy: stock.get(net_buy), reason: stock.get(reason), }) return pd.DataFrame(records)注意reason字段龙虎榜上榜原因比如“日涨幅偏离值达 7%”是后续做事件驱动信号的重要标签但同一只股票可能同时因为多个原因上榜导致出现在同一个榜单的多行里。清洗阶段不要做去重保留原始一对多关系信号层再按需求聚合。3.3 基本面指标与行业板块映射筛选器的原料从哪里来行情和资金流是日频数据基本面指标却不一样市盈率、市净率、ROE 大多按季度更新不需要每天抓。把这个认识写进存储模型能省一大半抓取量。我一般单独建两张表一张存股票基本信息与最新基本面快照一张存概念板块映射。CREATE TABLE stock_basic ( code VARCHAR(10) PRIMARY KEY, name VARCHAR(20) NOT NULL, industry VARCHAR(30), pe DECIMAL(10,2), pb DECIMAL(10,2), roe DECIMAL(6,2), update_date DATE ); CREATE TABLE stock_concept ( code VARCHAR(10) NOT NULL, concept VARCHAR(30) NOT NULL, PRIMARY KEY (code, concept) );行业字段在stock_basic里概念映射单独成表是因为两者语义不同行业是互斥的一只股票属于且只属于一个行业分类概念是叠加的一只股票可以同时属于“人工智能”“算力”“信创”多个概念。把它们放同一张表会导致更新逻辑复杂还会在信号层产生重复计数。基本面快照表则只需要保留最近一版每天抓取后REPLACE INTO覆盖即可没必要保留历史版本除非你要做基本面变化分析。4. 用清洗后的数据生成交易信号动量、资金流与龙虎榜的交叉验证底表建好之后写信号就是顺理成章的事。我常用的一个组合是“20 日动量 主力净流入为正 龙虎榜净买入为正”。这三个条件分别代表趋势、资金和事件三重确认单独用哪一个都不稳定动量追高风险大资金流有滞后龙虎榜覆盖范围太小。4.1 信号计算的核心逻辑20 日动量叠加主力净流入动量因子用 pandas 的 groupby 加 rolling 实现注意要先排序再计算否则前复权数据错位会让结果完全不可信。import pandas as pd def add_momentum(df: pd.DataFrame, window: int 20) - pd.DataFrame: 在日线数据上追加 20 日动量字段。 df df.sort_values([code, trade_date]).reset_index(dropTrue) df[ret] df.groupby(code)[close].pct_change() df[momentum] df.groupby(code)[ret].rolling(window).apply( lambda x: (1 x).prod() - 1, rawTrue ).reset_index(level0, dropTrue) return df.drop(columns[ret])这段代码里最容易写错的是reset_index(level0, dropTrue)。groupby 后接 rolling结果索引是“code 原始行号”两层结构不加这一步新字段的长度和原 DataFrame 会对不上。rolling(window)计算的是最近 20 个交易日的收益率乘积也就是区间累计涨幅而不是简单平均这个区别在动量语境里很关键。参数window我习惯用 20对应约一个自然月。A 股短周期轮动快5 日和 10 日动量噪音太大60 日动量又太钝等信号出来趋势差不多结束了。如果做的是 ETF 轮动窗口可以放宽到 60因为 ETF 的成分股分散短期动量不如个股敏感。4.2 每日自动输出候选池从数据到可操作清单的脚本动量和资金流算好后把它们跟龙虎榜合并输出候选池。这一步不需要复杂模型pandas 的 merge 加布尔筛选就够。把“主力净流入为正”和“龙虎榜净买入为正”同时作为条件过滤后的股票数量通常会从 5000 多只降到 30 到 100 只正好进入人工复核流程。def select_pool(quote_df, flow_df, lhb_df, date): 把当日动量、资金流和龙虎榜数据合到一起生成候选池。 merged quote_df.merge( flow_df[[code, main_net_inflow_pct]], oncode, howleft ) merged[on_lhb] merged[code].isin( lhb_df.loc[lhb_df[trade_date] date, code] ) pool merged[ (merged[momentum] 0.08) (merged[main_net_inflow_pct] 0) (merged[on_lhb]) ] return pool.sort_values(momentum, ascendingFalse)howleft表示以行情表为主表资金流缺失的股票保留下来但填充 NaN这样不会因为某一天资金流接口漏了一条数据就把整只股票从底表里剔除。on_lhb用 isin 判断比 inner join 更直观。momentum 0.08这个阈值不是拍脑袋定的是回测出来的过去三年满足这个阈值的股票次日常规收益分布最稳定。你可以先按 0.05 跑一个月观察数据分布再调不要一开始就追求精确阈值。候选池生成后我会再用基本面表过滤一次把 PE 为负、ST 状态的股票剔除。基本面表是季度更新过滤条件不宜设太紧否则候选池经常整个月都空着。5. InStock 落地避坑抓取限制、复权错位与数据不一致的五条记录这一章写的每一条都是我自己踩过的坑。数据管道跑起来不难难的是它连续跑三个月不出乱子。以下五条按频率排序基本覆盖了日频量化工具最常见的故障模式。5.1 抓取频率太高被封 IP不要在收盘一小时内并发拉全市场现象系统运行到第三到五天某天早上开始出现大量 403 响应再往后同一 IP 下任何接口都返回验证码页面。由于采集脚本只记录成功数据那天的行情表缺了 200 多只股票直到两天后做完整性校验才发现。原因收盘后所有人的定时任务都集中在 15:30 到 16:30 启动我的脚本在这个时段用 16 个线程并发拉全市场单秒请求数超过数据源风控阈值触发封禁。解决把并发改成单线程循环每只股票间隔 0.2 秒到 0.5 秒并把任务执行时间从 15:30 改到 17:00 以后避开高峰。封禁解除后在抓取函数里加了失败告警连续 5 次 403 就发邮件而不是默默跳过。另外免费接口的风控通常集中在“单 IP 请求频率”和“请求特征”两个维度。UA 和 Referer 是常数容易被识别成脚本。我的做法是准备两个用户代理轮流使用同时在凌晨跑一次小批量验证任务看 IP 是否被标记。5.2 前复权和后复权混用同一个 close 字段引发的信号失真现象某只股票在回测里走势跟行情软件对不上尤其是 2020 年以前的数据close 序列的涨跌幅明显偏大导致动量信号在除权日附近频繁误报。原因免费接口的复权数据有两种口径前复权随最新价变动历史数据会不断重算后复权相对稳定适合回测。但同一个接口在不同时间段抓下来的前复权数据基准不同日期越早偏差越大。我的存储表里只有close一个字段没记录复权类型清洗时把两种口径混在一起了。解决确定一个主口径我在日频信号里统一用后复权价计算动量用原始不复权价计算涨跌幅和成交量。这两类数据分开存储分别命名close_adj和close_raw。如果只想存一个字段就选后复权并定期全量重算一次保证整个历史区间口径一致。反过来说如果要计算当日涨跌幅必须用不复权价否则除权日当天会算出“跳空下跌”或“跳空上涨”的假信号。5.3 龙虎榜日期和收盘日期错位夜盘数据落到哪一天很重要现象每天早上看候选池发现不少股票的“龙虎榜净买入”跟当天涨幅对不上。比如 4 月 10 日早上生成候选池时标记某股票 4 月 9 日上了龙虎榜但这只股票 4 月 9 日的行情实际还没收盘。原因龙虎榜按交易所规则是 T 日收盘后公布 T 日上榜名单但部分数据提供方在晚上 10 点前返回的数据用的是 T-1 日榜单也有少数接口返回的 JSON 里日期字段是“公布日期”而不是“交易日期”。全链路没统一基准信号层拿到的是两个不同日期的数据。解决在清洗层强制统一龙虎榜数据一律以trade_date字段作为归属日期落库后写一个校验任务比对龙虎榜最大日期与行情表最大日期相差超过 1 天直接告警。信号层关联时永远用trade_date关联不允许用“导入时间”或“抓取时间”。5.4 增量更新覆盖历史数据缺了主键约束就会静默翻车现象某天检查数据库发现总行数从 500 万跌到 300 万最初以为是存储问题后来发现是更新逻辑写错了。原因早期用“先删后插”策略做增量更新删除条件只写了code ?没有限定trade_date ?。某只股票某天数据重跑时把该股票的全部历史记录删掉了再插入只有当天一行。解决两条硬措施。第一条是表结构加复合主键(trade_date, code)让重复插入直接报错或走ON DUPLICATE KEY UPDATE从机制上堵住误删。第二条是废弃“先删后插”改成按日期分片写入只操作trade_date 目标日期的记录。执行完立刻对比当天行数与源文件行数不一致就报警。如果表已经建好且没有主键可以先把重复数据清掉再补主键有主键约束后很多误操作在第一步就会被数据库拦住而不是等几天后数据对不上才发现。5.5 数据库连接池耗尽定时任务为什么会半夜失联现象连续几天凌晨的定时任务都失败但白天手动跑同样的脚本完全正常。日志里出现Too many connectionsMySQL 拒绝新连接。原因采集脚本里每次调用数据库都pymysql.connect新建连接用完没关。白天任务少连接数没到上限凌晨任务一启动同时并发几十个连接加上之前的连接没释放直接打满 MySQL 默认的 151 连接数。更隐蔽的是程序异常退出时finally里没有close()连接被占用直到超时。解决用连接池管理数据库会话把连接数上限控制在 20 以内。pymysql本身没有连接池实现我用DBUtils.PooledDB包了一层设置maxconnections20每次从池里借连接用完归还。同时给所有数据库操作套上了try/finally确保即使查询报错也能归还连接。from dbutils.pooled_db import PooledDB import pymysql pool PooledDB( creatorpymysql, maxconnections20, host127.0.0.1, userquant, passwordyour_password, databasestock, charsetutf8mb4, ) def query_one(sql, argsNone): conn pool.connection() try: with conn.cursor() as cur: cur.execute(sql, args) return cur.fetchone() finally: conn.close() # 归还连接而不是真的关闭这段代码里容易混淆的是conn.close()在PooledDB语义下close()不是关闭底层连接而是把连接还给池子。如果用原生pymysql写同样的代码close()就是真的断开连接。所以连接池的封装必须统一入口不要让部分代码走原生连接、部分走池子两套混用会把连接数逻辑彻底搞乱。6. 进阶验证技巧用一致性校验和回放测试让信号靠谱一点6.1 每日抓取后先做三行校验自动化跑一段时间后比抓取更重要的反而是验证。我每天在落库后跑一段简单的覆盖度检查用“预期标的集合”对比“实际抓到的标的集合”缺了就告警。def check_coverage(expected, actual, date): miss expected - actual extra actual - expected if miss: print(f[{date}] 缺失 {len(miss)} 个标的前 5 个: {sorted(miss)[:5]}) if extra: print(f[{date}] 多出 {len(extra)} 个标的可能代码映射有误) return not miss and not extra预期标的集合来自上一交易日收盘时的全市场股票快照加当日新上市列表。如果缺失数量突然从个位数变成几十个大概率不是网络原因而是代码规则变动。这个校验跑完我才会信任当天生成的候选池。6.2 用回放确认信号不是过拟合参数调多了难免过拟合我的自我检查方式是把当前参数放回六个月的历史数据里重跑一遍看信号池的平均表现是否稳定。如果 20 日动量阈值设为 8% 在近三个月有效但五个月前同一批股票平均收益为负说明当前参数大概率在拟合近期行情。回放不追求精确收益只看分布。我只需要知道满足条件的股票数量是否稳定、次日平均涨跌幅是否为正、最大回撤是否超出预期。任何一天满足条件的股票少于 5 只我都会把阈值调低再观察而不是手动往池子里塞股票。信号的稳定性来源于规则对噪声的不敏感而不是规则足够巧妙。我习惯每天把覆盖率数字直接打在任务日志第一行出现任何异常第一眼就能看到。数据管道这种系统不怕慢就怕坏得无声无息把校验做成强制步骤之后我的信号质量明显稳了一截。希望这套从抓取到校验的工程思路对你有帮助。本文还有配套的精品资源点击获取