refactor(health): unified HealthMonitor API — one class for all deployments

- HealthMonitor(app=...) → Flask /health endpoint
- HealthMonitor(http_port=8080) → standalone HTTP for CLI/VPS
- HealthMonitor(file_path='/tmp/qsl.heartbeat') → file-based for self-hosted
- HealthMonitor() → auto-detect (Firestore if GCP, otherwise file)
- Backward-compat: register_health_endpoint, FileHeartbeat, FirestoreHeartbeat

Co-Authored-By: Claude <noreply@anthropic.com>

Author: Pigbibi

scripts/qsl_watchdog.py

@@ -1,12 +1,15 @@
 #!/usr/bin/env python3
-"""Universal QSL watchdog — ping health endpoints and alert on failure.
+"""Universal QSL Watchdog — ping health endpoints, alert on failure.
+
+Works for ANY deployment — Cloud Run, VPS, bare metal, self-hosted.
 
-Works with any deployment type:
   Cloud Run:  qsl_watchdog.py --url https://my-service.run.app
-  VPS:        qsl_watchdog.py --file /var/run/qsl/heartbeat.json
-  Firestore:  qsl_watchdog.py --firestore health/alive
+  VPS HTTP:   qsl_watchdog.py --url http://my-vps:8080
+  File-based: qsl_watchdog.py --file /tmp/qsl.heartbeat
+
+Exit 0 = alive, 1 = dead.  Can be used with cron, GitHub Actions, or any scheduler.
 
-Exit code 0 = alive, 1 = dead (integrates with cron / GitHub Actions).
+Optional Telegram alert: set TELEGRAM_TOKEN + GLOBAL_TELEGRAM_CHAT_ID env vars.
 """
 from __future__ import annotations
 
@@ -15,56 +18,47 @@
 import sys
 
 
-def send_telegram(message: str) -> bool:
-    token = (os.environ.get("TELEGRAM_TOKEN") or os.environ.get("TG_TOKEN") or "").strip()
-    chat_id = (os.environ.get("GLOBAL_TELEGRAM_CHAT_ID") or os.environ.get("TG_CHAT_ID") or "").strip()
+def _telegram_alert(message: str) -> bool:
+    token = os.environ.get("TELEGRAM_TOKEN", "").strip()
+    chat_id = os.environ.get("GLOBAL_TELEGRAM_CHAT_ID", "").strip()
     if not token or not chat_id:
-        print("No Telegram config; cannot send alert.", file=sys.stderr)
         return False
     import json, urllib.request
-    url = f"https://api.telegram.org/bot{token}/sendMessage"
-    body = json.dumps({"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}).encode()
-    req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST")
+    req = urllib.request.Request(
+        f"https://api.telegram.org/bot{token}/sendMessage",
+        data=json.dumps({"chat_id": chat_id, "text": message}).encode(),
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
     try:
-        with urllib.request.urlopen(req, timeout=10) as resp:
-            return resp.status == 200
+        with urllib.request.urlopen(req, timeout=10) as r:
+            return r.status == 200
     except Exception:
         return False
 
 
 def main() -> int:
-    p = argparse.ArgumentParser(description="QSL health watchdog")
-    p.add_argument("--url", help="HTTP health endpoint URL")
-    p.add_argument("--file", help="Local heartbeat file path")
-    p.add_argument("--firestore", help="Firestore collection/document (e.g. health/alive)")
-    p.add_argument("--name", default="QSL Platform", help="Service name for alert messages")
-    p.add_argument("--max-age", type=int, default=300, help="Max heartbeat age in seconds")
+    p = argparse.ArgumentParser(description="Universal QSL health watchdog")
+    g = p.add_mutually_exclusive_group(required=True)
+    g.add_argument("--url", help="HTTP health endpoint URL (e.g. https://svc.run.app or http://vps:8080)")
+    g.add_argument("--file", help="Local heartbeat file path (e.g. /tmp/qsl.heartbeat)")
+    p.add_argument("--name", default="QSL Platform", help="Service name for alerts")
+    p.add_argument("--max-age", type=int, default=300, help="Max heartbeat age in seconds (default 300)")
     p.add_argument("--alert", action="store_true", help="Send Telegram alert on failure")
     args = p.parse_args()
 
-    from quant_platform_kit.common.health import check_service_alive, FileHeartbeat, FirestoreHeartbeat
+    from quant_platform_kit.common.health import check_alive
 
-    reader = None
-    if args.file:
-        reader = FileHeartbeat(args.file)
-    elif args.firestore:
-        parts = args.firestore.split("/")
-        reader = FirestoreHeartbeat(parts[0], parts[1] if len(parts) > 1 else "alive")
-
-    alive, detail = check_service_alive(
-        heartbeat_url=args.url,
-        heartbeat_reader=reader,
-        max_age_seconds=args.max_age,
-    )
+    alive, detail = check_alive(url=args.url or "", file_path=args.file or "", max_age_seconds=args.max_age)
 
     if alive:
         print(f"✅ {args.name}: {detail}")
         return 0
 
-    msg = f"🚨 *{args.name}* health check FAILED: `{detail}`"
-    print(f"❌ {args.name}: {detail}")
+    msg = f"🚨 *{args.name}* DOWN: {detail}"
+    print(f"❌ {msg}")
     if args.alert:
-        send_telegram(msg)
+        _telegram_alert(msg)
     return 1