resonance-engine/market_tension_injector.py

"""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 = {
    # ---- LEVELS (4-hour rolling means) ----
    # Calibrated 2026-06-08 on full April BTC distributions:
    # per_strength = |val| q90 -> strength 1.0  (so q99 -> ~1.7-2.0, cap takes it).
    # Mean-reverting tensions (bs_ratio, activity, cvd) compress heavily under
    # the 4hr mean (10-20× vs per-minute std). Persistent ones (funding) compress
    # only ~1.5×. Scales below reflect the post-aggregation distributions.
    "funding_level_per_strength":  1180.0,
    "funding_level_strength_cap":  2.1,

    "bs_ratio_level_per_strength": 0.060,
    "bs_ratio_level_strength_cap": 1.8,

    "activity_level_per_strength": 0.58,
    "activity_level_strength_cap": 2.2,

    # cvd level: tanh keeps it bounded; scale = q90 → q90 maps to tanh(1)=0.76.
    "cvd_level_tanh_scale":        1.24e6,

    # ---- VELOCITIES (4hr level diff) ----
    # Calibrated same way: per_strength = |vel| q90.
    "funding_vel_per_strength": 900.0,
    "funding_vel_strength_cap": 1.8,

    "bs_ratio_vel_per_strength": 0.078,
    "bs_ratio_vel_strength_cap": 1.7,

    "activity_vel_per_strength": 0.87,
    "activity_vel_strength_cap": 1.8,

    "cvd_vel_tanh_scale":        1.78e6,
}

# 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

# 4-hour aggregation window for LEVELS and VELOCITIES.
# Levels = rolling mean over last 4hr (past-only shift).
# Velocities = level(t) - level(t - LEVEL_WIN_MIN). The signal we found dominant
# was asymmetry_d240 (β=+4.77). Pre-aggregating to the same timescale tests
# whether the lattice's value is temporal smoothing (4hr-input matches it) or
# non-linear coupling (4hr-input fails, only minute-level + lattice memory works).
LEVEL_WIN_MIN = 240

# Lattice spatial geometry (lattice is 1024×1024, center 512,512)
CENTER     = (512.0, 512.0)
DIPOLE_OFF = 64.0    # dipole leg offset from center (funding x, bs_ratio y)
RING_R     = 192.0   # vorticity ring radius (was 96; doubled 2026-06-07 to
                     # register in the global vorticity_mean metric)
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

# --- VELOCITY (Δlevel over 4hr) spatial geometry — distinct from levels ---
# Levels and velocities of the SAME tension must occupy distinct spatial
# signatures so the lattice can't conflate them. Velocities use a different
# radius / scale than their level counterparts.
DIPOLE_OFF_VEL = 192.0   # funding/bs_ratio velocity dipoles wider than level dipoles
SIGMA_VEL      = 16.0    # tighter sigma for velocity injections (sharper, less DC)
RING_R_ACT_VEL = 320.0   # activity-velocity annular ring (same-sign legs = pure breathing)
RING_R_CVD_VEL = 96.0    # cvd-velocity counter-rotating ring (half the cvd-level radius)
ACT_VEL_N_LEGS = 4       # activity-velocity ring leg count

# 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 AND their 4-hour level
    aggregates plus 4-hour velocities. Eight tensions total.

    Naming:
      <name>            = per-minute native tension (kept for reference & analyzer)
      <name>_level      = 4-hour rolling mean (past-only via shift(1))
      <name>_vel        = level(t) - level(t - LEVEL_WIN_MIN)
    """
    out = df.copy()

    # --- funding: forward-fill the hourly value onto 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 ---
    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| ---
    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)
    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)

    # --- 4-HOUR LEVELS (rolling means, past-only via shift(1)) ---
    for col in ["funding_bps", "bs_ratio_signed", "activity_excess", "cvd_divergence"]:
        out[f"{col}_level"] = (out[col].astype(float).shift(1)
                                       .rolling(LEVEL_WIN_MIN, min_periods=60).mean())

    # --- 4-HOUR VELOCITIES (level(t) - level(t - LEVEL_WIN_MIN)) ---
    for col in ["funding_bps", "bs_ratio_signed", "activity_excess", "cvd_divergence"]:
        out[f"{col}_vel"] = out[f"{col}_level"] - out[f"{col}_level"].shift(LEVEL_WIN_MIN)

    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))


# ---- LEVEL ENCODERS (re-use original geometries, fed with 4hr-mean values) ----
def encode_funding_level(pub: zmq.Socket, val: float) -> None:
    """4hr-mean funding → x-axis dipole at ±DIPOLE_OFF."""
    if not np.isfinite(val): return
    s = float(np.clip(val / MAP_SCALE["funding_level_per_strength"],
                      -MAP_SCALE["funding_level_strength_cap"],
                      +MAP_SCALE["funding_level_strength_cap"]))
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    _send_inject(pub, cx - DIPOLE_OFF, cy, -s)
    _send_inject(pub, cx + DIPOLE_OFF, cy, +s)


