本文目录导读：
基于统计检验的Python脚本
import numpy as np
import pandas as pd
from scipy import stats
from scipy.spatial.distance import jensenshannon
import warnings
warnings.filterwarnings('ignore')
class DistributionShiftMonitor:
def __init__(self, reference_data, threshold=0.05):
"""
初始化监控器
reference_data: 参考数据集（训练数据分布）
threshold: 检测阈值
"""
self.reference_data = reference_data
self.threshold = threshold
self.reference_stats = self._calculate_statistics(reference_data)
def _calculate_statistics(self, data):
"""计算数据的统计特征"""
stats_dict = {
'mean': np.mean(data),
'std': np.std(data),
'median': np.median(data),
'percentiles': np.percentile(data, [25, 50, 75]),
'skewness': stats.skew(data),
'kurtosis': stats.kurtosis(data)
}
return stats_dict
def ks_test(self, new_data):
"""Kolmogorov-Smirnov检验"""
statistic, p_value = stats.ks_2samp(self.reference_data, new_data)
return {
'test': 'KS Test',
'statistic': statistic,
'p_value': p_value,
'drift_detected': p_value < self.threshold
}
def js_distance(self, new_data, bins=50):
"""Jensen-Shannon散度"""
# 计算直方图
hist_ref, edges = np.histogram(self.reference_data, bins=bins, density=True)
hist_new, _ = np.histogram(new_data, edges, density=True)
# 确保非零值
hist_ref = hist_ref + 1e-10
hist_new = hist_new + 1e-10
# 归一化
hist_ref = hist_ref / hist_ref.sum()
hist_new = hist_new / hist_new.sum()
js_div = jensenshannon(hist_ref, hist_new)
# JS距离的阈值通常设置为0.1
js_threshold = 0.1
return {
'test': 'JS Distance',
'distance': js_div,
'drift_detected': js_div > js_threshold
}
def chi_square_test(self, new_data, bins=20):
"""卡方检验"""
# 分箱
hist_ref, edges = np.histogram(self.reference_data, bins=bins)
hist_new, _ = np.histogram(new_data, edges)
# 执行卡方检验
chi2_stat, p_value = stats.chisquare(hist_new, f_exp=hist_ref)
return {
'test': 'Chi-Square Test',
'statistic': chi2_stat,
'p_value': p_value,
'drift_detected': p_value < self.threshold
}
def wasserstein_distance(self, new_data):
"""Wasserstein距离（Earth Mover's Distance）"""
w_dist = stats.wasserstein_distance(self.reference_data, new_data)
# Wasserstein距离阈值根据数据尺度调整
std_ref = self.reference_stats['std']
threshold = 0.1 * std_ref
return {
'test': 'Wasserstein Distance',
'distance': w_dist,
'normalized_distance': w_dist / std_ref,
'drift_detected': w_dist > threshold
}
def comprehensive_monitor(self, new_data):
"""综合监控报告"""
results = {
'reference_stats': self.reference_stats,
'new_data_stats': self._calculate_statistics(new_data),
'tests': {
'ks_test': self.ks_test(new_data),
'js_distance': self.js_distance(new_data),
'chi_square': self.chi_square_test(new_data),
'wasserstein': self.wasserstein_distance(new_data)
}
}
# 整体判定：多数测试检测到漂移
drift_count = sum(1 for test in results['tests'].values()
if test['drift_detected'])
results['overall_drift'] = drift_count >= 2
return results
# 使用示例
if __name__ == "__main__":
# 生成示例数据
np.random.seed(42)
reference_data = np.random.normal(0, 1, 10000)
shifted_data = np.random.normal(0.5, 1.2, 10000)
normal_data = np.random.normal(0, 1, 10000)
# 初始化监控器
monitor = DistributionShiftMonitor(reference_data, threshold=0.05)
# 监控有漂移的数据
print("="*50)
print("检测有漂移的数据:")
results_shifted = monitor.comprehensive_monitor(shifted_data)
print_results(results_shifted)
print("\n" + "="*50)
print("检测正常数据:")
results_normal = monitor.comprehensive_monitor(normal_data)
print_results(results_normal)
def print_results(results):
"""打印结果"""
print("\n参考数据统计:")
for key, value in results['reference_stats'].items():
print(f" {key}: {value:.3f}")
print("\n新数据统计:")
for key, value in results['new_data_stats'].items():
print(f" {key}: {value:.3f}")
print("\n检测结果:")
for test_name, test_results in results['tests'].items():
drift_status = "⚠️ 漂移" if test_results['drift_detected'] else "✅ 正常"
print(f" {test_results['test']}: {drift_status}")
overall = "⚠️ 检测到分布漂移" if results['overall_drift'] else "✅ 分布正常"
print(f"\n综合判定: {overall}")
实时监控脚本（带报警功能）
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import logging
import json
import smtplib
from email.mime.text import MIMEText
import time
class RealTimeDistributionMonitor:
def __init__(self, reference_distribution, config=None):
"""
实时分布监控器
reference_distribution: 参考分布数据或统计参数
config: 配置字典
"""
self.reference = reference_distribution
self.config = config or self._default_config()
self.alerts_history = []
self._setup_logging()
def _default_config(self):
return {
'window_size': 100, # 滑动窗口大小
'check_frequency': 10, # 检查频率（每N个样本）
'alert_threshold': 0.05, # 告警阈值
'email_alerts': False,
'email_config': {
'smtp_server': 'smtp.gmail.com',
'smtp_port': 587,
'sender': 'monitor@example.com',
'password': 'your_password',
'recipients': ['alert@example.com']
},
'slack_webhook': None,
'log_file': 'distribution_monitor.log'
}
def _setup_logging(self):
"""设置日志"""
logging.basicConfig(
filename=self.config['log_file'],
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def process_stream(self, data_stream, delay=0.01):
"""
处理数据流
data_stream: 数据生成器或迭代器
delay: 处理延迟（秒）
"""
buffer = []
sample_count = 0
for sample in data_stream:
buffer.append(sample)
sample_count += 1
# 每收集到一定数量样本进行检查
if sample_count % self.config['check_frequency'] == 0:
if len(buffer) >= self.config['window_size']:
window_data = buffer[-self.config['window_size']:]
self._check_and_alert(window_data)
time.sleep(delay)
# 清理旧数据
if len(buffer) > self.config['window_size'] * 2:
buffer = buffer[-self.config['window_size']:]
def _check_and_alert(self, window_data):
"""检查并触发告警"""
from scipy import stats
# 执行KS检验
statistic, p_value = stats.ks_2samp(self.reference, window_data)
alert_data = {
'timestamp': datetime.now().isoformat(),
'window_size': len(window_data),
'ks_statistic': statistic,
'p_value': p_value,
'drift_detected': p_value < self.config['alert_threshold'],
'window_stats': {
'mean': np.mean(window_data),
'std': np.std(window_data),
'percentiles': np.percentile(window_data, [25, 50, 75])
}
}
# 记录日志
self.logger.info(f"监测结果: {json.dumps(alert_data, indent=2)}")
# 如果检测到漂移，触发告警
if alert_data['drift_detected']:
self.alerts_history.append(alert_data)
self._trigger_alerts(alert_data)
def _trigger_alerts(self, alert_data):
"""触发告警机制"""
# 日志告警（已包含）
self.logger.warning(f"检测到分布漂移: {alert_data}")
# 打印告警
print(f"⚠️ 告警: 检测到数据分布漂移 (p-value: {alert_data['p_value']:.6f})")
# 邮件告警
if self.config['email_alerts']:
self._send_email_alert(alert_data)
# Slack告警（如果需要）
if self.config['slack_webhook']:
self._send_slack_alert(alert_data)
def _send_email_alert(self, alert_data):
"""发送邮件告警"""
try:
msg = MIMEText(f"""
数据分布漂移告警
时间: {alert_data['timestamp']}
KS统计量: {alert_data['ks_statistic']:.4f}
P值: {alert_data['p_value']:.6f}
窗口大小: {alert_data['window_size']}
统计信息:
- 均值: {alert_data['window_stats']['mean']:.3f}
- 标准差: {alert_data['window_stats']['std']:.3f}
- 中位数: {alert_data['window_stats']['percentiles'][1]:.3f}
""")
msg['Subject'] = '数据分布漂移告警'
msg['From'] = self.config['email_config']['sender']
msg['To'] = ', '.join(self.config['email_config']['recipients'])
server = smtplib.SMTP(
self.config['email_config']['smtp_server'],
self.config['email_config']['smtp_port']
)
server.starttls()
server.login(
self.config['email_config']['sender'],
self.config['email_config']['password']
)
server.send_message(msg)
server.quit()
except Exception as e:
self.logger.error(f"发送邮件告警失败: {e}")
def get_alerts_summary(self):
"""获取告警汇总"""
return {
'total_alerts': len(self.alerts_history),
'last_alert': self.alerts_history[-1] if self.alerts_history else None,
'alert_rate': len(self.alerts_history) / max(1, len(self.alerts_history) + 1)
}
# 使用示例
if __name__ == "__main__":
# 模拟数据流
def data_stream_generator():
np.random.seed(42)
phase = 0
while True:
if phase < 1000:
# 正常数据
yield np.random.normal(0, 1)
elif phase < 2000:
# 开始漂移
shift = (phase - 1000) / 1000
yield np.random.normal(shift, 1 + shift/2)
else:
# 严重漂移
yield np.random.normal(1, 1.5)
phase += 1
# 参考分布
reference_data = np.random.normal(0, 1, 10000)
# 初始化监控器
monitor = RealTimeDistributionMonitor(
reference_data,
config={
'window_size': 200,
'check_frequency': 50,
'alert_threshold': 0.01,
'log_file': 'real_time_monitor.log'
}
)
# 开始监控（运行时间有限）
print("开始实时监控...")
stream = data_stream_generator()
# 只处理前500个样本作为演示
for i, sample in enumerate(stream):
if i > 500: # 限制运行时间
break
# 在实际应用中，这里会调用实时监控
多维度分布监控脚本
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.covariance import EllipticEnvelope
from sklearn.preprocessing import StandardScaler
import plotly.graph_objects as go
from plotly.subplots import make_subplots
class MultiDimensionalShiftMonitor:
def __init__(self, reference_data, feature_names=None):
"""
多维数据分布监控器
reference_data: DataFrame或二维数组
feature_names: 特征名称列表
"""
if isinstance(reference_data, pd.DataFrame):
self.reference_df = reference_data
self.feature_names = reference_data.columns.tolist()
else:
self.reference_df = pd.DataFrame(reference_data, columns=feature_names)
self.feature_names = feature_names or [f'feature_{i}' for i in range(reference_data.shape[1])]
self.n_features = len(self.feature_names)
self.reference_statistics = self._calculate_multivariate_stats()
self.models = {}
def _calculate_multivariate_stats(self):
"""计算多维统计特征"""
stats = {
'mean': self.reference_df.mean(),
'covariance': self.reference_df.cov(),
'correlation': self.reference_df.corr(),
'feature_stats': {}
}
for col in self.feature_names:
stats['feature_stats'][col] = {
'mean': self.reference_df[col].mean(),
'std': self.reference_df[col].std(),
'skewness': self.reference_df[col].skew(),
'kurtosis': self.reference_df[col].kurtosis(),
'percentiles': self.reference_df[col].quantile([0.25, 0.5, 0.75]).to_dict()
}
return stats
def train_anomaly_detection_models(self):
"""训练异常检测模型"""
# 标准化数据
scaler = StandardScaler()
scaled_data = scaler.fit_transform(self.reference_df)
# Isolation Forest
isolation_forest = IsolationForest(
contamination=0.1,
random_state=42
)
isolation_forest.fit(scaled_data)
# Elliptic Envelope
elliptic_envelope = EllipticEnvelope(
contamination=0.1,
random_state=42
)
elliptic_envelope.fit(scaled_data)
self.models = {
'scaler': scaler,
'isolation_forest': isolation_forest,
'elliptic_envelope': elliptic_envelope
}
def per_feature_shift_detection(self, new_data):
"""单特征漂移检测"""
results = {}
for feature in self.feature_names:
ref_values = self.reference_df[feature]
new_values = new_data[feature] if isinstance(new_data, pd.DataFrame) else new_data[:, self.feature_names.index(feature)]
from scipy import stats
# KS检验
ks_stat, ks_p = stats.ks_2samp(ref_values, new_values)
# 均值差异检验
t_stat, t_p = stats.ttest_ind(ref_values, new_values)
# 分布差异度量
ref_hist, edges = np.histogram(ref_values, bins=20, density=True)
new_hist, _ = np.histogram(new_values, edges, density=True)
js_div = self._js_divergence(ref_hist, new_hist)
results[feature] = {
'ks_test': {
'statistic': ks_stat,
'p_value': ks_p,
'drift': ks_p < 0.05
},
'ttest': {
'statistic': t_stat,
'p_value': t_p,
'drift': t_p < 0.05
},
'js_divergence': js_div,
'statistics': {
'ref_mean': np.mean(ref_values),
'new_mean': np.mean(new_values),
'mean_change': np.mean(new_values) - np.mean(ref_values)
}
}
return results
def multivariate_shift_detection(self, new_data):
"""多维漂移检测"""
if not self.models:
self.train_anomaly_detection_models()
if isinstance(new_data, pd.DataFrame):
new_array = new_data.values
else:
new_array = new_data
# 标准化新数据
scaled_new = self.models['scaler'].transform(new_array)
# 使用Isolation Forest检测异常
if_scores = self.models['isolation_forest'].score_samples(scaled_new)
if_predictions = self.models['isolation_forest'].predict(scaled_new)
# 使用Elliptic Envelope检测异常
ee_scores = self.models['elliptic_envelope'].score_samples(scaled_new)
ee_predictions = self.models['elliptic_envelope'].predict(scaled_new)
# 计算Mahalanobis距离
mahalanobis_distances = self._mahalanobis_distance(
new_array,
self.reference_statistics['mean'].values,
self.reference_statistics['covariance'].values
)
return {
'if_anomaly_ratio': np.mean(if_predictions == -1),
'ee_anomaly_ratio': np.mean(ee_predictions == -1),
'mean_mahalanobis': np.mean(mahalanobis_distances),
'max_mahalanobis': np.max(mahalanobis_distances),
'if_scores': if_scores,
'ee_scores': ee_scores,
'mahalanobis_distances': mahalanobis_distances
}
def _js_divergence(self, p, q):
"""计算Jensen-Shannon散度"""
from scipy.spatial.distance import jensenshannon
return jensenshannon(p, q)
def _mahalanobis_distance(self, x, mean, cov):
"""计算马氏距离"""
try:
inv_cov = np.linalg.inv(cov)
distances = []
for observation in x:
diff = observation - mean
dist = np.sqrt(diff.T @ inv_cov @ diff)
distances.append(dist)
return np.array(distances)
except np.linalg.LinAlgError:
# 如果协方差矩阵不可逆，使用伪逆
inv_cov = np.linalg.pinv(cov)
distances = []
for observation in x:
diff = observation - mean
dist = np.sqrt(diff.T @ inv_cov @ diff)
distances.append(dist)
return np.array(distances)
def visualize_shift(self, new_data, save_path=None):
"""可视化分布漂移"""
per_feature_results = self.per_feature_shift_detection(new_data)
# 创建子图
n_features = len(self.feature_names)
fig = make_subplots(
rows=n_features, cols=2,
subplot_titles=[f'{feat} - 分布对比' for feat in self.feature_names] +
[f'{feat} - Q-Q图' for feat in self.feature_names]
)
for i, feature in enumerate(self.feature_names):
ref_values = self.reference_df[feature]
new_values = new_data[feature] if isinstance(new_data, pd.DataFrame) else new_data[:, i]
# 分布对比图
fig.add_trace(
go.Histogram(x=ref_values, name='参考分布', opacity=0.7),
row=i+1, col=1
)
fig.add_trace(
go.Histogram(x=new_values, name='新数据分布', opacity=0.7),
row=i+1, col=1
)
# Q-Q图
sorted_ref = np.sort(ref_values)
sorted_new = np.sort(new_values)
fig.add_trace(
go.Scatter(x=sorted_ref, y=sorted_new, mode='markers',
name=f'{feature} Q-Q'),
row=i+1, col=2
)
# 添加参考线
min_val = min(sorted_ref.min(), sorted_new.min())
max_val = max(sorted_ref.max(), sorted_new.max())
fig.add_trace(
go.Scatter(x=[min_val, max_val], y=[min_val, max_val],
mode='lines', line=dict(dash='dash', color='red'),
showlegend=False),
row=i+1, col=2
)
fig.update_layout(height=300*n_features, title_text="分布漂移可视化")
if save_path:
fig.write_html(save_path)
fig.show()
def comprehensive_report(self, new_data):
"""生成综合报告"""
per_feature = self.per_feature_shift_detection(new_data)
multivariate = self.multivariate_shift_detection(new_data)
report = {
'per_feature_analysis': per_feature,
'multivariate_analysis': multivariate,
'drift_summary': {
'features_with_drift': sum(1 for f in per_feature.values()
if any(t['drift'] for t in f.values()
if isinstance(t, dict) and 'drift' in t)),
'anomaly_ratio': multivariate['if_anomaly_ratio'],
'overall_drift_score': np.mean([
f['js_divergence'] for f in per_feature.values()
])
}
}
# 综合判定
drift_thresholds = {
'feature_drift_ratio': 0.3, # 30%的特征出现漂移
'anomaly_ratio': 0.15, # 15%的异常率
'js_divergence': 0.1 # JS散度阈值
}
report['overall_assessment'] = {
'drift_detected': (
report['drift_summary']['features_with_drift'] / self.n_features > drift_thresholds['feature_drift_ratio'] or
report['drift_summary']['anomaly_ratio'] > drift_thresholds['anomaly_ratio'] or
report['drift_summary']['overall_drift_score'] > drift_thresholds['js_divergence']
),
'severity': 'high' if report['drift_summary']['overall_drift_score'] > 0.2 else \
'medium' if report['drift_summary']['overall_drift_score'] > 0.1 else 'low',
'recommendations': self._generate_recommendations(report)
}
return report
def _generate_recommendations(self, report):
"""生成建议"""
recommendations = []
if report['drift_summary']['features_with_drift'] > self.n_features * 0.5:
recommendations.append("多个特征出现显著漂移，建议重新训练模型")
if report['multivariate_analysis']['if_anomaly_ratio'] > 0.2:
recommendations.append("数据异常比例较高，建议检查数据采集流程")
if report['drift_summary']['overall_drift_score'] > 0.15:
recommendations.append("整体分布漂移较大，建议实施模型更新策略")
if not recommendations:
recommendations.append("当前数据分布基本稳定，继续监控")
return recommendations
# 使用示例
if __name__ == "__main__":
# 生成示例数据
np.random.seed(42)
n_samples = 1000
n_features = 3
# 参考数据
reference_data = pd.DataFrame({
'feature_1': np.random.normal(0, 1, n_samples),
'feature_2': np.random.normal(10, 2, n_samples),
'feature_3': np.random.normal(-5, 3, n_samples)
})
# 漂移数据
shifted_data = pd.DataFrame({
'feature_1': np.random.normal(0.5, 1.5, n_samples),
'feature_2': np.random.normal(12, 3, n_samples),
'feature_3': np.random.normal(-3, 4, n_samples)
})
# 初始化监控器
monitor = MultiDimensionalShiftMonitor(reference_data)
# 执行检测
print("执行多维分布漂移检测...")
report = monitor.comprehensive_report(shifted_data)
# 打印结果
print("\n" + "="*60)
print("漂移检测报告")
print("="*60)
print(f"\n整体判定: {'⚠️ 检测到漂移' if report['overall_assessment']['drift_detected'] else '✅ 分布正常'}")
print(f"严重程度: {report['overall_assessment']['severity']}")
print(f"\n特征分析:")
for feature, results in report['per_feature_analysis'].items():
drift_flags = []
if results['ks_test']['drift']:
drift_flags.append('KS检验')
if isinstance(results.get('ttest'), dict) and results['ttest']['drift']:
drift_flags.append('T检验')
status = "⚠️ 漂移" if drift_flags else "✅ 正常"
print(f" {feature}: {status}")
if drift_flags:
print(f" 检测方法: {', '.join(drift_flags)}")
print(f"\n多维分析:")
print(f" 异常比例 (Isolation Forest): {report['multivariate_analysis']['if_anomaly_ratio']:.2%}")
print(f" 平均马氏距离: {report['multivariate_analysis']['mean_mahalanobis']:.2f}")
print(f"\n建议:")
for rec in report['overall_assessment']['recommendations']:
print(f" - {rec}")
# 可视化（可选）
# monitor.visualize_shift(shifted_data, save_path='shift_visualization.html')
自动化监控与告警系统
import schedule
import time
import json
from datetime import datetime
import pandas as pd
import numpy as np
from pathlib import Path
class AutomatedShiftMonitoringSystem:
def __init__(self, config_path='monitor_config.json'):
"""
自动化分布漂移监控系统
config_path: 配置文件路径
"""
self.config = self._load_config(config_path)
self.monitors = {}
self.setup_monitors()
def _load_config(self, config_path):
"""加载配置"""
default_config = {
'databases': {
'production': {
'type': 'csv',
'path': 'production_data.csv',
'check_interval': 3600 # 每小时检查一次
},
'staging': {
'type': 'csv',
'path': 'staging_data.csv',
'check_interval': 7200 # 每2小时检查一次
}
},
'reference_data': {
'path': 'reference_data.csv',
'update_interval': 86400 # 每天更新一次
},
'alerts': {
'channels': ['log', 'console'],
'email': {
'enabled': False,
'smtp_server': 'smtp.gmail.com',
'sender': 'monitor@example.com'
},
'slack': {
'enabled': False,
'webhook_url': None
}
},
'thresholds': {
'ks_test_p_value': 0.05,
'js_divergence': 0.1,
'anomaly_ratio': 0.15
},
'logging': {
'level': 'INFO',
'file': 'monitoring.log',
'max_size': 10 * 1024 * 1024 # 10MB
}
}
try:
with open(config_path, 'r') as f:
user_config = json.load(f)
# 合并配置
default_config.update(user_config)
except FileNotFoundError:
# 保存默认配置
with open(config_path, 'w') as f:
json.dump(default_config, f, indent=2)
print(f"已创建默认配置文件: {config_path}")
return default_config
def setup_monitors(self):
"""设置监控器"""
# 加载参考数据
reference_data = self._load_reference_data()
# 为每个数据库创建监控器
for db_name, db_config in self.config['databases'].items():
monitor = DistributionShiftMonitor(
reference_data,
threshold=self.config['thresholds']['ks_test_p_value']
)
self.monitors[db_name] = {
'monitor': monitor,
'config': db_config,
'last_check': None,
'alerts': []
}
def _load_reference_data(self):
"""加载参考数据"""
ref_path = self.config['reference_data']['path']
if Path(ref_path).exists():
data = pd.read_csv(ref_path)
return data.values.flatten() # 简化处理，实际应根据需求调整
else:
# 生成示例参考数据
print(f"警告: 参考数据文件 {ref_path} 不存在，使用生成的数据")
return np.random.normal(0, 1, 10000)
def check_data_sources(self):
"""检查所有数据源"""
results = {}
for db_name, monitor_info in self.monitors.items():
print(f"\n检查数据源: {db_name}")
try:
# 加载数据
new_data = self._load_data_source(db_name)
if new_data is not None:
# 执行检测
detection_result = monitor_info['monitor'].comprehensive_monitor(new_data)
# 更新监控信息
monitor_info['last_check'] = datetime.now()
# 记录结果
results[db_name] = {
'status': 'success',
'drift_detected': detection_result['overall_drift'],
'details': detection_result
}
# 如果检测到漂移，发送告警
if detection_result['overall_drift']:
self._send_alert(db_name, detection_result)
monitor_info['alerts'].append({
'timestamp': datetime.now().isoformat(),
'severity': 'high',
'details': detection_result
})
else:
results[db_name] = {
'status': 'error',
'message': '无法加载数据'
}
except Exception as e:
results[db_name] = {
'status': 'error',
'message': str(e)
}
print(f" 错误: {e}")
return results
def _load_data_source(self, db_name):
"""从数据源加载数据"""
config = self.config['databases'][db_name]
try:
if config['type'] == 'csv':
if Path(config['path']).exists():
data = pd.read_csv(config['path'])
# 假设数据在第一列或数字列
numeric_cols = data.select_dtypes(include=[np.number]).columns
if len(numeric_cols) > 0:
return data[numeric_cols[0]].values
else:
print(f" 警告: {config['path']} 中没有数值列")
return None
else:
print(f" 警告: 数据文件 {config['path']} 不存在")
# 返回模拟数据用于演示
return np.random.normal(0.2, 1.1, 1000)
elif config['type'] == 'database':
# 数据库连接代码（需要根据实际情况实现）
print(f" 数据库连接尚未实现: {config['path']}")
return None
else:
print(f" 不支持的数据源类型: {config['type']}")
return None
except Exception as e:
print(f" 加载数据失败: {e}")
return None
def _send_alert(self, source_name, detection_result):
"""发送告警"""
alert_message = f"""
数据分布漂移告警
来源: {source_name}
时间: {datetime.now().isoformat()}
详情: 检测到数据分布发生显著变化
"""
print(f"\n⚠️ 告警: {source_name} - 检测到分布漂移")
# 根据配置发送告警
for channel in self.config['alerts']['channels']:
if channel == 'log':
self._log_alert(alert_message)
elif channel == 'console':
print(alert_message)
elif channel == 'email' and self.config['alerts']['email']['enabled']:
self._send_email_alert(alert_message)
elif channel == 'slack' and self.config['alerts']['slack']['enabled']:
self._send_slack_alert(alert_message)
def _log_alert(self, message):
"""日志告警"""
import logging
logging.warning(message)
def _send_email_alert(self, message):
"""发送邮件告警（占位）"""
print("邮件告警功能待实现")
def _send_slack_alert(self, message):
"""发送Slack告警（占位）"""
print("Slack告警功能待实现")
def run_scheduled_checks(self):
"""运行定时检查"""
# 设置定时任务
for db_name, monitor_info in self.monitors.items():
interval = monitor_info['config']['check_interval']
schedule.every(interval).seconds.do(
self.check_single_source, db_name
)
print("开始定时监控任务...")
print(f"监控的数据源: {list(self.monitors.keys())}")
print("按 Ctrl+C 停止监控")
try:
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
print("\n监控已停止")
def check_single_source(self, db_name):
"""检查单个数据源"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 检查 {db_name}")
results = self.check_data_sources()
if db_name in results:
result = results[db_name]
if result['status'] == 'success':
drift_status = "漂移" if result['drift_detected'] else "正常"
print(f" {db_name}: {drift_status}")
def get_monitoring_report(self):
"""获取监控报告"""
report = {
'timestamp': datetime.now().isoformat(),
'data_sources': {},
'summary': {
'total_sources': len(self.monitors),
'sources_with_drift': 0,
'total_alerts': 0
}
}
