Python数据API数据质量怎么度量

wen python案例 2026-07-04 215

本文目录导读：

Python数据API数据质量怎么度量

基础数据质量维度与度量方法
整合的数据质量评分框架
常用专业工具
实际应用建议

在Python数据API中度量数据质量,通常从准确性、完整性、一致性、及时性和唯一性五个维度入手，下面我会结合Python生态（如Pandas、Great Expectations、Pydantic）给出具体的度量方法和代码示例。

基础数据质量维度与度量方法

1 完整性（Completeness）

度量：缺失值比例

import pandas as pd
def completeness_check(df, columns=None):
    """计算每列/指定列的完整性"""
    if columns is None:
        columns = df.columns
    total_rows = len(df)
    completeness = {}
    for col in columns:
        non_null_count = df[col].notna().sum()
        completeness[col] = non_null_count / total_rows * 100
    return pd.Series(completeness)
# 示例
data = pd.DataFrame({
    'user_id': [1, 2, None, 4, 5],
    'email': ['a@b.com', None, 'c@d.com', 'e@f.com', None]
})
print(completeness_check(data))

2 准确性（Accuracy）

度量：值与真实/预期值的匹配程度

import re
from datetime import datetime
def accuracy_check(df, rules):
    """
    rules: dict of {column: validation_function}
    返回每条记录的准确性得分
    """
    results = {}
    for col, func in rules.items():
        results[col] = df[col].apply(func).mean() * 100
    return results
