#!/usr/bin/env python3 """ Fleet Client — talks to the PHP coordinator (orb_fleet_api.php). Pure stdlib (urllib), no new pip dependencies. All methods return a dict on success or None on network failure — callers decide what "coordinator unreachable" means for them (claim → retry, signal_check → fail-open/closed per config). """ import json import socket import logging import urllib.request import urllib.error from typing import Any, Dict, List, Optional class FleetClient: def __init__(self, base_url: str, api_key: str, worker_name: str, timeout: int = 12, logger: Optional[logging.Logger] = None): self.base_url = base_url.rstrip("/") self.api_url = (base_url if base_url.endswith(".php") else self.base_url + "/orb_fleet_api.php") self.api_key = api_key self.worker_name = worker_name self.timeout = timeout self.logger = logger or logging.getLogger("FleetClient") self.worker_id: Optional[int] = None # ── Low-level POST ─────────────────────────────────────────────────────── def _post(self, action: str, data: Optional[Dict[str, Any]] = None, retries: int = 2) -> Optional[Dict[str, Any]]: payload = dict(data or {}) payload["action"] = action if self.worker_id is not None and "worker_id" not in payload: payload["worker_id"] = self.worker_id body = json.dumps(payload).encode("utf-8") for attempt in range(retries + 1): try: req = urllib.request.Request( self.api_url, data=body, method="POST", headers={ "Content-Type": "application/json", "X-Fleet-Key" : self.api_key, "User-Agent" : f"ORBFleetWorker/{self.worker_name}", }) with urllib.request.urlopen(req, timeout=self.timeout) as resp: out = json.loads(resp.read().decode("utf-8")) if not out.get("ok"): self.logger.warning(f"[fleet:{action}] server said: {out.get('error')}") # Coordinator lost our row (DB restore etc.) → re-register if out.get("reregister"): self.logger.warning("[fleet] coordinator asked to re-register") self.worker_id = None if self.register() and action != "register": payload["worker_id"] = self.worker_id body = json.dumps(payload).encode("utf-8") continue return out except (urllib.error.URLError, socket.timeout, json.JSONDecodeError, OSError) as e: if attempt < retries: self.logger.debug(f"[fleet:{action}] retry {attempt+1}: {e}") else: self.logger.warning(f"[fleet:{action}] unreachable: {e}") return None # ── Lifecycle ──────────────────────────────────────────────────────────── def register(self, hostname: str = "", version: str = "1.0") -> bool: if not hostname: try: hostname = socket.gethostname() except Exception: hostname = "unknown" r = self._post("register", { "worker_name": self.worker_name, "hostname" : hostname, "version" : version, }) if r and r.get("ok"): self.worker_id = int(r["worker_id"]) self.logger.info(f"✅ Registered with fleet as worker #{self.worker_id} " f"('{self.worker_name}')") return True return False def heartbeat(self, current_ticker: Optional[str] = None) -> Optional[Dict]: """Returns commands dict {'paused','pause_reason','quarantined','market_state', 'turbo','batch_size','check_interval'}.""" r = self._post("heartbeat", {"current_ticker": current_ticker}, retries=0) return r.get("commands") if r and r.get("ok") else None def progress(self, ticker: Optional[str], phase: str = "scanning") -> None: """Fire-and-forget live-activity ping → dashboard 'Now scanning' column.""" self._post("progress", {"ticker": ticker, "phase": phase}, retries=0) # ── Work queue ─────────────────────────────────────────────────────────── def claim(self, batch_size: Optional[int] = None) -> Optional[Dict]: """ Returns {'tickers':[{id,ticker,tv_url,...}], 'lease_token':.., 'retry_in':sec, 'paused':bool, 'pause_reason':str, 'turbo':bool} or None if unreachable. """ data = {} if batch_size: data["batch_size"] = batch_size return self._post("claim", data) def report_scan(self, ticker_id: int, lease_token: str, duration_sec: float, signal_found: bool = False, error: Optional[str] = None, ocr_ok: bool = True) -> Optional[Dict]: """ocr_ok=False flags a BLIND scan (blank/unreadable chart) — feeds the server-side quarantine counter. Returns {'quarantined':bool} on success.""" return self._post("report_scan", { "ticker_id" : ticker_id, "lease_token" : lease_token, "duration_sec": round(duration_sec, 2), "signal_found": bool(signal_found), "error" : error, "ocr_ok" : bool(ocr_ok), }, retries=1) def release_all(self) -> None: self._post("release_all", retries=1) # ── Global signal registry ─────────────────────────────────────────────── def signal_check(self, ticker: str, signal_type: str, entry=None, tp=None, sl=None, sig_time: Optional[str] = None) -> Optional[Dict]: """ Global fleet-wide dedup. Returns: {'duplicate': bool, 'signal_id': int, 'send_telegram': bool} or None if coordinator unreachable (caller applies fail-open policy). """ return self._post("signal_check", { "ticker" : ticker, "signal_type": signal_type, "entry" : entry, "tp" : tp, "sl" : sl, "time" : sig_time, }) def signal_tg_sent(self, fleet_signal_id: int) -> None: self._post("signal_tg_sent", {"signal_id": fleet_signal_id}, retries=1) def price_push(self, ticker: str, price: float) -> None: """Feed the coordinator's paper-trading engine with live prices the price monitors are already polling. Fire-and-forget.""" self._post("price_push", {"ticker": ticker, "price": price}, retries=0) def signal_outcome(self, fleet_signal_id: int, outcome: str, exit_price=None, max_fav_pct=None) -> None: """Report a price-monitor outcome (tp_hit / sl_hit / timeout) so the dashboard's win-rate table stays live.""" self._post("signal_outcome", { "signal_id" : fleet_signal_id, "outcome" : outcome, "exit_price" : exit_price, "max_fav_pct": max_fav_pct, }, retries=1) # ── Anomaly screenshots ────────────────────────────────────────────────── def screenshot_upload(self, ticker: str, reason: str, image_b64: str, note: Optional[str] = None) -> None: """Upload a (downscaled JPEG) base64 screenshot for dashboard review.""" self._post("screenshot_upload", { "ticker" : ticker, "reason" : reason, "note" : note, "image_b64": image_b64, }, retries=0)