示例成品 · 平台演示，按左边这组点选真跑出来的
改两个路径直接跑：`BASELINE_PATH` 是上一波/基准数据，`CURRENT_PATH` 是这波数据。输出字段级明细和跳过清单。如果某字段类型疑似从数值型变成非数值，会进跳过清单，这本身就是上游可能改分箱规则的信号。
```python
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_
点左边「开工 · 直接出成品」，出一份你自己的版本（文字免费）