feat(ml): add ensemble XGBoost+LightGBM+MLP with Optuna optimization

- train_ensemble.py: full training pipeline with 100-trial Optuna studies
  for XGBoost and LightGBM, MLP (256-128-64), SHAP feature selection,
  weighted soft-voting ensemble, benchmark report generation
- predict_v2.py: production prediction module with model cache invalidation
- combined_api.py: add /api/v1/predictions, /api/v1/model/status,
  /api/v1/model/invalidate-cache endpoints using ensemble model
- tests/test_ml_ensemble.py: regression, latency and API tests

Baseline XGBoost Precision@3: 0.5287 (holdout 20% temporal)
Deploy threshold: +5% = 0.5551

Co-Authored-By: Paperclip <noreply@paperclip.ing>

combined_api.py

@@ -3519,7 +3519,6 @@ def brave_search():
         return jsonify({"error": str(e)}), 500
 
 
-
 @app.route("/turf/api/predictions_analysis", methods=["GET"])
 def api_predictions_analysis():
     """Analyse des predictions vs resultats reels"""
@@ -3533,8 +3532,20 @@ def api_predictions_analysis():
     cursor = conn.cursor()
 
     stats = {
-        "canalturf": {"total": 0, "top1_pct": 0, "top3_pct": 0, "top5_pct": 0, "ze2_pct": 0},
+        "canalturf": {
-        "scoring":   {"total": 0, "top1_pct": 0, "top3_pct": 0, "top5_pct": 0, "ze2_pct": 0},
+            "total": 0,
+            "top1_pct": 0,
+            "top3_pct": 0,
+            "top5_pct": 0,
+            "ze2_pct": 0,
+        },
+        "scoring": {
+            "total": 0,
+            "top1_pct": 0,
+            "top3_pct": 0,
+            "top5_pct": 0,
+            "ze2_pct": 0,
+        },
     }
 
     for source in ["canalturf", "scoring"]:
