4. 数据获取与清洗：使用pandas-datareader获取历史汇率数据，处理缺失值与异常值

做汇率预测，第一步不是建模，而是搞数据。

我见过太多人，模型调得飞起，结果数据源就有问题。最后预测出来的东西，自己都不敢信。说白了，数据质量决定了模型的天花板。今天我们就来聊聊，怎么用 pandas-datareader 把历史汇率数据搞到手，再把它收拾得干干净净。

4.1 为什么选 pandas-datareader？

市面上能拿汇率数据的库不少，但我个人最常用 pandas-datareader。原因很简单：它把多个数据源的接口统一了。你写一套代码，就能从 Yahoo Finance、FRED、Alpha Vantage 这些地方拿数据。切换数据源，改个参数就行。

举个例子，你想拿美元兑人民币的汇率。用 FRED 的数据，代码就几行：

import pandas_datareader.data as web
import datetime

start = datetime.datetime(2010, 1, 1)
end = datetime.datetime(2023, 12, 31)

# 从 FRED 获取 DEXCHUS（中国/美国汇率）
df = web.DataReader('DEXCHUS', 'fred', start, end)
print(df.head())

跑完这段，你就拿到了从2010年到2023年的日度汇率数据。嗯，这里要注意：FRED 的代码是固定的，你得提前查好。比如欧元兑美元是 DEXUSEU，英镑是 DEXUSUK。

4.2 数据获取的常见坑

我在项目中遇到过几次数据获取失败的情况。总结下来，主要有三个坑：

• 网络问题：有些数据源需要翻墙，或者请求频率太高被限制。我建议加个重试机制。
• 日期格式：不同数据源的日期格式可能不一样。FRED 返回的是美国东部时间，Yahoo 用的是 UTC。你拿到数据后，第一件事就是统一时区。
• 数据缺失：周末和节假日没有交易数据。这不是异常，是正常现象。但如果你做日频预测，就得处理这些空缺。

import time
from functools import wraps

def retry(max_attempts=3, delay=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"第 {attempt+1} 次尝试失败: {e}")
                    time.sleep(delay)
            raise Exception("所有重试均失败")
        return wrapper
    return decorator

@retry(max_attempts=3)
def fetch_exchange_rate(code, source, start, end):
    return web.DataReader(code, source, start, end)

4.3 缺失值处理：别急着删

拿到数据后，第一件事就是检查缺失值。用 df.isnull().sum() 一看，你可能会发现周末全是 NaN。这很正常，外汇市场周末休市。

但有些缺失值就没那么友好了。比如某一天突然数据没了，可能是数据源的问题，也可能是那天发生了极端事件。怎么处理？我一般分三步走：

1. 先看缺失比例：如果缺失少于5%，可以用前向填充（ffill）或插值。如果超过20%，我建议重新检查数据源。
2. 区分缺失类型：周末缺失是结构性的，用 asfreq('D') 填充。随机缺失用插值，比如 df.interpolate(method='linear')。
3. 别用均值填充：汇率数据有趋势和波动，均值填充会破坏时序结构。我曾经犯过这个错，模型预测结果惨不忍睹。

4.4 异常值检测：那些离谱的跳变

汇率数据里，异常值往往意味着重大事件。比如2015年8月人民币突然贬值，或者2020年3月美元流动性危机。这些不是错误数据，而是市场信号。

但有些异常值纯粹是数据录入错误。比如某一天汇率突然跳了10%，第二天又跳回来。这种就需要处理。

我常用的方法是 Z-score 加滚动窗口：

import numpy as np

# 计算滚动均值和标准差
window = 20
df['rolling_mean'] = df['value'].rolling(window=window).mean()
df['rolling_std'] = df['value'].rolling(window=window).std()

# 计算 Z-score
df['z_score'] = (df['value'] - df['rolling_mean']) / df['rolling_std']

# 标记异常值（Z-score 绝对值大于3）
df['is_outlier'] = np.abs(df['z_score']) > 3

# 处理异常值：用滚动中位数替换
df['value_clean'] = np.where(
    df['is_outlier'],
    df['value'].rolling(window=window).median(),
    df['value']
)

你想想看，为什么用中位数而不是均值？因为中位数对极端值更鲁棒。如果异常值本身是真实的（比如黑天鹅事件），用中位数替换会保留趋势信息，而不是被异常值带偏。

4.5 数据清洗的完整流程

说了这么多，我们来画个流程图，把整个数据获取和清洗的逻辑串起来。

这个流程看起来简单，但每一步都有细节。比如在缺失值检查这一步，我建议你不仅看缺失数量，还要看缺失的分布。如果缺失集中在某一段时间，那可能是数据源的问题，而不是随机缺失。

4.6 实战：完整的数据清洗代码

最后，我把上面讲的东西整合成一个完整的函数。你直接拿去用就行：

def clean_exchange_rate_data(df, window=20, z_threshold=4):
    """
    清洗汇率数据
    - 填充周末缺失
    - 检测并处理异常值
    """
    # 1. 确保日期索引
    df.index = pd.to_datetime(df.index)
    
    # 2. 填充周末缺失（前向填充）
    df = df.asfreq('D').fillna(method='ffill')
    
    # 3. 检查剩余缺失值
    if df.isnull().sum().sum() > 0:
        df = df.interpolate(method='linear')
    
    # 4. 异常值检测
    df['rolling_mean'] = df['value'].rolling(window=window).mean()
    df['rolling_std'] = df['value'].rolling(window=window).std()
    df['z_score'] = (df['value'] - df['rolling_mean']) / df['rolling_std']
    
    # 5. 替换异常值
    df['value_clean'] = np.where(
        np.abs(df['z_score']) > z_threshold,
        df['value'].rolling(window=window).median(),
        df['value']
    )
    
    return df[['value_clean']]

这段代码我用了很多次，基本覆盖了大部分情况。但记住，没有万能代码。你拿到新数据后，最好先画个图看看趋势和波动。眼睛看一遍，比跑一百个统计检验都管用。

---

公众号:蓝海资料掘金营，微信deep3321