金融数据获取 (Financial Data APIs)
章节概述
没有数据就没有量化分析。本章介绍两个核心金融数据获取库：akshare（中国 A 股/基金/期货）和 yfinance（国际市场）。你将学会获取历史 OHLCV 行情、实时行情、基本面数据，以及如何将不同来源的数据标准化为统一的 DataFrame 格式。数据的”形状一致性”是所有量化策略的基石。
核心理念：量化分析的第一步永远是”数据工程”——获取、清洗、对齐、存储。用 C 程序员的思维来理解：数据就位（struct 对齐）后，才能开始计算。Python 的生态优势在于有现成的 API 库帮你完成数据获取，你只需要调用几个函数，就能得到结构化的 DataFrame。但要注意：每个数据源的格式都略有不同，标准化是必须做的工作。
第一节：akshare —— A 股数据的一站式入口
1.1 安装与基础架构
pip install akshare --upgradeakshare 提供 500+ 个数据接口，覆盖股票、基金、期货、债券、宏观数据等。所有接口返回 Pandas DataFrame。
1.2 获取 A 股历史行情（OHLCV）
python -c "
import akshare as ak
# 获取平安银行(000001)的历史日线数据
df = ak.stock_zh_a_hist(
symbol='000001',
period='daily',
start_date='20240101',
end_date='20241231',
adjust='qfq' # 前复权
)
print('Columns:', list(df.columns))
print('Shape:', df.shape)
print(df.head())
print()
print('dtypes:')
print(df.dtypes)
"关键参数说明：
| 参数 | 含义 | 可选值 |
|---|---|---|
symbol | 股票代码 | 6 位数字字符串，如 '000001' |
period | 数据周期 | daily, weekly, monthly |
start_date | 起始日期 | 'YYYYMMDD' 格式 |
end_date | 结束日期 | 'YYYYMMDD' 格式 |
adjust | 复权方式 | ''（不复权）、'qfq'（前复权）、'hfq'（后复权） |
1.3 获取实时行情
python -c "
import akshare as ak
# A 股实时行情（全部股票）
# df = ak.stock_zh_a_spot_em() # 东方财富源
# print(df.columns)
# print(df[['代码', '名称', '最新价', '涨跌幅', '成交量', '成交额']].head())
# 特定股票的实时数据
df = ak.stock_zh_a_hist(symbol='000001', period='daily', start_date='20260801', end_date='20260807')
print(df.tail())
"1.4 获取基金与指数数据
python -c "
import akshare as ak
# 获取沪深300指数成分股
# df = ak.index_stock_cons(symbol='000300')
# print(df.head())
# 获取指数历史数据
# df = ak.stock_zh_index_daily(symbol='sh000300')
# print(df.tail())
# 获取基金历史净值
# df = ak.fund_open_fund_info_em(symbol='000001', indicator='单位净值走势')
# print(df.head())
print('akshare provides 500+ data interfaces for Chinese markets')
"1.5 财报数据获取
python -c "
import akshare as ak
# 获取某股票的资产负债表
# df = ak.stock_financial_balance_sheet_by_report_em(symbol='000001')
# print(df.columns)
# 获取利润表
# df = ak.stock_financial_profit_sheet_by_report_em(symbol='000001')
# print(df.columns)
print('Financial statements available via akshare EM interfaces')
"第二节：yfinance —— 国际市场数据
2.1 安装与基础用法
pip install yfinance --upgradeyfinance 通过 Yahoo Finance API 获取全球市场的股票、ETF、指数、外汇、加密货币数据。
python -c "
import yfinance as yf
# 下载苹果公司历史数据
aapl = yf.download('AAPL', start='2024-01-01', end='2024-12-31')
print('Columns:', list(aapl.columns))
print('Shape:', aapl.shape)
print(aapl.tail())
"2.2 多股票同时下载
python -c "
import yfinance as yf
# 同时下载多只股票
tickers = ['AAPL', 'GOOGL', 'MSFT', 'TSLA']
data = yf.download(tickers, start='2024-01-01', end='2024-06-30')
print('Shape:', data.shape) # Multi-level columns
print('Column levels:', data.columns.names)
print()
# 只看收盘价
print('Close prices:')
print(data['Close'].tail())
"2.3 获取基本面数据
python -c "
import yfinance as yf
msft = yf.Ticker('MSFT')
# 基本信息
info = msft.info
print('Market Cap:', info.get('marketCap'))
print('PE Ratio:', info.get('trailingPE'))
print('Sector:', info.get('sector'))
# 获取财务报表
# balance_sheet = msft.balance_sheet
# income_stmt = msft.financials
# cashflow = msft.cashflow
"2.4 获取分红和拆股数据
python -c "
import yfinance as yf
aapl = yf.Ticker('AAPL')
dividends = aapl.dividends
splits = aapl.splits
print('Recent dividends:')
print(dividends.tail())
print()
print('Stock splits:')
print(splits)
"第三节：数据标准化 —— 统一不同数据源的格式
3.1 为什么需要标准化
不同数据源的列名、日期格式、复权方式各有不同。写策略之前必须将所有输入标准化为统一的格式：
python -c "
import pandas as pd
import numpy as np
# 定义标准 OHLCV 列名映射
COLUMN_MAP = {
'open': ['open', 'Open', 'OPEN', '开盘'],
'high': ['high', 'High', 'HIGH', '最高'],
'low': ['low', 'Low', 'LOW', '最低'],
'close': ['close', 'Close', 'CLOSE', '收盘'],
'volume': ['volume', 'Volume', 'VOLUME', '成交量'],
'date': ['date', 'Date', '日期', 'trade_date'],
'symbol': ['symbol', '代码', '股票代码'],
'adj_factor': ['adj_factor', '复权因子'],
}
def standardize_columns(df, symbol=None):
# 列名小写化
df.columns = [c.lower().strip() for c in df.columns]
# 映射到标准列名
rename = {}
for std_name, aliases in COLUMN_MAP.items():
for alias in aliases:
if alias.lower() in df.columns:
rename[alias.lower()] = std_name
df = df.rename(columns=rename)
if 'symbol' not in df.columns and symbol:
df['symbol'] = symbol
return df
print('Column standardization function defined')
"3.2 日期格式统一化
python -c "
import pandas as pd
# 不同数据源的日期格式
samples = ['2024-01-15', '20240115', '2024/01/15', '2024年01月15日']
for s in samples:
try:
dt = pd.to_datetime(s)
print(f'{s:20s} -> {dt.strftime(\"%Y-%m-%d\")}')
except Exception as e:
print(f'{s:20s} -> ERROR: {e}')
"3.3 复权处理标准化
python -c "
import numpy as np
import pandas as pd
# 如果数据源只提供不复权价格，手动计算复权
# 复权因子通常是累计的
def apply_adjustment(df):
'''
使用复权因子将不复权价格转换为前复权价格
'''
if 'adj_factor' not in df.columns:
# 若无复权因子，以后复权因子为 1.0
df['adj_factor'] = 1.0
# 前复权：当前价格不变，历史价格按累计因子调整
last_factor = df['adj_factor'].iloc[-1]
for col in ['open', 'high', 'low', 'close']:
if col in df.columns:
df[col] = df[col] * df['adj_factor'] / last_factor
return df
print('Adjustment function defined')
print('Forward adjustment: historical prices adjusted down, latest unchanged')
"3.4 数据完整性检查
python -c "
import pandas as pd
import numpy as np
def validate_ohlcv(df):
checks = {}
# 基本检查
checks['has_columns'] = all(
c in df.columns for c in ['open', 'high', 'low', 'close', 'volume']
)
checks['no_nan'] = df[['open', 'high', 'low', 'close', 'volume']].notna().all().all()
checks['high_ge_low'] = (df['high'] >= df['low']).all()
checks['high_ge_open'] = (df['high'] >= df['open']).all()
checks['high_ge_close'] = (df['high'] >= df['close']).all()
checks['low_le_open'] = (df['low'] <= df['open']).all()
checks['low_le_close'] = (df['low'] <= df['close']).all()
checks['non_negative'] = (df[['open', 'high', 'low', 'close', 'volume']] >= 0).all().all()
checks['sorted_by_date'] = df['date'].is_monotonic_increasing
all_passed = all(checks.values())
for check, result in checks.items():
status = 'OK' if result else 'FAIL'
print(f' [{status}] {check}')
return all_passed
print('Validation function defined')
"第四节：数据本地存储 —— HDF5 与 Parquet
4.1 为什么本地存储
量化回测会反复读取历史数据。每次都从网络获取不仅慢，而且不稳定。把数据缓存在本地才是正道。
4.2 HDF5 存储（Pandas 原生支持）
python -c "
import pandas as pd
import numpy as np
import os
# 创建示例数据
np.random.seed(42)
dates = pd.date_range('2024-01-01', periods=100, freq='B')
df = pd.DataFrame({
'open': 100 + np.cumsum(np.random.randn(100) * 2),
'high': 100 + np.cumsum(np.random.randn(100) * 2),
'low': 100 + np.cumsum(np.random.randn(100) * 2),
'close': 100 + np.cumsum(np.random.randn(100) * 2),
'volume': np.random.randint(1e6, 5e6, 100),
}, index=dates)
# 写入 HDF5
h5_path = '/tmp/market_data.h5'
df.to_hdf(h5_path, key='AAPL', mode='w')
print(f'Data written to {h5_path}')
# 读取
df_loaded = pd.read_hdf(h5_path, key='AAPL')
print(f'Loaded {len(df_loaded)} rows')
# 文件大小
print(f'File size: {os.path.getsize(h5_path)} bytes')
"4.3 Parquet 格式（列式存储，更快）
python -c "
import pandas as pd
import numpy as np
import os
np.random.seed(42)
dates = pd.date_range('2024-01-01', periods=1000, freq='B')
df = pd.DataFrame({
'symbol': ['AAPL'] * 1000,
'date': dates,
'close': 150 + np.cumsum(np.random.randn(1000) * 2),
'volume': np.random.randint(1e6, 1e7, 1000),
})
parquet_path = '/tmp/market_data.parquet'
df.to_parquet(parquet_path, index=False)
loaded = pd.read_parquet(parquet_path)
print(f'Parquet: {os.path.getsize(parquet_path)} bytes, {len(loaded)} rows')
"存储格式对比：
| 格式 | 压缩率 | 读取速度 | Pandas 支持 | 跨语言支持 |
|---|---|---|---|---|
| CSV | 差 | 慢 | ||
| HDF5 | 好 | 快 | 一般 | |
| Parquet | 极好 | 极快 | (pyarrow) | |
| Feather | 中 | 极快 | (pyarrow) |
练习
以下题目用于验证本章所学内容：
| 题号 | 题目 | 链接 | 涉及知识点 |
|---|---|---|---|
| — | 本章无对应力扣题 | — | 请用动手练习题自检 |