3、数据获取与清洗：使用Tushare/Akshare获取宏观数据、缺失值处理、异常值检测
各位同学，咱们直接进入正题。宏观择时信号的质量，说白了就取决于两个东西：数据的准确性和时效性。你想想看，如果拿到的CPI数据都是错的，那后面再牛的模型也是白搭。这一章，我就带你走一遍我从数据获取到清洗的全流程，把那些坑都踩平了。
3.1 数据源的选择：Tushare vs. Akshare
我个人习惯把这两个库比作「左膀右臂」。Tushare是老牌劲旅，数据质量高，但需要积分；Akshare是后起之秀，免费且覆盖广，但偶尔会有小脾气。我在项目中通常这样搭配：
- Tushare：用于获取核心宏观指标（GDP、CPI、PMI、社融等），因为它的历史数据回溯最完整。
- Akshare：用于获取高频或实时数据（如央行公开市场操作、shibor利率），因为它的更新速度更快。
3.2 实战：用Tushare拉取宏观数据
先看代码。假设我们要获取近5年的CPI和PPI数据：
import tushare as ts
import pandas as pd
# 设置token（记得换成你自己的）
ts.set_token('你的token')
pro = ts.pro_api()
# 获取CPI数据
df_cpi = pro.cn_cpi(start_date='20190101', end_date='20241231')
print(df_cpi.head())
嗯，这里要注意。Tushare返回的数据格式是宽表，月份是索引，指标是列。我个人习惯先把它转成长表，方便后续合并：
# 转成长表
df_cpi_long = df_cpi.melt(id_vars=['month'], var_name='indicator', value_name='value')
print(df_cpi_long.head())
为什么会这样？因为后续你要合并多个指标（CPI、PPI、PMI），长表格式用pandas的merge操作会顺手很多。我在项目中吃过亏，一开始用宽表合并，结果字段名冲突，排查了半天。
3.3 用Akshare补充高频数据
Akshare的接口风格比较「野」，但胜在直接。比如获取Shibor利率：
import akshare as ak
# 获取Shibor数据
df_shibor = ak.rate_interbank(market="上海银行间同业拆放利率",
symbol="Shibor",
indicator="隔夜")
print(df_shibor.tail())
df['date'] = pd.to_datetime(df['date'])。
3.4 缺失值处理：别让NaN毁了你的信号
宏观数据最头疼的就是缺失值。比如节假日、数据发布延迟、统计口径调整，都会导致NaN。我总结了一套「三步走」策略：
- 先看缺失比例：如果某个指标缺失超过30%，我建议直接放弃，或者找替代指标。
- 再判断缺失类型：是随机缺失还是结构性缺失？比如2020年2月的PMI数据异常低，那是疫情导致的，不能简单填充。
- 最后选填充方法：
| 缺失类型 | 推荐方法 | 代码示例 |
|---|---|---|
| 随机缺失（少量） | 前向填充（ffill） | df.fillna(method='ffill') |
| 周期性数据 | 线性插值 | df.interpolate(method='linear') |
| 结构性缺失 | 模型预测填充 | 用ARIMA或简单回归 |
举个例子，CPI数据偶尔会缺失某个月份。我一般先用前向填充，因为宏观数据有惯性，上个月的值往往能代表这个月。但如果连续缺失3个月以上，我会改用插值：
# 前向填充
df_cpi['cpi_filled'] = df_cpi['cpi'].fillna(method='ffill')
# 线性插值（适用于连续缺失）
df_cpi['cpi_interp'] = df_cpi['cpi'].interpolate(method='linear')
3.5 异常值检测：揪出那些「离谱」的数据
异常值在宏观数据里很常见。比如某个月份的M2增速突然飙升到20%，那大概率是统计口径调整了。我常用的检测方法有三种：
- 3σ原则：适用于正态分布的数据。超出均值±3倍标准差的值视为异常。
- IQR方法：适用于偏态分布。用四分位距（Q3-Q1）来界定异常。
- 移动窗口法：适用于时间序列。看当前值是否偏离前后N期的均值太远。
我个人最常用的是移动窗口法，因为宏观数据有趋势性，单纯用全局统计容易误判。比如GDP增速在下降通道中，用3σ可能会把正常值判为异常：
# 移动窗口法检测异常
window = 12 # 12个月窗口
df['rolling_mean'] = df['gdp'].rolling(window=window).mean()
df['rolling_std'] = df['gdp'].rolling(window=window).std()
df['is_outlier'] = (abs(df['gdp'] - df['rolling_mean']) > 3 * df['rolling_std'])
3.6 数据清洗的完整流程（SVG流程图）
下面这张图是我在项目中实际使用的清洗流程，你可以直接拿去用：
3.7 实战案例：清洗社融数据
最后，我拿社融数据举个例子，把整个流程串起来。社融数据经常有季节性波动和统计口径调整，是清洗的「硬骨头」：
# 1. 获取数据
df_sr = pro.cn_social_financing(start_date='20180101', end_date='20241231')
# 2. 缺失值处理（社融偶尔会缺失某个月）
df_sr['total'] = df_sr['total'].fillna(method='ffill')
# 3. 异常值检测（用移动窗口）
df_sr['ma'] = df_sr['total'].rolling(window=12).mean()
df_sr['std'] = df_sr['total'].rolling(window=12).std()
df_sr['z_score'] = (df_sr['total'] - df_sr['ma']) / df_sr['std']
# 标记异常
df_sr['flag'] = df_sr['z_score'].apply(lambda x: '异常' if abs(x) > 3 else '正常')
# 4. 查看异常值
print(df_sr[df_sr['flag'] == '异常'][['month', 'total', 'z_score']])
好了，数据获取和清洗这块就讲到这里。记住一句话：数据清洗花的时间，永远值得。你花80%的时间把数据搞干净，后面20%的时间就能把模型跑出花来。