cbazzaandClaude Sonnet 5
ec7290ece5
AutoUpdateAndLearn.update_data() now falls back to LottoWebUpdater (lotto.de scraping) when the API chain (Lottoland/GitHub/lottoAPI) returns no new data, matching the pattern already in production on the Eurojackpot sibling project. Required installing beautifulsoup4 into the venv - its absence would have crashed the entire update script the moment the fallback import was added, not just the fallback path itself, so this needed verifying before wiring it in. Also adds tests/test_popularity_score.py (pytest): regression coverage for the popularity/EV score against known overplayed combinations (1-2-3-4-5-6, 5-10-15-20-25-30) and the Superzahl EV ordering. Verified against the pre-fix code (commit8c2089d) that these tests would actually have caught the sum-extremity bug fixed ine05f1f1- this had previously only been caught by manual spot-checking. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
488 lines
16 KiB
Python
Executable File
488 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Automatischer Lotto 6aus49 Update & Learning Workflow
|
||
|
||
Nach jeder Ziehung:
|
||
1. Aktualisiert Daten von API
|
||
2. Evaluiert letzte generierte Tipps
|
||
3. Real-Time Learning Update
|
||
4. Performance-Report
|
||
|
||
Verwendung:
|
||
python auto_update_and_learn.py
|
||
|
||
Oder als Cronjob (nach Ziehung):
|
||
0 21 * * 3,0 cd /path/to/Lotto && python scripts/automation/auto_update_and_learn.py
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
from datetime import datetime, timedelta
|
||
import json
|
||
import pandas as pd
|
||
|
||
# Path Setup
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
project_dir = os.path.dirname(os.path.dirname(script_dir))
|
||
sys.path.insert(0, project_dir)
|
||
|
||
from scripts.utils.update_from_api import LottoAPIUpdater
|
||
from scripts.utils.update_from_web import LottoWebUpdater
|
||
from scripts.generators.ultimate_ai_ml_hybrid_generator import UltimateAIMLHybridGenerator
|
||
from scripts.utils.notifier import LottoNotifier
|
||
|
||
|
||
class AutoUpdateAndLearn:
|
||
"""Automatisches Update und Learning System."""
|
||
|
||
def __init__(self, data_dir: str):
|
||
self.data_dir = data_dir
|
||
# CSV file is in Lotto/data folder
|
||
# data_dir = .../Lotto/data
|
||
# We need .../Lotto/data/AlleLottozahlen.csv
|
||
self.data_file = os.path.join(data_dir, "AlleLottozahlen.csv")
|
||
self.tips_dir = os.path.join(data_dir, "generated_tips")
|
||
self.reports_dir = os.path.join(data_dir, "performance_reports")
|
||
self.learning_log = os.path.join(data_dir, "learning_log.json")
|
||
|
||
os.makedirs(self.reports_dir, exist_ok=True)
|
||
|
||
# Initialisiere Notifier
|
||
self.notifier = LottoNotifier()
|
||
|
||
print("🤖 AUTOMATISCHES UPDATE & LEARNING SYSTEM - LOTTO 6AUS49")
|
||
print("=" * 70)
|
||
|
||
def load_learning_log(self) -> dict:
|
||
"""Lädt Learning Log."""
|
||
if os.path.exists(self.learning_log):
|
||
with open(self.learning_log, 'r') as f:
|
||
return json.load(f)
|
||
return {"updates": []}
|
||
|
||
def save_learning_log(self, log: dict):
|
||
"""Speichert Learning Log."""
|
||
with open(self.learning_log, 'w') as f:
|
||
json.dump(log, f, indent=2)
|
||
|
||
def check_for_new_draw(self) -> dict:
|
||
"""
|
||
Prüft ob neue Ziehung verfügbar ist.
|
||
|
||
Returns:
|
||
Dict mit neuer Ziehung oder None
|
||
"""
|
||
print("\n🔍 PRÜFE AUF NEUE ZIEHUNG")
|
||
print("=" * 70)
|
||
|
||
log = self.load_learning_log()
|
||
|
||
# Lade aktuelle Daten
|
||
try:
|
||
df = pd.read_csv(self.data_file, sep=';')
|
||
df['datum'] = pd.to_datetime(df['datum'], format='%Y-%m-%d')
|
||
|
||
# Sortiere nach Datum absteigend
|
||
df = df.sort_values('datum', ascending=False)
|
||
|
||
latest_draw = df.iloc[0] # Neueste Ziehung
|
||
latest_date = latest_draw['datum']
|
||
|
||
print(f" 📅 Neueste Ziehung in Daten: {latest_date.strftime('%Y-%m-%d')}")
|
||
|
||
# Prüfe ob schon verarbeitet
|
||
if log["updates"]:
|
||
last_processed = log["updates"][-1].get("draw_date")
|
||
if last_processed == latest_date.strftime('%Y-%m-%d'):
|
||
print(f" ⏭️ Bereits verarbeitet")
|
||
return None
|
||
|
||
print(f" ✅ Neue Ziehung gefunden!")
|
||
|
||
return {
|
||
'date': latest_date,
|
||
'Z1': int(latest_draw['Z1']),
|
||
'Z2': int(latest_draw['Z2']),
|
||
'Z3': int(latest_draw['Z3']),
|
||
'Z4': int(latest_draw['Z4']),
|
||
'Z5': int(latest_draw['Z5']),
|
||
'Z6': int(latest_draw['Z6']),
|
||
'SZ': int(latest_draw['SZ']) if pd.notna(latest_draw['SZ']) else None
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f" ❌ Fehler beim Prüfen: {e}")
|
||
return None
|
||
|
||
def update_data(self) -> bool:
|
||
"""
|
||
Aktualisiert Daten von API.
|
||
|
||
Returns:
|
||
True bei Erfolg
|
||
"""
|
||
print("\n📥 AKTUALISIERE DATEN VON API")
|
||
print("=" * 70)
|
||
|
||
try:
|
||
updater = LottoAPIUpdater(self.data_file)
|
||
# 'all': Lottoland zuerst (liefert die neueste Ziehung meist noch am
|
||
# Ziehungstag), GitHub-Archiv als Quelle für die Historie/Fallback -
|
||
# das GitHub-Archiv allein hinkt wiederholt tagelang hinterher.
|
||
success = updater.update(api_name='all', create_backup=True)
|
||
|
||
if success:
|
||
print(" ✅ Daten erfolgreich aktualisiert")
|
||
else:
|
||
print(" ⚠️ API-Update ohne neue Daten, versuche Web-Scraper...")
|
||
try:
|
||
scraper = LottoWebUpdater(self.data_file)
|
||
success = scraper.update(create_backup=False)
|
||
if success:
|
||
print(" ✅ Web-Scraper erfolgreich")
|
||
else:
|
||
print(" ⚠️ Web-Scraper ohne neue Daten")
|
||
except Exception as scrape_err:
|
||
print(f" ❌ Web-Scraper Fehler: {scrape_err}")
|
||
|
||
return success
|
||
|
||
except Exception as e:
|
||
print(f" ❌ Fehler beim Update: {e}")
|
||
return False
|
||
|
||
def evaluate_tips(self, new_draw: dict) -> dict:
|
||
"""
|
||
Evaluiert letzte generierte Tipps gegen neue Ziehung.
|
||
|
||
Args:
|
||
new_draw: Dict mit neuer Ziehung
|
||
|
||
Returns:
|
||
Evaluierungs-Ergebnisse
|
||
"""
|
||
print("\n🎯 EVALUIERE LETZTE TIPPS")
|
||
print("=" * 70)
|
||
|
||
# Finde Tipps-Datei die VOR der Ziehung generiert wurde
|
||
if not os.path.exists(self.tips_dir):
|
||
print(" ⚠️ Keine Tipps zum Evaluieren")
|
||
return {}
|
||
|
||
tip_files = sorted(
|
||
[f for f in os.listdir(self.tips_dir) if f.endswith('.csv')],
|
||
reverse=True
|
||
)
|
||
|
||
if not tip_files:
|
||
print(" ⚠️ Keine Tipps-Dateien gefunden")
|
||
return {}
|
||
|
||
# Finde Tip-Datei die vor der Ziehung erstellt wurde
|
||
draw_date = new_draw['date']
|
||
selected_tip_file = None
|
||
|
||
for tip_file in tip_files:
|
||
# Parse Timestamp aus Dateiname: weekly_lotto_tips_YYYYMMDD_HHMMSS.csv
|
||
try:
|
||
parts = tip_file.replace('.csv', '').split('_')
|
||
tip_date_str = parts[-2] # YYYYMMDD
|
||
tip_date = pd.to_datetime(tip_date_str, format='%Y%m%d')
|
||
|
||
# Nehme erste Datei die vor der Ziehung war
|
||
if tip_date < draw_date:
|
||
selected_tip_file = tip_file
|
||
break
|
||
except:
|
||
continue
|
||
|
||
if not selected_tip_file:
|
||
# Fallback: nehme älteste Datei
|
||
selected_tip_file = tip_files[-1]
|
||
|
||
latest_tips_file = os.path.join(self.tips_dir, selected_tip_file)
|
||
print(f" 📁 Evaluiere: {selected_tip_file}")
|
||
|
||
try:
|
||
tips_df = pd.read_csv(latest_tips_file)
|
||
|
||
# Extrahiere gezogene Zahlen
|
||
drawn_main = [new_draw[f'Z{i}'] for i in range(1, 7)]
|
||
drawn_sz = new_draw.get('SZ')
|
||
|
||
print(f" 🎲 Gezogene Zahlen: {drawn_main} + SZ: {drawn_sz}")
|
||
print()
|
||
|
||
results = []
|
||
best_matches = {'main': 0, 'sz': False, 'tip': None}
|
||
|
||
for _, tip in tips_df.iterrows():
|
||
# Parse Hauptzahlen (Column name is 'Numbers' in Lotto)
|
||
main_str = tip['Numbers'] if 'Numbers' in tip else tip.get('Main_Numbers', '')
|
||
|
||
# Handle both formats: "[1, 2, 3]" and "1-2-3"
|
||
if main_str.startswith('['):
|
||
# Parse Python list format
|
||
import ast
|
||
tip_main = ast.literal_eval(main_str)
|
||
else:
|
||
# Parse dash-separated format
|
||
tip_main = [int(n) for n in main_str.split('-')]
|
||
|
||
# Parse Superzahl
|
||
tip_sz = int(tip['Superzahl'])
|
||
|
||
# Zähle Treffer
|
||
main_matches = len(set(tip_main) & set(drawn_main))
|
||
sz_match = (tip_sz == drawn_sz) if drawn_sz is not None else False
|
||
|
||
results.append({
|
||
'tip_number': tip['Tip_Number'],
|
||
'strategy': tip['Strategy'],
|
||
'main_matches': main_matches,
|
||
'sz_match': sz_match,
|
||
'total_score': main_matches + (1 if sz_match else 0)
|
||
})
|
||
|
||
# Track best
|
||
total_score = main_matches + (1 if sz_match else 0)
|
||
best_total = best_matches['main'] + (1 if best_matches['sz'] else 0)
|
||
|
||
if total_score > best_total:
|
||
best_matches = {
|
||
'main': main_matches,
|
||
'sz': sz_match,
|
||
'tip': tip['Tip_Number']
|
||
}
|
||
|
||
# Ausgabe
|
||
print(f" {'Tip':<5} {'Strategie':<15} {'Main':<6} {'SZ':<5} {'Score':<7} {'Bewertung'}")
|
||
print(" " + "-" * 60)
|
||
|
||
for r in results:
|
||
rating = self._get_match_rating(r['main_matches'], r['sz_match'])
|
||
sz_indicator = "✅" if r['sz_match'] else "⚪"
|
||
print(f" #{r['tip_number']:<4} {r['strategy']:<15} "
|
||
f"{r['main_matches']:<6} {sz_indicator:<5} "
|
||
f"{r['total_score']:<7} {rating}")
|
||
|
||
print()
|
||
print(f" 🏆 Bester Tipp: #{best_matches['tip']} "
|
||
f"({best_matches['main']} Main" +
|
||
(f" + SZ" if best_matches['sz'] else "") + ")")
|
||
|
||
return {
|
||
'file': tip_files[0],
|
||
'results': results,
|
||
'best': best_matches,
|
||
'avg_main_matches': sum(r['main_matches'] for r in results) / len(results),
|
||
'sz_match_rate': sum(1 for r in results if r['sz_match']) / len(results)
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f" ❌ Fehler bei Evaluation: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return {}
|
||
|
||
def _get_match_rating(self, main: int, sz_match: bool) -> str:
|
||
"""Bewertung der Treffer."""
|
||
if main == 6 and sz_match:
|
||
return "🏆 JACKPOT!"
|
||
elif main == 6:
|
||
return "💰 Klasse 2"
|
||
elif main == 5 and sz_match:
|
||
return "💰 Klasse 3"
|
||
elif main == 5:
|
||
return "💰 Klasse 4"
|
||
elif main == 4 and sz_match:
|
||
return "💵 Klasse 5"
|
||
elif main == 4:
|
||
return "💵 Klasse 6"
|
||
elif main == 3 and sz_match:
|
||
return "✅ Klasse 7"
|
||
elif main == 3:
|
||
return "✅ Klasse 8"
|
||
elif main == 2 and sz_match:
|
||
return "👍 Klasse 9"
|
||
elif main >= 2:
|
||
return "👍 OK"
|
||
else:
|
||
return "⚪ Niedrig"
|
||
|
||
def perform_learning_update(self, new_draw: dict) -> bool:
|
||
"""
|
||
Führt Real-Time Learning Update durch.
|
||
|
||
Args:
|
||
new_draw: Dict mit neuer Ziehung
|
||
|
||
Returns:
|
||
True bei Erfolg
|
||
"""
|
||
print("\n🧠 REAL-TIME LEARNING UPDATE")
|
||
print("=" * 70)
|
||
|
||
try:
|
||
# Initialisiere Generator
|
||
generator = UltimateAIMLHybridGenerator(
|
||
self.data_file,
|
||
fast_mode=True
|
||
)
|
||
|
||
# Learning Update (falls vorhanden)
|
||
if hasattr(generator, 'real_time_learner'):
|
||
generator.real_time_learner.learn_from_result(new_draw)
|
||
print(" ✅ Learning Update durchgeführt")
|
||
|
||
# Performance Tracking (falls vorhanden)
|
||
if hasattr(generator, 'performance_tracker'):
|
||
generator.performance_tracker.evaluate_predictions(new_draw)
|
||
print(" 📊 Performance getrackt")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" ⚠️ Learning Update nicht verfügbar: {e}")
|
||
print(" ℹ️ Generator funktioniert weiterhin normal")
|
||
return True
|
||
|
||
def generate_report(self, new_draw: dict, evaluation: dict):
|
||
"""Generiert Performance-Report."""
|
||
print("\n📊 PERFORMANCE-REPORT")
|
||
print("=" * 70)
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
report_file = os.path.join(
|
||
self.reports_dir,
|
||
f"report_{timestamp}.txt"
|
||
)
|
||
|
||
report = []
|
||
report.append("=" * 70)
|
||
report.append("LOTTO 6AUS49 PERFORMANCE REPORT")
|
||
report.append("=" * 70)
|
||
report.append(f"Erstellt: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
report.append("")
|
||
report.append(f"ZIEHUNG VOM {new_draw['date'].strftime('%Y-%m-%d')}")
|
||
report.append("-" * 70)
|
||
report.append(f"Hauptzahlen: {[new_draw[f'Z{i}'] for i in range(1, 7)]}")
|
||
report.append(f"Superzahl: {new_draw.get('SZ', 'N/A')}")
|
||
report.append("")
|
||
|
||
if evaluation:
|
||
report.append("EVALUATION DER TIPPS")
|
||
report.append("-" * 70)
|
||
report.append(f"Datei: {evaluation['file']}")
|
||
report.append(f"Tipps evaluiert: {len(evaluation['results'])}")
|
||
report.append(f"Avg Main Matches: {evaluation['avg_main_matches']:.2f}")
|
||
report.append(f"SZ Match Rate: {evaluation['sz_match_rate']*100:.1f}%")
|
||
report.append(f"Bester Tipp: #{evaluation['best']['tip']}")
|
||
report.append(f" - Main Treffer: {evaluation['best']['main']}")
|
||
report.append(f" - SZ Match: {'Ja' if evaluation['best']['sz'] else 'Nein'}")
|
||
|
||
report.append("")
|
||
report.append("=" * 70)
|
||
|
||
# Speichern
|
||
with open(report_file, 'w') as f:
|
||
f.write('\n'.join(report))
|
||
|
||
# Ausgabe
|
||
for line in report:
|
||
print(line)
|
||
|
||
print(f"\n📁 Report gespeichert: {os.path.basename(report_file)}")
|
||
|
||
def run(self):
|
||
"""Führt kompletten Workflow aus."""
|
||
print()
|
||
print("=" * 70)
|
||
print("START: AUTOMATISCHER UPDATE & LEARNING WORKFLOW")
|
||
print("=" * 70)
|
||
|
||
# 1. Update Daten
|
||
print("\n[SCHRITT 1/4] Daten aktualisieren")
|
||
self.update_data()
|
||
|
||
# 2. Prüfe auf neue Ziehung
|
||
print("\n[SCHRITT 2/4] Neue Ziehung prüfen")
|
||
new_draw = self.check_for_new_draw()
|
||
|
||
if not new_draw:
|
||
print("\n⏭️ Keine neue Ziehung - Workflow beendet")
|
||
return True
|
||
|
||
# 3. Evaluiere Tipps
|
||
print("\n[SCHRITT 3/4] Tipps evaluieren")
|
||
evaluation = self.evaluate_tips(new_draw)
|
||
|
||
# 4. Learning Update
|
||
print("\n[SCHRITT 4/4] Learning Update")
|
||
self.perform_learning_update(new_draw)
|
||
|
||
# Report
|
||
self.generate_report(new_draw, evaluation)
|
||
|
||
# Log Update
|
||
log = self.load_learning_log()
|
||
log["updates"].append({
|
||
"timestamp": datetime.now().isoformat(),
|
||
"draw_date": new_draw['date'].strftime('%Y-%m-%d'),
|
||
"evaluation": evaluation.get('best', {}),
|
||
"avg_matches": {
|
||
'main': evaluation.get('avg_main_matches', 0),
|
||
'sz_rate': evaluation.get('sz_match_rate', 0)
|
||
}
|
||
})
|
||
self.save_learning_log(log)
|
||
|
||
# Sende Benachrichtigung
|
||
try:
|
||
best_match = {
|
||
'main_matches': evaluation.get('best', {}).get('main', 0),
|
||
'sz_match': evaluation.get('best', {}).get('sz', False)
|
||
}
|
||
|
||
evaluation_summary = {
|
||
'avg_main': evaluation.get('avg_main_matches', 0),
|
||
'sz_rate': evaluation.get('sz_match_rate', 0)
|
||
}
|
||
|
||
self.notifier.send_draw_results(new_draw, evaluation_summary, best_match)
|
||
except Exception as e:
|
||
print(f"⚠️ Benachrichtigung fehlgeschlagen: {e}")
|
||
|
||
print("\n" + "=" * 70)
|
||
print("✅ WORKFLOW ERFOLGREICH ABGESCHLOSSEN")
|
||
print("=" * 70)
|
||
|
||
return True
|
||
|
||
|
||
def main():
|
||
"""Hauptfunktion."""
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="Automatisches Update & Learning nach Ziehung"
|
||
)
|
||
parser.add_argument(
|
||
'--data-dir',
|
||
type=str,
|
||
default="/Users/sebastianfrohlich/Projekte/Lotto/data",
|
||
help='Daten-Verzeichnis'
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
# Run Workflow
|
||
workflow = AutoUpdateAndLearn(args.data_dir)
|
||
success = workflow.run()
|
||
|
||
sys.exit(0 if success else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|