本文目录导读：
我来为您编写一个监控数据标签质量的Python脚本,包括完整性、准确性、一致性和时效性检查：
基础标签质量监控脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据标签质量监控脚本
功能：监控标签的完整性、准确性、一致性和时效性
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import json
import logging
from typing import Dict, List, Optional, Tuple
import warnings
warnings.filterwarnings('ignore')
class LabelQualityMonitor:
"""标签质量监控器"""
def __init__(self, config_path: Optional[str] = None):
"""
初始化监控器
Args:
config_path: 配置文件路径
"""
self.setup_logging()
self.config = self.load_config(config_path) if config_path else self.default_config()
self.quality_report = {}
def setup_logging(self):
"""设置日志"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('label_quality_monitor.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def default_config(self) -> Dict:
"""默认配置"""
return {
'completeness': {
'threshold': 0.95, # 完整性阈值
'required_fields': ['label_id', 'label_name', 'create_time']
},
'accuracy': {
'threshold': 0.90, # 准确性阈值
'valid_value_ranges': {
'age': (0, 150),
'score': (0, 100),
'probability': (0, 1)
}
},
'consistency': {
'threshold': 0.85, # 一致性阈值
'check_duplicates': True,
'check_format': True
},
'timeliness': {
'max_delay_days': 7, # 最大延迟天数
'update_frequency': 'daily'
}
}
def load_config(self, config_path: str) -> Dict:
"""加载配置文件"""
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
self.logger.info(f"配置文件加载成功: {config_path}")
return config
except Exception as e:
self.logger.error(f"配置文件加载失败: {e}")
return self.default_config()
def check_completeness(self, df: pd.DataFrame) -> Dict:
"""
检查标签完整性
Args:
df: 标签数据DataFrame
Returns:
完整性检查结果
"""
results = {
'status': 'pass',
'details': {},
'issues': []
}
# 1. 检查必填字段
required_fields = self.config['completeness']['required_fields']
for field in required_fields:
if field not in df.columns:
results['status'] = 'fail'
results['issues'].append(f"缺少必填字段: {field}")
continue
null_count = df[field].isnull().sum()
total_count = len(df)
null_rate = null_count / total_count if total_count > 0 else 1
results['details'][field] = {
'total': total_count,
'null_count': int(null_count),
'null_rate': float(f"{null_rate:.4f}"),
'status': 'pass' if null_rate <= (1 - self.config['completeness']['threshold']) else 'warn'
}
if null_rate > (1 - self.config['completeness']['threshold']):
results['issues'].append(f"字段 '{field}' 缺失率过高: {null_rate:.2%}")
# 2. 检查整体完整性
completeness_score = 1 - df.isnull().mean().mean()
results['completeness_score'] = float(f"{completeness_score:.4f}")
if completeness_score < self.config['completeness']['threshold']:
results['status'] = 'warn'
results['issues'].append(f"整体完整性偏低: {completeness_score:.2%}")
return results
def check_accuracy(self, df: pd.DataFrame) -> Dict:
"""
检查标签准确性
Args:
df: 标签数据
Returns:
准确性检查结果
"""
results = {
'status': 'pass',
'details': {},
'issues': []
}
# 检查数值范围
value_ranges = self.config['accuracy']['valid_value_ranges']
for field, (min_val, max_val) in value_ranges.items():
if field not in df.columns:
continue
numeric_col = pd.to_numeric(df[field], errors='coerce')
out_of_range = (numeric_col < min_val) | (numeric_col > max_val)
out_of_range_count = out_of_range.sum()
results['details'][field] = {
'min_value': float(numeric_col.min()),
'max_value': float(numeric_col.max()),
'out_of_range_count': int(out_of_range_count),
'status': 'pass' if out_of_range_count == 0 else 'warn'
}
if out_of_range_count > 0:
results['issues'].append(f"字段 '{field}' 存在 {out_of_range_count} 个越界值")
results['status'] = 'warn'
# 检查数据类型
for col in df.columns:
if df[col].dtype == 'object':
# 检查字符串字段中的异常字符
invalid_records = df[col].str.contains(r'[^a-zA-Z0-9\u4e00-\u9fff\s\.\-_]', na=False).sum()
if invalid_records > 0:
results['details'][col] = results['details'].get(col, {})
results['details'][col]['invalid_characters'] = int(invalid_records)
results['issues'].append(f"字段 '{col}' 存在 {invalid_records} 个异常字符")
# 计算准确性得分
accuracy_score = 1 - len(results['issues']) / max(len(df.columns), 1)
results['accuracy_score'] = float(f"{accuracy_score:.4f}")
if accuracy_score < self.config['accuracy']['threshold']:
results['status'] = 'warn'
return results
def check_consistency(self, df: pd.DataFrame) -> Dict:
"""
检查标签一致性
Args:
df: 标签数据
Returns:
一致性检查结果
"""
results = {
'status': 'pass',
'details': {},
'issues': []
}
# 1. 检查重复标签
if self.config['consistency']['check_duplicates']:
duplicate_count = df.duplicated().sum()
results['details']['duplicates'] = {
'count': int(duplicate_count),
'status': 'pass' if duplicate_count == 0 else 'warn'
}
if duplicate_count > 0:
results['issues'].append(f"发现 {duplicate_count} 条重复数据")
results['status'] = 'warn'
# 2. 检查字段格式一致性
if self.config['consistency']['check_format']:
for col in df.select_dtypes(include=['object']).columns:
if 'date' in col.lower() or 'time' in col.lower():
# 检查日期格式
date_formats = df[col].dropna().apply(
lambda x: len(str(x)) if pd.notna(x) else 0
).value_counts()
if len(date_formats) > 1:
results['details'][col] = {
'format_inconsistency': True,
'unique_formats': date_formats.to_dict(),
'status': 'warn'
}
results['issues'].append(f"字段 '{col}' 存在日期格式不一致")
results['status'] = 'warn'
# 3. 检查值域一致性
for col in df.select_dtypes(include=['category', 'object']).columns:
if df[col].nunique() < 20: # 枚举值字段
value_counts = df[col].value_counts()
expected_values = value_counts.index.tolist()
# 检查是否有新出现的值
unexpected_values = set(df[col].unique()) - set(expected_values)
if unexpected_values:
results['details'][col] = results['details'].get(col, {})
results['details'][col]['unexpected_values'] = list(unexpected_values)
results['issues'].append(f"字段 '{col}' 出现未预期的值: {unexpected_values}")
results['status'] = 'warn'
# 计算一致性得分
consistency_score = max(0, 1 - len(results['issues']) * 0.1)
results['consistency_score'] = float(f"{consistency_score:.4f}")
if consistency_score < self.config['consistency']['threshold']:
results['status'] = 'warn'
return results
def check_timeliness(self, df: pd.DataFrame) -> Dict:
"""
检查标签时效性
Args:
df: 标签数据
Returns:
时效性检查结果
"""
results = {
'status': 'pass',
'details': {},
'issues': []
}
current_time = datetime.now()
# 查找时间字段
time_fields = [col for col in df.columns if any(
keyword in col.lower() for keyword in ['time', 'date', 'create', 'update', 'timestamp']
)]
if not time_fields:
results['issues'].append("未找到时间字段，无法检查时效性")
results['status'] = 'warn'
return results
for time_field in time_fields[:3]: # 最多检查3个时间字段
if time_field not in df.columns:
continue
try:
# 转换时间字段
time_col = pd.to_datetime(df[time_field], errors='coerce')
# 计算时间延迟
time_delay = (current_time - time_col.max()).days
max_delay = self.config['timeliness']['max_delay_days']
results['details'][time_field] = {
'latest_update': time_col.max().strftime('%Y-%m-%d %H:%M:%S'),
'time_delay_days': time_delay,
'max_acceptable_delay': max_delay,
'status': 'pass' if time_delay <= max_delay else 'warn'
}
if time_delay > max_delay:
results['issues'].append(
f"字段 '{time_field}' 更新延迟 {time_delay} 天，超过最大容忍值 {max_delay} 天"
)
results['status'] = 'warn'
# 检查数据分布时效性
date_counts = time_col.dt.date.value_counts().sort_index()
recent_days = date_counts.tail(7)
results['details'][time_field]['recent_7days_updates'] = int(recent_days.sum())
except Exception as e:
self.logger.warning(f"时间字段 '{time_field}' 处理失败: {e}")
return results
def generate_quality_metrics(self, df: pd.DataFrame) -> Dict:
"""
生成综合质量指标
Args:
df: 标签数据
Returns:
综合质量指标
"""
metrics = {
'basic_info': {
'total_records': len(df),
'total_fields': len(df.columns),
'memory_usage': f"{df.memory_usage(deep=True).sum() / 1024**2:.2f} MB",
'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
'completeness': self.check_completeness(df),
'accuracy': self.check_accuracy(df),
'consistency': self.check_consistency(df),
'timeliness': self.check_timeliness(df)
}
# 计算综合评分
scores = []
for key in ['completeness', 'accuracy', 'consistency', 'timeliness']:
score_key = f'{key}_score'
if score_key in metrics[key]:
scores.append(metrics[key][score_key])
metrics['overall_score'] = float(f"{np.mean(scores):.4f}") if scores else 0
# 添加建议
metrics['recommendations'] = self.generate_recommendations(metrics)
return metrics
def generate_recommendations(self, metrics: Dict) -> List[str]:
"""
生成改进建议
Args:
metrics: 质量指标
Returns:
建议列表
"""
recommendations = []
# 基于各维度检查结果生成建议
for dimension in ['completeness', 'accuracy', 'consistency', 'timeliness']:
dimension_result = metrics.get(dimension, {})
issues = dimension_result.get('issues', [])
if issues:
recommendations.append(f"{dimension.capitalize()}改进建议:")
for issue in issues[:5]: # 最多5条建议
recommendations.append(f" - {issue}")
if not recommendations:
recommendations.append("当前标签质量良好，无需特别改进")
return recommendations
def run_quality_check(self, df: pd.DataFrame, output_path: Optional[str] = None) -> Dict:
"""
运行完整质量检查
Args:
df: 标签数据
output_path: 输出报告路径
Returns:
质量检查结果
"""
self.logger.info("开始标签质量检查...")
# 运行检查
quality_metrics = self.generate_quality_metrics(df)
# 输出报告
if output_path:
self.export_report(quality_metrics, output_path)
# 记录结果
self.quality_report = quality_metrics
self.logger.info(f"标签质量检查完成，综合评分: {quality_metrics['overall_score']:.4f}")
return quality_metrics
def export_report(self, metrics: Dict, output_path: str):
"""
导出质量报告
Args:
metrics: 质量指标
output_path: 输出路径
"""
try:
# 转换为可序列化的格式
report = json.dumps(metrics, ensure_ascii=False, indent=2, default=str)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(report)
self.logger.info(f"质量报告已导出: {output_path}")
except Exception as e:
self.logger.error(f"导出报告失败: {e}")
# 使用示例
def main():
"""主函数示例"""
# 1. 创建测试数据
np.random.seed(42)
test_data = {
'label_id': range(1, 1001),
'label_name': [f'label_{i}' for i in range(1, 1001)],
'create_time': pd.date_range(start='2024-01-01', periods=1000, freq='D'),
'age': np.random.randint(0, 100, 1000).tolist() + [200], # 包含异常值
'score': np.random.uniform(0, 100, 1000),
'category': np.random.choice(['A', 'B', 'C', 'D'], 1000)
}
# 添加缺失值
test_data['label_name'][:50] = np.nan
test_data['age'][-10:] = np.nan
df = pd.DataFrame(test_data)
# 2. 初始化监控器
monitor = LabelQualityMonitor()
# 3. 运行质量检查
quality_report = monitor.run_quality_check(
df,
output_path='label_quality_report.json'
)
# 4. 打印结果摘要
print("\n" + "="*50)
print("标签质量监控报告")
print("="*50)
print(f"记录总数: {quality_report['basic_info']['total_records']}")
print(f"综合评分: {quality_report['overall_score']:.4f}")
print("\n问题列表:")
for dimension in ['completeness', 'accuracy', 'consistency', 'timeliness']:
issues = quality_report[dimension].get('issues', [])
if issues:
print(f"\n{dimension.capitalize()}:")
for issue in issues:
print(f" ⚠️ {issue}")
print("\n改进建议:")
for rec in quality_report['recommendations']:
print(f" 💡 {rec}")
if __name__ == "__main__":
main()
高级功能扩展
class AdvancedLabelQualityMonitor(LabelQualityMonitor):
"""高级标签质量监控器"""
def __init__(self, config_path=None):
super().__init__(config_path)
self.alert_threshold = 0.8
self.ml_validator = None
def check_distribution(self, df: pd.DataFrame) -> Dict:
"""
检查数据分布
Args:
df: 标签数据
"""
results = {}
for col in df.select_dtypes(include=[np.number]).columns:
stats = {
'mean': float(df[col].mean()),
'std': float(df[col].std()),
'skew': float(df[col].skew()),
'kurtosis': float(df[col].kurtosis()),
'distribution_type': 'normal' if abs(df[col].skew()) < 1 else 'skewed'
}
results[col] = stats
return results
def detect_anomalies(self, df: pd.DataFrame) -> Dict:
"""
异常检测（使用IQR方法）
Args:
df: 标签数据
"""
anomalies = {}
for col in df.select_dtypes(include=[np.number]).columns:
Q1 = df[col].quantile(0.25)
Q3 = df[col].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
anomaly_mask = (df[col] < lower_bound) | (df[col] > upper_bound)
anomalies[col] = {
'count': int(anomaly_mask.sum()),
'indices': df.index[anomaly_mask].tolist()[:10], # 仅保存前10个
'bounds': {'lower': float(lower_bound), 'upper': float(upper_bound)}
}
return {'anomalies': anomalies}
定时监控脚本
# schedule_monitor.py
"""
定时执行标签质量监控
"""
import schedule
import time
from datetime import datetime
import pandas as pd
from label_quality_monitor import LabelQualityMonitor
def scheduled_quality_check():
"""定时执行质量检查"""
print(f"\n[{datetime.now()}] 开始定时标签质量检查...")
try:
# 加载数据（实际应用中从数据库或API获取）
df = pd.read_parquet('latest_labels.parquet')
# 执行检查
monitor = LabelQualityMonitor()
report = monitor.run_quality_check(df,
output_path=f'reports/quality_report_{datetime.now().strftime("%Y%m%d")}.json')
# 检查是否需要告警
if report['overall_score'] < 0.8:
send_alert(report)
except Exception as e:
print(f"检查失败: {e}")
send_error_alert(str(e))
def send_alert(report):
"""发送告警（示例）"""
print(f"⚠️ 标签质量评分过低: {report['overall_score']}")
# 实际应用中发送邮件、短信或企业微信通知
def send_error_alert(error_msg):
"""发送错误告警"""
print(f"❌ 监控脚本执行出错: {error_msg}")
# 设置定时任务
schedule.every().day.at("03:00").do(scheduled_quality_check) # 每天凌晨3点
schedule.every().monday.at("09:00").do(scheduled_quality_check) # 每周一上午9点
if __name__ == "__main__":
print("标签质量监控定时任务启动...")
# 立即执行一次
scheduled_quality_check()
# 持续运行
while True:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次
使用建议
- 配置管理：使用JSON或YAML配置文件管理阈值和规则
- 增量监控：仅检查新增或变更的数据，提高效率
- 可视化报告：集成Grafana或自建Dashboard展示质量趋势
- 告警通知：接入企业微信、钉钉、邮件等通知渠道
- 历史对比：保存历史质量报告，分析质量变化趋势
- 自动化修复：对可自动修复的问题实施自动化脚本
这个脚本提供了完整的标签质量监控框架,您可以根据实际业务需求进行调整和扩展。
