算法工程师的专业底线：CTR模型上线前的三道校验

import pandas as pd

def audit_leakage(df: pd.DataFrame, label_ts: str, feats: list,
                  tol_sec: int = 60, thresh: float = 0.01) -> dict:
    """
    df: 训练样本；label_ts: 标签时间戳列名
    每个特征需有 <feat>__ts 列，记录该特征值实际可得时间
    tol_sec: 允许的时钟容差（秒）
    返回疑似穿越的特征及越界比例
    """
    leak = {}
    for f in feats:
        ts = pd.to_datetime(df[f"{f}__ts"])
        gap = (ts - pd.to_datetime(df[label_ts])).dt.total_seconds()
        # 特征可得时间晚于标签时间 + 容差 => 穿越
        bad_ratio = (gap > tol_sec).mean()
        if bad_ratio > thresh:
            leak[f] = round(float(bad_ratio), 4)
    return leak

import numpy as np

def psi(expected: np.ndarray, actual: np.ndarray, bins: int = 10) -> float:
    """群体稳定性指数：量化训练/服务分布偏移"""
    qs = np.quantile(expected, np.linspace(0, 1, bins + 1))
    qs[0], qs[-1] = -np.inf, np.inf
    e = np.histogram(expected, bins=qs)[0] / len(expected) + 1e-6
    a = np.histogram(actual, bins=qs)[0] / len(actual) + 1e-6
    return float(np.sum((a - e) * np.log(a / e)))

def check_consistency(offline: pd.DataFrame, online: pd.DataFrame,
                      feats: list, psi_thr: float = 0.1) -> dict:
    report = {}
    for f in feats:
        o, n = offline[f].fillna(0), online[f].fillna(0)
        # 1) 逐样本数值一致性：允许浮点/精度误差
        diff = (o.values - n.values)
        rel = np.abs(diff) / (np.abs(o.values) + 1e-6)
        # 2) 分布稳定性
        p = psi(o.values, n.values)
        bad = (rel > 1e-3).mean() > 0.01 or p > psi_thr
        if bad:
            report[f] = {"rel_mismatch": round(float((rel > 1e-3).mean()), 4),
                         "psi": round(p, 4)}
    return report

from scipy.stats import norm

def sample_size(p0: float, mde_rel: float, alpha=0.05, power=0.8) -> int:
    """p0: 基线CTR; mde_rel: 相对提升，如0.03表示+3%"""
    p1 = p0 * (1 + mde_rel)
    pbar = (p0 + p1) / 2
    za, zb = norm.ppf(1 - alpha / 2), norm.ppf(power)
    n = ((za * np.sqrt(2 * pbar * (1 - pbar))
          + zb * np.sqrt(p0 * (1 - p0) + p1 * (1 - p1))) ** 2) / (p1 - p0) ** 2
    return int(np.ceil(n))