LSTM时间序列足球赛果概率预测：Elo特征与Dash看板

简介：SoccerPredictor 是一份面向有 Python 与机器学习基础读者的足球赛果预测项目源码，围绕英超比赛胜负及双倍机会投注，将预测任务建模为时间序列分类，并为每支球队构建神经网络模型联合训练，适合参考 TensorFlow、Keras、LSTM 与体育数据建模实践。压缩包共 29 个文件、约 2.41MB，以 14 个 py 脚本为核心，辅以 pickle 数据/模型、png 可视化图表、css 样式及 requirements、README 等说明，目录含 trainer、backtester、visualizer 等模块，便于按训练、回测与展示流程理解项目。已有 1258 人学习，作为体育预测与时间序列案例具备一定参考热度。读者可从中获取特征组织、模型训练、结果可视化与回测评估的完整实现思路，并借鉴模块划分搭建相似项目；需注意作者未发布数据集，当前更适合基于附带文件做可视化与回测演示，实际训练需自行准备数据。

1. 先泼一盆冷水：足球比赛结果预测到底能做到什么程度

周六晚上朋友丢过来一句「明天那场谁赢」，你要是回「五五开」，等于没答。SoccerPredictor 这类项目干的活，就是把这种口头猜测变成一条可复现的流水线：把历史比分表清洗成时间序列样本，喂进 Keras 里的 LSTM，输出主胜、平、客胜三个概率，再用 Dash 搭一块能点选球队的看板。它适合两类人：想把机器学习时间序列从教程挪到真实脏数据上练手的 Python 开发者，以及想给球队做赛前复盘分析的球迷工程师。

先把预期摆正。足球比赛结果预测的准确率天花板很低，多数类基线（永远猜主胜）在常见联赛里就有 45% 左右，模型能稳定跑到 52%～55% 已经算合格，而平局永远是玄学。它给的是概率分布，不是结论，冷门照样发生。想清楚这一点，后面所有工程细节才有意义——评估看 log loss 而不是准确率，特征只用赛前已知信息，回测按赛季滚动而不是随机切分。

2. 架构分层与目录：pandas 洗数据、Keras 建模、Dash 出图的三段式

2.1 四层结构：数据层、特征层、模型层、展示层

这套流水线拆开看只有四层，依赖方向严格单向，app/可以 importsrc/，反过来绝对不行。这条规矩听起来像教条，但它决定了你后面能不能把看板换掉、把模型换掉而不动数据层。

数据层负责把每个赛季一份的原始比分 CSV 收进来，统一列名、统一日期、统一队名写法，落成 parquet 缓存。特征层是一堆无状态的纯函数，输入一张按日期排序的比赛表，输出一张加了 Elo、近期进球、休息天数等列的宽表——所谓无状态，就是不许在函数里存全局变量，同一份输入必须永远得到同一份输出，否则回测时你会怀疑人生。模型层只做三件事：把宽表切成(样本, 时间步, 特征)的三维张量、训练、保存。展示层是 Dash，负责把一次推理的结果画成柱状图。

为什么是 LSTM 而不是直接上 LightGBM？说实话，在纯表格特征上树模型经常打平甚至更强，这是血泪经验。但项目的定位是时间序列建模：每支球队最近 N 场的状态是一个变长序列，LSTM 的共享编码器能天然处理「主队序列 + 客队序列」这种双输入结构，还能顺带输出比分回归。工程上我一般两个都留：LightGBM 当基线，LSTM 当主模型，看板上同时显示两条曲线，谁的效果差就把谁换掉。

2.2 目录结构与文件职责

SoccerPredictor/ ├── data/ │ ├── raw/ # 原始赛季比分 CSV，一个赛季一个文件 │ └── processed/ # 清洗后的 parquet，特征表缓存在这里 ├── src/ │ ├── config.py # 路径、赛季范围、窗口长度等常量 │ ├── clean.py # 队名归一、日期解析、标签生成 │ ├── features.py # Elo、滚动状态、休息天数 │ ├── dataset.py # 滑窗切样本、按时间切分训练/验证/测试 │ ├── model.py # Keras 双头 LSTM 定义 │ ├── train.py # 训练入口，产出 .keras 与 scaler.joblib │ └── predict.py # 单场/批量推理封装 ├── app/ │ ├── app.py # Dash 入口与回调 │ └── assets/style.css ├── models/ │ ├── lstm_two_head.keras │ └── preprocess.joblib ├── notebooks/ # 探索期的东西，别进生产 └── requirements.txt

