3个实战案例教你用Python睽违数据：保姆级教程

看了一堆教程还是不会写项目？别急，这篇保姆级教程带你用Python处理睽违数据，从入门到实战，3个真实案例拆解，让你直接上手。

项目目标

睽违数据指的是在时间序列中，某些数据点与前后数据存在显著偏离，但又不是简单的异常值。比如在建筑工地的施工进度记录中，某天突然停工三天，再复工时进度数据会有跳跃。这种数据如果直接平均，会严重影响后续预测。

我们今天要做的，是用Python识别并处理这类睽违数据。目标很明确：输入原始数据，输出处理后的平滑数据，同时保留关键变化点的信息。

目录结构

整个项目结构很简单，就三个文件：

project/
├── data/
│   └── progress.csv      # 原始施工进度数据
├── src/
│   ├── __init__.py
│   ├── detector.py       # 睽违检测核心逻辑
│   └── processor.py      # 数据处理与平滑
└── main.py               # 主入口

先创建这些目录和文件，代码都写在对应位置。data目录放CSV文件，src放核心逻辑，main.py负责串联整个流程。

核心代码实现

先看detector.py，这是检测睽违的核心。我们不用复杂的统计模型，就用移动平均加阈值判断，简单有效。

import numpy as np
import pandas as pdclass KuivieDetector:def __init__(self, window=7, threshold=2.0):"""window: 移动平均窗口大小，默认7天threshold: 判定睽违的标准差倍数，默认2.0"""self.window = windowself.threshold = thresholddef detect(self, series: pd.Series) -> pd.Series:"""输入时间序列，返回睽违标记（1=睽违，0=正常）"""# 计算移动平均，填充NaN用前后值ma = series.rolling(window=self.window, center=True, min_periods=1).mean()# 计算残差residual = series - ma# 计算残差的标准差std = residual.rolling(window=self.window, center=True, min_periods=1).std()# 归一化残差z_score = residual / std.replace(0, np.nan)# 标记睽违点：|z_score| > thresholdflags = (z_score.abs() > self.threshold).astype(int)# 处理边界NaNflags[flags.isna()] = 0return flags

逐行讲解：

第8-11行，初始化窗口和阈值。窗口7天是因为建筑施工通常以周为单位观察，阈值2.0是统计学常用标准，表示偏离两个标准差以上才认为异常。

第17行，计算移动平均。center=True让平均窗口居中，min_periods=1保证边界也有值。

第20-23行，计算残差和标准差。残差是实际值减平均值，标准差衡量波动程度。

第25-27行，归一化残差。除以标准差后，z_score就是偏离多少个标准差。超过阈值就标记为睽违。

第30行，处理边界产生的NaN，统一设为0，避免后续出错。

再看processor.py，负责处理睽违数据：

import numpy as np
import pandas as pdclass KuivieProcessor:def __init__(self, method='interpolate'):"""method: 处理方式，'interpolate'插值，'forward'前向填充"""self.method = methoddef process(self, series: pd.Series, flags: pd.Series) -> pd.Series:"""输入原始序列和睽违标记，返回处理后序列"""result = series.copy()# 找到所有睽违点索引indices = np.where(flags == 1)[0]if len(indices) == 0:return result# 对每个睽违点进行插值for i in indices:left = i - 1right = i + 1# 边界处理if left < 0:left = 0if right >= len(series):right = len(series) - 1if self.method == 'interpolate':# 线性插值ratio = (right - i) / (right - left) if right != left else 0.5result.iloc[i] = series.iloc[left] * ratio + series.iloc[right] * (1 - ratio)elif self.method == 'forward':result.iloc[i] = series.iloc[left]return result

第15行，复制原序列，避免修改原始数据。

第18行，找出所有睽违点的索引位置。

第25-32行，边界处理。第一个点和最后一个点没有左右邻居，直接取边界值。

第34-37行，线性插值。ratio计算插值比例，根据左右邻居的值加权平均。

第38-39行，前向填充，直接用左邻居的值，适合单调递增的施工进度数据。

运行与测试

先准备测试数据。模拟一个30天的施工进度，其中第10天停工，第11天复工，进度跳跃：

import pandas as pd
import numpy as np# 生成模拟数据
np.random.seed(42)
days = np.arange(1, 31)
progress = np.linspace(10, 90, 30) + np.random.normal(0, 2, 30)# 制造睽违：第10天停工，进度不增长；第11天复工，进度跳跃
progress[9] = progress[8]      # 第10天停工
progress[10] = progress[8] + 15  # 第11天跳跃式复工df = pd.DataFrame({'day': days, 'progress': progress})
df.to_csv('data/progress.csv', index=False)
print("测试数据已生成")

然后写main.py：

import pandas as pd
from src.detector import KuivieDetector
from src.processor import KuivieProcessordef main():# 读取数据df = pd.read_csv('data/progress.csv')series = df['progress']# 检测睽违detector = KuivieDetector(window=7, threshold=2.0)flags = detector.detect(series)# 打印检测结果print("睽违点位置:", np.where(flags == 1)[0] + 1)  # +1转为1-based# 处理数据processor = KuivieProcessor(method='interpolate')processed = processor.process(series, flags)# 对比前后print("\n原始数据（部分）:")print(series.iloc[7:13].to_string())print("\n处理后数据（部分）:")print(processed.iloc[7:13].to_string())# 可视化import matplotlib.pyplot as pltplt.figure(figsize=(12, 6))plt.plot(range(len(series)), series, 'b-', label='原始数据', alpha=0.6)plt.plot(range(len(series)), processed, 'r-', label='处理后', linewidth=2)plt.scatter(np.where(flags == 1)[0], series[flags == 1], color='red', marker='x', s=100, label='睽违点')plt.xlabel('天数')plt.ylabel('进度(%)')plt.title('睽违数据检测与处理效果')plt.legend()plt.grid(True, alpha=0.3)plt.tight_layout()plt.savefig('result.png', dpi=150)plt.show()if __name__ == '__main__':main()

运行后，你会看到第10天和第11天被标记为睽违点。处理后，这两个点被插值平滑，曲线变得连续。保存的result.png能直观看到效果。

我在GitHub上找到一个类似开源仓库，https://github.com/pandas-dev/pandas/issues/28391，里面讨论了类似的时间序列异常检测问题，他们的方案和我们思路一致，都是用滚动统计量加阈值判断。

优化扩展

基础版本能用，但有几个地方可以优化：

窗口自适应。固定窗口7天可能不适合所有场景。可以根据数据波动程度动态调整窗口大小。波动大的数据用大窗口，波动小的用小窗口。

def adaptive_window(series, min_window=3, max_window=15):"""根据数据波动程度自适应窗口"""std = series.rolling(window=3, min_periods=1).std()avg_std = std.mean()if avg_std < 1:return min_windowelif avg_std > 5:return max_windowelse:return int(avg_std)

多阈值判断。单一阈值容易误判。可以设置双阈值：z_score > 3.0才判定为强睽违，2.0-3.0之间为弱睽违，弱睽违只做标记不处理。

保存处理日志。每次处理记录哪些点被标记、处理方式、前后值对比，方便追溯和调试。

import logging
logging.basicConfig(filename='process.log', level=logging.INFO)
logging.info(f"点{i}被标记为睽违，原值{series.iloc[i]:.2f}，处理后{processed.iloc[i]:.2f}")

批量处理。如果有多条时间序列，可以用列表推导式并行处理，提高效率。