数据质量六维度实战教程

规则编号: DQ-CHECK-003
维度: 完整性
对象: asset_position 持仓表, 近 7 天分区
判定: valuation_date 空值计数为 0
阈值: 0
频率: 每个交易日 02:00 批次结束后
处置: 失败则告警并挂起当日估值发布
责任人: 数据应用组

-- 完整性: 持仓表估值日为空
SELECT COUNT(*) AS null_count
FROM asset_position
WHERE valuation_date IS NULL;

-- 唯一性: 同一账户、证券、估值日重复
SELECT account_id, security_code, valuation_date, COUNT(*) AS cnt
FROM asset_position
GROUP BY account_id, security_code, valuation_date
HAVING COUNT(*) > 1;

-- 准确性: 持仓数量或冻结数量为负
SELECT COUNT(*) AS negative_count
FROM asset_position
WHERE quantity < 0 OR frozen_quantity < 0;

-- 一致性: 持仓市值与总账余额按日核对
SELECT p.account_id, p.valuation_date
FROM asset_position p
JOIN account_balance b
  ON b.account_id = p.account_id
 AND b.biz_date = p.valuation_date
WHERE ABS(p.market_value - b.balance_amount) > 0.01;

-- 时效性: 最近 7 天估值数据是否到齐
SELECT MAX(valuation_date) AS latest_valuation_date
FROM asset_position
WHERE valuation_date >= CURRENT_DATE - INTERVAL 7 DAY;

-- 有效性: 证券代码不在证券字典中
SELECT DISTINCT p.security_code
FROM asset_position p
LEFT JOIN security_dict d
  ON d.security_code = p.security_code
WHERE d.security_code IS NULL;

上游文件 rows 与下游表记录 rows 对账:
1) 行数: 上游计数 == 下游计数
2) 金额: SUM(amount) 偏差 <= 0.01
3) 键集: 双方业务键集合相等, 差集即异常
4) 抽样: 对相同键的行做字段拼接并 md5, 两侧一致
全部通过 -> 批次标记 SUCCESS
任一失败 -> 进入对账异常队列并告警

def test_日终持仓加工_正确汇总在途与冻结():
    input_trades = 样例交易流水(成交 2 笔, 冻结 1 笔)
    output = run_position_pipeline(input_trades)
    assert output.quantity == 2
    assert output.frozen_quantity == 1
    assert output.total_amount == Decimal("9999.99")

def test_持仓表与总账余额_同一估值日口径一致():
    positions = query("SELECT * FROM asset_position WHERE valuation_date = TODAY")
    balances = query("SELECT * FROM account_balance WHERE biz_date = TODAY")
    for pos in positions:
        bal = balances[pos.account_id]
        assert abs(pos.market_value - bal.balance_amount) <= 0.01

-- 近 7 天交易金额分布
SELECT AVG(amount) AS avg_amount,
       STDDEV(amount) AS std_amount,
       COUNT(*) AS cnt
FROM asset_trade
WHERE trade_time >= NOW() - INTERVAL 7 DAY;
-- 与训练分布比较, 偏差超过阈值触发漂移告警