数据漂移：特征分布随时间发生变化
您将通过逐步改变输入特征来模拟漂移，计算 Population Stability Index（PSI）和 KL 散度，并设置基于阈值的警报。
数据漂移：特征分布随时间发生变化 是 CoddyKit 上的免费 Machine Learning Academy 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Machine Learning Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Machine Learning Academy 课程共包含 4 节课。
什么是数据漂移
数据漂移（也称为协变量偏移）是指部署后输入特征的统计分布相对于训练期间所见的分布发生变化。例如，在 2022 年交易模式上训练的欺诈检测模型，到 2024 年可能会遇到非常不同的交易金额和商户类别。模型学到的决策边界不再匹配新的数据分布，从而导致只有通过监控才能发现的隐性性能下降。
模拟漂移：逐渐变化的特征
为了研究漂移，我们可以模拟特征均值随时间逐渐变化的情况。在生产环境中，这可能代表用户行为的季节性变化、影响购买力的经济变化，或不断演变的欺诈模式。绘制每周的特征分布，可以发现变化何时具有统计显著性，并应触发重新训练提醒。
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
np.random.seed(42)
# Training distribution: income ~ Normal(50000, 10000)
train_income = np.random.normal(50000, 10000, 5000)
# Production weeks 1-12: mean gradually shifts from 50k to 65k
prod_weeks = []
for week in range(1, 13):
shifted_mean = 50000 + week * 1250 # +1250 per week
week_data = np.random.normal(shifted_mean, 10000, 500)
prod_weeks.append({'week': week, 'income': week_data})
print('Training mean:', train_income.mean().round(0))
for pw in [prod_weeks[0], prod_weeks[5], prod_weeks[-1]]:
print(f'Week {pw["week"]} mean: {pw["income"].mean().round(0)}')总体稳定性指数（PSI）
总体稳定性指数（PSI） 是金融行业中最广泛使用的特征漂移检测指标。它通过分箱并测量各区间比例的差异来比较两个分布。PSI 低于 0.1 表示没有显著变化；0.1 至 0.2 表示发生了中等程度的变化，需要进一步调查；高于 0.2 表示发生了严重漂移，需要立即采取重新训练措施。
import numpy as np
def compute_psi(reference, current, buckets=10):
breakpoints = np.percentile(reference, np.linspace(0, 100, buckets + 1))
breakpoints[0], breakpoints[-1] = -np.inf, np.inf
ref_pcts = np.histogram(reference, bins=breakpoints)[0] / len(reference)
cur_pcts = np.histogram(current, bins=breakpoints)[0] / len(current)
# Avoid division by zero
ref_pcts = np.where(ref_pcts == 0, 1e-6, ref_pcts)
cur_pcts = np.where(cur_pcts == 0, 1e-6, cur_pcts)
psi = np.sum((cur_pcts - ref_pcts) * np.log(cur_pcts / ref_pcts))
return round(psi, 4)
train_income = np.random.normal(50000, 10000, 5000)
week6_income = np.random.normal(57500, 10000, 500)
week12_income = np.random.normal(65000, 10000, 500)
print('PSI week 6:', compute_psi(train_income, week6_income))
print('PSI week 12:', compute_psi(train_income, week12_income))使用 KL 散度衡量漂移
KL 散度（Kullback-Leibler 散度）用于衡量一个概率分布与参考分布之间的差异程度。它始终为非负值，且只有在两个分布完全相同时才为零。与 PSI 不同，KL 散度是不对称的：D(P||Q) ≠ D(Q||P)。在漂移检测中，请计算每个特征的训练直方图与生产直方图之间的 KL 散度，并在其超过经过校准的阈值时发出提醒。
import numpy as np
from scipy.stats import entropy
from scipy.special import kl_div
def compute_kl_divergence(reference, current, buckets=20):
bins = np.linspace(
min(reference.min(), current.min()),
max(reference.max(), current.max()),
buckets + 1
)
ref_hist, _ = np.histogram(reference, bins=bins, density=True)
cur_hist, _ = np.histogram(current, bins=bins, density=True)
# Add small epsilon to avoid log(0)
ref_hist = ref_hist + 1e-10
cur_hist = cur_hist + 1e-10
kl = entropy(cur_hist, ref_hist) # KL(current || reference)
return round(kl, 4)
train = np.random.normal(0, 1, 5000)
current = np.random.normal(0.5, 1.2, 1000)
print('KL divergence:', compute_kl_divergence(train, current))统计检验：Kolmogorov-Smirnov 检验
Kolmogorov-Smirnov（KS）检验 是一种非参数统计检验，用于衡量两个累积分布函数之间的最大差异。请使用 scipy.stats.ks_2samp 比较参考样本和生产样本。该检验会返回一个统计量（数值越大表示差异越大）和一个 p 值（数值越小表示统计显著性越强）。p 值低于 0.05 表示分布差异具有统计显著性。
import numpy as np
from scipy.stats import ks_2samp
np.random.seed(42)
train = np.random.normal(50000, 10000, 5000)
week_results = []
for week in range(1, 13):
prod_sample = np.random.normal(50000 + week * 1250, 10000, 500)
stat, p_value = ks_2samp(train, prod_sample)
week_results.append((week, round(stat, 4), round(p_value, 4)))
alert = 'DRIFT ALERT' if p_value < 0.05 else 'OK'
print(f'Week {week:2d}: KS={stat:.4f} p={p_value:.4f} {alert}')监控多个特征
生产数据集包含许多特征，任何一个特征都可能发生漂移。请在每个监控周期计算每一列的漂移分数，从而监控所有输入特征。使用热力图可视化各特征随时间变化的 PSI 或 KS 统计量。漂移分数持续较高的特征，是模型性能下降时最需要怀疑的对象，应在根因分析期间优先调查。
import numpy as np
import pandas as pd
from scipy.stats import ks_2samp
# Simulate reference and production with drift in some features
np.random.seed(42)
n_features = 5
reference = pd.DataFrame(
np.random.normal(0, 1, (5000, n_features)),
columns=[f'feature_{i}' for i in range(n_features)]
)
# Introduce drift in features 1 and 3
production = pd.DataFrame(
np.random.normal([0, 1.5, 0, 2.0, 0], 1, (1000, n_features)),
columns=reference.columns
)
drift_report = {}
for col in reference.columns:
stat, p_val = ks_2samp(reference[col], production[col])
drift_report[col] = {'ks_stat': round(stat, 4), 'p_value': round(p_val, 4),
'drift': p_val < 0.05}
for feature, result in drift_report.items():
status = '*** DRIFT ***' if result['drift'] else 'stable'
print(f'{feature}: KS={result["ks_stat"]} {status}')设置提醒阈值
提醒阈值应根据历史数据进行校准，而不是任意选择。一种常见方法是：在训练集与多个留出的验证划分之间计算漂移指标，从而建立无漂移条件下的分数基线分布。将提醒阈值设置为该基线的第 99 个百分位数——任何高于此水平的生产分数，都不太可能来自同一分布，表明确实发生了漂移。
import numpy as np
from scipy.stats import ks_2samp
train = np.random.normal(0, 1, 5000)
# Calibrate: compute KS statistic between train and 100 random validation splits
calibration_scores = []
for _ in range(100):
val_sample = np.random.normal(0, 1, 500) # same distribution
stat, _ = ks_2samp(train, val_sample)
calibration_scores.append(stat)
threshold_99 = np.percentile(calibration_scores, 99)
print(f'No-drift KS scores -- mean: {np.mean(calibration_scores):.4f}')
print(f'99th percentile threshold: {threshold_99:.4f}')
# Any production score above this triggers an alert
prod_sample_drifted = np.random.normal(0.5, 1, 500) # shifted
stat, _ = ks_2samp(train, prod_sample_drifted)
print(f'Production KS: {stat:.4f} -- ALERT: {stat > threshold_99}')分类特征中的漂移
对于分类特征，漂移意味着各类别的频率分布发生变化。例如，随着产品向全球扩展，国家字段可能会从训练数据中的 80% USA 变为生产数据中的 60% USA。请使用卡方检验或Jensen-Shannon 散度比较分类分布。生产环境中出现训练数据里不存在的新类别是一个特殊情况——如果标签编码器在拟合时没有使用 handle_unknown='ignore'，就会产生 KeyError。
import numpy as np
from scipy.stats import chi2_contingency
# Training distribution
train_countries = np.random.choice(
['US', 'GB', 'DE', 'FR'], p=[0.7, 0.15, 0.1, 0.05], size=5000
)
# Production: more EU traffic
prod_countries = np.random.choice(
['US', 'GB', 'DE', 'FR'], p=[0.5, 0.2, 0.2, 0.1], size=1000
)
categories = ['US', 'GB', 'DE', 'FR']
train_counts = [np.sum(train_countries == c) for c in categories]
prod_counts = [np.sum(prod_countries == c) for c in categories]
contingency = np.array([train_counts, prod_counts])
chi2, p_value, dof, _ = chi2_contingency(contingency)
print(f'Chi-squared: {chi2:.2f}, p-value: {p_value:.4f}')
print('Categorical drift detected:', p_value < 0.05)随时间记录漂移指标
只有持续记录指标并可视化趋势，漂移监控才有用。请将每周漂移分数存储在时序数据库（InfluxDB、PostgreSQL，甚至 CSV 文件）中，并使用滚动窗口绘制这些分数。漂移分数突然升高，可以精确定位数据 pipeline 发生变化或外部事件改变用户行为的那一周，从而帮助团队更快地进行根因分析。
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
records = []
base_date = datetime(2024, 1, 1)
for week in range(12):
prod_sample = np.random.normal(week * 0.1, 1, 500) # gradual drift
from scipy.stats import ks_2samp
train_sample = np.random.normal(0, 1, 5000)
stat, p = ks_2samp(train_sample, prod_sample)
records.append({
'date': base_date + timedelta(weeks=week),
'ks_stat': round(stat, 4),
'p_value': round(p, 4),
'alert': p < 0.05
})
df = pd.DataFrame(records)
print(df.to_string(index=False))
# In production: df.to_sql('drift_log', engine, if_exists='append')检测到漂移后该怎么办
检测到漂移后，可以根据严重程度采取多种应对策略。轻微漂移：提高监控频率并调查原因。中等漂移：使用最近的数据窗口触发重新训练。严重漂移：考虑旧的特征工程是否仍适用于新的分布，必要时重新设计特征。请始终记录漂移事件及其业务背景（例如营销活动或平台变更）。
def respond_to_drift(psi_score, ks_p_value):
if psi_score < 0.1 and ks_p_value > 0.05:
return 'No action needed. All features stable.'
elif psi_score < 0.2 and ks_p_value > 0.01:
return ('Moderate drift detected. '
'Increase monitoring to daily. '
'Schedule retraining for next cycle.')
else:
return ('SEVERE DRIFT. '
'Trigger emergency retraining now. '
'Consider feature engineering review. '
'Alert data engineering team.')
# Example
print(respond_to_drift(0.05, 0.3)) # stable
print(respond_to_drift(0.15, 0.04)) # moderate
print(respond_to_drift(0.35, 0.001)) # severe可视化随时间变化的特征漂移
将漂移可视化为时间序列，有助于利益相关者了解其严重程度和发生时间。请按每周滚动的方式，绘制每个受监控特征的 PSI 或 KS 统计量。当某个特征曲线越过提醒阈值时，请在图表上标注发生日期以及导致变化的业务事件类型（例如营销活动、价格变更或季节性模式）。这样可以将原始统计数据转化为可执行的商业洞察。
import matplotlib.pyplot as plt
import numpy as np
weeks = list(range(1, 13))
psi_feature_A = [0.01, 0.02, 0.03, 0.05, 0.07, 0.11, 0.18, 0.22, 0.25, 0.24, 0.23, 0.26]
psi_feature_B = [0.01, 0.01, 0.02, 0.01, 0.02, 0.03, 0.03, 0.04, 0.05, 0.06, 0.07, 0.08]
plt.figure(figsize=(10, 4))
plt.plot(weeks, psi_feature_A, marker='o', label='amount_usd (drifting)')
plt.plot(weeks, psi_feature_B, marker='s', label='merchant_category (stable)')
plt.axhline(0.1, color='orange', linestyle='--', label='Mild drift threshold')
plt.axhline(0.25, color='red', linestyle='--', label='Severe drift threshold')
plt.annotate('Pricing change', xy=(6, 0.11), xytext=(6, 0.16), arrowprops=dict(arrowstyle='->'))
plt.xlabel('Week')
plt.ylabel('PSI')
plt.legend()
plt.title('PSI Trend — Feature-Level Drift Monitoring')
plt.tight_layout()
plt.savefig('psi_trend.png', dpi=150)快速检查
请测试您对本课中 Python 机器学习概念的理解。
课程回顾
在本课中，您学到了：数据漂移是训练时与生产时输入特征分布发生的变化，会导致模型无声地退化、PSI 和 KS 检验可以提供量化的漂移分数，并使用行业标准的提醒阈值（PSI > 0.2 = 严重漂移），以及漂移监控应覆盖所有输入特征，并随时间记录分数，以支持趋势分析和根因识别。接下来，我们将区分概念漂移——即输入与目标标签之间的关系本身发生变化，这是一种更隐蔽也更危险的现象。
用 AI 导师学习 Python — 免费
在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。
- 课程
- 30
- 课程
- 120
常见问题解答
「数据漂移：特征分布随时间发生变化」课时是免费的吗？
是的 — 「数据漂移：特征分布随时间发生变化」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Machine Learning Academy 课程的其余内容，请升级到 CoddyKit PRO。 Machine Learning Academy 课程共包含 4 节课。
「数据漂移：特征分布随时间发生变化」这节课中我会学到什么？
您将通过逐步改变输入特征来模拟漂移，计算 Population Stability Index（PSI）和 KL 散度，并设置基于阈值的警报。 你通过在浏览器中直接运行的动手代码来练习 Machine Learning Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 Machine Learning Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 Machine Learning Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。
「数据漂移：特征分布随时间发生变化」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 Machine Learning Academy 课中编写并运行代码吗？
能。每节 Machine Learning Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 数据漂移：特征分布随时间发生变化
- 概念漂移：X 与 Y 之间的关系发生变化
- 监控预测分布与置信度分数
- 使用 Evidently AI 构建数据漂移告警流水线