""" Is a team's result sequence (W/D/L) a *predictive* Markov chain -- and is it EXPLOITABLE vs the closing line? Three escalating tests on football-data.co.uk results (league_data/*.csv): A. Structure : pooled first-order transition matrix P(next | last) vs the unconditional base rate (chi-square test of independence). B. Prediction : train P(FTR | home_last, away_last) on seasons <= 2324, test on 2425+2526. Compare log-loss & Brier to (i) base rates and (ii) the de-vigged closing odds (the market). C. Exploit : on the test set, bet the outcome where the Markov model most disagrees with the market (higher prob), flat-staked at B365 closing odds. ROI + binomial significance. No look-ahead: a match's state is each team's PREVIOUS result within the same (league, season); train/test split is strictly by season. """ import csv, glob, os, math from collections import defaultdict, Counter DIR = r'E:\soccer\league_data' TRAIN_MAX = 2324 # seasons with code <= this are training TEST_SEASONS = {2425, 2526} def _season(path): b = os.path.basename(path) try: return int(b.split('_')[1].split('.')[0]) except Exception: return None def _f(x): try: return float(x) except (TypeError, ValueError): return None def _date(s): for fmt in ('%d/%m/%Y', '%d/%m/%y'): try: from datetime import datetime return datetime.strptime(s, fmt) except (ValueError, TypeError): pass return None def load(): """Return list of match dicts with each team's PRIOR result filled in.""" matches = [] for path in glob.glob(os.path.join(DIR, '*.csv')): season = _season(path); if season is None: continue league = os.path.basename(path).split('_')[0] rows = [] with open(path, newline='', encoding='latin-1') as fh: for r in csv.DictReader(fh): ftr = (r.get('FTR') or '').strip().upper() if ftr not in ('H','D','A'): continue d = _date(r.get('Date')) if d is None: continue rows.append((d, r, ftr, league, season)) rows.sort(key=lambda t: t[0]) last = {} # team -> their last result letter W/D/L (within this season) for d, r, ftr, lg, sea in rows: ht, at = r.get('HomeTeam'), r.get('AwayTeam') hp, ap = last.get(ht), last.get(at) # prior states (None if first game) matches.append(dict(d=d, league=lg, season=sea, ftr=ftr, home=ht, away=at, hp=hp, ap=ap, oh=_f(r.get('B365H')), od=_f(r.get('B365D')), oa=_f(r.get('B365A')))) # update each team's last result from their own perspective last[ht] = 'W' if ftr=='H' else ('D' if ftr=='D' else 'L') last[at] = 'W' if ftr=='A' else ('D' if ftr=='D' else 'L') return matches def devig(oh, od, oa): if not (oh and od and oa): return None ih, idr, ia = 1/oh, 1/od, 1/oa; s = ih+idr+ia return {'H': ih/s, 'D': idr/s, 'A': ia/s} # ---- A. structure: team-perspective next|last transition matrix ---- def test_A(matches): # build per-team sequences again, but as W/D/L from team perspective seqs = defaultdict(list) for m in sorted(matches, key=lambda x:x['d']): seqs[(m['league'],m['season'],m['home'])].append('W' if m['ftr']=='H' else ('D' if m['ftr']=='D' else 'L')) seqs[(m['league'],m['season'],m['away'])].append('W' if m['ftr']=='A' else ('D' if m['ftr']=='D' else 'L')) trans = defaultdict(Counter); base = Counter() for seq in seqs.values(): for a,b in zip(seq, seq[1:]): trans[a][b]+=1; base[b]+=1 states=['W','D','L']; tot=sum(base.values()) print('A. First-order result memory (team perspective), N transitions =', tot) print(' unconditional: ' + ' '.join(f'{s}={base[s]/tot:.3f}' for s in states)) chi=0.0 for a in states: na=sum(trans[a].values()) row=' '.join(f'{b}:{trans[a][b]/na:.3f}' for b in states) print(f' after {a} (n={na:5d}): {row}') for b in states: exp=na*base[b]/tot; obs=trans[a][b] if exp>0: chi+=(obs-exp)**2/exp df=(3-1)*(3-1) try: from scipy import stats; p=stats.chi2.sf(chi,df) except Exception: p=float('nan') print(f' chi-square vs independence: chi2={chi:.1f}, df={df}, p={p:.2e}') # ---- B & C ---- def test_BC(matches): train=[m for m in matches if m['season']<=TRAIN_MAX and m['hp'] and m['ap']] test =[m for m in matches if m['season'] in TEST_SEASONS and m['hp'] and m['ap']] # learn P(FTR | hp, ap) cnt=defaultdict(Counter) for m in train: cnt[(m['hp'],m['ap'])][m['ftr']]+=1 base=Counter(m['ftr'] for m in train); bt=sum(base.values()) baserate={k:base[k]/bt for k in ('H','D','A')} def markov_p(m): c=cnt[(m['hp'],m['ap'])]; n=sum(c.values()) if n<30: return dict(baserate) # fall back if too few return {k:(c[k]+baserate[k]*5)/(n+5) for k in ('H','D','A')} # mild smoothing # B. scoring def logloss_brier(preds): ll=br=0.0; n=0 for m,p in preds: y=m['ftr'] ll+=-math.log(max(p[y],1e-9)) br+=sum((p[k]-(1.0 if k==y else 0.0))**2 for k in ('H','D','A')) n+=1 return ll/n, br/n, n mk=[(m,markov_p(m)) for m in test] bs=[(m,dict(baserate)) for m in test] mq=[(m,devig(m['oh'],m['od'],m['oa'])) for m in test if devig(m['oh'],m['od'],m['oa'])] print('\nB. Out-of-sample prediction on', len(test), 'test matches (24/25+25/26):') for nm,pr in [('Markov (form)',mk),('Base rate ',bs),('Market (B365)',mq)]: ll,br,n=logloss_brier(pr) print(f' {nm}: n={n:5d} logloss={ll:.4f} brier={br:.4f}') # C. exploit vs market print('\nC. Exploitable vs closing line? (flat $1 bets at B365, test set)') staked=ret=wins=n=0 for m in test: mq_=devig(m['oh'],m['od'],m['oa']) if not mq_: continue mp=markov_p(m) # outcome where Markov most exceeds the market-implied prob pick=max(('H','D','A'), key=lambda k: mp[k]-mq_[k]) if mp[pick]-mq_[pick] <= 0: continue # no disagreement -> no bet odds={'H':m['oh'],'D':m['od'],'A':m['oa']}[pick] n+=1; staked+=1 if m['ftr']==pick: ret+=odds-1; wins+=1 else: ret-=1 roi=100*ret/staked if staked else 0 # binomial vs implied print(f' bets={n} wins={wins} net={ret:+.1f} units ROI={roi:+.1f}%') if __name__=='__main__': M=load() print(f'Loaded {len(M)} matches, seasons {min(m["season"] for m in M)}-{max(m["season"] for m in M)}, ' f'{len(set(m["league"] for m in M))} leagues\n') test_A(M) test_BC(M)