16-EFT.WP.Methods.Cleaning v1.0 | 附录B 契约库与策略卡

目录 ／ 文档-技术白皮书 ／ 16-EFT.WP.Methods.Cleaning v1.0

附录B 契约库与策略卡

一句话目标：提供可直接装配于 assert_contract 的标准契约模板与场景化策略卡，统一阈值、窗口、处置与审计口径。

I. 总则与适用范围

1. 适用对象
   • 批处理数据集 ds、在线负载 req/resp、事件流消息 msg。
   • 覆盖模式、计量、时基、路径与到达时、缺失与插补、异常与漂移、去重与参照、发布与审计。
2. 基本符号与窗口
   • 内部计算在 tau_mono，发布在 ts。
   • 统计窗口 Delta_t 与分位函数 Q_p(x)（如 Q_0.99(x)=P99）。
   • 掩码 m ∈ {0,1} 的语义由策略定义，默认 m=1 表示缺失。
3. 契约执行与落盘
   • 契约以命名断言集合 tests.* 传入 assert_contract(ds, tests)。
   • 失败分级 sev ∈ {S0,S1,S2,S3}，并触发处置动作与审计链更新。
   • 任何涉及 T_arr 的契约必须并行计算两口径与 delta_form 并落盘。

II. 契约命名与层级

• tests.schema.*：模式与键类
• tests.units.*：单位与量纲类
• tests.time.*：时间轴与同步类
• tests.path.*：路径与到达时类
• tests.missing.*：缺失与插补类
• tests.range.*：范围与值域类
• tests.outlier.*：异常与漂移类
• tests.relate.*：去重与参照类
• tests.release.*：发布与审计类

III. 契约模板清单（断言规格）

1. tests.schema.unique(keys)
   • 目标：unique(keys)。
   • 失败：发现冲突组。sev=S2（默认）。
2. tests.schema.foreign_key(child.key -> parent.key)
   • 目标：外键可解。
   • 失败：孤儿记录计数超阈 orphan_max。sev=S2|S3。
3. tests.units.dim(expr, target_dim)
   • 目标：check_dim( expr - as_dim(target_dim) ) = 0。
   • 失败：量纲不守恒。sev=S3。
4. tests.units.unit_system(system)
   • 目标：字段单位属于 system，必要时可转换。
   • 失败：不可转换字段计数超阈。sev=S2。
5. tests.time.monotone(field)
   • 目标：non_decreasing(field)。
   • 失败：逆序或回拨。sev=S2。
6. tests.time.sync_bounds(offset_max, skew_max, J_max)
   • 目标：|offset| ≤ offset_max, |skew| ≤ skew_max, J ≤ J_max。
   • 失败：超界。sev=S1|S2。
7. tests.path.monotone(ell)
   • 目标：non_decreasing(ell)。
   • 失败：路径回退。sev=S2。
8. tests.path.length(L_min)
   • 目标：L_gamma = ( ∫_gamma 1 d ell ) ≥ L_min。
   • 失败：路径长度不足。sev=S1。
9. tests.path.arrival_forms(c_ref, tol_Tarr, tolP99_Tarr)
   • 目标：
     1. T_arr_1 = ( 1 / c_ref ) * ( ∫_{gamma(ell)} n_eff d ell )
     2. T_arr_2 = ( ∫_{gamma(ell)} ( n_eff / c_ref ) d ell )
     3. delta_form = | T_arr_1 - T_arr_2 |
     4. 断言：delta_form ≤ tol_Tarr 且 Q_0.99(delta_form) ≤ tolP99_Tarr。
   • 失败：sev=S2|S3。
10. tests.missing.coverage(m, max_ratio)
    • 目标：mean(m) ≤ max_ratio（默认 m=1 表示缺失）。
    • 失败：覆盖率不足。sev=S1。
11. tests.missing.impute_tagged(fields)
    • 目标：被插补字段均具来源与 RefCond 标注。
    • 失败：未标注插补。sev=S2。
12. tests.range.within(field, lo, hi, inclusive)
    • 目标：field ∈ [lo,hi] 或 (lo,hi)。
    • 失败：越界比例超阈。sev=S1|S2。
13. tests.outlier.rate(method, fields, rate_max)
    • 目标：离群率 ≤ rate_max。
    • 失败：sev=S1。
14. tests.outlier.drift(method, ref, psi_max | ks_max)
    • 目标：漂移度量 ≤ 阈值。
    • 失败：sev=S1|S2。
15. tests.relate.dedup_resolved(keys, conflicts_max)
    • 目标：去重后冲突残留 ≤ conflicts_max。
    • 失败：sev=S2。
16. tests.release.manifest_signed()
    • 目标：发布前存在 manifest，含 hash_sha256(blob) 与 signature。
    • 失败：sev=S3。

IV. 策略卡模板（结构与字段）

1. 字段
   • name：策略卡名称
   • scene：适用场景
   • Delta_t：统计窗口
   • thresholds：阈值集合（含 tol_Tarr, tolP99_Tarr, offset/skew/J 等）
   • actions：按 sev 的处置映射
   • audit：审计与告警通道
   • notes：补充约束
