Upload icpe_ingest_relay.py

icpe_ingest_relay.py

@@ -1,219 +1,219 @@
-#!/usr/bin/env python3
-"""
-╔══════════════════════════════════════════════════════════════════════════════╗
-║  K1RL QUASAR — ICPE Ingest Relay  v1.1  (HARDCODED — TESTING ONLY)          ║
-║  (icpe_ingest_relay.py)                                                      ║
-║  Asset: V75                                                                   ║
-╠══════════════════════════════════════════════════════════════════════════════╣
-║  WHAT IT DOES                                                                ║
-║    Sits inside telegram_notifier.py and intercepts every valid signal        ║
-║    from _on_signal() — flips AND non-flips — then POSTs the full            ║
-║    structured JSON payload to RICPE /ingest on the main PIL space.          ║
-║                                                                              ║
-║  ⚠️  TESTING-ONLY BUILD                                                      ║
-║    ICPE_INGEST_URL, INTERNAL_API_KEY, and ASSET_SYMBOL are HARDCODED        ║
-║    below instead of read from HF Space Secrets. HF Secrets still           ║
-║    override these if present, so this is safe to leave in place — but      ║
-║    revert to the Secrets-driven build before relying on this long-term,    ║
-║    since hardcoded URLs/keys won't survive a space rename or key rotation. ║
-║                                                                              ║
-║  INTEGRATION INTO telegram_notifier.py  (4 surgical edits)                  ║
-║    1.  [top of file]  from icpe_ingest_relay import IcpeRelay               ║
-║    2.  [__init__]     self._icpe = IcpeRelay()                               ║
-║    3.  [_on_signal]   self._icpe.push(...)                                   ║
-║    4.  [finally]      self._icpe.shutdown()                                  ║
-║                                                                              ║
-║  DATA FLOW                                                                   ║
-║    Redis → _on_signal() → IcpeRelay.push()                                  ║
-║                         → ThreadPool → POST https://karlcon-portfolio-intelligence.hf.space/ricpe/ingest                    ║
-║                                      → RICPE ContextStore                   ║
-║                                      → PIL GET /context/{sym}              ║
-╚══════════════════════════════════════════════════════════════════════════════╝
-"""
-
-import logging
-import os
-import time
-from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime, timezone, timedelta
-from typing import Any, Dict, Optional
-
-import requests
-
-log = logging.getLogger("QUASAR.ICPE.Relay")
-
-# ─────────────────────────────────────────────────────────────────────────────
-#  Config — HARDCODED for testing. HF Secrets (if set) still take priority.
-# ─────────────────────────────────────────────────────────────────────────────
-_HARDCODED_URL     = "https://karlcon-portfolio-intelligence.hf.space/ricpe"
-_HARDCODED_API_KEY = ""   # ← leave blank: RICPE has no INTERNAL_API_KEY set yet
-_HARDCODED_SYMBOL  = "V75"
-
-_BASE_URL   = os.environ.get("ICPE_INGEST_URL", _HARDCODED_URL).rstrip("/")
-_API_KEY    = os.environ.get("INTERNAL_API_KEY", _HARDCODED_API_KEY)
-_SYM_ENV    = os.environ.get("ASSET_SYMBOL", _HARDCODED_SYMBOL).strip().upper()
-
-_CAT        = timezone(timedelta(hours=2))
-
-# ─────────────────────────────────────────────────────────────────────────────
-#  IcpeRelay
-# ─────────────────────────────────────────────────────────────────────────────
-
-class IcpeRelay:
-    """
-    Non-blocking ICPE ingest relay.
-
-    One instance per telegram_notifier process.  Fires a POST to RICPE /ingest
-    for every valid signal received from Redis — the RICPE ContextStore uses the
-    full stream (not just flips) to compute:
-        • flip_rate_per_min
-        • signal_frequency_hz
-        • momentum_bias
-        • directional_stability
-    which are injected into the quasar_pil.py LLM reasoning chain via
-    GET /context/{symbol}.
-
-    Thread-safety: push() is safe to call from the Redis listener thread
-    (same as send_telegram).  The thread pool is isolated; it never touches
-    the asyncio event loop.
-    """
-
-    def __init__(self, symbol: Optional[str] = None):
-        self.symbol  = (symbol or _SYM_ENV or "UNKNOWN").upper()
-        self.url     = _BASE_URL        # hardcoded above, env overrides if set
-        self.api_key = _API_KEY
-        self._pool   = ThreadPoolExecutor(max_workers=2, thread_name_prefix="icpe")
-        self._ok:   int = 0
-        self._fail: int = 0
-
-        if not self.url:
-            log.warning(
-                "[IcpeRelay] ICPE_INGEST_URL not set — relay DISABLED.\n"
-                "           Set ICPE_INGEST_URL in HF Space Secrets to enable."
-            )
-        else:
-            log.info(f"[IcpeRelay] {self.symbol} ──► {self.url}/ingest  (hardcoded build)")
-
-    # ── Public API ────────────────────────────────────────────────────────────
-
-    def push(
-        self,
-        *,
-        action: str,
-        price: float,
-        prev_action: str       = "",
-        is_flip: bool          = False,
-        flip_number: int       = 0,
-        agent: str             = "",
-        keys: int              = 1,
-        total_signals: int     = 0,
-        session_start: float   = 0.0,
-        raw_data: Optional[Dict] = None,
-    ) -> None:
-        """
-        Build the RICPE-compatible payload from every valid _on_signal() event
-        and submit it to the thread pool.  Returns immediately — never blocks
-        the Redis listener thread.
-
-        Sends for EVERY signal (flip AND non-flip) so RICPE builds the full
-        momentum and frequency picture for the PIL reasoning chain.
-        """
-        if not self.url:
-            return
-
-        now = time.time()
-        payload: Dict[str, Any] = {
-            # ── RICPE /ingest required fields ─────────────────────────────
-            "symbol":      self.symbol,
-            "action":      action.upper(),
-            "price":       round(float(price), 8),
-            # ── RICPE /ingest optional enrichment ─────────────────────────
-            "is_flip":     bool(is_flip),
-            "prev_action": prev_action.upper() if prev_action else "",
-            "flip_number": int(flip_number),
-            "agent":       agent.upper() if agent else "",
-            "keys":        int(keys),
-            # ── Session context — stored in raw_payload on RICPE ──────────
-            "total_signals": int(total_signals),
-            "session_min":   round((now - session_start) / 60, 2) if session_start else 0.0,
-            # ── Timestamps ────────────────────────────────────────────────
-            "ts_unix": now,
-            "ts_cat":  datetime.now(_CAT).strftime("%H:%M:%S CAT"),
-            # ── Full raw Redis payload ─────────────────────────────────────
-            # Stored by RICPE in SignalEvent.raw_payload for PIL to inspect.
-            "raw": raw_data or {},
-        }
-        self._pool.submit(self._post, payload)
-
-    def stats(self) -> Dict[str, Any]:
-        """Return live counters — useful to expose on a /health endpoint."""
-        return {
-            "symbol":      self.symbol,
-            "icpe_url":    self.url or "(not configured)",
-            "sends_ok":    self._ok,
-            "sends_fail":  self._fail,
-            "enabled":     bool(self.url),
-        }
-
-    def shutdown(self) -> None:
-        """Call from the notifier's finally block."""
-        self._pool.shutdown(wait=False)
-        log.info(f"[IcpeRelay] Shutdown — ok={self._ok}  fail={self._fail}")
-
-    # ── Internal ──────────────────────────────────────────────────────────────
-
-    def _post(self, payload: Dict[str, Any]) -> bool:
-        """
-        Synchronous HTTP POST — runs inside the thread pool, NOT the event loop.
-
-        Retry policy:
-          • Retry once after 1.5 s on connection / timeout errors.
-          • Never retry 4xx (bad payload) — avoids hammering RICPE with a
-            malformed body.
-          • Silently drops after two failed attempts; trading logic is unaffected.
-        """
-        url     = self.url + "/ingest"
-        headers = {"Content-Type": "application/json"}
-        if self.api_key:
-            headers["x-api-key"] = self.api_key
-
-        for attempt in range(2):
-            try:
-                resp = requests.post(url, json=payload, headers=headers, timeout=8)
-
-                if resp.status_code == 200:
-                    self._ok += 1
-                    log.debug(
-                        f"[IcpeRelay] ✅ {self.symbol} "
-                        f"{payload.get('action')} @ {payload.get('price')} "
-                        f"flip={payload.get('is_flip')}"
-                    )
-                    return True
-
-                # Bad payload — don't retry
-                if resp.status_code in (400, 422):
-                    self._fail += 1
-                    log.warning(
-                        f"[IcpeRelay] ⚠️ Bad payload "
-                        f"({resp.status_code}): {resp.text[:200]}"
-                    )
-                    return False
-
-                # Other HTTP error — retry once
-                log.warning(
-                    f"[IcpeRelay] HTTP {resp.status_code} (attempt {attempt + 1})"
-                )
-
-            except requests.exceptions.ConnectionError as exc:
-                log.warning(f"[IcpeRelay] Connection error (attempt {attempt + 1}): {exc}")
-            except requests.exceptions.Timeout:
-                log.warning(f"[IcpeRelay] Timeout 8 s (attempt {attempt + 1})")
-            except Exception as exc:
-                log.error(f"[IcpeRelay] ❌ Unexpected error: {exc}")
-                break   # don't retry unknown errors
-
-            if attempt == 0:
-                time.sleep(1.5)   # brief wait before retry
-
-        self._fail += 1
-        return False
+#!/usr/bin/env python3
+"""
+╔══════════════════════════════════════════════════════════════════════════════╗
+║  K1RL QUASAR — ICPE Ingest Relay  v1.1  (HARDCODED — TESTING ONLY)          ║
+║  (icpe_ingest_relay.py)                                                      ║
+║  Asset: V75                                                                   ║
+╠══════════════════════════════════════════════════════════════════════════════╣
+║  WHAT IT DOES                                                                ║
+║    Sits inside telegram_notifier.py and intercepts every valid signal        ║
+║    from _on_signal() — flips AND non-flips — then POSTs the full            ║
+║    structured JSON payload to RICPE /ingest on the main PIL space.          ║
+║                                                                              ║
+║  ⚠️  TESTING-ONLY BUILD                                                      ║
+║    ICPE_INGEST_URL, INTERNAL_API_KEY, and ASSET_SYMBOL are HARDCODED        ║
+║    below instead of read from HF Space Secrets. HF Secrets still           ║
+║    override these if present, so this is safe to leave in place — but      ║
+║    revert to the Secrets-driven build before relying on this long-term,    ║
+║    since hardcoded URLs/keys won't survive a space rename or key rotation. ║
+║                                                                              ║
+║  INTEGRATION INTO telegram_notifier.py  (4 surgical edits)                  ║
+║    1.  [top of file]  from icpe_ingest_relay import IcpeRelay               ║
+║    2.  [__init__]     self._icpe = IcpeRelay()                               ║
+║    3.  [_on_signal]   self._icpe.push(...)                                   ║
+║    4.  [finally]      self._icpe.shutdown()                                  ║
+║                                                                              ║
+║  DATA FLOW                                                                   ║
+║    Redis → _on_signal() → IcpeRelay.push()                                  ║
+║                         → ThreadPool → POST https://q-karl-portfolio-intelligence.hf.space/ricpe/ingest                    ║
+║                                      → RICPE ContextStore                   ║
+║                                      → PIL GET /context/{sym}              ║
+╚══════════════════════════════════════════════════════════════════════════════╝
+"""
+
+import logging
+import os
+import time
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone, timedelta
+from typing import Any, Dict, Optional
+
+import requests
+
+log = logging.getLogger("QUASAR.ICPE.Relay")
+
+# ────────────────────────���────────────────────────────────────────────────────
+#  Config — HARDCODED for testing. HF Secrets (if set) still take priority.
+# ─────────────────────────────────────────────────────────────────────────────
+_HARDCODED_URL     = "https://q-karl-portfolio-intelligence.hf.space/ricpe"
+_HARDCODED_API_KEY = ""   # ← leave blank: RICPE has no INTERNAL_API_KEY set yet
+_HARDCODED_SYMBOL  = "V75"
+
+_BASE_URL   = os.environ.get("ICPE_INGEST_URL", _HARDCODED_URL).rstrip("/")
+_API_KEY    = os.environ.get("INTERNAL_API_KEY", _HARDCODED_API_KEY)
+_SYM_ENV    = os.environ.get("ASSET_SYMBOL", _HARDCODED_SYMBOL).strip().upper()
+
+_CAT        = timezone(timedelta(hours=2))
+
+# ─────────────────────────────────────────────────────────────────────────────
+#  IcpeRelay
+# ─────────────────────────────────────────────────────────────────────────────
+
+class IcpeRelay:
+    """
+    Non-blocking ICPE ingest relay.
+
+    One instance per telegram_notifier process.  Fires a POST to RICPE /ingest
+    for every valid signal received from Redis — the RICPE ContextStore uses the
+    full stream (not just flips) to compute:
+        • flip_rate_per_min
+        • signal_frequency_hz
+        • momentum_bias
+        • directional_stability
+    which are injected into the quasar_pil.py LLM reasoning chain via
+    GET /context/{symbol}.
+
+    Thread-safety: push() is safe to call from the Redis listener thread
+    (same as send_telegram).  The thread pool is isolated; it never touches
+    the asyncio event loop.
+    """
+
+    def __init__(self, symbol: Optional[str] = None):
+        self.symbol  = (symbol or _SYM_ENV or "UNKNOWN").upper()
+        self.url     = _BASE_URL        # hardcoded above, env overrides if set
+        self.api_key = _API_KEY
+        self._pool   = ThreadPoolExecutor(max_workers=2, thread_name_prefix="icpe")
+        self._ok:   int = 0
+        self._fail: int = 0
+
+        if not self.url:
+            log.warning(
+                "[IcpeRelay] ICPE_INGEST_URL not set — relay DISABLED.\n"
+                "           Set ICPE_INGEST_URL in HF Space Secrets to enable."
+            )
+        else:
+            log.info(f"[IcpeRelay] {self.symbol} ──► {self.url}/ingest  (hardcoded build)")
+
+    # ── Public API ────────────────────────────────────────────────────────────
+
+    def push(
+        self,
+        *,
+        action: str,
+        price: float,
+        prev_action: str       = "",
+        is_flip: bool          = False,
+        flip_number: int       = 0,
+        agent: str             = "",
+        keys: int              = 1,
+        total_signals: int     = 0,
+        session_start: float   = 0.0,
+        raw_data: Optional[Dict] = None,
+    ) -> None:
+        """
+        Build the RICPE-compatible payload from every valid _on_signal() event
+        and submit it to the thread pool.  Returns immediately — never blocks
+        the Redis listener thread.
+
+        Sends for EVERY signal (flip AND non-flip) so RICPE builds the full
+        momentum and frequency picture for the PIL reasoning chain.
+        """
+        if not self.url:
+            return
+
+        now = time.time()
+        payload: Dict[str, Any] = {
+            # ── RICPE /ingest required fields ─────────────────────────────
+            "symbol":      self.symbol,
+            "action":      action.upper(),
+            "price":       round(float(price), 8),
+            # ── RICPE /ingest optional enrichment ─────────────────────────
+            "is_flip":     bool(is_flip),
+            "prev_action": prev_action.upper() if prev_action else "",
+            "flip_number": int(flip_number),
+            "agent":       agent.upper() if agent else "",
+            "keys":        int(keys),
+            # ── Session context — stored in raw_payload on RICPE ──────────
+            "total_signals": int(total_signals),
+            "session_min":   round((now - session_start) / 60, 2) if session_start else 0.0,
+            # ── Timestamps ────────────────────────────────────────────────
+            "ts_unix": now,
+            "ts_cat":  datetime.now(_CAT).strftime("%H:%M:%S CAT"),
+            # ── Full raw Redis payload ─────────────────────────────────────
+            # Stored by RICPE in SignalEvent.raw_payload for PIL to inspect.
+            "raw": raw_data or {},
+        }
+        self._pool.submit(self._post, payload)
+
+    def stats(self) -> Dict[str, Any]:
+        """Return live counters — useful to expose on a /health endpoint."""
+        return {
+            "symbol":      self.symbol,
+            "icpe_url":    self.url or "(not configured)",
+            "sends_ok":    self._ok,
+            "sends_fail":  self._fail,
+            "enabled":     bool(self.url),
+        }
+
+    def shutdown(self) -> None:
+        """Call from the notifier's finally block."""
+        self._pool.shutdown(wait=False)
+        log.info(f"[IcpeRelay] Shutdown — ok={self._ok}  fail={self._fail}")
+
+    # ── Internal ──────────────────────────────────────────────────────────────
+
+    def _post(self, payload: Dict[str, Any]) -> bool:
+        """
+        Synchronous HTTP POST — runs inside the thread pool, NOT the event loop.
+
+        Retry policy:
+          • Retry once after 1.5 s on connection / timeout errors.
+          • Never retry 4xx (bad payload) — avoids hammering RICPE with a
+            malformed body.
+          • Silently drops after two failed attempts; trading logic is unaffected.
+        """
+        url     = self.url + "/ingest"
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["x-api-key"] = self.api_key
+
+        for attempt in range(2):
+            try:
+                resp = requests.post(url, json=payload, headers=headers, timeout=8)
+
+                if resp.status_code == 200:
+                    self._ok += 1
+                    log.debug(
+                        f"[IcpeRelay] ✅ {self.symbol} "
+                        f"{payload.get('action')} @ {payload.get('price')} "
+                        f"flip={payload.get('is_flip')}"
+                    )
+                    return True
+
+                # Bad payload — don't retry
+                if resp.status_code in (400, 422):
+                    self._fail += 1
+                    log.warning(
+                        f"[IcpeRelay] ⚠️ Bad payload "
+                        f"({resp.status_code}): {resp.text[:200]}"
+                    )
+                    return False
+
+                # Other HTTP error — retry once
+                log.warning(
+                    f"[IcpeRelay] HTTP {resp.status_code} (attempt {attempt + 1})"
+                )
+
+            except requests.exceptions.ConnectionError as exc:
+                log.warning(f"[IcpeRelay] Connection error (attempt {attempt + 1}): {exc}")
+            except requests.exceptions.Timeout:
+                log.warning(f"[IcpeRelay] Timeout 8 s (attempt {attempt + 1})")
+            except Exception as exc:
+                log.error(f"[IcpeRelay] ❌ Unexpected error: {exc}")
+                break   # don't retry unknown errors
+
+            if attempt == 0:
+                time.sleep(1.5)   # brief wait before retry
+
+        self._fail += 1
+        return False