def encode_bsratio_level(pub: zmq.Socket, val: float) -> None:
    """4hr-mean bs_ratio → y-axis dipole at ±DIPOLE_OFF."""
    if not np.isfinite(val): return
    s = float(np.clip(val / MAP_SCALE["bs_ratio_level_per_strength"],
                      -MAP_SCALE["bs_ratio_level_strength_cap"],
                      +MAP_SCALE["bs_ratio_level_strength_cap"]))
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    _send_inject(pub, cx, cy - DIPOLE_OFF, -s)
    _send_inject(pub, cx, cy + DIPOLE_OFF, +s)


def encode_activity_level(pub: zmq.Socket, val: float) -> None:
    """4hr-mean activity → isotropic center blob."""
    if not np.isfinite(val): return
    s = float(np.clip(val / MAP_SCALE["activity_level_per_strength"],
                      -MAP_SCALE["activity_level_strength_cap"],
                      +MAP_SCALE["activity_level_strength_cap"]))
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    _send_inject(pub, cx, cy, s)


def encode_cvd_level(pub: zmq.Socket, val: float) -> None:
    """4hr-mean cvd_div → 6-leg alternating-sign ring at R=192."""
    if not np.isfinite(val): return
    s = math.tanh(val / MAP_SCALE["cvd_level_tanh_scale"])
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    sigma = SIGMA * 0.75
    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)
        sign = s if (k % 2 == 0) else -s
        _send_inject(pub, x, y, sign, sigma)


# ---- VELOCITY ENCODERS (new, geometrically distinct from level encoders) ----
def encode_funding_vel(pub: zmq.Socket, val: float) -> None:
    """4hr funding velocity → x-axis dipole at ±DIPOLE_OFF_VEL=192, sigma=16.
    Wider separation and tighter sigma than the level dipole → distinct signature."""
    if not np.isfinite(val): return
    s = float(np.clip(val / MAP_SCALE["funding_vel_per_strength"],
                      -MAP_SCALE["funding_vel_strength_cap"],
                      +MAP_SCALE["funding_vel_strength_cap"]))
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    _send_inject(pub, cx - DIPOLE_OFF_VEL, cy, -s, SIGMA_VEL)
    _send_inject(pub, cx + DIPOLE_OFF_VEL, cy, +s, SIGMA_VEL)


def encode_bsratio_vel(pub: zmq.Socket, val: float) -> None:
    """4hr bs_ratio velocity → y-axis dipole at ±DIPOLE_OFF_VEL=192, sigma=16."""
    if not np.isfinite(val): return
    s = float(np.clip(val / MAP_SCALE["bs_ratio_vel_per_strength"],
                      -MAP_SCALE["bs_ratio_vel_strength_cap"],
                      +MAP_SCALE["bs_ratio_vel_strength_cap"]))
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    _send_inject(pub, cx, cy - DIPOLE_OFF_VEL, -s, SIGMA_VEL)
    _send_inject(pub, cx, cy + DIPOLE_OFF_VEL, +s, SIGMA_VEL)


def encode_activity_vel(pub: zmq.Socket, val: float) -> None:
    """4hr activity velocity → 4-leg same-sign annular ring at R=320.
    Same-sign legs = pure 'breathing' mode (expansion vs contraction),
    geometrically distinct from any dipole or alternating ring."""
    if not np.isfinite(val): return
    s = float(np.clip(val / MAP_SCALE["activity_vel_per_strength"],
                      -MAP_SCALE["activity_vel_strength_cap"],
                      +MAP_SCALE["activity_vel_strength_cap"]))
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    for k in range(ACT_VEL_N_LEGS):
        theta = 2 * math.pi * k / ACT_VEL_N_LEGS
        x = cx + RING_R_ACT_VEL * math.cos(theta)
        y = cy + RING_R_ACT_VEL * math.sin(theta)
        _send_inject(pub, x, y, s, SIGMA_VEL)


