""" The comprehensive attempt: every feature we can build, every model worth trying. (read-only, trades nothing) Prior attempts tested features in ISOLATION -- rolling form alone (nonprice_edge_test.py), then Elo alone (elo_model_test.py). Both lost to the closing line. This combines everything and tunes properly, so that "we didn't try hard enough" stops being a live explanation. FEATURES - Elo at THREE timescales (K=10 slow / 20 / 40 fast), goal-difference weighted, walk-forward so a match is always scored with the rating held BEFORE it. - Home-specific and away-specific Elo (teams differ home vs away). - Rest days per side + difference (congestion). - Rolling points / goals for / goals against over 5 and 10, per side. - VENUE-specific form (home team's home record, away team's away record). - Head-to-head: last-3 meetings goal difference. - Season progress, and the market's own probability as a base. MODELS: multinomial logistic across a regularisation sweep, gradient boosting across depth/rate, random forest, and a simple averaging ensemble. SUCCESS CRITERION, fixed in advance: `market + features` must beat `market` on a TIME-based holdout, with a paired t > 1.96 on per-fixture log loss. Anything else is noise. A betting simulation is NOT evidence -- elo_model_test showed a +2.7% "profit" that the no-model control revealed to be a favourite-longshot/devig artifact, so any betting result here gets the same control. Usage: python research\\full_model_test.py """ import sys import csv import math import statistics from datetime import datetime from collections import defaultdict, deque sys.path.insert(0, r'E:\soccer\main') sys.path.insert(0, r'E:\soccer\research') import daily_signals as DS import nonprice_edge_test as T FIX = r'E:\soccer\raw_api\fixtures_index.csv' IDX = {'H': 0, 'D': 1, 'A': 2} def _i(x): try: return int(float(x)) except (TypeError, ValueError): return None def load(): out = [] with open(FIX, newline='', encoding='utf-8') as fh: for r in csv.DictReader(fh): hg, ag = _i(r.get('home_goals')), _i(r.get('away_goals')) if hg is None or ag is None: continue try: d = datetime.strptime(r['date'][:10], '%Y-%m-%d').date() except (ValueError, KeyError): continue out.append(dict(date=d, div=r.get('div'), home=r.get('home_team'), away=r.get('away_team'), hg=hg, ag=ag)) out.sort(key=lambda x: x['date']) return out def featurise(fx): """Walk forward once, attaching only information available BEFORE each match.""" elo = {k: defaultdict(lambda: 1500.0) for k in (10, 20, 40)} elo_venue = {'h': defaultdict(lambda: 1500.0), 'a': defaultdict(lambda: 1500.0)} last = {} form = defaultdict(lambda: deque(maxlen=10)) # (pts, gf, ga) vform = {'h': defaultdict(lambda: deque(maxlen=10)), 'a': defaultdict(lambda: deque(maxlen=10))} h2h = defaultdict(lambda: deque(maxlen=3)) played = defaultdict(int) def agg(dq, n): s = list(dq)[-n:] if not s: return [0.0, 0.0, 0.0] return [sum(x[i] for x in s) / len(s) for i in range(3)] for f in fx: h, a = f['home'], f['away'] row = [] for K in (10, 20, 40): row += [elo[K][h], elo[K][a], elo[K][h] - elo[K][a]] row += [elo_venue['h'][h], elo_venue['a'][a], elo_venue['h'][h] - elo_venue['a'][a]] rh = (f['date'] - last[h]).days if h in last else 7 ra = (f['date'] - last[a]).days if a in last else 7 row += [min(rh, 30), min(ra, 30), min(rh, 30) - min(ra, 30)] for n in (5, 10): fh_, fa_ = agg(form[h], n), agg(form[a], n) row += fh_ + fa_ + [x - y for x, y in zip(fh_, fa_)] row += agg(vform['h'][h], 5) + agg(vform['a'][a], 5) key = tuple(sorted([h, a])) hh = list(h2h[key]) row += [sum(x for x in hh) / len(hh) if hh else 0.0, len(hh)] row += [min(played[h], 40), min(played[a], 40)] f['x'] = row # ---- updates (after the features are frozen) ---- gd = f['hg'] - f['ag'] sc = 1.0 if gd > 0 else (0.5 if gd == 0 else 0.0) mult = math.log(abs(gd) + 1) + 1 for K in (10, 20, 40): exp = 1.0 / (1.0 + 10 ** ((elo[K][a] - elo[K][h] - 65.0) / 400.0)) d = K * mult * (sc - exp) elo[K][h] += d elo[K][a] -= d expv = 1.0 / (1.0 + 10 ** ((elo_venue['a'][a] - elo_venue['h'][h]) / 400.0)) dv = 20 * mult * (sc - expv) elo_venue['h'][h] += dv elo_venue['a'][a] -= dv ph = 3 if gd > 0 else (1 if gd == 0 else 0) form[h].append((ph, f['hg'], f['ag'])) form[a].append((3 - ph if ph != 1 else 1, f['ag'], f['hg'])) vform['h'][h].append((ph, f['hg'], f['ag'])) vform['a'][a].append((3 - ph if ph != 1 else 1, f['ag'], f['hg'])) h2h[key].append(gd if h == key[0] else -gd) played[h] += 1 played[a] += 1 last[h] = last[a] = f['date'] return fx def main(): import numpy as np from sklearn.linear_model import LogisticRegression from sklearn.ensemble import HistGradientBoostingClassifier, RandomForestClassifier from sklearn.preprocessing import StandardScaler fx = featurise(load()) odds = T.load_odds() rows = [] for f in fx: k = (f['date'], T.norm(DS.fdco_name(f['home'])), T.norm(DS.fdco_name(f['away']))) hit = odds.get(k) if not hit: continue p, res = hit rows.append(dict(date=f['date'], p=p, res=res, x=f['x'])) rows.sort(key=lambda r: r['date']) print(f'matched fixtures: {len(rows)} features per fixture: {len(rows[0]["x"])}') cut = int(len(rows) * 0.7) tr, te = rows[:cut], rows[cut:] print(f' train {len(tr)} (to {tr[-1]["date"]}) test {len(te)} ' f'({te[0]["date"]} on)\n') ytr = [r['res'] for r in tr] yte = [r['res'] for r in te] Xtr = np.array([r['x'] for r in tr], dtype=float) Xte = np.array([r['x'] for r in te], dtype=float) sc = StandardScaler().fit(Xtr) Xtr, Xte = sc.transform(Xtr), sc.transform(Xte) Mtr = np.array([[math.log(max(v, 1e-9)) for v in r['p']] for r in tr]) Mte = np.array([[math.log(max(v, 1e-9)) for v in r['p']] for r in te]) mk = [r['p'] for r in te] def ll_vec(P): return [-math.log(max(P[i][IDX[yte[i]]], 1e-9)) for i in range(len(yte))] base_v = ll_vec(mk) base = statistics.mean(base_v) print(f'{"model":48s} {"log loss":>9s} {"vs mkt":>9s} {"paired t":>9s}') print(f'{"MARKET (Pinnacle closing)":48s} {base:9.5f} {"--":>9s} {"--":>9s}') results = [] def ev(label, A, B, model): try: model.fit(A, ytr) except Exception as e: print(f'{label:48s} FAILED {e}') return pr = model.predict_proba(B) o = list(model.classes_) P = [[pr[i][o.index(k)] for k in ('H', 'D', 'A')] for i in range(len(yte))] v = ll_vec(P) d = [base_v[i] - v[i] for i in range(len(v))] mu = statistics.mean(d) se = statistics.pstdev(d) / math.sqrt(len(d)) t = mu / se if se > 0 else 0.0 results.append((mu, t, label, P)) flag = ' *** BEATS MARKET ***' if t > 1.96 else '' print(f'{label:48s} {statistics.mean(v):9.5f} {mu:+9.5f} {t:+9.2f}{flag}') XM_tr, XM_te = np.hstack([Mtr, Xtr]), np.hstack([Mte, Xte]) for C in (0.003, 0.01, 0.03, 0.1, 0.3, 1.0): ev(f'market + ALL features, logistic C={C}', XM_tr, XM_te, LogisticRegression(C=C, max_iter=4000, multi_class='multinomial')) for depth, lr, it in ((2, 0.03, 400), (3, 0.05, 300), (4, 0.05, 300), (6, 0.03, 500)): ev(f'market + ALL, GBM d={depth} lr={lr}', XM_tr, XM_te, HistGradientBoostingClassifier(max_depth=depth, max_iter=it, learning_rate=lr, l2_regularization=1.0, random_state=0)) ev('market + ALL, RandomForest', XM_tr, XM_te, RandomForestClassifier(n_estimators=400, min_samples_leaf=25, n_jobs=-1, random_state=0)) ev('features ONLY, GBM d=4', Xtr, Xte, HistGradientBoostingClassifier(max_depth=4, max_iter=300, learning_rate=0.05, random_state=0)) # ensemble of the best few good = sorted(results, reverse=True)[:3] if len(good) >= 2: P = [[sum(g[3][i][j] for g in good) / len(good) for j in range(3)] for i in range(len(yte))] v = ll_vec(P) d = [base_v[i] - v[i] for i in range(len(v))] mu = statistics.mean(d) se = statistics.pstdev(d) / math.sqrt(len(d)) print(f'{"ENSEMBLE (top 3 averaged)":48s} {statistics.mean(v):9.5f} ' f'{mu:+9.5f} {mu/se:+9.2f}' f'{" *** BEATS MARKET ***" if mu/se > 1.96 else ""}') print(f'\nSuccess criterion set in advance: paired t > 1.96 vs the market.') win = [r for r in results if r[1] > 1.96] print(f'{len(win)} of {len(results)} configurations met it.') if not win: print('None. The closing line already contains everything these features know.') if __name__ == '__main__': main()