2. 数据质量基础：数据质量六大维度在运费场景的映射

在TMS运费数据治理中，数据质量并非抽象概念，而是直接决定运费计算准确性、结算效率与业务决策可信度的关键因素。本节将数据质量六大经典维度——准确性、完整性、一致性、及时性、唯一性、有效性——逐一映射到运费场景，帮助您建立从理论到实践的诊断框架。

2.1 六大维度定义与运费场景映射总览

下表将每个维度的通用定义、运费场景中的具体表现、以及典型异常案例进行对照，便于快速建立场景化理解。

2.2 维度交叉分析：运费场景中的典型质量问题链

单一维度的异常往往不是孤立事件。以下通过两个高频运费场景，展示维度之间的因果传导关系。

场景一：里程字段异常链

1. 完整性缺失：GPS设备离线导致某段路径点缺失 → 里程计算不完整
2. 准确性下降：基于不完整路径计算的里程偏离真实值（如少算30%）
3. 一致性冲突：TMS系统使用“规划里程”补全，而财务系统使用“实际里程”核算，两套数据口径不一致
4. 有效性失效：补全后的里程超出合理范围（如单日运输里程＞1500km）

场景二：费用金额重复支付链

1. 唯一性失效：运单号重复导致同一笔运费生成两条应付记录
2. 准确性失真：两条记录金额相同，合计后应付金额翻倍
3. 及时性风险：月末关账时重复记录未被识别，导致应付账款虚增
4. 一致性破坏：财务系统与TMS对账时，总金额差异无法解释

2.3 六大维度在运费数据清洗中的优先级矩阵

并非所有维度都需要在清洗初期投入同等资源。根据对运费准确性和结算效率的影响程度，建议按以下优先级排序：

2.4 维度映射的SVG架构图：运费数据质量六边形

以下SVG图将六大维度组织为相互连接的六边形结构，中心为“运费数据质量”，外围节点标注每个维度在运费场景中的关键检查项，帮助形成全局视角。

2.5 维度检查的Python实现示例：基于六大维度的数据质量评分函数

以下代码展示如何将六大维度抽象为可量化的质量评分函数，用于自动化扫描运费数据表。该函数返回每个维度的得分（0~100）及异常记录数。

import pandas as pd
import numpy as np

def freight_data_quality_score(df: pd.DataFrame, config: dict) -> dict:
    """
    运费数据质量六大维度评分函数
    
    Parameters:
    -----------
    df : DataFrame
        包含运费字段的TMS数据表
    config : dict
        各维度的检查配置，例如：
        {
            'accuracy': {'mileage_tolerance': 0.05},  # 里程允许偏差5%
            'completeness': {'required_fields': ['mileage', 'total_freight', 'origin', 'dest']},
            'consistency': {'mileage_source': 'gps'},  # 以GPS里程为基准
            'timeliness': {'max_delay_hours': 72},
            'uniqueness': {'key': 'shipment_id'},
            'validity': {'mileage_range': (0, 2000), 'freight_min': 0}
        }
    
    Returns:
    --------
    dict : 各维度得分及异常记录数
    """
    results = {}
    total = len(df)
    
    # 1. 准确性：检查里程偏差是否在容忍范围内
    if 'mileage_actual' in df.columns and 'mileage_planned' in df.columns:
        deviation = abs(df['mileage_actual'] - df['mileage_planned']) / df['mileage_planned']
        accuracy_errors = (deviation > config['accuracy']['mileage_tolerance']).sum()
        results['accuracy'] = {
            'score': round((1 - accuracy_errors / total) * 100, 2),
            'error_count': int(accuracy_errors)
        }
    
    # 2. 完整性：检查必填字段空值率
    required = config['completeness']['required_fields']
    missing_mask = df[required].isnull().any(axis=1)
    completeness_errors = missing_mask.sum()
    results['completeness'] = {
        'score': round((1 - completeness_errors / total) * 100, 2),
        'error_count': int(completeness_errors)
    }
    
    # 3. 一致性：以GPS里程为基准，检查规划里程偏差
    if 'mileage_gps' in df.columns and 'mileage_tms' in df.columns:
        consistency_errors = (abs(df['mileage_gps'] - df['mileage_tms']) > 10).sum()
        results['consistency'] = {
            'score': round((1 - consistency_errors / total) * 100, 2),
            'error_count': int(consistency_errors)
        }
    
    # 4. 及时性：检查运输完成到费用生成的时间差
    if 'delivery_time' in df.columns and 'freight_create_time' in df.columns:
        delay = (pd.to_datetime(df['freight_create_time']) - 
                 pd.to_datetime(df['delivery_time'])).dt.total_seconds() / 3600
        timeliness_errors = (delay > config['timeliness']['max_delay_hours']).sum()
        results['timeliness'] = {
            'score': round((1 - timeliness_errors / total) * 100, 2),
            'error_count': int(timeliness_errors)
        }
    
    # 5. 唯一性：检查主键重复
    key = config['uniqueness']['key']
    uniqueness_errors = df.duplicated(subset=[key]).sum()
    results['uniqueness'] = {
        'score': round((1 - uniqueness_errors / total) * 100, 2),
        'error_count': int(uniqueness_errors)
    }
    
    # 6. 有效性：检查里程范围与费用正负
    mileage_min, mileage_max = config['validity']['mileage_range']
    validity_errors = (
        (df['mileage'] < mileage_min) | 
        (df['mileage'] > mileage_max) | 
        (df['total_freight'] < config['validity']['freight_min'])
    ).sum()
    results['validity'] = {
        'score': round((1 - validity_errors / total) * 100, 2),
        'error_count': int(validity_errors)
    }
    
    return results

