3. 数据采集与清洗：支付数据源与质量治理

-- 交易数据完整性检查
SELECT 
  COUNT(*) AS total_records,
  SUM(CASE WHEN txn_amount IS NULL THEN 1 ELSE 0 END) AS null_amount,
  SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) AS null_user_id,
  SUM(CASE WHEN device_id IS NULL THEN 1 ELSE 0 END) AS null_device_id
FROM raw_transactions
WHERE dt = '2025-03-01';

-- 准确性检查：金额异常
SELECT txn_id, txn_amount
FROM raw_transactions
WHERE txn_amount <= 0 OR txn_amount > 1000000;

-- 一致性检查：用户手机号比对
SELECT a.user_id, a.phone AS txn_phone, b.phone AS user_phone
FROM raw_transactions a
JOIN dim_user b ON a.user_id = b.user_id
WHERE a.phone != b.phone;

import pandas as pd
import numpy as np

# 加载清洗后的交易数据
df = pd.read_parquet('cleaned_transactions.parquet')

# 1. 业务规则过滤：金额必须为正且小于100万
df = df[(df['txn_amount'] > 0) & (df['txn_amount'] < 1_000_000)]

# 2. Z-Score检测金额异常
z_scores = np.abs((df['txn_amount'] - df['txn_amount'].mean()) / df['txn_amount'].std())
df['is_amount_outlier'] = (z_scores > 3).astype(int)

# 3. 时序异常：同一用户1小时内交易次数超过5次
df['txn_hour'] = df['txn_time'].dt.floor('H')
user_hourly_counts = df.groupby(['user_id', 'txn_hour']).size().reset_index(name='hourly_txn_count')
df = df.merge(user_hourly_counts, on=['user_id', 'txn_hour'], how='left')
df['is_freq_outlier'] = (df['hourly_txn_count'] > 5).astype(int)

# 4. 异常值处理：标记但不删除，保留给模型判断
df['outlier_flag'] = df['is_amount_outlier'] | df['is_freq_outlier']
print(f"异常记录占比: {df['outlier_flag'].mean():.2%}")