基于Flask搭建回归预测服务：如何对新输入数据复现训练阶段的特征工程？

这个问题太常见了——很多人训练模型时只想着存模型，却忘了特征工程的“上下文”，导致新数据处理逻辑和训练时不一致，最终预测结果完全不可靠。别担心，我们可以通过封装特征工程逻辑+保存拟合参数的方式完美解决，具体步骤如下：

1. 把特征工程逻辑封装成可复用的组件 #

你需要把训练时从原始20列到最终35列特征的所有操作，写成一个可重复调用的类（推荐用类似sklearn的fit/transform模式，这样和机器学习生态兼容）。比如：

import pickle
import numpy as np
from sklearn.preprocessing import OneHotEncoder, PowerTransformer
import pandas as pd

class FeatureEngineer:
    def __init__(self):
        # 根据你的实际训练流程，定义需要处理的列和组件
        self.log_cols = ["col1", "col2"]  # 要做对数变换的列
        self.boxcox_cols = ["col3", "col4"]  # 要做Box-Cox变换的列
        self.onehot_cols = ["col5", "col6"]  # 要做One-Hot编码的列
        self.drop_cols = ["col7", "col8"]  # 因多重共线性删除的列
        self.boxcox_transformer = PowerTransformer(method='box-cox')
        self.onehot_encoder = OneHotEncoder(sparse_output=False, handle_unknown='ignore')
        self.train_fill_values = None  # 保存缺失值填充的均值/中位数
    
    def fit(self, df):
        # 用训练数据拟合所有需要的参数
        # 1. 处理缺失值（和训练时逻辑一致）
        self.train_fill_values = df[self.log_cols + self.boxcox_cols].median()
        df = df.fillna(self.train_fill_values)
        
        # 2. 拟合Box-Cox变换参数
        self.boxcox_transformer.fit(df[self.boxcox_cols])
        
        # 3. 拟合One-Hot编码的类别
        self.onehot_encoder.fit(df[self.onehot_cols])
        
        return self
    
    def transform(self, df):
        # 对新数据执行完全一致的特征工程
        # 1. 缺失值填充（复用训练时的填充值）
        df = df.fillna(self.train_fill_values)
        
        # 2. 对数变换（用log1p避免0值报错）
        df[self.log_cols] = np.log1p(df[self.log_cols])
        
        # 3. Box-Cox变换（复用训练时拟合的lambda值）
        df[self.boxcox_cols] = self.boxcox_transformer.transform(df[self.boxcox_cols])
        
        # 4. One-Hot编码
        onehot_df = pd.DataFrame(
            self.onehot_encoder.transform(df[self.onehot_cols]),
            columns=self.onehot_encoder.get_feature_names_out(self.onehot_cols),
            index=df.index
        )
        df = pd.concat([df, onehot_df], axis=1)
        
        # 5. 删除指定列（原分类列也要删掉）
        df = df.drop(columns=self.drop_cols + self.onehot_cols)
        
        # 确保列顺序和训练时完全一致（关键！模型输入顺序不能乱）
        expected_cols = pickle.load(open("expected_feature_order.pkl", "rb"))
        df = df[expected_cols]
        
        return df

2. 重新拟合并保存特征工程组件 #

用你的训练原始数据重新跑一遍fit方法，把拟合好的特征工程组件和特征顺序保存下来：

# 加载训练用的原始数据
train_df = pd.read_csv("train_raw_data.csv")

# 拟合特征工程组件
fe = FeatureEngineer()
fe.fit(train_df)

# 保存拟合后的组件
pickle.dump(fe, open("feature_engineer.pkl", "wb"))

# 保存最终特征的列顺序（保证和模型训练时的输入顺序一致）
processed_train_df = fe.transform(train_df)
pickle.dump(list(processed_train_df.columns), open("expected_feature_order.pkl", "wb"))

3. 在Flask服务中复用特征工程 #

现在你的Flask服务需要同时加载feature_engineer.pkl、expected_feature_order.pkl和训练好的模型，流程如下：

from flask import Flask, request, jsonify
import pickle
import pandas as pd

app = Flask(__name__)

# 加载预保存的组件
fe = pickle.load(open("feature_engineer.pkl", "rb"))
model = pickle.load(open("trained_regression_model.pkl", "rb"))

@app.route("/predict", methods=["POST"])
def predict():
    # 接收原始20列数据（JSON格式）
    raw_data = request.get_json()
    df = pd.DataFrame([raw_data])
    
    # 执行和训练时完全一致的特征工程
    processed_features = fe.transform(df)
    
    # 输入模型预测
    prediction = model.predict(processed_features)
    
    return jsonify({"prediction": float(prediction[0])})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

关键注意事项 #

• 参数绝对不能重新拟合：所有依赖训练数据的参数（比如Box-Cox的lambda值、One-Hot的类别、缺失值填充值）必须复用训练时的，绝对不能在预测阶段重新拟合！
• 处理未知类别：One-Hot编码时设置handle_unknown='ignore'，避免新数据出现训练时没有的类别导致服务报错。
• 列顺序必须严格匹配：模型的输入特征顺序和训练时必须完全一致，否则预测结果会完全错误。
• 异常值处理：如果训练时对异常值做了截断/替换，记得把相关阈值也保存下来，预测时执行同样的操作。

内容的提问来源于stack exchange，提问作者chink