03 · 数据契约

   ① 你不在他的 KPI 里
      上游团队的目标是"业务功能上线"，不是"下游模型别挂"
      他甚至【不知道】有你这个消费者

   ② 下游是不可见的
      一张表可能有 40 个消费者：报表、风控、推荐、财务对账、外部接口
      上游打开表的元信息，看到的是：什么都没有

   ③ 改动成本严重不对称  ⭐⭐
      上游改一行 SQL / 一个枚举值：    5 分钟
      下游发现异常 → 定位 → 修复 → 回溯：3 天
      ——「省下 5 分钟」的收益归他，「花掉 3 天」的成本归你

dataset: dwd_feed_impression_di
owner: feed-client@team          # 4 所有权
oncall: "#feed-data-oncall"
consumers:                        # 7 消费者清单 ⭐
  - {team: rank-algo, usage: "CTR 模型负样本", contact: "@zhang"}
  - {team: bi,        usage: "日报曝光量",    contact: "@li"}

semantics:                        # 2 语义 ⭐⭐
  impression:
    definition: "item 进入视口且停留 >= 1000ms"
    not_counted: ["预加载", "后台刷新", "滑动中未停留"]
    unit: "次"
    timezone: "Asia/Shanghai"     # dt 分区按此时区切
    since_version: "v3"           # 该口径自哪一版生效

schema:                           # 1 Schema
  - {name: dt,         type: date,   nullable: false}
  - {name: user_id,    type: string, nullable: false}
  - {name: item_id,    type: string, nullable: false}
  - {name: channel,    type: string, nullable: false,
     enum: [app_home, app_search, h5, mini_program]}   # ⚠️ 枚举必须列全集
  - {name: dwell_ms,   type: int,    nullable: true, min: 0, max: 600000}
primary_key: [dt, user_id, item_id, ts]

slo:                              # 3 质量 SLO
  freshness: "T+1 07:00 前产出"
  row_count: {min: 3.0e8, max: 5.0e8}   # ⭐ 绝对量区间，比"波动 ±40%"严
  null_rate: {user_id: 0.0, dwell_ms: 0.05}
  uniqueness: {primary_key: 1.0}

change_policy:                    # 5 + 6
  breaking: ["删字段", "改类型", "改主键", "改口径定义", "缩小枚举"]
  non_breaking: ["加字段", "加枚举值（需下游确认）"]
  notice_days: 14
  dual_write_days: 7

   上游产出                落地入库                 你消费
   ────────                ────────                 ────────
   闸①：生产端 CI          闸②：到达即校验          闸③：训练入口断言
   ────────                ────────                 ────────
   在上游的 PR 里跑         数据一落盘就跑            train.py 第一行就跑
   schema diff             SLO 断言                 你自己的假设
   → 破坏性变更            → 失败则【隔离到          → 不满足就 raise
     直接 block 合并 ⭐      quarantine】而不是
                            让它往下流 ⭐⭐

   成本：改一次 CI 配置      成本：一个调度任务         成本：20 行代码
   收益：把问题挡在源头      收益：脏数据不污染下游      收益：兜底，永远该有

import pandas as pd

def check_contract(df: pd.DataFrame, contract: dict):
    """按契约校验一个 DataFrame。返回 (是否通过, 违规列表)。
    contract 结构见本章 YAML：schema / primary_key / slo。"""
    v = []
    sch = {c["name"]: c for c in contract.get("schema", [])}

    for name, spec in sch.items():
        if name not in df.columns:
            v.append(("P0", name, "字段缺失", ""))
            continue
        s = df[name]
        if not spec.get("nullable", True) and s.isna().any():
            v.append(("P0", name, "不可空字段出现空值", f"{s.isna().mean():.2%}"))
        if "enum" in spec:
            allowed = set(spec["enum"])
            unknown = set(s.dropna().unique()) - allowed
            # ⭐ 枚举出现未知值 = 上游加了新取值却没通知，是最高频的静默破坏
            if unknown:
                share = s.isin(unknown).mean()
                v.append(("P1", name, f"未知枚举值 {sorted(unknown)[:5]}", f"占比 {share:.2%}"))
        if "min" in spec and pd.api.types.is_numeric_dtype(s) and s.min() < spec["min"]:
            v.append(("P1", name, "低于取值下限", f"{s.min()}"))
        if "max" in spec and pd.api.types.is_numeric_dtype(s) and s.max() > spec["max"]:
            # ⭐ 上限断言能抓住"单位从元改成分"这类 ×100 事故
            v.append(("P1", name, "超出取值上限", f"{s.max()}"))

    pk = contract.get("primary_key")
    if pk and all(c in df.columns for c in pk):
        dup = df.duplicated(subset=pk).sum()
        if dup:
            v.append(("P0", ",".join(pk), "主键重复", f"{dup} 行"))

    slo = contract.get("slo", {})
    rc = slo.get("row_count")
    if rc:
        n = len(df)
        if n < rc["min"] or n > rc["max"]:
            level = "P0" if n < rc["min"] / 2 else "P1"
            v.append((level, "<table>", "行数越界", f"{n} 不在 [{rc['min']:.0f},{rc['max']:.0f}]"))
    for col, thr in (slo.get("null_rate") or {}).items():
        if col in df.columns and df[col].isna().mean() > thr:
            v.append(("P2", col, "缺失率超阈值", f"{df[col].isna().mean():.2%} > {thr:.2%}"))

    blocking = [x for x in v if x[0] in ("P0", "P1")]
    return len(blocking) == 0, pd.DataFrame(v, columns=["级别", "字段", "问题", "详情"])