@@ -3585,5 +3596,219 @@ def api_predictions_analysis():
     return jsonify({"stats": stats, "period": {"start": start_date, "end": end_date}})
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# /api/v1/predictions  — Ensemble model endpoint (Sprint 6-7 ML Upgrade)
+# ─────────────────────────────────────────────────────────────────────────────
+_predict_v2 = None
+
+
+def _load_predict_v2():
+    """Lazy import of predict_v2 module (ensemble model)."""
+    global _predict_v2
+    if _predict_v2 is None:
+        try:
+            import importlib.util, sys
+
+            spec = importlib.util.spec_from_file_location(
+                "predict_v2", "/home/h3r7/turf_saas/predict_v2.py"
+            )
+            mod = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(mod)
+            _predict_v2 = mod
+        except Exception as e:
+            import logging
+
+            logging.error(f"[v1/predictions] predict_v2 import failed: {e}")
+    return _predict_v2
+
+
+@app.route("/api/v1/predictions", methods=["GET"])
+@app.route("/turf/api/v1/predictions", methods=["GET"])
+def api_v1_predictions():
+    """
+    Ensemble ML predictions using XGBoost + LightGBM + MLP (Optuna-tuned).
+    Query params:
+      - date: YYYY-MM-DD (default: today / latest available)
+      - reunion: int (default: all)
+      - course: int (default: all)
+    """
+    import time as _time
+
+    t0 = _time.perf_counter()
+
+    mod = _load_predict_v2()
+    if mod is None:
+        # Graceful fallback: redirect to legacy ml_predictions
+        return jsonify(
+            {
+                "error": "Ensemble model not available yet",
+                "fallback": "/api/ml_predictions",
+                "message": "Model is still training. Use /api/ml_predictions for legacy XGBoost predictions.",
+            }
+        ), 503
+
+    ensemble = mod.load_ensemble()
+    if ensemble is None:
+        return jsonify(
+            {
+                "error": "Ensemble model file not found",
+                "model_path": str(mod.ENSEMBLE_PATH),
+                "message": "Run train_ensemble.py to generate the model.",
+                "fallback": "/api/ml_predictions",
+            }
+        ), 503
+
+    date_param = request.args.get("date", None)
+    reunion_param = request.args.get("reunion", None)
+    course_param = request.args.get("course", None)
+
+    conn = sqlite3.connect("/home/h3r7/turf_saas/turf.db")
+    conn.row_factory = sqlite3.Row
+
+    # Determine date to use
+    if date_param:
+        date_used = date_param
+    else:
+        row = conn.execute(
+            "SELECT MAX(date_programme) as d FROM pmu_partants"
+        ).fetchone()
+        date_used = (
+            row["d"] if row and row["d"] else datetime.now().strftime("%Y-%m-%d")
+        )
+
+    # Build query
+    where_clauses = ["p.date_programme = ?"]
+    params = [date_used]
+    if reunion_param:
+        where_clauses.append("p.num_reunion = ?")
+        params.append(int(reunion_param))
+    if course_param:
+        where_clauses.append("p.num_course = ?")
+        params.append(int(course_param))
+
+    query = f"""
+        SELECT p.*, c.distance, c.discipline, c.specialite,
+               c.nb_declares_partants, c.montant_prix, c.penetrometre_intitule,
+               c.libelle as course_libelle, c.libelle_court as hippodrome,
+               c.heure_depart_str, c.parcours
+        FROM pmu_partants p
+        LEFT JOIN pmu_courses c ON p.date_programme = c.date_programme
+            AND p.num_reunion = c.num_reunion AND p.num_course = c.num_course
+        WHERE {" AND ".join(where_clauses)}
+        ORDER BY p.num_reunion, p.num_course, p.num_pmu
+    """
+    rows = conn.execute(query, params).fetchall()
+    conn.close()
+
+    if not rows:
+        return jsonify(
+            {
+                "date": date_used,
+                "model_version": mod.get_model_version(),
+                "predictions": [],
+                "message": f"No partants found for date {date_used}",
+            }
+        )
+
+    # Convert to list of dicts
+    partants = [dict(r) for r in rows]
+
+    # Run ensemble prediction
+    preds = mod.predict_top3(partants, model=ensemble)
+
+    # Group by race
+    races = {}
+    for pred in preds:
+        key = f"R{pred.get('num_reunion', 0)}C{pred.get('num_course', 0)}"
+        if key not in races:
+            # Find race metadata from partants
+            for p in partants:
+                if p.get("num_reunion") == pred.get("num_reunion") and p.get(
+                    "num_course"
+                ) == pred.get("num_course"):
+                    races[key] = {
+                        "reunion": pred.get("num_reunion"),
+                        "course": pred.get("num_course"),
+                        "label": key,
+                        "race_name": p.get("course_libelle", ""),
+                        "hippodrome": p.get("hippodrome", ""),
+                        "heure": p.get("heure_depart_str", ""),
+                        "discipline": p.get("discipline", ""),
+                        "distance": p.get("distance", 0),
+                        "horses": [],
+                    }
+                    break
+        if key in races:
+            races[key]["horses"].append(pred)
+
+    latency_ms = (_time.perf_counter() - t0) * 1000
+
+    return jsonify(
+        {
+            "date": date_used,
+            "model_version": mod.get_model_version(),
+            "latency_ms": round(latency_ms, 1),
+            "total_horses": len(preds),
+            "races": list(races.values()),
+        }
+    )
+
+
+@app.route("/api/v1/model/invalidate-cache", methods=["POST"])
+@app.route("/turf/api/v1/model/invalidate-cache", methods=["POST"])
+def api_v1_invalidate_cache():
+    """Force reload of ensemble model on next prediction call."""
+    mod = _load_predict_v2()
+    if mod:
+        mod.invalidate_model_cache()
+        return jsonify({"status": "ok", "message": "Model cache invalidated"})
+    return jsonify({"status": "error", "message": "predict_v2 module not loaded"}), 500
+
+
+@app.route("/api/v1/model/status", methods=["GET"])
+@app.route("/turf/api/v1/model/status", methods=["GET"])
+def api_v1_model_status():
+    """Return ensemble model status and version."""
+    import os as _os
+    from pathlib import Path as _Path
+
+    ensemble_path = _Path("/home/h3r7/turf_saas/models/ensemble_top3.pkl")
+    benchmark_path = _Path("/home/h3r7/turf_saas/models/benchmark_report.json")
+
+    status = {
+        "ensemble_available": ensemble_path.exists(),
+        "ensemble_path": str(ensemble_path),
+    }
+    if ensemble_path.exists():
+        mtime = _os.path.getmtime(str(ensemble_path))
+        status["last_trained"] = datetime.fromtimestamp(mtime).isoformat()
+
+    if benchmark_path.exists():
+        try:
+            with open(benchmark_path) as f:
+                import json as _json
+
+                report = _json.load(f)
+            status["benchmark"] = {
+                "baseline_precision_at3": report.get("baseline", {}).get(
+                    "precision_at3"
+                ),
+                "ensemble_precision_at3": report.get("ensemble", {}).get(
+                    "precision_at3"
+                ),
+                "delta": report.get("delta_precision_at3"),
+                "deployed": report.get("deploy"),
+                "run_date": report.get("run_date"),
+            }
+        except Exception:
+            pass
+
+    mod = _load_predict_v2()
+    if mod and ensemble_path.exists():
+        status["model_version"] = mod.get_model_version()
+
+    return jsonify(status)
+
+
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=8790, debug=False)

predict_v2.py

