cbazzaandClaude Sonnet 4.5
428403557e
Major improvements: - Deep Learning integration with PyTorch LSTM (Bidirectional, 128→64 units) - Hybrid predictor: 40% RandomForest + 60% Deep Learning - LaunchAgent for automatic weekly tip generation (Tue/Fri 21:00) - Health-Check system with auto-recovery and Telegram alerts - Model caching and intelligent retraining logic - Updated CSV data and generated tips - Performance reports for recent draws Technical details: - PyTorch used instead of TensorFlow (Python 3.14 compatibility) - Apple Silicon MPS acceleration support - Sequence learning with 20-draw history - Early stopping and learning rate scheduling 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
346 lines
11 KiB
Python
346 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Health-Check System mit Auto-Recovery
|
|
======================================
|
|
|
|
Prüft System-Gesundheit und behebt automatisch Probleme:
|
|
- CSV-Datei vorhanden und aktuell?
|
|
- Models trainiert und verfügbar?
|
|
- Learning State konsistent?
|
|
- Logs rotieren?
|
|
- Telegram-Bot erreichbar?
|
|
|
|
Bei Problemen:
|
|
- Auto-Retry
|
|
- Telegram-Alerts
|
|
- Logging
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Dict, List, Tuple
|
|
import pandas as pd
|
|
|
|
# Add parent dir to path
|
|
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.notifier import LottoNotifier
|
|
|
|
|
|
class HealthCheck:
|
|
"""System Health-Check mit Auto-Recovery."""
|
|
|
|
def __init__(self, data_dir: str, lottery_name: str = "Lotto"):
|
|
self.data_dir = data_dir
|
|
self.lottery_name = lottery_name
|
|
self.notifier = LottoNotifier()
|
|
|
|
self.health_log = os.path.join(data_dir, "health_check.json")
|
|
self.issues = []
|
|
self.warnings = []
|
|
self.recovered = []
|
|
|
|
print(f"🏥 HEALTH-CHECK SYSTEM - {lottery_name}")
|
|
print("=" * 70)
|
|
|
|
def run_all_checks(self) -> bool:
|
|
"""Führt alle Health-Checks aus."""
|
|
all_ok = True
|
|
|
|
checks = [
|
|
("CSV Data", self.check_csv_data),
|
|
("ML Models", self.check_ml_models),
|
|
("Learning State", self.check_learning_state),
|
|
("Logs", self.check_logs),
|
|
("Telegram", self.check_telegram),
|
|
("Disk Space", self.check_disk_space)
|
|
]
|
|
|
|
for name, check_func in checks:
|
|
print(f"\n🔍 Checking: {name}...", end=" ", flush=True)
|
|
try:
|
|
status, message = check_func()
|
|
if status == "OK":
|
|
print(f"✅ {message}")
|
|
elif status == "WARNING":
|
|
print(f"⚠️ {message}")
|
|
self.warnings.append(f"{name}: {message}")
|
|
elif status == "ERROR":
|
|
print(f"❌ {message}")
|
|
self.issues.append(f"{name}: {message}")
|
|
all_ok = False
|
|
elif status == "RECOVERED":
|
|
print(f"🔧 {message}")
|
|
self.recovered.append(f"{name}: {message}")
|
|
except Exception as e:
|
|
print(f"❌ Exception: {e}")
|
|
self.issues.append(f"{name}: Exception - {e}")
|
|
all_ok = False
|
|
|
|
# Summary
|
|
print("\n" + "=" * 70)
|
|
self._print_summary()
|
|
|
|
# Save health log
|
|
self._save_health_log(all_ok)
|
|
|
|
# Send alert if issues
|
|
if self.issues:
|
|
self._send_alert()
|
|
|
|
return all_ok
|
|
|
|
def check_csv_data(self) -> Tuple[str, str]:
|
|
"""Prüft CSV-Datei."""
|
|
csv_file = os.path.join(self.data_dir, "AlleLottozahlen.csv")
|
|
|
|
if not os.path.exists(csv_file):
|
|
return ("ERROR", f"CSV file not found: {csv_file}")
|
|
|
|
# Check age
|
|
mtime = os.path.getmtime(csv_file)
|
|
age_days = (time.time() - mtime) / 86400
|
|
|
|
if age_days > 10:
|
|
return ("WARNING", f"CSV file is {age_days:.1f} days old")
|
|
|
|
# Check content
|
|
try:
|
|
df = pd.read_csv(csv_file, sep=';')
|
|
if len(df) < 100:
|
|
return ("ERROR", f"CSV has only {len(df)} rows")
|
|
|
|
return ("OK", f"{len(df):,} draws, {age_days:.1f} days old")
|
|
except Exception as e:
|
|
return ("ERROR", f"CSV parse error: {e}")
|
|
|
|
def check_ml_models(self) -> Tuple[str, str]:
|
|
"""Prüft ML Models."""
|
|
models_dir = os.path.join(self.data_dir, "ultimate_ml_models")
|
|
|
|
if not os.path.exists(models_dir):
|
|
return ("WARNING", "No models cache found (will train on next run)")
|
|
|
|
# Check RandomForest models
|
|
rf_model = os.path.join(models_dir, "trained_models.pkl")
|
|
if os.path.exists(rf_model):
|
|
age_days = (time.time() - os.path.getmtime(rf_model)) / 86400
|
|
status = "OK" if age_days < 10 else "WARNING"
|
|
msg = f"RandomForest models {age_days:.1f} days old"
|
|
else:
|
|
status = "WARNING"
|
|
msg = "RandomForest models not found"
|
|
|
|
# Check Deep Learning models
|
|
dl_model = os.path.join(models_dir, "deep_learning", "lstm_model_49.pth")
|
|
if os.path.exists(dl_model):
|
|
age_days = (time.time() - os.path.getmtime(dl_model)) / 86400
|
|
msg += f", LSTM {age_days:.1f} days old"
|
|
else:
|
|
msg += ", LSTM not found"
|
|
|
|
return (status, msg)
|
|
|
|
def check_learning_state(self) -> Tuple[str, str]:
|
|
"""Prüft Learning State."""
|
|
state_file = os.path.join(self.data_dir, "learning_state.json")
|
|
|
|
if not os.path.exists(state_file):
|
|
return ("WARNING", "No learning state found")
|
|
|
|
try:
|
|
with open(state_file, 'r') as f:
|
|
state = json.load(f)
|
|
|
|
cycles = state.get('learning_cycle', 0)
|
|
last_update = state.get('last_update', '')
|
|
|
|
if not last_update:
|
|
return ("WARNING", f"{cycles} cycles, no last_update timestamp")
|
|
|
|
last_dt = datetime.fromisoformat(last_update)
|
|
age_days = (datetime.now() - last_dt).days
|
|
|
|
if age_days > 10:
|
|
return ("WARNING", f"{cycles} cycles, last update {age_days} days ago")
|
|
|
|
return ("OK", f"{cycles} cycles, last update {age_days} days ago")
|
|
|
|
except Exception as e:
|
|
return ("ERROR", f"State parse error: {e}")
|
|
|
|
def check_logs(self) -> Tuple[str, str]:
|
|
"""Prüft und rotiert Logs."""
|
|
logs_dir = os.path.join(os.path.dirname(self.data_dir), "logs")
|
|
|
|
if not os.path.exists(logs_dir):
|
|
os.makedirs(logs_dir, exist_ok=True)
|
|
return ("RECOVERED", "Created logs directory")
|
|
|
|
# Check log sizes
|
|
total_size = 0
|
|
large_logs = []
|
|
|
|
for log_file in Path(logs_dir).glob("*.log"):
|
|
size_mb = log_file.stat().st_size / 1024 / 1024
|
|
total_size += size_mb
|
|
|
|
if size_mb > 50: # > 50 MB
|
|
large_logs.append(log_file.name)
|
|
|
|
# Rotate large logs
|
|
if large_logs:
|
|
for log_name in large_logs:
|
|
self._rotate_log(os.path.join(logs_dir, log_name))
|
|
|
|
return ("RECOVERED", f"Rotated {len(large_logs)} large logs, total {total_size:.1f} MB")
|
|
|
|
return ("OK", f"Total size {total_size:.1f} MB")
|
|
|
|
def check_telegram(self) -> Tuple[str, str]:
|
|
"""Prüft Telegram-Bot."""
|
|
config = self.notifier.config
|
|
|
|
if not config.get("telegram", {}).get("enabled"):
|
|
return ("WARNING", "Telegram disabled in config")
|
|
|
|
bot_token = config.get("telegram", {}).get("bot_token")
|
|
if not bot_token or bot_token == "YOUR_BOT_TOKEN":
|
|
return ("WARNING", "Telegram bot_token not configured")
|
|
|
|
# Simple check: Token format
|
|
if len(bot_token) < 20 or ':' not in bot_token:
|
|
return ("ERROR", "Invalid bot_token format")
|
|
|
|
return ("OK", "Telegram configured")
|
|
|
|
def check_disk_space(self) -> Tuple[str, str]:
|
|
"""Prüft Festplatten-Speicher."""
|
|
import shutil
|
|
|
|
usage = shutil.disk_usage(self.data_dir)
|
|
free_gb = usage.free / 1024 / 1024 / 1024
|
|
percent_free = (usage.free / usage.total) * 100
|
|
|
|
if percent_free < 10:
|
|
return ("ERROR", f"Only {free_gb:.1f} GB free ({percent_free:.1f}%)")
|
|
elif percent_free < 20:
|
|
return ("WARNING", f"{free_gb:.1f} GB free ({percent_free:.1f}%)")
|
|
|
|
return ("OK", f"{free_gb:.1f} GB free ({percent_free:.1f}%)")
|
|
|
|
def _rotate_log(self, log_path: str):
|
|
"""Rotiert ein Log-File."""
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
backup_path = f"{log_path}.{timestamp}"
|
|
|
|
os.rename(log_path, backup_path)
|
|
print(f" 📦 Rotated: {os.path.basename(log_path)} → {os.path.basename(backup_path)}")
|
|
|
|
def _print_summary(self):
|
|
"""Druckt Zusammenfassung."""
|
|
print("\n📊 SUMMARY:")
|
|
|
|
if not self.issues and not self.warnings and not self.recovered:
|
|
print(" ✅ All checks passed - System healthy!")
|
|
return
|
|
|
|
if self.recovered:
|
|
print(f"\n 🔧 Auto-Recovered ({len(self.recovered)}):")
|
|
for item in self.recovered:
|
|
print(f" • {item}")
|
|
|
|
if self.warnings:
|
|
print(f"\n ⚠️ Warnings ({len(self.warnings)}):")
|
|
for item in self.warnings:
|
|
print(f" • {item}")
|
|
|
|
if self.issues:
|
|
print(f"\n ❌ Issues ({len(self.issues)}):")
|
|
for item in self.issues:
|
|
print(f" • {item}")
|
|
|
|
def _save_health_log(self, all_ok: bool):
|
|
"""Speichert Health-Log."""
|
|
log_entry = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"status": "OK" if all_ok else "ISSUES",
|
|
"issues": self.issues,
|
|
"warnings": self.warnings,
|
|
"recovered": self.recovered
|
|
}
|
|
|
|
# Load existing log
|
|
if os.path.exists(self.health_log):
|
|
with open(self.health_log, 'r') as f:
|
|
log_data = json.load(f)
|
|
else:
|
|
log_data = {"checks": []}
|
|
|
|
# Append new entry
|
|
log_data["checks"].append(log_entry)
|
|
|
|
# Keep only last 100 entries
|
|
log_data["checks"] = log_data["checks"][-100:]
|
|
|
|
# Save
|
|
with open(self.health_log, 'w') as f:
|
|
json.dump(log_data, f, indent=2)
|
|
|
|
def _send_alert(self):
|
|
"""Sendet Telegram-Alert bei Problemen."""
|
|
message = f"🚨 *HEALTH-CHECK ALERT - {self.lottery_name}*\n\n"
|
|
message += f"❌ *{len(self.issues)} Issues detected:*\n"
|
|
|
|
for issue in self.issues:
|
|
message += f"• {issue}\n"
|
|
|
|
if self.warnings:
|
|
message += f"\n⚠️ {len(self.warnings)} Warnings:\n"
|
|
for warning in self.warnings:
|
|
message += f"• {warning}\n"
|
|
|
|
message += f"\n🕐 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
|
|
|
try:
|
|
self.notifier._send_telegram(message)
|
|
except Exception as e:
|
|
print(f" ⚠️ Could not send alert: {e}")
|
|
|
|
|
|
def main():
|
|
"""Main function."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="System Health-Check")
|
|
parser.add_argument(
|
|
'--data-dir',
|
|
type=str,
|
|
default="/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data",
|
|
help='Data directory'
|
|
)
|
|
parser.add_argument(
|
|
'--lottery',
|
|
type=str,
|
|
default="Lotto",
|
|
help='Lottery name (Lotto or Eurojackpot)'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Run health check
|
|
checker = HealthCheck(args.data_dir, args.lottery)
|
|
success = checker.run_all_checks()
|
|
|
|
sys.exit(0 if success else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|