上一篇画 K 线图时,我们代码里「偷偷」用过一次 akshare:拉了平安银行的日线数据。当时为了不喧宾夺主,只说了一句「失败就用本地模拟数据」。数据是量化的原料——没有数据,后面所有的指标计算、策略回测都是空中楼阁。所以本篇专门把 akshare 讲透:它是什么、怎么装、怎么拉数据、拉下来的字段都是什么意思,以及最重要的——怎么写才能让代码在接口变动或断网时也不至于直接崩溃。
为什么选 akshare 而不是其他数据源?因为它是免费、开源的 Python 财经数据接口库,覆盖 A 股、基金、期货、宏观等海量数据,不需要注册账号和申请 token,装上就能用,对个人学习者非常友好。等以后你进入机构场景,再去接触 Wind、iFinD 这类付费终端也不迟。
本篇目标
学完本篇,你能够做到:
- 独立安装 akshare,并用
ak.stock_zh_a_hist拉取单只 A 股 2020-01-01 至 2024-12-31 的前复权日线行情。 - 说清楚返回 DataFrame 每一列(日期、开盘、收盘、最高、最低、成交量、成交额、振幅、涨跌幅、涨跌额、换手率)的含义和单位。
- 用 for 循环批量下载多只股票的行情,并在每次请求之间加间隔,做一个「礼貌的爬取者」。
- 写出带
try/except与本地造数据 fallback 的健壮代码,数据源不可用时程序照样能跑通。 - 学会去 akshare 官方文档查某个接口的最新用法,接口改名时自己能救自己。
核心概念
akshare 是什么,接口怎么「长」
akshare(开源项目,官网文档在 akshare.akfamily.xyz)把几十个公开数据网站整理成了统一的 Python 接口。它的接口命名有规律:ak.品类_市场_内容,比如 ak.stock_zh_a_hist 就是「股票(stock)+ 中国(zh)+ A 股(a)+ 历史行情(hist)」。记住这个规律,你去文档里找接口时会快很多。
调用一个接口,本质上就是「带参数发一次请求,拿回一个 DataFrame」。所以拉数据的代码就是普通函数调用,没有黑魔法。
划重点:akshare 是对公开网站数据的封装,网站一改版,接口就可能失效或改名。这不是你的错,也不是 akshare 的错,属于免费数据的常态。对策有两个:一是升级 akshare 到最新版试试;二是去官方文档搜接口名,看最新签名长什么样。
前复权、后复权:不复权的历史数据是「假」的
上市公司会分红、送股,除权除息那天股价会突然「跳空下跌」,比如 10 送 10 后股价直接腰斩——但你的钱包并没有损失一半。如果直接用原始价格算收益率,就会把这个「假下跌」算成真亏损。复权(Adjust)就是把这个跳空修正掉:
- 前复权(qfq):以当前价格为基准,把历史价格按分红送股调整。历史价格会变,但最新价格和行情软件显示一致。回测和研究一般用它。
- 后复权(hfq):以上市首日为基准往后调整,历史价格不变,最新价格可能大得离谱。适合算长期真实收益。
- 不复权(空):原始价格,只适合看最近没发生除权的行情。
本系列统一用前复权。细节原理在后面讲回测陷阱的篇章还会再提,现在记住「拉历史数据先 adjust="qfq"」即可。
返回字段逐个认识
ak.stock_zh_a_hist 返回的 DataFrame 常见列如下(不同版本列名或有增减,以你实际拿到的为准):
| 列名 | 含义 | 单位/说明 |
|---|---|---|
| 日期 | 交易日 | 字符串,需转成 datetime |
| 股票代码 | 证券代码 | 如 000001 |
| 开盘 / 收盘 / 最高 / 最低 | 当日四个价格 | 元(前复权后) |
| 成交量 | 当日成交股数 | 手(1 手 = 100 股),部分版本直接给股 |
| 成交额 | 当日成交金额 | 元 |
| 振幅 | (最高-最低)/昨收 | 百分比数值 |
| 涨跌幅 | 相对昨收的涨跌 | 百分比数值,如 1.25 表示 1.25% |
| 涨跌额 | 相对昨收的价格变化 | 元 |
| 换手率 | 成交量/流通股本 | 百分比数值 |
拿到任何一份新数据,第一件事永远是 df.head() 加 df.dtypes 看看结构和类型,别急着上手算。
礼貌爬取与异常处理
连续快速请求几百次,可能被数据源限流甚至临时封禁。两个好习惯:每次请求之间 time.sleep() 停零点几秒;用 try/except 包住每次请求,单只失败不影响整体,把失败的记下来事后补拉。再配合本地造数据 fallback,教学代码就能在任何网络环境下跑通。
完整代码
安装命令(建议同时升级到最新版):
uv pip install akshare pandas --upgrade
下面代码分两段。第一段:拉单只股票,统一区间 2020-01-01 至 2024-12-31,做列名整理和日期转换;拉不到就用固定种子的本地模拟数据顶上。
import time
import numpy as np
import pandas as pd
START, END = "20200101", "20241231" # 系列统一示例区间
def fetch_one(symbol: str) -> pd.DataFrame:
"""拉一只股票的前复权日线;失败则用本地模拟数据 fallback。"""
try:
import akshare as ak # 放进函数内,没装也不影响 fallback
df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
start_date=START, end_date=END, adjust="qfq")
df = df.rename(columns={"日期": "date", "开盘": "open", "收盘": "close",
"最高": "high", "最低": "low", "成交量": "volume",
"成交额": "amount", "涨跌幅": "pct_chg"})
df["date"] = pd.to_datetime(df["date"])
df = df.set_index("date").sort_index()
print(f"[{symbol}] akshare 拉取成功,共 {len(df)} 个交易日")
return df[["open", "close", "high", "low", "volume", "pct_chg"]]
except Exception as e:
print(f"[{symbol}] akshare 不可用({type(e).__name__}),使用本地模拟数据")
rng = np.random.default_rng(int(symbol)) # 用代码当种子,每只股票序列固定
n = 1215 # 2020-2024 约 1215 个交易日
idx = pd.bdate_range("2020-01-02", periods=n)
base = {"000001": 16.0, "600036": 38.0, "000858": 140.0}.get(symbol, 20.0)
close = base * np.exp(np.cumsum(rng.normal(0.0002, 0.02, n)))
open_ = close * (1 + rng.normal(0, 0.005, n))
high = np.maximum(open_, close) * (1 + np.abs(rng.normal(0, 0.008, n)))
low = np.minimum(open_, close) * (1 - np.abs(rng.normal(0, 0.008, n)))
vol = rng.integers(30, 500, n) * 1e4
pct = np.append(np.nan, (close[1:] / close[:-1] - 1) * 100) # 首日无昨收
out = pd.DataFrame({"open": open_, "close": close, "high": high,
"low": low, "volume": vol, "pct_chg": pct}, index=idx)
out.index.name = "date"
return out
df_000001 = fetch_one("000001")
print(df_000001.head())
print(df_000001.dtypes)
time.sleep(1) # 单次拉取后稍作停歇,养成礼貌习惯
第二段:循环下载多只股票,组装成一个「代码 → DataFrame」的字典,并演示怎么从字典里取一只股票出来画图验证。
symbols = ["000001", "600036", "000858"] # 平安银行、招商银行、五粮液
data: dict[str, pd.DataFrame] = {}
for i, sym in enumerate(symbols):
data[sym] = fetch_one(sym)
if i < len(symbols) - 1:
time.sleep(1) # 每次请求间隔 1 秒,降低被限流的风险
# 汇总检查:每只股票的行数、起止日期
for sym, df in data.items():
print(f"{sym}: {df.index.min().date()} ~ {df.index.max().date()},{len(df)} 行")
# 取出一只,快速画收盘价验证数据形态
import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"] = ["SimHei"] # 中文显示,第 6 篇讲过
plt.rcParams["axes.unicode_minus"] = False
df_600036 = data["600036"]
plt.figure(figsize=(11, 5))
plt.plot(df_600036.index, df_600036["close"], linewidth=1.1, label="招商银行 前复权收盘价")
plt.title("2020-2024 收盘价走势(数据不可用时为模拟数据)")
plt.ylabel("价格(元)")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
第三段:拉到手的本地数据先存一份 CSV——「今天存的」就是第 8 篇要清洗的原料,提前感受一下数据落地的流程:
# ---- 预告第 8 篇:把拉好的数据保存到本地 ----
import os
os.makedirs("data", exist_ok=True) # 建一个数据目录,重复运行也不会报错
for sym, df in data.items():
path = f"data/{sym}_daily.csv"
df.to_csv(path, encoding="utf-8-sig") # utf-8-sig 带 BOM,Excel 打开不乱码
size_kb = os.path.getsize(path) / 1024
print(f"{sym} 已保存到 {path}({size_kb:.1f} KB)")
# 验证:读回一只,确认日期索引是否需要重新解析
check = pd.read_csv(f"data/{symbols[0]}_daily.csv", index_col="date",
parse_dates=["date"])
print(f"读回 {symbols[0]}:{len(check)} 行,索引类型 = {check.index.dtype}")
代码逐段解释
第 1 段:单只拉取函数。 fetch_one() 把「拉数据」封装成一个输入代码、输出标准 DataFrame 的函数,这是为后面批量下载做准备。函数内第一行才 import akshare,这样没装 akshare 的读者不会在 import 阶段就报错,而是自然走到 except 分支。列名重命名成统一英文小写,是整个系列约定俗成的「标准列名」——第 8 篇清洗数据、第 9 篇算收益率都直接沿用。本地 fallback 用 int(symbol) 做随机种子,好处是:同一只股票不管跑多少次,模拟序列都一样,代码可复现;不同股票序列不同,符合「每只股票行情不同」的直觉。示例输出示意(数值为示意,实际以运行为准):
[000001] akshare 拉取成功,共 1215 个交易日
open close high low volume pct_chg
date
2020-01-02 16.55 16.45 16.60 16.30 783415 0.15
2020-01-03 16.52 16.36 16.60 16.28 668782 -0.55
...
open float64
close float64
volume int64
pct_chg float64
第 2 段:批量下载。 用字典推导式的思路换成显式 for 循环,逐只调用 fetch_one 存进 data 字典,键是 6 位代码,值是标准 DataFrame。每两次请求之间 time.sleep(1)。之后打印每只股票的起止日期和行数做「到货验收」——养成这个习惯,能第一时间发现拉到了空数据或区间不对。最后画一张收盘价图眼见为实。示例输出示意:
[000001] akshare 拉取成功,共 1215 个交易日
[600036] akshare 拉取成功,共 1215 个交易日
[000858] akshare 不可用(KeyError),使用本地模拟数据
000001: 2020-01-02 ~ 2024-12-31,1215 行
600036: 2020-01-02 ~ 2024-12-31,1215 行
000858: 2020-01-02 ~ 2024-12-31,1215 行
注意上面示意里第三只股票走了 fallback——这正是我们想要的效果:一只失败,整体照常完成,最后你只需单独补拉失败的代码即可。
第 3 段:落地成 CSV。 os.makedirs(..., exist_ok=True) 确保目录存在且重复运行不报错;encoding="utf-8-sig" 是给 Windows 用户的贴心参数——带 BOM 头的 UTF-8 让 Excel 双击打开中文不乱码。读回验证时再次出现了 parse_dates=["date"]——「CSV 不保存类型」这个话题,第 8 篇会正式展开。示例输出示意:
000001 已保存到 data/000001_daily.csv(58.3 KB)
600036 已保存到 data/600036_daily.csv(57.9 KB)
000858 已保存到 data/000858_daily.csv(58.1 KB)
读回 000001:1215 行,索引类型 = datetime64[ns]
划重点:
data这个「代码 → 行情 DataFrame」的字典结构,就是你自己最简单的「本地数据库」雏形。第 8 篇我们会把它落盘成 CSV、SQLite 和 Parquet,变成真正的本地存储。
关于接口时效性的声明。 本篇写就时,ak.stock_zh_a_hist(symbol, period, start_date, end_date, adjust) 是可用签名;akshare 迭代很快,等你读到这篇时,参数名或返回列名可能已经变化。遇到报错先做三件事:uv pip install akshare --upgrade 升级;去官方文档搜接口名看最新签名;用 print(df.columns) 打印实际返回列。教程给的是方法和骨架,最新细节以官方文档为准。
常见坑与小练习
坑 1:报错 AttributeError: module 'akshare' has no attribute ... 或列名对不上。 现象:照着教程写却报接口不存在。原因:akshare 版本差异,接口改名或迁移。避免方式:升级到最新版;仍不行就查官方文档确认当前接口名。这也是本篇坚持「数据层全部包 try/except + fallback」的原因。
坑 2:收益率算出莫名其妙的巨幅亏损。 现象:某天涨跌幅是 -33%,但你记得那只是普通分红除权。原因:用了不复权数据,把除权跳空当成了真实亏损。避免方式:拉历史数据一律 adjust="qfq" 前复权。
坑 3:连续拉几十只股票后被限流。 现象:前面正常,中途开始报超时或空数据。原因:请求太密集。避免方式:每请求之间 time.sleep(1) 左右;批量任务放在非交易高峰跑;失败列表记录下来稍后重试。
坑 4:date 列是字符串,切片和画图全部出错。 现象:df.loc["2020-01"] 报错,或 X 轴顺序错乱。原因:日期没转 datetime 就当了索引。避免方式:拿到数据第一件事 pd.to_datetime + set_index + sort_index,本篇 fetch_one 已内置。
坑 5:成交量单位误当「股」。 现象:换手或金额计算差 100 倍。原因:部分数据源成交量单位是「手」,1 手 = 100 股。避免方式:用 df.head() 对照行情软件核对一两个数字,确认单位后再用。
小练习 1: 把股票列表扩到 5 只(自选两只加入),循环下载并打印每只的「区间总涨跌幅」:(末日收盘/首日收盘 - 1) * 100,注意用前复权价。
思路提示:
first, last = df["close"].iloc[0], df["close"].iloc[-1];fallback 数据也能算,结果为示意。
小练习 2: 给 fetch_one 增加重试逻辑:失败后自动重试 2 次,每次间隔 3 秒,仍失败才走本地模拟数据。
思路提示:把 try/except 套进
for attempt in range(3):循环里,成功就return,用尽次数再 fallback。
小练习 3: 打印某只股票 pct_chg 列的最大值和最小值对应的两天的完整行情(用 idxmax / idxmin),对照当天 K 线看看长什么样。
思路提示:
df.loc[df["pct_chg"].idxmax()];把这两天和前后各两天一起打印,更有上下文。
本文策略仅用于教学,不构成投资建议,实盘前请充分测试。