"""market_tension_injector.py — feed REAL market tensions into the lattice.

Replaces the synthetic z-score/STR_CAP scheme of inject_continuous_stream.py.

The previous injectors took market variables, z-scored them away to a
dimensionless ±0.30 blob at (512,512), and asked the lattice to find
structure. That destroyed all native magnitude and unit information. A
5σ flow and a 3σ flow produced identical injections. Predictably, six
"different variables" collapsed to one transfer function.

This injector encodes four NATIVE market tensions, in physical units,
as DISTINCT spatial forcing patterns the lattice can mechanically
respond to differently:

  ┌──────────────────────┬────────────────────────┬─────────────────────────┐
  │ tension              │ physical interpretation│ lattice forcing pattern │
  ├──────────────────────┼────────────────────────┼─────────────────────────┤
  │ funding_bps          │ directional cost-to-   │ x-axis DIPOLE           │
  │   (signed bps annual)│ carry, longs pulled vs │ (asymmetry forcing)     │
  │                      │ shorts                 │                         │
  │                      │                        │                         │
  │ bs_ratio_signed      │ flow directional       │ x-axis OFFSET blob      │
  │   (-0.5..+0.5)       │ imbalance THIS minute  │ (velocity bias)         │
  │                      │                        │                         │
  │ activity_excess      │ excitation relative to │ ISOTROPIC center blob   │
  │   (>= -1, unbounded) │ own 500-min baseline   │ (density / energy)      │
  │                      │                        │                         │
  │ cvd_divergence       │ coiled spring — flow   │ VORTICITY RING          │
  │   ($ / bps signed)   │ without price response │ (rotational forcing)    │
  └──────────────────────┴────────────────────────┴─────────────────────────┘

SCALING DISCIPLINE
  No z-scoring. No rolling normalisation. Each tension is scaled to lattice
  units by a FIXED PHYSICAL CONSTANT documented in MAP_SCALE below. Cross-
  regime amplitude information is preserved: a 100bps funding day produces
  exactly 2× the dipole strength of a 50bps day. A panic flow imbalance
  produces 5× the velocity bias of a calm one. The lattice sees true
  market amplitude.

ARMS
  A — control. No injection. Fresh baseline of the lattice's natural
      variability over the experiment window. (Re-captured even though
      we have an old arm A — field state may have drifted.)
  T — tension. All four patterns injected per minute, each from its native
      market value via fixed physical scale.

PROTOCOL
  - Run arm A first (n minutes), save parquet
  - Cooldown 20 min (let field settle to a new baseline)
  - Run arm T (same n minutes), save parquet
  - Analyzer compares: does field state under arm T predict realized
    vol / direction at h=5,15,60 BETTER than (a) arm A field, and (b) the
    raw tensions themselves?

REQUIREMENTS
  - Lattice daemon alive on 5556 (telemetry PUB) / 5557 (cmd SUB).
  - Must be launched via WSL setsid nohup (proven pattern from contstream).
  - Tag: --agent-owner=PAPERTRADER per AGENTS.md ledger (this is research,
    PaperTrader scope; lattice daemon itself is RESONANCE, untouched).

Output: /mnt/d/Resonance_Engine/traj/tension_<RUN_ID>/
    meta.json
    funding_history.parquet           (raw HL fundingHistory cached)
    arm_A_no_inject.parquet
    arm_T_tension.parquet
    progress.log
"""
from __future__ import annotations
import argparse, glob, json, math, threading, time, urllib.request
from pathlib import Path

import numpy as np
import pandas as pd
import zmq

# ───────────────────────── config ─────────────────────────
DATA_ROOT  = "/mnt/d/PaperTrader/research/hl_data/minutes"
COIN       = "BTC"

# Default to a single day for the first run. Override with --days-glob.
DEFAULT_DAYS_GLOB = "20260415*"

