Godstrument/workers/world_crypto.py

#!/usr/bin/env python3
"""Live trade ticks from Binance — the fastest, most chaotic sources.

One websocket tap, four markets: BTC (the lead, legacy keys), ETH, SOL, and
EUR/USD via the EURUSDT pair — a real FX rate riding the same firehose. Every
trade fires price + tick-to-tick velocity.

And a trend kit on BTC, computed on 5-second closes so it is stable no matter
how fast the trades arrive: the MACD line (fast EMA minus slow), Wilder's
RSI(14), and the cross bells — /gs/crypto/cross_up rings 1.0 the moment the
fast average climbs over the slow (a golden cross, in miniature), cross_down
when it falls through. The market's mood, as playable signals.
"""
import argparse
import asyncio
import json
import urllib.parse
import urllib.request

from pythonosc.udp_client import SimpleUDPClient

NAME = "world_crypto"
SYMBOLS = {                    # binance symbol -> (price key, velocity key)
    "BTCUSDT": ("crypto/price", "crypto/vel"),   # legacy keys: BTC is the lead
    "ETHUSDT": ("crypto/eth", "crypto/eth_vel"),
    "SOLUSDT": ("crypto/sol", "crypto/sol_vel"),
    "EURUSDT": ("fx/eurusd", "fx/eurusd_vel"),   # the euro, live, off the same tap
}
# The cryptos ride raw trades; EURUSDT is thinly traded, so it rides the 1s
# miniTicker instead — a steady pulse rather than sparse, minutes-apart ticks.
STREAM_OF = {"EURUSDT": "@miniTicker"}
WS_URL = ("wss://stream.binance.com:9443/stream?streams="
          + "/".join(s.lower() + STREAM_OF.get(s, "@trade") for s in SYMBOLS))
REST_URL = ("https://api.binance.com/api/v3/ticker/price?symbols="
            + urllib.parse.quote(json.dumps(sorted(SYMBOLS), separators=(",", ":"))))
POLL_S = 2.0  # REST fallback poll interval
USER_AGENT = "godstrument/1.0 (+world_crypto)"
TREND_SYMBOL = "BTCUSDT"       # the trend kit listens to the lead


class TrendKit:
    """MACD line, Wilder RSI and cross bells on sampled closes.

    Tickless: feed it (t, price) whenever one arrives; it takes a close every
    `sample_s` seconds and returns {osc_key: value} to emit (empty between
    samples). EMAs seed on the first close; RSI stays quiet until it has
    rsi_n changes to chew on.
    """

    def __init__(self, fast=12, slow=60, rsi_n=14, sample_s=5.0):
        self.kf, self.ks = 2.0 / (fast + 1), 2.0 / (slow + 1)
        self.rsi_n, self.sample_s = rsi_n, sample_s
        self.ema_f = self.ema_s = None
        self.avg_gain = self.avg_loss = None
        self.changes = []          # pre-seed buffer for the RSI
        self.last_close = None
        self.next_t = None
        self.macd_sign = 0

    def on_price(self, t, price):
        if self.next_t is None:
            self.next_t = t + self.sample_s
        if t < self.next_t:
            return {}
        self.next_t = t + self.sample_s
        return self._close(float(price))

    def _close(self, c):
        out = {}
        # EMAs & the MACD line
        if self.ema_f is None:
            self.ema_f = self.ema_s = c
        else:
            self.ema_f += self.kf * (c - self.ema_f)
            self.ema_s += self.ks * (c - self.ema_s)
        macd = self.ema_f - self.ema_s
        out["crypto/macd"] = macd
        # the cross bells — ring on the sign flip, silent otherwise
        sign = 1 if macd > 0 else (-1 if macd < 0 else 0)
        out["crypto/cross_up"] = 1.0 if (sign > 0 and self.macd_sign < 0) else 0.0
        out["crypto/cross_down"] = 1.0 if (sign < 0 and self.macd_sign > 0) else 0.0
        if sign:
            self.macd_sign = sign
        # Wilder RSI
        if self.last_close is not None:
            ch = c - self.last_close
            gain, loss = max(ch, 0.0), max(-ch, 0.0)
            if self.avg_gain is None:
                self.changes.append((gain, loss))
                if len(self.changes) >= self.rsi_n:
                    self.avg_gain = sum(g for g, _ in self.changes) / self.rsi_n
                    self.avg_loss = sum(l for _, l in self.changes) / self.rsi_n
            else:
                a = 1.0 / self.rsi_n
                self.avg_gain += a * (gain - self.avg_gain)
                self.avg_loss += a * (loss - self.avg_loss)
            if self.avg_gain is not None:
                rs = self.avg_gain / self.avg_loss if self.avg_loss > 1e-12 else float("inf")
                out["crypto/rsi"] = 100.0 - 100.0 / (1.0 + rs)
        self.last_close = c
        return out


