From 49a83ba135a05302c390d3f5208023660b050860 Mon Sep 17 00:00:00 2001
From: Benjamin Adovasio <benjamin@adovasio.com>
Date: Mon, 9 Feb 2026 18:28:33 -0500
Subject: [PATCH] added spot_feed,py

---
 spot_feed.py | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 98 insertions(+)
 create mode 100644 spot_feed.py

diff --git a/spot_feed.py b/spot_feed.py
new file mode 100644
index 0000000..4ba8bf6
--- /dev/null
+++ b/spot_feed.py
@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import time
+from collections import deque
+from dataclasses import dataclass
+from typing import Deque, Dict, Tuple
+
+import requests
+
+
+@dataclass
+class SpotPoint:
+    ts: float
+    price: float
+
+
+class SpotFeed:
+    """
+    Simple polling spot feed (Coinbase spot). Keeps rolling history for volatility + jump checks.
+    """
+    def __init__(self, urls: Dict[str, str], lookback_seconds: int):
+        self.urls = urls
+        self.lookback_seconds = lookback_seconds
+        self.history: Dict[str, Deque[SpotPoint]] = {sym: deque() for sym in urls.keys()}
+
+    def _fetch(self, url: str) -> float:
+        r = requests.get(url, timeout=10)
+        r.raise_for_status()
+        data = r.json()
+        # Coinbase shape: {"data": {"amount": "70428.82", "currency": "USD"}}
+        return float(data["data"]["amount"])
+
+    def update(self) -> Dict[str, float]:
+        now = time.time()
+        out: Dict[str, float] = {}
+        for sym, url in self.urls.items():
+            px = self._fetch(url)
+            out[sym] = px
+            dq = self.history[sym]
+            dq.append(SpotPoint(ts=now, price=px))
+            # trim
+            cutoff = now - self.lookback_seconds
+            while dq and dq[0].ts < cutoff:
+                dq.popleft()
+        return out
+
+    def latest(self, sym: str) -> float:
+        dq = self.history[sym]
+        if not dq:
+            raise RuntimeError(f"No spot data yet for {sym}")
+        return dq[-1].price
+
+    def returns_over_window(self, sym: str, window_seconds: int) -> float:
+        dq = self.history[sym]
+        if len(dq) < 2:
+            return 0.0
+        now = dq[-1].ts
+        cutoff = now - window_seconds
+        # find earliest point >= cutoff
+        base = dq[0]
+        for p in dq:
+            if p.ts >= cutoff:
+                base = p
+                break
+        if base.price <= 0:
+            return 0.0
+        return (dq[-1].price / base.price) - 1.0
+
+    def realized_vol(self, sym: str) -> float:
+        """
+        Very conservative realized vol estimate from simple returns in the stored history.
+        Returns a per-second sigma (not annualized).
+        """
+        dq = self.history[sym]
+        if len(dq) < 3:
+            return 0.0
+        rets = []
+        for i in range(1, len(dq)):
+            p0 = dq[i - 1].price
+            p1 = dq[i].price
+            if p0 > 0:
+                rets.append((p1 / p0) - 1.0)
+        if len(rets) < 2:
+            return 0.0
+
+        # compute stddev of returns per sample
+        mean = sum(rets) / len(rets)
+        var = sum((r - mean) ** 2 for r in rets) / (len(rets) - 1)
+
+        # Estimate average sampling interval
+        dt_avg = (dq[-1].ts - dq[0].ts) / max(1, (len(dq) - 1))
+        if dt_avg <= 0:
+            return 0.0
+
+        # Convert return std per sample to per-second sigma
+        std_per_sample = var ** 0.5
+        sigma_per_second = std_per_sample / (dt_avg ** 0.5)
+        return sigma_per_second