# fixed physical scales — never refit, never z-score
# (the whole point of this injector)
MAP_SCALE = {
    # funding_rate from HL is per-hour. Annualised bps = rate * 8760 * 10000.
    # 100bps annualised → 1.0 dipole strength leg.
    # Cap at ±2.0 leg strength (i.e. ±200bps annualised funding).
    "funding_bps_per_strength": 100.0,
    "funding_strength_cap":     2.0,

    # bs_ratio_signed ranges natively ±0.5 (since ratio is 0..1, minus 0.5).
    # Scale ×2 so full ±0.5 → ±1.0 strength.
    "bs_ratio_per_strength":    0.5,
    "bs_ratio_strength_cap":    1.0,

    # activity_excess = (trade_count / rolling_500min_mean) - 1.
    # Centred at 0. -1 = zero activity. +2 = 3× baseline. +5 = 6× baseline.
    # Scale: 3.0 of excess → 1.0 strength. Cap at 5.
    "activity_per_strength":    3.0,
    "activity_strength_cap":    5.0,

    # cvd_divergence = sum(signed_flow_usd, 60min) / (|price_change_60min_bps| + eps)
    # Units: USD per bps. April BTC: median |val|=$566k, 90th=$2.5M, 99th=$13M, max=$100M.
    # Tanh scale 2e6 keeps the bulk (q50 strength 0.28, q90 0.85) in the responsive
    # part of the curve, while ~8% of extreme minutes saturate near ±1.0.
    # Calibrated 2026-06-07 on 43k April minutes.
    "cvd_div_tanh_scale":       2e6,
}

# ROLLING WINDOWS (computed per source minute, past-only by shift)
ROLL_ACTIVITY_WIN = 500   # 500-min rolling baseline for activity_excess
ROLL_CVD_WIN      = 60    # 60-min summed CVD and price change
EPS_BPS           = 0.5   # epsilon to prevent /0 in cvd_div denominator

# Lattice spatial geometry (lattice is 1024×1024, center 512,512)
CENTER     = (512.0, 512.0)
DIPOLE_OFF = 64.0    # dipole leg offset from center
OFFSET_OFF = 64.0    # velocity-bias single-blob offset from center
RING_R     = 96.0    # vorticity ring radius
SIGMA      = 32.0    # gaussian sigma of each injection blob

# Vorticity ring: N legs around the ring, alternating signs to create curl.
# 4 legs is the minimum for a clean dipole-quadrupole rotation. 6 is smoother.
VORTICITY_N_LEGS = 6

# Timing
PER_MINUTE_MS         = 100     # 10× realtime (proven in throughput probe)
WAIT_AFTER_INJECT_MS  = 80      # extra room for the 4 force-patterns to settle
COOLDOWN_BETWEEN_ARMS_S = 1200  # 20 min, same as contstream

# ZMQ
TEL_ADDR = "tcp://127.0.0.1:5556"
CMD_ADDR = "tcp://127.0.0.1:5557"

CHANNELS = ["asymmetry", "coherence", "stress_xx", "stress_yy", "stress_xy",
            "vorticity_mean", "vel_mean", "vel_max", "vel_var"]

# Funding fetch
HL_INFO_URL = "https://api.hyperliquid.xyz/info"


# ───────────────────────── logging ─────────────────────────
RUN_ID  = time.strftime("%Y%m%dT%H%M%S")
OUT_DIR = Path(f"/mnt/d/Resonance_Engine/traj/tension_{RUN_ID}")
PROGRESS = OUT_DIR / "progress.log"


def log(msg: str) -> None:
    line = f"[{time.strftime('%Y-%m-%dT%H:%M:%S')}] {msg}"
    print(line, flush=True)
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    with PROGRESS.open("a") as f:
        f.write(line + "\n")


# ───────────────────────── ZMQ telemetry subscriber ─────────────────────────
class LatestTel:
    """Background thread holding the most-recent telemetry frame only."""

    def __init__(self, addr: str):
        self.ctx = zmq.Context.instance()
        self.sock = self.ctx.socket(zmq.SUB)
        self.sock.connect(addr)
        self.sock.setsockopt(zmq.SUBSCRIBE, b"")
        self.sock.setsockopt(zmq.RCVHWM, 2000)
        self.latest: dict | None = None
        self.latest_wall: float = 0.0
        self.n_seen = 0
        self._stop = threading.Event()
        self._t = threading.Thread(target=self._run, daemon=True)
        self._t.start()

    def _run(self):
        while not self._stop.is_set():
            if self.sock.poll(100):
                try:
                    raw = self.sock.recv_string(zmq.NOBLOCK)
                    self.latest = json.loads(raw)
                    self.latest_wall = time.time()
                    self.n_seen += 1
                except Exception:
                    pass

    def snapshot(self) -> tuple[dict | None, float]:
        return self.latest, self.latest_wall

    def stop(self):
        self._stop.set()
        self._t.join(timeout=2)
        try:
            self.sock.close(0)
        except Exception:
            pass


