2. 金融数据获取与清洗：宏观经济指标

任务与环境

从 FRED 公开 CSV 接口（fredgraph.csv，无需 ）拉取 GDP、CPI、联邦基金利率三个宏观指标，统一为季度频率，对齐时间轴并处理缺失值，输出一张干净的宽表。依赖：pandas、requests（pip install pandas requests）。

实现思路

完整代码

import io
import pandas as pd
import requests

SERIES = {
    "GDP": "GDP",              # 季度，名义 GDP（十亿美元）
    "CPI": "CPIAUCSL",         # 月度，CPI 指数
    "FEDFUNDS": "FEDFUNDS",    # 月度，联邦基金利率
}
BASE = "https://fred.stlouisfed.org/graph/fredgraph.csv?id={sid}"

def fetch_series(series_id: str) -> pd.Series:
    """从 FRED 拉取单个序列，返回以日期为索引的 Series。"""
    resp = requests.get(BASE.format(sid=series_id), timeout=30)
    resp.raise_for_status()
    df = pd.read_csv(io.StringIO(resp.text))
    df.columns = ["date", "value"]
    df["date"] = pd.to_datetime(df["date"])
    s = df.set_index("date")["value"]
    # FRED 用 "." 表示缺失
    s = pd.to_numeric(s, errors="coerce")
    return s.sort_index()

def to_quarterly(s: pd.Series) -> pd.Series:
    """统一到季度频率：取每季度最后一个观测值。"""
    return s.resample("QE").last()

def align_and_fill(series_dict: dict) -> pd.DataFrame:
    """对齐时间轴并处理缺失值。"""
    frames = {name: to_quarterly(fetch_series(sid))
              for name, sid in series_dict.items()}
    wide = pd.concat(frames, axis=1).sort_index()
    # 前向填充：季度指标发布滞后，用上一期值补当期缺口
    wide = wide.ffill()
    # 丢弃因起始时间不一致导致的前导 NaN 行
    wide = wide.dropna(how="any")
    return wide

if __name__ == "__main__":
    df = align_and_fill(SERIES)
    print(df.tail())
    print("shape:", df.shape)
    print("index freq:", pd.infer_freq(df.index))
    print("any NaN:", df.isna().any().any())

运行与验证

直接运行 python macro_clean.py。判定通过条件：

• 打印的 index freq 为 QE（季末频率）；
• any NaN 为 False，即宽表无缺失值；
• 列名包含 GDP、CPI、FEDFUNDS，且 shape 行数等于三序列共同覆盖的季度数。

若网络受限，可把 fetch_series 中的 requests.get 换成读取本地已下载的 CSV 文件，其余逻辑不变。

关键点

• resample("QE").last() 把月度 CPI/利率降为季度，与 GDP 对齐；GDP 本身已是季度，重采样不改变其值。
• ffill() 处理发布滞后造成的缺口；dropna(how="any") 只丢弃三序列尚未同时存在的早期季度，避免用填充值伪造历史。
• FRED 缺失标记 "." 通过 pd.to_numeric(errors="coerce") 转为 NaN，再统一填充。