经典AI算法与编程实战：在喧嚣的生成式浪潮中，守住确定性根基

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline

# 模拟一份含噪声的电商用户数据
df = pd.DataFrame({
    'age': [25, 32, np.nan, 45, 29, 51],
    'income': [5000, 12000, 8000, 30000, 6000, 4500],
    'category': ['electronics', 'clothing', 'electronics', 'books', np.nan, 'clothing'],
    'click_rate': [0.02, 0.15, 0.08, 0.01, 0.22, 0.03],
    'is_purchase': [0, 1, 0, 1, 1, 0]  # 目标标签
})

# ---- 编程实战第一原则：永远在Pipeline里做变换，避免数据泄露 ----
numeric_features = ['age', 'income']
categorical_features = ['category']

# 数值特征：中位数填充 + 标准化
numeric_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# 类别特征：众数填充 + 简单编码 (实际中常用Target Encoding，但此处演示基础)
categorical_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('encoder', LabelEncoder())  # 注意：LabelEncoder需包裹为OneHotEncoder或Ordinal，此处仅为示意
])

# 组合预处理器 (实际生产推荐使用OneHotEncoder处理低基数类别)
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, numeric_features),
        ('cat', categorical_transformer, categorical_features)
    ])

import xgboost as xgb
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import cross_val_score

# 封装完整的Pipeline
model_pipeline = Pipeline(steps=[
    ('preprocessor', preprocessor),
    ('classifier', xgb.XGBClassifier(
        n_estimators=100,
        learning_rate=0.1,
        max_depth=5,
        subsample=0.8,
        colsample_bytree=0.8,
        eval_metric='logloss',
        use_label_encoder=False,
        random_state=42
    ))
])

# 编程实战黄金法则：先交叉验证看稳定性，再划分测试集
X = df.drop('is_purchase', axis=1)
y = df['is_purchase']

# 5折交叉验证的AUC均值，远胜于单次Hold-Out
cv_scores = cross_val_score(model_pipeline, X, y, cv=5, scoring='roc_auc')
print(f"交叉验证AUC均值: {np.mean(cv_scores):.4f} (+/- {np.std(cv_scores):.4f})")

# 确认稳定后，再拆分训练/测试做最终评估
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
model_pipeline.fit(X_train, y_train)
y_pred_proba = model_pipeline.predict_proba(X_test)[:, 1]
print(f"测试集AUC: {roc_auc_score(y_test, y_pred_proba):.4f}")

from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler

# 假设有用户行为特征矩阵
behavior_data = np.random.randn(1000, 5)  # 1000个用户，5维行为
scaler = StandardScaler()
scaled_data = scaler.fit_transform(behavior_data)

# DBSCAN 天然支持异常点检测（-1标签）
# eps（邻域半径）和min_samples是灵魂参数，需结合K距离图调试
clustering = DBSCAN(eps=0.5, min_samples=10).fit(scaled_data)
outlier_mask = clustering.labels_ == -1
print(f"检测到异常用户数: {np.sum(outlier_mask)}")
# 异常用户可能是机器流量或极端的价值用户，需业务侧重点分析

# 实战中最有效的过拟合防御：早停 + 验证集
model = xgb.XGBClassifier(
    n_estimators=1000,  # 开大，让早停决定何时停
    early_stopping_rounds=50,
    eval_set=[(X_val, y_val)],
    verbose=False
)

import joblib
# 保存
joblib.dump(model_pipeline, 'user_click_pipeline_v1.2.joblib')
# 加载（建议在Docker容器中锁定库版本）
loaded_model = joblib.load('user_click_pipeline_v1.2.joblib')

# FastAPI 推理接口的高效写法（伪代码）
@app.post("/predict")
async def predict(features: List[float]):  # 直接接收数组，而非复杂嵌套JSON
    import numpy as np
    arr = np.array(features).reshape(1, -1)
    # 如果Pipeline依赖列名，需用字典映射，此处假设预处理已外置
    proba = loaded_model.predict_proba(arr)[0][1]
    return {"risk_score": round(proba, 4)}