@@ -0,0 +1,387 @@
+#!/usr/bin/env python3
+"""
+Ensemble prediction module for /api/v1/predictions.
+
+Loads the trained ensemble model and provides a high-level predict_top3()
+function compatible with the existing combined_api.py interface.
+
+Cache: model is loaded once at import time (or on first call).
+       Invalidation: reload if models/ensemble_top3.pkl mtime changes.
+"""
+
+import logging
+import os
+import pickle
+import re
+import threading
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+import pandas as pd
+from sklearn.preprocessing import LabelEncoder
+
+logger = logging.getLogger(__name__)
+
+MODELS_DIR = Path("/home/h3r7/turf_saas/models")
+ENSEMBLE_PATH = MODELS_DIR / "ensemble_top3.pkl"
+
+# ── Cache ─────────────────────────────────────────────────────────────────────
+_model_cache = {
+    "ensemble": None,
+    "mtime": None,
+    "lock": threading.Lock(),
+}
+
+# ── Feature list (must match train_ensemble.py FEATURE_COLS) ─────────────────
+FEATURE_COLS = [
+    "age",
+    "sexe_enc",
+    "nombre_courses",
+    "nombre_victoires",
+    "nombre_places",
+    "tx_victoire",
+    "tx_place",
+    "forme_recente",
+    "tendance_num",
+    "gains_annee_en_cours",
+    "cote_direct",
+    "cote_reference",
+    "distance",
+    "nb_partants",
+    "discipline_enc",
+    "specialite_enc",
+    "oeilleres_enc",
+    "tendance_cote_enc",
+    "penetrometre_intitule_enc",
+    "form_1",
+    "form_2",
+    "form_3",
+    "form_4",
+    "form_5",
+    "form_weighted",
+    "form_avg",
+    "form_best",
+    "form_worst",
+    "win_ratio",
+    "place_ratio",
+    "implied_prob",
+    "win_rate_adj",
+    "place_rate_adj",
+    "earnings_per_race",
+    "cote_diff",
+    "cote_ratio",
+    "rang_cote",
+    "ratio_cote_field",
+    "distance_cat",
+    "age_win_interact",
+    "is_favorite",
+    "poids",
+    "prize_norm",
+]
+
+
+# ── Encoders (built per-prediction batch for live data) ──────────────────────
+def _fit_encoder(values, default):
+    le = LabelEncoder()
+    unique = list(set(str(v) if v else default for v in values)) + [default]
+    le.fit(unique)
+    return le
+
+
+def _safe_transform(le: LabelEncoder, value, default: str):
+    v = str(value) if value else default
+    if v not in le.classes_:
+        v = default
+    return int(le.transform([v])[0])
+
+
+# ── Model loading with auto-invalidation ─────────────────────────────────────
+def load_ensemble(force: bool = False) -> Optional[object]:
+    """Load ensemble model, reload if file changed."""
+    with _model_cache["lock"]:
+        if not ENSEMBLE_PATH.exists():
+            return None
+        mtime = ENSEMBLE_PATH.stat().st_mtime
+        if force or _model_cache["ensemble"] is None or mtime != _model_cache["mtime"]:
+            try:
+                with open(ENSEMBLE_PATH, "rb") as f:
+                    _model_cache["ensemble"] = pickle.load(f)
+                _model_cache["mtime"] = mtime
+                logger.info(f"[predict_v2] Loaded ensemble model from {ENSEMBLE_PATH}")
+            except Exception as e:
+                logger.error(f"[predict_v2] Failed to load ensemble: {e}")
+                return None
+        return _model_cache["ensemble"]
+
+
+def invalidate_model_cache():
+    """Force reload on next prediction call."""
+    with _model_cache["lock"]:
+        _model_cache["mtime"] = None
+
+
+# ── Feature engineering for live pmu_partants rows ───────────────────────────
+def _parse_musique(musique) -> list:
+    if not musique or pd.isna(str(musique)):
+        return [0, 0, 0, 0, 0]
+    try:
+        clean = re.sub(r"\(\d+\)", "", str(musique))
+        numbers = re.findall(r"\d+", clean)
+        result = [int(n) for n in numbers[:5]]
+        result += [0] * (5 - len(result))
+        return result[:5]
+    except Exception:
+        return [0, 0, 0, 0, 0]
+
+
+def build_feature_df(partants: list) -> pd.DataFrame:
+    """
+    Convert a list of pmu_partants dicts to a feature DataFrame.
+
+    Expected keys (same as pmu_partants columns):
+        date_programme, num_reunion, num_course, num_pmu,
+        age, sexe, musique, nombre_courses, nombre_victoires, nombre_places,
+        gains_annee_en_cours, handicap_poids, oeilleres, cote_direct,
+        cote_reference, tendance_cote, favoris, tx_victoire, tx_place,
+        forme_recente, tendance_forme, indicateur_inedit,
+        distance, discipline, specialite, nb_declares_partants,
+        montant_prix, penetrometre_intitule
+    """
+    if not partants:
+        return pd.DataFrame()
+
+    df = pd.DataFrame(partants)
+
+    # ── Categorical encoders fitted on this batch ─────────────────────────────
+    le_sexe = _fit_encoder(df.get("sexe", ["U"]), "U")
+    le_oeilleres = _fit_encoder(df.get("oeilleres", ["SANS"]), "SANS")
+    le_discipline = _fit_encoder(df.get("discipline", ["UNKNOWN"]), "UNKNOWN")
+    le_specialite = _fit_encoder(df.get("specialite", ["UNKNOWN"]), "UNKNOWN")
+    le_tendance = _fit_encoder(df.get("tendance_cote", ["STABLE"]), "STABLE")
+    le_penet = _fit_encoder(df.get("penetrometre_intitule", ["BON"]), "BON")
+
+    df["sexe_enc"] = df["sexe"].apply(lambda v: _safe_transform(le_sexe, v, "U"))
+    df["oeilleres_enc"] = df["oeilleres"].apply(
+        lambda v: _safe_transform(le_oeilleres, v, "SANS")
+    )
+    df["discipline_enc"] = df.get("discipline", pd.Series(["UNKNOWN"] * len(df))).apply(
+        lambda v: _safe_transform(le_discipline, v, "UNKNOWN")
+    )
+    df["specialite_enc"] = df.get("specialite", pd.Series(["UNKNOWN"] * len(df))).apply(
+        lambda v: _safe_transform(le_specialite, v, "UNKNOWN")
+    )
+    df["tendance_cote_enc"] = df.get(
+        "tendance_cote", pd.Series(["STABLE"] * len(df))
+    ).apply(lambda v: _safe_transform(le_tendance, v, "STABLE"))
+    df["penetrometre_intitule_enc"] = df.get(
+        "penetrometre_intitule", pd.Series(["BON"] * len(df))
+    ).apply(lambda v: _safe_transform(le_penet, v, "BON"))
+
+    # ── Musique ────────────────────────────────────────────────────────────────
+    music_parsed = df["musique"].apply(_parse_musique)
+    for i in range(5):
+        df[f"form_{i + 1}"] = music_parsed.apply(lambda x: x[i])
+    weights = np.array([0.4, 0.25, 0.15, 0.12, 0.08])
+    df["form_weighted"] = music_parsed.apply(
+        lambda x: sum(w * v for w, v in zip(weights, x))
+    )
+    df["form_avg"] = music_parsed.apply(np.mean)
+    df["form_best"] = music_parsed.apply(min)
+    df["form_worst"] = music_parsed.apply(max)
+
+    # ── Numeric features ───────────────────────────────────────────────────────
+    for col in [
+        "nombre_courses",
+        "nombre_victoires",
+        "nombre_places",
+        "tx_victoire",
+        "tx_place",
+        "forme_recente",
+        "tendance_forme",
+        "gains_annee_en_cours",
+        "cote_direct",
+        "cote_reference",
+        "distance",
+        "handicap_poids",
+        "age",
+        "montant_prix",
+        "nb_declares_partants",
+    ]:
+        if col not in df.columns:
+            df[col] = 0.0
+        df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0)
+
+    df["tendance_num"] = df["tendance_forme"].fillna(0)
+    df["win_ratio"] = df["nombre_victoires"] / df["nombre_courses"].replace(0, 1)
+    df["place_ratio"] = df["nombre_places"] / df["nombre_courses"].replace(0, 1)
+    df["implied_prob"] = 1.0 / df["cote_direct"].replace(0, np.nan)
+    df["win_rate_adj"] = df["tx_victoire"] * np.log1p(df["nombre_courses"])
+    df["place_rate_adj"] = df["tx_place"] * np.log1p(df["nombre_courses"])
+    df["earnings_per_race"] = df["gains_annee_en_cours"] / df["nombre_courses"].replace(
+        0, 1
+    )
+    df["cote_diff"] = (df["cote_direct"] - df["cote_reference"]).fillna(0)
+    df["cote_ratio"] = (
+        df["cote_direct"] / df["cote_reference"].replace(0, np.nan)
+    ).fillna(1)
+
+    # ── Per-race rank features ─────────────────────────────────────────────────
+    if "num_reunion" in df.columns and "num_course" in df.columns:
+        grp = ["date_programme", "num_reunion", "num_course"]
+        # Some fields may be missing
+        for g in grp:
+            if g not in df.columns:
+                df[g] = 0
+        df["rang_cote"] = df.groupby(grp)["cote_direct"].rank(
+            method="min", na_option="bottom"
+        )
+        race_mean = df.groupby(grp)["cote_direct"].transform("mean")
+        df["ratio_cote_field"] = df["cote_direct"] / race_mean.replace(0, np.nan)
+        df["nb_partants"] = df.groupby(grp)["cote_direct"].transform("count")
+    else:
+        df["rang_cote"] = 1.0
+        df["ratio_cote_field"] = 1.0
+        df["nb_partants"] = df.get("nb_declares_partants", pd.Series([10] * len(df)))
+
+    df["distance_cat"] = pd.cut(
+        df["distance"].fillna(1600),
+        bins=[0, 1400, 1800, 2200, 2600, 10000],
+        labels=[1, 2, 3, 4, 5],
+    ).astype(float)
+    df["age_win_interact"] = df["age"] * df["tx_victoire"]
+    df["is_favorite"] = (
+        df.get("favoris", pd.Series([0] * len(df))).fillna(0).astype(int)
+    )
+    df["poids"] = df["handicap_poids"].fillna(60)
+    df["prize_norm"] = np.log1p(df["montant_prix"].fillna(0))
+
+    return df
+
+
+# ── Main prediction function ───────────────────────────────────────────────────
+def predict_top3(partants: list, model=None) -> list:
+    """
+    Given a list of partant dicts (from pmu_partants), return predictions.
+
+    Returns list of {horse_name, num_pmu, prob_top3, prob_top1_approx, ...}
+    sorted by prob_top3 descending.
+
+    Falls back to empty list if model not available.
+    """
+    t_start = time.perf_counter()
+
+    if model is None:
+        model = load_ensemble()
+    if model is None:
+        logger.warning("[predict_v2] Ensemble model not available — no predictions")
+        return []
+
+    df = build_feature_df(partants)
+    if df.empty:
+        return []
+
+    available = [c for c in FEATURE_COLS if c in df.columns]
+    X = df[available].fillna(0)
+
+    try:
+        proba = model.predict_proba(X)[:, 1]
+    except Exception as e:
+        logger.error(f"[predict_v2] predict_proba failed: {e}")
+        return []
+
+    latency_ms = (time.perf_counter() - t_start) * 1000
+
+    results = []
+    for i, (p, row) in enumerate(zip(proba, partants)):
+        results.append(
+            {
+                "horse_name": row.get("nom", row.get("horse_name", f"H{i}")),
+                "num_pmu": row.get("num_pmu", i + 1),
+                "num_reunion": row.get("num_reunion"),
+                "num_course": row.get("num_course"),
+                "prob_top3": round(float(p) * 100, 1),
+                # approx top1 from top3 score (divide by ~2.5 empirically)
+                "prob_top1": round(float(p) / 2.5 * 100, 1),
+                "ml_score": round(float(p) * 100, 1),
+                "recommendation": "top3"
+                if p >= 0.40
+                else ("watch" if p >= 0.28 else "pass"),
+                "is_value_bet": int(
+                    p >= 0.35 and float(row.get("cote_direct", 0) or 0) > 10
+                ),
+                "model_version": getattr(model, "version", "ensemble_v1"),
+            }
+        )
+
+    results.sort(key=lambda x: x["prob_top3"], reverse=True)
+
+    # Mark top-3 predicted
+    for i, r in enumerate(results[:3]):
+        r["predicted_rank"] = i + 1
+
+    if results:
+        logger.info(
+            f"[predict_v2] {len(results)} horses predicted in {latency_ms:.1f} ms "
+            f"({latency_ms / len(results):.2f} ms/horse)"
+        )
+
+    return results
+
+
+# ── API-compatible wrapper keeping model_version & structure ──────────────────
+def get_model_version() -> str:
+    m = load_ensemble()
+    if m is None:
+        return "ensemble_v1_not_loaded"
+    return getattr(m, "version", "ensemble_v1")
+
+
+if __name__ == "__main__":
+    # Quick self-test
+    import sqlite3
+
+    conn = sqlite3.connect("/home/h3r7/turf_saas/turf.db")
+    rows = conn.execute(
+        """SELECT p.*, c.distance, c.discipline, c.specialite,
+                  c.nb_declares_partants, c.montant_prix, c.penetrometre_intitule
+           FROM pmu_partants p
+           LEFT JOIN pmu_courses c ON p.date_programme=c.date_programme
+                AND p.num_reunion=c.num_reunion AND p.num_course=c.num_course
+           WHERE p.date_programme=(SELECT MAX(date_programme) FROM pmu_partants)
+             AND p.num_reunion=1 AND p.num_course=1
+           LIMIT 20"""
+    ).fetchall()
+    conn.close()
+
+    if not rows:
+        print("No data found for self-test")
+    else:
+        cols = [d[0] for d in conn.description] if hasattr(conn, "description") else []
+        # Fallback column list
+        import sqlite3 as sq3
+
+        conn2 = sq3.connect("/home/h3r7/turf_saas/turf.db")
+        cur = conn2.execute(
+            """SELECT p.*, c.distance, c.discipline, c.specialite,
+                      c.nb_declares_partants, c.montant_prix, c.penetrometre_intitule
+               FROM pmu_partants p
+               LEFT JOIN pmu_courses c ON p.date_programme=c.date_programme
+                    AND p.num_reunion=c.num_reunion AND p.num_course=c.num_course
+               WHERE p.date_programme=(SELECT MAX(date_programme) FROM pmu_partants)
+                 AND p.num_reunion=1 AND p.num_course=1
+               LIMIT 20"""
+        )
+        cols = [d[0] for d in cur.description]
+        rows2 = cur.fetchall()
+        conn2.close()
+
+        partants = [dict(zip(cols, row)) for row in rows2]
+        preds = predict_top3(partants)
+        print(f"Self-test: {len(preds)} predictions")
+        for p in preds[:5]:
+            print(
+                f"  {p['horse_name']:20s}  prob_top3={p['prob_top3']}%  rec={p['recommendation']}"
+            )

