Python脚本如何修复局部同步缺失数据

wen python案例 2026-06-25 277

本文目录导读：

时间序列数据的插值方法
前向/后向填充
使用scipy的高级插值
移动平均修复
多变量修复（利用相关变量）
完整修复示例
异常值检测与修复
选择建议

我来介绍几种修复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)