A股历史数据的获取看起来是个再普通不过的需求但真正动过手的人都知道这里面藏着的坑比想象中多得多。行情软件导出的CSV格式五花八门不同数据源复权方式不统一停牌日的处理逻辑各家不一样指数成分股还会随时间调整。我前后做过好几个量化回测和数据分析的小项目每次重新拉数据都会遇到新问题。这篇内容就把我这些年抓取A股个股和指数历史数据的完整经验梳理一遍从数据源选型、接口调用、复权处理、批量抓取策略到数据校验尽量把每个环节的决策逻辑和踩坑细节都讲清楚。不管你是刚接触量化、想拿历史数据跑个简单回测的新手还是已经用过一些接口但总觉得数据对不上的老手应该都能从中找到有用的东西。1. 数据源选型免费接口的能力边界在哪里1.1 常见免费数据源的横向对比做A股历史数据抓取第一步永远是选数据源。市面上能用的免费接口就那么几个各有各的脾气。我整理了一张对比表方便你根据自己的需求快速定位数据源个股日线指数日线分钟线复权支持历史深度调用限制某财经数据接口A支持支持有限前复权/后复权/不复权1990年代至今有频率限制某财经数据接口B支持支持支持前复权/后复权2000年代至今较宽松某开源数据包支持支持不支持前复权/后复权依赖上游无显式限制某交易所公开数据支持支持不支持不复权较完整需自行解析这张表里的信息是基于我实际使用后的感受总结的具体细节可能随时间变化但大方向不会差太多。选哪个核心看三件事你需要多深的历史、要不要分钟级数据、复权处理是否必须。1.2 为什么我最终选择了组合方案单靠一个数据源很难覆盖所有需求。我的做法是日线级别数据用某开源数据包做主力因为它封装好了复权逻辑调用简单返回的就是标准DataFrame省去了大量清洗工作。指数数据用财经接口补充因为开源包对指数的覆盖有时候不够全特别是某些行业指数和自定义指数。分钟线数据单独走一个接口只在需要做日内策略回测时才拉取。这样组合的好处是日常回测用开源包就够了速度快、代码简洁遇到开源包拉不到的数据再用接口补分钟线这种重资源的需求单独处理不拖累主流程。注意不管你用哪个数据源都建议先拉一小段数据做交叉验证。不同源之间的收盘价可能有几分钱的差异这很正常但如果差异超过1%那就要查清楚是哪边的复权处理出了问题。1.3 环境准备与依赖安装不管你选哪个方案Python环境是基础。我习惯用conda建独立环境避免包冲突conda create -n astock python3.10 conda activate astock pip install pandas numpy requests akshare tushare这里解释一下每个包的作用。pandas和numpy是数据处理的基础不用多说。requests用于直接调HTTP接口。akshare是我常用的开源数据包覆盖了股票、基金、期货等多个品种。tushare是另一个可选方案它的优势在于数据规范程度高但部分接口需要积分。安装完成后建议先跑一个简单的测试确认数据能正常拉取import akshare as ak # 测试拉取某只个股的日线数据 df ak.stock_zh_a_hist(symbol000001, perioddaily, start_date20240101, end_date20241231, adjustqfq) print(df.head()) print(df.shape)如果这段代码能正常输出数据说明环境没问题。如果报错大概率是网络问题或者包版本不兼容先升级到最新版试试。2. 个股历史数据抓取从单只到批量的完整实现2.1 单只个股日线数据的获取与字段解读先看最基础的操作——拉一只股票的全部历史日线。以某只虚构的制造业个股为例import akshare as ak import pandas as pd def fetch_stock_daily(symbol, start_date, end_date, adjustqfq): 获取个股日线数据 symbol: 6位股票代码如 000001 adjust: qfq-前复权, hfq-后复权, -不复权 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustadjust ) # 重命名列为英文方便后续处理 df.columns [date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover] df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df df fetch_stock_daily(000001, 20200101, 20241231) print(df.tail())返回的字段里有几个容易混淆的地方需要说清楚。amplitude是振幅计算方式是最高价-最低价/昨收价。pct_change是涨跌幅百分比。change是涨跌额。turnover是换手率。这些字段在不同数据源里的命名可能不同但含义基本一致。2.2 前复权、后复权、不复权到底怎么选这是新手最容易迷糊的地方。我用一个生活化的类比来解释假设你持有一张100元的购物卡商场搞活动送了你20元卡里变成120元。这时候不复权就是直接记录120元前复权是把之前的100元也按比例调整让价格曲线连续后复权则是保持历史价格不变把后来的价格往上调整。具体到量化回测前复权适合看历史走势和技术指标因为价格曲线是连续的不会因为除权除息出现跳空。后复权适合计算真实收益率因为它保留了历史成本的真实购买力。不复权适合需要真实交易价格的场景比如计算手续费、印花税。我的习惯是做技术分析用前复权做收益回测用后复权做交易成本核算用不复权。三个都拉一份存着用的时候按需取。# 同时拉取三种复权方式的数据 df_qfq fetch_stock_daily(000001, 20200101, 20241231, adjustqfq) df_hfq fetch_stock_daily(000001, 20200101, 20241231, adjusthfq) df_raw fetch_stock_daily(000001, 20200101, 20241231, adjust)2.3 批量抓取全市场个股的工程化方案单只拉取很简单但你要做全市场回测就得面对几千只股票。这时候不能傻乎乎地循环调用得考虑效率和稳定性。我的方案是分三步走第一步获取全市场股票列表。def get_all_stock_codes(): 获取当前所有A股代码 df ak.stock_zh_a_spot_em() codes df[代码].tolist() return codes第二步设计带重试和限速的抓取函数。import time import random def fetch_with_retry(symbol, start_date, end_date, adjustqfq, max_retries3): 带重试机制的抓取 for attempt in range(max_retries): try: df fetch_stock_daily(symbol, start_date, end_date, adjust) return df except Exception as e: if attempt max_retries - 1: print(f{symbol} 抓取失败: {e}) return None time.sleep(random.uniform(1, 3)) return None第三步批量执行并保存。def batch_fetch(codes, start_date, end_date, save_dir./data): import os os.makedirs(save_dir, exist_okTrue) for i, code in enumerate(codes): filepath os.path.join(save_dir, f{code}.csv) if os.path.exists(filepath): continue # 断点续传 df fetch_with_retry(code, start_date, end_date) if df is not None: df.to_csv(filepath, indexFalse) # 每抓10只休息一下 if (i 1) % 10 0: time.sleep(random.uniform(2, 5)) if (i 1) % 100 0: print(f已完成 {i1}/{len(codes)})这个方案的核心是断点续传和随机延迟。断点续传保证你中断后不用从头再来随机延迟则是为了降低被目标服务器限制的概率。实测下来全市场5000多只股票日线数据全部拉完大概需要几个小时具体取决于网络状况和接口响应速度。实操心得不要一次性把几千只股票全部拉完再存而是边拉边存。我早期写过一个版本是全部拉完再统一保存结果跑到一半程序崩了前面几个小时白费。后来改成每只股票单独存一个CSV稳得多。3. 指数历史数据抓取个股之外的另一个战场3.1 常见指数代码与获取方式指数数据和个股数据在接口层面是分开的。常见的指数包括上证系列、深证系列、沪深300、中证500、创业板指等。获取方式和个股类似但代码格式不同def fetch_index_daily(symbol, start_date, end_date): 获取指数日线数据 symbol: 指数代码如 000001上证指数、399001深证成指 df ak.stock_zh_index_daily_em(symbolsymbol, start_datestart_date, end_dateend_date) df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df # 拉取沪深300 df_hs300 fetch_index_daily(000300, 20200101, 20241231) print(df_hs300.tail())指数数据一般不需要复权处理因为指数本身是计算出来的不存在除权除息的问题。但要注意不同数据源对指数代码的命名规则可能不同有的用带前缀的格式如sh000001有的用纯数字。用之前先确认清楚。3.2 指数成分股数据的获取与动态调整做指数增强策略或者行业轮动光有指数点位不够还需要成分股列表。这个数据比指数本身复杂得多因为成分股会定期调整。def get_index_components(index_code): 获取指数成分股 df ak.index_stock_cons_csindex(symbolindex_code) return df这里有个关键问题你拿到的是当前成分股不是历史成分股。如果你用当前的成分股去回测三年前的策略会产生严重的幸存者偏差。比如某只股票三年前还没进指数但你现在用当前成分股列表去回测就相当于提前知道了它会进指数回测结果会虚高。处理这个问题有两个思路。一是找提供历史成分股的数据源但免费渠道很少。二是接受这个偏差但在回测报告里明确说明。我一般用第二种因为对于个人研究来说精确的历史成分股数据获取成本太高。3.3 指数与个股数据的对齐处理当你同时有个股和指数数据时经常需要把它们对齐到同一个时间轴上。这里有个细节个股有停牌指数没有。所以对齐的时候不能简单用日期做内连接否则停牌期间的数据会丢失。我的做法是def align_stock_index(stock_df, index_df): 将个股数据对齐到指数交易日 # 以指数交易日为准左连接个股数据 merged index_df[[date, close]].rename(columns{close: index_close}) merged merged.merge(stock_df[[date, close]].rename(columns{close: stock_close}), ondate, howleft) # 停牌日用前值填充 merged[stock_close] merged[stock_close].ffill() return merged停牌日用前值填充是常见做法但要注意如果停牌时间很长填充出来的数据会失真。一般停牌超过一周的我会在回测时标记出来单独处理。4. 数据清洗与质量校验让数据真正能用4.1 常见数据问题与清洗规则原始数据拉下来不能直接用。我总结了几类常见问题问题类型表现处理方式缺失值某天数据为空区分停牌和真缺失停牌用前值填充真缺失标记异常值涨跌幅超过10%非ST核对是否除权日确认后修正重复行同一日期出现两次去重保留第一条价格为零收盘价为0通常是数据错误删除该行日期不连续缺少交易日与交易日历比对补全或标记清洗代码示例def clean_stock_data(df): 清洗个股数据 # 去重 df df.drop_duplicates(subset[date], keepfirst) # 删除价格为0的行 df df[df[close] 0] # 删除涨跌幅异常的行非ST股票涨跌幅超过11% df df[df[pct_change].abs() 11] # 按日期排序 df df.sort_values(date).reset_index(dropTrue) return df4.2 交易日历的构建与使用A股交易日历不是简单的周一到周五还有节假日。构建准确的交易日历是数据对齐的基础。def get_trade_calendar(start_date, end_date): 获取交易日历 df ak.tool_trade_date_hist_sina() df[trade_date] pd.to_datetime(df[trade_date]) mask (df[trade_date] start_date) (df[trade_date] end_date) return df[mask][trade_date].tolist()有了交易日历你就可以检查数据是否完整。比如你拉了某只股票2024年全年的数据理论上应该有约242个交易日的数据如果实际只有200条那就要查是停牌还是数据缺失。4.3 数据校验的自动化脚本每次拉完数据都手动检查不现实我写了一个自动校验脚本def validate_data(df, symbol, trade_dates): 校验数据完整性 issues [] # 检查日期是否都在交易日历中 invalid_dates set(df[date]) - set(trade_dates) if invalid_dates: issues.append(f包含非交易日: {invalid_dates}) # 检查是否有重复日期 if df[date].duplicated().any(): issues.append(存在重复日期) # 检查价格逻辑最高价 收盘价 最低价 if not (df[high] df[close]).all(): issues.append(最高价小于收盘价) if not (df[close] df[low]).all(): issues.append(收盘价小于最低价) # 检查涨跌幅与价格变化是否一致 df[calc_pct] df[close].pct_change() * 100 diff (df[calc_pct] - df[pct_change]).abs() if (diff 0.5).any(): issues.append(涨跌幅与价格变化不一致) if issues: print(f{symbol} 数据问题: {issues}) return issues这个脚本帮我发现过好几次数据问题。有一次某只股票的数据里连续几天的收盘价完全一样查下来是数据源那边出了问题重新拉取后就正常了。5. 存储方案与增量更新策略5.1 文件存储 vs 数据库存储的选择数据量小的时候CSV文件足够用。但当你有了全市场几年的日线数据CSV的读写效率就成了瓶颈。我对比过几种方案存储方式优点缺点适用场景CSV简单、通用读写慢、占空间少量数据、临时分析Parquet压缩率高、读写快需要额外依赖中等规模、频繁读取SQLite单文件、支持SQL并发差个人项目、结构化查询PostgreSQL功能强、并发好需要部署维护团队协作、大规模我现在的做法是原始数据存Parquet分析时读入内存。Parquet的列式存储对只读几列的场景特别友好速度比CSV快好几倍。# 存为Parquet df.to_parquet(f./data/{symbol}.parquet, indexFalse) # 读取 df pd.read_parquet(f./data/{symbol}.parquet)5.2 增量更新的实现逻辑全量拉取一次之后后续只需要更新最新几天的数据。增量更新的核心是记录每只股票最后更新的日期def incremental_update(symbol, save_dir./data): 增量更新单只股票数据 filepath os.path.join(save_dir, f{symbol}.parquet) if os.path.exists(filepath): old_df pd.read_parquet(filepath) last_date old_df[date].max() start_date (last_date pd.Timedelta(days1)).strftime(%Y%m%d) else: old_df None start_date 20200101 end_date pd.Timestamp.now().strftime(%Y%m%d) if start_date end_date: return # 已是最新 new_df fetch_with_retry(symbol, start_date, end_date) if new_df is not None and len(new_df) 0: if old_df is not None: combined pd.concat([old_df, new_df], ignore_indexTrue) combined combined.drop_duplicates(subset[date], keeplast) else: combined new_df combined.to_parquet(filepath, indexFalse)这个逻辑看起来简单但有个坑除权除息会导致历史数据变化。如果你用前复权每次除权后之前的所有历史价格都会变。所以增量更新前复权数据时不能只追加新数据而应该重新拉取全量。后复权和不复权数据则不受影响可以安全地增量更新。实操心得我的做法是不复权和后复权数据用增量更新前复权数据每隔一段时间全量刷新一次。这样既保证了效率又避免了复权导致的偏差。5.3 数据版本管理的小技巧数据源会变接口会调整今天拉的数据和明天拉的可能有细微差异。我习惯在保存数据时加一个元数据文件记录拉取时间、数据源、复权方式等信息import json from datetime import datetime def save_with_metadata(df, symbol, save_dir./data, adjustqfq): filepath os.path.join(save_dir, f{symbol}.parquet) df.to_parquet(filepath, indexFalse) meta { symbol: symbol, adjust: adjust, last_update: datetime.now().isoformat(), rows: len(df), start_date: str(df[date].min()), end_date: str(df[date].max()) } meta_path os.path.join(save_dir, f{symbol}_meta.json) with open(meta_path, w) as f: json.dump(meta, f, ensure_asciiFalse, indent2)这个习惯看起来多余但当你几个月后回头看数据想确认当时用的是什么复权方式、数据截止到哪天这个元数据文件就派上用场了。6. 实战中踩过的坑与应对方案6.1 接口频率限制与反爬应对免费接口都有频率限制超了要么返回空数据要么直接封IP。我踩过最惨的一次是写了个循环没有加延迟跑了不到200只股票就被限制了等了半天才恢复。应对方案就三条加延迟、加重试、加断点续传。延迟建议用随机值不要用固定值因为固定间隔的请求模式很容易被识别。重试要设置最大次数避免死循环。断点续传保证中断后不用重来。# 推荐的延迟策略 time.sleep(random.uniform(0.5, 2.0)) # 每抓N只后额外休息 if (i 1) % 50 0: time.sleep(random.uniform(5, 10))6.2 复权数据不一致的排查过程有一次我用两个不同数据源拉同一只股票的前复权数据发现某天的收盘价差了3%以上。排查过程是这样的第一步拉取不复权数据对比发现不复权数据是一致的。说明问题出在复权因子上。第二步找到该股票的分红送配记录发现那天正好是除权除息日。第三步分别计算两个数据源的复权因子发现一个用的是除权日前收盘价计算另一个用的是除权日当天开盘价计算。两种算法都有道理但结果不同。第四步确认哪个数据源的算法更符合我的回测需求最终选择了用前收盘价计算的那个。这个经历告诉我不同数据源的复权结果可能有差异做回测时一定要固定用一个数据源不要混用。6.3 停牌、退市股票的数据处理全市场回测绕不开停牌和退市股票。停牌股票的数据会有缺失退市股票则直接没有后续数据。处理方式停牌用前值填充但在回测时标记出来避免在停牌期间产生交易信号。退市保留历史数据但在股票池中剔除。回测时要注意不能用当前还在交易的股票列表去回测历史否则会有幸存者偏差。def handle_suspended(df, trade_dates): 处理停牌数据 # 补全缺失的交易日 full_dates pd.DataFrame({date: trade_dates}) merged full_dates.merge(df, ondate, howleft) # 标记停牌日 merged[is_suspended] merged[close].isna() # 前值填充 merged[close] merged[close].ffill() return merged6.4 数据量增大后的性能优化当你的股票池从几十只扩大到几千只性能问题就来了。我做过几个优化第一用Parquet替代CSV读取速度提升明显。第二只读需要的列比如只做收盘价分析就只读date和close两列。第三用多进程并行抓取但要注意控制并发数别把接口打挂了。from concurrent.futures import ThreadPoolExecutor def parallel_fetch(codes, max_workers5): with ThreadPoolExecutor(max_workersmax_workers) as executor: results list(executor.map(fetch_with_retry, codes)) return results并发数建议控制在3到5之间太高容易触发限制太低效率上不去。这个平衡点需要根据实际接口的承受能力来调。6.5 数据更新时机的选择什么时候更新数据也有讲究。收盘后更新当然可以但要注意有些数据源在收盘后一段时间内数据还没完全更新。我的经验是交易日晚上7点以后更新当天的数据比较稳妥。如果做日内策略需要盘中数据那就得用分钟线接口日线接口盘中拉到的当天数据可能不完整。另外周末和节假日不要更新因为不会有新数据白白浪费请求次数。可以在脚本里加一个判断def is_trade_day(date): 判断是否为交易日 trade_dates get_trade_calendar(20200101, 20301231) return pd.Timestamp(date) in trade_dates7. 从数据到策略一个完整的回测数据准备示例7.1 构建回测数据集把前面所有环节串起来一个完整的回测数据准备流程是这样的def prepare_backtest_data(stock_codes, index_code, start_date, end_date): 准备回测数据集 # 1. 获取交易日历 trade_dates get_trade_calendar(start_date, end_date) # 2. 获取指数数据 index_df fetch_index_daily(index_code, start_date, end_date) # 3. 批量获取个股数据 stock_data {} for code in stock_codes: df fetch_with_retry(code, start_date, end_date, adjusthfq) if df is not None: df clean_stock_data(df) stock_data[code] df # 4. 对齐到统一时间轴 aligned_data {} for code, df in stock_data.items(): aligned align_stock_index(df, index_df) aligned_data[code] aligned return aligned_data, index_df7.2 数据质量报告在正式回测前生成一份数据质量报告确认数据可用def generate_quality_report(aligned_data, trade_dates): 生成数据质量报告 report [] for code, df in aligned_data.items(): total_days len(trade_dates) actual_days df[stock_close].notna().sum() suspended_days df[is_suspended].sum() report.append({ code: code, total_days: total_days, actual_days: actual_days, suspended_days: suspended_days, coverage: f{actual_days/total_days*100:.1f}% }) report_df pd.DataFrame(report) print(report_df.describe()) return report_df覆盖率低于90%的股票建议在回测中剔除否则停牌期间填充的数据会严重影响策略表现。7.3 常见回测数据陷阱最后说几个回测中容易踩的数据陷阱陷阱一未来函数。用了当天收盘后才知道的信息去模拟当天交易。比如用当天的最高价作为买入价但你在盘中根本不知道最高价是多少。陷阱二幸存者偏差。只用当前还在交易的股票回测忽略了已经退市的股票。这会让回测收益虚高。陷阱三复权方式不一致。个股用前复权指数用不复权两者对比时会产生偏差。陷阱四停牌期间交易。停牌股票被前值填充后价格曲线是平的如果策略在此时产生交易信号实际是无法成交的。避开这些陷阱你的回测结果才靠谱。数据准备阶段多花点时间做校验比回测跑完发现结果不对再回头查要省事得多。我在实际使用中发现数据抓取这件事代码写起来不难难的是对各种边界情况的处理。停牌、除权、退市、数据源差异每一个都可能让你的分析结果出现偏差。建议刚开始做的时候先用少量股票跑通全流程确认数据质量没问题再扩展到全市场。另外养成记录数据来源和更新时间的习惯过几个月回头看你会感谢自己当初多写了那几行元数据。