模型预测前怎么验证表格输入？拦截缺列、错类型和越界值

表格模型能运行一次，不代表任意输入都可以直接预测。预测前应先核对必需列，再把非法类型、越界值和允许填补的缺失值分开处理；拒绝的原始行要保留原因，合格行进入模型时要保持原始行号。这样才能避免把错误字符串转成空值后悄悄填补，得到看似成功的结果。

本例制定一份演示契约：age 必须存在、非空、18–100；income 必须存在，数值范围为 0–10,000,000，允许单元格缺失。两列可接受可解析的数字字符串，布尔值、非数值字符串和无穷大拒绝。这些范围只用于演示，不能作为真实年龄、收入或业务决策标准。

代码已在 Windows、Python 3.11.15、pandas 3.0.6、NumPy 2.4.6、scikit-learn 1.9.1 的 CPU 环境运行。训练数据与标签是人为构造的六行样本，仅用于验证“先审核、再填补、最后预测”的调用顺序，没有实际预测业务含义。

缺列、缺失值和错类型要分别定义

pandas.to_numeric 文档规定 errors='coerce'会把无法解析的值设为 NaN。这里先保存原始缺失掩码，再做转换，才能区分原本缺失与转换失败。isna 文档说明，空字符串和无穷大通常不算 NA，因此还需要类型与有限性检查。

先在独立目录建立环境，以下固定为本文版本；macOS/Linux 激活环境使用 source .venv/bin/activate。

python -m venv .venv
.venv\Scripts\activate
python -m pip install pandas==3.0.6 numpy==2.4.6 scikit-learn==1.9.1

保留拒绝行，再预测合法行

把完整代码保存为 validate_input.py，执行 python validate_input.py。函数会返回标准化后的合格行和带错误原因的拒绝行；缺列与重复列名直接抛错，调用方应停止本次预测并返回明确的输入问题。

import numpy as np
import pandas as pd
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

RULES = {'age': (18, 100, False), 'income': (0, 10_000_000, True)}

def validate_input(frame):
    if not frame.columns.is_unique:
        raise ValueError('duplicate column names')
    missing_columns = sorted(set(RULES) - set(frame.columns))
    if missing_columns:
        raise ValueError(f'missing columns: {missing_columns}')
    reasons = [[] for _ in range(len(frame))]
    normalized = pd.DataFrame(index=frame.index)
    for column, (low, high, allow_missing) in RULES.items():
        original = frame[column]
        missing = original.isna()
        is_bool = original.map(lambda x: isinstance(x, (bool, np.bool_)))
        values = pd.to_numeric(original, errors='coerce').astype(float)
        illegal = (~missing) & (values.isna() | is_bool)
        nonfinite = (~missing) & (~illegal) & (~np.isfinite(values))
        outside = (~missing) & (~illegal) & (~nonfinite) & ((values < low) | (values > high))
        masks = [(illegal, 'invalid_numeric'), (nonfinite, 'nonfinite'),
                 (outside, 'out_of_range')]
        if not allow_missing:
            masks.append((missing, 'required_value_missing'))
        for mask, reason in masks:
            for pos in np.flatnonzero(mask.to_numpy()):
                reasons[pos].append(f'{column}:{reason}')
        normalized[column] = values
    valid = np.array([not item for item in reasons])
    rejected = frame.iloc[np.flatnonzero(~valid)].copy()
    rejected['validation_errors'] = [';'.join(reasons[pos]) for pos in np.flatnonzero(~valid)]
    accepted = normalized.iloc[np.flatnonzero(valid)].copy()
    return accepted, rejected

