Scruff AI
576 lines
25 KiB
Python
576 lines
25 KiB
Python
"""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.
|
||
# Calibrated 2026-06-07 on 720 hourly rows from April BTC HL fundingHistory.
|
||
# Real distribution: median |val|=620bps, q90=1265bps, q99=2320bps, max=3835bps.
|
||
# 3000bps annualised → 1.0 dipole-leg strength.
|
||
# Second rebalance 2026-06-07: previous scale (1000bps/strength, cap 2.5)
|
||
# left funding firing as a near-constant DC offset at strength ~1.3 every
|
||
# minute, dominating the stress tensor. Tripling the scale shrinks each
|
||
# dipole leg to ~0.4 and lets minute-to-minute variance show through.
|
||
"funding_bps_per_strength": 3000.0,
|
||
"funding_strength_cap": 2.5,
|
||
|
||
# bs_ratio_signed ranges natively ±0.5 (since ratio is 0..1, minus 0.5).
|
||
# Scale ×2 so full ±0.5 → ±1.0 strength.
|
||
# Cap ±3.0 (rebalance 2026-06-07): single-blob encoding was mute (ρ<0.15)
|
||
# because a single density blob doesn't couple to global telemetry channels.
|
||
# Second rebalance 2026-06-07: bs_ratio is now a Y-AXIS DIPOLE (mechanically
|
||
# equivalent to funding's x-axis dipole, rotated 90°). Couples cleanly to
|
||
# stress_yy and vel_mean on equal footing with funding.
|
||
"bs_ratio_per_strength": 0.5,
|
||
"bs_ratio_strength_cap": 3.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 (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
|
||
|
||
# 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 dipole (mechanically equivalent to funding's
|
||
x-axis dipole, rotated 90°). Positive = buy pressure pushes +y leg positive,
|
||
-y leg negative. Couples to global stress_yy / vel_mean.
|
||
Second rebalance 2026-06-07: was a single offset blob that didn't couple
|
||
to any global telemetry channel."""
|
||
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
|
||
_send_inject(pub, cx, cy - DIPOLE_OFF, -strength)
|
||
_send_inject(pub, cx, cy + DIPOLE_OFF, +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)")
|
||
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:")
|
||
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,
|
||
"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 = [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()
|