fix: Update predict model

Author: devWorld335

.env.example

@@ -31,9 +31,19 @@ PAPER_TRADING=0
 MARKET_INTERVAL_SECONDS=300
 ASSET = btc
 
+# Selective-bet gates (arb_bot_paper_predict.py)
+# Only trade when model fair prob, edge vs market, entry price, regime, and signals align.
+SELECTIVE_MIN_FAIR_PROB=0.53
+SELECTIVE_MIN_EDGE=0.12
+SELECTIVE_MAX_ENTRY_PRICE=0.40
+SELECTIVE_MIN_CONFIDENCE=53
+SELECTIVE_MIN_SIGNALS=2
+
 CHAINLINK_BTC_FEED_ID=
+CHAINLINK_TWAP_30S_FEED_ID=
+CHAINLINK_TWAP_WINDOW_SECONDS=30
 CHAINLINK_STREAMS_API_KEY=
-CHAINLINK_STREAMS_API_SECRET= 
+CHAINLINK_STREAMS_API_SECRET=
 
 
 

ecosystem.config.cjs

@@ -4,19 +4,23 @@
  * Requires: cwd = src/ so `config`, `service`, `client` imports work.
  * python-dotenv walks up from cwd to find `.env` at the repo root.
  *
- * Usage:
+ * Usage (run from the project root so the config file is found):
  *   mkdir -p logs
- *   pm2 start ecosystem.config.cjs
- *   pm2 start ecosystem.config.cjs --only pm-scalp
- *   pm2 start ecosystem.config.cjs --only pm-scalp --env live   # PAPER_TRADING=0
+ *   pm2 start ecosystem.config.cjs --only crypto-analytics   # analytics watcher
+ *   pm2 start ecosystem.config.cjs --only arb-predict        # paper bot + prediction model
+ *   pm2 logs arb-predict
  *
- * Logs: ./logs/pm-scalp-*.log
+ * Logs: ./logs/<app>-out.log and ./logs/<app>-error.log
  */
 const path = require("path");
 
 const projectRoot = path.resolve(__dirname);
 const srcRoot = path.join(projectRoot, "src");
 const botScript = path.join(srcRoot, "script", "market_maker_arb.py");
+const analyticsScript = path.join(srcRoot, "script", "crypto_analytics.py");
+const predictBotScript = path.join(srcRoot, "script", "arb_bot_paper_predict.py");
+// Use the project venv interpreter so requests/websocket-client are available.
+const venvPython = path.join(projectRoot, "venv", "bin", "python");
 
 module.exports = {
   apps: [
@@ -43,5 +47,43 @@ module.exports = {
       merge_logs: true,
       time: true,
     },
+    {
+      // Real-time Coinbase analytics + per-market UP/DOWN prediction logging.
+      //   pm2 start ecosystem.config.cjs --only crypto-analytics
+      name: "crypto-analytics",
+      script: analyticsScript,
+      args: "--watch --no-clear --refresh 5",
+      cwd: srcRoot,
+      interpreter: venvPython,
+      interpreter_args: "-u",
+      instances: 1,
+      autorestart: true,
+      watch: false,
+      max_restarts: 20,
+      min_uptime: "15s",
+      error_file: path.join(projectRoot, "logs", "crypto-analytics-error.log"),
+      out_file: path.join(projectRoot, "logs", "crypto-analytics-out.log"),
+      merge_logs: true,
+      time: true,
+    },
+    {
+      // Paper arb bot WITH the integrated prediction model (logs token +
+      // score + confidence each market, then the real chainlink result).
+      //   pm2 start ecosystem.config.cjs --only arb-predict
+      name: "arb-predict",
+      script: predictBotScript,
+      cwd: srcRoot,
+      interpreter: venvPython,
+      interpreter_args: "-u",
+      instances: 1,
+      autorestart: true,
+      watch: false,
+      max_restarts: 20,
+      min_uptime: "15s",
+      error_file: path.join(projectRoot, "logs", "arb-predict-error.log"),
+      out_file: path.join(projectRoot, "logs", "arb-predict-out.log"),
+      merge_logs: true,
+      time: true,
+    },
   ],
 };

