太好了，你这套 【预测前的输入数据质检】 的思路是非常专业、非常有必要的！ 我帮你把这个需求整理成一个可落地的、可执行的标准方案，并给你示例代码示意，方便你直接落地到你的多输入多输出场景。
✅ 1️⃣ 你的检测目标拆解
你要做的：
| 检测项 | 要点 | 常用做法 |
|---|---|---|
| 1. 缺失值检测 | 每个必需列不能缺失 | 列表比对 + Null 检查 |
| 2. 嵌入向量值范围检测 | 如果有离散变量做过embedding，检测新类别是否超训练集 | 比如 Gender 原本只有 {M, F}，来了 X 就要报错或用默认向量 |
| 3. 归一化/标准化值范围检测 | 预测输入是否超出训练分布 | 保存 min/max（归一化）或 z-score clip（标准化） |
✅ 2️⃣ 【输入列完整性检测】
训练时必备列列表：
# 假设你的特征列
REQUIRED_COLUMNS = ['age', 'salary', 'city', 'industry']
# 预测输入
missing_cols = [col for col in REQUIRED_COLUMNS if col not in X_test.columns]
if missing_cols:
raise ValueError(f"⚠️ 缺少列: {missing_cols}")
# 检查是否有 NaN
nan_cols = X_test.columns[X_test.isnull().any()].tolist()
if nan_cols:
raise ValueError(f"⚠️ 存在缺失值: {nan_cols}")
✅ 3️⃣ 【嵌入向量范围检测】
如果你的某个分类列是需要做 embedding 的，比如 city，那要确保预测时的新类别存在于训练时保存的词表里。
示例：
# 假设训练时保存了 city 的词表
trained_city_vocab = {'New York', 'LA', 'Chicago'}
# 预测时输入
new_cities = set(X_test['city'].unique()) - trained_city_vocab
if new_cities:
raise ValueError(f"⚠️ city 列存在未知类别: {new_cities}")
✅ 4️⃣ 【归一化/标准化分布检测】
4.1 如果是 MinMaxScaler：
保存训练时每个特征的 min/max：
scaler_params = {
'salary': {'min': 3000, 'max': 15000},
'age': {'min': 20, 'max': 60},
}
检测是否超出：
for col, stats in scaler_params.items():
val_min, val_max = X_test[col].min(), X_test[col].max()
if val_min < stats['min'] or val_max > stats['max']:
print(f"⚠️ 特征 {col} 超出训练范围 [{stats['min']}, {stats['max']}], "
f"当前输入范围 [{val_min}, {val_max}]")
4.2 如果是 StandardScaler：
你保存了 mean 和 std，同时也要保存原始 min/max，用来检测：
scaler_params = {
'salary': {'mean': 8000, 'std': 2000, 'min': 3000, 'max': 15000},
}
for col, stats in scaler_params.items():
raw_min, raw_max = X_test[col].min(), X_test[col].max()
if raw_min < stats['min'] or raw_max > stats['max']:
print(f"⚠️ 特征 {col} 超出训练分布")
然后对标准化后的 z-score 也可以 clip：
X_scaled = (X_test - mean) / std
X_scaled = X_scaled.clip(-3, 3)
✅ 5️⃣ 【整合版完整流程示例】
下面是一个可执行的结构化示例，展示多输入多输出 + 嵌入 + 缺失值 + 分布检测怎么放到一起👇
import pandas as pd
import numpy as np
def validate_inputs(X_test: pd.DataFrame,
required_columns: list,
embedding_vocab: dict,
scaler_params: dict):
# 缺失列检测
missing_cols = [col for col in required_columns if col not in X_test.columns]
if missing_cols:
raise ValueError(f"⚠️ 缺少列: {missing_cols}")
# 缺失值检测
nan_cols = X_test.columns[X_test.isnull().any()].tolist()
if nan_cols:
raise ValueError(f"⚠️ 存在缺失值: {nan_cols}")
# 嵌入词表检测
for col, vocab in embedding_vocab.items():
new_vals = set(X_test[col].unique()) - vocab
if new_vals:
raise ValueError(f"⚠️ 特征 {col} 有未知类别: {new_vals}")
# 归一化/标准化范围检测
for col, stats in scaler_params.items():
raw_min, raw_max = X_test[col].min(), X_test[col].max()
if raw_min < stats['min'] or raw_max > stats['max']:
print(f"⚠️ 特征 {col} 输入超出训练分布 [{stats['min']}, {stats['max']}], "
f"当前 [{raw_min}, {raw_max}]")
# ========== 使用示例 ==========
# 必需列
REQUIRED_COLUMNS = ['age', 'salary', 'city']
# 嵌入词表示例
EMBEDDING_VOCAB = {
'city': {'New York', 'LA', 'Chicago'}
}
# 归一化参数示例
SCALER_PARAMS = {
'age': {'min': 20, 'max': 60},
'salary': {'min': 3000, 'max': 15000},
}
# 假设你的预测数据
X_test = pd.DataFrame({
'age': [25, 70], # 70超出
'salary': [8000, 20000], # 20000超出
'city': ['LA', 'Paris'] # Paris未知
})
validate_inputs(X_test, REQUIRED_COLUMNS, EMBEDDING_VOCAB, SCALER_PARAMS)
✅ 6️⃣ 总结 — 上线预测前的必做 3 件事
| 检测项 | 怎么做 | 是否必须 |
|---|---|---|
| 缺失值 / 缺列 | 列表比对 + isnull 检查 | 必须 |
| 嵌入词表检测 | 训练时保存类别集合 | 建议 |
| 分布范围检测 | 保存 min/max，做 clip 或告警 | 必须 |
##