2. 处置级别映射（默认）
   • S0：记录审计，仅告警
   • S1：降权或局部隔离
   • S2：整体隔离到 topic_quarantine 或 ds_quarantine，需人工复核
   • S3：回滚/阻断发布，启动应急流程

V. 策略卡样例（推荐默认）

1. SC-Offline-Daily
   • scene：T+1 批处理发布
   • Delta_t = 24 h
   • thresholds
     1. tests.schema.unique(pk)
     2. tests.units.unit_system("SI")
     3. tests.units.dim( t_arr, "[T]" )
     4. tests.time.monotone(ts)
     5. tests.path.arrival_forms(c_ref=const, tol_Tarr=5e-6 s, tolP99_Tarr=1e-5 s)
     6. tests.missing.coverage(m, max_ratio=0.05)
     7. tests.range.within(q_score, 0, 1, inclusive=true)
   • actions：S1→warn, S2→quarantine_batch, S3→rollback_release
   • audit：写入审计链与日报
2. SC-Online-API
   • scene：在线读服务
   • Delta_t = 5 min
   • thresholds
     1. tests.time.sync_bounds(offset_max=5 ms, skew_max=50 ppm, J_max=3 ms)
     2. tests.path.arrival_forms(c_ref=const, tol_Tarr=2e-6 s, tolP99_Tarr=5e-6 s)
     3. tests.outlier.rate(method="MAD", fields=[x], rate_max=0.01)
   • actions：S1→degrade_feature, S2→shadow_read+quarantine, S3→circuit_break
   • audit：实时告警与仪表板
3. SC-Stream-Topic
   • scene：事件流清洗
   • Delta_t = 15 min 滚动
   • thresholds
     1. tests.time.monotone(tau_mono)
     2. tests.missing.coverage(m, max_ratio=0.02)
     3. tests.outlier.drift(method="PSI", ref="last_7d", psi_max=0.1)
   • actions：S1→tag, S2→route_to(topic_quarantine), S3→halt_partition
   • audit：偏移、重放指令落盘

VI. 计算与度量口径

1. 分位与窗口
   • Q_p(x) 在 Delta_t 上以有界内存算法估计，记录 u(Q_p)。
   • 对于到达时：同时产出 Q_0.50(delta_form) 与 Q_0.99(delta_form)。
2. 误差与预算
   与 TS.sli.* 对齐，允许策略卡引用 err_rate、lat_ms、fresh_age 等以联动处置。

VII. 契约 DSL（建议表达）

1. 基本形式
   • UNIQUE(pk)
   • FK(child.key -> parent.key)
   • DIM("t_arr","[T]")
   • UNIT_SYSTEM("SI")
   • MONOTONE("ts")
   • SYNC_BOUNDS(offset<=5 ms, skew<=50 ppm, J<=3 ms)
   • ARRIVAL_FORMS(c_ref=const, tol=5e-6 s, tolP99=1e-5 s)
   • COVERAGE("m", <=0.05)
   • WITHIN("q_score", 0, 1, inclusive=true)
   • OUTLIER_RATE(method="MAD", fields=["x"], <=0.01)
   • DRIFT(method="PSI", ref="last_7d", <=0.1)
   • DEDUP_RESOLVED(keys=["pk"], conflicts_max=0)
   • MANIFEST_SIGNED()
2. 组合与条件
   ALL( ... )、ANY( ... )、IF tag("lab") THEN RELAX(OUTLIER_RATE, x2)。

VIII. 失败分级到处置映射（统一规则）

• S0（信息）：仅审计与面板展示。
• S1（轻度）：打标降权，允许继续流水。
• S2（严重）：隔离对象到 ds_quarantine 或 topic_quarantine，阻断对外发布。
• S3（致命）：回滚或冻结当期发布，触发应急与变更审核。

IX. 集成示例（assert_contract 的测试组）

1. 批处理最小契约组（示例）
   • UNIQUE(pk)
   • FK(child.pid -> parent.pid)
   • UNIT_SYSTEM("SI")
   • DIM("t_arr","[T]")
   • MONOTONE("ts")
   • ARRIVAL_FORMS(c_ref=c_ref_2024Q4, tol=5e-6 s, tolP99=1e-5 s)
   • COVERAGE("m", <=0.05)
   • WITHIN("q_score", 0, 1, inclusive=true)
   • MANIFEST_SIGNED()
2. 事件流最小契约组（示例）
   • MONOTONE("tau_mono")
   • SYNC_BOUNDS(offset<=5 ms, skew<=50 ppm, J<=3 ms)
   • OUTLIER_RATE(method="MAD", fields=["Xi"], <=0.01)

X. 审计与回退协同

1. 审计链
   • 任一次契约执行均生成 audit_event，含 who, when, tests, pass, hash_prev，并计算 hash_sha256(blob)。
   • 发布冻结前必须通过 MANIFEST_SIGNED()，并记录 signature。
2. 回退策略
   S3 触发 freeze_release 回滚到上一个签名版本；对流式则暂停分区并挂起重放。

小结
本附录定义了从断言语义到场景化阈值与处置动作的可复用契约库，并以策略卡统一窗口、阈值与审计联动。装配 tests.* 于 assert_contract，即可在批、在线与流式场景实现“可声明、可审计、可回退”的清洗质量闸门。