requirements.txt

@@ -12,6 +12,7 @@ py-clob-client-v2>=1.0.0
 # WebSocket support
 websocket-client==1.6.4
 py-chainlink-streams>=0.3.4
+polymarket-client>=0.3.0
 # Date/time utilities
 python-dateutil>=2.8.2
 black==24.4.2

src/client/chainlink_client/chainlink_client.py

@@ -38,12 +38,53 @@
 STREAM_RESTART_DELAY = 5
 
 
+def _decode_v2_report_data(report_blob: bytes) -> dict:
+    """Decode v2 report blob (TWAP feeds). py_chainlink_streams only supports v3."""
+    from eth_abi import decode  # pyright: ignore[reportMissingImports]
+
+    types = [
+        "bytes32",  # feedId
+        "uint32",   # validFromTimestamp
+        "uint32",   # observationsTimestamp
+        "uint192",  # nativeFee
+        "uint192",  # linkFee
+        "uint32",   # expiresAt
+        "int192",   # benchmarkPrice (TWAP)
+    ]
+    decoded = decode(types, report_blob)
+    return {
+        "observationsTimestamp": decoded[2],
+        "benchmarkPrice": decoded[6],
+    }
+
+
+def _extract_report_price(report: "ReportResponse") -> tuple[float, Optional[int]]:
+    """Decode benchmark (v3) or TWAP (v2) price from a Chainlink report."""
+    schema = ReportResponse.get_schema_version(report.feed_id)
+    if schema == 3:
+        prices = report.get_decoded_prices()
+        obs_ts = prices.get("observationsTimestamp")
+        return float(prices.get("benchmarkPrice", 0.0)), int(obs_ts) if obs_ts is not None else None
+    if schema == 2:
+        structure = ReportResponse._decode_report_structure(report.full_report)
+        data = _decode_v2_report_data(structure["reportBlob"])
+        price = ReportResponse.convert_fixed_point_to_decimal(data["benchmarkPrice"])
+        return price, int(data.get("observationsTimestamp", report.observations_timestamp))
+    raise ValueError(f"Unsupported Chainlink report schema v{schema} for feed {report.feed_id}")
+
+
 class ChainlinkClient:
-    def __init__(self, feed_id: str, start_stream_thread: bool = True):
+    def __init__(
+        self,
+        feed_id: str,
+        start_stream_thread: bool = True,
+        price_config_key: str = "CURRENT_PRICE",
+    ):
         self.streams_config = None
         self.client = None
         self.feed_id = feed_id
         self.feed_ids = [feed_id] if feed_id else []
+        self.price_config_key = price_config_key
         self._price_history: deque = deque(maxlen=3000)
         self._last_price: Optional[float] = None
         self._last_update_time: Optional[int] = None
@@ -68,9 +109,20 @@ def __init__(self, feed_id: str, start_stream_thread: bool = True):
             else:
                 logger.warning("Chainlink streams client unavailable; stream features are disabled.")
 
-        if start_stream_thread and self.client is not None:
-            self.thread = threading.Thread(target=self._stream, daemon=True)
+        if not self.feed_id:
+            logger.warning(
+                "Chainlink feed_id missing for %s; stream and historical fetch disabled.",
+                self.price_config_key,
+            )
+
+        if start_stream_thread and self.client is not None and self.feed_id:
+            self.thread = threading.Thread(
+                target=self._stream,
+                daemon=True,
+                name=f"chainlink-{self.price_config_key}",
+            )
             self.thread.start()
+
     def _price_at_ago_ms(self, history: List[Tuple[int, float]], now_ms: int, ago_ms: int) -> Optional[float]:
         """Return price from history closest to (now_ms - ago_ms)."""
         target = now_ms - ago_ms
