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.0self.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})批量处理。如果有多条时间序列，可以用列表推导式并行处理，提高效率。
小结
这个保姆级教程带你看完了睽违数据处理的完整流程：从检测逻辑、数据处理、测试验证到优化扩展。核心思路就是滚动统计加阈值判断，简单但有效。
建筑施工中的进度数据、设备运行数据、天气监测数据，很多场景都会遇到睽违问题。这套方法可以直接复用，改改参数就能适配不同数据。
代码都在上面，复制就能跑。遇到问题，先检查窗口大小和阈值是否合适，这两个参数对结果影响最大。
你更常用插值还是前向填充来处理睽违点？评论区交流 3个实战案例教你用Python睽违数据：保姆级教程
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.0self.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})批量处理。如果有多条时间序列，可以用列表推导式并行处理，提高效率。
小结
这个保姆级教程带你看完了睽违数据处理的完整流程：从检测逻辑、数据处理、测试验证到优化扩展。核心思路就是滚动统计加阈值判断，简单但有效。
建筑施工中的进度数据、设备运行数据、天气监测数据，很多场景都会遇到睽违问题。这套方法可以直接复用，改改参数就能适配不同数据。
代码都在上面，复制就能跑。遇到问题，先检查窗口大小和阈值是否合适，这两个参数对结果影响最大。
你更常用插值还是前向填充来处理睽违点？评论区交流