Akshare 财经数据抓取实战：量化交易的第一步

为什么选择 Akshare？

对于我们的量化研究库来说，Akshare 提供了极其丰富的接口，涵盖了从股票、期货到宏观经济的全维度数据。 . 环境准备确保您的 Python 环境已安装。由于 Akshare 依赖较多，建议在独立虚拟环境中操作。安装命令：

pip install akshare --upgrade

核心代码实现

以下是在我的 Minisforum N5 Pro 服务器上测试通过的代码片段：

import akshare as ak

A. 股票行情数据获取 A 股所有股票的最新实时行情。


# 获取日线数据
df = ak.stock_zh_a_hist(symbol="000001", period="daily", adjust="qfq")
print(df.head())

B. 历史行情数据（日线/周线/月线）获取特定股票的历史数据，适合用于回测。

stock_zh_a_hist_df = ak.stock_zh_a_hist(symbol="600519", period="daily", start_date="20230101", end_date="20231231", adjust="qfq")
print(stock_zh_a_hist_df)


# 获取当前时刻 A 股所有股票的实时行情
stock_zh_a_spot_em_df = ak.stock_zh_a_spot_em()
print(stock_zh_a_spot_em_df.head())

# 中国 PMI 数据
macro_china_pmi_df = ak.macro_china_pmi_index()
print(macro_china_pmi_df)

C. 股息红利数据既然您的目标是长期股息红利，以下接口非常关键：

# 获取个股的分红送股数据
stock_history_dividend_df = ak.stock_history_dividend_detail(symbol="600036", indicator="分红")
print(stock_history_dividend_df)

财务报表


# 获取贵州茅台的资产负债表 (按报告期)
stock_balance_sheet_df = ak.stock_balance_sheet_by_report_em(symbol="600519")
print(stock_balance_sheet_df.head())
# 获取利润表
stock_profit_sheet_df = ak.stock_profit_sheet_by_report_em(symbol="600519")
print(stock_profit_sheet_df.head())
# 获取现金流量表
stock_cash_flow_sheet_df = ak.stock_cash_flow_sheet_by_report_em(symbol="600519")
print(stock_cash_flow_sheet_df.head())
# 获取主要财务指标
stock_financial_analysis_indicator_df = ak.stock_financial_analysis_indicator(symbol="600519")
print(stock_financial_analysis_indicator_df)

常见应用场景示例

场景：筛选高股息率股票结合多个接口，您可以编写脚本自动计算当前的股息率情况

步骤	接口示例
1. 获取成分股	ak.index_stock_cons(symbol=“000016”) (上证50)
2. 获取财务报表	ak.stock_financial_analysis_indicator()
3. 获取最新市值	ak.stock_zh_a_spot_em

针对“股息红利”的专项数据

既然您的目标是红利策略，以下两个接口能提供更直观的支持：

接口名称	功能	适用场景
ak.stock_history_dividend()	历史分红详情	查看公司历年送股、转增、派现数据。
ak.stock_zh_a_gdhs_detail_em()	股东户数详情	观察筹码集中度，判断大股东动向。

同步A股全市场历史日线

为了后续能训练你的TFT机器学习模型，我们需要一个自动化的同步脚本。这个脚本会做三件事： 1.从Akshare 抓取数据。 2.自动在quant_lab库中创建表。 3.关键：自动将表转换为TimescaleDB的Hypertable（超表），实现极致的查询性能。在你的Ubuntu VM或本地开发机上运行

SQL脚本

 -- 1. 创建量化专用库
CREATE DATABASE quant_lab;

-- 2. 切换到新库
\c quant_lab

-- 3. 在新库也要开启插件
CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;
-- 开启压缩，这对于 NAS 存储非常友好，能节省约 90% 空间
ALTER TABLE stock_daily_hists SET (
  timescaledb.compress,
  timescaledb.compress_segmentby = 'symbol'
);

Python脚本：sync_all_stocks.py

import akshare as ak
import pandas as pd
from sqlalchemy import create_engine, text
# 1. 连接到你的新库 quant_lab
DB_URL = "postgresql://postgres:password@ip/quant_lab"
engine = create_engine(DB_URL)
def sync_daily_data(symbol="000001"):
    print(f"正在抓取 {symbol}...")
    # 获取历史日线 (前复权)
    df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="qfq")
    # 字段清洗：对接数据库命名规范
    df = df[['日期', '开盘', '收盘', '最高', '最低', '成交量', '成交额']]
    df.columns = ['trade_date', 'open', 'close', 'high', 'low', 'volume', 'turnover']
    df['symbol'] = symbol
    df['trade_date'] = pd.to_datetime(df['trade_date'])

    # 2. 写入数据库
    # 如果表不存在则创建，如果存在则追加数据
    df.to_sql('stock_daily_hists', engine, if_exists='append', index=False)

def init_hypertable():
    # 3. 将表转为 Hypertable (只需执行一次)
    with engine.connect() as conn:
        conn.execute(text("SELECT create_hypertable('stock_daily_hists', 'trade_date', if_not_exists => TRUE);"))
        conn.commit()
    print("✅ Hypertable 转换完成！")

if __name__ == "__main__":
    # 先运行一次初始化
    init_hypertable()
    # 尝试同步一只股票测试
    sync_daily_data("000001")

stock_df.to_csv("000001_daily.csv", index=False)