@@ -85,7 +137,6 @@ def _price_at_ago_ms(self, history: List[Tuple[int, float]], now_ms: int, ago_ms
                 best_price = price
         return best_price
 
-
     @property
     def last_price(self) -> Optional[float]:
         """Last received trade price, or None if no update yet."""
@@ -115,8 +166,8 @@ def get_price_at_timestamp(self, timestamp: int | float | str) -> float:
             logger.warning("Invalid timestamp for Chainlink report: %r", timestamp)
             return 0.0
         report = self.client.get_report(self.feed_id, ts_int)
-        prices = report.get_decoded_prices()
-        return float(prices.get("benchmarkPrice", 0.0))
+        price, _obs_ts = _extract_report_price(report)
+        return price
 
     def _stream(self) -> None:
         if self.client is None:
@@ -142,14 +193,20 @@ def _stream(self) -> None:
 
     def _on_connection_status(self, is_connected: bool, host: str, _origin: str) -> None:
         if is_connected:
-            logger.info("Chainlink stream connected to %s", host)
+            logger.info("Chainlink stream connected to %s (%s)", host, self.price_config_key)
         else:
-            logger.warning("Chainlink stream disconnected from %s", host)
+            logger.warning("Chainlink stream disconnected from %s (%s)", host, self.price_config_key)
 
     async def _process_report(self, report_data: dict) -> None:
         if ReportResponse is None:
             return
-        report = ReportResponse.from_dict(report_data)
-        prices = report.get_decoded_prices()
-        price = float(prices.get("benchmarkPrice", 0.0))
-        config.CURRENT_PRICE = price
+        try:
+            report = ReportResponse.from_dict(report_data)
+            price, obs_ts = _extract_report_price(report)
+        except Exception:
+            logger.exception("Failed to decode Chainlink report for %s", self.price_config_key)
+            return
+        self._last_price = price
+        if obs_ts is not None:
+            self._last_update_time = int(obs_ts) * 1000
+        setattr(config, self.price_config_key, price)

src/client/coinbase_client/coinbase_client.py

@@ -32,6 +32,7 @@
 import sys
 import threading
 import time
+from collections import deque
 from pathlib import Path
 from typing import Any, Optional
 
@@ -73,20 +74,29 @@ class CoinbaseClient:
     of that update so callers can detect staleness.
     """
 
-    def __init__(self, debug: bool = False) -> None:
+    def __init__(self, debug: bool = False, buffer_seconds: float = 0.0) -> None:
         """
         Initialize CoinbaseClient.
 
         Args:
             debug: If True, log status/error events received from the
                 child process to stdout. Off by default to keep the
                 trading logs uncluttered.
+            buffer_seconds: If > 0, keep a rolling, thread-safe buffer of recent
+                trades (price/size/side/ts) spanning this many seconds. This is
+                what the prediction model reads from, so a single Coinbase WS
+                feeds both the live price and the analytics — no second socket.
         """
         asset = (os.getenv("ASSET") or "BTC").strip().upper() or "BTC"
         self.symbol = f"{asset}-USD"
 
         self._debug = debug
 
+        # Optional rolling trade buffer (enabled when buffer_seconds > 0).
+        self._buffer_seconds = float(buffer_seconds)
+        self._trade_buffer: deque = deque()
+        self._buffer_lock = threading.Lock()
+
         # Child-process plumbing — created lazily in start().
         self._ws_proc: Optional[mp.Process] = None
         self._event_q: Optional[Any] = None  # mp.Queue
@@ -190,6 +200,61 @@ def last_update_time(self) -> Optional[int]:
         """Wall-clock ms timestamp of last price update, or None."""
         return self._last_update_time
 
+    # ------------------------------------------------------------------
+    # Rolling trade buffer (read by the prediction model)
+    # ------------------------------------------------------------------
+
+    def _append_trade(self, ts_s: float, price: float, size: Any, side: Any, tid: Any) -> None:
+        """Append one trade and prune to the retention window. Thread-safe."""
+        try:
+            size_f = float(size) if size is not None else 0.0
+        except (TypeError, ValueError):
+            size_f = 0.0
+        rec = {
+            "ts": ts_s,
+            "price": price,
+            "size": size_f,
+            "side": str(side) if side is not None else "",
+            "tid": int(tid) if isinstance(tid, (int, float)) else None,
+        }
+        cutoff = time.time() - self._buffer_seconds
+        with self._buffer_lock:
+            self._trade_buffer.append(rec)
+            while self._trade_buffer and self._trade_buffer[0]["ts"] < cutoff:
+                self._trade_buffer.popleft()
+
+    def get_trades(self, since: Optional[float] = None) -> list[dict]:
+        """Return a copy of buffered trades (oldest -> newest).
+
+        Each item is ``{"ts", "price", "size", "side", "tid"}`` with ``ts`` in
+        epoch seconds. Pass ``since`` (epoch seconds) to filter to recent trades.
+        Returns an empty list if buffering is disabled.
+        """
+        with self._buffer_lock:
+            trades = list(self._trade_buffer)
+        if since is not None:
+            trades = [t for t in trades if t["ts"] >= since]
+        return trades
+
+    def seed_trades(self, trades: list[dict]) -> None:
+        """Merge externally-fetched (e.g. REST backfill) trades into the buffer,
+        de-duplicating by ``tid`` and keeping the buffer time-sorted/pruned."""
+        if not trades:
+            return
+        cutoff = time.time() - self._buffer_seconds
+        with self._buffer_lock:
+            seen = {t["tid"] for t in self._trade_buffer if t.get("tid") is not None}
+            merged = list(self._trade_buffer)
+            for t in trades:
+                tid = t.get("tid")
+                if tid is not None and tid in seen:
+                    continue
+                if tid is not None:
+                    seen.add(tid)
+                merged.append(t)
+            merged.sort(key=lambda r: r["ts"])
+            self._trade_buffer = deque(t for t in merged if t["ts"] >= cutoff)
+
     # ------------------------------------------------------------------
     # REST helpers (unchanged behaviour)
     # ------------------------------------------------------------------
@@ -288,6 +353,9 @@ def _dispatch_loop(self) -> None:
                 self._last_update_time = (
                     int(ts_ms) if isinstance(ts_ms, (int, float)) else int(time.time() * 1000)
                 )
+                if self._buffer_seconds > 0:
+                    ts_s = (float(ts_ms) / 1000.0) if isinstance(ts_ms, (int, float)) else time.time()
+                    self._append_trade(ts_s, price_f, msg.get("size"), side, msg.get("trade_id"))
                 continue
 
             if etype == "_status":

src/client/coinbase_client/coinbase_ws_worker.py

@@ -15,7 +15,8 @@
 
 Wire protocol on ``event_q`` (child -> parent)
 ----------------------------------------------
-- ``{"event_type": "match",   "price": float, "ts_ms": int}`` per trade.
+- ``{"event_type": "match",   "price": float, "side": str, "size": float,
+     "trade_id": int|None, "ts_ms": int}`` per trade.
 - ``{"event_type": "_status", "connected": bool}`` on (dis)connect.
 - ``{"event_type": "_error",  "error": str}`` on protocol/library errors.
 """
@@ -133,9 +134,13 @@ def on_message(ws, message):  # noqa: ARG001
             return
         try:
             # print("data", data)
-            # size = float(data["last_size"])
             side = data["side"]
             price = float(data["price"])
+            try:
+                size = float(data.get("last_size") or 0.0)
+            except (TypeError, ValueError):
+                size = 0.0
+            tid = data.get("trade_id")
             # best_bid = float(data["best_bid"])
             # best_ask = float(data["best_ask"])
             # best_bid_size = float(data["best_bid_size"])
@@ -149,6 +154,8 @@ def on_message(ws, message):  # noqa: ARG001
             "event_type": "match",
             "price": price,
             "side": side,
+            "size": size,
+            "trade_id": tid,
             "ts_ms": int(time.time() * 1000),
         })
 

src/client/polymarket_twap_client/__init__.py

@@ -0,0 +1,3 @@
+from .polymarket_twap_client import PolymarketTwapClient
+
+__all__ = ["PolymarketTwapClient"]