aa81535379
Right-click any source: its actual number, live, with a two-minute
sparkline right in the menu — and 📌 pin it to the ticker, where your
watchlist scrolls ahead of the news with live values. The Binance tap
widens to ETH, SOL and the euro itself (EUR/USD on a 1s miniTicker,
since the pair trades thin), the ECB's daily fixes arrive as glacial
drones (GBP/USD, USD/JPY, AUD/USD), and Bitcoin speaks its own
technical analysis on 5s closes: the MACD line, Wilder's RSI(14), and
cross bells that ring once at every golden and death cross. All of it
routed through the Spirits like any money — to the solar plexus, to
Hod, to Nidavellir, to fehu.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
#!/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
|