tests/test_ml_ensemble.py

@@ -0,0 +1,333 @@
+"""
+Tests ML Ensemble — HRT-32 Sprint 6-7
+Tests de régression, benchmark et latence pour le nouveau modèle ensemble.
+
+Usage:
+    pytest tests/test_ml_ensemble.py -v
+    pytest tests/test_ml_ensemble.py -v -m regression
+    pytest tests/test_ml_ensemble.py -v -m latency
+"""
+
+import json
+import os
+import pickle
+import sqlite3
+import time
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import pytest
+import requests
+
+BASE_URL = os.environ.get("APP_URL", "http://localhost:8790")
+DB_PATH = os.environ.get("DB_PATH", "/home/h3r7/turf_saas/turf.db")
+MODELS_DIR = Path("/home/h3r7/turf_saas/models")
+ENSEMBLE_PATH = MODELS_DIR / "ensemble_top3.pkl"
+BENCHMARK_PATH = MODELS_DIR / "benchmark_report.json"
+
+
+# ─── Fixtures ────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture(scope="session")
+def ensemble_model():
+    """Load ensemble model (skip tests if not yet trained)."""
+    if not ENSEMBLE_PATH.exists():
+        pytest.skip(
+            f"Ensemble model not found at {ENSEMBLE_PATH}. Run train_ensemble.py first."
+        )
+    with open(ENSEMBLE_PATH, "rb") as f:
+        return pickle.load(f)
+
+
+@pytest.fixture(scope="session")
+def benchmark_report():
+    """Load benchmark report (skip if not generated)."""
+    if not BENCHMARK_PATH.exists():
+        pytest.skip(f"Benchmark report not found at {BENCHMARK_PATH}.")
+    with open(BENCHMARK_PATH) as f:
+        return json.load(f)
+
+
+@pytest.fixture(scope="session")
+def holdout_data():
+    """Load holdout slice (last 20% temporal) for regression tests."""
+    conn = sqlite3.connect(DB_PATH)
+    df = pd.read_sql_query(
+        """
+        SELECT p.*, c.distance, c.discipline, c.specialite,
+               c.nb_declares_partants, c.montant_prix, c.penetrometre_intitule
+        FROM pmu_partants p
+        LEFT JOIN pmu_courses c ON p.date_programme=c.date_programme
+            AND p.num_reunion=c.num_reunion AND p.num_course=c.num_course
+        WHERE p.ordre_arrivee > 0
+        ORDER BY p.date_programme, p.num_reunion, p.num_course, p.num_pmu
+        """,
+        conn,
+    )
+    conn.close()
+    n = len(df)
+    cutoff = int(n * 0.80)
+    return df.iloc[cutoff:].copy()
+
+
+@pytest.fixture(scope="session")
+def predict_v2():
+    """Import predict_v2 module."""
+    import importlib.util
+
+    spec = importlib.util.spec_from_file_location(
+        "predict_v2", "/home/h3r7/turf_saas/predict_v2.py"
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# ─── Model Existence Tests ────────────────────────────────────────────────────
+
+
+class TestModelFiles:
+    """Verify all expected model files exist."""
+
+    def test_ensemble_model_exists(self):
+        assert ENSEMBLE_PATH.exists(), f"Ensemble model missing: {ENSEMBLE_PATH}"
+
+    def test_benchmark_report_exists(self):
+        assert BENCHMARK_PATH.exists(), f"Benchmark report missing: {BENCHMARK_PATH}"
+
+    def test_models_dir_contains_expected_files(self):
+        expected = ["ensemble_top3.pkl", "benchmark_report.json", "benchmark_report.md"]
+        for fname in expected:
+            assert (MODELS_DIR / fname).exists(), f"Missing: {MODELS_DIR / fname}"
+
+
+# ─── Benchmark Tests ──────────────────────────────────────────────────────────
+
+
+class TestBenchmark:
+    """Validate benchmark metrics from the training report."""
+
+    @pytest.mark.regression
+    def test_ensemble_beats_baseline_or_meets_threshold(self, benchmark_report):
+        """Ensemble Precision@3 must be >= baseline XGBoost."""
+        baseline = benchmark_report["baseline"]["precision_at3"]
+        ensemble = benchmark_report["ensemble"]["precision_at3"]
+        assert ensemble >= baseline, (
+            f"Ensemble Precision@3 {ensemble:.4f} < baseline {baseline:.4f}"
+        )
+
+    @pytest.mark.regression
+    def test_ensemble_auc_above_random(self, benchmark_report):
+        """Ensemble AUC must be > 0.60 (significantly above random 0.50)."""
+        auc = benchmark_report["ensemble"]["auc"]
+        assert auc > 0.60, f"Ensemble AUC {auc:.4f} <= 0.60"
+
+    @pytest.mark.regression
+    def test_optuna_ran_minimum_trials(self, benchmark_report):
+        """Optuna must have run at least 100 trials per model."""
+        n_trials = benchmark_report["optuna"]["n_trials"]
+        assert n_trials >= 100, f"Only {n_trials} Optuna trials (minimum 100 required)"
+
+    @pytest.mark.regression
+    def test_no_precision_regression(self, benchmark_report):
+        """Ensemble Precision@3 must not be below naive random baseline (~30%)."""
+        ensemble_p3 = benchmark_report["ensemble"]["precision_at3"]
+        assert ensemble_p3 >= 0.30, (
+            f"Precision@3 {ensemble_p3:.4f} is below random baseline (~0.30)"
+        )
+
+    def test_benchmark_has_all_required_models(self, benchmark_report):
+        """Benchmark must include results for all 3 models."""
+        required = {"xgboost", "lightgbm", "mlp"}
+        found = set(benchmark_report.get("individual_models", {}).keys())
+        missing = required - found
+        assert not missing, f"Missing model benchmarks: {missing}"
+
+
+# ─── Regression Tests ─────────────────────────────────────────────────────────
+
+
+class TestPrecisionRegression:
+    """Holdout regression: ensure precision doesn't degrade."""
+
+    @pytest.mark.regression
+    def test_precision_at3_on_holdout(self, ensemble_model, holdout_data):
+        """Precision@3 on holdout must be above naive baseline."""
+        from predict_v2 import build_feature_df, FEATURE_COLS
+
+        df = holdout_data.copy()
+        df["top3"] = (df["ordre_arrivee"] <= 3).astype(int)
+
+        partants = df.to_dict("records")
+        feature_df = build_feature_df(partants)
+        available = [c for c in FEATURE_COLS if c in feature_df.columns]
+        X = feature_df[available].fillna(0)
+
+        proba = ensemble_model.predict_proba(X)[:, 1]
+
+        # Per-race Precision@3
+        tmp = df[["date_programme", "num_reunion", "num_course"]].copy()
+        tmp["proba"] = proba
+        tmp["actual"] = df["top3"].values
+
+        precisions = []
+        for _, group in tmp.groupby(["date_programme", "num_reunion", "num_course"]):
+            if len(group) >= 3:
+                top3_pred = group.nlargest(3, "proba")
+                precisions.append(top3_pred["actual"].sum() / 3.0)
+
+        p_at3 = float(np.mean(precisions)) if precisions else 0.0
+        print(f"\n  Holdout Precision@3: {p_at3:.4f} over {len(precisions)} races")
+
+        # Must beat random baseline (30%)
+        assert p_at3 >= 0.30, f"Holdout Precision@3 {p_at3:.4f} < 0.30"
+
+    @pytest.mark.regression
+    def test_no_all_zero_predictions(self, ensemble_model, holdout_data):
+        """Ensemble must not predict 0 probability for all horses."""
+        from predict_v2 import build_feature_df, FEATURE_COLS
+
+        partants = holdout_data.head(50).to_dict("records")
+        feature_df = build_feature_df(partants)
+        available = [c for c in FEATURE_COLS if c in feature_df.columns]
+        X = feature_df[available].fillna(0)
+
+        proba = ensemble_model.predict_proba(X)[:, 1]
+        assert proba.max() > 0.01, "All predictions are near 0 — model appears broken"
+        assert proba.std() > 0.01, (
+            "All predictions have identical probability — no discrimination"
+        )
+
+
+# ─── Latency Tests ────────────────────────────────────────────────────────────
+
+
+class TestPredictionLatency:
+    """Prediction latency must be < 200ms per race."""
+
+    @pytest.mark.latency
+    def test_single_race_latency(self, ensemble_model, holdout_data):
+        """Prediction for a single race (<=20 horses) must be < 200ms."""
+        from predict_v2 import build_feature_df, FEATURE_COLS
+
+        # Take one race
+        first_race = (
+            holdout_data.groupby(["date_programme", "num_reunion", "num_course"])
+            .first()
+            .reset_index()
+            .iloc[0]
+        )
+        mask = (
+            (holdout_data["date_programme"] == first_race["date_programme"])
+            & (holdout_data["num_reunion"] == first_race["num_reunion"])
+            & (holdout_data["num_course"] == first_race["num_course"])
+        )
+        race_df = holdout_data[mask]
+        partants = race_df.to_dict("records")
+
+        # Warm-up
+        feature_df = build_feature_df(partants)
+        available = [c for c in FEATURE_COLS if c in feature_df.columns]
+        X = feature_df[available].fillna(0)
+        ensemble_model.predict_proba(X)
+
+        # Timed run
+        t0 = time.perf_counter()
+        for _ in range(10):
+            ensemble_model.predict_proba(X)
+        elapsed_ms = (time.perf_counter() - t0) / 10 * 1000
+
+        print(f"\n  Single-race latency: {elapsed_ms:.2f} ms ({len(partants)} horses)")
+        assert elapsed_ms < 200, (
+            f"Prediction latency {elapsed_ms:.1f} ms exceeds 200 ms limit"
+        )
+
+    @pytest.mark.latency
+    def test_full_day_latency(self, ensemble_model, holdout_data):
+        """Prediction for a full day (all races) must complete < 5 seconds."""
+        from predict_v2 import build_feature_df, FEATURE_COLS
+
+        # Take one day
+        day = holdout_data["date_programme"].iloc[0]
+        day_df = holdout_data[holdout_data["date_programme"] == day]
+        partants = day_df.to_dict("records")
+
+        feature_df = build_feature_df(partants)
+        available = [c for c in FEATURE_COLS if c in feature_df.columns]
+        X = feature_df[available].fillna(0)
+
+        t0 = time.perf_counter()
+        proba = ensemble_model.predict_proba(X)
+        elapsed_ms = (time.perf_counter() - t0) * 1000
+
+        print(
+            f"\n  Full day latency: {elapsed_ms:.2f} ms ({len(partants)} horses, {day})"
+        )
+        assert elapsed_ms < 5000, (
+            f"Full-day prediction {elapsed_ms:.0f} ms exceeds 5s limit"
+        )
+
+
+# ─── API Endpoint Tests ───────────────────────────────────────────────────────
+
+
+class TestV1PredictionsAPI:
+    """Tests for the new /api/v1/predictions endpoint."""
+
+    def _api_available(self):
+        try:
+            requests.get(f"{BASE_URL}/api/v1/model/status", timeout=3)
+            return True
+        except Exception:
+            return False
+
+    @pytest.mark.api
+    def test_model_status_endpoint(self):
+        """GET /api/v1/model/status returns valid JSON."""
+        if not self._api_available():
+            pytest.skip("API server not running")
+        resp = requests.get(f"{BASE_URL}/api/v1/model/status", timeout=10)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "ensemble_available" in data
+
+    @pytest.mark.api
+    def test_v1_predictions_no_500(self):
+        """GET /api/v1/predictions must not return 5xx."""
+        if not self._api_available():
+            pytest.skip("API server not running")
+        resp = requests.get(f"{BASE_URL}/api/v1/predictions", timeout=30)
+        assert resp.status_code < 500, (
+            f"Server error: {resp.status_code}\n{resp.text[:200]}"
+        )
+
+    @pytest.mark.api
+    def test_v1_predictions_returns_json(self):
+        """GET /api/v1/predictions returns valid JSON with expected keys."""
+        if not self._api_available():
+            pytest.skip("API server not running")
+        resp = requests.get(f"{BASE_URL}/api/v1/predictions", timeout=30)
+        if resp.status_code == 503:
+            pytest.skip("Ensemble model not yet deployed")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "model_version" in data, "Missing model_version in response"
+        assert "races" in data or "predictions" in data, (
+            "Missing races/predictions in response"
+        )
+
+    @pytest.mark.api
+    def test_v1_predictions_latency(self):
+        """GET /api/v1/predictions must respond in < 3 seconds."""
+        if not self._api_available():
+            pytest.skip("API server not running")
+        resp = requests.get(f"{BASE_URL}/api/v1/predictions", timeout=30)
+        if resp.status_code == 503:
+            pytest.skip("Ensemble model not yet deployed")
+        # Check API-reported latency
+        if resp.status_code == 200:
+            data = resp.json()
+            latency = data.get("latency_ms", 0)
+            assert latency < 3000, f"API latency {latency:.0f} ms > 3000 ms"