Files
turf_saas/backtest.py
2026-04-25 17:18:43 +02:00

326 lines
9.6 KiB
Python

#!/usr/bin/env python3
"""
Backtest - Analyse des performances des prédictions
Calcule ROI, précision, et métriques par type de pari
"""
import sqlite3
import os
from datetime import datetime, timedelta
from collections import defaultdict
DB_PATH = os.environ.get("DB_PATH", "/home/h3r7/turf_scraper/turf.db")
def get_db():
return sqlite3.connect(DB_PATH)
def calculate_backtest(start_date=None, end_date=None):
"""Calcule les statistiques de backtest"""
conn = get_db()
if not start_date:
start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
if not end_date:
end_date = datetime.now().strftime('%Y-%m-%d')
# Récupérer toutes les recommandations avec résultats
query = """
SELECT
r.date,
r.race_name,
r.type_pari,
r.cheval1,
r.cheval2,
r.numero1,
r.numero2,
r.cote,
r.mise,
r.resultat,
r.gain_potentiel,
h.position as position_reel
FROM recommendations r
LEFT JOIN results h ON h.date = r.date
AND h.race_name LIKE '%' || SUBSTR(r.race_name, 1, 20) || '%'
AND h.horse_name = r.cheval1
WHERE r.resultat IS NOT NULL
AND r.resultat != ''
AND r.date BETWEEN ? AND ?
"""
cursor = conn.execute(query, (start_date, end_date))
rows = cursor.fetchall()
if not rows:
conn.close()
return {
'summary': {
'total_bets': 0,
'message': 'Aucune donnée disponible pour cette période'
}
}
# Métriques globales
stats = {
'total_bets': len(rows),
'gagne': 0,
'perdu': 0,
'mise_totale': 0,
'gain_total': 0,
'by_type': defaultdict(lambda: {
'count': 0, 'gagne': 0, 'mise': 0, 'gain': 0
})
}
details = []
for row in rows:
date, race_name, type_pari, cheval1, cheval2, numero1, numero2, cote, mise, resultat, gain_potentiel, position = row
mise = float(mise or 1)
cote = float(cote or 1)
stats['mise_totale'] += mise
stats['by_type'][type_pari]['mise'] += mise
if resultat == 'GAGNE':
stats['gagne'] += 1
stats['perdu'] += 0
stats['gain_total'] += mise * cote
stats['by_type'][type_pari]['gagne'] += 1
stats['by_type'][type_pari]['gain'] += mise * cote
result = 'GAGNE'
else:
stats['perdu'] += 1
stats['gain_total'] += 0
result = 'PERDU'
details.append({
'date': date,
'race_name': race_name[:30] if race_name else '',
'type_pari': type_pari,
'cheval': cheval1,
'cote': cote,
'mise': mise,
'resultat': result,
'gain': mise * cote if result == 'GAGNE' else 0
})
# Calculer ROI
roi = 0
if stats['mise_totale'] > 0:
roi = ((stats['gain_total'] - stats['mise_totale']) / stats['mise_totale']) * 100
# Calculer précision
precision = 0
if stats['total_bets'] > 0:
precision = (stats['gagne'] / stats['total_bets']) * 100
# Métriques par type de pari
by_type_results = {}
for pari_type, data in stats['by_type'].items():
pari_roi = 0
if data['mise'] > 0:
pari_roi = ((data['gain'] - data['mise']) / data['mise']) * 100
pari_precision = 0
if data['count'] > 0:
pari_precision = (data['gagne'] / data['count']) * 100
by_type_results[pari_type] = {
'count': data['count'],
'gagne': data['gagne'],
'mise': data['mise'],
'gain': data['gain'],
'roi': round(pari_roi, 1),
'precision': round(pari_precision, 1)
}
# Précision par rang de prédiction
precision_by_rank = get_precision_by_rank(conn, start_date, end_date)
conn.close()
return {
'period': {
'start': start_date,
'end': end_date
},
'summary': {
'total_bets': stats['total_bets'],
'gagne': stats['gagne'],
'perdu': stats['perdu'],
'precision': round(precision, 1),
'mise_totale': round(stats['mise_totale'], 2),
'gain_total': round(stats['gain_total'], 2),
'roi': round(roi, 1)
},
'by_type': by_type_results,
'precision_by_rank': precision_by_rank,
'details': details[-50:] # Last 50 bets
}
def get_precision_by_rank(conn, start_date, end_date):
"""Calcule la précision par rang de prédiction"""
query = """
SELECT
p.prediction_rank,
COUNT(*) as total,
SUM(CASE WHEN p.prediction_rank = r.position THEN 1 ELSE 0 END) as hits,
SUM(CASE WHEN r.position <= 3 THEN 1 ELSE 0 END) as top3
FROM predictions p
INNER JOIN results r ON r.date = p.date
AND r.horse_name = p.horse_name
WHERE p.date BETWEEN ? AND ?
AND p.prediction_rank > 0
AND r.position > 0
GROUP BY p.prediction_rank
ORDER BY p.prediction_rank
"""
cursor = conn.execute(query, (start_date, end_date))
rows = cursor.fetchall()
results = {}
for rank, total, hits, top3 in rows:
precision = (hits / total * 100) if total > 0 else 0
top3_rate = (top3 / total * 100) if total > 0 else 0
results[f'rank_{rank}'] = {
'total': total,
'hits': hits,
'precision': round(precision, 1),
'top3': top3,
'top3_rate': round(top3_rate, 1)
}
return results
def get_daily_stats(days=30):
"""Statistiques quotidiennes sur les N derniers jours"""
conn = get_db()
end_date = datetime.now().strftime('%Y-%m-%d')
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
query = """
SELECT
date,
COUNT(*) as total_bets,
SUM(CASE WHEN resultat = 'GAGNE' THEN 1 ELSE 0 END) as gagne,
SUM(CASE WHEN resultat = 'PERDU' THEN 1 ELSE 0 END) as perdu,
SUM(CASE WHEN resultat = 'GAGNE' THEN mise * cote ELSE 0 END) as gains,
SUM(mise) as mises
FROM recommendations
WHERE date BETWEEN ? AND ?
AND resultat IS NOT NULL
AND resultat != ''
GROUP BY date
ORDER BY date DESC
"""
cursor = conn.execute(query, (start_date, end_date))
rows = cursor.fetchall()
daily = []
for date, total, gagne, perdu, gains, mises in rows:
roi = ((gains - mises) / mises * 100) if mises > 0 else 0
precision = (gagne / total * 100) if total > 0 else 0
daily.append({
'date': date,
'total_bets': total,
'gagne': gagne,
'perdu': perdu,
'precision': round(precision, 1),
'mises': round(mises, 2),
'gains': round(gains, 2),
'roi': round(roi, 1)
})
conn.close()
return daily
def get_weekly_summary():
"""Résumé hebdomadaire"""
conn = get_db()
# Semaine en cours
today = datetime.now()
week_start = today - timedelta(days=today.weekday())
week_start_str = week_start.strftime('%Y-%m-%d')
query = """
SELECT
COUNT(*) as total,
SUM(CASE WHEN resultat = 'GAGNE' THEN 1 ELSE 0 END) as gagne,
SUM(mise) as mise,
SUM(CASE WHEN resultat = 'GAGNE' THEN mise * cote ELSE 0 END) as gain
FROM recommendations
WHERE date >= ? AND resultat IS NOT NULL AND resultat != ''
"""
cursor = conn.execute(query, (week_start_str,))
row = cursor.fetchone()
total, gagne, mise, gain = row
weekly = {
'week_start': week_start_str,
'total_bets': total or 0,
'gagne': gagne or 0,
'mise': round(mise or 0, 2),
'gain': round(gain or 0, 2),
'roi': round(((gain - mise) / mise * 100) if mise and mise > 0 else 0, 1),
'precision': round((gagne / total * 100) if total and total > 0 else 0, 1)
}
# Semaine dernière
last_week_start = week_start - timedelta(days=7)
last_week_end = week_start - timedelta(days=1)
query = """
SELECT
COUNT(*) as total,
SUM(CASE WHEN resultat = 'GAGNE' THEN 1 ELSE 0 END) as gagne,
SUM(mise) as mise,
SUM(CASE WHEN resultat = 'GAGNE' THEN mise * cote ELSE 0 END) as gain
FROM recommendations
WHERE date BETWEEN ? AND ?
AND resultat IS NOT NULL
AND resultat != ''
"""
cursor = conn.execute(query, (last_week_start.strftime('%Y-%m-%d'), last_week_end.strftime('%Y-%m-%d')))
row = cursor.fetchone()
total, gagne, mise, gain = row
weekly['last_week'] = {
'total_bets': total or 0,
'gagne': gagne or 0,
'mise': round(mise or 0, 2),
'gain': round(gain or 0, 2),
'roi': round(((gain - mise) / mise * 100) if mise and mise > 0 else 0, 1),
'precision': round((gagne / total * 100) if total and total > 0 else 0, 1)
}
conn.close()
return weekly
if __name__ == "__main__":
import json
print("="*60)
print("BACKTEST - Analyse des performances")
print("="*60)
# Test
result = calculate_backtest()
print(json.dumps(result, indent=2, default=str))