A股历史数据的获取几乎是每个做量化、做回测、做数据分析的人绕不开的第一道坎。我自己从最早手动从网页上复制粘贴到后来写脚本批量拉取中间踩过的坑能写满一整页纸。这篇文章就把我这些年抓取个股和指数历史数据的完整思路、代码实现和避坑经验一次性讲清楚。不管你是刚入门想跑通第一个策略的新手还是已经有一定基础但想找更稳定数据源的老手应该都能从里面找到有用的东西。核心关键词就三个Python、A股历史数据、个股与指数。围绕这三个词我会从数据源选型、接口调用、数据清洗、批量抓取、常见报错排查几个维度展开最后给出一套可以直接复用的代码框架。1. 数据源选型与整体思路拆解1.1 为什么数据源的选择决定了整个项目的成败抓A股数据这件事技术难度其实不高真正让人头疼的是数据源。我见过太多人代码写得漂漂亮亮结果接口一改、字段一变整个脚本全废。所以第一步不是写代码而是想清楚你要的数据从哪来这个来源稳不稳定字段全不全频率限制是多少。目前市面上常见的免费数据来源大致分几类。一类是开源财经数据接口库封装好了直接调用省事但依赖维护者的更新频率一类是财经门户网站的数据页面需要自己解析HTML或JSON还有一类是交易所或官方渠道公布的公开数据文件。这三类各有优劣我一般会根据数据量和更新频率来组合使用。对于个股日线、周线、月线这类历史行情开源接口库是最省心的选择。它们通常已经帮你处理好了复权、停牌、字段命名等细节。对于指数数据同样可以用这类库获取但要注意指数代码的格式和个股不一样需要单独处理。如果你需要分钟级甚至Tick级数据免费渠道基本很难满足那就得考虑付费数据服务了。1.2 复权这件事不理解清楚数据全白抓很多人抓完数据直接拿收盘价去算收益率结果发现和行情软件对不上。问题十有八九出在复权上。A股个股会因为分红、送股、配股导致股价出现跳空如果不做复权处理历史价格序列就是断裂的。复权分前复权、后复权和不复权三种。前复权是以最新价格为基准向前调整适合做技术分析和回测因为你看当前价格就是真实价格后复权是以上市首日为基准向后调整适合计算长期真实收益率不复权就是原始价格适合研究分红送配本身的影响。我的建议是做策略回测一律用前复权数据做长期收益归因用后复权做事件研究比如分红公告影响用不复权。大部分开源接口都支持通过参数指定复权方式抓取时一定要显式设置不要用默认值因为不同库的默认值不一样。1.3 整体架构设计从单次抓取到批量流水线一个能用的抓取方案不能只是能跑一次而要能稳定地、可重复地跑很多次。我的整体思路分三层第一层是配置层把股票代码列表、指数代码列表、时间范围、复权方式、输出路径这些参数集中管理第二层是抓取层负责调用接口、处理异常、控制频率第三层是存储层负责把数据落地成CSV或数据库方便后续读取。这样分层的好处是当数据源需要更换时只需要改抓取层配置和存储不用动。当需要增加新的数据字段时也只需要在抓取层做扩展。我早期把所有逻辑写在一个脚本里后来数据源一换就得重写教训很深刻。2. 核心细节解析与实操要点2.1 个股历史数据抓取的关键参数抓个股数据时有几个参数必须搞清楚。第一个是股票代码格式。不同接口对代码的要求不一样有的要带交易所前缀如sh600000、sz000001有的只要纯数字600000有的要带后缀600000.SH。抓之前先确认清楚否则会返回空数据或者报错。第二个是时间范围。大部分接口用start和end两个日期参数格式通常是YYYYMMDD或YYYY-MM-DD。要注意的是有些接口的end日期是包含当天的有些是不包含的这个差异会导致你少抓一天数据。我的习惯是end日期统一往后多写一天抓完再自己过滤。第三个是频率参数。日线用daily周线用weekly月线用monthly。如果你要分钟线参数通常是1min、5min、15min、30min、60min。分钟线数据量大抓取时要特别注意频率限制。第四个是复权参数。常见取值是qfq前复权、hfq后复权、空字符串不复权。这个参数不设置的话不同库行为不同有的默认前复权有的默认不复权一定要显式指定。2.2 指数数据抓取的特殊之处指数和个股最大的区别在于指数没有复权概念因为指数本身就是通过成分股计算出来的不存在分红送配导致的除权问题。所以抓指数数据时复权参数通常不需要设置或者设置了也会被忽略。但指数有自己的坑。第一指数代码格式和个股不同上证指数是000001但平安银行也是000001只是交易所不同。所以抓指数时必须明确指定这是指数不能和个股混在一起。第二不同接口对指数的支持程度不一样有些库对个股支持很好但指数数据不全或者更新不及时。第三指数的历史数据起点不同有的指数只有几年数据有的有十几年抓之前最好先查一下指数的发布日。我一般会把指数代码单独维护一个列表和个股列表分开管理。常见的宽基指数包括上证综指、深证成指、沪深300、中证500、创业板指等每个指数对应的代码格式要提前确认好。2.3 频率控制与反爬策略免费接口通常都有频率限制有的是每分钟多少次有的是每天多少次。如果你不管不顾地循环抓几千只股票轻则被封IP一段时间重则直接被拉黑。我踩过最惨的一次是半夜跑批量任务第二天发现IP被限了整整一天所有工作停摆。合理的做法是第一在每次请求之间加延时一般0.3到1秒比较安全具体看接口的容忍度第二用try-except包裹每次请求遇到限流错误就sleep更长时间再重试第三把已经抓成功的数据及时落盘这样即使中途中断下次可以从断点继续不用从头再来。还有一个技巧是分批抓取。比如把5000只股票分成10批每批500只批与批之间休息几分钟。这样既能控制总时长又能降低被封的风险。如果你数据量特别大可以考虑在多个时间段分散抓取比如白天抓一部分晚上抓一部分。2.4 数据字段的完整性与命名规范抓下来的数据字段是否完整直接决定了后续能不能用。日线数据最基本的字段包括日期、开盘价、最高价、最低价、收盘价、成交量、成交额。有些接口还会返回涨跌幅、换手率、振幅等衍生字段这些有则更好没有也可以自己算。字段命名上我建议统一成英文小写加下划线比如date、open、high、low、close、volume、amount。不要直接用接口返回的中文列名因为中文列名在不同系统间传递容易出编码问题。如果接口返回的是中文抓完立刻重命名。另外要注意数据类型的统一。日期统一成datetime类型价格统一成float成交量统一成int或float。我遇到过因为某只股票成交量特别大导致整列被识别成object类型的情况后来每次抓完都强制做一次类型转换。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先把环境搭好。Python版本建议3.8以上太老的版本有些库不支持。核心依赖就几个数据抓取库、数据处理库、存储库。安装命令如下pip install akshare pandas numpy如果你用的是其他数据接口库把akshare换成对应的库名即可。pandas和numpy是数据处理必备基本所有方案都会用到。安装完之后先做个简单测试确认库能正常导入并且能拿到数据。不要一上来就写批量脚本先用单只股票跑通再扩展。import akshare as ak import pandas as pd # 测试单只股票日线数据 df ak.stock_zh_a_hist(symbol600000, perioddaily, start_date20230101, end_date20231231, adjustqfq) print(df.head()) print(df.shape)如果这段代码能正常输出数据说明环境没问题。如果报错先检查网络连接再检查库版本最后检查接口是否变更。3.2 单只个股历史数据抓取完整实现单只股票抓取是所有批量操作的基础。下面这段代码是我常用的模板包含了参数设置、异常处理和基本的数据清洗import akshare as ak import pandas as pd import time def fetch_stock_daily(symbol, start_date, end_date, adjustqfq): 抓取单只A股个股日线数据 symbol: 纯数字代码如 600000 start_date/end_date: 格式 YYYYMMDD adjust: qfq前复权 / hfq后复权 / 不复权 try: df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustadjust ) if df is None or df.empty: print(f{symbol} 返回空数据) return None # 重命名列为英文 df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 振幅: amplitude, 涨跌幅: pct_change, 涨跌额: change, 换手率: turnover }) # 类型转换 df[date] pd.to_datetime(df[date]) for col in [open, close, high, low, amount, amplitude, pct_change, change, turnover]: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) if volume in df.columns: df[volume] pd.to_numeric(df[volume], errorscoerce) df df.sort_values(date).reset_index(dropTrue) df[symbol] symbol return df except Exception as e: print(f{symbol} 抓取失败: {e}) return None这段代码有几个细节值得说。第一adjust参数显式传入不依赖默认值。第二列重命名放在最前面后面所有操作都用英文列名。第三类型转换用errorscoerce遇到无法转换的值变成NaN而不是直接报错。第四最后加了一列symbol方便后续合并多只股票的数据。3.3 指数历史数据抓取实现指数抓取和个股类似但接口函数不同参数也有差异def fetch_index_daily(symbol, start_date, end_date): 抓取指数日线数据 symbol: 指数代码如 000001 表示上证指数 try: df ak.stock_zh_index_daily_em(symbolsymbol) if df is None or df.empty: print(f指数 {symbol} 返回空数据) return None df df.rename(columns{ date: date, open: open, close: close, high: high, low: low, volume: volume, amount: amount }) df[date] pd.to_datetime(df[date]) df df[(df[date] pd.to_datetime(start_date)) (df[date] pd.to_datetime(end_date))] df df.sort_values(date).reset_index(dropTrue) df[symbol] symbol return df except Exception as e: print(f指数 {symbol} 抓取失败: {e}) return None注意指数接口返回的列名可能已经是英文也可能和个股不同抓完先print一下看看实际列名再决定怎么重命名。另外指数数据我习惯抓全量再按日期过滤因为指数数据量不大全量抓一次也就几秒钟。3.4 批量抓取与断点续传批量抓取的核心是循环加延时加异常处理。下面是我常用的批量框架import os import time def batch_fetch_stocks(symbol_list, start_date, end_date, adjustqfq, save_dir./data/stock): os.makedirs(save_dir, exist_okTrue) success, fail [], [] for i, symbol in enumerate(symbol_list): filepath os.path.join(save_dir, f{symbol}.csv) # 断点续传已存在则跳过 if os.path.exists(filepath): print(f[{i1}/{len(symbol_list)}] {symbol} 已存在跳过) success.append(symbol) continue df fetch_stock_daily(symbol, start_date, end_date, adjust) if df is not None and not df.empty: df.to_csv(filepath, indexFalse, encodingutf-8-sig) success.append(symbol) print(f[{i1}/{len(symbol_list)}] {symbol} 成功{len(df)}条) else: fail.append(symbol) print(f[{i1}/{len(symbol_list)}] {symbol} 失败) time.sleep(0.5) # 频率控制 print(f\n完成成功{len(success)}只失败{len(fail)}只) if fail: print(f失败列表{fail}) return success, fail断点续传这个设计非常关键。我早期没有做这个每次中断都得从头再来几千只股票抓一次要几个小时中途断一次心态直接崩。加了断点续传之后中断了重新跑一遍已经抓好的自动跳过只补缺失的部分。3.5 数据存储方案对比数据存哪里取决于你的使用场景。我列了一个对比表存储方式优点缺点适用场景CSV文件简单直观无需额外依赖读取慢不支持并发数据量小个人使用SQLite单文件支持SQL查询并发写入弱中等数据量本地分析MySQL/PostgreSQL支持并发查询快需要额外部署大数据量多人协作Parquet压缩率高读取快需要pyarrow依赖大数据量列式分析HDF5读写快支持复杂结构文件损坏风险科学计算场景我个人的习惯是抓取阶段先存CSV方便检查和调试抓完之后统一导入SQLite或Parquet做后续分析。这样既保证了抓取过程的简单可靠又保证了分析阶段的高效。4. 常见问题与排查技巧实录4.1 接口报错与数据为空的排查思路抓取过程中最常见的问题就是报错或者返回空数据。我整理了一个排查清单按顺序检查问题现象可能原因排查方法解决方案返回空DataFrame代码格式不对检查是否需要交易所前缀调整代码格式返回空DataFrame时间范围无交易日检查日期是否在上市之后调整时间范围报连接错误网络问题或接口变更单独测试接口检查网络更新库版本报限流错误请求频率过高查看错误信息增加延时降低频率字段缺失接口版本变更print列名对比更新重命名映射数据对不上复权方式不一致对比行情软件统一复权参数我遇到最多的是代码格式问题。比如某只股票是深市的代码需要加sz前缀但我按沪市格式传了结果返回空。后来我写了一个代码格式转换函数根据代码首位数字自动判断交易所省了很多事。4.2 复权数据与行情软件对不上的原因这个问题我被问过无数次。明明抓的是前复权数据但和某行情软件一比价格就是差那么一点。原因通常有三个。第一复权基准日不同。前复权是以最新交易日为基准的但不同软件用的最新可能不是同一天。比如你昨天抓的数据今天软件更新了基准日变了价格自然对不上。解决办法是抓取时记录抓取日期对比时用同一天的数据。第二复权算法细节不同。有的软件用等比复权有的用等差复权分红再投资的假设也不一样。这个没法完全对齐只能接受微小差异。一般来说差异在千分之一以内是正常的。第三数据源本身有误。免费数据源偶尔会有错误数据尤其是早期的历史数据。如果发现某一天价格明显异常可以对比多个数据源确认。4.3 停牌、退市股票的处理A股有停牌和退市的情况抓取时如果不处理会导致数据缺失或者报错。停牌的股票在停牌期间没有交易数据接口返回的数据里会直接跳过这些日期这是正常的。但如果你做回测需要自己处理这些缺失日期比如用前一日收盘价填充或者直接跳过。退市股票的处理更麻烦。有些接口对退市股票不返回数据有些返回但代码格式不同。我的做法是维护一个退市股票列表抓取时单独处理抓不到就记录到失败列表不阻塞其他股票。还有一个细节是ST股票。ST股票的涨跌停限制是5%和普通股票的10%不同。如果你做涨跌停相关的策略需要单独标记ST股票。这个信息通常在股票基本信息接口里能拿到。4.4 数据清洗的常见坑抓下来的原始数据不能直接用必须清洗。我总结几个常见的坑。第一个坑是重复数据。有些接口在特定情况下会返回重复行尤其是当时间范围跨越了接口的数据更新点时。抓完一定要drop_duplicates按date列去重。第二个坑是缺失值。成交量、成交额偶尔会有缺失可能是接口问题也可能是当天确实没有数据。处理方式取决于你的用途做价格分析可以忽略做成交量分析就需要填充或剔除。第三个坑是异常值。偶尔会出现价格明显错误的数据比如收盘价是开盘价的10倍。这种通常是数据源的问题需要设置合理的阈值过滤比如涨跌幅超过20%的非ST、非新股就标记为可疑。第四个坑是日期格式不统一。有的接口返回2023-01-01有的返回20230101有的返回2023/01/01。统一用pd.to_datetime转换它能自动识别大部分格式。4.5 提升抓取效率的实用技巧最后分享几个提升效率的技巧。并发抓取。如果接口允许可以用多线程并发抓取。但要注意并发数不要太高一般3到5个线程就够了太高容易被限流。用concurrent.futures的ThreadPoolExecutor实现起来很简单。增量更新。不要每次都抓全量历史数据只抓上次抓取之后的新数据然后追加到已有文件里。这样能把每次更新的时间从几小时缩短到几分钟。缓存机制。对于不常变的数据比如股票基本信息、指数成分股列表抓一次存本地下次直接用不用重复请求。日志记录。每次抓取都记录日志包括抓取时间、成功数量、失败列表、耗时等。出问题的时候日志是排查的第一手资料。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(fetch.log, encodingutf-8), logging.StreamHandler() ] )这套日志配置我每个抓取项目都会用FileHandler写文件StreamHandler输出到控制台两边都不耽误。5. 一套可直接复用的完整代码框架5.1 项目目录结构设计把前面所有内容整合成一个可复用的项目目录结构建议这样组织stock_data_fetcher/ ├── config.py # 配置参数 ├── fetcher.py # 抓取核心逻辑 ├── storage.py # 存储逻辑 ├── utils.py # 工具函数 ├── main.py # 主入口 ├── data/ │ ├── stock/ # 个股数据 │ └── index/ # 指数数据 └── logs/ # 日志文件这样拆分的好处是每个模块职责单一修改一处不影响其他部分。config.py集中管理所有参数换数据源或者改时间范围只需要改这一个文件。5.2 配置与工具函数config.py的内容大概是这样# config.py START_DATE 20200101 END_DATE 20241231 ADJUST qfq STOCK_SAVE_DIR ./data/stock INDEX_SAVE_DIR ./data/index REQUEST_DELAY 0.5 MAX_RETRY 3 # 指数列表 INDEX_LIST [ 000001, # 上证指数 399001, # 深证成指 000300, # 沪深300 000905, # 中证500 399006, # 创业板指 ]工具函数里我一般放代码格式转换、日期处理、重试装饰器这些通用逻辑# utils.py import time from functools import wraps def retry(max_retry3, delay2): def decorator(func): wraps(func) def wrapper(*args, **kwargs): for i in range(max_retry): try: return func(*args, **kwargs) except Exception as e: if i max_retry - 1: raise print(f第{i1}次失败{delay}秒后重试: {e}) time.sleep(delay) return wrapper return decorator def normalize_stock_code(code): 统一股票代码格式为纯数字 code str(code).strip().upper() for prefix in [SH, SZ, BJ]: if code.startswith(prefix): code code[len(prefix):] for suffix in [.SH, .SZ, .BJ]: if code.endswith(suffix): code code[:-len(suffix)] return code.zfill(6)重试装饰器这个设计很实用网络请求偶尔失败是正常的自动重试能大幅提升成功率。代码格式转换函数则解决了不同数据源代码格式不统一的问题。5.3 主流程串联main.py把所有模块串起来# main.py import os import pandas as pd from config import * from fetcher import fetch_stock_daily, fetch_index_daily from utils import retry import time def get_stock_list(): 获取全部A股代码列表这里用接口获取 import akshare as ak df ak.stock_info_a_code_name() return df[code].tolist() def main(): # 抓取个股 stock_list get_stock_list() print(f共{len(stock_list)}只股票待抓取) for i, code in enumerate(stock_list): code normalize_stock_code(code) filepath os.path.join(STOCK_SAVE_DIR, f{code}.csv) if os.path.exists(filepath): continue df fetch_stock_daily(code, START_DATE, END_DATE, ADJUST) if df is not None: df.to_csv(filepath, indexFalse, encodingutf-8-sig) time.sleep(REQUEST_DELAY) # 抓取指数 for code in INDEX_LIST: df fetch_index_daily(code, START_DATE, END_DATE) if df is not None: filepath os.path.join(INDEX_SAVE_DIR, f{code}.csv) df.to_csv(filepath, indexFalse, encodingutf-8-sig) if __name__ __main__: main()这个框架跑起来之后你只需要改config.py里的参数就能适配不同的抓取需求。想换数据源改fetcher.py想换存储方式改storage.py想加新的数据字段在fetcher里扩展。5.4 数据质量校验抓完之后一定要做质量校验不然拿着脏数据做分析结论全是错的。我一般检查这几项def validate_data(df, symbol): 数据质量校验 issues [] # 检查是否有数据 if df is None or df.empty: return [f{symbol}: 数据为空] # 检查日期是否连续排除周末和节假日 df df.sort_values(date) date_diff df[date].diff().dt.days if (date_diff 10).any(): issues.append(f{symbol}: 存在超过10天的数据缺口) # 检查价格逻辑 if (df[high] df[low]).any(): issues.append(f{symbol}: 最高价低于最低价) if (df[close] 0).any(): issues.append(f{symbol}: 存在非正收盘价) # 检查涨跌幅异常 if pct_change in df.columns: abnormal df[df[pct_change].abs() 21] if len(abnormal) 0: issues.append(f{symbol}: 存在{len(abnormal)}条涨跌幅异常记录) return issues这个校验函数会返回所有发现的问题跑完批量抓取后统一执行一遍把有问题的股票挑出来人工检查。我实测下来免费数据源大概有1%到3%的股票会存在各种小问题提前发现比事后返工强得多。6. 抓取之外数据后续使用的一些经验数据抓下来只是第一步怎么用好才是关键。我分享几个实际使用中的体会。数据对齐问题。不同股票的历史数据长度不一样有的上市早有的上市晚。做多股票对比分析时需要先对齐日期通常用pivot或者merge操作。我一般会构建一个统一的交易日历把所有股票数据reindex到这个日历上缺失的用NaN填充。性能优化。当股票数量到几千只时逐个读取CSV会很慢。这时候要么把数据合并成一个大文件要么用数据库。我试过把5000只股票合并成一个Parquet文件读取速度比逐个读CSV快了几十倍。数据更新策略。历史数据不是抓一次就完事了每天都有新数据产生。我的做法是每周更新一次只抓最近一周的数据追加到已有文件里。这样既保证了数据新鲜度又不会浪费太多时间。版本管理。数据源会变接口会变抓取逻辑也会变。我习惯给每次抓取的数据打上版本标签记录抓取日期和数据源版本。这样当发现数据问题时能快速定位是哪次抓取出的问题。最后说一个我踩过的大坑。早期我抓数据没有做去重结果某次接口返回了重复数据我没发现直接拿去做回测回测结果好得离谱后来排查了半天才发现是重复数据导致的。从那以后我每次抓完数据第一件事就是drop_duplicates这个习惯救了我很多次。数据抓取这件事技术门槛不高但细节特别多。把上面这些点都注意到基本能覆盖90%以上的场景。剩下的10%就得靠实际跑起来之后慢慢积累了。遇到问题不要慌先看日志再对比数据最后查接口文档大部分问题都能定位到。 
长河君 · 2026/10/13 1:29:21 · 阅读 1.2k Python抓取A股历史数据：个股与指数接口、复权与批量实战
A股历史数据的获取几乎是每个做量化、做回测、做数据分析的人绕不开的第一道坎。我自己从最早手动从网页上复制粘贴到后来写脚本批量拉取中间踩过的坑能写满一整页纸。这篇文章就把我这些年抓取个股和指数历史数据的完整思路、代码实现和避坑经验一次性讲清楚。不管你是刚入门想跑通第一个策略的新手还是已经有一定基础但想找更稳定数据源的老手应该都能从里面找到有用的东西。核心关键词就三个Python、A股历史数据、个股与指数。围绕这三个词我会从数据源选型、接口调用、数据清洗、批量抓取、常见报错排查几个维度展开最后给出一套可以直接复用的代码框架。1. 数据源选型与整体思路拆解1.1 为什么数据源的选择决定了整个项目的成败抓A股数据这件事技术难度其实不高真正让人头疼的是数据源。我见过太多人代码写得漂漂亮亮结果接口一改、字段一变整个脚本全废。所以第一步不是写代码而是想清楚你要的数据从哪来这个来源稳不稳定字段全不全频率限制是多少。目前市面上常见的免费数据来源大致分几类。一类是开源财经数据接口库封装好了直接调用省事但依赖维护者的更新频率一类是财经门户网站的数据页面需要自己解析HTML或JSON还有一类是交易所或官方渠道公布的公开数据文件。这三类各有优劣我一般会根据数据量和更新频率来组合使用。对于个股日线、周线、月线这类历史行情开源接口库是最省心的选择。它们通常已经帮你处理好了复权、停牌、字段命名等细节。对于指数数据同样可以用这类库获取但要注意指数代码的格式和个股不一样需要单独处理。如果你需要分钟级甚至Tick级数据免费渠道基本很难满足那就得考虑付费数据服务了。1.2 复权这件事不理解清楚数据全白抓很多人抓完数据直接拿收盘价去算收益率结果发现和行情软件对不上。问题十有八九出在复权上。A股个股会因为分红、送股、配股导致股价出现跳空如果不做复权处理历史价格序列就是断裂的。复权分前复权、后复权和不复权三种。前复权是以最新价格为基准向前调整适合做技术分析和回测因为你看当前价格就是真实价格后复权是以上市首日为基准向后调整适合计算长期真实收益率不复权就是原始价格适合研究分红送配本身的影响。我的建议是做策略回测一律用前复权数据做长期收益归因用后复权做事件研究比如分红公告影响用不复权。大部分开源接口都支持通过参数指定复权方式抓取时一定要显式设置不要用默认值因为不同库的默认值不一样。1.3 整体架构设计从单次抓取到批量流水线一个能用的抓取方案不能只是能跑一次而要能稳定地、可重复地跑很多次。我的整体思路分三层第一层是配置层把股票代码列表、指数代码列表、时间范围、复权方式、输出路径这些参数集中管理第二层是抓取层负责调用接口、处理异常、控制频率第三层是存储层负责把数据落地成CSV或数据库方便后续读取。这样分层的好处是当数据源需要更换时只需要改抓取层配置和存储不用动。当需要增加新的数据字段时也只需要在抓取层做扩展。我早期把所有逻辑写在一个脚本里后来数据源一换就得重写教训很深刻。2. 核心细节解析与实操要点2.1 个股历史数据抓取的关键参数抓个股数据时有几个参数必须搞清楚。第一个是股票代码格式。不同接口对代码的要求不一样有的要带交易所前缀如sh600000、sz000001有的只要纯数字600000有的要带后缀600000.SH。抓之前先确认清楚否则会返回空数据或者报错。第二个是时间范围。大部分接口用start和end两个日期参数格式通常是YYYYMMDD或YYYY-MM-DD。要注意的是有些接口的end日期是包含当天的有些是不包含的这个差异会导致你少抓一天数据。我的习惯是end日期统一往后多写一天抓完再自己过滤。第三个是频率参数。日线用daily周线用weekly月线用monthly。如果你要分钟线参数通常是1min、5min、15min、30min、60min。分钟线数据量大抓取时要特别注意频率限制。第四个是复权参数。常见取值是qfq前复权、hfq后复权、空字符串不复权。这个参数不设置的话不同库行为不同有的默认前复权有的默认不复权一定要显式指定。2.2 指数数据抓取的特殊之处指数和个股最大的区别在于指数没有复权概念因为指数本身就是通过成分股计算出来的不存在分红送配导致的除权问题。所以抓指数数据时复权参数通常不需要设置或者设置了也会被忽略。但指数有自己的坑。第一指数代码格式和个股不同上证指数是000001但平安银行也是000001只是交易所不同。所以抓指数时必须明确指定这是指数不能和个股混在一起。第二不同接口对指数的支持程度不一样有些库对个股支持很好但指数数据不全或者更新不及时。第三指数的历史数据起点不同有的指数只有几年数据有的有十几年抓之前最好先查一下指数的发布日。我一般会把指数代码单独维护一个列表和个股列表分开管理。常见的宽基指数包括上证综指、深证成指、沪深300、中证500、创业板指等每个指数对应的代码格式要提前确认好。2.3 频率控制与反爬策略免费接口通常都有频率限制有的是每分钟多少次有的是每天多少次。如果你不管不顾地循环抓几千只股票轻则被封IP一段时间重则直接被拉黑。我踩过最惨的一次是半夜跑批量任务第二天发现IP被限了整整一天所有工作停摆。合理的做法是第一在每次请求之间加延时一般0.3到1秒比较安全具体看接口的容忍度第二用try-except包裹每次请求遇到限流错误就sleep更长时间再重试第三把已经抓成功的数据及时落盘这样即使中途中断下次可以从断点继续不用从头再来。还有一个技巧是分批抓取。比如把5000只股票分成10批每批500只批与批之间休息几分钟。这样既能控制总时长又能降低被封的风险。如果你数据量特别大可以考虑在多个时间段分散抓取比如白天抓一部分晚上抓一部分。2.4 数据字段的完整性与命名规范抓下来的数据字段是否完整直接决定了后续能不能用。日线数据最基本的字段包括日期、开盘价、最高价、最低价、收盘价、成交量、成交额。有些接口还会返回涨跌幅、换手率、振幅等衍生字段这些有则更好没有也可以自己算。字段命名上我建议统一成英文小写加下划线比如date、open、high、low、close、volume、amount。不要直接用接口返回的中文列名因为中文列名在不同系统间传递容易出编码问题。如果接口返回的是中文抓完立刻重命名。另外要注意数据类型的统一。日期统一成datetime类型价格统一成float成交量统一成int或float。我遇到过因为某只股票成交量特别大导致整列被识别成object类型的情况后来每次抓完都强制做一次类型转换。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先把环境搭好。Python版本建议3.8以上太老的版本有些库不支持。核心依赖就几个数据抓取库、数据处理库、存储库。安装命令如下pip install akshare pandas numpy如果你用的是其他数据接口库把akshare换成对应的库名即可。pandas和numpy是数据处理必备基本所有方案都会用到。安装完之后先做个简单测试确认库能正常导入并且能拿到数据。不要一上来就写批量脚本先用单只股票跑通再扩展。import akshare as ak import pandas as pd # 测试单只股票日线数据 df ak.stock_zh_a_hist(symbol600000, perioddaily, start_date20230101, end_date20231231, adjustqfq) print(df.head()) print(df.shape)如果这段代码能正常输出数据说明环境没问题。如果报错先检查网络连接再检查库版本最后检查接口是否变更。3.2 单只个股历史数据抓取完整实现单只股票抓取是所有批量操作的基础。下面这段代码是我常用的模板包含了参数设置、异常处理和基本的数据清洗import akshare as ak import pandas as pd import time def fetch_stock_daily(symbol, start_date, end_date, adjustqfq): 抓取单只A股个股日线数据 symbol: 纯数字代码如 600000 start_date/end_date: 格式 YYYYMMDD adjust: qfq前复权 / hfq后复权 / 不复权 try: df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustadjust ) if df is None or df.empty: print(f{symbol} 返回空数据) return None # 重命名列为英文 df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 振幅: amplitude, 涨跌幅: pct_change, 涨跌额: change, 换手率: turnover }) # 类型转换 df[date] pd.to_datetime(df[date]) for col in [open, close, high, low, amount, amplitude, pct_change, change, turnover]: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) if volume in df.columns: df[volume] pd.to_numeric(df[volume], errorscoerce) df df.sort_values(date).reset_index(dropTrue) df[symbol] symbol return df except Exception as e: print(f{symbol} 抓取失败: {e}) return None这段代码有几个细节值得说。第一adjust参数显式传入不依赖默认值。第二列重命名放在最前面后面所有操作都用英文列名。第三类型转换用errorscoerce遇到无法转换的值变成NaN而不是直接报错。第四最后加了一列symbol方便后续合并多只股票的数据。3.3 指数历史数据抓取实现指数抓取和个股类似但接口函数不同参数也有差异def fetch_index_daily(symbol, start_date, end_date): 抓取指数日线数据 symbol: 指数代码如 000001 表示上证指数 try: df ak.stock_zh_index_daily_em(symbolsymbol) if df is None or df.empty: print(f指数 {symbol} 返回空数据) return None df df.rename(columns{ date: date, open: open, close: close, high: high, low: low, volume: volume, amount: amount }) df[date] pd.to_datetime(df[date]) df df[(df[date] pd.to_datetime(start_date)) (df[date] pd.to_datetime(end_date))] df df.sort_values(date).reset_index(dropTrue) df[symbol] symbol return df except Exception as e: print(f指数 {symbol} 抓取失败: {e}) return None注意指数接口返回的列名可能已经是英文也可能和个股不同抓完先print一下看看实际列名再决定怎么重命名。另外指数数据我习惯抓全量再按日期过滤因为指数数据量不大全量抓一次也就几秒钟。3.4 批量抓取与断点续传批量抓取的核心是循环加延时加异常处理。下面是我常用的批量框架import os import time def batch_fetch_stocks(symbol_list, start_date, end_date, adjustqfq, save_dir./data/stock): os.makedirs(save_dir, exist_okTrue) success, fail [], [] for i, symbol in enumerate(symbol_list): filepath os.path.join(save_dir, f{symbol}.csv) # 断点续传已存在则跳过 if os.path.exists(filepath): print(f[{i1}/{len(symbol_list)}] {symbol} 已存在跳过) success.append(symbol) continue df fetch_stock_daily(symbol, start_date, end_date, adjust) if df is not None and not df.empty: df.to_csv(filepath, indexFalse, encodingutf-8-sig) success.append(symbol) print(f[{i1}/{len(symbol_list)}] {symbol} 成功{len(df)}条) else: fail.append(symbol) print(f[{i1}/{len(symbol_list)}] {symbol} 失败) time.sleep(0.5) # 频率控制 print(f\n完成成功{len(success)}只失败{len(fail)}只) if fail: print(f失败列表{fail}) return success, fail断点续传这个设计非常关键。我早期没有做这个每次中断都得从头再来几千只股票抓一次要几个小时中途断一次心态直接崩。加了断点续传之后中断了重新跑一遍已经抓好的自动跳过只补缺失的部分。3.5 数据存储方案对比数据存哪里取决于你的使用场景。我列了一个对比表存储方式优点缺点适用场景CSV文件简单直观无需额外依赖读取慢不支持并发数据量小个人使用SQLite单文件支持SQL查询并发写入弱中等数据量本地分析MySQL/PostgreSQL支持并发查询快需要额外部署大数据量多人协作Parquet压缩率高读取快需要pyarrow依赖大数据量列式分析HDF5读写快支持复杂结构文件损坏风险科学计算场景我个人的习惯是抓取阶段先存CSV方便检查和调试抓完之后统一导入SQLite或Parquet做后续分析。这样既保证了抓取过程的简单可靠又保证了分析阶段的高效。4. 常见问题与排查技巧实录4.1 接口报错与数据为空的排查思路抓取过程中最常见的问题就是报错或者返回空数据。我整理了一个排查清单按顺序检查问题现象可能原因排查方法解决方案返回空DataFrame代码格式不对检查是否需要交易所前缀调整代码格式返回空DataFrame时间范围无交易日检查日期是否在上市之后调整时间范围报连接错误网络问题或接口变更单独测试接口检查网络更新库版本报限流错误请求频率过高查看错误信息增加延时降低频率字段缺失接口版本变更print列名对比更新重命名映射数据对不上复权方式不一致对比行情软件统一复权参数我遇到最多的是代码格式问题。比如某只股票是深市的代码需要加sz前缀但我按沪市格式传了结果返回空。后来我写了一个代码格式转换函数根据代码首位数字自动判断交易所省了很多事。4.2 复权数据与行情软件对不上的原因这个问题我被问过无数次。明明抓的是前复权数据但和某行情软件一比价格就是差那么一点。原因通常有三个。第一复权基准日不同。前复权是以最新交易日为基准的但不同软件用的最新可能不是同一天。比如你昨天抓的数据今天软件更新了基准日变了价格自然对不上。解决办法是抓取时记录抓取日期对比时用同一天的数据。第二复权算法细节不同。有的软件用等比复权有的用等差复权分红再投资的假设也不一样。这个没法完全对齐只能接受微小差异。一般来说差异在千分之一以内是正常的。第三数据源本身有误。免费数据源偶尔会有错误数据尤其是早期的历史数据。如果发现某一天价格明显异常可以对比多个数据源确认。4.3 停牌、退市股票的处理A股有停牌和退市的情况抓取时如果不处理会导致数据缺失或者报错。停牌的股票在停牌期间没有交易数据接口返回的数据里会直接跳过这些日期这是正常的。但如果你做回测需要自己处理这些缺失日期比如用前一日收盘价填充或者直接跳过。退市股票的处理更麻烦。有些接口对退市股票不返回数据有些返回但代码格式不同。我的做法是维护一个退市股票列表抓取时单独处理抓不到就记录到失败列表不阻塞其他股票。还有一个细节是ST股票。ST股票的涨跌停限制是5%和普通股票的10%不同。如果你做涨跌停相关的策略需要单独标记ST股票。这个信息通常在股票基本信息接口里能拿到。4.4 数据清洗的常见坑抓下来的原始数据不能直接用必须清洗。我总结几个常见的坑。第一个坑是重复数据。有些接口在特定情况下会返回重复行尤其是当时间范围跨越了接口的数据更新点时。抓完一定要drop_duplicates按date列去重。第二个坑是缺失值。成交量、成交额偶尔会有缺失可能是接口问题也可能是当天确实没有数据。处理方式取决于你的用途做价格分析可以忽略做成交量分析就需要填充或剔除。第三个坑是异常值。偶尔会出现价格明显错误的数据比如收盘价是开盘价的10倍。这种通常是数据源的问题需要设置合理的阈值过滤比如涨跌幅超过20%的非ST、非新股就标记为可疑。第四个坑是日期格式不统一。有的接口返回2023-01-01有的返回20230101有的返回2023/01/01。统一用pd.to_datetime转换它能自动识别大部分格式。4.5 提升抓取效率的实用技巧最后分享几个提升效率的技巧。并发抓取。如果接口允许可以用多线程并发抓取。但要注意并发数不要太高一般3到5个线程就够了太高容易被限流。用concurrent.futures的ThreadPoolExecutor实现起来很简单。增量更新。不要每次都抓全量历史数据只抓上次抓取之后的新数据然后追加到已有文件里。这样能把每次更新的时间从几小时缩短到几分钟。缓存机制。对于不常变的数据比如股票基本信息、指数成分股列表抓一次存本地下次直接用不用重复请求。日志记录。每次抓取都记录日志包括抓取时间、成功数量、失败列表、耗时等。出问题的时候日志是排查的第一手资料。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(fetch.log, encodingutf-8), logging.StreamHandler() ] )这套日志配置我每个抓取项目都会用FileHandler写文件StreamHandler输出到控制台两边都不耽误。5. 一套可直接复用的完整代码框架5.1 项目目录结构设计把前面所有内容整合成一个可复用的项目目录结构建议这样组织stock_data_fetcher/ ├── config.py # 配置参数 ├── fetcher.py # 抓取核心逻辑 ├── storage.py # 存储逻辑 ├── utils.py # 工具函数 ├── main.py # 主入口 ├── data/ │ ├── stock/ # 个股数据 │ └── index/ # 指数数据 └── logs/ # 日志文件这样拆分的好处是每个模块职责单一修改一处不影响其他部分。config.py集中管理所有参数换数据源或者改时间范围只需要改这一个文件。5.2 配置与工具函数config.py的内容大概是这样# config.py START_DATE 20200101 END_DATE 20241231 ADJUST qfq STOCK_SAVE_DIR ./data/stock INDEX_SAVE_DIR ./data/index REQUEST_DELAY 0.5 MAX_RETRY 3 # 指数列表 INDEX_LIST [ 000001, # 上证指数 399001, # 深证成指 000300, # 沪深300 000905, # 中证500 399006, # 创业板指 ]工具函数里我一般放代码格式转换、日期处理、重试装饰器这些通用逻辑# utils.py import time from functools import wraps def retry(max_retry3, delay2): def decorator(func): wraps(func) def wrapper(*args, **kwargs): for i in range(max_retry): try: return func(*args, **kwargs) except Exception as e: if i max_retry - 1: raise print(f第{i1}次失败{delay}秒后重试: {e}) time.sleep(delay) return wrapper return decorator def normalize_stock_code(code): 统一股票代码格式为纯数字 code str(code).strip().upper() for prefix in [SH, SZ, BJ]: if code.startswith(prefix): code code[len(prefix):] for suffix in [.SH, .SZ, .BJ]: if code.endswith(suffix): code code[:-len(suffix)] return code.zfill(6)重试装饰器这个设计很实用网络请求偶尔失败是正常的自动重试能大幅提升成功率。代码格式转换函数则解决了不同数据源代码格式不统一的问题。5.3 主流程串联main.py把所有模块串起来# main.py import os import pandas as pd from config import * from fetcher import fetch_stock_daily, fetch_index_daily from utils import retry import time def get_stock_list(): 获取全部A股代码列表这里用接口获取 import akshare as ak df ak.stock_info_a_code_name() return df[code].tolist() def main(): # 抓取个股 stock_list get_stock_list() print(f共{len(stock_list)}只股票待抓取) for i, code in enumerate(stock_list): code normalize_stock_code(code) filepath os.path.join(STOCK_SAVE_DIR, f{code}.csv) if os.path.exists(filepath): continue df fetch_stock_daily(code, START_DATE, END_DATE, ADJUST) if df is not None: df.to_csv(filepath, indexFalse, encodingutf-8-sig) time.sleep(REQUEST_DELAY) # 抓取指数 for code in INDEX_LIST: df fetch_index_daily(code, START_DATE, END_DATE) if df is not None: filepath os.path.join(INDEX_SAVE_DIR, f{code}.csv) df.to_csv(filepath, indexFalse, encodingutf-8-sig) if __name__ __main__: main()这个框架跑起来之后你只需要改config.py里的参数就能适配不同的抓取需求。想换数据源改fetcher.py想换存储方式改storage.py想加新的数据字段在fetcher里扩展。5.4 数据质量校验抓完之后一定要做质量校验不然拿着脏数据做分析结论全是错的。我一般检查这几项def validate_data(df, symbol): 数据质量校验 issues [] # 检查是否有数据 if df is None or df.empty: return [f{symbol}: 数据为空] # 检查日期是否连续排除周末和节假日 df df.sort_values(date) date_diff df[date].diff().dt.days if (date_diff 10).any(): issues.append(f{symbol}: 存在超过10天的数据缺口) # 检查价格逻辑 if (df[high] df[low]).any(): issues.append(f{symbol}: 最高价低于最低价) if (df[close] 0).any(): issues.append(f{symbol}: 存在非正收盘价) # 检查涨跌幅异常 if pct_change in df.columns: abnormal df[df[pct_change].abs() 21] if len(abnormal) 0: issues.append(f{symbol}: 存在{len(abnormal)}条涨跌幅异常记录) return issues这个校验函数会返回所有发现的问题跑完批量抓取后统一执行一遍把有问题的股票挑出来人工检查。我实测下来免费数据源大概有1%到3%的股票会存在各种小问题提前发现比事后返工强得多。6. 抓取之外数据后续使用的一些经验数据抓下来只是第一步怎么用好才是关键。我分享几个实际使用中的体会。数据对齐问题。不同股票的历史数据长度不一样有的上市早有的上市晚。做多股票对比分析时需要先对齐日期通常用pivot或者merge操作。我一般会构建一个统一的交易日历把所有股票数据reindex到这个日历上缺失的用NaN填充。性能优化。当股票数量到几千只时逐个读取CSV会很慢。这时候要么把数据合并成一个大文件要么用数据库。我试过把5000只股票合并成一个Parquet文件读取速度比逐个读CSV快了几十倍。数据更新策略。历史数据不是抓一次就完事了每天都有新数据产生。我的做法是每周更新一次只抓最近一周的数据追加到已有文件里。这样既保证了数据新鲜度又不会浪费太多时间。版本管理。数据源会变接口会变抓取逻辑也会变。我习惯给每次抓取的数据打上版本标签记录抓取日期和数据源版本。这样当发现数据问题时能快速定位是哪次抓取出的问题。最后说一个我踩过的大坑。早期我抓数据没有做去重结果某次接口返回了重复数据我没发现直接拿去做回测回测结果好得离谱后来排查了半天才发现是重复数据导致的。从那以后我每次抓完数据第一件事就是drop_duplicates这个习惯救了我很多次。数据抓取这件事技术门槛不高但细节特别多。把上面这些点都注意到基本能覆盖90%以上的场景。剩下的10%就得靠实际跑起来之后慢慢积累了。遇到问题不要慌先看日志再对比数据最后查接口文档大部分问题都能定位到。