""" Does NON-PRICE information (API-Football team stats) beat the market? (read-only) THE QUESTION. Everything mined so far has been derived from prices, and [[project_edge_mine_exhausted]] showed that whole space is structurally dead: predictable movement sits exactly where the spread is widest. The one untouched hypothesis is genuinely different information -- form, shots, possession, discipline -- and we already hold 70k team-match rows of it in raw_api/. THE ONLY TEST THAT MATTERS. Not "can a model predict football" -- of course it can, badly. The question is whether it knows something the CLOSING PRICE does not. So: baseline = Pinnacle closing odds, devigged (the market) model = API features alone combined = API features + the market's own probability If `combined` does not beat `baseline` out-of-sample on log loss, the features add nothing the price has not already absorbed, and there is no edge to trade regardless of how good the model looks in isolation. DISCIPLINE. - Rolling features are .shift(1) lagged in refresh_api_football.build_features, so no same-match leakage. Verified before writing this. - TIME-BASED split (train on earlier seasons, test on later). Random CV would leak: teams' form is autocorrelated and the market improves over time. - Any edge found must then clear the ~0.67c half-spread -- see project_xlow_spread_verdict. Log-loss improvement alone is not money. Usage: python research\\nonprice_edge_test.py """ import sys import csv import glob import math from collections import defaultdict from datetime import datetime sys.path.insert(0, r'E:\soccer\main') import daily_signals as DS FEAT = r'E:\soccer\raw_api\match_features.csv' LEAGUE = r'E:\soccer\league_data\*.csv' IDX = {'H': 0, 'D': 1, 'A': 2} def _f(x): try: v = float(x) return v if v == v else None except (TypeError, ValueError): return None def norm(s): return (s or '').strip().lower().replace('.', '').replace('-', ' ') def load_odds(): """(date, home_fd, away_fd) -> devigged Pinnacle closing probs.""" out = {} for p in glob.glob(LEAGUE): try: with open(p, newline='', encoding='utf-8', errors='replace') as fh: for r in csv.DictReader(fh): res = (r.get('FTR') or '').strip().upper() if res not in IDX: continue o = [_f(r.get(c)) for c in ('PSCH', 'PSCD', 'PSCA')] if not all(o) or any(v <= 1 for v in o): o = [_f(r.get(c)) for c in ('PSH', 'PSD', 'PSA')] if not all(o) or any(v <= 1 for v in o): continue d = (r.get('Date') or '').strip() for fmt in ('%d/%m/%Y', '%d/%m/%y'): try: dt = datetime.strptime(d, fmt).date() break except ValueError: dt = None if not dt: continue imp = [1 / v for v in o] s = sum(imp) out[(dt, norm(r.get('HomeTeam')), norm(r.get('AwayTeam')))] = \ ([v / s for v in imp], res) except Exception: continue return out def load_features(): """fixture -> (home_feats, away_feats, meta). One row per fixture.""" byfix = defaultdict(dict) cols = None with open(FEAT, newline='', encoding='utf-8') as fh: for r in csv.DictReader(fh): if cols is None: cols = sorted(c for c in r if c.startswith('api_')) side = 'h' if r.get('is_home') in ('1', '1.0', 'True') else 'a' byfix[r['fixture_id']][side] = r out = [] for fid, d in byfix.items(): if 'h' not in d or 'a' not in d: continue h, a = d['h'], d['a'] try: dt = datetime.strptime(h['date'][:10], '%Y-%m-%d').date() except ValueError: continue hv = [_f(h.get(c)) for c in cols] av = [_f(a.get(c)) for c in cols] if any(v is None for v in hv) or any(v is None for v in av): continue out.append(dict(fid=fid, date=dt, div=h.get('div'), home=h.get('home_team'), away=h.get('away_team'), hv=hv, av=av)) return out, cols def main(): odds = load_odds() feats, cols = load_features() print(f'odds rows: {len(odds)} feature fixtures: {len(feats)}') rows = [] for f in feats: key = (f['date'], norm(DS.fdco_name(f['home'])), norm(DS.fdco_name(f['away']))) hit = odds.get(key) if not hit: continue p, res = hit rows.append(dict(date=f['date'], p=p, res=res, x=f['hv'] + f['av'] + [x - y for x, y in zip(f['hv'], f['av'])])) print(f'MATCHED fixtures (features + Pinnacle + result): {len(rows)}') if len(rows) < 500: print('too few matched -- name translation is the bottleneck, not the idea') return rows.sort(key=lambda r: r['date']) 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') import numpy as np from sklearn.ensemble import HistGradientBoostingClassifier def ll(probs, ys): return -sum(math.log(max(p[IDX[y]], 1e-9)) for p, y in zip(probs, ys)) / len(ys) ytr = [r['res'] for r in tr] yte = [r['res'] for r in te] Xtr = np.array([r['x'] for r in tr]) Xte = np.array([r['x'] for r in te]) Mtr = np.array([r['p'] for r in tr]) Mte = np.array([r['p'] for r in te]) base = ll([r['p'] for r in te], yte) print(f'{"model":34s} {"test log loss":>14s} vs market') print(f'{"MARKET (Pinnacle closing)":34s} {base:14.5f} --') for label, A, B in (('API features ALONE', Xtr, Xte), ('API features + market prob', np.hstack([Xtr, Mtr]), np.hstack([Xte, Mte]))): m = HistGradientBoostingClassifier(max_depth=4, max_iter=300, learning_rate=0.06, random_state=0) m.fit(A, ytr) pr = m.predict_proba(B) order = list(m.classes_) probs = [[pr[i][order.index(k)] for k in ('H', 'D', 'A')] for i in range(len(yte))] v = ll(probs, yte) print(f'{label:34s} {v:14.5f} {base - v:+.5f}' f'{" BEATS MARKET" if v < base else ""}') print('\nA negative delta means the model is WORSE than simply reading the price.') print('Only "features + market" beating "market" is evidence of new information;') print('and even then it must clear the ~0.67c half-spread to be tradeable.') if __name__ == '__main__': main()