Godstrument/workers/world_dealgod.py

"""
world_dealgod.py — your own market as an instrument. The dealgod feed.

Reads price history straight out of the basegod warehouse (dealgod.price_events,
~4M rows) on the M1 Ultra over the tailnet, bins it into an activity curve and an
average-price curve, and replays them compressed on a loop — your deal-flow as a
control source, warpable and jog-scrubbable like the other time-warp workers.

READ-ONLY: it issues a single SELECT (server-side GROUP BY does the work) over
SSH as johnking. No writes ever touch the master. Point --ssh at a host with
dashboard_ro if you want strict least-privilege networked access instead.

Emits:
  /gs/market/velocity   normalized market activity (price events per time bin)
  /gs/market/turnover   normalized average price of what's moving
"""

from __future__ import annotations
import argparse
import os
import subprocess
import sys
import threading
import time

from pythonosc.udp_client import SimpleUDPClient
from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import ThreadingOSCUDPServer

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from timewarp import ReplayBuffer, Jog

PSQL = "/opt/homebrew/opt/postgresql@16/bin/psql"

# quote-free SQL (make_interval / to_timestamp avoid single quotes over ssh)
QUERY = (
    "SELECT extract(epoch from bucket), cnt, avgp FROM ("
    "SELECT date_bin(make_interval(mins=>{binmin}), scraped_at, to_timestamp(0)) AS bucket, "
    "count(*) AS cnt, avg(price) FILTER (WHERE price BETWEEN 0 AND 100000) AS avgp "
    "FROM price_events WHERE scraped_at > now() - make_interval(days => {days}) "
    "GROUP BY bucket ORDER BY bucket) s"
)


def fetch(ssh_host, days, binmin):
    sql = QUERY.format(days=days, binmin=binmin)
    remote = f'{PSQL} -d dealgod -tA -F"|" -c "{sql}"'
    r = subprocess.run(
        ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", ssh_host, remote],
        capture_output=True, text=True, timeout=45)
    rows = []
    for line in r.stdout.splitlines():
        p = line.split("|")
        if len(p) >= 3 and p[0]:
            try:
                rows.append((float(p[0]), float(p[1]), float(p[2] or 0)))
            except ValueError:
                pass
    return rows, r.stderr.strip()


def normalize(vals):
    lo, hi = min(vals), max(vals)
    return [((v - lo) / (hi - lo) if hi > lo else 0.5) for v in vals]


def build(rows, into):
    times = [r[0] for r in rows]
    vel = ReplayBuffer("continuous", into)
    vel.load_continuous(times, normalize([r[1] for r in rows]))
    turn = ReplayBuffer("continuous", into)
    turn.load_continuous(times, normalize([r[2] for r in rows]))
    return vel, turn


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--jog-port", type=int, default=9104)
    ap.add_argument("--ssh", default="johnking@100.91.239.7")
    ap.add_argument("--days", type=int, default=7)
    ap.add_argument("--binmin", type=int, default=15, help="minutes per bin")
    ap.add_argument("--into", type=float, default=300.0, help="loop length (s)")
    ap.add_argument("--refresh", type=float, default=900.0, help="re-query every Ns")
    args = ap.parse_args()

    client = SimpleUDPClient(args.host, args.port)
    jog = Jog()
    state = {"vel": None, "turn": None}

    def load():
        try:
            rows, err = fetch(args.ssh, args.days, args.binmin)
        except Exception as e:   # SSH timeout, missing ssh binary, tailnet stall
            print(f"[dealgod] query failed ({e}); retrying")
            return
        if len(rows) >= 2:
            vel, turn = build(rows, args.into)
            state["vel"], state["turn"] = vel, turn
            span = (rows[-1][0] - rows[0][0]) / 86400
            print(f"[dealgod] {len(rows)} bins over {span:.1f}d "
                  f"-> loop {args.into:g}s  (jog udp/{args.jog_port})")
        else:
            print(f"[dealgod] no data ({err or 'empty'}); retrying")

    load()
    while state["vel"] is None:
        time.sleep(10)
        load()

    disp = Dispatcher()
    disp.map("/jog/rate", jog.on_rate)
    disp.map("/jog/nudge", jog.on_nudge)
    srv = ThreadingOSCUDPServer(("127.0.0.1", args.jog_port), disp)
    threading.Thread(target=srv.serve_forever, daemon=True).start()

    last_refresh = time.time()
    period = 1.0 / 30.0
    while True:
        v, t = state["vel"], state["turn"]
        if v and t:
            v.tick(period, jog.rate_mult, jog.nudge)
            t.tick(period, jog.rate_mult, jog.nudge)
            client.send_message("/gs/market/velocity", float(v.current()))
            client.send_message("/gs/market/turnover", float(t.current()))
        if time.time() - last_refresh > args.refresh:
            last_refresh = time.time()
            threading.Thread(target=load, daemon=True).start()
        time.sleep(period)


if __name__ == "__main__":
    main()