def make_emitter(client):
    """Per-symbol price+velocity emit, plus the trend kit on the lead."""
    last = {}
    kit = TrendKit()

    def emit(t, symbol, price):
        keys = SYMBOLS.get(symbol)
        if not keys:
            return
        client.send_message("/gs/" + keys[0], float(price))
        prev = last.get(symbol)
        client.send_message("/gs/" + keys[1], 0.0 if prev is None else abs(price - prev))
        last[symbol] = price
        if symbol == TREND_SYMBOL:
            for k, v in kit.on_price(t, price).items():
                client.send_message("/gs/" + k, float(v))
    return emit


async def run_websocket(emit):
    """Stream trades over the combined websocket."""
    import websockets

    loop = asyncio.get_running_loop()
    async with websockets.connect(WS_URL, ping_interval=20, ping_timeout=20) as ws:
        async for raw in ws:
            try:
                msg = json.loads(raw)["data"]
                symbol = msg["s"]
                price = float(msg["p"] if "p" in msg else msg["c"])  # trade / miniTicker close
            except (ValueError, KeyError, TypeError):
                continue  # malformed / unknown frame; skip
            emit(loop.time(), symbol, price)


async def poll_rest(emit):
    """Fallback: poll the combined REST ticker forever at POLL_S."""
    loop = asyncio.get_running_loop()
    while True:
        try:
            def fetch():
                req = urllib.request.Request(REST_URL, headers={"User-Agent": USER_AGENT})
                with urllib.request.urlopen(req, timeout=10) as resp:
                    return json.loads(resp.read().decode())

            for row in await loop.run_in_executor(None, fetch):
                emit(loop.time(), row["symbol"], float(row["price"]))
        except Exception as e:  # transient network / parse error — keep going
            print(f"[{NAME}] REST poll error: {e}; retrying")
        await asyncio.sleep(POLL_S)


def selftest():
    """The one runnable check: the trend kit's math, on synthetic prices."""
    kit = TrendKit(fast=3, slow=8, rsi_n=3, sample_s=1.0)
    t, out = 0.0, {}
    seen = {"up": 0, "down": 0}
    rsis = []
    for i in range(60):                        # rise then fall then rise
        price = 100 + (i if i < 20 else (40 - i if i < 40 else i - 40))
        t += 1.0
        out = kit.on_price(t + 0.5, price)     # t+0.5 always past the 1s boundary
        seen["up"] += out.get("crypto/cross_up", 0) > 0
        seen["down"] += out.get("crypto/cross_down", 0) > 0
        if "crypto/rsi" in out:
            rsis.append(out["crypto/rsi"])
    assert "crypto/macd" in out, "macd should emit every close"
    assert seen["down"] >= 1 and seen["up"] >= 1, f"expected both crosses, saw {seen}"
    assert all(0.0 <= r <= 100.0 for r in rsis) and rsis, "rsi must stay in [0,100]"
    assert min(rsis) < 30 < 70 < max(rsis), "rsi should swing wide on a full reversal"
    assert kit.on_price(t + 1.2, 100.0) == {} ,  "between samples the kit is silent"
    print(f"[{NAME}] selftest ok — {len(rsis)} rsi samples, crosses {seen}")


async def main():
    parser = argparse.ArgumentParser(description="Binance trade ticks (+BTC trend kit) -> OSC")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=9000)
    parser.add_argument("--selftest", action="store_true", help="check the trend math and exit")
    args = parser.parse_args()
    if args.selftest:
        selftest()
        return

    emit = make_emitter(SimpleUDPClient(args.host, args.port))
    print(f"[{NAME}] emitting {', '.join('/gs/' + k for k, _ in SYMBOLS.values())} "
          f"+ btc trend kit (macd/rsi/cross) every <1s (ws) / {POLL_S:g}s (rest)")

    backoff = 1.0
    while True:
        try:
            await run_websocket(emit)
            # Clean disconnect (stream ended): reconnect quickly.
            print(f"[{NAME}] websocket closed; reconnecting")
            backoff = 1.0
            await asyncio.sleep(1.0)   # avoid a hot loop on an instant re-close
        except Exception as e:
            # Bridge the gap on REST for the backoff window, then always retry
            # the socket — never latch onto REST for the rest of the session.
            print(f"[{NAME}] websocket error: {e}; REST bridge {backoff:g}s, then retry")
            try:
                await asyncio.wait_for(poll_rest(emit), timeout=backoff)
            except asyncio.TimeoutError:
                pass
            except Exception as e2:
                print(f"[{NAME}] REST bridge error: {e2}")
                await asyncio.sleep(min(backoff, POLL_S))
            backoff = min(backoff * 2, 30.0)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass