cbazzaandClaude Sonnet 4.6
4fc488c668
Data: - Add 12 missing draws (2026-01-09, 2026-04-03 to 2026-05-15) - Retrain LSTM and RF models on complete 952-draw dataset - Run retroactive learning for all 12 skipped draws Automation: - Add eurojackpot-zahlen.eu web scraper as fallback when APIs fail - Fixes recurring gap problem caused by Sazka detail-endpoint change Generator improvements: - Add hard structural filters (sum 95-165, min 1 even + 1 odd per tip) - Add long-term frequency prior as learner dampener (alpha=0.2) - Fix ENSEMBLE strategy generating duplicate tips (greedy → weighted random) - Fix non-vectorized frequency calculation (stack().value_counts()) - Fix Division-by-Zero edge case in set_frequency_prior - Fix potential KeyError on euro_ai_score in structural filter loop .gitignore: - Exclude data/backups/, weekly tip CSVs, performance reports Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
480 lines
16 KiB
Python
480 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Automatischer Eurojackpot 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 * * 2,5 cd /path/to/eurojackpot && source venv/bin/activate && 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 EurojackpotAPIUpdater
|
|
from scripts.utils.update_from_eurojackpot_zahlen_eu import EurojackpotUpdater as WebScraper
|
|
from scripts.generators.ultimate_ai_ml_eurojackpot_generator import UltimateAIMLEurojackpotGenerator
|
|
from scripts.utils.notifier import EurojackpotNotifier
|
|
|
|
|
|
class AutoUpdateAndLearn:
|
|
"""Automatisches Update und Learning System."""
|
|
|
|
def __init__(self, data_dir: str):
|
|
self.data_dir = data_dir
|
|
# CSV file is in Eurojackpot/data folder
|
|
# data_dir = .../Eurojackpot/data
|
|
# We need .../Eurojackpot/data/AlleEurojackpotzahlen.csv
|
|
self.data_file = os.path.join(data_dir, "AlleEurojackpotzahlen.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 = EurojackpotNotifier()
|
|
|
|
print("🤖 AUTOMATISCHES UPDATE & LEARNING SYSTEM")
|
|
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 und nimm neueste Ziehung
|
|
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']),
|
|
'SZ1': int(latest_draw['SZ1']),
|
|
'SZ2': int(latest_draw['SZ2'])
|
|
}
|
|
|
|
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 = EurojackpotAPIUpdater(self.data_file)
|
|
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 = WebScraper(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_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, 6)]
|
|
drawn_euro = [new_draw['SZ1'], new_draw['SZ2']]
|
|
|
|
print(f" 🎲 Gezogene Zahlen: {drawn_main} + Euro {drawn_euro}")
|
|
print()
|
|
|
|
results = []
|
|
best_matches = {'main': 0, 'euro': 0, 'tip': None}
|
|
|
|
for _, tip in tips_df.iterrows():
|
|
# Parse Hauptzahlen
|
|
main_str = tip['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 Eurozahlen
|
|
euro_str = tip['Euro_Numbers']
|
|
if euro_str.startswith('['):
|
|
import ast
|
|
tip_euro = ast.literal_eval(euro_str)
|
|
else:
|
|
tip_euro = [int(n) for n in euro_str.split('-')]
|
|
|
|
# Zähle Treffer
|
|
main_matches = len(set(tip_main) & set(drawn_main))
|
|
euro_matches = len(set(tip_euro) & set(drawn_euro))
|
|
|
|
total_matches = main_matches + euro_matches
|
|
|
|
results.append({
|
|
'tip_number': tip['Tip_Number'],
|
|
'strategy': tip['Strategy'],
|
|
'main_matches': main_matches,
|
|
'euro_matches': euro_matches,
|
|
'total_matches': total_matches
|
|
})
|
|
|
|
# Track best
|
|
if total_matches > (best_matches['main'] + best_matches['euro']):
|
|
best_matches = {
|
|
'main': main_matches,
|
|
'euro': euro_matches,
|
|
'tip': tip['Tip_Number']
|
|
}
|
|
|
|
# Ausgabe
|
|
print(f" {'Tip':<5} {'Strategie':<15} {'Main':<6} {'Euro':<6} {'Gesamt':<8} {'Bewertung'}")
|
|
print(" " + "-" * 60)
|
|
|
|
for r in results:
|
|
rating = self._get_match_rating(r['main_matches'], r['euro_matches'])
|
|
print(f" #{r['tip_number']:<4} {r['strategy']:<15} "
|
|
f"{r['main_matches']:<6} {r['euro_matches']:<6} "
|
|
f"{r['total_matches']:<8} {rating}")
|
|
|
|
print()
|
|
print(f" 🏆 Bester Tipp: #{best_matches['tip']} "
|
|
f"({best_matches['main']} Main + {best_matches['euro']} Euro)")
|
|
|
|
return {
|
|
'file': tip_files[0],
|
|
'results': results,
|
|
'best': best_matches,
|
|
'avg_main_matches': sum(r['main_matches'] for r in results) / len(results),
|
|
'avg_euro_matches': sum(r['euro_matches'] for r in results) / 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, euro: int) -> str:
|
|
"""Bewertung der Treffer."""
|
|
total = main + euro
|
|
|
|
if main == 5 and euro == 2:
|
|
return "🏆 JACKPOT!"
|
|
elif main == 5 and euro == 1:
|
|
return "💰 Klasse 2"
|
|
elif main == 5 and euro == 0:
|
|
return "💰 Klasse 3"
|
|
elif main == 4 and euro == 2:
|
|
return "💰 Klasse 4"
|
|
elif main == 4 and euro == 1:
|
|
return "💵 Klasse 5"
|
|
elif total >= 3:
|
|
return "✅ Gut"
|
|
elif total >= 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 = UltimateAIMLEurojackpotGenerator(
|
|
self.data_file,
|
|
fast_mode=True
|
|
)
|
|
|
|
# Learning Update
|
|
generator.real_time_learner.learn_from_result(new_draw)
|
|
|
|
# Performance Tracking
|
|
generator.performance_tracker.evaluate_predictions(new_draw)
|
|
|
|
print(" ✅ Learning Update durchgeführt")
|
|
print(" 📊 Models wurden angepasst")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Fehler beim Learning: {e}")
|
|
return False
|
|
|
|
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("EUROJACKPOT 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, 6)]}")
|
|
report.append(f"Eurozahlen: [{new_draw['SZ1']}, {new_draw['SZ2']}]")
|
|
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"Avg Euro Matches: {evaluation['avg_euro_matches']:.2f}")
|
|
report.append(f"Bester Tipp: #{evaluation['best']['tip']}")
|
|
report.append(f" - Main Treffer: {evaluation['best']['main']}")
|
|
report.append(f" - Euro Treffer: {evaluation['best']['euro']}")
|
|
|
|
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),
|
|
'euro': evaluation.get('avg_euro_matches', 0)
|
|
}
|
|
})
|
|
self.save_learning_log(log)
|
|
|
|
# Sende Benachrichtigung
|
|
try:
|
|
best_match = {
|
|
'main_matches': evaluation.get('best', {}).get('main', 0),
|
|
'euro_matches': evaluation.get('best', {}).get('euro', 0)
|
|
}
|
|
|
|
evaluation_summary = {
|
|
'avg_main': evaluation.get('avg_main_matches', 0),
|
|
'avg_euro': evaluation.get('avg_euro_matches', 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/Eurojackpot/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()
|