inputs = pd.DataFrame({
    'age': [28, 'abc', 17, 35, None, True, 45],
    'income': [None, 2000, 3000, 'oops', 4000, np.inf, 8000]
})
accepted, rejected = validate_input(inputs)
assert accepted.index.tolist() == [0, 6]
assert rejected.index.tolist() == [1, 2, 3, 4, 5]
assert pd.isna(accepted.loc[0, 'income'])
assert 'age:invalid_numeric' in rejected.loc[1, 'validation_errors']
assert 'age:out_of_range' in rejected.loc[2, 'validation_errors']
assert 'income:invalid_numeric' in rejected.loc[3, 'validation_errors']
assert 'age:required_value_missing' in rejected.loc[4, 'validation_errors']
assert 'income:nonfinite' in rejected.loc[5, 'validation_errors']
print('accepted_rows:', accepted.index.tolist())
print('rejected_rows:', rejected.index.tolist())
for index, row in rejected.iterrows():
    print('row', index, row['validation_errors'])
rejected.to_csv('rejected_input.csv', index_label='input_row', encoding='utf-8-sig')
try:
    validate_input(inputs.drop(columns='income'))
except ValueError as error:
    print('missing_column_rejected:', str(error))
else:
    raise AssertionError('Missing input column was accepted')
train = pd.DataFrame({'age': [22, 26, 34, 41, 50, 63],
                      'income': [2500, 3000, 4500, 6500, 8000, 9000]})
labels = np.array([0, 0, 0, 1, 1, 1])
model = make_pipeline(SimpleImputer(strategy='median'), StandardScaler(),
                      LogisticRegression(max_iter=500))
model.fit(train, labels)
print('training_income_median:', model.named_steps['simpleimputer'].statistics_[1])
predictions = pd.DataFrame({'prediction': model.predict(accepted)}, index=accepted.index)
assert predictions.index.tolist() == [0, 6]
print('predicted_only_accepted_rows:', predictions.index.tolist())
print('predictions:', predictions['prediction'].tolist())

检查输出是否遵守契约

本例实际输出如下。原始行号从 0 开始，0 与 6 通过，1–5 被拒绝。

accepted_rows: [0, 6]
rejected_rows: [1, 2, 3, 4, 5]
row 1 age:invalid_numeric
row 2 age:out_of_range
row 3 income:invalid_numeric
row 4 age:required_value_missing
row 5 age:invalid_numeric;income:nonfinite
missing_column_rejected: missing columns: ['income']
training_income_median: 5500.0
predicted_only_accepted_rows: [0, 6]
predictions: [0, 1]

行 2 的 age 为 17，被明确记为 age:out_of_range 并拒绝进入预测。

行 0 的 income 缺失是契约允许的情况，训练集收入中位数 5500 用来填补它。这个中位数来自训练阶段，而不是根据这批预测请求临时计算。SimpleImputer 官方文档说明 median 策略按列使用中位数填补；预测阶段沿用已经拟合的统计量。

行 1 的 abc 没有进入填补器；行 5 同时触发 age:invalid_numeric 和 income:nonfinite，两个原因都保留。删除 income 列的单独测试产生 missing columns 错误。最后仅对原始行 0 和 6 预测，输出与这两行的索引对应，没有为拒绝行编造预测。

查看当前目录下的 rejected_input.csv，应能找到原始输入行号、原值和 validation_errors。检查通过数加拒绝数是否为 7，预测行号是否与通过行一致。不要只看函数返回非空或终端没有异常就判断全批合格。

换成真实模型前要冻结哪些规则

1. 把 RULES 改成训练模型真正需要的字段名、单位、范围和空值策略；列顺序也要与训练一致。
2. 明确数字字符串是否可以接收。本例允许可解析的字符串，但若业务接口要求原生数值类型，就要先加严格类型校验。
3. 对 ID、枚举类别、日期、时间区间等建立独立规则，不能把它们统一转成浮点数。货币或高精度整数也应按业务精度协议处理。
4. 留存输入版本、拒绝原因和行号映射；如果行索引本身不唯一，另加稳定请求行编号。
5. 空批次或全部被拒绝时，返回明确状态并停止 model.predict；不要把零行结果标成成功完成全部预测。

这些代码只验证单行字段契约。它没有检测跨字段矛盾、输入分布漂移或对抗样本，也没有证明合法输入必然预测正确。输入合法性和模型效果需要各自的验证证据。