predict.py这个文件值得单独说一句。看板、批量回测、离线验证全都调它，不许任何地方绕过它自己拼特征。我见过太多项目在app.py里又写了一遍特征逻辑，结果线上线下的窗口长度差一个，概率全歪，排查半天才发现是两套代码。

2.3 环境搭建与依赖版本：一次装干净的顺序

python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 锁小版本，Keras 2 和 Keras 3 的保存/加载行为不一样，混用必翻车 pip install "tensorflow==2.15.*" pandas numpy scikit-learn joblib pyarrow pip install "dash>=2.14" "plotly>=5.18" python -c "import tensorflow as tf, keras; print(tf.__version__, keras.__version__)"

tensorflow==2.15.*这个写法不是洁癖。2.16 之后默认捆绑 Keras 3，model.save("xxx.keras")和load_model的兼容行为变了，老代码直接 import 就报错。机器上没有独显也无所谓，这批数据量下 CPU 训练一轮 LSTM 也就几分钟，为它折腾 CUDA 不划算。

注意：先跑通python -m src.train --epochs 1再做任何改动。一轮训练能把数据读取、序列构造、模型编译三条链路的问题全部暴露出来，比在 Jupyter 里逐格试快得多。

3. 特征管道拆解：Elo 评分、滚动状态和滑窗样本怎么拼出来

3.1 原始比分表清洗：队名、日期、标签三件事

原始 CSV 只有日期、主客队名、全场进球数这些字段，脏的地方集中在三处：日期格式跨赛季不一致、同一支球队在不同文件里两种拼写、个别行缺进球数。

import pandas as pd RAW_COLS = ["Date", "HomeTeam", "AwayTeam", "FTHG", "FTAG"] ALIASES = {"Man United": "Man Utd", "Nott'm Forest": "Nottingham Forest"} def load_season(path: str) -> pd.DataFrame: df = pd.read_csv(path, usecols=lambda c: c in RAW_COLS) # dayfirst=True 对应 31/12/2023 这种写法，解析失败的行直接变 NaT，后面统一丢 df["Date"] = pd.to_datetime(df["Date"], dayfirst=True, errors="coerce") df = df.dropna(subset=["Date", "HomeTeam", "AwayTeam", "FTHG", "FTAG"]) df["FTHG"] = df["FTHG"].astype(int) df["FTAG"] = df["FTAG"].astype(int) for col in ("HomeTeam", "AwayTeam"): df[col] = df[col].str.strip().replace(ALIASES) # 标签自己算，不依赖文件里的结果列：2=主胜 1=平 0=客胜 df["label"] = (df["FTHG"] > df["FTAG"]).astype(int) * 2 \ + (df["FTHG"] == df["FTAG"]).astype(int) return df.sort_values("Date").reset_index(drop=True)

errors="coerce"是关键参数，它把解析不了的值变成NaT而不是抛异常，一个赛季几千行里只要有几行格式错，硬抛会让整条流水线卡住。ALIASES这张别名表看着土，但没有它，滚动特征会把同一支球队当成两支，主队近六场均值永远是空的。别问我怎么知道的。

标签的编码顺序直接对应模型 softmax 的输出下标：0客胜、1平、2主胜。后面画图的时候x=["客胜","平","主胜"]必须和这个顺序对齐，错一位，看板上显示的「主胜 62%」其实是客胜概率，这种翻车最难发现，因为数字本身看着很合理。

3.2 Elo 与滚动状态：赛前特征的唯一合法来源

Elo 的思路很朴素：每支球队一个分数，赛后按预期胜率和实际结果的差值更新。它天然满足「只用赛前信息」这条铁律——记录的是更新之前的分数。

import numpy as np def add_elo(df: pd.DataFrame, k: float = 20, home_adv: float = 65, base: float = 1500) -> pd.DataFrame: elo, rows = {}, [] for r in df.itertuples(): eh, ea = elo.get(r.HomeTeam, base), elo.get(r.AwayTeam, base) # 先落盘赛前评分，再更新，顺序反了就是把答案喂给模型 rows.append((eh, ea, eh - ea + home_adv)) exp_home = 1 / (1 + 10 ** ((ea - eh - home_adv) / 400)) score_home = 1.0 if r.FTHG > r.FTAG else (0.5 if r.FTHG == r.FTAG else 0.0) delta = k * (score_home - exp_home) elo[r.HomeTeam], elo[r.AwayTeam] = eh + delta, ea - delta return pd.DataFrame(rows, columns=["elo_h", "elo_a", "elo_diff"], index=df.index)

