本文目录导读：
我来介绍几种修复Python中局部同步缺失数据的方法：
时间序列数据的插值方法
线性插值
import pandas as pd
import numpy as np
# 创建示例数据（包含缺失值）
df = pd.DataFrame({
'timestamp': pd.date_range('2024-01-01', periods=10, freq='H'),
'value': [10, np.nan, 15, np.nan, np.nan, 20, 18, np.nan, 22, 25]
})
# 线性插值
df['value_interpolated'] = df['value'].interpolate(method='linear')
print("线性插值结果:")
print(df[['timestamp', 'value', 'value_interpolated']])
时间序列插值
# 时间序列插值（考虑时间间隔）
df['value_time'] = df['value'].interpolate(method='time')
print("时间序列插值:")
print(df[['timestamp', 'value', 'value_time']])
前向/后向填充
# 前向填充
df['value_ffill'] = df['value'].fillna(method='ffill')
print("前向填充:")
print(df[['timestamp', 'value', 'value_ffill']])
# 后向填充
df['value_bfill'] = df['value'].fillna(method='bfill')
print("后向填充:")
print(df[['timestamp', 'value', 'value_bfill']])
使用scipy的高级插值
from scipy import interpolate
import numpy as np
# 示例数据
x = np.array([0, 1, 2, 3, 4, 5]) # 已知点
y = np.array([0, np.nan, 4, np.nan, np.nan, 10])
# 移除NaN值进行拟合
mask = ~np.isnan(y)
x_clean = x[mask]
y_clean = y[mask]
# 创建插值函数
f = interpolate.interp1d(x_clean, y_clean, kind='cubic', fill_value='extrapolate')
# 在所有点上插值
y_interpolated = f(x)
print("使用scipy的三次插值:", y_interpolated)
移动平均修复
# 使用移动平均
df['value_rolling'] = df['value'].rolling(window=3, min_periods=1).mean()
print("移动平均修复:")
print(df[['timestamp', 'value', 'value_rolling']])
多变量修复（利用相关变量）
# 创建相关变量
df['related_var'] = [5, 8, 7, 10, 9, 12, 11, 14, 13, 16]
# 使用线性回归预测缺失值
from sklearn.linear_model import LinearRegression
# 准备训练数据（非缺失值）
train_data = df[df['value'].notna()]
X_train = train_data[['related_var']]
y_train = train_data['value']
# 训练模型
model = LinearRegression()
model.fit(X_train, y_train)
# 预测缺失值
missing_data = df[df['value'].isna()]
X_missing = missing_data[['related_var']]
df.loc[df['value'].isna(), 'value_regression'] = model.predict(X_missing)
print("回归修复结果:")
print(df)
完整修复示例
def fix_missing_data(data, method='interpolate', window=3):
"""
综合修复数据缺失函数
参数:
data: pandas Series或DataFrame
method: 'interpolate', 'ffill', 'bfill', 'rolling', 'auto'
window: 移动平均窗口大小
"""
if method == 'interpolate':
return data.interpolate(method='linear')
elif method == 'ffill':
return data.fillna(method='ffill')
elif method == 'bfill':
return data.fillna(method='bfill')
elif method == 'rolling':
return data.rolling(window=window, min_periods=1).mean()
elif method == 'auto':
# 自动选择最佳方法
missing_pct = data.isna().sum() / len(data) * 100
if missing_pct < 5:
return data.interpolate(method='linear')
elif missing_pct < 20:
return data.rolling(window=3, min_periods=1).mean()
else:
return data.fillna(method='ffill')
else:
raise ValueError("Unsupported method")
# 使用示例
fixed_data = fix_missing_data(df['value'], method='auto')
print("修复后的数据:", fixed_data)
异常值检测与修复
def detect_and_fix_outliers(data, threshold=3):
"""
检测并修复异常值
"""
# Z-score方法
z_scores = np.abs((data - data.mean()) / data.std())
outliers = z_scores > threshold
# 替换异常值为中位数
data_fixed = data.copy()
data_fixed[outliers] = data.median()
return data_fixed
# 使用示例
data_with_outliers = pd.Series([10, 12, 11, 100, 13, 12, 150, 11, 13, 12])
fixed_data = detect_and_fix_outliers(data_with_outliers)
print("修复异常值前:", data_with_outliers.values)
print("修复异常值后:", fixed_data.values)
选择建议
- 数据缺失较少 (<5%)：使用线性插值
- 连续缺失：使用前向填充
- 周期性数据：使用时间序列插值
- 有相关变量：使用回归修复
- 噪声数据：使用移动平均
根据你的具体数据特点和应用场景选择合适的修复方法。
