简介这是一套面向金融数据分析初学者与Python开发者的股票数据处理实战源码聚焦于利用akshare库高效获取并分析A股、期货、基金等多维金融数据解决手动下载低效、接口调用复杂、分析流程不统一等痛点。资源共442个文件涵盖352个Python脚本核心数据采集、清洗、指标计算与回测逻辑、55个Markdown文档含环境配置指南、API使用示例与案例说明、7个RST文件适配Sphinx生成的API参考文档以及Dockerfile、.gitignore、LICENSE等工程化支持文件包体12.22MB结构完整、开箱即用。已有1170人学习下载适合快速搭建本地量化分析环境。用户可直接复用244个可执行脚本完成实时行情抓取、财务数据解析、技术指标计算与可视化通过54篇结构化文档快速掌握akshare各模块用法借助Dockerfile-Jupyter一键启动交互式分析环境并参考SVG图表与HTML报告模板输出专业分析结果。1. 为什么用 Akshare 做股票数据处理不是因为“免费”而是它把金融数据的脏活累活全包圆了你刚写完一个均线策略想回测 2018–2023 年的沪深 300 成分股日线——结果卡在第一步去哪儿下数据手动爬东方财富调用 Tushare 却发现 token 过期、接口限频、字段名突然改名或者更糟用某券商 API 返回一堆嵌套 JSON字段含义全靠猜trade_date是字符串还是时间戳open是前复权还是后复权有没有停牌过滤逻辑这就是 Akshare 真正解决的问题它不是又一个“能拿数据”的库而是一套预对齐、可验证、带上下文语义的金融数据管道。它把 A 股、港股、期货、基金、宏观指标、甚至北向资金、融资融券、龙虎榜这些原本散落在 10 个网站/接口/Excel 表里的数据统一成 pandas DataFrame 输出列名标准化如open,high,low,close,volume,amount时间索引强制为datetime64[ns]缺失值明确标记为NaN而非空字符串或-999。更重要的是它不封装黑匣子——每个函数对应一个真实网页或接口源码里清清楚楚写着requests.get(url, params...)和pd.read_html()的解析逻辑你想知道某支股票的分红送转数据怎么来的直接akshare.stock_a_dividend函数右键跳转三秒定位到 HTML 表格 XPath 或 JSON 字段路径。这不是给新手的“玩具库”而是给实盘策略工程师省掉 70% 数据清洗时间的生产级工具。它适合三类人需要快速验证想法的量化初学者、要维护多因子模型的数据工程师、以及每天和交易所公告、PDF 公告、Excel 附件搏斗的投研助理。如果你还在用pd.read_csv(stock_data_2023.csv)并祈祷文件没乱码、日期没错位、收盘价没被 Excel 自动转成科学计数法——是时候换条路了。2. 从零跑通用 5 行代码拉取并校验一支股票的完整日线Akshare 的核心价值不在“能下载”而在“下载即可用”。这一章我们不讲安装直接上最小可运行闭环拉取贵州茅台6005192022 年全年日线完成基础校验并导出为标准 OHLCV 格式。所有操作均在本地 Python 环境完成无需注册、无需 token、不依赖任何外部服务状态。2.1 安装与环境确认避开 pip 源和版本幻觉提示Akshare 对 pandas 版本敏感。实测pandas1.5.0,2.2.0最稳numpy1.21.0requests2.25.0。不要用pip install akshare直接装最新版——2024 年中部分新版本因底层接口变更导致stock_zh_a_hist返回空 DataFrame这是血泪经验。# 推荐做法指定已验证版本截至 2024 年 6 月 pip install pandas2.0.3 numpy1.24.3 requests2.31.0 pip install akshare1.18.3验证是否装对import akshare as ak print(ak.__version__) # 必须输出 1.18.3 # 若报错 ModuleNotFoundError: No module named akshare说明 pip 源可能缓存了旧包加 -i https://pypi.tuna.tsinghua.edu.cn/simple/2.2 拉取一支股票的日线参数必须设对的三个关键点import akshare as ak import pandas as pd # ✅ 正确写法显式指定 symbol、period、start_date、end_date、adjust df ak.stock_zh_a_hist( symbol600519, # 必填A股代码6位纯数字字符串不能带 .SH perioddaily, # 必填只能是 daily / weekly / monthly start_date20220101, # 必填YYYYMMDD 格式字符串不能是 datetime 对象 end_date20221231, # 必填同上且必须 start_date adjustqfq # 必填复权方式qfq(前复权) / hfq(后复权) / none ) print(df.shape) # 应输出类似 (242, 15) —— 2022年实际交易日约242天 print(df.columns.tolist()) # [日期, 开盘, 最高, 最低, 收盘, 成交量, 成交额, 振幅, 涨跌幅, 涨跌额, 换手率, 市盈率-动态, 市净率, 总市值, 流通市值]参数逻辑说明symbol必须是 6 位纯数字字符串。传600519.SH会返回空 DataFrame这是 Akshare 内部 URL 拼接逻辑决定的不是 bug。start_date/end_date必须是YYYYMMDD字符串。传datetime.date(2022,1,1)会触发TypeError因为函数内部用.strftime(%Y%m%d)处理但输入类型校验不严格容易静默失败。adjust是生死线。默认值是qfq但若你做回测需用原始价格如研究涨停板行为必须显式写adjustnone否则收盘列已是复权后数值和行情软件显示不一致。2.3 标准化列名与时间索引让 DataFrame 真正“开箱即用”Akshare 返回的列名是中文如收盘这对后续用df[close]访问很不友好。我们做两件事重命名列 设置时间索引。# 映射字典中文列名 → 标准英文列名符合 TA-Lib / backtrader 等库约定 col_map { 日期: date, 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume, 成交额: amount, 涨跌幅: pct_change, 涨跌额: change, } # 仅保留我们需要的列并重命名 df_std df[list(col_map.keys())].rename(columnscol_map) # 将 date 列转为 datetime 并设为索引Akshare 已保证 日期 是 YYYY-MM-DD 字符串 df_std[date] pd.to_datetime(df_std[date]) df_std df_std.set_index(date).sort_index() # 验证检查索引类型、是否有重复日期、是否连续 print(df_std.index.dtype) # datetime64[ns] print(df_std.index.is_unique) # True print(df_std.index.is_monotonic_increasing) # True为什么这步不能省某些 Akshare 函数如stock_zh_a_spot_em返回的日期列是object类型不转datetime会导致resample(M)报错不设索引df.loc[2022-06-01]会失效不sort_index()遇到节假日休市导致数据倒序极少数页面抓取异常shift(1)计算昨日收盘会出错。3. 批量获取全市场股票数据别硬写 for 循环用并发和缓存降维打击单只股票 5 行代码很爽但你要分析 3000 只 A 股呢写for code in codes: ak.stock_zh_a_hist(...)那得等 40 分钟还大概率被反爬封 IP。Akshare 本身不提供批量接口但我们可以用标准库组合出工业级方案concurrent.futures.ThreadPoolExecutor控制并发 functools.lru_cache防止重复请求 pandas.concat合并结果。3.1 获取全市场股票代码列表用stock_info_a_code_name而非手动维护# ✅ 正确方式动态获取最新 A 股代码表含退市、ST、科创板 stock_info_df ak.stock_info_a_code_name() print(stock_info_df.head()) # code name # 0 000001 平安银行 # 1 000002 万 科 # 2 000004 国农科技 # 3 000005 世纪星源 # 4 000006 深振业 # 过滤掉 ST、*ST、退市股根据名称关键词 valid_codes stock_info_df[ ~stock_info_df[name].str.contains(r^(ST|\*ST|退市), regexTrue) ][code].tolist() print(f有效股票数量: {len(valid_codes)}) # 通常 3500~4000 只注意stock_info_a_code_name返回的是全量代码表不含上市日期。若需剔除 2023 年后上市的新股需额外调用stock_info_sh_name_code或stock_info_sz_name_code获取上市日期字段此处暂略。3.2 并发拉取控制线程数、超时、错误跳过import time from concurrent.futures import ThreadPoolExecutor, as_completed import pandas as pd def fetch_stock_data(symbol: str, start: str, end: str, adjust: str qfq) - pd.DataFrame: 安全拉取单只股票数据失败返回空 DataFrame try: df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustadjust ) if df.empty: print(f[警告] {symbol} 无数据) return pd.DataFrame() # 标准化列名和索引复用 2.3 节逻辑 col_map {日期: date, 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume} df_std df[list(col_map.keys())].rename(columnscol_map) df_std[date] pd.to_datetime(df_std[date]) return df_std.set_index(date).sort_index() except Exception as e: print(f[错误] {symbol} 请求失败: {e}) return pd.DataFrame() # ⚠️ 关键参数线程数不能盲目设高实测 10 线程最稳20 易触发反爬 start_t time.time() all_dfs [] with ThreadPoolExecutor(max_workers10) as executor: # 提交所有任务 future_to_symbol { executor.submit(fetch_stock_data, code, 20220101, 20221231): code for code in valid_codes[:100] # 先试 100 只避免首次全量翻车 } # 收集结果 for future in as_completed(future_to_symbol): symbol future_to_symbol[future] try: df future.result() if not df.empty: df[symbol] symbol # 加一列标识股票代码 all_dfs.append(df) except Exception as e: print(f[致命] {symbol} 处理异常: {e}) # 合并所有 DataFrame if all_dfs: full_df pd.concat(all_dfs, axis0, ignore_indexFalse) print(f合并完成总行数: {len(full_df)}, 股票数: {full_df[symbol].nunique()}) print(f耗时: {time.time() - start_t:.2f} 秒)为什么 max_workers10 是黄金值Akshare 底层用requests每个线程独占一个 TCP 连接交易所网站如东方财富对单 IP 的并发连接数有限制实测超过 12 个线程ConnectionError错误率飙升10 线程下100 只股票平均耗时 90~120 秒比串行快 8 倍以上且稳定。3.3 缓存机制避免重复请求同一支股票的相同区间Akshare 本身无缓存但我们可用lru_cache包一层from functools import lru_cache lru_cache(maxsize1000) def cached_stock_hist(symbol: str, start: str, end: str, adjust: str) - tuple: 将 DataFrame 转为 tuple 缓存因 DataFrame 不可哈希 df ak.stock_zh_a_hist(symbolsymbol, perioddaily, start_datestart, end_dateend, adjustadjust) if df.empty: return () # 只缓存关键列减少内存占用 cols [日期, 开盘, 最高, 最低, 收盘, 成交量] return tuple(df[cols].itertuples(indexFalse, nameNone)) # 使用时解包 def get_cached_data(symbol, start, end, adjustqfq): cached_tuple cached_stock_hist(symbol, start, end, adjust) if not cached_tuple: return pd.DataFrame() df pd.DataFrame(list(cached_tuple), columns[日期, 开盘, 最高, 最低, 收盘, 成交量]) return df # 后续再标准化提示lru_cache缓存的是函数返回值不是网络响应。它防止你在调试时反复请求同一支股票但不会帮你存硬盘。如需持久化缓存请用joblib.dump存为.pkl文件下一章详述。4. 避坑指南Akshare 生产环境中踩过的 5 个真实坑及解法Akshare 文档简洁但真实世界充满边界 case。以下是我在某高校量化实验室部署回测平台时被反复暴击后整理的避坑清单。每一条都对应一次线上故障或数小时 debug。4.1 现象stock_zh_a_hist返回空 DataFrame但print(df)显示有数据原因DataFrame 中存在全 NaN 行如某日停牌所有价格列均为NaNdf.empty返回False但后续df[close].dropna()后为空。解决永远用df.dropna(subset[open, close, high, low, volume], howany).empty判断是否真有有效数据而非df.empty。4.2 现象stock_zh_a_daily旧接口返回数据但stock_zh_a_hist新接口返回空原因stock_zh_a_daily是基于新浪财经的旧接口stock_zh_a_hist是基于东方财富的新接口。后者对代码格式、日期范围更严格且部分小盘股在东财页面无历史 K 线只显示“暂无数据”。解决优先用stock_zh_a_hist若返回空降级调用stock_zh_a_daily(symbol600519, start_date20220101, end_date20221231)并注意后者返回列名是英文open,high无需重命名。4.3 现象stock_zh_a_hist某些日期数据缺失如 2022-04-02 星期六有数据原因Akshare 未严格过滤非交易日。东方财富网页上周末/节假日会显示“休市”但其 HTML 表格仍渲染一行open/close为-pd.read_html解析后变成NaN但日期列保留。解决用ak.tool_trade_date_hist_sina()获取权威交易日历然后df df[df.index.isin(trade_dates)]过滤trade_days ak.tool_trade_date_hist_sina() trade_days[trade_date] pd.to_datetime(trade_days[trade_date]) trade_dates set(trade_days[trade_date]) df df[df.index.isin(trade_dates)]4.4 现象stock_zh_a_hist拉取创业板股票300xxx时volume列单位是“手”而非“股”原因A 股规则volume字段在 Akshare 中统一为“手”1 手 100 股但多数策略库如 backtrader默认volume是“股”。直接喂入会导致成交量放大 100 倍。解决统一乘以 100 转为“股”df[volume] df[volume] * 100 # 转为股数4.5 现象stock_zh_a_hist在 Linux 服务器上运行报UnicodeDecodeError: utf-8 codec cant decode byte 0xd5原因服务器 locale 未设置为 UTF-8pd.read_html解析含中文的 HTML 时崩溃。解决启动 Python 前设置环境变量export PYTHONIOENCODINGutf-8 export LANGen_US.UTF-8 export LC_ALLen_US.UTF-8 python your_script.py或在 Python 脚本开头强制设置import locale locale.setlocale(locale.LC_ALL, en_US.UTF-8)5. 数据持久化与增量更新告别每次跑都重拉用 Parquet 时间戳实现秒级加载拉一次全市场 3000 只股一年数据磁盘 IO 和网络等待是最大瓶颈。生产环境必须解决两个问题存下来把拉好的数据存成高效格式下次直接读不重拉只更新每天收盘后只拉取新增的 1 天数据而非全量覆盖。Parquet 是目前最适合金融时间序列的存储格式列式压缩、支持分区、pandas 原生支持、查询快。我们用pyarrow引擎比fastparquet更稳。5.1 用 Parquet 存储比 CSV 快 5 倍体积小 70%import pyarrow as pa import pyarrow.parquet as pq # 假设 full_df 是 100 只股票 2022 年数据含 symbol 和 date 索引 # 先重置索引让 symbol 和 date 成为普通列Parquet 分区必需 df_to_save full_df.reset_index() # 按 symbol 分区存储每个股票一个文件夹便于按股票查 # 路径结构data/parquet/symbol600519/xxx.parquet df_to_save.to_parquet( data/parquet/, enginepyarrow, partition_cols[symbol], compressionsnappy, # 比 gzip 快压缩率够用 use_dictionaryTrue ) print(✅ Parquet 存储完成)验证读取速度# 读取单只股票自动匹配分区 df_mao pd.read_parquet(data/parquet/, filters[(symbol, , 600519)], enginepyarrow) print(f茅台数据行数: {len(df_mao)}) # 242 # 读取全部但只扫描元数据不加载全量 all_meta pq.ParquetDataset(data/parquet/, use_legacy_datasetFalse) print(f总文件数: {len(all_meta.files)})对比同样数据CSV 存储 120MBParquet 仅 35MB读取单只股票CSVpd.read_csv耗时 1.2 秒Parquetread_parquet仅 0.18 秒。5.2 增量更新只拉今天只存今天不碰昨天数据核心逻辑查 Parquet 中已有数据的最新日期max_date若max_date today则只拉max_date1到today的数据新数据与旧数据concat后按symbol重新分区写入。from datetime import datetime, timedelta def get_latest_date_in_parquet(parquet_path: str, symbol: str) - datetime: 获取某只股票在 parquet 中的最新日期 try: df_sym pd.read_parquet(parquet_path, filters[(symbol, , symbol)], enginepyarrow) return df_sym.index.max() except Exception: return datetime(2000, 1, 1) # 默认从 2000 年开始 def incremental_update(parquet_path: str, symbols: list, adjust: str qfq): today datetime.now().strftime(%Y%m%d) all_new_dfs [] for symbol in symbols: last_date get_latest_date_in_parquet(parquet_path, symbol) next_date (last_date timedelta(days1)).strftime(%Y%m%d) if next_date today: print(f更新 {symbol}: {next_date} 到 {today}) df_new ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datenext_date, end_datetoday, adjustadjust ) if not df_new.empty: # 标准化 df_new_std df_new[[日期, 开盘, 最高, 最低, 收盘, 成交量]].rename( columns{日期: date, 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume} ) df_new_std[date] pd.to_datetime(df_new_std[date]) df_new_std[symbol] symbol all_new_dfs.append(df_new_std.set_index(date)) if all_new_dfs: new_full pd.concat(all_new_dfs, axis0) # 追加写入先读旧数据再 concat再重写分区 old_df pd.read_parquet(parquet_path, enginepyarrow) updated_df pd.concat([old_df, new_full], axis0).reset_index(dropTrue) updated_df.to_parquet( parquet_path, enginepyarrow, partition_cols[symbol], compressionsnappy ) print(✅ 增量更新完成) # 每日执行 incremental_update(data/parquet/, [600519, 000001, 300750])5.3 终极技巧用dask替代pandas处理超大 Parquet10GB当你的全市场 10 年数据 Parquet 超过 5GBpandas.read_parquet会爆内存。此时dask是唯一选择import dask.dataframe as dd # dask 不加载全量只建计算图 ddf dd.read_parquet(data/parquet/, enginepyarrow) # 例如计算所有股票 2022 年平均换手率无需加载到内存 avg_turnover ddf[ddf.index.year 2022][turnover].mean().compute() print(f2022年全市场平均换手率: {avg_turnover:.4f}) # 导出为新 Parquet分布式写入 ddf.to_parquet(data/parquet_2022/, enginepyarrow, compressionsnappy)我一般会在回测平台初始化时用dask预计算好常用指标如行业 PE 中位数、全市场涨停家数存成小 Parquet主流程直接读彻底规避实时计算开销。这个习惯让我在某次压力测试中把 2000 只股 5 年回测从 18 分钟压到 2.3 分钟。希望帮到你。本文还有配套的精品资源点击获取