def encode_cvd_vel(pub: zmq.Socket, val: float) -> None:
    """4hr cvd_div velocity → 6-leg alternating-sign ring at R=96.
    Half the level-ring radius → inner counter-rotating vorticity. Sign-flipped
    relative to level so rotation opposes the level ring's curl when both fire.
    """
    if not np.isfinite(val): return
    s = math.tanh(val / MAP_SCALE["cvd_vel_tanh_scale"])
    if abs(s) < 1e-6: return
    cx, cy = CENTER
    for k in range(VORTICITY_N_LEGS):
        theta = 2 * math.pi * k / VORTICITY_N_LEGS + (math.pi / VORTICITY_N_LEGS)  # rotated 30°
        x = cx + RING_R_CVD_VEL * math.cos(theta)
        y = cy + RING_R_CVD_VEL * math.sin(theta)
        sign = -s if (k % 2 == 0) else s  # opposite rotation to level ring
        _send_inject(pub, x, y, sign, SIGMA_VEL)


# ───────────────────────── 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,
        # per-minute (kept for reference / analyzer continuity)
        "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"),
        # 4-hour levels (what arm T actually injects)
        "funding_bps_level":     tensions.get("funding_bps_level"),
        "bs_ratio_signed_level": tensions.get("bs_ratio_signed_level"),
        "activity_excess_level": tensions.get("activity_excess_level"),
        "cvd_divergence_level":  tensions.get("cvd_divergence_level"),
        # 4-hour velocities (what arm T actually injects)
        "funding_bps_vel":     tensions.get("funding_bps_vel"),
        "bs_ratio_signed_vel": tensions.get("bs_ratio_signed_vel"),
        "activity_excess_vel": tensions.get("activity_excess_vel"),
        "cvd_divergence_vel":  tensions.get("cvd_divergence_vel"),
    }
    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: 8 TENSION INJECTIONS per minute (4 levels + 4 velocities, 4hr window)")

    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 = {
            # per-minute (reference)
            "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"),
            # 4-hour levels (injected)
            "funding_bps_level":     row.get("funding_bps_level"),
            "bs_ratio_signed_level": row.get("bs_ratio_signed_level"),
            "activity_excess_level": row.get("activity_excess_level"),
            "cvd_divergence_level":  row.get("cvd_divergence_level"),
            # 4-hour velocities (injected)
            "funding_bps_vel":     row.get("funding_bps_vel"),
            "bs_ratio_signed_vel": row.get("bs_ratio_signed_vel"),
            "activity_excess_vel": row.get("activity_excess_vel"),
            "cvd_divergence_vel":  row.get("cvd_divergence_vel"),
        }

        if arm == "T" and pub is not None:
            # 4 LEVELS — existing spatial geometries, fed with 4hr means
            encode_funding_level (pub, tensions["funding_bps_level"])
            encode_bsratio_level (pub, tensions["bs_ratio_signed_level"])
            encode_activity_level(pub, tensions["activity_excess_level"])
            encode_cvd_level     (pub, tensions["cvd_divergence_level"])
            # 4 VELOCITIES — new geometries, fed with 4hr deltas
            encode_funding_vel (pub, tensions["funding_bps_vel"])
            encode_bsratio_vel (pub, tensions["bs_ratio_signed_vel"])
            encode_activity_vel(pub, tensions["activity_excess_vel"])
            encode_cvd_vel     (pub, tensions["cvd_divergence_vel"])
            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)")
    ap.add_argument("--cooldown-s", type=int, default=COOLDOWN_BETWEEN_ARMS_S,
                    help="seconds to wait between arms (only applies after arm T)")
    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 (per-minute + 4hr levels + 4hr velocities):")
    for col in ["funding_bps", "bs_ratio_signed", "activity_excess", "cvd_divergence",
                "funding_bps_level", "bs_ratio_signed_level",
                "activity_excess_level", "cvd_divergence_level",
                "funding_bps_vel", "bs_ratio_signed_vel",
                "activity_excess_vel", "cvd_divergence_vel"]:
        s = df[col]
        finite = int(np.isfinite(s.astype(float)).sum())
        log(f"  {col:<26}  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, "dipole_off_vel": DIPOLE_OFF_VEL,
            "ring_r": RING_R,
            "ring_r_act_vel": RING_R_ACT_VEL, "ring_r_cvd_vel": RING_R_CVD_VEL,
            "sigma": SIGMA, "sigma_vel": SIGMA_VEL,
            "vorticity_n_legs": VORTICITY_N_LEGS, "act_vel_n_legs": ACT_VEL_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,
            "level_win_min":         LEVEL_WIN_MIN,
        },
        "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 = [a.strip().upper() for a in args.arms.split(",")]
        for ai, arm in enumerate(arms_to_run):
            if arm not in ("A", "T"):
                log(f"skipping unknown arm: {arm}")
                continue
            # Only cooldown AFTER arm T (the perturbing arm).
            # Arm A is passive, no field settling needed before/after it.
            if ai > 0 and arms_to_run[ai - 1] == "T":
                log(f"--- cooldown {args.cooldown_s}s (let field settle after arm T) ---")
                time.sleep(args.cooldown_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()