# 示例规则
validation_rules = {
    'email': lambda x: bool(re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', str(x))) if pd.notna(x) else False,
    'age': lambda x: 0 <= x <= 150 if pd.notna(x) else False,
    'date': lambda x: isinstance(x, datetime) if pd.notna(x) else False
}
print(accuracy_check(data, validation_rules))

3 一致性（Consistency）

度量：数据格式、单位或逻辑关系的统一性

def consistency_check(df, consistency_rules):
    """
    检查字段间/字段内的一致性
    如：开始日期 <= 结束日期
    """
    issues = []
    for rule_name, condition in consistency_rules.items():
        violations = (~df.eval(condition)).sum()
        issues.append({
            'rule': rule_name,
            'violations': violations,
            'consistency_score': (len(df) - violations) / len(df) * 100
        })
    return pd.DataFrame(issues)
# 示例
df = pd.DataFrame({
    'start_date': pd.to_datetime(['2023-01-01', '2023-02-01', '2023-03-01']),
    'end_date': pd.to_datetime(['2023-01-10', '2023-01-15', '2023-02-20'])
})
rules = {
    'date_range_valid': 'start_date <= end_date'
}
print(consistency_check(df, rules))

4 及时性（Timeliness）

度量：数据延迟、更新频率、新鲜度

from datetime import datetime, timedelta
def timeliness_check(df, date_column, max_lag_hours=24):
    """检查数据延迟是否在可接受范围内"""
    now = datetime.now()
    df['data_lag_hours'] = (now - df[date_column]).dt.total_seconds() / 3600
    timely = (df['data_lag_hours'] <= max_lag_hours).mean() * 100
    return {
        'timeliness_score': timely,
        'max_lag_hours': df['data_lag_hours'].max(),
        'avg_lag_hours': df['data_lag_hours'].mean()
    }
# 示例
df = pd.DataFrame({
    'event_time': [datetime.now() - timedelta(hours=i*2) for i in range(10)]
})
print(timeliness_check(df, 'event_time', max_lag_hours=12))

5 唯一性（Uniqueness）

度量：重复记录比例

def uniqueness_check(df, key_columns):
    """检查关键字段的唯一性"""
    total_records = len(df)
    duplicates = df.duplicated(subset=key_columns, keep=False).sum()
    return {
        'total_records': total_records,
        'duplicate_records': duplicates,
        'uniqueness_score': (total_records - duplicates) / total_records * 100
    }
# 示例
df = pd.DataFrame({
    'user_id': [1, 2, 2, 4, 5],  # ID为2的记录重复
    'name': ['Alice', 'Bob', 'Bob', 'David', 'Eve']
})
print(uniqueness_check(df, ['user_id']))

整合的数据质量评分框架

class DataQualityMetrics:
    def __init__(self, df):
        self.df = df
        self.results = {}
    def calculate_completeness(self, columns=None):
        cols = columns or self.df.columns
        scores = {}
        for col in cols:
            non_null_pct = self.df[col].notna().mean() * 100
            scores[col] = round(non_null_pct, 2)
        self.results['completeness'] = scores
        return scores
    def calculate_uniqueness(self, key_columns):
        dup_count = self.df.duplicated(subset=key_columns).sum()
        score = (1 - dup_count / len(self.df)) * 100
        self.results['uniqueness'] = round(score, 2)
        return score
    def calculate_accuracy(self, rules):
        scores = {}
        for col, func in rules.items():
            valid = self.df[col].apply(func).sum()
            score = (valid / len(self.df)) * 100
            scores[col] = round(score, 2)
        self.results['accuracy'] = scores
        return scores
    def calculate_timeliness(self, date_col, max_lag_hours=24):
        lag = (datetime.now() - self.df[date_col]).dt.total_seconds() / 3600
        timely_pct = (lag <= max_lag_hours).mean() * 100
        self.results['timeliness'] = round(timely_pct, 2)
        return timely_pct
    def calculate_consistency(self, rules):
        scores = {}
        for rule_name, condition in rules.items():
            valid = self.df.eval(condition).sum()
            score = (valid / len(self.df)) * 100
            scores[rule_name] = round(score, 2)
        self.results['consistency'] = scores
        return scores
    def overall_score(self, weights=None):
        """计算综合质量评分"""
        if weights is None:
            weights = {
                'completeness': 0.3,
                'accuracy': 0.25,
                'consistency': 0.2,
                'timeliness': 0.15,
                'uniqueness': 0.1
            }
        # 获取各维度平均分
        avg_scores = {}
        for dim in ['completeness', 'accuracy', 'consistency', 'timeliness', 'uniqueness']:
            if dim in self.results:
                if isinstance(self.results[dim], dict):
                    avg_scores[dim] = sum(self.results[dim].values()) / len(self.results[dim])
                else:
                    avg_scores[dim] = self.results[dim]
        total = sum(avg_scores.get(dim, 0) * weights.get(dim, 0) for dim in weights)
        return round(total, 2)
# 使用示例
df = pd.DataFrame({
    'user_id': [1, 2, 2, 4, 5],
    'name': ['Alice', 'Bob', 'Bob', 'David', None],
    'email': ['a@b.com', 'b@c.com', None, 'd@e.com', 'e@f.com'],
    'created_at': [datetime.now() - timedelta(hours=i*3) for i in range(5)]
})
metrics = DataQualityMetrics(df)
# 逐维度计算
completeness = metrics.calculate_completeness()
uniqueness = metrics.calculate_uniqueness(['user_id'])
accuracy = metrics.calculate_accuracy({
    'email': lambda x: bool(re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', str(x))) if pd.notna(x) else False
})
timeliness = metrics.calculate_timeliness('created_at', max_lag_hours=24)
print(f"Completeness: {completeness}")
print(f"Uniqueness: {uniqueness}")
print(f"Accuracy: {accuracy}")
print(f"Timeliness: {timeliness}")
print(f"Overall Score: {metrics.overall_score()}")

常用专业工具

1 Great Expectations（推荐）

import great_expectations as ge
# 将DataFrame转为Great Expectations格式
ge_df = ge.from_pandas(df)
# 定义期望/规则
ge_df.expect_column_values_to_not_be_null('user_id')
ge_df.expect_column_values_to_match_regex('email', r'^[\w\.-]+@[\w\.-]+\.\w+$')
ge_df.expect_column_values_to_be_unique('user_id')
# 生成数据质量报告
results = ge_df.validate()
print(results)

2 Pydantic（API输入验证）

from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional
class UserAPIInput(BaseModel):
    user_id: int = Field(..., ge=0)
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: Optional[int] = Field(None, ge=0, le=150)
    @validator('name')
    def name_must_be_alpha(cls, v):
        if not v.isalpha():
            raise ValueError('Name must contain only letters')
        return v
# 自动验证API输入数据质量
try:
    user = UserAPIInput(**raw_data)
    print("Data quality check passed")
except Exception as e:
    print(f"Data quality violation: {e}")

实际应用建议

分阶段度量：
- 输入阶段：使用Pydantic/Schema验证
- 处理阶段：使用Pandas内置检查
- 输出阶段：使用Great Expectations生成报告
阈值设定：
- 关键字段（如ID）：完整性≥99.9%
- 业务字段（如金额）：准确性≥99%
- 辅助字段（如备注）：完整性≥80%

监控与告警：

def alert_if_data_quality_drops(quality_score, threshold=95):
 if quality_score < threshold:
     send_alert(f"Data quality score dropped to {quality_score}%")