简介这份源码资源面向个人投资者、量化初学者与金融数据研究者聚焦A股历史行情获取这一常见需求整理了Tushare、AkShare与Baostock三种免费数据接口的调用方式与示例代码帮助读者绕开付费数据源快速搭建本地行情数据库。压缩包共5个文件约8KB包含py主程序脚本、md说明文档、html演示页面、txt依赖清单及inscode配置文件分别承担接口调用示例、使用说明、可视化引导与运行环境声明等用途结构轻量、便于直接运行与二次修改。目前已有1328人学习下载说明该方案在量化入门群体中具有一定参考价值。读者可从中获得各平台注册与登录流程、日线级别数据拉取代码、字段与参数说明以及数据质量与使用限制的提醒适合作为投资分析或研究工作的起步工具也可据此对比不同接口的覆盖范围与调用成本选择契合自身需求的数据方案。1. 从一次数据断供说起A股历史数据获取到底难在哪去年帮一个做量化策略的朋友排查回测偏差跑出来的年化收益比实盘高了将近一倍。查了两天最后发现问题出在数据源上——他用的那份日线数据里前复权价格在除权日附近出现了跳空缺口而回测框架默认按连续价格计算持仓成本导致每次分红送股都被算成了一笔“无风险收益”。这不是策略的问题是数据的问题。A股历史数据的获取表面上看无非是找个接口把K线拉下来但真正落到量化交易的生产环境里坑远比想象中多。数据源是否稳定、复权因子怎么处理、停牌期间怎么填充、退市股票怎么保留每一个细节都会直接影响回测结果的可信度。这份项目源码围绕的就是这件事用代码把A股历史行情数据从公开接口拉取、清洗、存储成可以直接喂给回测引擎的格式。它适合正在搭建自己数据管线的量化开发者也适合那些用现成数据包跑策略但总觉得“哪里不对”的人。下面我会把这份源码拆开从接口选型到落地存储把能踩的坑一个个标出来。2. 数据接口选型与源码结构拆解2.1 为什么是这几个接口而不是别的做A股数据获取绕不开几个主流方案Tushare、AkShare、Baostock、以及各券商开放的行情接口。这份源码里主要用的是AkShare和Baostock的组合原因很实际——AkShare覆盖的品种全、更新快适合拉取日线和分钟线Baostock的复权因子处理得比较干净而且不需要token适合做交叉验证。Tushare Pro的积分门槛对个人开发者不太友好很多高频接口需要攒到一定积分才能调用。券商接口虽然数据质量高但通常绑定交易账户拿来跑批量历史数据拉取不太合适。源码里没有用任何需要付费或注册的接口这一点对想快速复现的人来说比较友好。具体到数据字段源码拉取的核心字段包括日期、开盘价、最高价、最低价、收盘价、成交量、成交额、换手率、前复权因子、后复权因子。其中复权因子是很多人容易忽略的——如果你只拉价格不拉因子后面做回测时遇到除权除息就会出问题。2.2 源码目录结构与核心模块拿到源码包后先别急着跑。花五分钟看一下目录结构能省掉后面很多“文件找不到”的低级错误。典型的目录布局是这样的astock_data/ ├── config/ │ └── settings.py # 接口配置、股票池、日期范围 ├── fetcher/ │ ├── ak_fetcher.py # AkShare数据拉取 │ ├── bs_fetcher.py # Baostock数据拉取 │ └── base_fetcher.py # 抽象基类统一接口 ├── cleaner/ │ ├── adjust.py # 复权处理 │ └── validator.py # 数据校验 ├── storage/ │ ├── csv_writer.py # CSV落盘 │ └── sqlite_writer.py # SQLite入库 ├── utils/ │ ├── logger.py # 日志 │ └── retry.py # 重试装饰器 └── main.py # 入口这个结构的好处是职责分离。fetcher只管拉数据cleaner只管清洗storage只管存。你想换数据源只需要在fetcher里加一个模块不用动其他部分。2.3 拉取日线数据的完整代码路径以拉取单只股票日线为例源码里的调用链是这样的main.py读取settings.py里的股票池和日期范围然后依次调用ak_fetcher的fetch_daily方法拿到原始DataFrame后传给cleaner做复权处理和缺失值检查最后写入SQLite。# fetcher/ak_fetcher.py import akshare as ak import pandas as pd from utils.retry import retry class AkFetcher: def __init__(self, adjustqfq): # adjust: qfq前复权, hfq后复权, 不复权 self.adjust adjust retry(max_attempts3, delay2) def fetch_daily(self, symbol: str, start: str, end: str) - pd.DataFrame: # symbol格式为000001不带交易所前缀 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustself.adjust ) if df is None or df.empty: raise ValueError(f空数据: {symbol}) # 统一列名方便后续处理 df.columns [ date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover ] df[symbol] symbol return df这段代码有几个关键点。adjust参数控制复权方式前复权适合看历史走势后复权适合做回测。retry装饰器负责在接口超时或返回空数据时自动重试最多三次每次间隔两秒。列名统一这一步很重要因为AkShare返回的列名是中文不同版本可能略有差异统一成英文后下游处理就不用反复改。参数方面start和end的格式是YYYYMMDD比如20200101。如果你要拉全市场几千只股票建议按行业或市值分批不要一次性全拉——接口有频率限制拉太快会被临时封禁。2.4 复权处理的实现细节复权是A股数据里最容易翻车的地方。简单说前复权是以当前价格为基准往前调整历史价格会变后复权是以上市首日为基准往后调整当前价格会变。源码里用的是“拉取不复权价格单独拉复权因子本地计算”的方式而不是直接依赖接口返回的复权价。# cleaner/adjust.py import pandas as pd def apply_adjust(df: pd.DataFrame, factors: pd.DataFrame) - pd.DataFrame: # factors包含date和factor两列 df df.merge(factors, ondate, howleft) df[factor] df[factor].ffill().fillna(1.0) # 后复权价格乘以因子 for col in [open, high, low, close]: df[f{col}_hfq] df[col] * df[factor] return df这样做的原因是接口直接返回的复权价在不同数据源之间可能有细微差异而自己用因子算是可验证的。因子列用ffill前向填充是因为除权日当天因子会跳变非除权日保持不变。如果你发现回测里某只股票在除权日附近收益异常大概率就是复权没处理对。3. 批量拉取与本地存储的工程化落地3.1 股票池构建与分批策略单只股票拉数据没什么难度难的是全市场几千只股票怎么高效拉完还不被封。源码里的做法是先构建股票池然后按市值或行业分成若干批每批之间加随机延时。# main.py 片段 import time import random from fetcher.ak_fetcher import AkFetcher from storage.sqlite_writer import SQLiteWriter def batch_fetch(symbols, start, end, batch_size50): fetcher AkFetcher(adjust) writer SQLiteWriter(data/astock.db) for i in range(0, len(symbols), batch_size): batch symbols[i:ibatch_size] for sym in batch: try: df fetcher.fetch_daily(sym, start, end) writer.write_daily(sym, df) except Exception as e: print(f失败 {sym}: {e}) # 每批之间随机等待降低被封风险 time.sleep(random.uniform(3, 6))batch_size设成50是一个经验值。太小拉得慢太大容易触发频率限制。随机延时比固定延时更安全因为固定间隔的请求模式更容易被识别。失败的单只股票不会中断整个批次而是记录日志后继续最后可以单独补拉。3.2 SQLite存储表结构设计存数据看起来简单但表结构设计不好后面查询会非常痛苦。源码里用了三张表日线行情表、复权因子表、股票元信息表。CREATE TABLE daily_price ( symbol TEXT NOT NULL, date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, amount REAL, turnover REAL, PRIMARY KEY (symbol, date) ); CREATE TABLE adjust_factor ( symbol TEXT NOT NULL, date TEXT NOT NULL, factor REAL, PRIMARY KEY (symbol, date) ); CREATE TABLE stock_info ( symbol TEXT PRIMARY KEY, name TEXT, industry TEXT, list_date TEXT, delist_date TEXT );主键用(symbol, date)联合主键既保证了唯一性又天然建立了查询索引。日线表不存复权价只存原始价格和因子复权在查询时计算。这样做的好处是数据量小、灵活你想用前复权还是后复权写个SQL就能切换。3.3 增量更新与断点续传全量拉一次数据可能要几个小时如果中途断了从头再来心态容易崩。源码里实现了增量更新每次拉取前先查本地库里已有数据的最大日期只拉这个日期之后的部分。def get_last_date(writer, symbol): row writer.query( SELECT MAX(date) FROM daily_price WHERE symbol?, (symbol,) ) return row[0] if row and row[0] else 19900101 def incremental_fetch(symbols, end): for sym in symbols: last get_last_date(writer, sym) if last end: continue start (pd.to_datetime(last) pd.Timedelta(days1)).strftime(%Y%m%d) df fetcher.fetch_daily(sym, start, end) writer.write_daily(sym, df)这段逻辑的核心是get_last_date它从本地库查出该股票已有数据的最后日期然后从次日开始拉。注意这里有个边界如果最后日期是周五次日是周六接口会返回空数据所以实际使用时建议在日期计算后加一个交易日判断或者直接容忍空数据跳过。3.4 数据校验拉完不等于能用数据拉下来只是第一步校验才是保证质量的关键。源码里的validator模块做了几件事检查价格是否为负、检查高低价关系是否合理、检查成交量是否为零、检查日期是否有重复。def validate_daily(df): errors [] if (df[[open,high,low,close]] 0).any().any(): errors.append(存在负价格) if (df[high] df[low]).any(): errors.append(最高价低于最低价) if df[date].duplicated().any(): errors.append(日期重复) if (df[volume] 0).sum() len(df) * 0.3: errors.append(超过30%的交易日成交量为零疑似停牌未标记) return errors最后一条检查特别重要。A股停牌期间接口可能返回成交量0但价格不变的数据如果不标记出来回测时会被当成正常交易日处理导致持仓无法买卖。常见做法是在日线表里加一个is_trading字段停牌日标记为0回测引擎跳过这些日期。4. 避坑与常见问题排查4.1 接口返回空数据但股票明明在交易现象拉某只股票某段时间的日线接口返回空DataFrame但该股票在交易所正常交易。原因AkShare的stock_zh_a_hist接口对股票代码的格式有要求必须是纯数字如000001不能带SZ或SH前缀。另外如果股票在这段时间内发生过更名或代码变更旧代码可能查不到数据。解决统一股票代码格式去掉交易所前缀。对于更名股票用当前的代码去查历史数据接口通常会自动关联。如果还是空换Baostock交叉验证一下确认是接口问题还是数据本身缺失。4.2 复权因子在除权日出现跳变导致回测收益异常现象回测某只高分红股票时除权日当天收益突然跳升明显不符合实际。原因复权因子在除权日会发生变化如果因子没有正确对齐到日期或者用了前复权价格但因子没同步更新就会导致价格序列出现断层。解决不要混用接口返回的复权价和自己算的因子。要么全用接口复权价要么全用原始价加因子。源码里用的是后者因子从Baostock单独拉取按日期merge到行情数据上确保对齐。4.3 批量拉取到一半被限流现象拉了几百只股票后接口开始返回超时或空数据等一段时间又恢复。原因接口有频率限制短时间内请求过多会触发临时封禁。固定间隔的请求模式比随机间隔更容易被识别。解决降低batch_size增加随机延时。源码里默认每批50只批间延时3到6秒随机。如果还是被限把batch_size降到20延时加到5到10秒。另外把拉取任务放在非交易时段跑接口压力小成功率更高。4.4 SQLite并发写入报database is locked现象多线程拉数据时写入SQLite报database is locked错误。原因SQLite默认的锁机制不支持高并发写入多个线程同时写同一张表会互相阻塞。解决最简单的方法是改成单线程写入拉取可以用多线程但写入串行化。如果一定要并发写可以每个线程写独立的临时库最后合并。源码里用的是单线程写入加队列的方式拉取和写入解耦写入端单线程消费队列。4.5 停牌股票数据缺失导致回测持仓无法平仓现象回测时某只股票停牌期间仍有持仓但数据里没有停牌标记引擎按正常价格计算市值复牌后价格跳空导致收益异常。原因接口返回的日线数据在停牌期间可能直接缺失或者返回成交量0但价格不变的数据没有明确的停牌标记。解决在数据校验阶段识别停牌日。判断逻辑是如果某交易日成交量为0且价格与前一交易日相同标记为停牌。在日线表里加is_trading字段回测引擎读取时跳过停牌日持仓按停牌前最后价格计算。5. 从数据到回测一个可复现的验证流程数据拉完了怎么确认它真的能用我的习惯是跑一个最小验证流程选一只分红送股比较多的股票拉取包含除权日的区间数据分别用不复权、前复权、后复权三种方式计算区间收益然后和行情软件上的数据对比。import sqlite3 import pandas as pd conn sqlite3.connect(data/astock.db) # 拉取某只股票2020年到2023年的数据 df pd.read_sql( SELECT * FROM daily_price WHERE symbol000001 AND date BETWEEN 20200101 AND 20231231 ORDER BY date, conn ) factors pd.read_sql( SELECT * FROM adjust_factor WHERE symbol000001 ORDER BY date, conn ) # 不复权收益 raw_return df[close].iloc[-1] / df[close].iloc[0] - 1 # 后复权收益 df df.merge(factors, ondate, howleft) df[factor] df[factor].ffill().fillna(1.0) df[close_hfq] df[close] * df[factor] hfq_return df[close_hfq].iloc[-1] / df[close_hfq].iloc[0] - 1 print(f不复权收益: {raw_return:.2%}) print(f后复权收益: {hfq_return:.2%})跑完之后把后复权收益和行情软件上同一区间的“后复权涨幅”对比。如果差异在0.5%以内说明数据处理基本正确。如果差异大优先检查因子对齐和除权日处理。还有一个验证技巧选一只经历过10送10的股票看除权日当天不复权价格是否腰斩后复权价格是否连续。如果后复权价格在除权日出现跳空说明因子没算对。参数方面验证区间建议至少包含一次分红和一次送股否则测不出复权处理的问题。股票选择上银行股分红稳定但送股少中小板股票送股多但数据可能不全可以各选一只交叉验证。从那以后我每次拉完新数据都会先跑一遍这个验证流程确认复权处理没问题再喂给回测引擎。数据这件事宁可多花半小时验证也别等回测跑出漂亮曲线之后才发现是数据在骗你。希望帮到你。本文还有配套的精品资源点击获取