实用脚本如何利用历史大数据建模预测？

wen 实用脚本 2026-10-21 2

实用脚本如何利用历史大数据建模预测？

这是一个涉及数据工程 + 特征工程 + 建模 + 部署的完整链路问题，下面我按"从脚本角度"给出可落地的思路和代码骨架。

实用脚本如何利用历史大数据建模预测？

整体链路（脚本视角）

历史大数据 → 数据清洗脚本 → 特征工程脚本 → 训练脚本 → 评估脚本 → 预测脚本 → 定时调度

关键原则：每个环节一个脚本，可复用、可重跑、可版本化。

数据准备脚本

目标：把原始大数据（日志/交易/传感器/业务库）转成建模可用的宽表。

# prep_data.py
import pandas as pd
from sqlalchemy import create_engine
def load_raw(engine, start, end):
    sql = f"""
    SELECT user_id, ts, amount, category, channel
    FROM events
    WHERE ts BETWEEN '{start}' AND '{end}'
    """
    return pd.read_sql(sql, engine)
def clean(df):
    df = df.drop_duplicates(subset=["user_id", "ts"])
    df["ts"] = pd.to_datetime(df["ts"])
    df = df[df["amount"] >= 0]
    return df
if __name__ == "__main__":
    engine = create_engine("postgresql://...")
    df = clean(load_raw(engine, "2022-01-01", "2024-12-31"))
    df.to_parquet("data/clean.parquet")   # 列式存储，后续读取快

要点：

用 Parquet 替代 CSV（体积小、读取快、保留类型）。
时间切分：训练/验证/测试 按时间切，不要随机切（防数据泄漏）。

特征工程脚本

历史大数据的价值 90% 在特征，常用模式：

# features.py
import pandas as pd
def build_features(df):
    df = df.sort_values(["user_id", "ts"])
    g = df.groupby("user_id")
    # 1) 滑窗统计（近7/30天行为）
    for w in [7, 30]:
        df[f"amt_sum_{w}d"] = g["amount"].transform(
            lambda s: s.rolling(f"{w}D", on=df.loc[s.index, "ts"]).sum()
        )
        df[f"cnt_{w}d"] = g["amount"].transform(
            lambda s: s.rolling(f"{w}D", on=df.loc[s.index, "ts"]).count()
        )
    # 2) 差分/趋势
    df["amt_diff"] = g["amount"].diff()
    df["days_since_last"] = g["ts"].diff().dt.days
    # 3) 类别编码
    df = pd.get_dummies(df, columns=["category", "channel"], drop_first=True)
    return df

要点：

滑窗特征务必只用当前时刻之前的数据（.shift(1) 或时间窗口严格左闭）。
类别型用 target encoding 或 hashing（高基数时）。
特征存成特征库（Feast / 自建 parquet），训练和线上共用一套逻辑。

训练与验证脚本

# train.py
import pandas as pd, joblib, mlflow
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import roc_auc_score
import lightgbm as lgb
df = pd.read_parquet("data/features.parquet")
X, y = df.drop(columns=["label", "ts", "user_id"]), df["label"]
tscv = TimeSeriesSplit(n_splits=5)
aucs = []
for tr, va in tscv.split(X):
    model = lgb.LGBMClassifier(n_estimators=500, learning_rate=0.05)
    model.fit(X.iloc[tr], y.iloc[tr])
    aucs.append(roc_auc_score(y.iloc[va], model.predict_proba(X.iloc[va])[:,1]))
print("CV AUC:", sum(aucs)/len(aucs))
# 全量重训 + 记录
final = lgb.LGBMClassifier(**model.get_params()).fit(X, y)
joblib.dump(final, "models/lgbm_v1.pkl")
with mlflow.start_run():
    mlflow.log_metric("cv_auc", sum(aucs)/len(aucs))
    mlflow.sklearn.log_model(final, "model")

要点：

一定是 TimeSeriesSplit，不是 KFold。
用 MLflow / DVC 管理模型和数据的版本。
评估指标要和业务对齐（AUC 只是排序能力，还要看 Recall@K、收益、成本）。

预测脚本（批 + 实时）

批量预测（每天跑）：

# predict_batch.py
import joblib, pandas as pd
from features import build_features
model = joblib.load("models/lgbm_v1.pkl")
def predict(date):
    raw = pd.read_parquet(f"data/raw/{date}.parquet")
    feat = build_features(raw)
    X = feat.drop(columns=["user_id","ts"], errors="ignore")
    feat["score"] = model.predict_proba(X)[:, 1]
    feat[["user_id","score"]].to_parquet(f"out/pred_{date}.parquet")
if __name__ == "__main__":
    import sys; predict(sys.argv[1])

实时预测：把 build_features 拆成在线特征服务（Redis 特征缓存 + 实时聚合），模型导出为 ONNX / PMML 供低延迟推理。

调度与监控脚本

# crontab 示例：每天 02:00 跑数据+训练，06:00 出预测
0 2 * * *  cd /opt/ml && python prep_data.py && python features.py
30 2 * * * cd /opt/ml && python train.py
0 6 * * *  cd /opt/ml && python predict_batch.py $(date +\%F)

监控脚本要监控三件事：

数据漂移（PSI / KS 检验特征分布）
预测漂移（输出分布变化）
效果衰减（有 label 后滞后评估 AUC / 命中率）

# monitor.py 片段
from scipy.stats import ks_2samp
def drift(train_col, live_col):
    return ks_2samp(train_col, live_col).pvalue < 0.01  # 触发告警

实用脚本的几条铁律

原则	说明
时间严格性	特征、切分、验证全程用时间序，禁未来信息
可复现	数据版本 + 代码版本 + 参数固定随机种子
脚本单一职责	prep / feat / train / predict / monitor 分开
特征复用	训练和线上用同一份 `build_features` 代码
回滚机制	模型带版本号，线上可一键切回上一版
小步迭代	先跑通 baseline（逻辑回归/单特征），再加复杂度

一个最小可运行骨架

ml_project/
├── config.yaml
├── data/{raw,clean,features}/
├── models/
├── out/
├── prep_data.py
├── features.py
├── train.py
├── predict_batch.py
├── monitor.py
└── run.sh        # 串联全流程

run.sh：

set -e
python prep_data.py
python features.py
python train.py
python predict_batch.py $(date +%F)
python monitor.py

这样一套脚本,就形成了 "历史大数据 → 特征 → 模型 → 预测 → 监控" 的闭环，可稳定按天/小时运行，并逐步迭代模型效果。

需要的话,我可以针对你的具体场景（比如销量预测 / 用户流失 / 风控评分 / 设备故障）给出更贴合的脚本模板和特征设计，你现在的历史数据大概是什么类型、什么规模？