99 lines
2.9 KiB
Python
99 lines
2.9 KiB
Python
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
|