3、宏观经济数据获取:使用pandas-datareader获取FRED数据、使用akshare获取国内宏观数据、API调用基础

做宏观因子回测,第一步就是搞数据。这活儿看着简单,其实坑不少。我见过太多人花80%的时间在数据清洗上,就是因为源头没选对。

今天咱们就聊聊怎么高效拿到宏观经济数据。说白了就三个路子:FRED(美联储经济数据库)、akshare(国产神器)、以及通用的API调用。我一个个说。

3.1 用pandas-datareader拉FRED数据

FRED是美联储旗下的经济数据库。全球宏观研究者几乎都在用。它覆盖了美国及全球主要经济体的GDP、CPI、失业率、利率等核心指标。

我个人习惯用pandas-datareader来拉FRED数据。这库封装好了,几行代码就能搞定。

核心逻辑:你只需要知道FRED的「系列ID」(比如GDP是"GDP"、CPI是"CPIAUCSL"),然后调用DataReader函数即可。

先安装依赖:

pip install pandas-datareader

然后看代码:

import pandas as pd
import pandas_datareader.data as web
from datetime import datetime

# 定义要获取的指标
series_dict = {
    'GDP': 'GDP',           # 美国实际GDP
    'CPI': 'CPIAUCSL',      # 美国CPI
    'UNRATE': 'UNRATE',     # 美国失业率
    'FEDFUNDS': 'FEDFUNDS'  # 联邦基金利率
}

# 设置时间范围
start = datetime(2000, 1, 1)
end = datetime(2024, 12, 31)

# 批量获取数据
df = web.DataReader(list(series_dict.keys()), 'fred', start, end)

# 看一眼数据
print(df.head())

嗯,这里要注意:FRED的数据频率不一样。GDP是季度数据,CPI是月度数据,利率是日度数据。你拿回来之后,得统一频率。我一般用resample转成月度或季度。

小技巧:FRED的系列ID可以在官网搜索。比如你想找「中国GDP」,搜"China GDP"就能找到对应的ID。不过FRED主要覆盖美国数据,中国的指标相对有限。

我在项目中遇到过一个问题:FRED的API有调用次数限制。免费用户每分钟最多调用120次。如果你要拉几百个指标,建议分批处理,中间加个time.sleep(1)

避坑指南:我曾经一次性拉了500个指标,结果被FRED封了IP半小时。后来学乖了,每次最多拉50个,间隔1秒。

3.2 用akshare拿国内宏观数据

做A股或国内宏观研究,FRED就不太够用了。这时候我推荐akshare。这库是国人开发的,覆盖了央行、统计局、交易所等官方数据源。

akshare的特点是什么?说白了就是「全而杂」。它把国内公开的宏观数据几乎都扒下来了,但数据格式不太统一,需要你花点心思清洗。

先安装:

pip install akshare

拿几个常用指标看看:

import akshare as ak

# 1. 中国GDP(季度)
df_gdp = ak.macro_china_gdp()
print(df_gdp.tail())

# 2. 中国CPI(月度)
df_cpi = ak.macro_china_cpi_monthly()
print(df_cpi.tail())

# 3. 中国PMI(月度)
df_pmi = ak.macro_china_pmi()
print(df_pmi.tail())

# 4. 社会融资规模(月度)
df_social_financing = ak.macro_china_social_financing()
print(df_social_financing.tail())

你看,函数名很直观。但有个问题:akshare返回的DataFrame列名是中文的,而且不同函数的列名风格不一致。比如GDP的列名是「季度」「国内生产总值」,CPI的列名是「月份」「当月同比」。你得手动统一。

我的做法:写一个统一的清洗函数,把列名转成英文,日期转成标准格式,缺失值用前向填充。

def clean_akshare_data(df, date_col, value_col, new_name):
    """
    统一清洗akshare数据
    """
    df = df[[date_col, value_col]].copy()
    df.columns = ['date', new_name]
    df['date'] = pd.to_datetime(df['date'])
    df = df.sort_values('date')
    df = df.set_index('date')
    # 缺失值用前向填充
    df = df.fillna(method='ffill')
    return df

# 使用示例
df_gdp_clean = clean_akshare_data(
    df_gdp, '季度', '国内生产总值', 'GDP_China'
)

akshare还有一个好处:它支持实时数据。比如你想拿今天的Shibor利率,直接调ak.rate_interbank()就行。这在做高频宏观因子时特别有用。

个人经验:akshare的数据源有时会更新延迟。比如统计局每月10号左右发布CPI,但akshare可能要晚1-2天才能同步。如果你做实时策略,建议用官方API或付费数据源。

3.3 API调用基础

FRED和akshare本质上都是对API的封装。但有时候你需要自己写API调用,比如对接Wind、Bloomberg、或者一些付费数据服务。

API调用的核心就三步:

  1. 构造请求URL:包含基础地址、参数、API Key
  2. 发送请求:用requests库的GET或POST方法
  3. 解析响应:通常是JSON格式,转成DataFrame

我拿一个虚拟的宏观数据API举个例子:

import requests
import pandas as pd

# 假设的API配置
API_KEY = 'your_api_key_here'
BASE_URL = 'https://api.macrodata.com/v1'

# 构造请求参数
params = {
    'series': 'CPI',
    'country': 'CN',
    'start_date': '2020-01-01',
    'end_date': '2024-12-31',
    'api_key': API_KEY
}

# 发送GET请求
response = requests.get(f'{BASE_URL}/series', params=params)

# 检查状态码
if response.status_code == 200:
    data = response.json()
    # 解析JSON为DataFrame
    df = pd.DataFrame(data['observations'])
    print(df.head())
else:
    print(f'请求失败,状态码:{response.status_code}')

这里有几个坑,我踩过不止一次:

  • API Key泄露:千万别把Key硬编码在代码里。用环境变量或配置文件。
  • 频率限制:很多API有每分钟/每天的调用上限。建议用time.sleep控制节奏。
  • 数据格式:返回的JSON结构千奇百怪。有的把日期放在"date"字段,有的放在"time"。你得先打印出来看看。

避坑指南:我曾经在循环里调API,忘了加延迟,结果被对方服务器拉黑了。后来我写了个重试机制,遇到429错误(Too Many Requests)就自动等待30秒再试。

3.4 本章知识体系

为了让你更直观地理解,我画了一张流程图。它展示了从数据源到最终DataFrame的完整链路。

宏观经济数据获取流程 FRED数据库 国内数据源 其他API pandas-datareader akshare requests + JSON解析 统一格式的DataFrame(日期索引 + 数值列) 进入宏观因子回测模块

你看,流程其实不复杂。关键是把数据源、工具、输出这三层打通。我个人的建议是:

  • 做全球宏观,优先用FRED + pandas-datareader
  • 做国内宏观,优先用akshare
  • 做定制化需求,自己写API调用

数据拿到手之后,别急着用。先检查一下日期范围、缺失值、异常值。我见过有人拿到的CPI数据全是NaN,因为没注意akshare的列名变了。嗯,这些细节决定了你的回测结果靠不靠谱。

总结一句话:数据获取是宏观因子回测的「地基」。地基不稳,后面盖的楼再漂亮也是危房。


专注资料整理