# 使用示例
if __name__ == "__main__":
    # 模拟运费数据
    sample_data = pd.DataFrame({
        'shipment_id': ['S001', 'S001', 'S002', 'S003', 'S004'],
        'mileage': [450, 450, -100, 800, 2500],
        'mileage_actual': [445, 445, 0, 790, 2480],
        'mileage_planned': [460, 460, 120, 810, 2520],
        'mileage_gps': [448, 448, None, 795, 2490],
        'mileage_tms': [450, 450, -100, 800, 2500],
        'total_freight': [5000, 5000, 0, 8500, 30000],
        'origin': ['北京', '北京', '上海', '广州', '深圳'],
        'dest': ['上海', '上海', None, '成都', '乌鲁木齐'],
        'delivery_time': ['2024-01-10 18:00', '2024-01-10 18:00', 
                          '2024-01-11 09:00', '2024-01-12 14:00', '2024-01-13 20:00'],
        'freight_create_time': ['2024-01-10 20:00', '2024-01-10 20:00',
                                '2024-01-15 10:00', '2024-01-12 16:00', '2024-01-14 08:00']
    })
    
    config = {
        'accuracy': {'mileage_tolerance': 0.05},
        'completeness': {'required_fields': ['mileage', 'total_freight', 'origin', 'dest']},
        'consistency': {'mileage_source': 'gps'},
        'timeliness': {'max_delay_hours': 72},
        'uniqueness': {'key': 'shipment_id'},
        'validity': {'mileage_range': (0, 2000), 'freight_min': 0}
    }
    
    scores = freight_data_quality_score(sample_data, config)
    for dim, result in scores.items():
        print(f"{dim}: 得分 {result['score']}%, 异常 {result['error_count']} 条")

2.6 维度映射的清洗优先级决策流程

以下SVG流程图展示在实际运费数据清洗项目中，如何根据六大维度的检查结果，动态决定清洗优先级与处理策略。

2.7 本章小结：从维度认知到清洗行动

• 六大维度是运费数据质量的“体检指标”，每个维度对应具体的业务风险点：准确性影响金额、完整性影响可用性、一致性影响对账、及时性影响结算周期、唯一性影响重复支付、有效性影响自动化处理。
• 维度之间存在因果传导链，清洗方案需从维度交叉处入手。例如，里程准确性异常往往由完整性缺失（GPS断点）或一致性冲突（规划 vs 实际口径）引发。
• 优先级矩阵建议：P0（准确性、完整性）需实时拦截；P1（一致性、有效性）可批量扫描；P2（唯一性、及时性）适合事后审计。
• Python评分函数提供了可落地的自动化检查工具，建议嵌入ETL数据质量门禁，实现“脏数据不入结算管道”。