1. 数据抓取这件事为什么值得认真做一遍做量化研究、写复盘工具、跑回测策略甚至只是想在本地建一个自己的行情数据库第一步永远绕不开同一件事把历史数据拿到手。A股市场的数据获取渠道看起来不少但真正用起来就会发现各有各的坑——有的接口限流严重有的字段残缺不全有的复权处理含糊不清还有的用着用着就改了规则。我前后折腾过好几套方案最后沉淀下来的还是基于 Python 自己抓取这一条路原因很简单可控。这篇文章聊的就是怎么用 Python 把 A 股个股和指数的历史数据抓下来存成自己能随时调用的格式。核心关键词就三个个股历史数据、指数历史数据、Python 抓取。适合谁看如果你会一点 Python 基础语法知道什么是列表和字典但还没系统做过数据抓取那这篇基本可以照着抄作业如果你已经用过一些现成的数据接口但总觉得数据质量不稳定、想自己掌控全流程那这里面关于复权、字段校验、增量更新的部分应该对你有用。需要先说明一点数据抓取本身是个中性技术活关键在于你抓什么、怎么用、是否遵守数据来源方的使用条款。本文讨论的全部是公开市场行情数据用于个人研究学习不涉及任何交易建议也不涉及任何敏感内容。所有代码示例里的接口地址、参数命名都以通用实践为准实际使用时请以你选用的数据源官方文档为准。我自己的使用场景比较典型本地维护一个 SQLite 数据库每天收盘后跑一次增量更新周末做一次全量校验。个股大概几千只指数几十个全量拉一遍历史数据在合理配置下大概十几分钟。下面把这套流程拆开讲。2. 整体方案设计与技术选型思路2.1 为什么优先考虑轻量级方案而不是重型框架很多人一上来就想搭一套分布式抓取系统用上消息队列、多节点调度、容器编排。我的建议是先别。个人研究场景下数据量其实没有想象中那么大。A 股个股按 5000 只算每只股票 20 年日线数据大约 5000 条记录全量也就 2500 万条左右这个量级用单机 SQLite 或者单机 PostgreSQL 完全扛得住。上重型框架带来的运维成本、调试成本远大于它带来的性能收益。所以我的方案骨架是这样的数据获取层用requests或httpx发 HTTP 请求拿到 JSON 或 CSV 格式的原始数据数据解析层用pandas做结构化处理统一字段名、统一日期格式、统一数值类型数据存储层SQLite 做本地存储按“个股日线”“指数日线”分表调度层一个主脚本 配置文件支持全量初始化和增量更新两种模式这个架构的好处是每一层职责清晰出问题容易定位。比如数据对不上先看是请求层拿到的原始数据有问题还是解析层字段映射错了还是存储层写入时类型转换出了岔子。2.2 数据源的选择逻辑与取舍数据源这块我不点名具体平台只讲选择逻辑。你需要关注几个维度维度说明为什么重要字段完整性是否包含开高低收、成交量、成交额、复权因子缺字段会导致回测失真历史深度能回溯到多久以前长周期策略需要足够历史更新时效收盘后多久能拿到当日数据影响增量更新调度时间请求限制单位时间内的请求次数上限决定抓取节奏和并发策略复权支持是否提供前复权/后复权数据或因子除权除息处理的核心我一般会同时准备两个数据源做交叉校验。主源用来日常抓取备源用来抽样比对。如果两边对同一只股票同一天的收盘价差异超过千分之一就触发人工检查。这个习惯帮我发现过好几次数据源本身的字段错位问题。2.3 复权处理绕不过去的核心问题这是新手最容易踩坑的地方。股票除权除息之后价格会出现跳空如果不做复权处理直接拿原始价格算收益率会得到完全错误的结果。举个例子某股票10送10除权日价格从20元变成10元如果你不做处理系统会认为当天跌了50%但实际上股东权益没有变化。复权分两种前复权以当前价格为基准向前调整历史价格。优点是当前价格和实际盘面一致适合看盘和短线分析。缺点是每次除权后整个历史序列都会变不适合做需要稳定历史数据的回测。后复权以上市首日为基准向后调整。优点是历史数据一旦生成就不再变化适合回测。缺点是当前价格和盘面不一致。我的做法是存储原始行情 复权因子在需要的时候动态计算前复权或后复权价格。这样既保留了原始数据又能灵活切换复权方式。复权因子的计算逻辑是前复权价 原始价 × (当前复权因子 / 该日复权因子) 后复权价 原始价 × (该日复权因子 / 首日复权因子)这个公式看起来简单但实际实现时要注意复权因子的精度和累积误差。我一般保留小数点后6位并且在每次除权事件后重新校验一遍。3. 核心细节解析与实操要点3.1 请求层的稳定性设计请求层看起来就是发个 HTTP 请求但实际写起来要考虑的事情不少。我总结了几条硬性要求第一必须设置超时。不设超时的请求在弱网环境下会挂死整个脚本卡住。我一般设连接超时5秒读取超时15秒。第二必须处理重试。网络抖动是常态一次失败不代表接口有问题。用指数退避策略第一次等1秒第二次等2秒第三次等4秒最多重试3次。第三必须控制频率。不管数据源有没有明说限流自己都要有节制。我一般控制在每秒2到5个请求具体看数据源的表现。太快容易被封太慢抓不完。第四必须伪装成正常浏览器。设置合理的 User-Agent 和 Referer这是基本的礼貌也能避免很多莫名其妙的拒绝。一个典型的请求函数长这样import requests import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(): session requests.Session() retry Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504] ) adapter HTTPAdapter(max_retriesretry) session.mount(http://, adapter) session.mount(https://, adapter) session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Referer: https://example.com/ }) return session注意status_forcelist里加上 429 是因为很多数据源用这个状态码表示请求过频。遇到 429 不要硬刚退避时间要拉长。3.2 数据解析的字段标准化不同数据源返回的字段名五花八门有的叫trade_date有的叫date有的叫日期。如果不做标准化后面每换一个数据源就要改一遍下游代码。我的做法是定义一个统一的字段映射表标准字段名含义常见别名trade_date交易日期date, 日期, dtopen开盘价open_price, 开盘high最高价high_price, 最高low最低价low_price, 最低close收盘价close_price, 收盘volume成交量vol, 成交量amount成交额turnover, 成交额adj_factor复权因子factor, 复权因子解析的时候统一转成标准字段名日期统一转成YYYY-MM-DD格式数值统一转成float类型。这一步做完下游代码就再也不用关心数据源是谁了。还有一个细节停牌日的处理。停牌当天没有交易数据但有些数据源会返回一条全零记录有些直接跳过。我的做法是统一跳过不补零。因为补零会在计算收益率时产生 -100% 的假信号。3.3 存储层的表结构设计SQLite 建表语句我一般这么写CREATE TABLE IF NOT EXISTS stock_daily ( ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, amount REAL, adj_factor REAL, PRIMARY KEY (ts_code, trade_date) ); CREATE TABLE IF NOT EXISTS index_daily ( ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, amount REAL, PRIMARY KEY (ts_code, trade_date) );用(ts_code, trade_date)做联合主键天然支持INSERT OR REPLACE的幂等写入。这样重复抓取同一天的数据不会产生重复记录增量更新的时候特别省心。索引方面除了主键自带的索引我还会在trade_date上单独建一个索引方便按日期范围查询。个股表数据量大按日期查的场景很多这个索引不能省。3.4 增量更新的判断逻辑全量抓取只在初始化时做一次之后都是增量更新。增量更新的核心是怎么知道哪些数据需要更新。我的逻辑是查询本地数据库中每只股票的最新交易日期从数据源获取该股票的最新交易日期如果本地最新日期 数据源最新日期则抓取缺失区间的数据如果本地没有该股票记录则做全量抓取这里有个坑新上市的股票和退市的股票。新上市的股票本地没有记录走全量逻辑没问题。退市的股票数据源可能不再更新但本地已有数据要保留不能因为查不到就删掉。还有一个坑数据源的“最新交易日”可能不准。有些数据源在收盘后一段时间内返回的最新日期还是前一天。我的做法是加一个校验如果当前时间是交易日收盘后2小时以内允许数据源最新日期比实际最新交易日少一天不报错。3.5 并发策略与资源控制单线程抓取太慢全量初始化可能要几个小时。用并发是必然的但并发度要控制好。我的经验值数据源明确限流的并发度设为1到2靠重试和退避来保证成功率数据源没有明确限流的并发度设为4到8用concurrent.futures.ThreadPoolExecutor控制不管什么情况都要加一个全局的请求间隔比如每个请求之间至少间隔0.2秒用线程池的示例from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_all_stocks(stock_list, max_workers4): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(fetch_one_stock, code): code for code in stock_list } for future in as_completed(future_map): code future_map[future] try: data future.result() results.append((code, data)) except Exception as e: print(f{code} 抓取失败: {e}) return results注意线程池的max_workers不是越大越好。我试过设到16结果数据源直接返回429反而拖慢了整体进度。4到8是比较稳妥的区间。4. 完整实操流程与关键环节实现4.1 环境准备与依赖安装先把环境搭起来。Python 版本建议3.8以上太老的版本有些库不支持。pip install requests pandas numpy sqlite3sqlite3是 Python 标准库自带的不用单独装。如果你打算用 PostgreSQL把sqlite3换成psycopg2就行。目录结构我一般这么组织project/ ├── config.py # 配置文件 ├── fetcher.py # 请求层 ├── parser.py # 解析层 ├── storage.py # 存储层 ├── updater.py # 调度层 └── data/ └── market.db # SQLite 数据库文件分层的好处是每部分可以单独测试。比如数据源改了返回格式只需要改parser.py其他部分不动。4.2 个股列表的获取与维护抓个股历史数据之前得先知道有哪些股票。个股列表本身也是会变的——新股上市、老股退市、代码变更。我的做法是每周更新一次个股列表存到数据库里。个股列表需要包含的字段股票代码带市场后缀如.SH、.SZ股票名称上市日期退市日期未退市为空当前状态上市/退市/暂停上市有了上市日期和退市日期抓历史数据的时候就能确定每只股票的有效时间范围避免请求不存在的数据。4.3 单只股票历史数据的完整抓取流程以一只股票为例走一遍完整流程第一步确定抓取区间。从上市日期到当前日期或者从本地最新日期到当前日期。第二步构造请求参数。一般需要股票代码、起始日期、结束日期、复权类型这几个参数。第三步发送请求并处理响应。拿到 JSON 后先检查返回状态确认数据有效再解析。第四步解析并标准化字段。按前面说的映射表统一字段名和类型。第五步写入数据库。用INSERT OR REPLACE保证幂等。第六步记录抓取日志。记录股票代码、抓取区间、记录条数、耗时、是否成功。这个日志在排查问题时非常有用。代码骨架def fetch_one_stock(ts_code, start_date, end_date): params { code: ts_code, start: start_date, end: end_date, adjust: none # 先抓原始数据 } resp session.get(API_URL, paramsparams, timeout(5, 15)) resp.raise_for_status() raw resp.json() if not raw.get(data): return [] df pd.DataFrame(raw[data]) df standardize_columns(df) df df[(df[trade_date] start_date) (df[trade_date] end_date)] return df.to_dict(records)4.4 指数历史数据的抓取差异指数和个股的抓取流程基本一致但有几个差异点要注意第一指数没有复权概念。指数本身就是一篮子股票的价格加权不存在除权除息所以不需要复权因子字段。第二指数的代码规则不同。上证指数、深证成指、创业板指各有各的代码前缀抓取前要确认代码格式。第三指数的成交量含义不同。个股成交量是股数指数成交量是成分股成交量之和单位可能不一样做分析时要注意区分。第四指数的历史深度通常更长。很多指数有20年以上的历史数据抓取时要注意数据源是否支持这么长的区间。我一般把指数和个股分两个脚本处理虽然代码有重复但逻辑清晰维护起来方便。4.5 数据质量校验的实操方法数据抓下来不等于能用必须做校验。我常用的校验规则校验项规则处理方式价格逻辑high max(open, close), low min(open, close)标记异常人工检查价格范围价格 0过滤掉非正价格日期连续性相邻记录日期差不超过10个自然日超过则检查是否停牌成交量volume 0负值标记异常重复记录(ts_code, trade_date) 唯一主键约束自动保证校验脚本我一般单独写一个跑完输出一份报告总共多少条记录多少条异常异常分布在哪些股票哪些日期。这份报告我每次全量更新后都会看一遍。4.6 增量更新的调度实现增量更新我一般用两种触发方式手动触发跑一个脚本检查所有股票的最新日期抓取缺失部分。适合偶尔补数据。定时触发用系统的定时任务每个交易日收盘后固定时间跑一次。Linux 下用crontabWindows 下用任务计划程序。定时任务的脚本要加锁防止上一次还没跑完下一次就启动了。我用文件锁实现import fcntl def acquire_lock(lock_file): fp open(lock_file, w) try: fcntl.flock(fp, fcntl.LOCK_EX | fcntl.LOCK_NB) return fp except IOError: return None拿到锁就继续跑拿不到就退出等下一次调度。4.7 抓取性能的实测数据说一下我实测的性能数据供参考单只股票20年日线数据约5000条记录单次请求耗时约0.5到1.5秒5000只股票全量抓取4线程并发约40到60分钟每日增量更新约5000只股票各抓1条4线程并发约5到8分钟数据库全量查询某只股票全部历史SQLite 响应时间在10毫秒以内这个性能对于个人研究完全够用。如果你要更快的速度可以考虑把 SQLite 换成 PostgreSQL或者把数据存成 Parquet 文件用 DuckDB 查询但那是另一个话题了。5. 常见问题与排查技巧实录5.1 请求被拒绝或返回空数据这是最常见的问题。排查顺序检查请求头。User-Agent 和 Referer 是否设置正确。有些数据源对这两个字段很敏感。检查请求频率。是不是发太快了。把并发降到1加长请求间隔再试一次。检查参数格式。日期格式是YYYYMMDD还是YYYY-MM-DD股票代码带不带后缀这些细节很容易搞错。检查数据源状态。换个时间再试或者换个数据源交叉验证。我遇到过一次某数据源对请求间隔有隐藏限制每秒超过3个请求就返回空数据但不报错。这种最坑因为脚本不会抛异常只是默默拿到空结果。后来我加了“连续空结果告警”连续5次拿到空数据就暂停并打印警告。5.2 数据字段缺失或类型错误不同数据源、不同接口版本返回的字段可能不一样。排查方法先把原始响应打印出来看看实际有哪些字段对比字段映射表看是否有新增或改名的字段检查数值字段是否被返回成了字符串我遇到过一次成交量字段在某些记录里返回的是字符串12345在另一些记录里返回的是数字12345。pandas读进来之后整列变成object类型后续计算全错。解决办法是在解析层强制类型转换df[volume] pd.to_numeric(df[volume], errorscoerce)errorscoerce会把无法转换的值变成NaN然后统一处理NaN。5.3 复权因子对不上的问题复权因子是数据质量的重灾区。常见问题因子精度不够导致长周期复权价格累积误差因子更新不及时除权后几天才更新不同数据源的因子计算方式不一致我的处理方式是以原始价格为准复权因子只做参考。回测时如果需要复权价格用自己计算的因子不用数据源直接给的复权价格。自己算的因子可以控制精度和更新时机。5.4 数据库写入性能问题SQLite 默认每条INSERT都自动提交写入大量数据时非常慢。解决办法是批量写入加事务conn sqlite3.connect(market.db) conn.execute(BEGIN) for record in records: conn.execute(INSERT OR REPLACE INTO stock_daily VALUES (?,?,?,?,?,?,?,?,?), record) conn.commit()实测批量写入比逐条写入快10倍以上。如果数据量特别大还可以用executemanyconn.executemany(INSERT OR REPLACE INTO stock_daily VALUES (?,?,?,?,?,?,?,?,?), records) conn.commit()5.5 常见问题速查表现象可能原因排查方法解决方案请求返回空频率过高/参数错误降频重试检查参数加间隔核对参数格式字段缺失数据源改版打印原始响应更新字段映射表类型错误混合类型返回检查 dtype强制类型转换复权价异常因子精度/更新问题对比原始价自算因子写入慢逐条提交看写入耗时批量事务数据重复主键未生效查重复记录检查主键定义增量漏抓最新日期判断错对比本地和源加校验逻辑6. 几个我踩过的坑和对应的经验6.1 不要相信“最新交易日”这个字段很多数据源会返回一个“最新交易日”字段但这个字段经常不准。有的数据源在收盘后几小时内还显示前一天有的在节假日显示错误。我的做法是自己维护一个交易日历用交易日历判断当前应该抓到哪一天而不是依赖数据源返回的字段。交易日历可以从数据源获取也可以自己根据节假日规则生成。我一般每年初更新一次把全年的交易日列出来。6.2 停牌股票的处理要特别小心停牌股票在停牌期间没有交易数据但复牌后价格可能大幅跳空。如果回测时把停牌期间当成连续交易日会得到错误的收益率。我的处理方式是在数据中标记停牌区间回测时跳过停牌期间或者用停牌前的最后价格填充。6.3 数据库备份不能省我吃过一次亏增量更新脚本有个 bug把某只股票的历史数据覆盖成了只有最近一个月的数据。因为没有备份只能重新全量抓取花了将近一个小时。从那以后我每次全量更新前都会备份数据库文件增量更新前也会导出一份关键表的快照。6.4 日志要记到文件里控制台打印的日志关掉窗口就没了。我现在的做法是所有抓取日志同时输出到控制台和文件文件按日期切分保留最近30天。排查问题时直接翻日志文件比重新跑一遍脚本快得多。6.5 抓取节奏要像正常人访问这一点很微妙但很重要。如果你的请求模式看起来像机器人——固定间隔、固定顺序、24小时不间断——很容易被识别。我的做法是加入随机间隔比如基础间隔0.5秒上下浮动0.3秒抓取顺序也打乱不按代码顺序依次抓。这些小技巧能显著降低被限制的概率。6.6 数据源要准备备选没有任何一个数据源是100%可靠的。我至少准备两个数据源主源挂了切备源。切换逻辑可以自动也可以手动我一般做成配置项改个配置就能切换。两个数据源的字段映射表分开维护解析层根据配置选择对应的映射表。6.7 定期做全量校验增量更新跑久了难免会有数据不一致。我每个月做一次全量校验随机抽100只股票从数据源重新抓取全部历史数据和本地数据库比对。发现差异就记录分析原因。这个习惯帮我发现过好几次增量更新的边界问题。6.8 代码要能重复跑抓取脚本一定要设计成可以重复跑的。今天跑一半断了明天接着跑不能出问题。这就是为什么我用INSERT OR REPLACE而不是INSERT为什么增量更新要判断最新日期而不是无脑全量抓。可重复执行的脚本才是好脚本。6.9 数据格式要统一我见过有人把日期存成20240101有人存成2024-01-01有人存成2024/01/01。混在一起用的时候各种转换错误。我的做法是入库时统一转成YYYY-MM-DD字符串格式查询时如果需要其他格式再转换。这样数据库里的数据永远是干净的。6.10 不要过度优化最后一条经验不要一开始就追求完美。先把数据抓下来能跑通能用然后再优化。我见过太多人花大量时间设计完美的架构结果数据还没抓下来。先用最简单的方案跑通全流程再根据实际遇到的问题逐步优化这才是正确的顺序。这套方案我用了挺长时间从最初的单线程脚本到现在带并发和增量更新的版本中间迭代了很多次。核心逻辑其实不复杂难的是各种边界情况的处理和数据质量的保证。如果你刚开始做这件事建议先从一只股票、一个指数开始跑通了再扩展到全市场。遇到问题不要慌大部分问题都能通过降频、重试、校验这三招解决。