核对特征数据是否稳定

改两个路径直接跑：BASELINE_PATH 是上一波/基准数据，CURRENT_PATH 是这波数据。输出字段级明细和跳过清单。如果某字段类型疑似从数值型变成非数值，会进跳过清单，这本身就是上游可能改分箱规则的信号。

import pandas as pd
import numpy as np

# 改成你的文件路径：基线=上一波数据，当前=这波数据
BASELINE_PATH = "baseline.csv"
CURRENT_PATH = "current.csv"
OUTPUT_PATH = "numeric_feature_stability_check.csv"
SKIP_OUTPUT_PATH = "skipped_fields.csv"

def read_df(path):
    if isinstance(path, pd.DataFrame):
        return path.copy()
    p = str(path)
    if p.endswith(".parquet"):
        return pd.read_parquet(p)
    if p.endswith(".xlsx"):
        return pd.read_excel(p)
    return pd.read_csv(p)

def series_metrics(s):
    s = pd.to_numeric(s, errors="coerce")
    n = len(s)
    non_null = int(s.notna().sum())
    null_cnt = int(s.isna().sum())

    coverage = non_null / n if n else np.nan
    null_rate = null_cnt / n if n else np.nan

    if non_null == 0:
        return coverage, null_rate, np.nan, np.nan, np.nan

    return (
        coverage,
        null_rate,
        s.quantile(0.10),
        s.quantile(0.50),
        s.quantile(0.90),
    )

def pct_change(cur, base):
    if pd.isna(cur) or pd.isna(base):
        return np.nan
    if base == 0:
        return 0.0 if cur == 0 else np.inf
    return (cur - base) / abs(base)

def compare_numeric_features(
    baseline,
    current,
    coverage_delta_threshold=0.05,   # 覆盖率/空值率变化超过5个百分点算异常
    quantile_pct_threshold=0.10,     # 分位数相对变化超过10%算异常
):
    baseline = read_df(baseline)
    current = read_df(current)

    rows = []
    skipped = []
    all_cols = list(dict.fromkeys(list(baseline.columns) + list(current.columns)))

    for col in all_cols:
        if col not in baseline.columns or col not in current.columns:
            skipped.append((col, "字段只在单侧存在，跳过"))
            continue

        base_raw = baseline[col]
        curr_raw = current[col]
        base_is_num = pd.api.types.is_numeric_dtype(base_raw)
        curr_is_num = pd.api.types.is_numeric_dtype(curr_raw)

        # 本任务只核对数值型字段
        if not base_is_num and not curr_is_num:
            skipped.append((col, "非数值字段，按要求跳过"))
            continue

        base_series = pd.to_numeric(base_raw, errors="coerce")
        curr_series = pd.to_numeric(curr_raw, errors="coerce")

        # 如果某侧原生不是数值，强制转数值后有效比例又太低，大概率是类型已经变了
        if not base_is_num and base_series.notna().mean() < 0.5:
            skipped.append((col, "基线侧疑似从数值型变成非数值/字符串，跳过"))
            continue
        if not curr_is_num and curr_series.notna().mean() < 0.5:
            skipped.append((col, "当前侧疑似从数值型变成非数值/字符串，跳过"))
            continue

        base_cov, base_null, base_p10, base_p50, base_p90 = series_metrics(base_series)
        curr_cov, curr_null, curr_p10, curr_p50, curr_p90 = series_metrics(curr_series)

        # 某侧全空，无法计算分位数，跳过
        if pd.isna(base_p50) or pd.isna(curr_p50):
            skipped.append((col, "某侧全空，无法计算分位数，跳过"))
            continue

        coverage_delta = curr_cov - base_cov
        null_rate_delta = curr_null - base_null
        p10_delta = curr_p10 - base_p10
        p50_delta = curr_p50 - base_p50
        p90_delta = curr_p90 - base_p90

        p10_pct = pct_change(curr_p10, base_p10)
        p50_pct = pct_change(curr_p50, base_p50)
        p90_pct = pct_change(curr_p90, base_p90)

        flags = []
        if abs(coverage_delta) > coverage_delta_threshold:
            flags.append("覆盖率变化大")
        if abs(null_rate_delta) > coverage_delta_threshold:
            flags.append("空值率变化大")
        if abs(p10_pct) > quantile_