Initial commit: Lotto number generator project
This project includes multiple AI/ML-based lottery number generators for German Lotto 6aus49, including pattern analysis, weighted predictions, and hybrid approaches. Features automated weekly tip generation, performance tracking, and Telegram bot integration. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
cbazzaandClaude Sonnet 4.5 committed
commit
f6106b8333
48 files changed
+25168
No files matched your search
Executable
+452
@@ -0,0 +1,452 @@
|
||||
#!/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.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)
|
||||
success = updater.update(api_name='github', create_backup=True)
|
||||
|
||||
if success:
|
||||
print(" ✅ Daten erfolgreich aktualisiert")
|
||||
else:
|
||||
print(" ⚠️ Update ohne neue Daten")
|
||||
|
||||
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 neueste Tipps-Datei
|
||||
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 {}
|
||||
|
||||
latest_tips_file = os.path.join(self.tips_dir, tip_files[0])
|
||||
print(f" 📁 Evaluiere: {tip_files[0]}")
|
||||
|
||||
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/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/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()
|
||||
@@ -0,0 +1,301 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Wöchentlicher Lotto 6aus49 Tipp-Generator
|
||||
|
||||
Automatisiert:
|
||||
1. Prüft ob neue Tipps nötig sind (basierend auf letzter Generierung)
|
||||
2. Generiert 10 Ultimate Tipps
|
||||
3. Speichert mit Timestamp
|
||||
4. Trackt Generierungs-Historie
|
||||
|
||||
Verwendung:
|
||||
python weekly_tip_generator.py
|
||||
|
||||
Oder als Cronjob:
|
||||
0 9 * * 3,6 cd /path/to/lotto && source venv/bin/activate && python scripts/automation/weekly_tip_generator.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
import json
|
||||
|
||||
# Füge Parent-Verzeichnisse zum Path hinzu
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_dir = os.path.dirname(os.path.dirname(script_dir))
|
||||
generators_dir = os.path.join(project_dir, 'scripts', 'generators')
|
||||
sys.path.insert(0, project_dir)
|
||||
sys.path.insert(0, generators_dir)
|
||||
|
||||
# Import des Ultimate Generators
|
||||
from scripts.generators.ultimate_ai_ml_hybrid_generator import UltimateAIMLHybridGenerator
|
||||
from scripts.utils.notifier import LottoNotifier
|
||||
|
||||
|
||||
class WeeklyTipGenerator:
|
||||
"""Automatischer wöchentlicher Tipp-Generator für Lotto 6aus49."""
|
||||
|
||||
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.history_file = os.path.join(self.tips_dir, "generation_history.json")
|
||||
|
||||
os.makedirs(self.tips_dir, exist_ok=True)
|
||||
|
||||
# Initialisiere Notifier
|
||||
self.notifier = LottoNotifier()
|
||||
|
||||
print("🤖 AUTOMATISCHER WÖCHENTLICHER LOTTO 6AUS49 TIPP-GENERATOR")
|
||||
print("=" * 70)
|
||||
|
||||
def load_history(self) -> dict:
|
||||
"""Lädt Generierungs-Historie."""
|
||||
if os.path.exists(self.history_file):
|
||||
with open(self.history_file, 'r') as f:
|
||||
return json.load(f)
|
||||
return {"generations": []}
|
||||
|
||||
def save_history(self, history: dict):
|
||||
"""Speichert Historie."""
|
||||
with open(self.history_file, 'w') as f:
|
||||
json.dump(history, f, indent=2)
|
||||
|
||||
def needs_new_tips(self) -> bool:
|
||||
"""
|
||||
Prüft ob neue Tipps nötig sind.
|
||||
|
||||
Logik:
|
||||
- Lotto 6aus49: Mittwoch & Samstag Ziehungen
|
||||
- Generiere Tipps wenn:
|
||||
a) Noch nie generiert
|
||||
b) Letzte Generierung > 3 Tage her
|
||||
c) Es ist Dienstag oder Freitag (vor Ziehung)
|
||||
"""
|
||||
history = self.load_history()
|
||||
|
||||
if not history["generations"]:
|
||||
print(" ℹ️ Noch nie Tipps generiert")
|
||||
return True
|
||||
|
||||
last_gen = history["generations"][-1]
|
||||
last_date = datetime.fromisoformat(last_gen["timestamp"])
|
||||
days_since = (datetime.now() - last_date).days
|
||||
|
||||
print(f" 📅 Letzte Generierung: {last_date.strftime('%Y-%m-%d %H:%M')}")
|
||||
print(f" ⏱️ Vor {days_since} Tagen")
|
||||
|
||||
# Wenn > 3 Tage her
|
||||
if days_since > 3:
|
||||
print(" ✅ Mehr als 3 Tage her - neue Tipps nötig")
|
||||
return True
|
||||
|
||||
# Prüfe Wochentag (0=Montag, 2=Mittwoch, 5=Samstag)
|
||||
today = datetime.now().weekday()
|
||||
|
||||
# Dienstag (vor Mittwoch-Ziehung)
|
||||
if today == 1 and days_since >= 1:
|
||||
print(" ✅ Dienstag - generiere für Mittwoch-Ziehung")
|
||||
return True
|
||||
|
||||
# Freitag (vor Samstag-Ziehung)
|
||||
if today == 4 and days_since >= 1:
|
||||
print(" ✅ Freitag - generiere für Samstag-Ziehung")
|
||||
return True
|
||||
|
||||
print(" ⏭️ Keine neuen Tipps nötig")
|
||||
return False
|
||||
|
||||
def generate_tips(self, num_tips: int = 10, force: bool = False) -> bool:
|
||||
"""
|
||||
Generiert neue Tipps.
|
||||
|
||||
Args:
|
||||
num_tips: Anzahl Tipps
|
||||
force: Ignoriere needs_new_tips Check
|
||||
|
||||
Returns:
|
||||
True bei Erfolg
|
||||
"""
|
||||
print("\n🎯 TIPP-GENERIERUNG")
|
||||
print("=" * 70)
|
||||
|
||||
# Check ob nötig
|
||||
if not force and not self.needs_new_tips():
|
||||
print("\n⏭️ Keine Generierung nötig")
|
||||
return True
|
||||
|
||||
print(f"\n🚀 Generiere {num_tips} Ultimate Lotto 6aus49 Tipps...")
|
||||
print("-" * 70)
|
||||
|
||||
try:
|
||||
# Initialisiere Generator
|
||||
generator = UltimateAIMLHybridGenerator(
|
||||
self.data_file,
|
||||
fast_mode=True
|
||||
)
|
||||
|
||||
# Generiere Tipps
|
||||
tips = generator.generate_ultimate_tips(num_tips=num_tips)
|
||||
|
||||
if not tips:
|
||||
print("\n❌ Keine Tipps generiert")
|
||||
return False
|
||||
|
||||
# Speichere Tipps
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
output_file = os.path.join(
|
||||
self.tips_dir,
|
||||
f"weekly_lotto_tips_{timestamp}.csv"
|
||||
)
|
||||
|
||||
self._export_tips_to_csv(tips, output_file)
|
||||
|
||||
# Update Historie
|
||||
history = self.load_history()
|
||||
history["generations"].append({
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"num_tips": len(tips),
|
||||
"file": os.path.basename(output_file),
|
||||
"avg_confidence": sum(t['confidence'] for t in tips) / len(tips),
|
||||
"avg_quality": sum(t['quality'] for t in tips) / len(tips)
|
||||
})
|
||||
self.save_history(history)
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
print("✅ TIPPS ERFOLGREICH GENERIERT")
|
||||
print(f"📁 Datei: {os.path.basename(output_file)}")
|
||||
print(f"📊 Anzahl: {len(tips)}")
|
||||
print(f"🎯 Avg Confidence: {history['generations'][-1]['avg_confidence']:.4f}")
|
||||
print(f"💎 Avg Quality: {history['generations'][-1]['avg_quality']:.4f}")
|
||||
print("=" * 70)
|
||||
|
||||
# Sende Benachrichtigung
|
||||
try:
|
||||
# Finde besten Tipp (höchste Confidence)
|
||||
best_tip = max(tips, key=lambda t: t.get('confidence', 0))
|
||||
|
||||
# Formatiere für Notification
|
||||
best_tip_formatted = {
|
||||
'numbers': best_tip.get('numbers', []),
|
||||
'superzahl': best_tip.get('superzahl', 0),
|
||||
'confidence': best_tip.get('confidence', 0),
|
||||
'strategy': best_tip.get('strategy', 'UNKNOWN')
|
||||
}
|
||||
|
||||
timestamp_formatted = datetime.now().strftime('%Y-%m-%d %H:%M')
|
||||
self.notifier.send_tips_generated(tips, timestamp_formatted, best_tip_formatted)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Benachrichtigung fehlgeschlagen: {e}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Fehler bei Generierung: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def _export_tips_to_csv(self, tips, filepath):
|
||||
"""Exportiert Tips als CSV."""
|
||||
import pandas as pd
|
||||
|
||||
rows = []
|
||||
for tip in tips:
|
||||
numbers_str = '-'.join([str(n) for n in tip['numbers']])
|
||||
|
||||
rows.append({
|
||||
'Tip_Number': tip['tip_number'],
|
||||
'Numbers': numbers_str,
|
||||
'Superzahl': tip['superzahl'],
|
||||
'Strategy': tip['strategy'],
|
||||
'AI_Score': f"{tip['ai_score']:.4f}",
|
||||
'Pattern_Weight': f"{tip['pattern_weight']:.4f}",
|
||||
'Confidence': f"{tip['confidence']:.4f}",
|
||||
'Quality': f"{tip['quality']:.4f}"
|
||||
})
|
||||
|
||||
df_export = pd.DataFrame(rows)
|
||||
df_export.to_csv(filepath, index=False)
|
||||
print(f"\n💾 Tips exported to: {filepath}")
|
||||
|
||||
def show_history(self):
|
||||
"""Zeigt Generierungs-Historie."""
|
||||
history = self.load_history()
|
||||
|
||||
if not history["generations"]:
|
||||
print("\n ℹ️ Noch keine Generierungen")
|
||||
return
|
||||
|
||||
print("\n📊 GENERIERUNGS-HISTORIE")
|
||||
print("=" * 70)
|
||||
print(f"{'Nr':<4} {'Datum':<20} {'Tips':<6} {'Confidence':<12} {'Quality':<12} {'Datei'}")
|
||||
print("-" * 70)
|
||||
|
||||
for i, gen in enumerate(reversed(history["generations"][-10:]), 1):
|
||||
timestamp = datetime.fromisoformat(gen["timestamp"])
|
||||
print(f"{i:<4} {timestamp.strftime('%Y-%m-%d %H:%M'):<20} "
|
||||
f"{gen['num_tips']:<6} {gen['avg_confidence']:<12.4f} "
|
||||
f"{gen['avg_quality']:<12.4f} {gen['file']}")
|
||||
|
||||
print("-" * 70)
|
||||
print(f"Total: {len(history['generations'])} Generierungen")
|
||||
|
||||
|
||||
def main():
|
||||
"""Hauptfunktion."""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Wöchentlicher Lotto 6aus49 Tipp-Generator"
|
||||
)
|
||||
parser.add_argument(
|
||||
'--force',
|
||||
action='store_true',
|
||||
help='Generiere Tipps auch wenn nicht nötig'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--num-tips',
|
||||
type=int,
|
||||
default=10,
|
||||
help='Anzahl Tipps zu generieren (default: 10)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--history',
|
||||
action='store_true',
|
||||
help='Zeige nur Historie'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--data-dir',
|
||||
type=str,
|
||||
default="/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data",
|
||||
help='Daten-Verzeichnis'
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Initialisiere
|
||||
generator = WeeklyTipGenerator(args.data_dir)
|
||||
|
||||
# Zeige Historie wenn gewünscht
|
||||
if args.history:
|
||||
generator.show_history()
|
||||
return
|
||||
|
||||
# Generiere Tipps
|
||||
success = generator.generate_tips(
|
||||
num_tips=args.num_tips,
|
||||
force=args.force
|
||||
)
|
||||
|
||||
# Zeige Historie
|
||||
generator.show_history()
|
||||
|
||||
sys.exit(0 if success else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,241 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Model Evaluation & Reporting Module
|
||||
Für umfassende ML-Model-Evaluation
|
||||
"""
|
||||
|
||||
import numpy as np
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from collections import defaultdict
|
||||
|
||||
try:
|
||||
from sklearn.model_selection import cross_val_score
|
||||
from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_error
|
||||
SKLEARN_AVAILABLE = True
|
||||
except ImportError:
|
||||
SKLEARN_AVAILABLE = False
|
||||
|
||||
|
||||
class ModelEvaluator:
|
||||
"""Evaluiert und dokumentiert ML-Modelle."""
|
||||
|
||||
def __init__(self, cache_path=None):
|
||||
self.cache_path = cache_path
|
||||
self.evaluation_results = {}
|
||||
self.report_file = os.path.join(cache_path, 'model_evaluation.json') if cache_path else None
|
||||
|
||||
def evaluate_model(self, model, X_train, X_test, y_train, y_test, model_name, number):
|
||||
"""
|
||||
Umfassende Model-Evaluation.
|
||||
|
||||
Returns:
|
||||
dict with metrics
|
||||
"""
|
||||
if not SKLEARN_AVAILABLE:
|
||||
return {'error': 'scikit-learn not available'}
|
||||
|
||||
results = {
|
||||
'model_name': model_name,
|
||||
'number': number,
|
||||
'timestamp': datetime.now().isoformat(),
|
||||
'data_size': {
|
||||
'train': len(X_train),
|
||||
'test': len(X_test)
|
||||
}
|
||||
}
|
||||
|
||||
try:
|
||||
# 1. Training Score
|
||||
y_train_pred = model.predict(X_train)
|
||||
results['train_r2'] = r2_score(y_train, y_train_pred)
|
||||
results['train_mae'] = mean_absolute_error(y_train, y_train_pred)
|
||||
results['train_rmse'] = np.sqrt(mean_squared_error(y_train, y_train_pred))
|
||||
|
||||
# 2. Test Score
|
||||
y_test_pred = model.predict(X_test)
|
||||
results['test_r2'] = r2_score(y_test, y_test_pred)
|
||||
results['test_mae'] = mean_absolute_error(y_test, y_test_pred)
|
||||
results['test_rmse'] = np.sqrt(mean_squared_error(y_test, y_test_pred))
|
||||
|
||||
# 3. Overfit Detection
|
||||
results['overfitting'] = results['train_r2'] - results['test_r2']
|
||||
results['is_overfit'] = results['overfitting'] > 0.2
|
||||
|
||||
# 4. Cross-Validation (3-fold for speed)
|
||||
try:
|
||||
cv_scores = cross_val_score(model, X_train, y_train, cv=3, scoring='r2')
|
||||
results['cv_mean'] = float(np.mean(cv_scores))
|
||||
results['cv_std'] = float(np.std(cv_scores))
|
||||
results['cv_scores'] = [float(s) for s in cv_scores]
|
||||
except Exception as e:
|
||||
results['cv_error'] = str(e)
|
||||
|
||||
# 5. Feature Importance (if available)
|
||||
if hasattr(model, 'feature_importances_'):
|
||||
importances = model.feature_importances_
|
||||
results['top_features'] = {
|
||||
f'feature_{i}': float(imp)
|
||||
for i, imp in enumerate(importances[:10]) # Top 10
|
||||
}
|
||||
results['feature_importance_sum'] = float(np.sum(importances))
|
||||
|
||||
# 6. Prediction Distribution
|
||||
results['pred_distribution'] = {
|
||||
'min': float(np.min(y_test_pred)),
|
||||
'max': float(np.max(y_test_pred)),
|
||||
'mean': float(np.mean(y_test_pred)),
|
||||
'std': float(np.std(y_test_pred))
|
||||
}
|
||||
|
||||
# 7. Quality Rating
|
||||
test_r2 = results['test_r2']
|
||||
if test_r2 > 0.7:
|
||||
results['quality'] = 'Excellent'
|
||||
elif test_r2 > 0.5:
|
||||
results['quality'] = 'Good'
|
||||
elif test_r2 > 0.3:
|
||||
results['quality'] = 'Fair'
|
||||
elif test_r2 > 0.1:
|
||||
results['quality'] = 'Poor'
|
||||
else:
|
||||
results['quality'] = 'Very Poor'
|
||||
|
||||
except Exception as e:
|
||||
results['error'] = str(e)
|
||||
|
||||
return results
|
||||
|
||||
def add_evaluation(self, number, model_name, results):
|
||||
"""Fügt Evaluation-Result hinzu."""
|
||||
key = f"{number}_{model_name}"
|
||||
self.evaluation_results[key] = results
|
||||
|
||||
def generate_summary_report(self):
|
||||
"""Generiert Zusammenfassungs-Report."""
|
||||
if not self.evaluation_results:
|
||||
return "Keine Evaluation-Daten vorhanden"
|
||||
|
||||
report = []
|
||||
report.append("=" * 80)
|
||||
report.append("MODEL EVALUATION SUMMARY")
|
||||
report.append("=" * 80)
|
||||
report.append(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
report.append(f"Total Evaluations: {len(self.evaluation_results)}")
|
||||
report.append("")
|
||||
|
||||
# Aggregate stats
|
||||
all_test_r2 = []
|
||||
all_cv_mean = []
|
||||
overfit_count = 0
|
||||
quality_dist = defaultdict(int)
|
||||
|
||||
for key, results in self.evaluation_results.items():
|
||||
if 'test_r2' in results:
|
||||
all_test_r2.append(results['test_r2'])
|
||||
if 'cv_mean' in results:
|
||||
all_cv_mean.append(results['cv_mean'])
|
||||
if results.get('is_overfit'):
|
||||
overfit_count += 1
|
||||
if 'quality' in results:
|
||||
quality_dist[results['quality']] += 1
|
||||
|
||||
# Overall Stats
|
||||
report.append("OVERALL STATISTICS")
|
||||
report.append("-" * 80)
|
||||
if all_test_r2:
|
||||
report.append(f"Test R² Score:")
|
||||
report.append(f" Mean: {np.mean(all_test_r2):.4f}")
|
||||
report.append(f" Median: {np.median(all_test_r2):.4f}")
|
||||
report.append(f" Std: {np.std(all_test_r2):.4f}")
|
||||
report.append(f" Min: {np.min(all_test_r2):.4f}")
|
||||
report.append(f" Max: {np.max(all_test_r2):.4f}")
|
||||
report.append("")
|
||||
|
||||
if all_cv_mean:
|
||||
report.append(f"Cross-Validation R² Score:")
|
||||
report.append(f" Mean: {np.mean(all_cv_mean):.4f}")
|
||||
report.append(f" Std: {np.std(all_cv_mean):.4f}")
|
||||
report.append("")
|
||||
|
||||
report.append(f"Overfitting Detection:")
|
||||
report.append(f" Overfit Models: {overfit_count}/{len(self.evaluation_results)}")
|
||||
report.append("")
|
||||
|
||||
report.append(f"Quality Distribution:")
|
||||
for quality in ['Excellent', 'Good', 'Fair', 'Poor', 'Very Poor']:
|
||||
count = quality_dist.get(quality, 0)
|
||||
pct = (count / len(self.evaluation_results)) * 100 if self.evaluation_results else 0
|
||||
report.append(f" {quality:12}: {count:3} ({pct:5.1f}%)")
|
||||
report.append("")
|
||||
|
||||
# Top 10 Best Models
|
||||
sorted_results = sorted(
|
||||
[(k, v) for k, v in self.evaluation_results.items() if 'test_r2' in v],
|
||||
key=lambda x: x[1]['test_r2'],
|
||||
reverse=True
|
||||
)[:10]
|
||||
|
||||
report.append("TOP 10 MODELS (by Test R²)")
|
||||
report.append("-" * 80)
|
||||
report.append(f"{'Number':<10} {'Model':<20} {'Test R²':<12} {'CV Mean':<12} {'Quality':<15}")
|
||||
report.append("-" * 80)
|
||||
|
||||
for key, results in sorted_results:
|
||||
number = results.get('number', 'N/A')
|
||||
model = results.get('model_name', 'N/A')
|
||||
test_r2 = results.get('test_r2', 0)
|
||||
cv_mean = results.get('cv_mean', 0)
|
||||
quality = results.get('quality', 'N/A')
|
||||
|
||||
report.append(f"{number:<10} {model:<20} {test_r2:<12.4f} {cv_mean:<12.4f} {quality:<15}")
|
||||
|
||||
report.append("")
|
||||
report.append("=" * 80)
|
||||
|
||||
return "\n".join(report)
|
||||
|
||||
def save_evaluation(self):
|
||||
"""Speichert Evaluation persistent."""
|
||||
if not self.report_file:
|
||||
return
|
||||
|
||||
try:
|
||||
os.makedirs(os.path.dirname(self.report_file), exist_ok=True)
|
||||
|
||||
data = {
|
||||
'timestamp': datetime.now().isoformat(),
|
||||
'total_evaluations': len(self.evaluation_results),
|
||||
'results': self.evaluation_results
|
||||
}
|
||||
|
||||
with open(self.report_file, 'w') as f:
|
||||
json.dump(data, f, indent=2)
|
||||
|
||||
print(f" 💾 Evaluation saved: {self.report_file}")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not save evaluation: {e}")
|
||||
|
||||
def load_evaluation(self):
|
||||
"""Lädt gespeicherte Evaluation."""
|
||||
if not self.report_file or not os.path.exists(self.report_file):
|
||||
return
|
||||
|
||||
try:
|
||||
with open(self.report_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
|
||||
self.evaluation_results = data.get('results', {})
|
||||
timestamp = data.get('timestamp', 'Unknown')
|
||||
|
||||
print(f" 📂 Evaluation loaded: {len(self.evaluation_results)} results (from {timestamp[:10]})")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not load evaluation: {e}")
|
||||
|
||||
def print_summary(self):
|
||||
"""Druckt Summary auf Console."""
|
||||
summary = self.generate_summary_report()
|
||||
print("\n" + summary)
|
||||
@@ -0,0 +1,340 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Lotto 6aus49 Benachrichtigungs-System
|
||||
|
||||
Unterstützt:
|
||||
- Telegram Bot Notifications
|
||||
- Email Notifications (optional)
|
||||
|
||||
Konfiguration über config/notifications.json
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import requests
|
||||
from typing import List, Dict, Any
|
||||
|
||||
|
||||
class LottoNotifier:
|
||||
"""Benachrichtigungs-System für Lotto 6aus49."""
|
||||
|
||||
def __init__(self, config_path: str = None):
|
||||
"""
|
||||
Initialisiert Notifier.
|
||||
|
||||
Args:
|
||||
config_path: Pfad zur Konfigurationsdatei
|
||||
"""
|
||||
if config_path is None:
|
||||
# Default: config/notifications.json im Projekt-Root
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.dirname(os.path.dirname(script_dir))
|
||||
config_path = os.path.join(project_root, 'config', 'notifications.json')
|
||||
|
||||
self.config_path = config_path
|
||||
self.config = self._load_config()
|
||||
|
||||
def _load_config(self) -> dict:
|
||||
"""Lädt Konfiguration."""
|
||||
if not os.path.exists(self.config_path):
|
||||
print(f"⚠️ Konfigurationsdatei nicht gefunden: {self.config_path}")
|
||||
return {
|
||||
"telegram": {"enabled": False},
|
||||
"email": {"enabled": False}
|
||||
}
|
||||
|
||||
try:
|
||||
with open(self.config_path, 'r') as f:
|
||||
config = json.load(f)
|
||||
return config
|
||||
except Exception as e:
|
||||
print(f"⚠️ Fehler beim Laden der Konfiguration: {e}")
|
||||
return {
|
||||
"telegram": {"enabled": False},
|
||||
"email": {"enabled": False}
|
||||
}
|
||||
|
||||
def send_tips_generated(self, tips: List[Dict], timestamp: str, best_tip: Dict):
|
||||
"""
|
||||
Sendet Benachrichtigung über generierte Tipps.
|
||||
|
||||
Args:
|
||||
tips: Liste aller generierten Tipps
|
||||
timestamp: Timestamp der Generierung
|
||||
best_tip: Bester Tipp (höchste Confidence)
|
||||
"""
|
||||
if self.config['telegram']['enabled']:
|
||||
self._send_telegram_tips(tips, timestamp, best_tip)
|
||||
|
||||
if self.config['email']['enabled']:
|
||||
self._send_email_tips(tips, timestamp, best_tip)
|
||||
|
||||
def _send_telegram_tips(self, tips: List[Dict], timestamp: str, best_tip: Dict):
|
||||
"""Sendet Telegram-Nachricht."""
|
||||
try:
|
||||
bot_token = self.config['telegram']['bot_token']
|
||||
chat_id = self.config['telegram']['chat_id']
|
||||
|
||||
# Formatiere Nachricht
|
||||
message = self._format_telegram_message(tips, timestamp, best_tip)
|
||||
|
||||
# Sende über Telegram Bot API
|
||||
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
|
||||
data = {
|
||||
"chat_id": chat_id,
|
||||
"text": message,
|
||||
"parse_mode": "Markdown"
|
||||
}
|
||||
|
||||
response = requests.post(url, data=data, timeout=10)
|
||||
|
||||
if response.status_code == 200:
|
||||
print("✅ Telegram-Benachrichtigung gesendet")
|
||||
else:
|
||||
print(f"⚠️ Telegram-Fehler: {response.status_code}")
|
||||
print(f" Response: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Telegram-Benachrichtigung fehlgeschlagen: {e}")
|
||||
|
||||
def _format_telegram_message(self, tips: List[Dict], timestamp: str, best_tip: Dict) -> str:
|
||||
"""Formatiert Telegram-Nachricht."""
|
||||
# Header
|
||||
message = "🎲 *LOTTO 6AUS49 - NEUE TIPPS GENERIERT*\n"
|
||||
message += "=" * 40 + "\n\n"
|
||||
|
||||
# Timestamp
|
||||
message += f"📅 *Generiert:* {timestamp}\n"
|
||||
message += f"📊 *Anzahl Tipps:* {len(tips)}\n\n"
|
||||
|
||||
# Bester Tipp
|
||||
message += "⭐ *BESTER TIPP:*\n"
|
||||
numbers_str = ' - '.join([f"{n:02d}" for n in best_tip['numbers']])
|
||||
message += f"🎯 Zahlen: `{numbers_str}`\n"
|
||||
message += f"🌟 Superzahl: `{best_tip['superzahl']}`\n"
|
||||
message += f"📈 Confidence: `{best_tip['confidence']:.4f}`\n"
|
||||
message += f"🎨 Strategie: `{best_tip['strategy']}`\n\n"
|
||||
|
||||
# Statistiken
|
||||
avg_conf = sum(t['confidence'] for t in tips) / len(tips)
|
||||
avg_qual = sum(t['quality'] for t in tips) / len(tips)
|
||||
|
||||
message += "📊 *STATISTIKEN:*\n"
|
||||
message += f"🎯 Ø Confidence: `{avg_conf:.4f}`\n"
|
||||
message += f"💎 Ø Quality: `{avg_qual:.4f}`\n\n"
|
||||
|
||||
# Top 3 Tipps
|
||||
message += "🏆 *TOP 3 TIPPS:*\n"
|
||||
sorted_tips = sorted(tips, key=lambda t: t['confidence'], reverse=True)[:3]
|
||||
|
||||
for i, tip in enumerate(sorted_tips, 1):
|
||||
nums = ' - '.join([f"{n:02d}" for n in tip['numbers']])
|
||||
message += f"{i}. `{nums}` + SZ `{tip['superzahl']}` "
|
||||
message += f"({tip['confidence']:.3f})\n"
|
||||
|
||||
message += "\n🍀 *Viel Glück!*"
|
||||
|
||||
return message
|
||||
|
||||
def _send_email_tips(self, tips: List[Dict], timestamp: str, best_tip: Dict):
|
||||
"""Sendet Email-Benachrichtigung."""
|
||||
# TODO: Email-Versand implementieren wenn gewünscht
|
||||
print("⚠️ Email-Benachrichtigung noch nicht implementiert")
|
||||
|
||||
def send_draw_results(self, draw: Dict, evaluation_summary: Dict, best_match: Dict):
|
||||
"""
|
||||
Sendet Benachrichtigung über Ziehungs-Ergebnisse und Tipp-Evaluation.
|
||||
|
||||
Args:
|
||||
draw: Dict mit Ziehungsdaten (date, Z1-Z6, SZ)
|
||||
evaluation_summary: Dict mit avg_main, sz_rate
|
||||
best_match: Dict mit main_matches, sz_match
|
||||
"""
|
||||
if self.config['telegram']['enabled']:
|
||||
self._send_telegram_draw_results(draw, evaluation_summary, best_match)
|
||||
|
||||
if self.config['email']['enabled']:
|
||||
self._send_email_draw_results(draw, evaluation_summary, best_match)
|
||||
|
||||
def _send_telegram_draw_results(self, draw: Dict, evaluation_summary: Dict, best_match: Dict):
|
||||
"""Sendet Telegram-Nachricht mit Ziehungsergebnissen."""
|
||||
try:
|
||||
bot_token = self.config['telegram']['bot_token']
|
||||
chat_id = self.config['telegram']['chat_id']
|
||||
|
||||
# Formatiere Nachricht
|
||||
message = self._format_draw_results_message(draw, evaluation_summary, best_match)
|
||||
|
||||
# Sende über Telegram Bot API
|
||||
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
|
||||
data = {
|
||||
"chat_id": chat_id,
|
||||
"text": message,
|
||||
"parse_mode": "Markdown"
|
||||
}
|
||||
|
||||
response = requests.post(url, data=data, timeout=10)
|
||||
|
||||
if response.status_code == 200:
|
||||
print("✅ Telegram-Benachrichtigung gesendet")
|
||||
else:
|
||||
print(f"⚠️ Telegram-Fehler: {response.status_code}")
|
||||
print(f" Response: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Telegram-Benachrichtigung fehlgeschlagen: {e}")
|
||||
|
||||
def _format_draw_results_message(self, draw: Dict, evaluation_summary: Dict, best_match: Dict) -> str:
|
||||
"""Formatiert Telegram-Nachricht für Ziehungsergebnisse."""
|
||||
# Header
|
||||
message = "🎲 *LOTTO 6AUS49 - ZIEHUNGSERGEBNISSE*\n"
|
||||
message += "=" * 40 + "\n\n"
|
||||
|
||||
# Ziehungsdatum
|
||||
draw_date = draw['date']
|
||||
if hasattr(draw_date, 'strftime'):
|
||||
date_str = draw_date.strftime('%d.%m.%Y')
|
||||
else:
|
||||
date_str = str(draw_date)
|
||||
|
||||
message += f"📅 *Ziehung vom:* {date_str}\n\n"
|
||||
|
||||
# Gezogene Zahlen
|
||||
drawn_numbers = [draw.get(f'Z{i}') for i in range(1, 7)]
|
||||
numbers_str = ' - '.join([f"{n:02d}" for n in drawn_numbers if n is not None])
|
||||
message += f"🎯 *Gewinnzahlen:* `{numbers_str}`\n"
|
||||
|
||||
if draw.get('SZ') is not None:
|
||||
message += f"🌟 *Superzahl:* `{draw['SZ']}`\n\n"
|
||||
else:
|
||||
message += "\n"
|
||||
|
||||
# Evaluation Summary
|
||||
message += "📊 *TIPP-EVALUATION:*\n"
|
||||
if evaluation_summary:
|
||||
avg_main = evaluation_summary.get('avg_main', 0)
|
||||
sz_rate = evaluation_summary.get('sz_rate', 0)
|
||||
message += f"🎯 Ø Treffer Hauptzahlen: `{avg_main:.2f}`\n"
|
||||
message += f"🌟 Superzahl-Rate: `{sz_rate*100:.1f}%`\n\n"
|
||||
|
||||
# Bester Tipp
|
||||
message += "🏆 *BESTER TIPP:*\n"
|
||||
if best_match:
|
||||
main_matches = best_match.get('main_matches', 0)
|
||||
sz_match = best_match.get('sz_match', False)
|
||||
|
||||
# Rating emoji
|
||||
if main_matches == 6 and sz_match:
|
||||
rating = "🏆 JACKPOT!"
|
||||
elif main_matches == 6:
|
||||
rating = "💰 Klasse 2"
|
||||
elif main_matches == 5 and sz_match:
|
||||
rating = "💰 Klasse 3"
|
||||
elif main_matches == 5:
|
||||
rating = "💰 Klasse 4"
|
||||
elif main_matches == 4:
|
||||
rating = "💵 Klasse 6"
|
||||
elif main_matches == 3:
|
||||
rating = "✅ Klasse 8"
|
||||
elif main_matches >= 2:
|
||||
rating = "👍 OK"
|
||||
else:
|
||||
rating = "⚪ Niedrig"
|
||||
|
||||
sz_indicator = "✅" if sz_match else "⚪"
|
||||
message += f"🎯 Treffer Hauptzahlen: `{main_matches}/6`\n"
|
||||
message += f"🌟 Superzahl: {sz_indicator}\n"
|
||||
message += f"📈 Bewertung: {rating}\n"
|
||||
|
||||
message += "\n🔄 *System wurde aktualisiert und trainiert!*"
|
||||
|
||||
return message
|
||||
|
||||
def _send_email_draw_results(self, draw: Dict, evaluation_summary: Dict, best_match: Dict):
|
||||
"""Sendet Email-Benachrichtigung."""
|
||||
# TODO: Email-Versand implementieren wenn gewünscht
|
||||
print("⚠️ Email-Benachrichtigung noch nicht implementiert")
|
||||
|
||||
def send_test_notification(self):
|
||||
"""Sendet Test-Benachrichtigung."""
|
||||
if self.config['telegram']['enabled']:
|
||||
try:
|
||||
bot_token = self.config['telegram']['bot_token']
|
||||
chat_id = self.config['telegram']['chat_id']
|
||||
|
||||
message = "🧪 *LOTTO 6AUS49 TEST*\n\n"
|
||||
message += "✅ Benachrichtigungs-System funktioniert!\n"
|
||||
message += f"📅 {os.popen('date').read().strip()}"
|
||||
|
||||
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
|
||||
data = {
|
||||
"chat_id": chat_id,
|
||||
"text": message,
|
||||
"parse_mode": "Markdown"
|
||||
}
|
||||
|
||||
response = requests.post(url, data=data, timeout=10)
|
||||
|
||||
if response.status_code == 200:
|
||||
print("✅ Test-Benachrichtigung erfolgreich gesendet")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ Fehler: {response.status_code}")
|
||||
print(f" Response: {response.text}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Test fehlgeschlagen: {e}")
|
||||
return False
|
||||
else:
|
||||
print("⚠️ Telegram nicht aktiviert in config/notifications.json")
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""Test-Funktion."""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description="Lotto 6aus49 Notifier Test")
|
||||
parser.add_argument(
|
||||
'--test',
|
||||
action='store_true',
|
||||
help='Sende Test-Benachrichtigung'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--config',
|
||||
type=str,
|
||||
help='Pfad zur Config-Datei'
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
notifier = LottoNotifier(config_path=args.config)
|
||||
|
||||
if args.test:
|
||||
notifier.send_test_notification()
|
||||
else:
|
||||
# Beispiel-Tipps
|
||||
example_tips = [
|
||||
{
|
||||
'numbers': [7, 14, 21, 28, 35, 42],
|
||||
'superzahl': 3,
|
||||
'confidence': 0.7234,
|
||||
'quality': 0.6891,
|
||||
'strategy': 'PURE-AI'
|
||||
},
|
||||
{
|
||||
'numbers': [2, 11, 19, 27, 36, 45],
|
||||
'superzahl': 7,
|
||||
'confidence': 0.6978,
|
||||
'quality': 0.6543,
|
||||
'strategy': 'HYBRID-OPT'
|
||||
}
|
||||
]
|
||||
|
||||
best = example_tips[0]
|
||||
notifier.send_tips_generated(example_tips, "2024-11-27 16:00", best)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,240 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Einfacher Lotto 6aus49 Data Updater
|
||||
|
||||
Manuelle Eingabe oder CSV-Import von neuen Ziehungen.
|
||||
Perfekt als Fallback wenn Web-Scraping oder APIs nicht funktionieren.
|
||||
"""
|
||||
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
import shutil
|
||||
import os
|
||||
|
||||
|
||||
def load_existing_data(data_file):
|
||||
"""Lädt existierende Daten."""
|
||||
print("📂 Lade existierende Daten...")
|
||||
|
||||
if not os.path.exists(data_file):
|
||||
print(" ⚠️ Datei existiert nicht - erstelle neue")
|
||||
return pd.DataFrame(columns=['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ'])
|
||||
|
||||
df = pd.read_csv(data_file, sep=';')
|
||||
df['datum'] = pd.to_datetime(df['datum'], format='%Y-%m-%d', errors='coerce')
|
||||
|
||||
print(f" ✅ {len(df)} Ziehungen geladen")
|
||||
if len(df) > 0:
|
||||
print(f" 📅 Neueste: {df['datum'].max().strftime('%Y-%m-%d')}")
|
||||
|
||||
return df
|
||||
|
||||
|
||||
def create_backup(data_file):
|
||||
"""Erstellt Backup."""
|
||||
if not os.path.exists(data_file):
|
||||
return
|
||||
|
||||
print("\n💾 Erstelle Backup...")
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
|
||||
# Backup im Lotto/data/backups Verzeichnis
|
||||
parent_dir = os.path.dirname(data_file)
|
||||
lotto_dir = os.path.join(parent_dir, "Lotto")
|
||||
backup_dir = os.path.join(lotto_dir, "data", "backups")
|
||||
os.makedirs(backup_dir, exist_ok=True)
|
||||
|
||||
backup_file = os.path.join(backup_dir, f"{os.path.basename(data_file)}.backup_{timestamp}")
|
||||
shutil.copy2(data_file, backup_file)
|
||||
print(f" ✅ Backup: {os.path.basename(backup_file)}")
|
||||
|
||||
|
||||
def manual_input_mode():
|
||||
"""Manuelle Eingabe von Ziehungen."""
|
||||
print("\n⌨️ MANUELLE EINGABE")
|
||||
print("=" * 70)
|
||||
print("Format: YYYY-MM-DD Z1 Z2 Z3 Z4 Z5 Z6 SZ")
|
||||
print("Beispiel: 2025-01-22 7 14 21 28 35 42 3")
|
||||
print("Leer lassen zum Beenden.\n")
|
||||
|
||||
draws = []
|
||||
while True:
|
||||
user_input = input(f"Ziehung {len(draws) + 1}: ").strip()
|
||||
|
||||
if not user_input:
|
||||
break
|
||||
|
||||
try:
|
||||
parts = user_input.split()
|
||||
if len(parts) != 8:
|
||||
print(" ❌ Ungültiges Format. Bitte 8 Werte eingeben.")
|
||||
continue
|
||||
|
||||
date_obj = datetime.strptime(parts[0], '%Y-%m-%d')
|
||||
numbers = [int(parts[i]) for i in range(1, 7)]
|
||||
superzahl = int(parts[7])
|
||||
|
||||
# Validierung
|
||||
if not all(1 <= n <= 49 for n in numbers):
|
||||
print(" ❌ Hauptzahlen müssen zwischen 1 und 49 liegen.")
|
||||
continue
|
||||
|
||||
if not 0 <= superzahl <= 9:
|
||||
print(" ❌ Superzahl muss zwischen 0 und 9 liegen.")
|
||||
continue
|
||||
|
||||
if len(set(numbers)) != 6:
|
||||
print(" ❌ Hauptzahlen müssen eindeutig sein.")
|
||||
continue
|
||||
|
||||
# Sortiere Zahlen
|
||||
numbers_sorted = sorted(numbers)
|
||||
|
||||
draws.append({
|
||||
'datum': date_obj,
|
||||
'Z1': numbers_sorted[0],
|
||||
'Z2': numbers_sorted[1],
|
||||
'Z3': numbers_sorted[2],
|
||||
'Z4': numbers_sorted[3],
|
||||
'Z5': numbers_sorted[4],
|
||||
'Z6': numbers_sorted[5],
|
||||
'SZ': superzahl
|
||||
})
|
||||
|
||||
print(f" ✅ Hinzugefügt: {date_obj.strftime('%Y-%m-%d')} | "
|
||||
f"{'-'.join(map(str, numbers_sorted))} | SZ: {superzahl}")
|
||||
|
||||
except ValueError as e:
|
||||
print(f" ❌ Fehler: {e}")
|
||||
continue
|
||||
|
||||
return draws
|
||||
|
||||
|
||||
def csv_import_mode():
|
||||
"""CSV-Import von Ziehungen."""
|
||||
print("\n📁 CSV-IMPORT")
|
||||
print("=" * 70)
|
||||
print("CSV-Format: datum;Z1;Z2;Z3;Z4;Z5;Z6;SZ")
|
||||
print("Beispiel: 2025-01-22;7;14;21;28;35;42;3\n")
|
||||
|
||||
csv_file = input("Pfad zur CSV-Datei: ").strip()
|
||||
|
||||
if not os.path.exists(csv_file):
|
||||
print(f" ❌ Datei nicht gefunden: {csv_file}")
|
||||
return []
|
||||
|
||||
try:
|
||||
df_import = pd.read_csv(csv_file, sep=';')
|
||||
df_import['datum'] = pd.to_datetime(df_import['datum'], format='%Y-%m-%d', errors='coerce')
|
||||
|
||||
draws = df_import.to_dict('records')
|
||||
print(f" ✅ {len(draws)} Ziehungen aus CSV geladen")
|
||||
|
||||
# Zeige erste 3
|
||||
for i, draw in enumerate(draws[:3]):
|
||||
nums = f"{draw['Z1']}-{draw['Z2']}-{draw['Z3']}-{draw['Z4']}-{draw['Z5']}-{draw['Z6']}"
|
||||
print(f" {i+1}. {draw['datum'].strftime('%Y-%m-%d')} | {nums} | SZ: {draw['SZ']}")
|
||||
|
||||
if len(draws) > 3:
|
||||
print(f" ... und {len(draws) - 3} weitere")
|
||||
|
||||
return draws
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Fehler beim Laden: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def merge_and_save(df_existing, new_draws, data_file):
|
||||
"""Merged und speichert Daten."""
|
||||
if not new_draws:
|
||||
print("\n⚠️ Keine neuen Ziehungen zum Hinzufügen")
|
||||
return False
|
||||
|
||||
print(f"\n🔀 Merge {len(new_draws)} neue Ziehung(en)...")
|
||||
|
||||
df_new = pd.DataFrame(new_draws)
|
||||
df_combined = pd.concat([df_existing, df_new], ignore_index=True)
|
||||
|
||||
# Duplikate entfernen
|
||||
before = len(df_combined)
|
||||
df_combined = df_combined.drop_duplicates(subset=['datum'], keep='first')
|
||||
after = len(df_combined)
|
||||
|
||||
if before - after > 0:
|
||||
print(f" 🗑️ {before - after} Duplikat(e) entfernt")
|
||||
|
||||
# Sortieren
|
||||
df_combined = df_combined.sort_values('datum', ascending=True).reset_index(drop=True)
|
||||
|
||||
# Speichern
|
||||
print(f"\n💾 Speichere {len(df_combined)} Ziehungen...")
|
||||
|
||||
df_export = df_combined.copy()
|
||||
df_export['datum'] = df_export['datum'].dt.strftime('%Y-%m-%d')
|
||||
|
||||
df_export.to_csv(data_file, sep=';', index=False)
|
||||
|
||||
print(f" ✅ Gespeichert: {data_file}")
|
||||
print(f" 📊 Vorher: {len(df_existing)} | Neu: {len(df_combined) - len(df_existing)} | Gesamt: {len(df_combined)}")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def main():
|
||||
"""Hauptfunktion."""
|
||||
print()
|
||||
print("=" * 70)
|
||||
print(" LOTTO 6AUS49 SIMPLE UPDATER")
|
||||
print(" Manuelle Eingabe oder CSV-Import")
|
||||
print("=" * 70)
|
||||
print()
|
||||
|
||||
# Datei-Pfad
|
||||
default_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data/AlleLottozahlen.csv"
|
||||
|
||||
print(f"Standard-Datei: {default_file}")
|
||||
use_default = input("\nStandard verwenden? (j/n): ").strip().lower()
|
||||
|
||||
if use_default in ['j', 'ja', 'y', 'yes', '']:
|
||||
data_file = default_file
|
||||
else:
|
||||
data_file = input("Pfad zur Datei: ").strip()
|
||||
|
||||
print()
|
||||
|
||||
# Lade existierende Daten
|
||||
df_existing = load_existing_data(data_file)
|
||||
|
||||
# Backup
|
||||
create_backup(data_file)
|
||||
|
||||
# Eingabe-Modus wählen
|
||||
print("\n📝 EINGABE-MODUS")
|
||||
print("=" * 70)
|
||||
print("1. Manuelle Eingabe (einzelne Ziehungen)")
|
||||
print("2. CSV-Import (mehrere Ziehungen)")
|
||||
print()
|
||||
|
||||
mode = input("Wahl (1/2): ").strip()
|
||||
|
||||
if mode == '2':
|
||||
new_draws = csv_import_mode()
|
||||
else:
|
||||
new_draws = manual_input_mode()
|
||||
|
||||
# Merge und speichern
|
||||
success = merge_and_save(df_existing, new_draws, data_file)
|
||||
|
||||
print()
|
||||
print("=" * 70)
|
||||
if success:
|
||||
print("✅ UPDATE ERFOLGREICH")
|
||||
else:
|
||||
print("⚠️ KEINE ÄNDERUNGEN")
|
||||
print("=" * 70)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,496 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Lotto 6aus49 API Data Updater
|
||||
|
||||
Lädt Lotto-Ziehungen von verschiedenen APIs:
|
||||
1. lottoAPI (https://lottoapi.herokuapp.com)
|
||||
2. Lotto.de API
|
||||
3. Fallback Quellen
|
||||
|
||||
Features:
|
||||
- Automatisches Laden von APIs
|
||||
- Fallback auf andere APIs wenn primäre fehlschlägt
|
||||
- Duplikate-Vermeidung
|
||||
- Automatisches Backup
|
||||
"""
|
||||
|
||||
import requests
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
import shutil
|
||||
import os
|
||||
from typing import List, Dict, Optional
|
||||
import time
|
||||
|
||||
|
||||
class LottoAPIUpdater:
|
||||
"""Updater für Lotto 6aus49 über APIs."""
|
||||
|
||||
def __init__(self, data_file: str):
|
||||
self.data_file = data_file
|
||||
self.backup_file = None
|
||||
self.df_existing = None
|
||||
|
||||
# API-Endpoints
|
||||
self.apis = {
|
||||
'github': {
|
||||
'url': 'https://johannesfriedrich.github.io/LottoNumberArchive/Lottonumbers_tidy_complete.json',
|
||||
'name': 'GitHub Lotto Archive',
|
||||
'parser': self._parse_github_archive
|
||||
},
|
||||
'lottoapi': {
|
||||
'url': 'https://lottoapi.herokuapp.com/lotto/6aus49/100',
|
||||
'name': 'lottoAPI (Backup)',
|
||||
'parser': self._parse_lottoapi
|
||||
}
|
||||
}
|
||||
|
||||
print("🔄 LOTTO 6AUS49 API UPDATER")
|
||||
print("=" * 70)
|
||||
print(f"📁 Datei: {os.path.basename(data_file)}")
|
||||
print("=" * 70)
|
||||
|
||||
def load_existing_data(self) -> bool:
|
||||
"""Lädt existierende Daten."""
|
||||
print("\n📂 Lade existierende Daten...")
|
||||
|
||||
try:
|
||||
if not os.path.exists(self.data_file):
|
||||
print(f" ⚠️ Datei existiert nicht - erstelle neue")
|
||||
self.df_existing = pd.DataFrame(
|
||||
columns=['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
)
|
||||
return True
|
||||
|
||||
self.df_existing = pd.read_csv(self.data_file, sep=';')
|
||||
|
||||
# Konvertiere Datum
|
||||
if 'datum' in self.df_existing.columns:
|
||||
self.df_existing['datum'] = pd.to_datetime(
|
||||
self.df_existing['datum'],
|
||||
format='%Y-%m-%d',
|
||||
errors='coerce'
|
||||
)
|
||||
|
||||
print(f" ✅ {len(self.df_existing)} Ziehungen geladen")
|
||||
|
||||
if len(self.df_existing) > 0:
|
||||
latest = self.df_existing['datum'].max()
|
||||
oldest = self.df_existing['datum'].min()
|
||||
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Fehler: {e}")
|
||||
return False
|
||||
|
||||
def create_backup(self) -> bool:
|
||||
"""Erstellt Backup."""
|
||||
if not os.path.exists(self.data_file):
|
||||
return True
|
||||
|
||||
try:
|
||||
print("\n💾 Erstelle Backup...")
|
||||
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
backup_dir = os.path.join(os.path.dirname(self.data_file), "data", "backups")
|
||||
os.makedirs(backup_dir, exist_ok=True)
|
||||
|
||||
filename = os.path.basename(self.data_file)
|
||||
self.backup_file = os.path.join(backup_dir, f"{filename}.backup_{timestamp}")
|
||||
|
||||
shutil.copy2(self.data_file, self.backup_file)
|
||||
print(f" ✅ Backup: {os.path.basename(self.backup_file)}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Backup-Fehler: {e}")
|
||||
return False
|
||||
|
||||
def fetch_from_api(self, api_name: str) -> List[Dict]:
|
||||
"""
|
||||
Holt Daten von spezifischer API.
|
||||
|
||||
Args:
|
||||
api_name: Name der API ('lottoapi')
|
||||
|
||||
Returns:
|
||||
Liste von Ziehungen
|
||||
"""
|
||||
if api_name not in self.apis:
|
||||
print(f" ❌ Unbekannte API: {api_name}")
|
||||
return []
|
||||
|
||||
api = self.apis[api_name]
|
||||
print(f"\n🌐 Versuche {api['name']}...")
|
||||
print(f" URL: {api['url']}")
|
||||
|
||||
try:
|
||||
headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
|
||||
}
|
||||
|
||||
response = requests.get(api['url'], headers=headers, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
print(f" ✅ Antwort erhalten ({len(response.content)} bytes)")
|
||||
|
||||
# Parse mit spezifischem Parser
|
||||
draws = api['parser'](response.json())
|
||||
|
||||
if draws:
|
||||
print(f" ✅ {len(draws)} Ziehungen extrahiert")
|
||||
else:
|
||||
print(f" ⚠️ Keine Ziehungen extrahiert")
|
||||
|
||||
return draws
|
||||
|
||||
except requests.RequestException as e:
|
||||
print(f" ❌ Netzwerkfehler: {e}")
|
||||
return []
|
||||
except Exception as e:
|
||||
print(f" ❌ Fehler: {e}")
|
||||
return []
|
||||
|
||||
def _parse_github_archive(self, data: List[Dict]) -> List[Dict]:
|
||||
"""
|
||||
Parst GitHub Lotto Archive Response.
|
||||
|
||||
Expected format:
|
||||
[
|
||||
{
|
||||
"id": 4963,
|
||||
"date": "26.11.2025",
|
||||
"variable": "Lottozahl",
|
||||
"value": 2
|
||||
},
|
||||
...
|
||||
]
|
||||
"""
|
||||
draws = []
|
||||
|
||||
try:
|
||||
# Gruppiere nach ID (jede ID = eine Ziehung)
|
||||
draws_by_id = {}
|
||||
|
||||
for entry in data:
|
||||
draw_id = entry.get('id')
|
||||
if draw_id not in draws_by_id:
|
||||
draws_by_id[draw_id] = {
|
||||
'date': entry.get('date'),
|
||||
'numbers': [],
|
||||
'superzahl': None
|
||||
}
|
||||
|
||||
variable = entry.get('variable')
|
||||
value = entry.get('value')
|
||||
|
||||
if variable == 'Lottozahl':
|
||||
draws_by_id[draw_id]['numbers'].append(value)
|
||||
elif variable == 'Superzahl':
|
||||
draws_by_id[draw_id]['superzahl'] = value
|
||||
|
||||
# Konvertiere zu unserem Format
|
||||
for draw_id, draw_data in draws_by_id.items():
|
||||
try:
|
||||
# Parse Datum (Format: DD.MM.YYYY)
|
||||
date_str = draw_data['date']
|
||||
day, month, year = date_str.split('.')
|
||||
date_obj = datetime(int(year), int(month), int(day))
|
||||
|
||||
# Sortiere Zahlen
|
||||
numbers = sorted(draw_data['numbers'])
|
||||
|
||||
# Validierung
|
||||
if len(numbers) != 6:
|
||||
continue
|
||||
|
||||
superzahl = draw_data['superzahl']
|
||||
if superzahl is None:
|
||||
continue
|
||||
|
||||
draw = {
|
||||
'datum': date_obj,
|
||||
'Z1': numbers[0],
|
||||
'Z2': numbers[1],
|
||||
'Z3': numbers[2],
|
||||
'Z4': numbers[3],
|
||||
'Z5': numbers[4],
|
||||
'Z6': numbers[5],
|
||||
'SZ': superzahl
|
||||
}
|
||||
|
||||
if self._validate_draw(draw):
|
||||
draws.append(draw)
|
||||
|
||||
except Exception as e:
|
||||
continue
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Parse-Fehler: {e}")
|
||||
|
||||
return draws
|
||||
|
||||
def _parse_lottoapi(self, data: List[Dict]) -> List[Dict]:
|
||||
"""
|
||||
Parst lottoAPI Response.
|
||||
|
||||
Expected format:
|
||||
[
|
||||
{
|
||||
"date": "2025-01-22",
|
||||
"numbers": [3, 9, 12, 24, 39, 45],
|
||||
"superzahl": 7
|
||||
},
|
||||
...
|
||||
]
|
||||
"""
|
||||
draws = []
|
||||
|
||||
try:
|
||||
for item in data:
|
||||
# Datum
|
||||
date_str = item.get('date')
|
||||
if not date_str:
|
||||
continue
|
||||
|
||||
date_obj = datetime.strptime(date_str, '%Y-%m-%d')
|
||||
|
||||
# Hauptzahlen (6 Zahlen)
|
||||
numbers = sorted(item.get('numbers', []))
|
||||
if len(numbers) != 6:
|
||||
continue
|
||||
|
||||
# Superzahl
|
||||
superzahl = item.get('superzahl')
|
||||
if superzahl is None:
|
||||
continue
|
||||
|
||||
draw = {
|
||||
'datum': date_obj,
|
||||
'Z1': numbers[0],
|
||||
'Z2': numbers[1],
|
||||
'Z3': numbers[2],
|
||||
'Z4': numbers[3],
|
||||
'Z5': numbers[4],
|
||||
'Z6': numbers[5],
|
||||
'SZ': superzahl
|
||||
}
|
||||
|
||||
if self._validate_draw(draw):
|
||||
draws.append(draw)
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Parse-Fehler: {e}")
|
||||
|
||||
return draws
|
||||
|
||||
def _validate_draw(self, draw: Dict) -> bool:
|
||||
"""Validiert eine Ziehung."""
|
||||
try:
|
||||
required_fields = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
if not all(field in draw for field in required_fields):
|
||||
return False
|
||||
|
||||
# Hauptzahlen (1-49)
|
||||
main_numbers = [draw[f'Z{i}'] for i in range(1, 7)]
|
||||
if not all(isinstance(n, int) and 1 <= n <= 49 for n in main_numbers):
|
||||
return False
|
||||
|
||||
if len(set(main_numbers)) != 6:
|
||||
return False
|
||||
|
||||
# Superzahl (0-9)
|
||||
if not isinstance(draw['SZ'], int) or not 0 <= draw['SZ'] <= 9:
|
||||
return False
|
||||
|
||||
if not isinstance(draw['datum'], datetime):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def fetch_from_all_apis(self) -> List[Dict]:
|
||||
"""
|
||||
Versucht alle APIs nacheinander.
|
||||
|
||||
Returns:
|
||||
Liste von Ziehungen
|
||||
"""
|
||||
print("\n🔍 SUCHE NACH DATEN VON APIs")
|
||||
print("=" * 70)
|
||||
|
||||
all_draws = []
|
||||
|
||||
# Versuche alle APIs (GitHub zuerst, da zuverlässig)
|
||||
for api_name in ['github', 'lottoapi']:
|
||||
draws = self.fetch_from_api(api_name)
|
||||
|
||||
if draws:
|
||||
all_draws.extend(draws)
|
||||
print(f" ✅ {len(draws)} Ziehungen von {self.apis[api_name]['name']}")
|
||||
break # Erste erfolgreiche API nutzen
|
||||
else:
|
||||
print(f" ⏭️ Weiter zur nächsten API...")
|
||||
|
||||
time.sleep(1) # Pause zwischen APIs
|
||||
|
||||
if not all_draws:
|
||||
print("\n ❌ Keine Daten von APIs erhalten")
|
||||
print(" 💡 Tipp: Verwende simple_update.py für manuelle Eingabe")
|
||||
|
||||
return all_draws
|
||||
|
||||
def merge_with_existing(self, new_draws: List[Dict]) -> pd.DataFrame:
|
||||
"""Merged neue Ziehungen mit existierenden Daten."""
|
||||
print(f"\n🔀 Merge mit existierenden Daten...")
|
||||
|
||||
if not new_draws:
|
||||
print(" ⚠️ Keine neuen Ziehungen zum Mergen")
|
||||
return self.df_existing
|
||||
|
||||
df_new = pd.DataFrame(new_draws)
|
||||
|
||||
if self.df_existing is None or len(self.df_existing) == 0:
|
||||
df_combined = df_new
|
||||
print(f" ✅ Neue Datei erstellt mit {len(df_combined)} Ziehungen")
|
||||
else:
|
||||
df_combined = pd.concat([self.df_existing, df_new], ignore_index=True)
|
||||
|
||||
before_dedup = len(df_combined)
|
||||
df_combined = df_combined.drop_duplicates(subset=['datum'], keep='first')
|
||||
after_dedup = len(df_combined)
|
||||
|
||||
duplicates = before_dedup - after_dedup
|
||||
if duplicates > 0:
|
||||
print(f" 🗑️ {duplicates} Duplikat(e) entfernt")
|
||||
|
||||
df_combined = df_combined.sort_values('datum', ascending=True)
|
||||
df_combined = df_combined.reset_index(drop=True)
|
||||
|
||||
new_entries = len(df_combined) - (len(self.df_existing) if self.df_existing is not None else 0)
|
||||
|
||||
print(f" ✅ Merge abgeschlossen:")
|
||||
print(f" Vorher: {len(self.df_existing) if self.df_existing is not None else 0} Ziehungen")
|
||||
print(f" Neue: {new_entries} Ziehungen")
|
||||
print(f" Gesamt: {len(df_combined)} Ziehungen")
|
||||
|
||||
return df_combined
|
||||
|
||||
def save_data(self, df: pd.DataFrame) -> bool:
|
||||
"""Speichert Daten."""
|
||||
try:
|
||||
print(f"\n💾 Speichere Daten...")
|
||||
|
||||
df_export = df.copy()
|
||||
df_export['datum'] = df_export['datum'].dt.strftime('%Y-%m-%d')
|
||||
|
||||
column_order = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
df_export = df_export[column_order]
|
||||
|
||||
df_export.to_csv(self.data_file, sep=';', index=False)
|
||||
|
||||
print(f" ✅ Gespeichert: {self.data_file}")
|
||||
print(f" 📊 {len(df)} Ziehungen total")
|
||||
|
||||
if len(df) > 0:
|
||||
latest = df['datum'].max()
|
||||
oldest = df['datum'].min()
|
||||
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Speicherfehler: {e}")
|
||||
|
||||
if self.backup_file and os.path.exists(self.backup_file):
|
||||
print(" 🔄 Stelle Backup wieder her...")
|
||||
shutil.copy2(self.backup_file, self.data_file)
|
||||
print(" ✅ Backup wiederhergestellt")
|
||||
|
||||
return False
|
||||
|
||||
def update(self, api_name: str = 'all', create_backup: bool = True) -> bool:
|
||||
"""
|
||||
Führt komplettes Update durch.
|
||||
|
||||
Args:
|
||||
api_name: 'all' oder 'lottoapi'
|
||||
create_backup: Backup erstellen
|
||||
|
||||
Returns:
|
||||
True bei Erfolg
|
||||
"""
|
||||
print("\n🚀 STARTE UPDATE")
|
||||
print("=" * 70)
|
||||
|
||||
if not self.load_existing_data():
|
||||
return False
|
||||
|
||||
if create_backup:
|
||||
self.create_backup()
|
||||
|
||||
if api_name == 'all':
|
||||
new_draws = self.fetch_from_all_apis()
|
||||
else:
|
||||
new_draws = self.fetch_from_api(api_name)
|
||||
|
||||
if not new_draws:
|
||||
print("\n❌ Keine Daten von APIs geladen")
|
||||
print("💡 Tipp: Verwende simple_update.py für manuelle Eingabe")
|
||||
return False
|
||||
|
||||
df_updated = self.merge_with_existing(new_draws)
|
||||
success = self.save_data(df_updated)
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
if success:
|
||||
print("✅ UPDATE ERFOLGREICH ABGESCHLOSSEN")
|
||||
else:
|
||||
print("❌ UPDATE FEHLGESCHLAGEN")
|
||||
print("=" * 70)
|
||||
|
||||
return success
|
||||
|
||||
|
||||
def main():
|
||||
"""Hauptfunktion."""
|
||||
import sys
|
||||
|
||||
print()
|
||||
print("=" * 70)
|
||||
print(" LOTTO 6AUS49 API UPDATER")
|
||||
print(" Unterstützt: GitHub Lotto Archive, lottoAPI")
|
||||
print("=" * 70)
|
||||
print()
|
||||
|
||||
default_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data/AlleLottozahlen.csv"
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
data_file = sys.argv[1]
|
||||
api_name = sys.argv[2] if len(sys.argv) > 2 else 'all'
|
||||
else:
|
||||
print(f"Standard-Datei: {default_file}")
|
||||
use_default = input("\nStandard verwenden? (j/n): ").strip().lower()
|
||||
|
||||
if use_default in ['j', 'ja', 'y', 'yes', '']:
|
||||
data_file = default_file
|
||||
else:
|
||||
data_file = input("Pfad zur Datei: ").strip()
|
||||
|
||||
api_name = 'all'
|
||||
|
||||
print()
|
||||
|
||||
updater = LottoAPIUpdater(data_file)
|
||||
success = updater.update(api_name=api_name)
|
||||
|
||||
sys.exit(0 if success else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,499 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Lotto 6aus49 Data Updater für lotto.de
|
||||
|
||||
Web-Scraper für https://www.lotto.de/lotto-6aus49/lottozahlen
|
||||
|
||||
Features:
|
||||
- Automatisches Scraping aller historischen Ziehungen
|
||||
- Duplikate-Vermeidung
|
||||
- Automatisches Backup vor Update
|
||||
- Validierung der Daten
|
||||
- Fortschrittsanzeige
|
||||
"""
|
||||
|
||||
import pandas as pd
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from datetime import datetime
|
||||
import shutil
|
||||
import os
|
||||
from typing import List, Dict, Optional
|
||||
import time
|
||||
|
||||
|
||||
class LottoWebUpdater:
|
||||
"""Updater für lotto.de Web-Scraping"""
|
||||
|
||||
def __init__(self, data_file: str):
|
||||
self.data_file = data_file
|
||||
self.url = "https://www.lotto.de/lotto-6aus49/lottozahlen"
|
||||
self.backup_file = None
|
||||
self.df_existing = None
|
||||
|
||||
print("🔄 LOTTO 6AUS49 WEB UPDATER")
|
||||
print("=" * 70)
|
||||
print(f"📁 Datei: {os.path.basename(data_file)}")
|
||||
print(f"🌐 Quelle: {self.url}")
|
||||
print("=" * 70)
|
||||
|
||||
def load_existing_data(self) -> bool:
|
||||
"""Lädt existierende Daten."""
|
||||
print("\n📂 Lade existierende Daten...")
|
||||
|
||||
try:
|
||||
if not os.path.exists(self.data_file):
|
||||
print(f" ⚠️ Datei existiert nicht - erstelle neue")
|
||||
self.df_existing = pd.DataFrame(
|
||||
columns=['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
)
|
||||
return True
|
||||
|
||||
self.df_existing = pd.read_csv(self.data_file, sep=';')
|
||||
|
||||
# Konvertiere Datum
|
||||
if 'datum' in self.df_existing.columns:
|
||||
self.df_existing['datum'] = pd.to_datetime(
|
||||
self.df_existing['datum'],
|
||||
format='%Y-%m-%d',
|
||||
errors='coerce'
|
||||
)
|
||||
|
||||
print(f" ✅ {len(self.df_existing)} Ziehungen geladen")
|
||||
|
||||
if len(self.df_existing) > 0:
|
||||
latest = self.df_existing['datum'].max()
|
||||
oldest = self.df_existing['datum'].min()
|
||||
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Fehler: {e}")
|
||||
return False
|
||||
|
||||
def create_backup(self) -> bool:
|
||||
"""Erstellt Backup."""
|
||||
if not os.path.exists(self.data_file):
|
||||
return True
|
||||
|
||||
try:
|
||||
print("\n💾 Erstelle Backup...")
|
||||
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
parent_dir = os.path.dirname(self.data_file)
|
||||
lotto_dir = os.path.join(parent_dir, "Lotto")
|
||||
backup_dir = os.path.join(lotto_dir, "data", "backups")
|
||||
os.makedirs(backup_dir, exist_ok=True)
|
||||
|
||||
filename = os.path.basename(self.data_file)
|
||||
self.backup_file = os.path.join(backup_dir, f"{filename}.backup_{timestamp}")
|
||||
|
||||
shutil.copy2(self.data_file, self.backup_file)
|
||||
print(f" ✅ Backup: {os.path.basename(self.backup_file)}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Backup-Fehler: {e}")
|
||||
return False
|
||||
|
||||
def fetch_draws_from_web(self, max_pages: int = 5) -> List[Dict]:
|
||||
"""
|
||||
Scrapt Ziehungen von lotto.de
|
||||
|
||||
Args:
|
||||
max_pages: Maximale Anzahl Seiten (jede Seite ~20 Ziehungen)
|
||||
|
||||
Returns:
|
||||
Liste von Ziehungen
|
||||
"""
|
||||
print(f"\n🌐 Lade Daten von {self.url}...")
|
||||
print(f" 📄 Lade bis zu {max_pages} Seiten...")
|
||||
|
||||
all_draws = []
|
||||
|
||||
try:
|
||||
headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
|
||||
}
|
||||
|
||||
for page in range(1, max_pages + 1):
|
||||
url = f"{self.url}?page={page}" if page > 1 else self.url
|
||||
|
||||
print(f"\n 📄 Seite {page}/{max_pages}...")
|
||||
|
||||
response = requests.get(url, headers=headers, timeout=15)
|
||||
response.raise_for_status()
|
||||
|
||||
soup = BeautifulSoup(response.content, 'html.parser')
|
||||
|
||||
# Finde Ziehungen auf dieser Seite
|
||||
page_draws = self._extract_draws_from_page(soup, page)
|
||||
|
||||
if not page_draws:
|
||||
print(f" ⚠️ Keine Ziehungen gefunden - Abbruch")
|
||||
break
|
||||
|
||||
all_draws.extend(page_draws)
|
||||
print(f" ✅ {len(page_draws)} Ziehungen extrahiert")
|
||||
|
||||
# Rate limiting
|
||||
if page < max_pages:
|
||||
time.sleep(1)
|
||||
|
||||
print(f"\n ✅ Gesamt: {len(all_draws)} Ziehungen von {page} Seite(n)")
|
||||
return all_draws
|
||||
|
||||
except requests.RequestException as e:
|
||||
print(f" ❌ Netzwerkfehler: {e}")
|
||||
return all_draws # Return was bisher geladen wurde
|
||||
except Exception as e:
|
||||
print(f" ❌ Fehler: {e}")
|
||||
return all_draws
|
||||
|
||||
def _extract_draws_from_page(self, soup: BeautifulSoup, page_num: int) -> List[Dict]:
|
||||
"""
|
||||
Extrahiert Ziehungen von einer Seite.
|
||||
|
||||
Args:
|
||||
soup: BeautifulSoup Objekt
|
||||
page_num: Seitennummer
|
||||
|
||||
Returns:
|
||||
Liste von Ziehungen
|
||||
"""
|
||||
draws = []
|
||||
|
||||
try:
|
||||
# Verschiedene Selektoren versuchen
|
||||
# 1. Versuche mit data-attribute
|
||||
result_divs = soup.find_all('div', class_=lambda x: x and 'result' in x.lower())
|
||||
|
||||
if not result_divs:
|
||||
# 2. Versuche table
|
||||
result_divs = soup.find_all('tr', class_=lambda x: x and ('draw' in x.lower() or 'result' in x.lower()))
|
||||
|
||||
if not result_divs:
|
||||
# 3. Versuche generische Container
|
||||
result_divs = soup.find_all('div', class_=lambda x: x and 'drawing' in x.lower())
|
||||
|
||||
print(f" 🔍 {len(result_divs)} potentielle Ergebnis-Container gefunden")
|
||||
|
||||
for i, elem in enumerate(result_divs):
|
||||
try:
|
||||
# Extrahiere Datum
|
||||
date_elem = elem.find('time') or elem.find(class_=lambda x: x and 'date' in x.lower())
|
||||
if not date_elem:
|
||||
# Versuche direkten Text
|
||||
date_text = self._extract_date_text(elem.get_text())
|
||||
else:
|
||||
date_text = date_elem.get_text().strip()
|
||||
|
||||
date_obj = self._parse_date(date_text)
|
||||
|
||||
# Extrahiere Zahlen - verschiedene Ansätze
|
||||
numbers = self._extract_numbers(elem)
|
||||
|
||||
if len(numbers) < 7: # Mindestens 6 Hauptzahlen + 1 Superzahl
|
||||
continue
|
||||
|
||||
main_numbers = sorted(numbers[:6])
|
||||
superzahl = numbers[6]
|
||||
|
||||
draw = {
|
||||
'datum': date_obj,
|
||||
'Z1': main_numbers[0],
|
||||
'Z2': main_numbers[1],
|
||||
'Z3': main_numbers[2],
|
||||
'Z4': main_numbers[3],
|
||||
'Z5': main_numbers[4],
|
||||
'Z6': main_numbers[5],
|
||||
'SZ': superzahl
|
||||
}
|
||||
|
||||
if self._validate_draw(draw):
|
||||
draws.append(draw)
|
||||
else:
|
||||
print(f" ⚠️ Element {i+1}: Validierung fehlgeschlagen")
|
||||
|
||||
except Exception as e:
|
||||
continue
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Fehler beim Extrahieren: {e}")
|
||||
|
||||
return draws
|
||||
|
||||
def _extract_date_text(self, text: str) -> str:
|
||||
"""Extrahiert Datum aus Text."""
|
||||
import re
|
||||
|
||||
# Suche nach Datum im Format DD.MM.YYYY oder YYYY-MM-DD
|
||||
patterns = [
|
||||
r'(\d{2}\.\d{2}\.\d{4})',
|
||||
r'(\d{4}-\d{2}-\d{2})',
|
||||
r'(\d{1,2}\.\s*\w+\s*\d{4})' # z.B. "22. Januar 2025"
|
||||
]
|
||||
|
||||
for pattern in patterns:
|
||||
match = re.search(pattern, text)
|
||||
if match:
|
||||
return match.group(1)
|
||||
|
||||
return text.strip()
|
||||
|
||||
def _extract_numbers(self, elem) -> List[int]:
|
||||
"""Extrahiert Zahlen aus Element."""
|
||||
numbers = []
|
||||
|
||||
# 1. Versuche mit span/div mit Klassen wie 'ball', 'number', etc.
|
||||
number_elems = elem.find_all(class_=lambda x: x and any(c in x.lower() for c in ['ball', 'number', 'zahl']))
|
||||
|
||||
if number_elems:
|
||||
for num_elem in number_elems:
|
||||
try:
|
||||
text = num_elem.get_text().strip()
|
||||
# Extrahiere nur Zahlen
|
||||
import re
|
||||
nums = re.findall(r'\d+', text)
|
||||
if nums:
|
||||
numbers.append(int(nums[0]))
|
||||
except:
|
||||
continue
|
||||
|
||||
# 2. Fallback: Alle Zahlen aus Text extrahieren
|
||||
if len(numbers) < 7:
|
||||
import re
|
||||
text = elem.get_text()
|
||||
# Finde alle Zahlen
|
||||
all_nums = re.findall(r'\b(\d+)\b', text)
|
||||
|
||||
# Filtere plausible Lotto-Zahlen (1-49 für Hauptzahlen, 0-9 für Superzahl)
|
||||
plausible = []
|
||||
for n in all_nums:
|
||||
num = int(n)
|
||||
if 1 <= num <= 49:
|
||||
plausible.append(num)
|
||||
elif 0 <= num <= 9 and len(plausible) >= 6:
|
||||
# Könnte Superzahl sein
|
||||
plausible.append(num)
|
||||
|
||||
if len(plausible) >= 7:
|
||||
numbers = plausible[:7]
|
||||
|
||||
return numbers
|
||||
|
||||
def _parse_date(self, date_str: str) -> datetime:
|
||||
"""Parst Datum."""
|
||||
import re
|
||||
|
||||
# Clean up
|
||||
date_str = date_str.strip()
|
||||
|
||||
# Format 1: DD.MM.YYYY
|
||||
try:
|
||||
return datetime.strptime(date_str, '%d.%m.%Y')
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Format 2: YYYY-MM-DD
|
||||
try:
|
||||
return datetime.strptime(date_str, '%Y-%m-%d')
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Format 3: DD. MONAT YYYY (z.B. "22. Januar 2025")
|
||||
months_de = {
|
||||
'januar': 1, 'februar': 2, 'märz': 3, 'april': 4,
|
||||
'mai': 5, 'juni': 6, 'juli': 7, 'august': 8,
|
||||
'september': 9, 'oktober': 10, 'november': 11, 'dezember': 12
|
||||
}
|
||||
|
||||
match = re.search(r'(\d{1,2})\.\s*(\w+)\s*(\d{4})', date_str.lower())
|
||||
if match:
|
||||
day = int(match.group(1))
|
||||
month_str = match.group(2)
|
||||
year = int(match.group(3))
|
||||
|
||||
if month_str in months_de:
|
||||
return datetime(year, months_de[month_str], day)
|
||||
|
||||
# Fallback: Aktuelles Datum
|
||||
print(f" ⚠️ Konnte Datum nicht parsen: {date_str}")
|
||||
return datetime.now()
|
||||
|
||||
def _validate_draw(self, draw: Dict) -> bool:
|
||||
"""Validiert eine Ziehung."""
|
||||
try:
|
||||
required_fields = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
if not all(field in draw for field in required_fields):
|
||||
return False
|
||||
|
||||
# Hauptzahlen (1-49)
|
||||
main_numbers = [draw[f'Z{i}'] for i in range(1, 7)]
|
||||
if not all(isinstance(n, int) and 1 <= n <= 49 for n in main_numbers):
|
||||
return False
|
||||
|
||||
if len(set(main_numbers)) != 6:
|
||||
return False
|
||||
|
||||
# Superzahl (0-9)
|
||||
if not isinstance(draw['SZ'], int) or not 0 <= draw['SZ'] <= 9:
|
||||
return False
|
||||
|
||||
if not isinstance(draw['datum'], datetime):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def merge_with_existing(self, new_draws: List[Dict]) -> pd.DataFrame:
|
||||
"""Merged neue Ziehungen mit existierenden Daten."""
|
||||
print(f"\n🔀 Merge mit existierenden Daten...")
|
||||
|
||||
if not new_draws:
|
||||
print(" ⚠️ Keine neuen Ziehungen zum Mergen")
|
||||
return self.df_existing
|
||||
|
||||
df_new = pd.DataFrame(new_draws)
|
||||
|
||||
if self.df_existing is None or len(self.df_existing) == 0:
|
||||
df_combined = df_new
|
||||
print(f" ✅ Neue Datei erstellt mit {len(df_combined)} Ziehungen")
|
||||
else:
|
||||
df_combined = pd.concat([self.df_existing, df_new], ignore_index=True)
|
||||
|
||||
before_dedup = len(df_combined)
|
||||
df_combined = df_combined.drop_duplicates(subset=['datum'], keep='first')
|
||||
after_dedup = len(df_combined)
|
||||
|
||||
duplicates = before_dedup - after_dedup
|
||||
if duplicates > 0:
|
||||
print(f" 🗑️ {duplicates} Duplikat(e) entfernt")
|
||||
|
||||
df_combined = df_combined.sort_values('datum', ascending=True)
|
||||
df_combined = df_combined.reset_index(drop=True)
|
||||
|
||||
new_entries = len(df_combined) - (len(self.df_existing) if self.df_existing is not None else 0)
|
||||
|
||||
print(f" ✅ Merge abgeschlossen:")
|
||||
print(f" Vorher: {len(self.df_existing) if self.df_existing is not None else 0} Ziehungen")
|
||||
print(f" Neue: {new_entries} Ziehungen")
|
||||
print(f" Gesamt: {len(df_combined)} Ziehungen")
|
||||
|
||||
return df_combined
|
||||
|
||||
def save_data(self, df: pd.DataFrame) -> bool:
|
||||
"""Speichert Daten."""
|
||||
try:
|
||||
print(f"\n💾 Speichere Daten...")
|
||||
|
||||
df_export = df.copy()
|
||||
df_export['datum'] = df_export['datum'].dt.strftime('%Y-%m-%d')
|
||||
|
||||
column_order = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
df_export = df_export[column_order]
|
||||
|
||||
df_export.to_csv(self.data_file, sep=';', index=False)
|
||||
|
||||
print(f" ✅ Gespeichert: {self.data_file}")
|
||||
print(f" 📊 {len(df)} Ziehungen total")
|
||||
|
||||
if len(df) > 0:
|
||||
latest = df['datum'].max()
|
||||
oldest = df['datum'].min()
|
||||
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Speicherfehler: {e}")
|
||||
|
||||
if self.backup_file and os.path.exists(self.backup_file):
|
||||
print(" 🔄 Stelle Backup wieder her...")
|
||||
shutil.copy2(self.backup_file, self.data_file)
|
||||
print(" ✅ Backup wiederhergestellt")
|
||||
|
||||
return False
|
||||
|
||||
def update(self, max_pages: int = 5, create_backup: bool = True) -> bool:
|
||||
"""
|
||||
Führt komplettes Update durch.
|
||||
|
||||
Args:
|
||||
max_pages: Maximale Anzahl Seiten zum Scrapen
|
||||
create_backup: Backup vor Update erstellen
|
||||
|
||||
Returns:
|
||||
True bei Erfolg
|
||||
"""
|
||||
print("\n🚀 STARTE UPDATE")
|
||||
print("=" * 70)
|
||||
|
||||
if not self.load_existing_data():
|
||||
return False
|
||||
|
||||
if create_backup:
|
||||
self.create_backup()
|
||||
|
||||
new_draws = self.fetch_draws_from_web(max_pages=max_pages)
|
||||
|
||||
if not new_draws:
|
||||
print("\n❌ Keine Daten von Website geladen")
|
||||
print("💡 Tipp: Verwende simple_update.py für manuelle Eingabe")
|
||||
return False
|
||||
|
||||
df_updated = self.merge_with_existing(new_draws)
|
||||
success = self.save_data(df_updated)
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
if success:
|
||||
print("✅ UPDATE ERFOLGREICH ABGESCHLOSSEN")
|
||||
else:
|
||||
print("❌ UPDATE FEHLGESCHLAGEN")
|
||||
print("=" * 70)
|
||||
|
||||
return success
|
||||
|
||||
|
||||
def main():
|
||||
"""Hauptfunktion."""
|
||||
import sys
|
||||
|
||||
print()
|
||||
print("=" * 70)
|
||||
print(" LOTTO 6AUS49 WEB UPDATER")
|
||||
print(" Quelle: lotto.de")
|
||||
print("=" * 70)
|
||||
print()
|
||||
|
||||
default_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data/AlleLottozahlen.csv"
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
data_file = sys.argv[1]
|
||||
max_pages = int(sys.argv[2]) if len(sys.argv) > 2 else 5
|
||||
else:
|
||||
print(f"Standard-Datei: {default_file}")
|
||||
use_default = input("\nStandard verwenden? (j/n): ").strip().lower()
|
||||
|
||||
if use_default in ['j', 'ja', 'y', 'yes', '']:
|
||||
data_file = default_file
|
||||
else:
|
||||
data_file = input("Pfad zur Datei: ").strip()
|
||||
|
||||
pages_input = input("\nAnzahl Seiten laden (1-10, default=5): ").strip()
|
||||
max_pages = int(pages_input) if pages_input else 5
|
||||
|
||||
print()
|
||||
|
||||
updater = LottoWebUpdater(data_file)
|
||||
success = updater.update(max_pages=max_pages)
|
||||
|
||||
sys.exit(0 if success else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Executable
+460
@@ -0,0 +1,460 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
CSV-Validator für Lotto 6aus49 und Eurojackpot
|
||||
|
||||
Validiert CSV-Dateien auf:
|
||||
- Korrekte Spaltenstruktur
|
||||
- Datumformat und -konsistenz
|
||||
- Zahlenbereich und Eindeutigkeit
|
||||
- Wochentag (nur Mi/Sa für Lotto, nur Di/Fr für Eurojackpot)
|
||||
- Chronologische Sortierung
|
||||
- Duplikate
|
||||
- Fehlende Werte
|
||||
- Zukunftsdaten
|
||||
"""
|
||||
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
import sys
|
||||
import os
|
||||
from typing import List, Dict, Tuple
|
||||
|
||||
|
||||
class LottoCSVValidator:
|
||||
"""Validator für Lotto 6aus49 CSV-Dateien."""
|
||||
|
||||
def __init__(self, csv_file: str):
|
||||
self.csv_file = csv_file
|
||||
self.df = None
|
||||
self.errors = []
|
||||
self.warnings = []
|
||||
self.lottery_type = self._detect_lottery_type()
|
||||
|
||||
def _detect_lottery_type(self) -> str:
|
||||
"""Erkennt ob Lotto oder Eurojackpot basierend auf Dateiname."""
|
||||
basename = os.path.basename(self.csv_file).lower()
|
||||
if 'eurojackpot' in basename:
|
||||
return 'eurojackpot'
|
||||
elif 'lotto' in basename:
|
||||
return 'lotto'
|
||||
else:
|
||||
# Versuche anhand der Spalten zu erkennen
|
||||
return 'unknown'
|
||||
|
||||
def load_csv(self) -> bool:
|
||||
"""Lädt CSV-Datei."""
|
||||
try:
|
||||
if not os.path.exists(self.csv_file):
|
||||
self.errors.append(f"❌ Datei existiert nicht: {self.csv_file}")
|
||||
return False
|
||||
|
||||
self.df = pd.read_csv(self.csv_file, sep=';')
|
||||
|
||||
# Auto-detect wenn noch unknown
|
||||
if self.lottery_type == 'unknown':
|
||||
if 'SZ2' in self.df.columns:
|
||||
self.lottery_type = 'eurojackpot'
|
||||
elif 'SZ' in self.df.columns:
|
||||
self.lottery_type = 'lotto'
|
||||
else:
|
||||
self.errors.append("❌ Kann Lottery-Typ nicht erkennen")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.errors.append(f"❌ Fehler beim Laden: {e}")
|
||||
return False
|
||||
|
||||
def validate_structure(self) -> bool:
|
||||
"""Validiert Spaltenstruktur."""
|
||||
if self.lottery_type == 'lotto':
|
||||
expected_cols = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
elif self.lottery_type == 'eurojackpot':
|
||||
expected_cols = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']
|
||||
else:
|
||||
self.errors.append("❌ Unbekannter Lottery-Typ")
|
||||
return False
|
||||
|
||||
actual_cols = list(self.df.columns)
|
||||
|
||||
if actual_cols != expected_cols:
|
||||
self.errors.append(f"❌ Spaltenstruktur falsch")
|
||||
self.errors.append(f" Erwartet: {expected_cols}")
|
||||
self.errors.append(f" Gefunden: {actual_cols}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def validate_dates(self) -> bool:
|
||||
"""Validiert Datumsspalte."""
|
||||
valid = True
|
||||
|
||||
# Prüfe auf leere Daten
|
||||
empty_dates = self.df[self.df['datum'].isna() | (self.df['datum'] == '')]
|
||||
if len(empty_dates) > 0:
|
||||
self.errors.append(f"❌ {len(empty_dates)} Zeile(n) mit fehlendem Datum:")
|
||||
for idx in empty_dates.index[:5]: # Zeige max 5
|
||||
self.errors.append(f" Zeile {idx + 2}")
|
||||
if len(empty_dates) > 5:
|
||||
self.errors.append(f" ... und {len(empty_dates) - 5} weitere")
|
||||
valid = False
|
||||
|
||||
# Konvertiere Datum
|
||||
try:
|
||||
self.df['datum'] = pd.to_datetime(self.df['datum'], format='%Y-%m-%d', errors='coerce')
|
||||
except Exception as e:
|
||||
self.errors.append(f"❌ Fehler beim Parsen der Datumsangaben: {e}")
|
||||
return False
|
||||
|
||||
# Prüfe auf ungültige Datumsangaben
|
||||
invalid_dates = self.df[self.df['datum'].isna()]
|
||||
if len(invalid_dates) > 0:
|
||||
self.errors.append(f"❌ {len(invalid_dates)} ungültige Datumsangaben")
|
||||
valid = False
|
||||
|
||||
# Prüfe auf Zukunftsdaten
|
||||
today = datetime.now()
|
||||
future_dates = self.df[self.df['datum'] > today]
|
||||
if len(future_dates) > 0:
|
||||
self.errors.append(f"❌ {len(future_dates)} Datum/Daten in der Zukunft:")
|
||||
for idx, row in future_dates.iterrows():
|
||||
self.errors.append(f" Zeile {idx + 2}: {row['datum'].strftime('%Y-%m-%d')}")
|
||||
valid = False
|
||||
|
||||
# Prüfe Wochentage
|
||||
if self.lottery_type == 'lotto':
|
||||
# Historisch:
|
||||
# 1955-1965: Sonntags (und manchmal Montags bei Feiertagen)
|
||||
# 1965-2000: Samstags
|
||||
# Ab 2000: Mittwoch + Samstag
|
||||
weekday_names = {0: 'Mo', 1: 'Di', 2: 'Mi', 3: 'Do', 4: 'Fr', 5: 'Sa', 6: 'So'}
|
||||
else: # eurojackpot
|
||||
# Eurojackpot: Dienstag + Freitag
|
||||
weekday_names = {0: 'Mo', 1: 'Di', 2: 'Mi', 3: 'Do', 4: 'Fr', 5: 'Sa', 6: 'So'}
|
||||
|
||||
wrong_weekdays = []
|
||||
for idx, row in self.df.iterrows():
|
||||
if pd.notna(row['datum']):
|
||||
weekday = row['datum'].weekday()
|
||||
date = row['datum']
|
||||
|
||||
if self.lottery_type == 'lotto':
|
||||
# Lotto: Historische Regeln
|
||||
if date < datetime(1965, 9, 1):
|
||||
# Bis Aug 1965: Sonntag (und Montag bei Feiertagen)
|
||||
if weekday not in [0, 6]: # Mo, So
|
||||
wrong_weekdays.append((idx, date))
|
||||
elif date < datetime(2000, 12, 1):
|
||||
# Sep 1965 - Nov 2000: Nur Samstag
|
||||
if weekday != 5: # Sa
|
||||
wrong_weekdays.append((idx, date))
|
||||
else:
|
||||
# Ab Dez 2000: Mittwoch + Samstag
|
||||
if weekday not in [2, 5]: # Mi, Sa
|
||||
wrong_weekdays.append((idx, date))
|
||||
else:
|
||||
# Eurojackpot: Dienstag + Freitag
|
||||
if weekday not in [1, 4]: # Di, Fr
|
||||
wrong_weekdays.append((idx, date))
|
||||
|
||||
if len(wrong_weekdays) > 0:
|
||||
# Filtere mögliche Feiertags-Sonderziehungen (< 5 pro Zeitraum = OK)
|
||||
if len(wrong_weekdays) <= 5:
|
||||
self.warnings.append(f"⚠️ {len(wrong_weekdays)} Ziehung(en) am ungewöhnlichen Wochentag (möglicherweise Feiertage):")
|
||||
for idx, date in wrong_weekdays:
|
||||
weekday_name = weekday_names[date.weekday()]
|
||||
self.warnings.append(f" Zeile {idx + 2}: {date.strftime('%Y-%m-%d')} ({weekday_name})")
|
||||
else:
|
||||
self.errors.append(f"❌ {len(wrong_weekdays)} Ziehung(en) am falschen Wochentag:")
|
||||
for idx, date in wrong_weekdays[:5]:
|
||||
weekday_name = weekday_names[date.weekday()]
|
||||
self.errors.append(f" Zeile {idx + 2}: {date.strftime('%Y-%m-%d')} ({weekday_name})")
|
||||
if len(wrong_weekdays) > 5:
|
||||
self.errors.append(f" ... und {len(wrong_weekdays) - 5} weitere")
|
||||
valid = False
|
||||
|
||||
return valid
|
||||
|
||||
def validate_numbers(self) -> bool:
|
||||
"""Validiert Zahlenwerte."""
|
||||
valid = True
|
||||
|
||||
if self.lottery_type == 'lotto':
|
||||
main_cols = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6']
|
||||
main_range = (1, 49)
|
||||
sz_cols = ['SZ']
|
||||
sz_range = (0, 9)
|
||||
else: # eurojackpot
|
||||
main_cols = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5']
|
||||
main_range = (1, 50)
|
||||
sz_cols = ['SZ1', 'SZ2']
|
||||
sz_range = (1, 12)
|
||||
|
||||
# Prüfe Hauptzahlen
|
||||
for col in main_cols:
|
||||
# Fehlende Werte
|
||||
missing = self.df[self.df[col].isna()]
|
||||
if len(missing) > 0:
|
||||
self.errors.append(f"❌ {len(missing)} fehlende Werte in Spalte {col}")
|
||||
valid = False
|
||||
|
||||
# Zahlenbereich
|
||||
out_of_range = self.df[
|
||||
(self.df[col] < main_range[0]) |
|
||||
(self.df[col] > main_range[1])
|
||||
]
|
||||
if len(out_of_range) > 0:
|
||||
self.errors.append(f"❌ {len(out_of_range)} Werte außerhalb {main_range} in {col}")
|
||||
for idx, row in out_of_range.head(3).iterrows():
|
||||
self.errors.append(f" Zeile {idx + 2}: {row[col]}")
|
||||
valid = False
|
||||
|
||||
# Prüfe Superzahl(en)
|
||||
for col in sz_cols:
|
||||
missing = self.df[self.df[col].isna()]
|
||||
if len(missing) > 0:
|
||||
self.warnings.append(f"⚠️ {len(missing)} fehlende Werte in Spalte {col}")
|
||||
|
||||
out_of_range = self.df[
|
||||
(self.df[col] < sz_range[0]) |
|
||||
(self.df[col] > sz_range[1])
|
||||
]
|
||||
if len(out_of_range) > 0:
|
||||
self.errors.append(f"❌ {len(out_of_range)} Werte außerhalb {sz_range} in {col}")
|
||||
valid = False
|
||||
|
||||
# Prüfe Eindeutigkeit der Hauptzahlen pro Zeile
|
||||
duplicate_numbers = []
|
||||
for idx, row in self.df.iterrows():
|
||||
main_numbers = [row[col] for col in main_cols if pd.notna(row[col])]
|
||||
if len(main_numbers) != len(set(main_numbers)):
|
||||
duplicate_numbers.append((idx, main_numbers))
|
||||
|
||||
if len(duplicate_numbers) > 0:
|
||||
self.errors.append(f"❌ {len(duplicate_numbers)} Zeile(n) mit doppelten Hauptzahlen:")
|
||||
for idx, numbers in duplicate_numbers[:5]:
|
||||
self.errors.append(f" Zeile {idx + 2}: {numbers}")
|
||||
if len(duplicate_numbers) > 5:
|
||||
self.errors.append(f" ... und {len(duplicate_numbers) - 5} weitere")
|
||||
valid = False
|
||||
|
||||
# Prüfe Sortierung der Hauptzahlen pro Zeile
|
||||
unsorted_rows = []
|
||||
for idx, row in self.df.iterrows():
|
||||
main_numbers = [row[col] for col in main_cols if pd.notna(row[col])]
|
||||
if main_numbers != sorted(main_numbers):
|
||||
unsorted_rows.append((idx, main_numbers))
|
||||
|
||||
if len(unsorted_rows) > 0:
|
||||
self.warnings.append(f"⚠️ {len(unsorted_rows)} Zeile(n) mit unsortierten Zahlen:")
|
||||
for idx, numbers in unsorted_rows[:3]:
|
||||
self.warnings.append(f" Zeile {idx + 2}: {numbers}")
|
||||
|
||||
# Prüfe Sortierung der Eurozahlen (nur Eurojackpot)
|
||||
if self.lottery_type == 'eurojackpot':
|
||||
unsorted_euro = []
|
||||
for idx, row in self.df.iterrows():
|
||||
if pd.notna(row['SZ1']) and pd.notna(row['SZ2']):
|
||||
if row['SZ1'] > row['SZ2']:
|
||||
unsorted_euro.append((idx, row['SZ1'], row['SZ2']))
|
||||
|
||||
if len(unsorted_euro) > 0:
|
||||
self.warnings.append(f"⚠️ {len(unsorted_euro)} Zeile(n) mit unsortierten Eurozahlen")
|
||||
|
||||
return valid
|
||||
|
||||
def validate_duplicates(self) -> bool:
|
||||
"""Prüft auf doppelte Datumssätze."""
|
||||
duplicates = self.df[self.df.duplicated(subset=['datum'], keep=False)]
|
||||
|
||||
if len(duplicates) > 0:
|
||||
self.errors.append(f"❌ {len(duplicates)} doppelte Datumseinträge gefunden:")
|
||||
for idx, row in duplicates.head(5).iterrows():
|
||||
self.errors.append(f" Zeile {idx + 2}: {row['datum'].strftime('%Y-%m-%d')}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def validate_chronology(self) -> bool:
|
||||
"""Prüft chronologische Sortierung."""
|
||||
if len(self.df) < 2:
|
||||
return True
|
||||
|
||||
unsorted = []
|
||||
for i in range(len(self.df) - 1):
|
||||
if self.df.iloc[i]['datum'] > self.df.iloc[i + 1]['datum']:
|
||||
unsorted.append((i, self.df.iloc[i]['datum'], self.df.iloc[i + 1]['datum']))
|
||||
|
||||
if len(unsorted) > 0:
|
||||
self.warnings.append(f"⚠️ {len(unsorted)} Stelle(n) mit falscher chronologischer Reihenfolge:")
|
||||
for idx, date1, date2 in unsorted[:3]:
|
||||
self.warnings.append(f" Zeile {idx + 2}: {date1.strftime('%Y-%m-%d')} > {date2.strftime('%Y-%m-%d')}")
|
||||
|
||||
return len(unsorted) == 0
|
||||
|
||||
def validate_completeness(self) -> bool:
|
||||
"""Prüft auf Lücken in den Ziehungen."""
|
||||
if len(self.df) < 2:
|
||||
return True
|
||||
|
||||
if self.lottery_type == 'lotto':
|
||||
# Lotto: 2x pro Woche (Mi, Sa)
|
||||
expected_days = 3.5 # Durchschnitt
|
||||
else:
|
||||
# Eurojackpot: 2x pro Woche (Di, Fr)
|
||||
expected_days = 3.5
|
||||
|
||||
gaps = []
|
||||
for i in range(len(self.df) - 1):
|
||||
date1 = self.df.iloc[i]['datum']
|
||||
date2 = self.df.iloc[i + 1]['datum']
|
||||
diff = (date2 - date1).days
|
||||
|
||||
# Wenn Lücke größer als 7 Tage, könnte Ziehung fehlen
|
||||
if diff > 7:
|
||||
gaps.append((i, date1, date2, diff))
|
||||
|
||||
if len(gaps) > 0:
|
||||
self.warnings.append(f"⚠️ {len(gaps)} mögliche Lücke(n) in den Ziehungen (>7 Tage):")
|
||||
for idx, date1, date2, diff in gaps[:5]:
|
||||
self.warnings.append(f" Zeile {idx + 2}: {date1.strftime('%Y-%m-%d')} → {date2.strftime('%Y-%m-%d')} ({diff} Tage)")
|
||||
if len(gaps) > 5:
|
||||
self.warnings.append(f" ... und {len(gaps) - 5} weitere")
|
||||
|
||||
return True
|
||||
|
||||
def get_statistics(self) -> Dict:
|
||||
"""Erstellt Statistiken."""
|
||||
if self.df is None or len(self.df) == 0:
|
||||
return {}
|
||||
|
||||
stats = {
|
||||
'total_draws': len(self.df),
|
||||
'date_range': (
|
||||
self.df['datum'].min().strftime('%Y-%m-%d'),
|
||||
self.df['datum'].max().strftime('%Y-%m-%d')
|
||||
),
|
||||
'years_covered': (self.df['datum'].max().year - self.df['datum'].min().year) + 1,
|
||||
}
|
||||
|
||||
return stats
|
||||
|
||||
def validate_all(self) -> bool:
|
||||
"""Führt alle Validierungen durch."""
|
||||
print(f"\n{'='*70}")
|
||||
print(f" CSV VALIDATOR")
|
||||
print(f" Typ: {self.lottery_type.upper()}")
|
||||
print(f"{'='*70}")
|
||||
print(f"\n📁 Datei: {os.path.basename(self.csv_file)}")
|
||||
|
||||
if not self.load_csv():
|
||||
return False
|
||||
|
||||
print(f"✅ Datei geladen: {len(self.df)} Zeilen")
|
||||
|
||||
# Alle Validierungen
|
||||
checks = [
|
||||
("Spaltenstruktur", self.validate_structure),
|
||||
("Datumsangaben", self.validate_dates),
|
||||
("Zahlenwerte", self.validate_numbers),
|
||||
("Duplikate", self.validate_duplicates),
|
||||
("Chronologie", self.validate_chronology),
|
||||
("Vollständigkeit", self.validate_completeness),
|
||||
]
|
||||
|
||||
print(f"\n🔍 VALIDIERUNG")
|
||||
print("="*70)
|
||||
|
||||
all_valid = True
|
||||
for check_name, check_func in checks:
|
||||
try:
|
||||
result = check_func()
|
||||
status = "✅" if result else "❌"
|
||||
print(f"{status} {check_name}")
|
||||
if not result:
|
||||
all_valid = False
|
||||
except Exception as e:
|
||||
print(f"❌ {check_name} - Fehler: {e}")
|
||||
all_valid = False
|
||||
|
||||
# Statistiken
|
||||
stats = self.get_statistics()
|
||||
if stats:
|
||||
print(f"\n📊 STATISTIKEN")
|
||||
print("="*70)
|
||||
print(f"Ziehungen gesamt: {stats['total_draws']}")
|
||||
print(f"Zeitraum: {stats['date_range'][0]} bis {stats['date_range'][1]}")
|
||||
print(f"Jahre: {stats['years_covered']}")
|
||||
|
||||
# Fehler ausgeben
|
||||
if self.errors:
|
||||
print(f"\n❌ FEHLER ({len(self.errors)})")
|
||||
print("="*70)
|
||||
for error in self.errors:
|
||||
print(error)
|
||||
|
||||
# Warnungen ausgeben
|
||||
if self.warnings:
|
||||
print(f"\n⚠️ WARNUNGEN ({len(self.warnings)})")
|
||||
print("="*70)
|
||||
for warning in self.warnings:
|
||||
print(warning)
|
||||
|
||||
# Zusammenfassung
|
||||
print(f"\n{'='*70}")
|
||||
if all_valid and not self.errors:
|
||||
print("✅ VALIDIERUNG ERFOLGREICH - Keine Fehler gefunden!")
|
||||
elif not self.errors and self.warnings:
|
||||
print("✅ VALIDIERUNG OK - Nur Warnungen (keine kritischen Fehler)")
|
||||
else:
|
||||
print("❌ VALIDIERUNG FEHLGESCHLAGEN")
|
||||
print("="*70)
|
||||
|
||||
return all_valid and len(self.errors) == 0
|
||||
|
||||
|
||||
def main():
|
||||
"""Hauptfunktion."""
|
||||
import sys
|
||||
|
||||
# Standard-Dateien
|
||||
default_files = {
|
||||
'lotto': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data/AlleLottozahlen.csv',
|
||||
'eurojackpot': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/AlleEurojackpotzahlen.csv'
|
||||
}
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
# Benutzerdefinierte Datei
|
||||
csv_file = sys.argv[1]
|
||||
validator = LottoCSVValidator(csv_file)
|
||||
success = validator.validate_all()
|
||||
sys.exit(0 if success else 1)
|
||||
else:
|
||||
# Validiere beide Standard-Dateien
|
||||
print("\n🎲 Validiere beide Lottery-Dateien...\n")
|
||||
|
||||
results = {}
|
||||
for lottery_type, csv_file in default_files.items():
|
||||
if os.path.exists(csv_file):
|
||||
validator = LottoCSVValidator(csv_file)
|
||||
results[lottery_type] = validator.validate_all()
|
||||
else:
|
||||
print(f"\n⚠️ {lottery_type.upper()}: Datei nicht gefunden: {csv_file}")
|
||||
results[lottery_type] = False
|
||||
|
||||
# Gesamtergebnis
|
||||
print("\n" + "="*70)
|
||||
print(" GESAMTERGEBNIS")
|
||||
print("="*70)
|
||||
for lottery_type, success in results.items():
|
||||
status = "✅" if success else "❌"
|
||||
print(f"{status} {lottery_type.upper()}")
|
||||
print("="*70)
|
||||
|
||||
all_success = all(results.values())
|
||||
sys.exit(0 if all_success else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Executable
+401
@@ -0,0 +1,401 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Ziehungs-Verifizierer für Lotto 6aus49 und Eurojackpot
|
||||
|
||||
Verifiziert Ziehungen gegen offizielle Datenquellen:
|
||||
1. GitHub Lotto Archive (für Lotto 6aus49)
|
||||
2. Eurojackpot-zahlen.eu (für Eurojackpot)
|
||||
|
||||
Prüft:
|
||||
- Vollständigkeit (fehlende Ziehungen)
|
||||
- Korrektheit (falsche Zahlen)
|
||||
- Duplikate
|
||||
"""
|
||||
|
||||
import pandas as pd
|
||||
import requests
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Tuple, Optional
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
class DrawVerifier:
|
||||
"""Verifiziert Ziehungen gegen offizielle Quellen."""
|
||||
|
||||
def __init__(self, csv_file: str, lottery_type: str):
|
||||
self.csv_file = csv_file
|
||||
self.lottery_type = lottery_type
|
||||
self.df_local = None
|
||||
self.df_official = None
|
||||
self.errors = []
|
||||
self.warnings = []
|
||||
self.info = []
|
||||
|
||||
def load_local_data(self) -> bool:
|
||||
"""Lädt lokale CSV-Datei."""
|
||||
try:
|
||||
if not os.path.exists(self.csv_file):
|
||||
self.errors.append(f"❌ Datei existiert nicht: {self.csv_file}")
|
||||
return False
|
||||
|
||||
self.df_local = pd.read_csv(self.csv_file, sep=';')
|
||||
self.df_local['datum'] = pd.to_datetime(
|
||||
self.df_local['datum'],
|
||||
format='%Y-%m-%d',
|
||||
errors='coerce'
|
||||
)
|
||||
|
||||
self.info.append(f"✅ Lokale Daten: {len(self.df_local)} Ziehungen")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.errors.append(f"❌ Fehler beim Laden: {e}")
|
||||
return False
|
||||
|
||||
def fetch_official_lotto_data(self) -> bool:
|
||||
"""Holt offizielle Lotto 6aus49 Daten von GitHub Archive."""
|
||||
try:
|
||||
url = 'https://johannesfriedrich.github.io/LottoNumberArchive/Lottonumbers_tidy_complete.json'
|
||||
|
||||
print("🌐 Lade offizielle Lotto-Daten von GitHub Archive...")
|
||||
|
||||
headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
|
||||
}
|
||||
|
||||
response = requests.get(url, headers=headers, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
# Parse gruppierte Daten
|
||||
draws_by_id = {}
|
||||
for entry in data:
|
||||
draw_id = entry.get('id')
|
||||
if draw_id not in draws_by_id:
|
||||
draws_by_id[draw_id] = {
|
||||
'date': entry.get('date'),
|
||||
'numbers': [],
|
||||
'superzahl': None
|
||||
}
|
||||
|
||||
variable = entry.get('variable')
|
||||
value = entry.get('value')
|
||||
|
||||
if variable == 'Lottozahl':
|
||||
draws_by_id[draw_id]['numbers'].append(value)
|
||||
elif variable == 'Superzahl':
|
||||
draws_by_id[draw_id]['superzahl'] = value
|
||||
|
||||
# Konvertiere zu DataFrame
|
||||
official_draws = []
|
||||
for draw_id, draw_data in draws_by_id.items():
|
||||
try:
|
||||
date_str = draw_data['date']
|
||||
day, month, year = date_str.split('.')
|
||||
date_obj = datetime(int(year), int(month), int(day))
|
||||
|
||||
numbers = sorted(draw_data['numbers'])
|
||||
|
||||
if len(numbers) == 6:
|
||||
official_draws.append({
|
||||
'datum': date_obj,
|
||||
'Z1': numbers[0],
|
||||
'Z2': numbers[1],
|
||||
'Z3': numbers[2],
|
||||
'Z4': numbers[3],
|
||||
'Z5': numbers[4],
|
||||
'Z6': numbers[5],
|
||||
'SZ': draw_data['superzahl']
|
||||
})
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
self.df_official = pd.DataFrame(official_draws)
|
||||
self.info.append(f"✅ Offizielle Daten: {len(self.df_official)} Ziehungen")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.errors.append(f"❌ Fehler beim Laden offizieller Daten: {e}")
|
||||
return False
|
||||
|
||||
def fetch_official_eurojackpot_data(self) -> bool:
|
||||
"""
|
||||
Holt offizielle Eurojackpot Daten.
|
||||
|
||||
Note: Da es keine vollständige öffentliche API gibt,
|
||||
beschränken wir uns auf Plausibilitätsprüfungen.
|
||||
"""
|
||||
self.warnings.append("⚠️ Keine vollständige offizielle Eurojackpot-API verfügbar")
|
||||
self.warnings.append(" Nur Plausibilitätsprüfungen möglich")
|
||||
|
||||
# Erstelle minimale "offizielle" Daten basierend auf erwarteten Ziehungsterminen
|
||||
# Dies ist nur für Vollständigkeitsprüfung
|
||||
|
||||
if len(self.df_local) > 0:
|
||||
start_date = self.df_local['datum'].min()
|
||||
end_date = datetime.now()
|
||||
|
||||
expected_dates = []
|
||||
current = start_date
|
||||
|
||||
while current <= end_date:
|
||||
# Eurojackpot: Dienstag (1) und Freitag (4)
|
||||
if current.weekday() in [1, 4]:
|
||||
expected_dates.append(current)
|
||||
current += timedelta(days=1)
|
||||
|
||||
self.df_official = pd.DataFrame({'datum': expected_dates})
|
||||
self.info.append(f"ℹ️ Erwartete Ziehungstermine: {len(expected_dates)}")
|
||||
|
||||
return True
|
||||
|
||||
def compare_completeness(self) -> List[datetime]:
|
||||
"""Prüft auf fehlende Ziehungen."""
|
||||
if self.df_official is None or self.df_local is None:
|
||||
return []
|
||||
|
||||
official_dates = set(self.df_official['datum'].dt.date)
|
||||
local_dates = set(self.df_local['datum'].dt.date)
|
||||
|
||||
missing = official_dates - local_dates
|
||||
extra = local_dates - official_dates
|
||||
|
||||
if missing:
|
||||
self.errors.append(f"❌ {len(missing)} fehlende Ziehung(en):")
|
||||
for date in sorted(missing)[:10]:
|
||||
self.errors.append(f" {date.strftime('%Y-%m-%d')}")
|
||||
if len(missing) > 10:
|
||||
self.errors.append(f" ... und {len(missing) - 10} weitere")
|
||||
|
||||
if extra:
|
||||
self.warnings.append(f"⚠️ {len(extra)} zusätzliche Ziehung(en) (nicht in offiziellen Daten):")
|
||||
for date in sorted(extra)[:5]:
|
||||
self.warnings.append(f" {date.strftime('%Y-%m-%d')}")
|
||||
if len(extra) > 5:
|
||||
self.warnings.append(f" ... und {len(extra) - 5} weitere")
|
||||
|
||||
return list(missing)
|
||||
|
||||
def compare_accuracy(self) -> int:
|
||||
"""Vergleicht Zahlenwerte mit offiziellen Daten."""
|
||||
if self.df_official is None or self.df_local is None:
|
||||
return 0
|
||||
|
||||
if self.lottery_type == 'eurojackpot':
|
||||
# Keine detaillierten offiziellen Daten verfügbar
|
||||
return 0
|
||||
|
||||
mismatches = 0
|
||||
|
||||
# Merge auf Datum
|
||||
merged = pd.merge(
|
||||
self.df_local,
|
||||
self.df_official,
|
||||
on='datum',
|
||||
suffixes=('_local', '_official'),
|
||||
how='inner'
|
||||
)
|
||||
|
||||
if self.lottery_type == 'lotto':
|
||||
cols_to_check = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
||||
else: # eurojackpot
|
||||
cols_to_check = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']
|
||||
|
||||
for idx, row in merged.iterrows():
|
||||
mismatch_cols = []
|
||||
|
||||
for col in cols_to_check:
|
||||
local_col = f"{col}_local"
|
||||
official_col = f"{col}_official"
|
||||
|
||||
if local_col in row and official_col in row:
|
||||
# Vergleiche nur wenn beide Werte vorhanden
|
||||
if pd.notna(row[local_col]) and pd.notna(row[official_col]):
|
||||
if row[local_col] != row[official_col]:
|
||||
mismatch_cols.append(
|
||||
f"{col}: {int(row[local_col])} ≠ {int(row[official_col])}"
|
||||
)
|
||||
|
||||
if mismatch_cols:
|
||||
mismatches += 1
|
||||
if mismatches == 1:
|
||||
self.errors.append("❌ Zahlen-Abweichungen gefunden:")
|
||||
|
||||
if mismatches <= 10:
|
||||
date_str = row['datum'].strftime('%Y-%m-%d')
|
||||
self.errors.append(f" {date_str}: {', '.join(mismatch_cols)}")
|
||||
|
||||
if mismatches > 10:
|
||||
self.errors.append(f" ... und {mismatches - 10} weitere Abweichungen")
|
||||
|
||||
return mismatches
|
||||
|
||||
def check_recent_draws(self, days: int = 30) -> None:
|
||||
"""Prüft besonders die letzten N Tage."""
|
||||
if self.df_local is None:
|
||||
return
|
||||
|
||||
cutoff = datetime.now() - timedelta(days=days)
|
||||
recent = self.df_local[self.df_local['datum'] >= cutoff]
|
||||
|
||||
self.info.append(f"ℹ️ Letzte {days} Tage: {len(recent)} Ziehungen")
|
||||
|
||||
if len(recent) == 0:
|
||||
self.warnings.append(f"⚠️ Keine Ziehungen in den letzten {days} Tagen!")
|
||||
|
||||
# Erwartete Anzahl berechnen
|
||||
if self.lottery_type == 'lotto':
|
||||
# 2x pro Woche
|
||||
expected = (days / 7) * 2
|
||||
else: # eurojackpot
|
||||
# 2x pro Woche
|
||||
expected = (days / 7) * 2
|
||||
|
||||
if len(recent) < expected * 0.8: # Toleranz 20%
|
||||
self.warnings.append(
|
||||
f"⚠️ Weniger Ziehungen als erwartet: {len(recent)} vs. ~{int(expected)}"
|
||||
)
|
||||
|
||||
def verify_all(self) -> bool:
|
||||
"""Führt komplette Verifikation durch."""
|
||||
print(f"\n{'='*70}")
|
||||
print(f" ZIEHUNGS-VERIFIZIERER")
|
||||
print(f" Typ: {self.lottery_type.upper()}")
|
||||
print(f"{'='*70}")
|
||||
print(f"\n📁 Datei: {os.path.basename(self.csv_file)}")
|
||||
|
||||
# Lade lokale Daten
|
||||
if not self.load_local_data():
|
||||
return False
|
||||
|
||||
# Lade offizielle Daten
|
||||
print()
|
||||
if self.lottery_type == 'lotto':
|
||||
if not self.fetch_official_lotto_data():
|
||||
return False
|
||||
else: # eurojackpot
|
||||
if not self.fetch_official_eurojackpot_data():
|
||||
return False
|
||||
|
||||
print(f"\n🔍 VERIFIKATION")
|
||||
print("="*70)
|
||||
|
||||
# Prüfungen
|
||||
print("Prüfe Vollständigkeit...")
|
||||
missing = self.compare_completeness()
|
||||
|
||||
if self.lottery_type == 'lotto':
|
||||
print("Prüfe Zahlenwerte...")
|
||||
mismatches = self.compare_accuracy()
|
||||
|
||||
print("Prüfe aktuelle Ziehungen...")
|
||||
self.check_recent_draws(30)
|
||||
|
||||
# Statistiken
|
||||
print(f"\n📊 STATISTIKEN")
|
||||
print("="*70)
|
||||
for info in self.info:
|
||||
print(info)
|
||||
|
||||
# Zusammenfassung
|
||||
if self.df_local is not None and self.df_official is not None:
|
||||
if self.lottery_type == 'lotto':
|
||||
overlap = len(pd.merge(
|
||||
self.df_local,
|
||||
self.df_official,
|
||||
on='datum',
|
||||
how='inner'
|
||||
))
|
||||
|
||||
if overlap > 0:
|
||||
print(f"\n✅ {overlap} Ziehungen in beiden Quellen")
|
||||
|
||||
# Genauigkeit
|
||||
if self.lottery_type == 'lotto':
|
||||
accuracy = ((overlap - (mismatches if 'mismatches' in locals() else 0)) / overlap * 100)
|
||||
print(f"✅ Genauigkeit: {accuracy:.1f}%")
|
||||
|
||||
# Fehler
|
||||
if self.errors:
|
||||
print(f"\n❌ FEHLER ({len(self.errors)})")
|
||||
print("="*70)
|
||||
for error in self.errors:
|
||||
print(error)
|
||||
|
||||
# Warnungen
|
||||
if self.warnings:
|
||||
print(f"\n⚠️ WARNUNGEN ({len(self.warnings)})")
|
||||
print("="*70)
|
||||
for warning in self.warnings:
|
||||
print(warning)
|
||||
|
||||
# Ergebnis
|
||||
print(f"\n{'='*70}")
|
||||
if not self.errors:
|
||||
if self.warnings:
|
||||
print("✅ VERIFIKATION OK - Nur Warnungen")
|
||||
else:
|
||||
print("✅ VERIFIKATION ERFOLGREICH - Alle Ziehungen korrekt!")
|
||||
else:
|
||||
print("❌ VERIFIKATION FEHLGESCHLAGEN - Fehler gefunden")
|
||||
print("="*70)
|
||||
|
||||
return len(self.errors) == 0
|
||||
|
||||
|
||||
def main():
|
||||
"""Hauptfunktion."""
|
||||
|
||||
# Standard-Dateien
|
||||
files = {
|
||||
'lotto': {
|
||||
'path': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data/AlleLottozahlen.csv',
|
||||
'type': 'lotto'
|
||||
},
|
||||
'eurojackpot': {
|
||||
'path': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/AlleEurojackpotzahlen.csv',
|
||||
'type': 'eurojackpot'
|
||||
}
|
||||
}
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
# Einzelne Datei
|
||||
csv_file = sys.argv[1]
|
||||
lottery_type = sys.argv[2] if len(sys.argv) > 2 else 'lotto'
|
||||
|
||||
verifier = DrawVerifier(csv_file, lottery_type)
|
||||
success = verifier.verify_all()
|
||||
sys.exit(0 if success else 1)
|
||||
|
||||
else:
|
||||
# Beide Dateien
|
||||
print("\n🎲 Verifiziere beide Lottery-Dateien...\n")
|
||||
|
||||
results = {}
|
||||
|
||||
for name, config in files.items():
|
||||
if os.path.exists(config['path']):
|
||||
verifier = DrawVerifier(config['path'], config['type'])
|
||||
results[name] = verifier.verify_all()
|
||||
else:
|
||||
print(f"\n⚠️ {name.upper()}: Datei nicht gefunden")
|
||||
results[name] = False
|
||||
|
||||
# Gesamtergebnis
|
||||
print("\n" + "="*70)
|
||||
print(" GESAMTERGEBNIS")
|
||||
print("="*70)
|
||||
for name, success in results.items():
|
||||
status = "✅" if success else "❌"
|
||||
print(f"{status} {name.upper()}")
|
||||
print("="*70)
|
||||
|
||||
all_success = all(results.values())
|
||||
sys.exit(0 if all_success else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user