# 用法：ok, report = check_contract(df, contract_dict)
#      if not ok: 写入 quarantine 分区，不要写主表  ⭐

import sys

BREAKING = {"删字段", "改类型", "改主键", "缩小枚举", "改口径定义", "改单位", "改时区"}

def diff_schema(old: dict, new: dict):
    """对比两版契约，输出变更清单和是否含破坏性变更。"""
    changes = []
    o = {c["name"]: c for c in old["schema"]}
    n = {c["name"]: c for c in new["schema"]}
    for name in o.keys() - n.keys():
        changes.append(("删字段", name, o[name].get("type"), None))
    for name in n.keys() - o.keys():
        changes.append(("加字段", name, None, n[name].get("type")))
    for name in o.keys() & n.keys():
        if o[name].get("type") != n[name].get("type"):
            changes.append(("改类型", name, o[name].get("type"), n[name].get("type")))
        if o[name].get("nullable") is False and n[name].get("nullable") is True:
            changes.append(("放宽可空", name, False, True))
        oe, ne = set(o[name].get("enum", [])), set(n[name].get("enum", []))
        if oe - ne:
            changes.append(("缩小枚举", name, sorted(oe - ne), None))
        if ne - oe:
            # ⭐ 新增枚举对机器兼容、对下游可能是破坏：单列出来强制下游确认
            changes.append(("新增枚举", name, None, sorted(ne - oe)))
    if old["primary_key"] != new["primary_key"]:
        changes.append(("改主键", "<pk>", old["primary_key"], new["primary_key"]))
    # ⭐ 语义变更靠版本号发现：口径改了就必须 bump since_version
    for k, sem in (new.get("semantics") or {}).items():
        old_sem = (old.get("semantics") or {}).get(k, {})
        if old_sem and old_sem.get("definition") != sem.get("definition"):
            changes.append(("改口径定义", k, old_sem.get("definition"), sem.get("definition")))
        for key, label in (("unit", "改单位"), ("timezone", "改时区")):
            if old_sem and old_sem.get(key) != sem.get(key):
                changes.append((label, k, old_sem.get(key), sem.get(key)))
    has_breaking = any(c[0] in BREAKING for c in changes)
    return has_breaking, changes

# CI 里：has_breaking, changes = diff_schema(old, new)
#        if has_breaking and not pr_has_label("approved-breaking"): sys.exit(1)

   系统：某平台的用户信用分模型，特征表里有一列 channel（注册渠道）
   训练时 channel 的取值：app_home / app_search / h5   （3 个）
   代码里的处理：
       CHANNEL_MAP = {"app_home": 0, "app_search": 1, "h5": 2}
       x = CHANNEL_MAP.get(row.channel, 3)     # ⚠️ 3 = "unknown"

   2 月 5 日：小程序端上线，埋点新增枚举值 mini_program
   上游的判断："只是加了个取值，字段没动，向后兼容" —— 没有通知任何人

   ✅ 字段还在，类型没变，非空率 100%
   ✅ 行数在正常区间（小程序是增量，总量甚至涨了）
   ✅ 训练任务、推理服务全部正常，零报错
   ✅ 整体 AUC 从 0.781 微降到 0.774 —— 在正常波动范围内 💀

   💀 真相：mini_program 流量从 0% 一路涨到 23%，
      这 23% 的用户 channel 全部被映射成 "unknown"(3)，
      而 "unknown" 在训练集里只占 0.4%，模型对它几乎没有学过
      → 这 23% 用户拿到的是一个近乎随机的分档

   ⭐ 分渠道拆开之后（事后才做的）：
      app_home / app_search / h5 :  AUC 0.784  （正常）
      mini_program              :  AUC 0.548  ⚠️ 接近随机

   发现延迟：6 周（2/5 → 3/19）
   发现方式：小程序运营抱怨"通过率忽高忽低，看不出规律"

   影响：小程序渠道 41 万次授信决策使用了近乎随机的信用分
        其中约 1.7 万笔本该拒绝的通过了，事后逾期率是正常组的 3.4 倍
        直接坏账增量 ≈ 216 万元
   修复：重建映射 + 回溯 90 天特征 + 重训 = 8 人日

   第 1 周：给你最重要的 3 张表，各写一份 30 行的契约（YAML 就行）⭐
           重点写【语义】和【消费者清单】——这两块光写下来就有价值
   第 2 周：把 check_contract 挂到训练入口（闸③），先只告警不阻断
   第 3–4 周：观察误报，收敛阈值；确认稳定后，把 P0/P1 改成真阻断
   第 2 个月：推动闸②（落地即校验 + quarantine 分区）
   第 3 个月：找上游谈闸①（CI 里跑 diff_schema）——
             ⭐ 这一步是政治问题不是技术问题，
                拿着前两个月攒下的【拦截记录】去谈，成功率高得多