crypto.price / crypto.vel β Bitcoin. Live trades off the Binance websocket firehose. Price is zscore-normalized over a 400-sample window (so it's "how unusual is this price vs the recent run," not an absolute dollar figure); volatility (crypto.vel) is minmax over 120 samples with an aggressive, snappy filter (min_cutoff 2.0, Ξ² 0.15) so it reacts in near-real-time. This is the fastest feed in the instrument: sub-second chaos, a genuine rhythm engine. BTC volatility drives the hats (amt 0.8, scurve) and nudges the tempo (amt 0.25) β when the market convulses, the percussion densifies and the track leans forward. And Bitcoin plays the lead melody (crypto.price β lead.note, quantized to minor pentatonic, root 60 / C, 2 octaves) β the price doesn't wobble formlessly, it picks notes in a key. The market's fear and greed, rendered as a melody line that's always in tune.
+The trading desk β more markets, and the market's mood as signals. The same Binance tap now carries ETH and SOL (crypto.eth, crypto.sol, each with its own velocity) and β because Binance quotes the euro itself β EUR/USD live (fx.eurusd, on a steady one-second pulse). The slower currencies arrive as ECB daily reference rates: fx.gbpusd, fx.usdjpy, fx.audusd β glacial, tectonic feeds like the national debt; drones, not melodies. And Bitcoin now speaks its own technical analysis, computed on five-second closes so it is stable no matter how fast the trades storm in: crypto.macd is the MACD line (fast EMA minus slow β momentum, z-scored so above-half means a bullish push), crypto.rsi is Wilder's RSI(14) (a natural mean-reverting wave: near 1 overbought, near 0 oversold β practically a market-driven LFO), and the cross bells: crypto.cross_up fires a single impulse the moment the fast average climbs over the slow (a golden cross in miniature) and crypto.cross_down when it falls through β patch them to a bell, a chord, the reverb throwing open, and the market's turning points become events you hear happen. Every one of these routes through the Spirits like any money: to the solar plexus, to Hod, to the dwarves in Nidavellir, to α fehu β wealth is strife among kinsmen.
+market.velocity / market.turnover β your dealgod market. Your own deal-flow, read from the basegod warehouse (dealgod.price_events, ~4M rows) over the tailnet β a single read-only aggregate query, no writes. Binned into an activity curve (velocity) and an average-price curve (turnover), replayed compressed and jog-scrubbable. Both are pre-normalized 0β1. Character: a replay with a playhead, so it's scratchable like a record. Your market drives the hats (amt 0.7) and nudges tempo (amt 0.2), mirroring BTC; pricier gear moving = more grit (turnover β saturation, amt 0.5, exp); and your market plays the lead (turnover β lead.note, G dorian, root 55, 2 octaves). This is the one voice that is yours β your commerce, given a melody in a mode with a little melancholy built in.
@@ -1166,6 +1168,19 @@ // short "what it is" for the output voices/params (sources use their label) const DESC = { + "crypto.eth": "ETH price, live trade ticks from Binance", + "crypto.eth_vel": "ETH tick-to-tick velocity β ether's jitter", + "crypto.sol": "SOL price, live trade ticks from Binance", + "crypto.sol_vel": "SOL tick-to-tick velocity", + "crypto.macd": "BTC momentum β fast EMA minus slow (the MACD line) on 5s closes; above the middle = bullish push", + "crypto.rsi": "BTC RSI(14) on 5s closes β near 1 overbought, near 0 oversold; a natural mean-reverting wave", + "crypto.cross_up": "golden cross β BTC's fast average just crossed above the slow. A bell rings for the bulls.", + "crypto.cross_down": "death cross β the fast average just fell through the slow. The bears' bell.", + "fx.eurusd": "EUR/USD, live from Binance's EURUSDT tap β the euro breathing in real time", + "fx.eurusd_vel": "EUR/USD tick-to-tick velocity", + "fx.gbpusd": "GBP/USD β cable, from the ECB daily fix (glacial: a drone, not a melody)", + "fx.usdjpy": "USD/JPY β the yen, from the ECB daily fix", + "fx.audusd": "AUD/USD β the aussie, from the ECB daily fix", "filter.cutoff": "master brightness β opens & closes the whole sound", "reverb.size": "the size of the space everything sits in", "delay.feedback": "echo β how much the sound repeats and smears", @@ -2304,7 +2319,11 @@ const s = msg.sources[k]; const n = ensureSource(k, s); if (s.label) n.label = s.label; - if (typeof s.raw === "number") n.raw = s.raw; + if (typeof s.raw === "number") { + n.raw = s.raw; + const hh = n.hist || (n.hist = []); // a couple of minutes of memory, for the sparkline + if (hh[hh.length - 1] !== s.raw || hh.length < 2) { hh.push(s.raw); if (hh.length > 120) hh.shift(); } + } if (typeof s.norm === "number") n.norm = clamp(s.norm, 0, 1); if (s.event) n.isEvent = true; n.muted = !!s.muted; @@ -2982,7 +3001,7 @@ if (["wiki","net","news","social","edit","econ","debt"].includes(p) || // V Β· abstract thought & money-as-idea /inflation/.test(k) || k === "market.turnover" || k === "crypto.price") return "mental"; if (/vel$|velocity/.test(k) || // VI Β· desire, fire, felt sensation - ["fire","crypto","market","light","ambient","audio","mic","me"].includes(p)) return "astral"; + ["fire","crypto","fx","market","light","ambient","audio","mic","me"].includes(p)) return "astral"; return "physical"; // VII Β· the dense body of the world } @@ -3029,7 +3048,7 @@ if (k === "astro.mars" || p === "fire" || k === "quake.event" || k === "astro.tension" || k === "world.hunger") return "gevurah"; if (/^sky\./.test(k)) return "tiferet"; // the solar heart if (/season|almanac\.(element|fertile)/.test(k)) return "netzach"; // Venus: nature's rhythms - if (["wiki","market","crypto","planes","econ","me"].includes(p) || /retro/.test(k)) return "hod"; // Mercury: mind & number + if (["wiki","market","crypto","fx","planes","econ","me"].includes(p) || /retro/.test(k)) return "hod"; // Mercury: mind & number if (/^astro\.moon|almanac\.dec|ambient|light\./.test(k)) return "yesod"; // Levanah: the astral machinery return "malkut"; // the Kingdom: body & world } @@ -3065,7 +3084,7 @@ if (p === "hand" || p === "tof" || p === "touch") // the candidate at the porch return /\.x$|open/.test(k) ? "boaz" : "jachin"; if (/^sky\.|sun_lon|_lon|zodiac/.test(k) || k === "astro.saturn" || k === "astro.mars") return "east"; // the heavens, read from the East - if (["crypto","market","econ","debt","me"].includes(p)) return "west"; // the wages, paid in the West + if (["crypto","fx","market","econ","debt","me"].includes(p)) return "west"; // the wages, paid in the West if (p === "audio" || /season|almanac|lifeexp|births/.test(k)) return "south"; // Beauty holds music & the arts if (/deaths|hunger|poverty|quake|pm25|aqi/.test(k)) return "north"; // what waits in the dark if (k === "sun.speed" || k === "astro.harmony" || k === "almanac.fertile") return "star"; // Providence @@ -3101,7 +3120,7 @@ if (k === "astro.mars" || p === "fire" || /tension|quake\.event|deaths/.test(k)) return "mars"; if (k === "sun.speed" || /^sky\./.test(k)) return "sun"; if (/season|almanac\.element|centroid/.test(k)) return "venus"; - if (["wiki","market","crypto","planes","iss","me"].includes(p) || /retro/.test(k)) return "mercury"; + if (["wiki","market","crypto","fx","planes","iss","me"].includes(p) || /retro/.test(k)) return "mercury"; return "moon"; // the sublunary world of flux } @@ -3134,7 +3153,7 @@ if (/ambient|light\.|harmony/.test(k)) return "alfheim"; if (p === "wiki" || p === "hand" || p === "planes" || /lifeexp|popvel/.test(k)) return "midgard"; if (/wind|pressure|tension|quake\.event/.test(k)) return "jotunheim"; // the storm-giants - if (["crypto","market","debt","econ","me"].includes(p)) return "nidavellir"; // the dwarves' gold & its curse + if (["crypto","fx","market","debt","econ","me"].includes(p)) return "nidavellir"; // the dwarves' gold & its curse if (p === "fire" || k === "weather.temp") return "muspelheim"; if (/pm25|aqi/.test(k)) return "niflheim"; // the poison-mists if (/deaths|hunger|poverty|quake\.depth/.test(k)) return "helheim"; @@ -3192,7 +3211,7 @@ if (/births|popvel|season|fertile|lifeexp|wind|harmony|almanac\.element/.test(k)) return "wood"; if (p === "fire" || /temp|mars|tension|ambient|light\.|sky\./.test(k) || k === "sun.speed") return "fire"; if (/saturn|quake|hunger|poverty|pressure|debt\.total/.test(k)) return "earth"; - if (["market","planes","audio","me"].includes(p) || /inflation|crypto\.price|debt\.vel/.test(k)) return "metal"; // coin, bells, metal birds + if (["market","fx","planes","audio","me"].includes(p) || /inflation|crypto\.price|debt\.vel/.test(k)) return "metal"; // coin, bells, metal birds return "water"; // the deep: moon, wisdom, the word } @@ -3231,7 +3250,7 @@ if (/quake\.depth|debt\.total|saturn|poverty/.test(k)) return "foundation"; // the deep structure beneath if (/tension|quake\.event|pm25|aqi|inflation|hunger/.test(k)) return "crossing"; // what stands in the way if (/births|popvel|season|fertile|almanac/.test(k)) return "before"; // what approaches - if (["crypto","market","fire","me"].includes(p) || k === "debt.vel") return "hopes"; // money & fire: hope and fear at once + if (["crypto","fx","market","fire","me"].includes(p) || k === "debt.vel") return "hopes"; // money & fire: hope and fear at once if (p === "weather" || p === "planes" || p === "wiki") return "house"; // the world around you, talking if (/lifeexp|astro\.moon|harmony/.test(k)) return "outcome"; // the culminating light return "present"; // what covers you right now @@ -3269,7 +3288,7 @@ function zodiacOf(key) { // by traditional rulership/affinity const k = (key || "").toLowerCase(), p = k.split(".")[0]; if (p === "fire") return "aries"; // Mars burns - if (p === "me" || k === "market.turnover" || k === "crypto.price") return "taurus"; // holdings + if (p === "me" || p === "fx" || k === "market.turnover" || k === "crypto.price") return "taurus"; // holdings & value: the second house if (p === "wiki" || /mercury_retro/.test(k)) return "gemini"; // the word; Mercury's own condition if (p === "weather" || k === "almanac.dec") return "cancer"; // the home sky if (p === "sun" || /^sky\.|light|ambient/.test(k)) return "leo"; // the light itself @@ -3441,7 +3460,7 @@ }); function futharkOf(key) { const k = (key || "").toLowerCase(), p = k.split(".")[0]; - if (p === "me" || k === "crypto.price" || /inflation/.test(k)) return "fehu"; // "wealth is strife among kinsmen" + if (p === "me" || p === "fx" || k === "crypto.price" || /inflation/.test(k)) return "fehu"; // "wealth is strife among kinsmen" if (p === "body" || k === "quake.depth") return "uruz"; // primal strength if (k === "quake.event") return "thurisaz"; // the giant's blow if (p === "wiki") return "ansuz"; // Odin's word @@ -3501,7 +3520,7 @@ if (/births|fertile/.test(k) || p === "natal") return "imbolc"; // first stirrings; your birth if (p === "fire" || ["hand", "tof", "touch", "body", "syn"].includes(p)) return "beltane"; // the fires, the dance, the union if (p === "sun" || /^sky\./.test(k)) return "litha"; // the sun at its height - if (["wiki", "audio", "market", "me", "crypto", "planes", "iss"].includes(p)) return "lughnasadh"; // the great fair + if (["wiki", "audio", "market", "me", "crypto", "fx", "planes", "iss"].includes(p)) return "lughnasadh"; // the great fair return "mabon"; // the reckoning of the year } function drawYearOverlay(t, g) { // the sun walking the year, live @@ -3568,7 +3587,7 @@ if (/harmony|fertile|births|light\.|ambient/.test(k)) return "white"; // silence, pregnant with possibility if (/deaths|debt|poverty/.test(k)) return "black"; // the silence with no future if (p === "body" || p === "fire" || k === "quake.event" || k === "astro.mars") return "red"; // heartbeat & flame: the drum - if (["wiki", "net", "news", "social", "edit", "crypto", "market", "econ", "me"].includes(p) || + if (["wiki", "net", "news", "social", "edit", "crypto", "fx", "market", "econ", "me"].includes(p) || /inflation/.test(k)) return "yellow"; // the trumpet, blown louder and louder if (p === "clock" || p === "audio" || /season|zodiac|hour|lifeexp/.test(k)) return "orange"; // the bell, rung on the hour if (["weather", "air", "planes", "plane"].includes(p) || /temp|wind|pressure|humid|rain|precip/.test(k)) return "green"; // nature at rest @@ -3674,7 +3693,7 @@ if (["wiki", "net", "news", "social", "edit", "audio"].includes(p) || k === "astro.harmony" || p === "syn") return "dui"; // the mouth and tongue; joy if (["weather", "air", "planes", "plane"].includes(p) || /wind|pressure|temp/.test(k)) return "xun"; // what rides the wind if (p === "iss" || p === "space" || p === "natal" || /^sky\.|_lon/.test(k) || p === "astro") return "qian"; // the heavens - if (["crypto", "market", "econ", "me"].includes(p)) return "qian"; // "Qian is jade, is gold" (licence: the markets) + if (["crypto", "fx", "market", "econ", "me"].includes(p)) return "qian"; // "Qian is jade, is gold" (licence: the markets) if (/births|fertile|poverty|hunger|popvel/.test(k) || p === "world" || p === "almanac") return "kun"; // the mother, the multitude, nourishment return "kun"; // the receptive earth holds the rest } @@ -4425,6 +4444,14 @@ const spacing = 46; let totalW = 0; const segs = []; + for (const pk of tickerPins) { // the watchlist β your chosen numbers, live, ahead of the news + const pn = sources.get(pk); + if (!pn) continue; + const label = "π " + (pn.label || pk) + " " + fmtVal(pn.raw); + const w = ctx.measureText(label).width; + segs.push({ label, w, col: pn.col }); + totalW += w + spacing; + } for (const e of eventTicker) { const ago = Math.max(0, Math.floor(now() - e.tAdded)); const label = e.text + " Β· " + ago + "s ago"; @@ -4708,7 +4735,21 @@ const ctxEl = document.createElement("div"); ctxEl.id = "ctxmenu"; document.body.appendChild(ctxEl); - function closeCtx() { ctxEl.classList.remove("open"); ctxEl.innerHTML = ""; } + let ctxLive = null; // the open menu's live-readout timer + function closeCtx() { ctxEl.classList.remove("open"); ctxEl.innerHTML = ""; if (ctxLive) { clearInterval(ctxLive); ctxLive = null; } } + function fmtVal(v) { // raw numbers, human-sized + if (typeof v !== "number" || !isFinite(v)) return "β"; + const a = Math.abs(v); + if (a >= 1e12) return (v / 1e12).toFixed(2) + "T"; + if (a >= 1e9) return (v / 1e9).toFixed(2) + "B"; + if (a >= 1e6) return (v / 1e6).toFixed(2) + "M"; + if (a >= 1000) return Math.round(v).toLocaleString("en-US"); + if (a >= 10) return v.toFixed(1); + return v.toFixed(3); + } + let tickerPins = new Set(); // your watchlist β live values riding the ticker + try { tickerPins = new Set(JSON.parse(localStorage.getItem("gs_pins") || "[]")); } catch (e) {} + function savePins() { try { localStorage.setItem("gs_pins", JSON.stringify([...tickerPins])); } catch (e) {} } window.addEventListener("mousedown", (e) => { if (!ctxEl.contains(e.target)) closeCtx(); }); window.addEventListener("keydown", (e) => { if (e.key === "Escape") closeCtx(); }); @@ -4742,6 +4783,7 @@ function openCtxMenu(node, px, py) { ctxEl.innerHTML = ""; + if (ctxLive) { clearInterval(ctxLive); ctxLive = null; } if (node) { const isSrc = node.kind === "source"; const t0 = tw(node.key); @@ -4751,6 +4793,40 @@ sp.textContent = node.key + (isSrc ? " Β· " + activeSkin.stationOf(node.key) + " " + activeSkin.word : " Β· voice"); hd.appendChild(b); hd.appendChild(sp); ctxEl.appendChild(hd); + if (isSrc) { // the actual data β live value + two minutes of shape + const lv = document.createElement("div"); + lv.style.cssText = "color:#9fd0ff;font-size:11px;padding:2px 12px 0"; + const spark = document.createElement("canvas"); + spark.width = 220; spark.height = 30; + spark.style.cssText = "display:block;margin:2px 12px 4px;width:220px;height:30px"; + const sc = spark.getContext("2d"); + const drawLive = () => { + lv.textContent = "live: " + fmtVal(node.raw) + + (typeof node.norm === "number" ? " Β· " + node.norm.toFixed(2) + " into the matrix" : ""); + const h = node.hist || []; + sc.clearRect(0, 0, spark.width, spark.height); + if (h.length > 1) { + let lo = Infinity, hi = -Infinity; + for (const v of h) { if (v < lo) lo = v; if (v > hi) hi = v; } + const span = hi - lo || 1; + sc.beginPath(); + for (let i = 0; i < h.length; i++) + sc.lineTo(4 + (i / (h.length - 1)) * (spark.width - 8), + spark.height - 4 - ((h[i] - lo) / span) * (spark.height - 8)); + sc.strokeStyle = rgba(node.col, 0.9); sc.lineWidth = 1.4; sc.stroke(); + sc.fillStyle = "rgba(150,170,200,0.6)"; sc.font = "8px 'Helvetica Neue', Arial, sans-serif"; + sc.textAlign = "left"; + sc.fillText(fmtVal(hi), 4, 8); sc.fillText(fmtVal(lo), 4, spark.height - 1); + } else { + sc.fillStyle = "rgba(150,170,200,0.4)"; sc.font = "9px 'Helvetica Neue', Arial, sans-serif"; + sc.fillText("gathering historyβ¦", 6, 18); + } + }; + drawLive(); + ctxLive = setInterval(drawLive, 400); + ctxEl.appendChild(lv); ctxEl.appendChild(spark); + } + ctxItem(node.muted ? "unmute" : "mute", () => toggleMuteLocal(node.key), "click"); if (isSrc) ctxItem(soloKey === node.key ? "unsolo" : "solo", () => { soloKey = soloKey === node.key ? null : node.key; }, "β₯-click"); @@ -4797,6 +4873,8 @@ if (isSrc) ctxItem(selection.has(node.key) ? "deselect (group)" : "select (group)", () => toggleSelect(node.key), "β-click"); ctxItem("open inspector tab", () => openInspector(node), "β§-click"); + if (isSrc) ctxItem(tickerPins.has(node.key) ? "π unpin from the ticker" : "π pin to the ticker", + () => { if (tickerPins.has(node.key)) tickerPins.delete(node.key); else tickerPins.add(node.key); savePins(); }); ctxItem("reset to neutral", () => resetTweaks(node.key), "ββ§-click"); ctxItem("π½ project this node", () => openStage({ type: "node", key: node.key })); const mapSub = ctxSub("β¨ map a key to this"); diff --git a/workers/world_crypto.py b/workers/world_crypto.py index af0e2da..f5d851b 100644 --- a/workers/world_crypto.py +++ b/workers/world_crypto.py @@ -1,53 +1,143 @@ #!/usr/bin/env python3 -"""BTC/USD live trade ticks from Binance β the fastest, most chaotic source. -Every sub-second trade fires the market's pulse; price-to-price velocity is raw -volatility you can hear, driving rhythm and jitter in the modulation matrix. +"""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" -WS_URL = "wss://stream.binance.com:9443/ws/btcusdt@trade" -REST_URL = "https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT" +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 -def emit(client, last_price, price): - """Send price + tick-to-tick velocity. Returns the new last_price.""" - client.send_message("/gs/crypto/price", float(price)) - vel = 0.0 if last_price is None else abs(price - last_price) - client.send_message("/gs/crypto/vel", float(vel)) - return price +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 -async def run_websocket(client): - """Stream trades over the websocket. Returns True if it ever connected - (so the caller keeps favoring the socket), False if it never connected - (so the caller can fall back to REST).""" - import websockets # part of the required stack; import here to keep top clean +def make_emitter(client): + """Per-symbol price+velocity emit, plus the trend kit on the lead.""" + last = {} + kit = TrendKit() - ever_connected = False - last_price = None + 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: - ever_connected = True async for raw in ws: try: - msg = json.loads(raw) - price = float(msg["p"]) + 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 / non-trade frame; skip - last_price = emit(client, last_price, price) - return ever_connected + continue # malformed / unknown frame; skip + emit(loop.time(), symbol, price) -async def poll_rest(client): - """Fallback: poll the REST ticker forever at POLL_S. Never returns.""" - last_price = None +async def poll_rest(emit): + """Fallback: poll the combined REST ticker forever at POLL_S.""" loop = asyncio.get_running_loop() while True: try: @@ -56,27 +146,53 @@ async def poll_rest(client): with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) - data = await loop.run_in_executor(None, fetch) - price = float(data["price"]) - last_price = emit(client, last_price, price) + 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 BTCUSDT trade ticks -> OSC") + 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 - client = SimpleUDPClient(args.host, args.port) - print(f"[{NAME}] emitting /gs/crypto/price,/gs/crypto/vel every <1s (ws) / {POLL_S:g}s (rest)") + 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(client) + await run_websocket(emit) # Clean disconnect (stream ended): reconnect quickly. print(f"[{NAME}] websocket closed; reconnecting") backoff = 1.0 @@ -86,7 +202,7 @@ async def main(): # 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(client), timeout=backoff) + await asyncio.wait_for(poll_rest(emit), timeout=backoff) except asyncio.TimeoutError: pass except Exception as e2: diff --git a/workers/world_fx.py b/workers/world_fx.py new file mode 100644 index 0000000..80b9694 --- /dev/null +++ b/workers/world_fx.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +"""The great currencies as drones β ECB daily reference rates via frankfurter.app. + +The ECB fixes its reference rates once per working day, so these feeds are +glacial: not melodies but tectonics, like the national debt. The pound, the +yen and the aussie drift by pips over hours β patch them somewhere slow (a +drone's weight, a filter's resting point) and hear the currencies lean on +each other. The euro is deliberately absent: fx.eurusd already ticks live +off the Binance tap in world_crypto.py. +""" +import argparse +import json +import time +import urllib.request + +from pythonosc.udp_client import SimpleUDPClient + +NAME = "world_fx" +URL = "https://api.frankfurter.app/latest?from=USD&to=JPY,GBP,AUD" +POLL_S = 900.0 # the fix changes once a day; 15 min is plenty +USER_AGENT = "godstrument/1.0 (+world_fx)" + + +def fetch(): + req = urllib.request.Request(URL, headers={"User-Agent": USER_AGENT}) + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read().decode()) + + +def emit(client, rates): + """USD-based rates -> the conventional market pairs.""" + out = {} + if rates.get("JPY"): + out["fx/usdjpy"] = float(rates["JPY"]) # yen per dollar, as quoted + if rates.get("GBP"): + out["fx/gbpusd"] = 1.0 / float(rates["GBP"]) # cable: dollars per pound + if rates.get("AUD"): + out["fx/audusd"] = 1.0 / float(rates["AUD"]) # dollars per aussie + for k, v in out.items(): + client.send_message("/gs/" + k, v) + return out + + +def main(): + parser = argparse.ArgumentParser(description="ECB daily FX fixes -> OSC") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=9000) + parser.add_argument("--once", action="store_true", help="one fetch+emit, then exit (doubles as the self-check)") + args = parser.parse_args() + client = SimpleUDPClient(args.host, args.port) + print(f"[{NAME}] emitting /gs/fx/usdjpy,/gs/fx/gbpusd,/gs/fx/audusd every {POLL_S:g}s") + + while True: + try: + out = emit(client, fetch().get("rates", {})) + assert all(v > 0 for v in out.values()), f"nonsense rates: {out}" + print(f"[{NAME}] " + " ".join(f"{k.split('/')[1]} {v:.4f}" for k, v in out.items())) + if args.once: + return + except Exception as e: # transient network / parse error β keep going + print(f"[{NAME}] error: {e}; retrying in {POLL_S:g}s") + if args.once: + raise + time.sleep(POLL_S) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + pass