k=20是更新步长，越大越敏感、越容易追着最近几场跑；home_adv=65是主场加成，折算成约 0.65 个球的优势。这两个参数不用精调，量级对了就行，真正影响效果的是接下来的滚动状态。

def add_form(df: pd.DataFrame, window: int = 6) -> pd.DataFrame: df = df.copy() df["match_id"] = range(len(df)) # 把一场比赛拆成主队视角、客队视角两行，才能按球队做时间序列统计 home = pd.DataFrame({ "match_id": df.match_id, "team": df.HomeTeam, "date": df.Date, "gf": df.FTHG, "ga": df.FTAG, "is_home": 1, "pts": df.label.map({2: 3, 1: 1, 0: 0}), }) away = pd.DataFrame({ "match_id": df.match_id, "team": df.AwayTeam, "date": df.Date, "gf": df.FTAG, "ga": df.FTHG, "is_home": 0, "pts": df.label.map({2: 0, 1: 1, 0: 3}), }) long = pd.concat([home, away]).sort_values(["team", "date"]).reset_index(drop=True) g = long.groupby("team") # shift(1) 一定不能漏：当前这场的比分不能出现在自己的赛前特征里 for col in ("gf", "ga", "pts"): long[f"{col}_roll"] = g[col].transform( lambda s: s.shift(1).rolling(window, min_periods=1).mean()) long["rest_days"] = g["date"].transform( lambda s: s.diff().dt.days.clip(upper=30)).fillna(14) keep = ["match_id", "gf_roll", "ga_roll", "pts_roll", "rest_days"] h = long[long.is_home == 1][keep].add_prefix("h_").rename(columns={"h_match_id": "match_id"}) a = long[long.is_home == 0][keep].add_prefix("a_").rename(columns={"a_match_id": "match_id"}) return df.merge(h, on="match_id").merge(a, on="match_id")

min_periods=1让升班马的前几场也有值，代价是这些值参考性弱，所以后面构造样本时会按历史场次过滤。clip(upper=30)是把休赛期截断——赛季末到赛季初隔了三个月，不截断的话「休息 96 天」会变成异常值把标准化拉歪。

3.3 滑窗切样本与按时间切分：别让模型偷看未来

模型不吃宽表，吃序列。每支球队最近 10 场的状态是一条(10, 5)的序列，主队一条、客队一条，再拼几个静态特征。

TEAM_FEATURES = ["gf", "ga", "pts", "rest_days", "is_home"] def team_series(df: pd.DataFrame, timesteps: int = 10) -> dict: """返回 {(team, match_id): (T, 5)}，只包含该场之前的历史""" seqs, buf = {}, {} for r in df.itertuples(): hist = buf.setdefault(r.HomeTeam if False else r.team, []) # 见下方长表构造 ...

上面这段是伪代码，真实实现要先摊成长表再遍历，完整版是这样：

def team_series(df: pd.DataFrame, timesteps: int = 10) -> dict: df = df.copy() df["match_id"] = range(len(df)) home = pd.DataFrame({"match_id": df.match_id, "team": df.HomeTeam, "date": df.Date, "gf": df.FTHG, "ga": df.FTAG, "is_home": 1, "pts": df.label.map({2: 3, 1: 1, 0: 0})}) away = pd.DataFrame({"match_id": df.match_id, "team": df.AwayTeam, "date": df.Date, "gf": df.FTAG, "ga": df.FTHG, "is_home": 0, "pts": df.label.map({2: 0, 1: 1, 0: 3})}) long = pd.concat([home, away]).sort_values(["team", "date", "match_id"]) seqs, buf = {}, {} for r in long.itertuples(): hist = buf.setdefault(r.team, []) # 先取窗口，再把本场结果压进缓冲：顺序反了就是数据泄漏 seqs[(r.team, r.match_id)] = ( np.asarray(hist[-timesteps:], dtype="float32") if len(hist) >= timesteps else None) hist.append([r.gf, r.ga, r.pts, 0.0, float(r.is_home)]) return seqs

len(hist) >= timesteps这个判断会丢掉每支球队的最前面若干场，联赛前几轮样本会少一大截，这是必须付的代价。返回None的位置在组装阶段整条丢弃，别用零填充糊过去——零填充会让模型以为这支球队最近十场一球未进。

def build_xy(df: pd.DataFrame, timesteps: int = 10): seqs = team_series(df, timesteps) df = df.assign(match_id=range(len(df))) Xh, Xa, Xs, y_cls, y_goals = [], [], [], [], [] for r in df.itertuples(): sh = seqs.get((r.HomeTeam, r.match_id)) sa = seqs.get((r.AwayTeam, r.match_id)) if sh is None or sa is None: continue Xh.append(sh); Xa.append(sa) Xs.append([r.elo_diff, r.h_rest_days - r.a_rest_days]) y_cls.append(r.label); y_goals.append([r.FTHG, r.FTAG]) return (np.stack(Xh), np.stack(Xa), np.asarray(Xs, "float32"), np.asarray(y_cls), np.asarray(y_goals, "float32"))

切分绝不能 shuffle。按日期排序后按比例切，训练集在前、验证集中间、测试集最后：

def time_split(df: pd.DataFrame, ratios=(0.7, 0.15, 0.15)): df = df.sort_values("Date").reset_index(drop=True) n, i, j = len(df), int(len(df) * ratios[0]), int(len(df) * sum(ratios[:2])) return df.iloc[:i], df.iloc[i:j], df.iloc[j:] train, val, test = time_split(features) # 标准化器只在训练集上 fit，漏了这一步测试集信息会渗进训练 scaler = StandardScaler().fit(train[["elo_diff", "h_rest_days", "a_rest_days"]])

注意：见过太多人用train_test_split(shuffle=True)切时间序列，验证集准确率能凭空高出十个点，然后上线一塌糊涂。时间序列的验证集必须比训练集「晚」。

4. 双头 LSTM 模型：一个主干同时预测胜平负和比分

4.1 共享 LSTM 主干 + 双输出的网络结构

两支球队的近期序列结构完全一样，没必要各训一个 LSTM。同一个LSTM(32)分别编码主队和客队，得到两个 32 维向量，再和静态特征拼起来进全连接层。这样参数量少一半，还能避免主客队学出两套不一致的标准。

import tensorflow as tf from tensorflow.keras import layers, Model, Input def build_model(timesteps: int, team_dim: int, static_dim: int, n_classes: int = 3): home_in = Input(shape=(timesteps, team_dim), name="home_seq") away_in = Input(shape=(timesteps, team_dim), name="away_seq") static_in = Input(shape=(static_dim,), name="static") # 共享编码器：主客队用同一层，参数复用 encoder = layers.LSTM(32, dropout=0.2) h = encoder(home_in) # (batch, 32) a = encoder(away_in) # (batch, 32) x = layers.Concatenate()([h, a, layers.Subtract()([h, a]), static_in]) x = layers.Dense(64, activation="relu")(x) x = layers.Dropout(0.3)(x) result = layers.Dense(n_classes, activation="softmax", name="result")(x) goals = layers.Dense(2, activation="softplus", name="goals")(x) # 进球非负 return Model([home_in, away_in, static_in], [result, goals])

layers.Subtract()显式把两队编码的差值塞进去，让网络直接看到「谁状态更好」这个一阶信号，收敛会快一些。softplus而不是relu作进球输出，是因为relu在 0 附近梯度为 0，进球数恰好为 0 的样本（比如 0:0）容易把输出压死。

4.2 训练配置、损失函数与基线对比

胜平负用稀疏交叉熵，比分用 Poisson 损失。进球数近似服从泊松分布，用 Poisson 比 MSE 更贴合，也让输出的量纲自然落在「期望进球」上。

model = build_model(timesteps=10, team_dim=5, static_dim=2) model.compile( optimizer=tf.keras.optimizers.Adam(1e-3), loss={"result": "sparse_categorical_crossentropy", "goals": "poisson"}, loss_weights={"result": 1.0, "goals": 0.3}, # 分类为主，回归当辅助任务 metrics={"result": ["accuracy"]}, ) callbacks = [ tf.keras.callbacks.EarlyStopping(monitor="val_loss", patience=5, restore_best_weights=True), tf.keras.callbacks.ReduceLROnPlateau(monitor="val_loss", factor=0.5, patience=2), ] model.fit( [Xh_train, Xa_train, Xs_train], {"result": y_cls_train, "goals": y_goals_train}, validation_data=([Xh_val, Xa_val, Xs_val], {"result": y_cls_val, "goals": y_goals_val}), epochs=60, batch_size=64, callbacks=callbacks, verbose=2, )

loss_weights里把回归头的权重压到 0.3，是因为比分回归的噪声比胜负大得多，权重给高了会把分类头带偏。EarlyStopping的patience=5在这批数据量下比较合适，再小会早停在第 6、7 轮，模型还没收敛。

评估别只看准确率：

from sklearn.metrics import log_loss, accuracy_score proba = model.predict([Xh_test, Xa_test, Xs_test])[0] print("log_loss", round(log_loss(y_cls_test, proba, labels=[0, 1, 2]), 4)) print("acc", round(accuracy_score(y_cls_test, proba.argmax(1)), 4)) print("baseline", round((y_cls_test == 2).mean(), 4)) # 全猜主胜

基线通常 0.44～0.47，模型能到 0.50～0.54 就有实用价值，但更该盯的是 log loss：一个准确率 0.52 但概率全挤在 0.34/0.33/0.33 的模型，等于没预测。合格的模型平局概率应该落在 0.22～0.30 区间浮动，而不是永远接近 0。

4.3 模型持久化与单场推理封装

训练完落两个文件：模型本体和预处理包。预处理包里带 scaler 和特征清单，缺一个都会在推理时对不上。

import joblib model.save("models/lstm_two_head.keras") joblib.dump({"scaler": scaler, "team_features": TEAM_FEATURES, "timesteps": 10}, "models/preprocess.joblib")

推理入口统一收在src/predict.py，对外只暴露一个函数：

def predict_match(home: str, away: str, date, feats, bundle) -> dict: """feats 是截至 date 的完整特征表，内部截断、构序列、标准化、前向一次""" past = feats[feats.Date < date] seqs = team_series(past, bundle["timesteps"]) # 复用训练期同一套逻辑 xh, xa, xs = assemble_one(home, away, past, seqs, bundle) proba, goals = model([xh, xa, xs], training=False) # 关键：training=False return {"away": float(proba[0][0]), "draw": float(proba[0][1]), "home": float(proba[0][2]), "score": goals[0].numpy().round(1).tolist()}

training=False这个参数看着不起眼，但漏掉它，Dropout 层在推理时还在随机丢神经元，同一场比赛连点两次按钮得到两组概率，用户第一反应就是「这系统坏了」。

5. 看板联调避坑：Dash 回调跑通后必须过的五道检查

5.1 一块看板的最小结构：布局、回调、复用同一套特征

Dash 的好处是不用写前端工程，一个文件就能出交互页面。布局部分是声明式的，回调部分用装饰器把输入输出连起来。

from dash import Dash, dcc, html, Input, Output import plotly.graph_objects as go app = Dash(__name__) teams = sorted(set(feats.HomeTeam) | set(feats.AwayTeam)) app.layout = html.Div([ html.H2("赛前预测面板"), dcc.Dropdown(id="home", options=teams, value=teams[0], clearable=False), dcc.Dropdown(id="away", options=teams, value=teams[1], clearable=False), dcc.DatePickerSingle(id="date", date=str(feats.Date.max().date())), html.Button("预测", id="run", n_clicks=0), dcc.Graph(id="prob-bar"), html.Div(id="score-line"), ]) @app.callback( Output("prob-bar", "figure"), Output("score-line", "children"), Input("run", "n_clicks"), Input("home", "value"), Input("away", "value"), Input("date", "date"), prevent_initial_call=True, ) def run_prediction(_, home, away, date): if home == away: return go.Figure(), "主客队不能是同一支" res = predict_match(home, away, pd.Timestamp(date), feats, bundle) fig = go.Figure(go.Bar( x=["客胜", "平", "主胜"], # 顺序必须和 label 编码对齐 y=[res["away"], res["draw"], res["home"]], marker_color=["#7f8c8d", "#f39c12", "#27ae60"])) fig.update_layout(yaxis_range=[0, 1], yaxis_title="概率") return fig, f"预期比分 {res['score'][0]:.1f} : {res['score'][1]:.1f}" if __name__ == "__main__": app.run(debug=True, port=8050, host="127.0.0.1")

回调里的三个Input用的是Input不是State，区别在于前者一变就触发、后者只在点按钮时才读。这段代码里两者混着用是故意的：比赛日期改了应该立刻重算，但如果每次都重算太慢，就把耗时的部分挪进按钮分支。

5.2 五条踩坑记录：从概率不变到内存上涨

一、验证集准确率 0.68，上线随便一测连 0.4 都不到。现象是训练日志漂亮得不像话，val_accuracy一路爬到 0.68。原因是切分用了随机 shuffle，加上滚动特征漏了shift(1)，模型等于开卷考试。解决是把time_split换成按日期切，所有rolling前面强制加 `shift(

本文还有配套的精品资源，点击获取