← 返回工具目录 · 全部教程

akshare 数据实操:行情数据自动落库

系列·数据篇 — 拉数、清洗、增量落库一条龙,让回测和 Agent 每天都有新鲜数据可用

01 · akshare 是什么,适合谁

akshare 是开源的 Python 财经数据接口库,聚合了东财、新浪等公开数据源,免费、无需注册 key。对个人研究者, 它是"从想法到数据"距离最短的一条路;代价是数据源非官方契约——接口偶尔会变,所以本篇最后有专门的避坑清单。

pip install akshare pandas -i https://pypi.tuna.tsinghua.edu.cn/simple

02 · 三类最常用接口

import akshare as ak

# 1) 单只股票日线(前复权,回测口径) — 东方财富源
df = ak.stock_zh_a_hist(symbol="000001", period="daily",
                        start_date="20230101", end_date="20250801", adjust="qfq")

# 2) 全市场实时快照(含最新价/涨跌幅/市值,盘中盘后都可拉)
spot = ak.stock_zh_a_spot_em()

# 3) 指数日线
idx = ak.stock_zh_index_daily(symbol="sh000001")

print(df.head())   # 注意:返回的是中文列名

中文列名不方便下游处理,统一 rename 成小写英文,这是所有 akshare 流水线的第一步:

import pandas as pd

RENAME = {"日期": "date", "开盘": "open", "最高": "high",
          "最低": "low", "收盘": "close", "成交量": "volume"}

df["date"] = pd.to_datetime(df["日期"])
df = df.rename(columns=RENAME).set_index("date")[["open", "high", "low", "close", "volume"]]

03 · 增量落库:SQLite 是个人研究的最佳默认

单文件、零运维、pandas 两行读写,数据量到千万行级别都够用。核心逻辑:每次只拉库里缺的那段:

import sqlite3, time
import akshare as ak

DB = "a_daily.db"

def update_one(code: str, conn):
    start = "20230101"
    # 库里已有就接着最后日期的下一天拉,实现增量
    exist = conn.execute(
        "SELECT MAX(date) FROM daily WHERE code=?", (code,)).fetchone()[0]
    if exist:
        start = (exist + pd.Timedelta(days=1)).strftime("%Y%m%d")

    df = ak.stock_zh_a_hist(symbol=code, period="daily",
                            start_date=start, end_date="20991231", adjust="qfq")
    if df.empty:
        return
    df["date"] = pd.to_datetime(df["日期"])
    df = df.rename(columns={"开盘": "open", "最高": "high", "最低": "low",
                            "收盘": "close", "成交量": "volume"})
    out = df[["date", "open", "high", "low", "close", "volume"]]
    out["code"] = code
    out.to_sql("daily", conn, if_exists="append", index=False)

conn = sqlite3.connect(DB)
conn.execute("""CREATE TABLE IF NOT EXISTS daily
    (code TEXT, date TIMESTAMP, open REAL, high REAL,
     low REAL, close REAL, volume REAL)""")
update_one("000001", conn)   # 实际项目里循环你的股票池
conn.close()

两个细节:time.sleep(0.5) 放进循环里做限频礼貌(别打挂数据源); 股票池列表可用 ak.stock_info_a_code_name() 拉全市场代码。

04 · 每天自动更新

把上面的脚本存成 update_daily.py,挂 crontab(服务器开机就一直跑):

crontab -e
# 交易日 18:30 跑增量更新(盘后数据基本定型)
30 18 * * 1-5 cd /home/quant/jobs && python update_daily.py >> logs/data_$(date +\%F).log 2>&1

05 · 避坑清单(每条都疼过)

• 盘中数据是半成品 — 日线接口盘中拉,当天那根 K 线是不完整的;自动任务统一排在盘后(18:30 之后)。
• 限频与封禁 — 循环里加 sleep;请求失败用指数退避重试,而不是立刻重试火上浇油。
• 复权口径 — 回测用 qfq 前复权;但前复权价会随分红除权整体漂移,落库的重要数据每隔几个月要全量刷新一遍,否则历史价格悄悄失真。
• 接口会变 — akshare 社区活跃,版本升级可能改名或换数据源;requirements.txt 里 pin 死版本,升级前先跑一遍回归。
• 退市股缺失 — 部分接口只返回当前在市股票,做全市场回测会有幸存者偏差,要有意识地在结论里打折。