4. 数据源接入：免费/付费数据源对比与API封装
做量化交易，第一件事就是搞数据。没有数据，策略就是空中楼阁。我见过不少新手，上来就写策略，结果回测跑得飞起，实盘一跑就崩——为什么？数据源不对，或者数据质量太差。
今天咱们就把数据源这件事彻底聊透。我会从免费到付费，从API封装到限流处理，把我在实战中踩过的坑、总结的经验，一次性讲清楚。
4.1 免费数据源：够用吗？
说实话，免费数据源对于学习、回测、甚至小规模实盘，完全够用。但你要知道它的边界在哪里。
Yahoo Finance
我个人最常用的免费数据源。接口简单，覆盖全球股票、ETF、指数、加密货币。用 yfinance 库，几行代码就能拉数据。
import yfinance as yf
# 获取苹果股票数据
aapl = yf.download('AAPL', start='2023-01-01', end='2023-12-31')
print(aapl.head())
但注意，Yahoo Finance 的数据有时会延迟，而且历史数据可能被调整过（比如拆股、分红）。我在项目中遇到过，回测时用的数据跟实际行情对不上，后来发现是 Yahoo 的调整因子没处理好。
Alpha Vantage
这个数据源提供更丰富的金融数据，包括基本面、技术指标、外汇、加密货币等。API 设计得很规范，但免费版有每分钟5次请求的限制。
import requests
API_KEY = 'your_api_key'
symbol = 'IBM'
url = f'https://www.alphavantage.co/query?function=TIME_SERIES_DAILY&symbol={symbol}&apikey={API_KEY}'
response = requests.get(url)
data = response.json()
Alpha Vantage 的数据质量比 Yahoo 高，但免费版的历史数据只有最近100天。如果你要做长周期回测，就得付费了。
Tushare（国内）
做A股的朋友，Tushare 是绕不开的。它提供股票、基金、期货、宏观经济等数据。免费版有积分限制，积分越高，能调用的数据越多。
import tushare as ts
# 设置token
ts.set_token('your_token')
pro = ts.pro_api()
# 获取日线数据
df = pro.daily(ts_code='000001.SZ', start_date='20230101', end_date='20231231')
print(df.head())
Tushare 的数据更新及时，但免费版每天有调用次数限制。我建议你先用免费版做原型验证，等策略稳定了再考虑付费。
4.2 付费数据源：值不值？
当你的策略开始赚钱，或者你需要更高质量的数据时，付费数据源就是必然选择。
| 数据源 | 价格 | 数据质量 | 适用场景 |
|---|---|---|---|
| Wind（万得） | 约2-5万/年 | 极高 | 机构级、高频、全品种 |
| Bloomberg | 约2万美元/年 | 极高 | 全球市场、实时数据 |
| Quandl | 按数据包收费 | 高 | 另类数据、宏观经济 |
| 聚宽/JQData | 约3000-1万/年 | 高 | A股、期货、回测 |
Wind 是我在机构工作时最常用的。它的数据覆盖全，更新快，但价格也确实贵。如果你只是个人交易者，我建议先用聚宽或 Tushare 付费版，性价比更高。
4.3 API封装：让数据调用更优雅
直接调用原始API，代码会变得很乱。我习惯封装一个统一的数据接口，这样切换数据源时，只需要改一行代码。
class DataSource:
def __init__(self, source='yahoo', api_key=None):
self.source = source
self.api_key = api_key
def get_daily_data(self, symbol, start, end):
if self.source == 'yahoo':
return self._get_yahoo_data(symbol, start, end)
elif self.source == 'alphavantage':
return self._get_alphavantage_data(symbol, start, end)
elif self.source == 'tushare':
return self._get_tushare_data(symbol, start, end)
else:
raise ValueError(f'Unknown source: {self.source}')
def _get_yahoo_data(self, symbol, start, end):
import yfinance as yf
return yf.download(symbol, start=start, end=end)
def _get_alphavantage_data(self, symbol, start, end):
# 具体实现略
pass
def _get_tushare_data(self, symbol, start, end):
# 具体实现略
pass
# 使用示例
ds = DataSource(source='yahoo')
data = ds.get_daily_data('AAPL', '2023-01-01', '2023-12-31')
这样封装的好处是，你可以在不同数据源之间自由切换。比如白天用 Tushare 做A股，晚上用 Yahoo 做美股，代码结构完全一致。
4.4 限流处理：别被API封了
免费API都有频率限制。我曾经因为没做限流，直接写了个循环去拉数据，结果IP被封了三天。嗯，从那以后我再也不敢忽视限流了。
限流的核心思路就两个：
- 控制请求频率：每次请求后等待固定时间
- 重试机制：请求失败后自动重试
import time
import requests
from functools import wraps
def rate_limit(max_calls=5, period=60):
"""限流装饰器：每分钟最多调用 max_calls 次"""
def decorator(func):
last_reset = time.time()
calls = 0
@wraps(func)
def wrapper(*args, **kwargs):
nonlocal last_reset, calls
# 重置计数器
if time.time() - last_reset > period:
calls = 0
last_reset = time.time()
# 检查是否超限
if calls >= max_calls:
sleep_time = period - (time.time() - last_reset)
if sleep_time > 0:
print(f'达到限流，等待 {sleep_time:.2f} 秒')
time.sleep(sleep_time)
calls = 0
last_reset = time.time()
calls += 1
return func(*args, **kwargs)
return wrapper
return decorator
@rate_limit(max_calls=5, period=60)
def fetch_data(url):
response = requests.get(url)
return response.json()
# 使用
for i in range(10):
data = fetch_data('https://api.example.com/data')
print(f'第 {i+1} 次请求成功')
这个装饰器可以复用到任何API调用上。你只需要设置每分钟最大请求次数，剩下的交给它处理。
4.5 知识体系总览
下面这张图，是我对数据源接入的整体理解。你可以把它当作一个决策树，帮你快速选择合适的数据源和封装方式。
这张图把数据源接入分成了三层：底层是数据源选择（免费 vs 付费），中间层是API封装，顶层是限流处理。每一层都有对应的技术方案和注意事项。
好了，数据源接入这部分就聊到这里。记住一句话：数据是量化交易的基础，但不要被数据绑架。先用免费数据跑通逻辑，再逐步升级到付费数据源。这样既省钱，又高效。