第3章：数据获取基础：金融数据源介绍与API调用

做量化投研，第一步就是搞定数据。

这话听起来简单，但我在项目里见过太多人栽在这上面。数据不对，后面所有模型都是白搭。今天咱们就把金融数据源这件事彻底聊透。

3.1 三大主流金融数据源

国内做量化，绕不开三个数据源：Tushare、AKShare、Yahoo Finance。我个人的习惯是，根据场景选工具，而不是死磕一个。

3.1.1 Tushare

Tushare 是国内老牌的金融数据接口，数据质量很高。它覆盖了股票、基金、期货、宏观经济等。不过要注意，Tushare 现在需要积分才能获取完整数据。新用户注册送100分，够你玩一阵子。

# Tushare 基础调用示例
import tushare as ts

# 设置 token（去官网注册获取）
ts.set_token('your_token_here')
pro = ts.pro_api()

# 获取日线数据
df = pro.daily(ts_code='000001.SZ', start_date='20240101', end_date='20240301')
print(df.head())

3.1.2 AKShare

AKShare 是后起之秀，完全开源免费。它的数据源来自各大财经网站，覆盖面极广。我曾经在做一个期货策略时，Tushare 的期货数据需要高积分，AKShare 直接免费搞定。

# AKShare 基础调用示例
import akshare as ak

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

3.1.3 Yahoo Finance

Yahoo Finance 主要做海外市场。美股、港股、全球指数，它都能拿。但有个坑——国内访问不稳定。我建议用 yfinance 这个 Python 库来调，它封装得比较好。

# Yahoo Finance 基础调用示例
import yfinance as yf

# 获取苹果股票数据
aapl = yf.Ticker("AAPL")
df = aapl.history(period="3mo")
print(df.head())

3.2 API 调用基础

说白了，API 调用就是向服务器发请求，拿数据回来。这里面有几个关键点，我踩过的坑不少。

3.2.1 请求频率限制

几乎所有数据源都有频率限制。Tushare 普通用户每秒最多调 2 次，AKShare 虽然没有明确限制，但调太快会被封 IP。我曾经写了个循环一次性拉了 5000 只股票的数据，结果 IP 被禁了 24 小时。

import time
import tushare as ts

pro = ts.pro_api()
stock_list = ['000001.SZ', '000002.SZ', '000003.SZ']

for stock in stock_list:
    df = pro.daily(ts_code=stock, start_date='20240101', end_date='20240301')
    print(f"{stock} 数据获取成功")
    time.sleep(0.5)  # 每次请求间隔 0.5 秒

3.2.2 错误处理

网络请求一定会出错。网络波动、服务器维护、参数错误，各种情况。我习惯用 try-except 包一层，再配合重试机制。

import time
import akshare as ak

def safe_get_data(symbol, retries=3):
    for i in range(retries):
        try:
            df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
                                    start_date="20240101", end_date="20240301")
            return df
        except Exception as e:
            print(f"第 {i+1} 次尝试失败: {e}")
            time.sleep(1)
    return None

df = safe_get_data("000001")
if df is not None:
    print("数据获取成功")

3.3 数据格式与字段说明

不同数据源返回的字段名不一样，但核心字段大同小异。我整理了一张对照表，方便你理解。

字段含义	Tushare	AKShare	Yahoo Finance
股票代码	ts_code	symbol	无（需自行指定）
交易日期	trade_date	date	Date
开盘价	open	开盘	Open
收盘价	close	收盘	Close
最高价	high	最高	High
最低价	low	最低	Low
成交量	vol	成交量	Volume

3.4 数据缓存策略

为什么要做缓存？原因很简单——快和省。快是读取速度，省是省 API 调用次数。我见过有人每次回测都重新拉一遍数据，白白浪费积分和时间。

3.4.1 本地文件缓存

最简单的方案，把数据存成 CSV 或 Parquet 文件。下次读取时先检查本地有没有。

import os
import pandas as pd

def get_data_with_cache(symbol, start_date, end_date):
    cache_file = f"data/{symbol}_{start_date}_{end_date}.parquet"
    
    # 检查缓存
    if os.path.exists(cache_file):
        print("从缓存读取数据")
        return pd.read_parquet(cache_file)
    
    # 没有缓存，从 API 获取
    print("从 API 获取数据")
    df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
                            start_date=start_date, end_date=end_date)
    
    # 保存到缓存
    os.makedirs("data", exist_ok=True)
    df.to_parquet(cache_file)
    
    return df

3.4.2 增量更新策略

全量拉取太慢了。我习惯只拉取最新数据，追加到已有文件里。比如每天收盘后，只拉当天数据。

def incremental_update(symbol):
    cache_file = f"data/{symbol}_full.parquet"
    
    # 读取已有数据
    if os.path.exists(cache_file):
        existing_df = pd.read_parquet(cache_file)
        last_date = existing_df['date'].max()
    else:
        existing_df = pd.DataFrame()
        last_date = None
    
    # 获取最新数据
    if last_date:
        new_df = ak.stock_zh_a_hist(symbol=symbol, period="daily",
                                    start_date=last_date, end_date=pd.Timestamp.today().strftime('%Y%m%d'))
        # 去重合并
        combined = pd.concat([existing_df, new_df]).drop_duplicates(subset=['date'])
    else:
        combined = ak.stock_zh_a_hist(symbol=symbol, period="daily",
                                      start_date="20200101", end_date=pd.Timestamp.today().strftime('%Y%m%d'))
    
    combined.to_parquet(cache_file)
    print(f"数据更新完成，共 {len(combined)} 条记录")

3.5 本章知识体系

下面这张图，帮你理清数据获取的整体流程。

数据获取这件事，说白了就是三个环节：选对源、调对接口、存好数据。每一步都有坑，但踩过一次就记住了。我个人觉得，先把 AKShare 玩熟，再慢慢扩展到其他数据源，是最稳妥的路径。