01 · 数据才是那个瓶颈

import numpy as np
from sklearn.datasets import make_classification
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score

rng = np.random.RandomState(0)
X, y = make_classification(n_samples=8000, n_features=25, n_informative=10,
                           flip_y=0.0, class_sep=0.9, random_state=0)
Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.3, random_state=0)

def flip(labels, rate, rs):
    """随机翻转 rate 比例的【训练】标签，模拟标注噪声"""
    y2 = labels.copy()
    idx = rs.choice(len(y2), int(len(y2) * rate), replace=False)
    y2[idx] = 1 - y2[idx]                       # ⭐ 只污染训练集，测试集永远是干净的
    return y2

models = {
    "LR":  LogisticRegression(max_iter=2000),
    "RF":  RandomForestClassifier(n_estimators=300, random_state=0),
    "GBDT": GradientBoostingClassifier(random_state=0),
}
for rate in (0.0, 0.05, 0.10):
    ytr_noisy = flip(ytr, rate, np.random.RandomState(42))
    row = []
    for name, m in models.items():
        m.fit(Xtr, ytr_noisy)
        row.append(f"{name}={roc_auc_score(yte, m.predict_proba(Xte)[:, 1]):.4f}")
    print(f"标签噪声 {rate:>4.0%} | " + "  ".join(row))

import numpy as np
import pandas as pd

def first_30_minutes(df: pd.DataFrame, label: str = None, time_col: str = None):
    """接手一份新数据的最小体检。返回一张问题清单。"""
    issues = []
    n = len(df)
    for c in df.columns:
        s = df[c]
        miss = s.isna().mean()
        if miss > 0.3:
            issues.append((c, "高缺失", f"{miss:.1%} —— 先别填，去问为什么缺（第5章）"))
        nun = s.nunique(dropna=True)
        if nun <= 1:
            issues.append((c, "常量列", "整列只有一个值，通常是上游没接通"))
        if nun == n and n > 100:
            issues.append((c, "疑似ID", "取值全不重复，别当特征用"))
        if s.dtype == object:
            top = s.value_counts(dropna=True).head(1)
            if len(top) and top.iloc[0] / max(s.notna().sum(), 1) > 0.9:
                issues.append((c, "单值垄断", f"'{top.index[0]}' 占 {top.iloc[0]/n:.1%}"))
        else:
            vc = s.value_counts().head(1)
            # ⭐ 某个具体数值占比异常高 → 极可能是默认值伪装成真值（第4章）
            if len(vc) and vc.iloc[0] / max(s.notna().sum(), 1) > 0.2:
                issues.append((c, "疑似哨兵值", f"值 {vc.index[0]} 占 {vc.iloc[0]/n:.1%}"))
    dup = df.duplicated().sum()
    if dup:
        issues.append(("<整行>", "精确重复", f"{dup} 行（{dup/n:.1%}）—— 见第6章"))
    if label is not None and label in df:
        pos = df[label].mean()
        issues.append((label, "正样本率", f"{pos:.3%} —— 低于1%要重新想指标"))
        for c in df.select_dtypes(include=[np.number]).columns:
            if c == label:
                continue
            r = df[[c, label]].corr().iloc[0, 1]
            # ⭐ 单特征与标签相关性过高，几乎总是泄漏，不是你运气好（第14章）
            if abs(r) > 0.8:
                issues.append((c, "疑似泄漏", f"与标签相关系数 {r:.3f}"))
    if time_col is not None and time_col in df:
        t = pd.to_datetime(df[time_col], errors="coerce")
        issues.append((time_col, "时间跨度", f"{t.min()} → {t.max()}，坏值 {t.isna().mean():.1%}"))
        by_day = t.dt.date.value_counts().sort_index()
        if len(by_day) > 7:
            ratio = by_day.max() / max(by_day.median(), 1)
            if ratio > 3:
                issues.append((time_col, "日量突刺", f"最大日/中位日 = {ratio:.1f}，疑似重跑或补数"))
    return pd.DataFrame(issues, columns=["列", "问题", "说明"])

# 用法：first_30_minutes(df, label="is_bad", time_col="event_time")