224 lines
7.4 KiB
Python
224 lines
7.4 KiB
Python
"""
|
|
Real AI Engine Backtest Script
|
|
==============================
|
|
Uses the ACTUAL models (V20/V25 Ensemble) to predict historical matches.
|
|
|
|
Usage:
|
|
python ai-engine/scripts/backtest_real.py
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import psycopg2
|
|
from psycopg2.extras import RealDictCursor
|
|
from datetime import datetime
|
|
|
|
# Add paths
|
|
AI_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT_DIR = os.path.dirname(AI_DIR)
|
|
sys.path.insert(0, ROOT_DIR)
|
|
|
|
# Fix for Windows path issues in scripts
|
|
if "scripts" in os.path.basename(AI_DIR):
|
|
ROOT_DIR = os.path.dirname(ROOT_DIR) # One level up if inside scripts folder
|
|
|
|
from services.single_match_orchestrator import get_single_match_orchestrator, MatchData
|
|
|
|
def get_clean_dsn() -> str:
|
|
return "postgresql://suggestbet:SuGGesT2026SecuRe@localhost:15432/boilerplate_db"
|
|
|
|
def run_backtest():
|
|
print("🚀 REAL AI BACKTEST: Sept 13, 2024 - Top Leagues")
|
|
print("🧠 Engine: V30 Ensemble (V20+V25)")
|
|
print("="*60)
|
|
|
|
# Load Top Leagues
|
|
leagues_path = os.path.join(ROOT_DIR, "top_leagues.json")
|
|
try:
|
|
with open(leagues_path, 'r') as f:
|
|
top_leagues = json.load(f)
|
|
league_ids = tuple(str(lid) for lid in top_leagues)
|
|
print(f"📋 Loaded {len(top_leagues)} top leagues.")
|
|
except Exception as e:
|
|
print(f"❌ Error loading top_leagues.json: {e}")
|
|
return
|
|
|
|
# Date Range (Sept 13, 2024)
|
|
start_dt = datetime(2024, 9, 13, 0, 0, 0)
|
|
end_dt = datetime(2024, 9, 13, 23, 59, 59)
|
|
start_ts = int(start_dt.timestamp() * 1000)
|
|
end_ts = int(end_dt.timestamp() * 1000)
|
|
|
|
dsn = get_clean_dsn()
|
|
conn = psycopg2.connect(dsn)
|
|
cur = conn.cursor(cursor_factory=RealDictCursor)
|
|
|
|
# Fetch Matches
|
|
cur.execute("""
|
|
SELECT m.id, m.match_name, m.home_team_id, m.away_team_id,
|
|
m.mst_utc, m.league_id, m.status, m.score_home, m.score_away,
|
|
t1.name as home_team, t2.name as away_team,
|
|
l.name as league_name
|
|
FROM matches m
|
|
LEFT JOIN teams t1 ON m.home_team_id = t1.id
|
|
LEFT JOIN teams t2 ON m.away_team_id = t2.id
|
|
LEFT JOIN leagues l ON m.league_id = l.id
|
|
WHERE m.mst_utc BETWEEN %s AND %s
|
|
AND m.league_id IN %s
|
|
AND m.status = 'FT'
|
|
ORDER BY m.mst_utc ASC
|
|
LIMIT 20 -- Limit to 20 matches to avoid running for hours on a single backtest
|
|
""", (start_ts, end_ts, league_ids))
|
|
|
|
rows = cur.fetchall()
|
|
print(f"📊 Found {len(rows)} finished matches. Starting AI Analysis...")
|
|
|
|
if not rows:
|
|
print("⚠️ No matches found for this date.")
|
|
cur.close()
|
|
conn.close()
|
|
return
|
|
|
|
# Initialize AI Engine
|
|
try:
|
|
orchestrator = get_single_match_orchestrator()
|
|
print("✅ AI Engine (SingleMatchOrchestrator) Loaded.")
|
|
except Exception as e:
|
|
print(f"❌ Failed to load AI Engine: {e}")
|
|
print("💡 Make sure models are trained/present in ai-engine/models/")
|
|
cur.close()
|
|
conn.close()
|
|
return
|
|
|
|
# ─── Backtest Loop ───
|
|
total_matches_analyzed = 0
|
|
bets_skipped = 0
|
|
bets_played = 0
|
|
bets_won = 0
|
|
total_profit = 0.0
|
|
|
|
# Thresholds matching the NEW Skip Logic
|
|
MIN_CONF = 45.0
|
|
|
|
start_time = time.time()
|
|
|
|
for i, row in enumerate(rows):
|
|
match_id = str(row['id'])
|
|
home_team = row['home_team']
|
|
away_team = row['away_team']
|
|
home_score = row['score_home']
|
|
away_score = row['score_away']
|
|
|
|
print(f"\n[{i+1}/{len(rows)}] Analyzing: {home_team} vs {away_team} ...")
|
|
|
|
try:
|
|
# 1. AI PREDICTION (Actual Model Call)
|
|
prediction = orchestrator.analyze_match(match_id)
|
|
|
|
if not prediction:
|
|
print(f" ⚠️ AI returned no prediction.")
|
|
continue
|
|
|
|
total_matches_analyzed += 1
|
|
|
|
# 2. Extract Main Pick
|
|
main_pick = prediction.get("main_pick") or {}
|
|
pick_name = main_pick.get("pick")
|
|
confidence = main_pick.get("confidence", 0)
|
|
odds = main_pick.get("odds", 0)
|
|
|
|
if not pick_name or not confidence:
|
|
print(f" ⚠️ No main pick found in prediction.")
|
|
continue
|
|
|
|
print(f" 🤖 Pick: {pick_name} | Conf: {confidence}% | Odds: {odds}")
|
|
|
|
# 3. Apply Skip Logic (New Backtest Logic)
|
|
if confidence < MIN_CONF:
|
|
print(f" 🚫 SKIPPED (Confidence {confidence}% < {MIN_CONF}%)")
|
|
bets_skipped += 1
|
|
continue
|
|
|
|
if odds > 0:
|
|
implied_prob = 1.0 / odds
|
|
my_prob = confidence / 100.0
|
|
if my_prob - implied_prob < -0.03: # Negative edge
|
|
print(f" 🚫 SKIPPED (Negative Edge)")
|
|
bets_skipped += 1
|
|
continue
|
|
|
|
# 4. Bet Played
|
|
bets_played += 1
|
|
print(f" 🎲 BET PLAYED: {pick_name} @ {odds}")
|
|
|
|
# 5. Resolve Bet
|
|
won = False
|
|
# Basic resolution logic (Need to parse pick_name like "1", "X", "2", "2.5 Üst", etc.)
|
|
pick_clean = str(pick_name).upper()
|
|
|
|
# MS
|
|
if pick_clean in ["1", "MS 1"] and home_score > away_score: won = True
|
|
elif pick_clean in ["X", "MS X"] and home_score == away_score: won = True
|
|
elif pick_clean in ["2", "MS 2"] and away_score > home_score: won = True
|
|
|
|
# OU25
|
|
elif "ÜST" in pick_clean or "OVER" in pick_clean:
|
|
if (home_score + away_score) > 2.5: won = True
|
|
elif "ALT" in pick_clean or "UNDER" in pick_clean:
|
|
if (home_score + away_score) < 2.5: won = True
|
|
|
|
# BTTS
|
|
elif "VAR" in pick_clean and home_score > 0 and away_score > 0: won = True
|
|
elif "YOK" in pick_clean and (home_score == 0 or away_score == 0): won = True
|
|
|
|
if won:
|
|
bets_won += 1
|
|
profit = odds - 1.0
|
|
print(f" ✅ WON! (+{profit:.2f} units)")
|
|
else:
|
|
profit = -1.0
|
|
print(f" ❌ LOST! (-1.00 units)")
|
|
|
|
total_profit += profit
|
|
|
|
except Exception as e:
|
|
print(f" 💥 Error during analysis: {e}")
|
|
|
|
elapsed = time.time() - start_time
|
|
|
|
# ─── FINAL REPORT ───
|
|
print("\n" + "="*60)
|
|
print("📈 REAL AI BACKTEST RESULTS")
|
|
print(f"🕒 Time taken: {elapsed:.1f} seconds")
|
|
print("="*60)
|
|
print(f"📊 Matches Analyzed: {total_matches_analyzed}")
|
|
print(f"🚫 Bets SKIPPED: {bets_skipped}")
|
|
print(f"✅ Bets PLAYED: {bets_played}")
|
|
|
|
if bets_played > 0:
|
|
win_rate = (bets_won / bets_played) * 100
|
|
roi = (total_profit / bets_played) * 100
|
|
yield_val = total_profit # Net Units
|
|
|
|
print(f"🏆 Bets Won: {bets_won}")
|
|
print(f"💀 Bets Lost: {bets_played - bets_won}")
|
|
print("-" * 40)
|
|
print(f" Win Rate: {win_rate:.2f}%")
|
|
print(f"💰 Total Profit (Units): {total_profit:.2f}")
|
|
print(f"📊 ROI: {roi:.2f}%")
|
|
|
|
if roi > 0:
|
|
print("🟢 STRATEGY IS PROFITABLE!")
|
|
else:
|
|
print("🔴 STRATEGY IS LOSING")
|
|
else:
|
|
print("⚠️ No bets were played. All were skipped or failed.")
|
|
|
|
cur.close()
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
run_backtest()
|