diff --git a/.gitignore b/.gitignore index 5307322c..0253e479 100644 --- a/.gitignore +++ b/.gitignore @@ -48,3 +48,5 @@ research/forward.log research/launchd.log research/.nightly.lock.d research/__pycache__/ +live/parquet/ +live/tape_sync.log diff --git a/HANDOFF.md b/HANDOFF.md index 01da463c..52106cb9 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -52,9 +52,14 @@ push through. 25s — research/params/requote_timing.json; `fak_retry_niche_s` override, `fak_retry_s` fallback+kill-switch; second rejection tags "twice"). - **wwf-recorder**: the FULL firehose (trades + order matches + comments - + crypto ticks, ~8M events/day, dual-socket ~99.9% capture, 25GB volume, - NOTHING deleted until the Mac verifiably ingests it) → nightly → - `live/rtds.duckdb` (trades + aux). Current-era ground truth for research. + + crypto ticks, ~8M events/day, dual-socket ~99.9% capture, 25GB volume). + **Stage-0 warehouse 2026-07-21**: the box folds its own segments → + zstd Parquet partitions on the volume (fold.py sidecar, row-parity + verified, manifest + Mac-ack deletion protocol — invariant STRONGER than + before); the Mac mirrors + appends into `live/rtds.duckdb` every 15 min + (`com.jaxperro.tape-sync` → recorder/sync_tape.py). Tape freshness: + nightly → ~15 min; the box no longer needs the Mac to stay healthy. + `live/parquet/` = complete durable layer (Stage-1 MotherDuck feedstock). - **VALUE experiment: CLOSED 2026-07-19** — sub-2¢ hypothesis refuted (1W/993L, 0.075x); post-mortem in value/PLAN.md; app destroyed. - **research/ (NEW 2026-07-20, SILO — never touches the bots)**: tape-era diff --git a/fly.recorder.toml b/fly.recorder.toml index 3e0a3ae4..6bedefb5 100644 --- a/fly.recorder.toml +++ b/fly.recorder.toml @@ -15,4 +15,4 @@ primary_region = "arn" [[vm]] size = "shared-cpu-1x" - memory = "256mb" + memory = "1gb" # fold sidecar: duckdb capped at 384MB; capture headroom diff --git a/live/daily.sh b/live/daily.sh index 8ad8915c..e202982d 100755 --- a/live/daily.sh +++ b/live/daily.sh @@ -103,8 +103,13 @@ echo "[daily] $(date '+%F %T') edge: parity-era per-signal edge vs fee hurdle -> # The bankroll-decision number (HANDOFF rev 13 / 2026-07-16): one row/day; # the verdict line rides the Discord digest footer below. python3 edge.py || echo "[daily] edge skipped" -echo "[daily] $(date '+%F %T') tape: ingest RTDS segments -> live/rtds.duckdb" -python3 ../recorder/ingest.py || echo "[daily] tape ingest skipped" +echo "[daily] $(date '+%F %T') tape: sync parquet mirror -> live/rtds.duckdb" +# Stage-0 warehouse (2026-07-21): the box folds segments to parquet itself; +# we mirror + append (recorder/sync_tape.py, also every 15 min via +# com.jaxperro.tape-sync). Old bulk ingest.py stays as the fallback for a +# fold-less recorder (it no-ops when fold has already eaten the segments). +python3 -u ../recorder/sync_tape.py || python3 ../recorder/ingest.py \ + || echo "[daily] tape sync skipped" echo "[daily] $(date '+%F %T') 6/7 dashboard" python3 dashboard.py mkdir -p history && cp watch_skilled.json "history/watch_$(date '+%Y%m%d').json" 2>/dev/null diff --git a/recorder/README.md b/recorder/README.md index 61a8ffeb..5716064b 100644 --- a/recorder/README.md +++ b/recorder/README.md @@ -18,21 +18,43 @@ never touch trading, a bot deploy can never gap the tape. 5s of silence is pathological; false trips cost nothing with a twin). - **Segments**: hour-rotated to the volume — `rtds_YYYYMMDD_HH.jsonl.gz` (trades, stable schema) + `aux_…` (everything else, raw payloads). -- **Preservation (user directive 2026-07-19): nothing is deleted until the - Mac has verifiably downloaded it.** Only `ingest.py` deletes, after a - committed transaction. The 25GB volume holds ~5 weeks un-pulled; the - recorder warns from 80% full and only at 95% drops the single oldest - hour per rotation, loudly (`⚠⚠ TAPE LOSS`) — a full disk would otherwise - kill the CURRENT tape too. +- **Preservation (user directive 2026-07-19, STRENGTHENED by Stage 0 + 2026-07-21): nothing is deleted without a verified second copy.** The + fold sidecar deletes a raw gz only after its Parquet re-reads with a + matching row count; the disk guard deletes Parquet only oldest-first and + only files the Mac mirror has ACKED. The recorder's own last-resort + guard (95% -> drop oldest, `⚠⚠ TAPE LOSS`) still protects the live tape + from a full disk. -## The nightly ingest (Mac, daily.sh) +## Stage-0 warehouse (2026-07-21): fold on the box, mirror on the Mac -`recorder/ingest.py`: lists closed segments over `flyctl ssh`, pulls each -(base64 — see issue #8 for the sftp upgrade), inserts into -`live/rtds.duckdb` (`trades` + `aux` tables; own DB file so it can never -fight cache.duckdb's writer lock), marks it in `ingested`, then deletes on -the box. Idempotent and resumable; a crash mid-segment re-ingests cleanly -(single transaction per segment). +- **`fold.py`** (sidecar in the same machine, capture is PID 1): every + 2 min, each closed gz segment becomes + `/data/parquet//date=YYYY-MM-DD/.parquet` (zstd), + row-parity verified, appended to `/data/parquet/manifest.jsonl`, then the + gz is deleted. The volume IS the warehouse: immutable Parquet any client + mirrors incrementally. duckdb capped at 384MB (1GB VM) so a busy-hour + fold can never starve capture. ~250-400MB/day of parquet -> months of + headroom, months more once mirrored+acked files get pruned. +- **`sync_tape.py`** (Mac, every 15 min via `com.jaxperro.tape-sync` + + from daily.sh): pulls new manifest entries over `flyctl sftp`, + row-verifies each file, appends its rows into `live/rtds.duckdb`'s + native tables (research keeps native speed — views-over-parquet was + rejected: per-asset point queries would crawl), records it in + `ingested`, ACKs the box (`/data/parquet/acks/.ok`). Tape freshness + went from "nightly, if the Mac was awake" to ~15 min whenever awake, and + the box no longer depends on the Mac to stay healthy for weeks. +- **`bootstrap_parquet.py`**: one-shot export of the pre-fold history + (2026-07-17..21, existed only in rtds.duckdb) into the mirror — ran + 2026-07-21, parity OK (13.84M trades + 3.07M aux). `live/parquet/` is + the complete durable layer Stage 1 (MotherDuck/ClickHouse) would consume. + +## The legacy bulk ingest (fallback) + +`recorder/ingest.py` (sftp-first, base64 fallback, integrity guards) stays +as the fallback path for a fold-less recorder; with fold running it finds +no gz segments and no-ops. Same idempotence keys (`ingested` by segment +name) as sync_tape, so the two can never double-insert. ## Ops diff --git a/recorder/bootstrap_parquet.py b/recorder/bootstrap_parquet.py new file mode 100644 index 00000000..0e17f14a --- /dev/null +++ b/recorder/bootstrap_parquet.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""One-shot Stage-0 bootstrap: export the history that exists ONLY in +live/rtds.duckdb (segments already deleted on the box) into the parquet +mirror, so live/parquet/ becomes the complete durable warehouse layer from +day one. Partitioned like fold.py's output (date=YYYY-MM-DD, UTC from ts), +one bootstrap_.parquet per family per day. Idempotent: skips files +that already exist. Run once after the last bulk ingest; verify counts.""" +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +LIVE = os.path.join(HERE, "..", "live") +DB = os.path.join(LIVE, "rtds.duckdb") +MIRROR = os.path.join(LIVE, "parquet") + + +def main(): + import duckdb + con = duckdb.connect(DB, read_only=True) + total = {"trades": 0, "aux": 0} + for fam, cols in (("trades", "ts, wallet, asset, cond, side, price, size, tx, title"), + ("aux", "ts, topic, type, payload")): + days = [d for (d,) in con.execute( + f"SELECT DISTINCT strftime(to_timestamp(ts), '%Y-%m-%d') FROM {fam} " + f"WHERE ts IS NOT NULL ORDER BY 1").fetchall()] + for day in days: + outdir = os.path.join(MIRROR, fam, f"date={day}") + os.makedirs(outdir, exist_ok=True) + out = os.path.join(outdir, f"bootstrap_{day.replace('-', '')}.parquet") + if os.path.exists(out): + print(f"[bootstrap] {fam} {day}: exists — skip") + continue + con.execute(f"""COPY (SELECT {cols} FROM {fam} + WHERE strftime(to_timestamp(ts), '%Y-%m-%d') = '{day}') + TO '{out}' (FORMAT PARQUET, COMPRESSION ZSTD)""") + n, = con.execute( + f"SELECT count(*) FROM read_parquet('{out}')").fetchone() + total[fam] += n + print(f"[bootstrap] {fam} {day}: {n:,} rows " + f"({os.path.getsize(out)//2**20}MB)") + for fam in ("trades", "aux"): + db_n, = con.execute(f"SELECT count(*) FROM {fam}").fetchone() + pq_n, = con.execute(f"""SELECT count(*) FROM + read_parquet('{os.path.join(MIRROR, fam)}/*/*.parquet')""").fetchone() + ok = "OK" if db_n == pq_n else "MISMATCH" + print(f"[bootstrap] parity {fam}: db {db_n:,} vs mirror {pq_n:,} — {ok}") + if db_n != pq_n: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/recorder/fly.Dockerfile b/recorder/fly.Dockerfile index 97646075..03594891 100644 --- a/recorder/fly.Dockerfile +++ b/recorder/fly.Dockerfile @@ -1,7 +1,11 @@ -# wwf-recorder — RTDS tape silo. One pip dep (websocket-client), no git, no -# repo clone: the recorder ships its code in the image (a tape must not -# depend on GitHub being up to boot). +# wwf-recorder — RTDS tape silo. Two pip deps (websocket-client for capture, +# duckdb for the parquet fold sidecar), no git, no repo clone: the recorder +# ships its code in the image (a tape must not depend on GitHub being up to +# boot). FROM python:3.12-slim -RUN pip install --no-cache-dir websocket-client +RUN pip install --no-cache-dir websocket-client duckdb COPY recorder/recorder.py /recorder.py -CMD ["python3", "-u", "/recorder.py"] +COPY recorder/fold.py /fold.py +COPY recorder/start.sh /start.sh +RUN chmod +x /start.sh +CMD ["/start.sh"] diff --git a/recorder/fold.py b/recorder/fold.py new file mode 100644 index 00000000..83cae4b5 --- /dev/null +++ b/recorder/fold.py @@ -0,0 +1,147 @@ +#!/usr/bin/env python3 +"""Stage-0 warehouse fold — runs ON the recorder box beside recorder.py. + +Every FOLD_S seconds: each closed gz segment becomes a zstd Parquet file +under /data/parquet//date=YYYY-MM-DD/.parquet, row-parity +verified, manifest-logged — and only then is the raw gz deleted. This moves +the deletion authority that used to live in the Mac's nightly ingest onto +the box, and makes the volume itself the warehouse: immutable Parquet +partitions any client can mirror incrementally (recorder/sync_tape.py). + +The capture invariant gets STRONGER, not weaker: + raw gz deleted only after its Parquet row count matches on re-read; + parquet deleted only by the disk guard, only oldest-first, and only + files the Mac mirror has ACKED (/data/parquet/acks/.ok) — + plus the recorder's own last-resort >95% guard still protects + the live tape. + +Crash-safe by re-entry: parquet-without-manifest refolds (overwrite); +manifest-without-rm re-deletes the leftover gz. duckdb is capped at 384MB +so a busy-hour fold can never starve the capture process (1GB VM). +Never touches the current (un-gzipped) hour. Never imports bot code. +""" +import json +import os +import time + +import duckdb + +SEG = os.environ.get("TAPE_DIR", "/data/segments") +PQ = os.environ.get("PARQUET_DIR", "/data/parquet") +ACKS = os.path.join(PQ, "acks") +MANIFEST = os.path.join(PQ, "manifest.jsonl") +FOLD_S = 120 +TRADE_COLS = ("{ts:'DOUBLE', wallet:'VARCHAR', asset:'VARCHAR', " + "cond:'VARCHAR', side:'VARCHAR', price:'DOUBLE', " + "size:'DOUBLE', tx:'VARCHAR', title:'VARCHAR'}") +AUX_COLS = ("{ts:'DOUBLE', topic:'VARCHAR', type:'VARCHAR', " + "payload:'VARCHAR'}") # VARCHAR = json text, matches rtds.duckdb + + +def log(m): + print(f"{time.strftime('%H:%M:%S')} [fold] {m}", flush=True) + + +def manifest_names(): + names = set() + try: + with open(MANIFEST) as fh: + for ln in fh: + try: + names.add(json.loads(ln)["segment"]) + except Exception: + pass + except FileNotFoundError: + pass + return names + + +def seg_date(name): + # rtds_YYYYMMDD_HH.jsonl.gz / aux_YYYYMMDD_HH.jsonl.gz -> YYYY-MM-DD + stamp = name.split("_", 1)[1][:8] + return f"{stamp[:4]}-{stamp[4:6]}-{stamp[6:8]}" + + +def fold_one(db, name): + fam = "aux" if name.startswith("aux_") else "trades" + cols = AUX_COLS if fam == "aux" else TRADE_COLS + src = os.path.join(SEG, name) + outdir = os.path.join(PQ, fam, f"date={seg_date(name)}") + os.makedirs(outdir, exist_ok=True) + out = os.path.join(outdir, name.replace(".jsonl.gz", ".parquet")) + rd = (f"read_json('{src}', format='newline_delimited', " + f"columns={cols}, ignore_errors=true)") + n_src, = db.execute(f"SELECT count(*) FROM {rd}").fetchone() + db.execute(f"COPY (SELECT * FROM {rd}) TO '{out}' " + f"(FORMAT PARQUET, COMPRESSION ZSTD)") + n_pq, = db.execute(f"SELECT count(*) FROM read_parquet('{out}')").fetchone() + if n_pq != n_src: + os.remove(out) + raise RuntimeError(f"parity {n_src} src != {n_pq} parquet") + with open(MANIFEST, "a") as fh: + fh.write(json.dumps({"segment": name, "family": fam, + "path": os.path.relpath(out, PQ), "rows": n_src, + "bytes": os.path.getsize(out), + "folded_at": int(time.time())}) + "\n") + fh.flush() + os.fsync(fh.fileno()) + os.remove(src) + log(f"{name} -> {os.path.basename(out)} ({n_src} rows, " + f"{os.path.getsize(out)//1024}KB)") + + +def disk_guard(): + """Volume filling: free the oldest ACKED parquet only. Unacked parquet + is never touched here — the Mac mirror is the second copy.""" + try: + st = os.statvfs(PQ) + free = st.f_bavail / st.f_blocks + if free >= 0.15: + return + acked = {f[:-3] for f in os.listdir(ACKS)} if os.path.isdir(ACKS) else set() + cand = [] + for root, _, files in os.walk(PQ): + for f in files: + if f.endswith(".parquet") and f in acked: + cand.append(os.path.join(root, f)) + cand.sort() # name-sorted = oldest first + if cand: + os.remove(cand[0]) + log(f"disk guard: volume {100*(1-free):.0f}% full — dropped " + f"acked {os.path.basename(cand[0])}") + else: + log(f"⚠ volume {100*(1-free):.0f}% full and NO acked parquet to " + "drop — is the Mac mirror running?") + except Exception as e: + log(f"disk guard error: {e}") + + +def main(): + os.makedirs(PQ, exist_ok=True) + os.makedirs(ACKS, exist_ok=True) + db = duckdb.connect() + db.execute("SET memory_limit='384MB'") + db.execute("SET threads=1") + log(f"folding {SEG} -> {PQ} every {FOLD_S}s") + while True: + try: + done = manifest_names() + for name in sorted(os.listdir(SEG)): + if not name.endswith(".gz"): + continue # current hour is plain .jsonl — skip + if name in done: # crash between manifest and rm + os.remove(os.path.join(SEG, name)) + log(f"{name}: already folded — removed leftover gz") + continue + try: + fold_one(db, name) + except Exception as e: + log(f"⚠ {name}: fold failed ({e}) — gz left for retry") + disk_guard() + except Exception as e: + log(f"⚠ loop error: {e}") + time.sleep(FOLD_S) + + +if __name__ == "__main__": + main() diff --git a/recorder/start.sh b/recorder/start.sh new file mode 100644 index 00000000..a8530d97 --- /dev/null +++ b/recorder/start.sh @@ -0,0 +1,6 @@ +#!/bin/sh +# wwf-recorder entrypoint: capture is PID 1 (its death restarts the machine, +# policy=always); the fold sidecar runs beside it and can die without ever +# touching capture (its own retry loop; refolds on next boot regardless). +python3 -u /fold.py & +exec python3 -u /recorder.py diff --git a/recorder/sync_tape.py b/recorder/sync_tape.py new file mode 100644 index 00000000..b058d2b1 --- /dev/null +++ b/recorder/sync_tape.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +"""Parquet mirror sync (Stage-0 warehouse, replaces the nightly bulk ingest). + +Pulls NEW parquet files the recorder's fold sidecar produced (/data/parquet +on the wwf-recorder volume) into live/parquet/, row-verifies each against +the box manifest, appends its rows into live/rtds.duckdb's NATIVE tables +(same tables as ever — research and tape_sharps keep native-table speed; +views-over-parquet was rejected: sim's per-asset point queries would turn +minutes into hours), marks it in `ingested`, and ACKs it back (touch +/data/parquet/acks/.ok — the fold disk-guard may only ever delete +ACKED parquet). The mirror itself is the durable warehouse layer Stage 1 +(MotherDuck/ClickHouse) would consume. + +Transport is the battle-tested flyctl sftp get (+ ssh console for acks); +no ingress on the recorder, by design. Quiet no-op when the box is +unreachable or another process holds the db write lock — files are +immutable and manifest-driven, so the next run catches up. Runs every 15 +min via com.jaxperro.tape-sync and from daily.sh (old bulk ingest.py kept +as fallback for a fold-less recorder). +""" +import json +import os +import shutil +import subprocess +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +LIVE = os.path.join(HERE, "..", "live") +MIRROR = os.path.join(LIVE, "parquet") +LOCAL_MANIFEST = os.path.join(MIRROR, ".mirror_manifest.jsonl") +DB = os.path.join(LIVE, "rtds.duckdb") +APP, PQ = "wwf-recorder", "/data/parquet" +FLYCTL = shutil.which("flyctl") or "/opt/homebrew/bin/flyctl" + + +def box(cmd, timeout=120): + r = subprocess.run([FLYCTL, "ssh", "console", "-a", APP, "-C", + f"bash -c '{cmd}'"], capture_output=True, text=True, + timeout=timeout, stdin=subprocess.DEVNULL) + return r.stdout + + +def sftp_get(remote, local, timeout=900): + subprocess.run([FLYCTL, "ssh", "sftp", "get", remote, local, "-a", APP], + capture_output=True, timeout=timeout, + stdin=subprocess.DEVNULL) + return os.path.exists(local) and os.path.getsize(local) > 0 + + +def load_manifest(path): + out = {} + try: + with open(path) as fh: + for ln in fh: + try: + d = json.loads(ln) + out[d["segment"]] = d + except Exception: + pass + except FileNotFoundError: + pass + return out + + +def connect_rw(duckdb, tries=3, wait=10): + """The 15-min cadence can collide with a long research read (one-writer- + OR-many-readers). Retry briefly, then yield to the next run.""" + for i in range(tries): + try: + return duckdb.connect(DB) + except Exception as e: + if i == tries - 1: + print(f"[sync] db locked ({str(e)[:60]}) — yielding to next run") + return None + time.sleep(wait) + + +def main(): + os.makedirs(MIRROR, exist_ok=True) + tmp_man = os.path.join(MIRROR, ".remote_manifest.tmp") + if not sftp_get(f"{PQ}/manifest.jsonl", tmp_man, timeout=120): + print("[sync] box unreachable or no manifest yet — nothing to do") + return 0 + remote = load_manifest(tmp_man) + local = load_manifest(LOCAL_MANIFEST) + new = [d for s, d in sorted(remote.items()) if s not in local] + os.remove(tmp_man) + if not new: + print(f"[sync] mirror current ({len(local)} files)") + return 0 + import duckdb + con = connect_rw(duckdb) + if con is None: + return 0 + have = {r[0] for r in con.execute("SELECT segment FROM ingested").fetchall()} + got = 0 + for d in new: + rel, seg = d["path"], d["segment"] + dst = os.path.join(MIRROR, rel) + os.makedirs(os.path.dirname(dst), exist_ok=True) + if not os.path.exists(dst) and not sftp_get(f"{PQ}/{rel}", dst): + print(f"[sync] {rel}: fetch failed — retry next run") + continue + try: + n, = con.execute( + f"SELECT count(*) FROM read_parquet('{dst}')").fetchone() + except Exception as e: + print(f"[sync] {rel}: unreadable ({e}) — dropped, retry next run") + try: + os.remove(dst) + except OSError: + pass + continue + if n != d["rows"]: + print(f"[sync] {rel}: rows {n} != manifest {d['rows']} — dropped") + os.remove(dst) + continue + # append into the native tables exactly once (segment-keyed), then + # mirror-manifest + ack. A crash between COMMIT and manifest write + # re-runs into the `have` guard — no double insert. + if seg not in have: + con.execute("BEGIN") + if d["family"] == "aux": + con.execute(f"""INSERT INTO aux + SELECT ts, topic, type, payload + FROM read_parquet('{dst}')""") + else: + con.execute(f"""INSERT INTO trades + SELECT ts, wallet, asset, cond, side, price, size, tx, title + FROM read_parquet('{dst}')""") + con.execute("INSERT INTO ingested VALUES (?,?,?)", + [seg, d["rows"], int(time.time())]) + con.execute("COMMIT") + with open(LOCAL_MANIFEST, "a") as fh: + fh.write(json.dumps(d) + "\n") + box(f"touch {PQ}/acks/{os.path.basename(rel)}.ok") + got += 1 + print(f"[sync] {rel}: {n} rows -> db + mirror + ack") + n, = con.execute("SELECT count(*) FROM trades").fetchone() + print(f"[sync] +{got}/{len(new)} files · rtds.duckdb now {n:,} trades") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/research/tape.py b/research/tape.py index 032c5a09..63a35fde 100644 --- a/research/tape.py +++ b/research/tape.py @@ -40,8 +40,17 @@ SYMBOLS = {"bitcoin": "btcusdt", "btc": "btcusdt", "dogecoin": "dogeusdt"} -def connect(): - return duckdb.connect(RTDS, read_only=True) +def connect(tries=4, wait=5): + """read_only, with a brief retry: the 15-min tape-sync (Stage 0, + 2026-07-21) holds the write lock for a few seconds per run.""" + import time + for i in range(tries): + try: + return duckdb.connect(RTDS, read_only=True) + except duckdb.IOException: + if i == tries - 1: + raise + time.sleep(wait) # ── tape proxy-resolution ───────────────────────────────────────────────────