# ───────────────────────── data loading ─────────────────────────
def load_btc(days_glob: str) -> pd.DataFrame:
    day_dirs = sorted(glob.glob(f"{DATA_ROOT}/{days_glob}"))
    if not day_dirs:
        raise RuntimeError(f"no day dirs matching {days_glob} in {DATA_ROOT}")
    log(f"loading {len(day_dirs)} day dirs ({day_dirs[0].split('/')[-1]} .. {day_dirs[-1].split('/')[-1]})")
    dfs = []
    for d in day_dirs:
        for f in sorted(glob.glob(f"{d}/*.parquet")):
            dfs.append(pd.read_parquet(f))
    df = pd.concat(dfs, ignore_index=True)
    df = df[df.coin == COIN].sort_values("minute").drop_duplicates("minute").reset_index(drop=True)
    log(f"loaded {len(df)} unique minutes of {COIN}")
    return df


def fetch_funding_history(coin: str, start_min: int, end_min: int,
                           cache: Path) -> pd.DataFrame:
    """Pull HL fundingHistory for coin between [start_min, end_min] (minutes-since-epoch).
    Returns DataFrame[minute_floor_hour, funding_bps_annual] joined-ready.
    Caches raw JSON in `cache`."""
    if cache.exists():
        log(f"using cached funding: {cache}")
        return pd.read_parquet(cache)

    start_ms = int(start_min) * 60 * 1000
    end_ms   = int(end_min)   * 60 * 1000
    log(f"fetching HL fundingHistory for {coin} {start_ms} -> {end_ms}")
    rows_all = []
    cursor = start_ms
    while cursor < end_ms:
        body = json.dumps({"type": "fundingHistory", "coin": coin,
                           "startTime": cursor, "endTime": end_ms}).encode()
        req = urllib.request.Request(HL_INFO_URL, data=body,
                                     headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(req, timeout=30) as r:
            chunk = json.loads(r.read().decode())
        if not chunk:
            break
        rows_all.extend(chunk)
        last_ts = int(chunk[-1].get("time", 0))
        if last_ts <= cursor:
            break
        cursor = last_ts + 1
        time.sleep(0.15)
    if not rows_all:
        log("WARNING: empty fundingHistory response")
        return pd.DataFrame(columns=["minute_hour", "funding_bps_annual"])
    df = pd.DataFrame(rows_all)
    df["minute_hour"] = (df["time"].astype("int64") // (1000 * 60))  # to minute-since-epoch
    df["funding_rate_per_hour"] = df["fundingRate"].astype(float)
    # annualized bps: per-hour rate × 8760 hours × 10000 bps
    df["funding_bps_annual"] = df["funding_rate_per_hour"] * 8760 * 10000.0
    df = df[["minute_hour", "funding_bps_annual"]].drop_duplicates("minute_hour").sort_values("minute_hour")
    cache.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(cache)
    log(f"funding rows: {len(df)}   "
        f"mean={df.funding_bps_annual.mean():+.2f}bps  "
        f"std={df.funding_bps_annual.std():+.2f}bps  "
        f"range={df.funding_bps_annual.min():+.2f}..{df.funding_bps_annual.max():+.2f}")
    return df


def compute_tensions(df: pd.DataFrame, funding: pd.DataFrame) -> pd.DataFrame:
    """Per minute, compute the four native tensions in physical units."""
    out = df.copy()

    # --- funding: forward-fill the hourly value onto each minute ---
    # join via the floor-of-hour for each minute
    out["minute_hour_floor"] = (out["minute"] // 60) * 60
    out = out.merge(funding.rename(columns={"minute_hour": "minute_hour_floor"}),
                    on="minute_hour_floor", how="left")
    out["funding_bps"] = out["funding_bps_annual"].ffill()

    # --- buy_sell ratio signed ---
    denom = (out["taker_buy_usd"] + out["taker_sell_usd"]).replace(0, np.nan)
    out["bs_ratio_signed"] = (out["taker_buy_usd"] / denom) - 0.5  # -0.5..+0.5

    # --- activity_excess: trade_count / rolling-500min-mean - 1 ---
    # past-only via shift(1)
    rolling_mean = out["trade_count"].astype(float).shift(1).rolling(
        window=ROLL_ACTIVITY_WIN, min_periods=50
    ).mean()
    out["activity_excess"] = (out["trade_count"].astype(float) / rolling_mean) - 1.0

    # --- cvd_divergence: 60min summed signed_flow / |60min bps price change| ---
    # past-only: window ends at minute t-1, then we ascribe to minute t.
    sflow = out["signed_flow_usd"].astype(float)
    cvd_60 = sflow.shift(1).rolling(window=ROLL_CVD_WIN, min_periods=10).sum()
    px = out["mid_price"].astype(float)
    # |price_change_60min| in bps, looking back from t-1 -> t-1-60
    px_now  = px.shift(1)
    px_then = px.shift(1 + ROLL_CVD_WIN)
    pct_chg_bps = ((px_now - px_then) / px_then * 10000.0).abs()
    out["cvd_divergence"] = cvd_60 / (pct_chg_bps + EPS_BPS)

    return out


# ───────────────────────── injection encoders ─────────────────────────
def _send_inject(pub: zmq.Socket, x: float, y: float,
                 strength: float, sigma: float = SIGMA) -> None:
    payload = {"cmd": "inject_density",
               "x": float(x), "y": float(y),
               "sigma": float(sigma),
               "strength": float(strength)}
    pub.send_string(json.dumps(payload))


def encode_funding_dipole(pub: zmq.Socket, funding_bps: float) -> None:
    """Funding → x-axis dipole. Positive funding (longs paying) pushes +x,
    negative funding (shorts paying) pushes -x."""
    if not np.isfinite(funding_bps):
        return
    strength = funding_bps / MAP_SCALE["funding_bps_per_strength"]
    strength = float(np.clip(strength, -MAP_SCALE["funding_strength_cap"],
                                       +MAP_SCALE["funding_strength_cap"]))
    if abs(strength) < 1e-6:
        return
    cx, cy = CENTER
    _send_inject(pub, cx - DIPOLE_OFF, cy, -strength)
    _send_inject(pub, cx + DIPOLE_OFF, cy, +strength)


def encode_bsratio_velocity(pub: zmq.Socket, bs_ratio_signed: float) -> None:
    """bs_ratio_signed → y-axis offset blob. Positive = more buying = +y push.
    Y-axis chosen to keep funding (x-dipole) and bs_ratio orthogonal."""
    if not np.isfinite(bs_ratio_signed):
        return
    strength = bs_ratio_signed / MAP_SCALE["bs_ratio_per_strength"]
    strength = float(np.clip(strength, -MAP_SCALE["bs_ratio_strength_cap"],
                                       +MAP_SCALE["bs_ratio_strength_cap"]))
    if abs(strength) < 1e-6:
        return
    cx, cy = CENTER
    sign = +1.0 if strength > 0 else -1.0
    _send_inject(pub, cx, cy + OFFSET_OFF * sign, abs(strength))


def encode_activity_isotropic(pub: zmq.Socket, activity_excess: float) -> None:
    """activity_excess → isotropic center blob. Excess is always relative to
    own baseline so -1 (zero activity) is a meaningful 'cold' state. We allow
    NEGATIVE strength too — quiet minutes pull energy out of the field."""
    if not np.isfinite(activity_excess):
        return
    strength = activity_excess / MAP_SCALE["activity_per_strength"]
    strength = float(np.clip(strength, -MAP_SCALE["activity_strength_cap"] / 3.0,
                                       +MAP_SCALE["activity_strength_cap"]))
    if abs(strength) < 1e-6:
        return
    cx, cy = CENTER
    _send_inject(pub, cx, cy, strength)


def encode_cvd_vorticity(pub: zmq.Socket, cvd_divergence: float) -> None:
    """cvd_divergence → vorticity ring around center. N legs alternating signs.
    Direction of rotation = sign of divergence. Tanh keeps magnitude bounded."""
    if not np.isfinite(cvd_divergence):
        return
    strength = math.tanh(cvd_divergence / MAP_SCALE["cvd_div_tanh_scale"])
    if abs(strength) < 1e-6:
        return
    cx, cy = CENTER
    sigma = SIGMA * 0.75   # tighter sigma so legs don't overlap excessively
    for k in range(VORTICITY_N_LEGS):
        theta = 2 * math.pi * k / VORTICITY_N_LEGS
        x = cx + RING_R * math.cos(theta)
        y = cy + RING_R * math.sin(theta)
        s = strength if (k % 2 == 0) else -strength
        _send_inject(pub, x, y, s, sigma)


# ───────────────────────── per-arm runner ─────────────────────────
def snapshot_row(tel: LatestTel, minute: int, arm: str,
                 tensions: dict) -> dict:
    snap, snap_wall = tel.snapshot()
    rec = {
        "minute":      int(minute),
        "arm":         arm,
        "snap_wall":   snap_wall,
        "snap_age_ms": (time.time() - snap_wall) * 1000 if snap_wall else None,
        "funding_bps":     tensions.get("funding_bps"),
        "bs_ratio_signed": tensions.get("bs_ratio_signed"),
        "activity_excess": tensions.get("activity_excess"),
        "cvd_divergence":  tensions.get("cvd_divergence"),
    }
    if snap is None:
        for ch in CHANNELS:
            rec[ch] = None
    else:
        for ch in CHANNELS:
            rec[ch] = snap.get(ch)
    return rec


def run_arm(tel: LatestTel, pub: zmq.Socket | None, arm: str,
            df: pd.DataFrame) -> pd.DataFrame:
    log(f"\n=== arm {arm}: starting ({len(df)} minutes) ===")
    if arm == "A":
        log("  arm A: NO INJECTIONS — sampling state at cadence only")
    elif arm == "T":
        log("  arm T: 4 NATIVE TENSION INJECTIONS per minute")

    records: list[dict] = []
    t_arm_start = time.time()
    last_status = t_arm_start
    n_minutes   = len(df)
    step_s      = PER_MINUTE_MS / 1000.0
    wait_inject_s = WAIT_AFTER_INJECT_MS / 1000.0

    for i in range(n_minutes):
        t_tick = time.time()
        row = df.iloc[i]
        tensions = {
            "funding_bps":     row.get("funding_bps"),
            "bs_ratio_signed": row.get("bs_ratio_signed"),
            "activity_excess": row.get("activity_excess"),
            "cvd_divergence":  row.get("cvd_divergence"),
        }

        if arm == "T" and pub is not None:
            encode_funding_dipole   (pub, tensions["funding_bps"])
            encode_bsratio_velocity (pub, tensions["bs_ratio_signed"])
            encode_activity_isotropic(pub, tensions["activity_excess"])
            encode_cvd_vorticity    (pub, tensions["cvd_divergence"])
            time.sleep(wait_inject_s)
        else:
            time.sleep(wait_inject_s)

        records.append(snapshot_row(tel, int(row.minute), arm, tensions))

        spent = time.time() - t_tick
        remaining = step_s - spent
        if remaining > 0:
            time.sleep(remaining)

        if time.time() - last_status > 60:
            elapsed = time.time() - t_arm_start
            pct = (i + 1) / n_minutes * 100
            rate_per_s = (i + 1) / elapsed if elapsed > 0 else 0
            eta_s = (n_minutes - i - 1) / rate_per_s if rate_per_s > 0 else 0
            cur_field = tel.latest or {}
            log(f"  arm {arm} {i+1}/{n_minutes} ({pct:.1f}%)  "
                f"rate={rate_per_s:.1f}min/s  ETA={eta_s/60:.1f}min  "
                f"asym={cur_field.get('asymmetry','?'):.3f}  "
                f"latest_tel_age={(time.time()-tel.latest_wall)*1000 if tel.latest_wall else -1:.0f}ms  "
                f"tel_seen={tel.n_seen}")
            last_status = time.time()

    log(f"  arm {arm} COMPLETE  records={len(records)}  total={(time.time()-t_arm_start)/60:.1f}min")
    return pd.DataFrame(records)


# ───────────────────────── main ─────────────────────────
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--days-glob", default=DEFAULT_DAYS_GLOB,
                    help="day-dir glob pattern under hl_data/minutes/")
    ap.add_argument("--max-minutes", type=int, default=None,
                    help="cap the number of minutes used (handy for shakedown)")
    ap.add_argument("--arms", default="A,T",
                    help="comma-separated arm names to run (default A,T)")
    ap.add_argument("--skip-funding", action="store_true",
                    help="don't fetch funding history (uses NaN; useful for dry run)")
    args = ap.parse_args()

    log(f"=== market_tension_injector run_id={RUN_ID} ===")
    log(f"out_dir={OUT_DIR}")
    log(f"args: days_glob={args.days_glob} max_minutes={args.max_minutes} "
        f"arms={args.arms}")

    # load data
    df = load_btc(args.days_glob)
    if args.max_minutes:
        df = df.head(args.max_minutes).reset_index(drop=True)
        log(f"capped to first {args.max_minutes} minutes")

    # funding
    funding = pd.DataFrame(columns=["minute_hour", "funding_bps_annual"])
    if not args.skip_funding:
        try:
            funding = fetch_funding_history(
                COIN,
                start_min=int(df.minute.min()) - 120,    # extra room for ffill
                end_min=int(df.minute.max()) + 120,
                cache=OUT_DIR / "funding_history.parquet"
            )
        except Exception as e:
            log(f"funding fetch failed ({e}); proceeding with NaN funding")

    # tensions
    df = compute_tensions(df, funding)
    log("tensions computed:")
    for col in ["funding_bps", "bs_ratio_signed", "activity_excess", "cvd_divergence"]:
        s = df[col]
        finite = int(np.isfinite(s.astype(float)).sum())
        log(f"  {col:<22}  n_finite={finite:>5}/{len(s)}  "
            f"mean={s.mean():+.4f}  std={s.std():+.4f}  "
            f"min={s.min():+.4f}  max={s.max():+.4f}")

    # write meta
    meta = {
        "run_id": RUN_ID,
        "coin": COIN,
        "data_root": DATA_ROOT,
        "days_glob": args.days_glob,
        "n_minutes": int(len(df)),
        "minute_range": [int(df.minute.min()), int(df.minute.max())],
        "map_scale": MAP_SCALE,
        "spatial": {
            "center": list(CENTER), "dipole_off": DIPOLE_OFF,
            "offset_off": OFFSET_OFF, "ring_r": RING_R,
            "sigma": SIGMA, "vorticity_n_legs": VORTICITY_N_LEGS,
        },
        "timing": {
            "per_minute_ms": PER_MINUTE_MS,
            "wait_after_inject_ms": WAIT_AFTER_INJECT_MS,
            "cooldown_between_arms_s": COOLDOWN_BETWEEN_ARMS_S,
        },
        "rolling_windows": {
            "activity_baseline_min": ROLL_ACTIVITY_WIN,
            "cvd_window_min":        ROLL_CVD_WIN,
            "eps_bps":               EPS_BPS,
        },
        "arms": args.arms.split(","),
    }
    (OUT_DIR / "meta.json").write_text(json.dumps(meta, indent=2))

    # ZMQ setup
    tel = LatestTel(TEL_ADDR)
    ctx = zmq.Context.instance()
    pub = ctx.socket(zmq.PUB)
    pub.connect(CMD_ADDR)
    log(f"ZMQ connected: SUB {TEL_ADDR}, PUB {CMD_ADDR}")
    log("warming up SUB socket (3s) ...")
    time.sleep(3)
    log(f"  initial tel_seen={tel.n_seen}  latest_age_ms="
        f"{(time.time()-tel.latest_wall)*1000 if tel.latest_wall else -1:.0f}")

    try:
        arms_to_run = args.arms.split(",")
        for ai, arm in enumerate(arms_to_run):
            arm = arm.strip().upper()
            if arm not in ("A", "T"):
                log(f"skipping unknown arm: {arm}")
                continue
            if ai > 0:
                log(f"--- cooldown {COOLDOWN_BETWEEN_ARMS_S}s ---")
                time.sleep(COOLDOWN_BETWEEN_ARMS_S)
            arm_df = run_arm(tel, pub if arm == "T" else None, arm, df)
            outf = OUT_DIR / f"arm_{arm}_{'tension' if arm=='T' else 'no_inject'}.parquet"
            arm_df.to_parquet(outf)
            log(f"  saved {outf.name}  ({len(arm_df)} rows)")
        log(f"\n=== ALL ARMS COMPLETE ===")
        log(f"output: {OUT_DIR}")
    finally:
        tel.stop()
        pub.close(0)
        ctx.term()


if __name__ == "__main__":
    main()