3. 数据采集与预处理

ETA预测模型的精度高度依赖输入数据的质量。本章聚焦于三类核心数据源——GPS轨迹数据、订单数据、天气数据的采集策略与清洗工程。我们将从数据源特征出发，设计采集管道，并针对每类数据特有的脏数据模式（漂移、缺失、异常、时区不一致等）给出可落地的清洗方案。

3.1 数据源特征与采集策略

三类数据在采集频率、数据量级、实时性要求上差异显著，需采用不同的采集管道。

3.2 GPS轨迹数据清洗实战

GPS轨迹数据是ETA预测的核心输入，但也是最容易产生脏数据的数据源。常见问题包括：定位漂移、信号丢失、时间戳异常、坐标重复等。

3.2.1 清洗流程

1. 去重与排序：按设备ID + 时间戳排序，去除完全重复的记录。
2. 时间戳校验：过滤时间戳早于1970-01-01或晚于当前时间+1小时的记录。
3. 坐标范围过滤：剔除经纬度超出城市边界（如北京：39.4-41.1, 115.4-117.5）的点。
4. 速度异常检测：计算相邻点间瞬时速度，剔除速度 > 120 km/h（城市道路）或 > 200 km/h（高速）的漂移点。
5. 停留点合并：连续5分钟以上速度 < 1 km/h 且坐标变化 < 50m 的点，合并为停留事件。
6. 轨迹分段：相邻点时间间隔 > 30分钟 或 距离 > 5km，自动切分为新轨迹段。

# GPS轨迹清洗核心逻辑（Python伪代码）
def clean_gps_trajectory(df: pd.DataFrame) -> pd.DataFrame:
    # 1. 排序与去重
    df = df.sort_values(['device_id', 'timestamp']).drop_duplicates(
        subset=['device_id', 'timestamp', 'lat', 'lng']
    )
    # 2. 时间戳过滤
    now = pd.Timestamp.now(tz='UTC')
    df = df[(df['timestamp'] > '1970-01-01') & (df['timestamp'] <= now + pd.Timedelta(hours=1))]
    # 3. 坐标范围过滤（以北京为例）
    df = df[(df['lat'].between(39.4, 41.1)) & (df['lng'].between(115.4, 117.5))]
    # 4. 速度异常检测（使用haversine计算距离）
    df['dist_km'] = haversine_vectorized(df['lat'], df['lng'], df['lat'].shift(1), df['lng'].shift(1))
    df['time_diff_h'] = (df['timestamp'] - df['timestamp'].shift(1)).dt.total_seconds() / 3600
    df['speed_kmh'] = df['dist_km'] / df['time_diff_h'].replace(0, np.nan)
    df = df[(df['speed_kmh'].isna()) | (df['speed_kmh'] <= 120)]
    return df.drop(columns=['dist_km', 'time_diff_h', 'speed_kmh'])

3.3 订单数据清洗实战

订单数据通常来自业务数据库，脏数据模式相对固定：字段缺失、状态机异常、时间字段时区不一致、重复支付等。

3.3.1 清洗规则清单

3.3.2 状态机校验示例

# 订单状态机校验
VALID_TRANSITIONS = {
    'created': ['paid', 'cancelled'],
    'paid': ['picked_up', 'cancelled'],
    'picked_up': ['delivered', 'cancelled'],
    'delivered': ['completed'],
    'completed': [],
    'cancelled': []
}

def validate_order_status(history: List[Dict]) -> bool:
    """校验单个订单的状态流转是否合法"""
    sorted_history = sorted(history, key=lambda x: x['timestamp'])
    for i in range(len(sorted_history) - 1):
        curr_status = sorted_history[i]['status']
        next_status = sorted_history[i+1]['status']
        if next_status not in VALID_TRANSITIONS.get(curr_status, []):
            return False
    return True

3.4 天气数据清洗实战

天气数据通常来自外部API，脏数据主要表现为：缺失值（传感器故障）、异常值（极端天气记录错误）、空间粒度不匹配（气象站与配送区域不重合）。

3.4.1 清洗策略

• 缺失值处理：温度、湿度等连续变量使用前后2小时均值插补；降水量、风向等离散变量使用最近邻气象站数据填充。
• 异常值检测：使用3-sigma原则或IQR方法，过滤温度超出[-20°C, 50°C]、风速 > 50 m/s 等明显异常记录。
• 空间匹配：将气象站坐标与配送区域（网格或行政区划）进行空间连接，每个区域取最近3个气象站的加权平均（反距离权重法）。

# 反距离权重插值（IDW）示例
def idw_interpolation(station_df, target_lat, target_lng, power=2):
    """根据气象站数据插值目标点的天气值"""
    distances = haversine_vectorized(
        station_df['lat'], station_df['lng'],
        target_lat, target_lng
    )
    weights = 1 / (distances ** power + 1e-10)
    weights /= weights.sum()
    interpolated_value = (station_df['value'] * weights).sum()
    return interpolated_value

3.5 多源数据对齐与时间窗口划分

清洗后的三类数据需要按时间维度和空间维度对齐，才能用于ETA模型训练。

3.6 数据质量监控与告警

清洗管道上线后，必须建立数据质量监控体系，及时发现数据源异常或清洗逻辑失效。

3.7 本章小结