A股历史数据获取方法：从数据源选型到本地落地的完整路径

简介这份项目源码面向个人投资者、量化初学者与金融数据研究者聚焦A股历史行情获取这一常见需求整理了Tushare、AkShare与Baostock三种免费数据平台的调用方式与示例代码帮助读者在缺乏付费数据源的情况下快速搭建可用的数据获取流程。资源包共5个文件包含Python脚本、Markdown说明、HTML演示页、依赖清单及配置文件压缩后约8KB体积轻量便于直接运行与二次修改。目前已有1328人学习下载说明其在量化入门场景中具有一定参考价值。读者可从中获得各平台接口的注册与登录要点、日线级别数据的获取示例、依赖安装说明以及关于数据质量与使用限制的提醒适合作为投资分析或研究工作的起步工具也能为后续策略回测与数据清洗提供基础素材。1. A股历史数据获取方法从数据源选型到本地落地的完整路径做量化回测、因子挖掘或者只是想把K线拉下来做统计绕不开的第一道坎就是历史行情数据从哪来。很多人一上来就找“最全的接口”结果要么被限流卡死要么拿到一堆复权口径混乱的数据回测跑出来的曲线跟实盘对不上。A股历史数据获取方法这件事核心不是“能不能拿到”而是“拿到的数据能不能直接用于计算”。我在几个模拟项目里反复折腾过不同方案最后稳定下来的组合是日线级别用轻量接口批量拉取分钟级别用本地缓存加增量更新财务和除权信息单独维护一张对照表。这套方法适合个人开发者、小团队做策略验证也适合需要把数据落到本地数据库做二次加工的场景。下面把选型理由、代码实现和踩过的坑一次讲清楚。2. 数据源选型免费接口、本地缓存与增量更新怎么搭2.1 三类常见数据源的适用边界市面上能拿到A股历史数据的途径大致分三类公开HTTP接口、券商/数据商提供的本地文件、以及自己从行情软件导出。公开接口里常见做法是用一些开源库封装的日线接口优点是调用简单、字段规整缺点是分钟级数据要么没有要么限制条数。本地文件类通常是按日期打包的压缩包适合一次性建库但更新需要自己写脚本。行情软件导出适合小样本核对不适合自动化。我一般会按数据频率分两条线日线及以下频率周线、月线直接用HTTP接口拉全量因为数据量小几千只股票几年数据也就几十兆分钟线则先用本地文件建底仓之后每天收盘后跑增量脚本补当天数据。这样既避开了接口对高频数据的限制又保证了历史深度。选型时重点看三个指标复权方式是否可指定、字段是否包含成交额和换手率、接口是否支持按日期区间批量拉取。很多接口默认只给不复权价格回测时如果忘了处理除权除息长周期策略会严重失真。2.2 用Python拉取日线数据的最小可用脚本下面这段代码用常见的开源数据接口拉取单只股票的日线数据并落成CSV。注意这里用的是前复权口径适合做回测如果你要做估值分析应该换成后复权或不复权。import pandas as pd import akshare as ak # 常见开源数据接口库 def fetch_daily(symbol: str, start: str, end: str, adjust: str qfq) - pd.DataFrame: 拉取A股日线数据 symbol: 6位代码如 000001 start/end: YYYYMMDD adjust: qfq前复权, hfq后复权, 不复权 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustadjust ) # 统一列名方便后续入库 df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 换手率: turnover }) df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df if __name__ __main__: data fetch_daily(000001, 20200101, 20241231, adjustqfq) data.to_csv(000001_daily_qfq.csv, indexFalse) print(data.tail())逻辑说明接口返回的原始列名是中文先重命名成英文统一字段避免后续换数据源时改代码。adjust参数控制复权方式前复权适合回测因为价格连续不会因为除权出现跳空。sort_values保证按日期升序很多接口返回是倒序的不排直接算收益率会出错。参数上start_date和end_date用字符串格式不要传datetime部分接口对类型敏感。如果拉全市场股票建议把代码列表单独存一个CSV用循环加time.sleep(0.5)控制频率否则容易被限流。2.3 分钟线数据的本地建库与增量更新分钟线数据量大单只股票一年的1分钟线大约6万条全市场5000只就是3亿条。直接每次从接口拉不现实常见做法是先下载历史分钟数据包通常按季度或按月打包解压后统一转成Parquet格式按股票代码分区存储之后每天收盘后跑增量脚本只拉当天数据追加。import os import pandas as pd from pathlib import Path def append_minute_data(symbol: str, new_df: pd.DataFrame, base_dir: str ./minute_data): 增量追加分钟数据按股票代码分目录存储 stock_dir Path(base_dir) / symbol stock_dir.mkdir(parentsTrue, exist_okTrue) file_path stock_dir / 1min.parquet if file_path.exists(): old_df pd.read_parquet(file_path) # 去重按时间戳保留最新一条 combined pd.concat([old_df, new_df], ignore_indexTrue) combined combined.drop_duplicates(subset[datetime], keeplast) combined combined.sort_values(datetime).reset_index(dropTrue) else: combined new_df.sort_values(datetime).reset_index(dropTrue) combined.to_parquet(file_path, indexFalse) return len(combined) # 假设new_df是从接口拉到的当天分钟数据包含datetime列 # append_minute_data(000001, new_df)这里用Parquet而不是CSV原因是Parquet列式存储压缩率高读取时只加载需要的列速度比CSV快一个数量级。按股票代码分目录避免单文件过大。去重逻辑用drop_duplicates按时间戳保留最后一条防止重复拉取导致数据膨胀。增量更新时注意时区问题接口返回的时间戳有时是UTC有时是北京时间入库前统一转成Asia/Shanghai否则和日线数据对齐时会错位。3. 数据清洗与复权处理把原始数据变成可回测的格式3.1 除权除息数据的获取与复权因子计算复权不是简单地把价格乘一个系数而是要根据除权除息事件逐日调整。常见做法是先拉取不复权价格和除权除息记录每股分红、送转比例然后计算复权因子。前复权因子 累计除权调整系数后复权因子 累计再投资系数。def calc_adjust_factor(dividends: pd.DataFrame, prices: pd.DataFrame) - pd.DataFrame: dividends: 包含 ex_date, cash_div, stock_div 的DataFrame prices: 包含 date, close 的DataFrame 返回带前复权因子的DataFrame prices prices.sort_values(date).copy() prices[factor] 1.0 for _, row in dividends.iterrows(): ex_date row[ex_date] cash row[cash_div] # 每股现金分红 stock row[stock_div] # 每股送转比例 # 除权除息日前一天的价格 mask prices[date] ex_date if not mask.any(): continue prev_close prices.loc[mask, close].iloc[-1] # 除权价 (前收盘 - 现金分红) / (1 送转比例) ex_price (prev_close - cash) / (1 stock) ratio ex_price / prev_close # 前复权除权日之前的价格乘以ratio prices.loc[mask, factor] * ratio prices[adj_close] prices[close] * prices[factor] return prices逻辑说明遍历每个除权除息日找到该日之前的所有交易日把前复权因子乘上除权比例。这样越早的价格被调整得越多最新价格保持不变。参数上cash_div和stock_div需要从分红送转数据里取注意单位是“每股”还是“每10股”很多数据源给的是每10股要除以10。如果不想自己算很多接口直接提供前复权价格但口径可能不一致。我一般会自己算一遍和接口给的前复权价格对比偏差超过0.1%就检查除权记录是否漏了。3.2 停牌、涨跌停与异常值的处理A股停牌期间没有行情但回测时如果直接跳过停牌日持仓的市值计算会出错。常见做法是停牌日保留前收盘价成交量设为0并在策略里标记该日不可交易。def fill_suspended_dates(df: pd.DataFrame, trade_calendar: pd.DatetimeIndex) - pd.DataFrame: 用交易日历对齐停牌日填充前收盘价成交量置0 df df.set_index(date).reindex(trade_calendar) df[close] df[close].ffill() df[open] df[open].fillna(df[close]) df[high] df[high].fillna(df[close]) df[low] df[low].fillna(df[close]) df[volume] df[volume].fillna(0) df[amount] df[amount].fillna(0) df[is_suspended] df[volume] 0 return df.reset_index().rename(columns{index: date})涨跌停处理更微妙涨停时买不进跌停时卖不出。回测里通常用close high low且涨幅达到限制来标记一字板然后在撮合时跳过。注意ST股票涨跌停幅度是5%科创板是20%这些规则要单独维护一张表。异常值方面最常见的是成交量单位不一致有的接口给“手”有的给“股”。统一成“股”再入库否则换手率算出来差100倍。4. 避坑与排查数据获取路上最容易翻车的五个点4.1 接口限流导致数据缺失现象批量拉取时前面几百只正常后面突然大量返回空DataFrame。原因多数免费接口有每分钟调用次数限制超了直接返回空或报错但代码没做异常捕获空数据被当成正常结果存下来。解决在循环里加try/except对空结果重试三次每次间隔递增同时记录失败列表事后单独补拉。4.2 复权口径混用导致回测失真现象用前复权价格算收益率但分红再投资没算进去长周期策略年化收益偏低。原因前复权只调整价格不模拟分红再买入。解决如果策略涉及分红再投资应该用后复权价格或者自己维护一个总收益指数。我一般回测用前复权但会在策略里单独加分红收益。4.3 时间戳时区不一致现象分钟数据和日线数据合并时日期对不上夜盘或早盘数据跑到前一天。原因部分接口返回UTC时间部分返回北京时间直接pd.to_datetime不指定时区会按本地时区解析。解决统一用pd.to_datetime(..., utcTrue).dt.tz_convert(Asia/Shanghai)再取日期。4.4 股票代码变更与退市处理现象回测某只股票时中途代码变了比如从“000001”变成“000001.SZ”或者退市后数据突然中断。原因数据源对代码格式不统一退市股票不再更新。解决维护一张代码映射表记录历史代码和当前代码的对应关系退市股票保留最后交易日之前的数据并在策略里标记不可交易。4.5 本地存储格式选错导致读取慢现象用CSV存分钟数据回测时读取一只股票要几秒全市场跑一遍要几小时。原因CSV是文本格式解析慢且不支持列裁剪。解决改用Parquet或HDF5按股票代码分区读取时只加载需要的列。实测同样数据量Parquet读取速度是CSV的5到10倍。5. 进阶技巧用DuckDB做本地查询与因子计算数据落到本地后下一步是快速查询和计算因子。传统做法用pandas但数据量上到千万行级别pandas内存吃紧。我最近的习惯是用DuckDB它可以直接查Parquet文件不用把数据全加载到内存而且支持SQL窗口函数算滚动因子很方便。import duckdb # 直接查询Parquet文件不用先读进pandas con duckdb.connect() query SELECT symbol, date, close, AVG(close) OVER (PARTITION BY symbol ORDER BY date ROWS 20 PRECEDING) AS ma20, STDDEV(close) OVER (PARTITION BY symbol ORDER BY date ROWS 20 PRECEDING) AS std20 FROM read_parquet(./daily_data/*/*.parquet) WHERE date 2024-01-01 df con.execute(query).fetchdf()这段SQL直接对Parquet目录做通配查询DuckDB会自动并行读取。ROWS 20 PRECEDING是窗口帧算20日均线和标准差。比pandas的rolling快很多而且内存占用低。参数上read_parquet支持通配符按目录分区时可以用hive_partitioning1自动解析分区列。另一个技巧是增量更新时用DuckDB的INSERT INTO直接追加避免全量重写。建一张持久化的表每天把新数据INSERT进去查询时走索引。这样即使数据量到亿级单只股票的查询也能在毫秒级返回。最后说一个我踩过的坑DuckDB默认对时间戳类型很敏感如果Parquet里存的是字符串日期窗口函数排序会按字典序而不是时间序。入库前一定把日期列转成DATE或TIMESTAMP类型否则算出来的均线是乱的。这个坑我翻了两次车后来在写入Parquet时就强制astype(datetime64[ns])再没出过问题。希望帮到你。本文还有配套的精品资源点击获取