#!/usr/bin/env python3 from __future__ import annotations """ dex_platform/backtest/portfolio_router_v4.py Portfolio-aware CL LP router with: - allocation-aware route arms; - DEMA/MA trend gate for BIO-like bull range-order entries; - event-level warmup replay before entry; - 12h/24h health-check loop while a virtual LP position is active; - withdraw/idle on regime break instead of blind periodic rebalance. This is still a backtest / paper-live validation tool. It never signs txs. """ import argparse import csv import json import math from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Tuple import numpy as np from cl_fee_replay_fast_npz_v3 import ( filter_time, liquidity_for_capital, liquidity_share, load_npz, score_row, sqrt_raw_token1_per_token0_from_price, unit_value_curve_from_sqrt, unit_value_one_from_sqrt, ) SCRIPT_VERSION = "portfolio_router_v4_healthcheck_2026_05_04" @dataclass(frozen=True) class RouteArm: name: str action: str # trade | idle capital_usd: float lower_pct: float upper_pct: float rebalance_hours: float mode: str reason: str priority: int = 0 IDLE_ARM = RouteArm("idle", "idle", 0.0, 0.0, 0.0, 0.0, "idle", "idle-no-route", 0) BIO_BULL_ARM = RouteArm( "bio_bull_periodic_2_90_336h", "trade", 600.0, 2.0, 90.0, 336.0, "periodic", "bio-bull-validated-candidate", 100, ) CHECK_TINY_BEAR_ARM = RouteArm( "check_tiny_bear_periodic_50_0.1_672h", "trade", 25.0, 50.0, 0.1, 672.0, "periodic", "check-tiny-bear-validated-candidate", 30, ) @dataclass(frozen=True) class RouterConfig: routing_mode: str total_capital_usd: float lookback_hours: float decision_hours: float gas_usd: float swap_cost_bps: float max_current_share_pct: float max_deploy_fraction: float # BIO rolling route gate bio_micro_min_drift_pct: float bio_micro_min_trend_ratio: float bio_macro_lookback_hours: float bio_macro_kill_drift_pct: float bio_bull_min_fee_budget_pct_day: float bio_bull_max_toxicity: float # CHECK tiny route gate check_bear_max_drift_pct: float check_bear_min_fee_budget_pct_day: float min_events_per_hour: float allow_synd: bool no_bootstrap: bool strict_quote_check: bool # Event-level warmup probe before entry prewarm_hours: float warmup_probe_hours: float warmup_min_span_fraction: float warmup_min_return_pct: float warmup_max_mdd_pct: float warmup_min_pnl_mdd: float warmup_min_time_in_range_pct: float warmup_max_p99_share_pct: float warmup_max_max_share_pct: float # DEMA/MA trend gate dema_enable: bool dema_fast_hours: float dema_slow_hours: float dema_slope_hours: float dema_min_slow_slope_pct: float dema_require_price_above_slow: bool dema_require_fast_above_slow: bool # Active-position health-check health_check_enable: bool health_min_micro_drift_pct: float health_probe_enable: bool health_probe_hours: float health_probe_min_return_pct: float health_probe_max_mdd_pct: float health_probe_min_pnl_mdd: float health_probe_min_time_in_range_pct: float health_probe_max_p99_share_pct: float exit_on_route_mismatch: bool rebalance_when_due: bool no_reentry_after_exit: bool health_max_total_dd_pct: float # ───────────────────────────────────────────────────────────────────────────── # Generic helpers # ───────────────────────────────────────────────────────────────────────────── def ts_to_iso(ts: int) -> str: return datetime.fromtimestamp(int(ts), tz=timezone.utc).isoformat().replace("+00:00", "Z") def parse_iso_ts(s: str) -> int: return int(datetime.fromisoformat(str(s).replace("Z", "+00:00")).timestamp()) def write_rows_csv(path: Path, rows: List[Dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) if not rows: path.write_text("", encoding="utf-8") return fields: List[str] = [] for row in rows: for key in row.keys(): if key not in fields: fields.append(key) with path.open("w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=fields) w.writeheader() w.writerows(rows) def parse_fee_specs(spec: str, metadata_fee_rate: float | None = None) -> List[Tuple[str, float]]: out: List[Tuple[str, float]] = [] for item in str(spec).split(","): item = item.strip() if not item: continue if item == "metadata": if metadata_fee_rate is None: raise ValueError("fee spec `metadata` requested, but NPZ has no meta fee_rate") out.append(("metadata", float(metadata_fee_rate))) continue if ":" not in item: raise ValueError(f"Bad fee spec: {item}; expected name:rate or metadata") name, rate = item.split(":", 1) if name.strip() == "metadata" and rate.strip().lower() in {"auto", "meta", "metadata"}: if metadata_fee_rate is None: raise ValueError("metadata fee requested, but NPZ has no meta fee_rate") out.append(("metadata", float(metadata_fee_rate))) else: out.append((name.strip(), float(rate))) if not out: raise ValueError("No fee specs") return out def infer_time_range(data: Dict[str, Any], time_from: str, time_to: str) -> Tuple[str, str]: meta = data.get("meta", {}) or {} tf = time_from tt = time_to if not tf and "timestamp_start" in meta: tf = ts_to_iso(int(meta["timestamp_start"])) if not tt and "timestamp_end" in meta: tt = ts_to_iso(int(meta["timestamp_end"]) + 1) if not tf or not tt: ts = data["ts"].astype(np.int64) tf = tf or ts_to_iso(int(ts[0])) tt = tt or ts_to_iso(int(ts[-1]) + 1) return tf, tt def canonical_decimals(meta: Dict[str, Any], fallback_dec0: int, fallback_dec1: int) -> Tuple[int, int]: token_dec = { "USDC": 6, "USDT": 6, "USDBC": 6, "DAI": 18, "BIO": 18, "CHECK": 18, "SYND": 18, "WETH": 18, "RSC": 18, "VIRTUAL": 18, } t0 = str(meta.get("token0", "")).upper() t1 = str(meta.get("token1", "")).upper() return int(token_dec.get(t0, fallback_dec0)), int(token_dec.get(t1, fallback_dec1)) def pool_key_from_meta_path(meta: Dict[str, Any], npz_path: str) -> str: text = " ".join([ str(meta.get("pool_name", "")), str(meta.get("token0", "")), str(meta.get("token1", "")), Path(npz_path).name, ]).lower() token0 = str(meta.get("token0", "")).upper() token1 = str(meta.get("token1", "")).upper() stable0 = token0 in {"USDC", "USDT", "DAI", "USDBC"} if not stable0: if "rsc" in text: return "unsupported_rsc_weth" if "virtual" in text: return "unsupported_virtual_weth" if "bio" in text: return "unsupported_inverted_bio_usdc" return "unsupported_non_usd_quote" if "bio" in text: return "bio_usdc" if "check" in text: return "check_usdc" if "synd" in text: return "synd_usdc" if token1 == "WETH" or "weth_usdc" in text: return "weth_usdc" return "unknown" def dataset_key_from_path(npz_path: str, meta: Dict[str, Any]) -> str: name = Path(npz_path).name.lower() pool_name = str(meta.get("pool_name", "")).lower() text = f"{name} {pool_name}" if "bio" in text and "combined" in text: return "bio_combined" if "bio" in text and "apr" in text: return "bio_apr" if "bio" in text and ("feb" in text or "mar" in text): return "bio_feb_mar" if "bio" in text and "may" in text: return "bio_may" if "check" in text and "scaled" in text: return "check_scaled_current_liq" if "check" in text and "025_new" in text: return "check_apr_new" if "check" in text: return "check_q1" if "weth" in text and "usdc" in text: return "weth_q1" if "synd" in text: return "synd_apr" return "unknown" def robust_pct(x: np.ndarray, q: float, default: float) -> float: if len(x) == 0: return default v = float(np.nanpercentile(x, q)) return v if np.isfinite(v) else default def max_drawdown_pct(equity: np.ndarray) -> float: if len(equity) == 0: return 0.0 peak = np.maximum.accumulate(np.where(equity > 0, equity, 1e-300)) dd = equity / peak - 1.0 return float(np.nanmin(dd) * 100.0) def share_stats_pct(share_in: np.ndarray) -> Tuple[float, float, float, float]: if len(share_in) == 0: return 0.0, 0.0, 0.0, 0.0 s = np.asarray(share_in, dtype=np.float64) * 100.0 return float(np.mean(s)), float(np.percentile(s, 95)), float(np.percentile(s, 99)), float(np.max(s)) # ───────────────────────────────────────────────────────────────────────────── # Regime / trend features # ───────────────────────────────────────────────────────────────────────────── def window_features( price: np.ndarray, input_usd: np.ndarray, ts: np.ndarray, end_idx: int, lookback_hours: float, fee_rate: float, capital: float, ) -> Dict[str, float]: end_ts = int(ts[end_idx]) start_ts = end_ts - int(lookback_hours * 3600) lo = int(np.searchsorted(ts, start_ts, side="left")) lo = min(max(0, lo), end_idx) p = price[lo : end_idx + 1] flow = input_usd[lo : end_idx + 1] t = ts[lo : end_idx + 1] if len(p) < 3: return { "lookback_points": float(len(p)), "drift_pct": 0.0, "path_pct": 0.0, "trend_ratio": 0.0, "vol_pct": 0.0, "move_p95_pct": 0.0, "flow_usd_per_hour": 0.0, "events_per_hour": 0.0, "toxicity": 0.0, "fee_budget_pct_per_day": 0.0, "span_hours": 0.0, } lr = np.diff(np.log(np.maximum(p, 1e-300))) abs_lr = np.abs(lr) drift_pct = float((p[-1] / p[0] - 1.0) * 100.0) path_pct = float(np.sum(abs_lr) * 100.0) trend_ratio = float(drift_pct / max(path_pct, 1e-9)) vol_pct = float(np.std(lr) * math.sqrt(max(1, len(lr))) * 100.0) move_p95_pct = robust_pct(abs_lr * 100.0, 95, 0.0) hours = max(1.0 / 60.0, (int(t[-1]) - int(t[0])) / 3600.0) span_hours = float((int(t[-1]) - int(t[0])) / 3600.0) flow_usd_per_hour = float(np.sum(flow) / hours) events_per_hour = float(len(p) / hours) toxicity = float(min(1.0, abs(drift_pct) / max(path_pct, 1e-9))) fee_budget_pct_per_day = float(fee_rate * flow_usd_per_hour * 24.0 / max(capital, 1e-9) * 100.0) return { "lookback_points": float(len(p)), "drift_pct": drift_pct, "path_pct": path_pct, "trend_ratio": trend_ratio, "vol_pct": vol_pct, "move_p95_pct": move_p95_pct, "flow_usd_per_hour": flow_usd_per_hour, "events_per_hour": events_per_hour, "toxicity": toxicity, "fee_budget_pct_per_day": fee_budget_pct_per_day, "span_hours": span_hours, } def _ema_series(x: np.ndarray, span: int) -> np.ndarray: span = max(2, int(span)) alpha = 2.0 / (span + 1.0) out = np.empty_like(x, dtype=np.float64) if len(x) == 0: return out out[0] = float(x[0]) for i in range(1, len(x)): out[i] = alpha * float(x[i]) + (1.0 - alpha) * out[i - 1] return out def _dema_series(x: np.ndarray, span: int) -> np.ndarray: e1 = _ema_series(x, span) e2 = _ema_series(e1, span) return 2.0 * e1 - e2 def dema_features(price: np.ndarray, ts: np.ndarray, end_idx: int, cfg: RouterConfig) -> Dict[str, float]: if not cfg.dema_enable: return { "dema_enabled": 0.0, "dema_gate_pass": 1.0, "dema_fast": 0.0, "dema_slow": 0.0, "dema_slow_slope_pct": 0.0, "dema_price_above_slow": 1.0, "dema_fast_above_slow": 1.0, "dema_span_hours": 0.0, } end_ts = int(ts[end_idx]) window_hours = max(cfg.dema_slow_hours * 2.2, cfg.dema_slow_hours + cfg.dema_slope_hours * 1.2, cfg.dema_fast_hours * 4.0) start_ts = end_ts - int(window_hours * 3600) lo = int(np.searchsorted(ts, start_ts, side="left")) lo = min(max(0, lo), end_idx) p = price[lo : end_idx + 1] t = ts[lo : end_idx + 1] span_hours = float((int(t[-1]) - int(t[0])) / 3600.0) if len(t) > 1 else 0.0 if len(p) < 8 or span_hours < cfg.dema_slow_hours * 0.5: return { "dema_enabled": 1.0, "dema_gate_pass": 0.0, "dema_fast": float(p[-1]) if len(p) else 0.0, "dema_slow": float(p[-1]) if len(p) else 0.0, "dema_slow_slope_pct": 0.0, "dema_price_above_slow": 0.0, "dema_fast_above_slow": 0.0, "dema_span_hours": span_hours, } points_per_hour = len(p) / max(span_hours, 1e-9) fast_span = max(3, int(round(cfg.dema_fast_hours * points_per_hour))) slow_span = max(fast_span + 2, int(round(cfg.dema_slow_hours * points_per_hour))) fast = _dema_series(p, fast_span) slow = _dema_series(p, slow_span) slope_ts = end_ts - int(cfg.dema_slope_hours * 3600) slope_local = int(np.searchsorted(t, slope_ts, side="left")) slope_local = min(max(0, slope_local), len(p) - 2) slow_now = float(slow[-1]) fast_now = float(fast[-1]) price_now = float(p[-1]) slow_then = float(slow[slope_local]) slow_slope_pct = float((slow_now / max(slow_then, 1e-300) - 1.0) * 100.0) price_above = 1.0 if price_now > slow_now else 0.0 fast_above = 1.0 if fast_now > slow_now else 0.0 pass_reasons = [] if cfg.dema_require_price_above_slow and not price_above: pass_reasons.append("price_below_slow") if cfg.dema_require_fast_above_slow and not fast_above: pass_reasons.append("fast_below_slow") if slow_slope_pct < cfg.dema_min_slow_slope_pct: pass_reasons.append("slow_slope") return { "dema_enabled": 1.0, "dema_gate_pass": 0.0 if pass_reasons else 1.0, "dema_fast": fast_now, "dema_slow": slow_now, "dema_slow_slope_pct": slow_slope_pct, "dema_price_above_slow": price_above, "dema_fast_above_slow": fast_above, "dema_span_hours": span_hours, } # ───────────────────────────────────────────────────────────────────────────── # Route selection and probes # ───────────────────────────────────────────────────────────────────────────── def locked_candidate_arm(dataset_key: str, pool_key: str, cfg: RouterConfig) -> RouteArm: if pool_key.startswith("unsupported"): return RouteArm(**{**IDLE_ARM.__dict__, "reason": f"idle-{pool_key}-needs-normalization"}) if dataset_key in {"bio_apr", "bio_may", "bio_combined"}: return BIO_BULL_ARM if dataset_key == "check_q1": return CHECK_TINY_BEAR_ARM if dataset_key == "check_scaled_current_liq": return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-check-scaled-liquidity-invalidates-old-edge"}) if dataset_key == "check_apr_new": return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-check-new-no-strict-pass-yet"}) if dataset_key == "weth_q1": return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-weth-no-pnl-mdd-edge"}) if pool_key == "synd_usdc" and not cfg.allow_synd: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-synd-disabled"}) if dataset_key == "bio_feb_mar": return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-bio-downtrend-no-pass"}) return IDLE_ARM def rolling_arm(pool_key: str, feat: Dict[str, float], macro_feat: Dict[str, float], dema: Dict[str, float], cfg: RouterConfig) -> RouteArm: if pool_key.startswith("unsupported"): return RouteArm(**{**IDLE_ARM.__dict__, "reason": f"idle-{pool_key}-needs-normalization"}) if pool_key == "weth_usdc": return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-weth-no-pnl-mdd-edge"}) if feat["lookback_points"] < 3: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-warmup"}) if feat["events_per_hour"] < cfg.min_events_per_hour: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-thin-events"}) if pool_key == "bio_usdc": if cfg.no_bootstrap and macro_feat.get("span_hours", 0.0) < cfg.bio_macro_lookback_hours * 0.80: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-bio-bootstrap-wait-macro-history"}) if macro_feat.get("lookback_points", 0.0) >= 20.0 and macro_feat.get("drift_pct", 0.0) <= cfg.bio_macro_kill_drift_pct: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-bio-macro-downtrend-kill"}) if cfg.dema_enable and dema.get("dema_gate_pass", 0.0) < 0.5: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-bio-dema-gate-fail"}) if ( feat["drift_pct"] >= cfg.bio_micro_min_drift_pct and feat["trend_ratio"] >= cfg.bio_micro_min_trend_ratio and feat["fee_budget_pct_per_day"] >= cfg.bio_bull_min_fee_budget_pct_day and feat["toxicity"] <= cfg.bio_bull_max_toxicity ): return BIO_BULL_ARM if feat["drift_pct"] < 0: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-bio-nonbull"}) return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-bio-weak-bull"}) if pool_key == "check_usdc": if feat["drift_pct"] <= cfg.check_bear_max_drift_pct and feat["fee_budget_pct_per_day"] >= cfg.check_bear_min_fee_budget_pct_day: return CHECK_TINY_BEAR_ARM return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-check-no-bear-edge"}) if pool_key == "synd_usdc" and not cfg.allow_synd: return RouteArm(**{**IDLE_ARM.__dict__, "reason": "idle-synd-disabled"}) return IDLE_ARM def current_capacity_ok(arm: RouteArm, price: float, active_liq: float, dec0: int, dec1: int, cfg: RouterConfig) -> Tuple[bool, float, float, float, float]: if arm.action != "trade" or arm.capital_usd <= 0.0: return True, 0.0, 0.0, 0.0, 0.0 deploy_cap = min(arm.capital_usd, cfg.total_capital_usd * cfg.max_deploy_fraction) lower = price * (1.0 - arm.lower_pct / 100.0) upper = price * (1.0 + arm.upper_pct / 100.0) our_liq = liquidity_for_capital(deploy_cap, price, lower, upper, dec0, dec1) share_pct = float(liquidity_share(our_liq, float(active_liq)) * 100.0) if our_liq > 0.0 else 0.0 return share_pct <= cfg.max_current_share_pct, share_pct, our_liq, lower, upper def single_arm_probe( arm: RouteArm, price: np.ndarray, input_usd: np.ndarray, active_liq: np.ndarray, sqrt_price: np.ndarray, ts: np.ndarray, end_idx: int, dec0: int, dec1: int, fee_rate: float, cfg: RouterConfig, probe_hours: float, min_return_pct: float, max_mdd_pct: float, min_pnl_mdd: float, min_time_in_range_pct: float, max_p99_share_pct: float, max_max_share_pct: float, min_span_fraction_base_hours: float, ) -> Dict[str, Any]: if arm.action != "trade" or arm.capital_usd <= 0: return {"pass": True, "reason": "idle-arm", "return_pct": 0.0, "mdd_pct": 0.0, "pnl_mdd": 0.0, "time_in_range_pct": 0.0, "p99_share_pct": 0.0, "max_share_pct": 0.0, "span_hours": 0.0} end_ts = int(ts[end_idx]) start_ts = end_ts - int(probe_hours * 3600) start_idx = int(np.searchsorted(ts, start_ts, side="left")) start_idx = min(max(0, start_idx), end_idx) span_hours = float((int(ts[end_idx]) - int(ts[start_idx])) / 3600.0) if end_idx > start_idx else 0.0 min_span = min_span_fraction_base_hours * cfg.warmup_min_span_fraction if end_idx - start_idx < 3 or span_hours < min_span: return {"pass": False, "reason": "insufficient-history", "return_pct": 0.0, "mdd_pct": 0.0, "pnl_mdd": 0.0, "time_in_range_pct": 0.0, "p99_share_pct": 0.0, "max_share_pct": 0.0, "span_hours": span_hours} cap0 = min(float(arm.capital_usd), cfg.total_capital_usd * cfg.max_deploy_fraction) cash = cap0 eq_parts: List[np.ndarray] = [] in_parts: List[np.ndarray] = [] share_parts: List[np.ndarray] = [] i = start_idx safety = 0 while i <= end_idx and safety < 10000: safety += 1 p0 = float(price[i]) lower = p0 * (1.0 - arm.lower_pct / 100.0) upper = p0 * (1.0 + arm.upper_pct / 100.0) our_liq = liquidity_for_capital(max(cash, 0.0), p0, lower, upper, dec0, dec1) if not np.isfinite(our_liq) or our_liq <= 0.0: return {"pass": False, "reason": "zero-liquidity", "return_pct": 0.0, "mdd_pct": 0.0, "pnl_mdd": 0.0, "time_in_range_pct": 0.0, "p99_share_pct": 0.0, "max_share_pct": 0.0, "span_hours": span_hours} next_ts = int(ts[i]) + int(max(1.0, arm.rebalance_hours) * 3600) j = int(np.searchsorted(ts, next_ts, side="left")) j = min(max(i + 1, j), end_idx + 1) sl = slice(i, j) pseg = price[sl] in_range = (pseg >= lower) & (pseg <= upper) share = liquidity_share(our_liq, active_liq[sl]) fees = np.zeros_like(pseg, dtype=np.float64) fees[in_range] = input_usd[sl][in_range] * fee_rate * share[in_range] fees_cum = np.cumsum(fees) unit = unit_value_curve_from_sqrt(sqrt_price[sl], pseg, lower, upper, dec0, dec1) eq = our_liq * unit + fees_cum if len(eq) == 0: break cash = float(eq[-1]) cost = cfg.gas_usd + cash * (cfg.swap_cost_bps / 10000.0) cash = max(0.0, cash - cost) eq_parts.append(eq) in_parts.append(in_range.astype(np.int8)) share_parts.append(share) i = j if not eq_parts: return {"pass": False, "reason": "empty-probe", "return_pct": 0.0, "mdd_pct": 0.0, "pnl_mdd": 0.0, "time_in_range_pct": 0.0, "p99_share_pct": 0.0, "max_share_pct": 0.0, "span_hours": span_hours} eq_all = np.concatenate(eq_parts) in_all = np.concatenate(in_parts).astype(bool) share_all = np.concatenate(share_parts) share_in = share_all[in_all] ret = float((eq_all[-1] / cap0 - 1.0) * 100.0) mdd = max_drawdown_pct(eq_all) pnl_mdd = ret / max(abs(mdd), 1e-9) if ret > 0 else 0.0 time_in = float(np.mean(in_all) * 100.0) if len(in_all) else 0.0 p99 = float(np.percentile(share_in, 99) * 100.0) if len(share_in) else 0.0 mx = float(np.max(share_in) * 100.0) if len(share_in) else 0.0 reasons = [] if ret < min_return_pct: reasons.append("return") if abs(mdd) > max_mdd_pct: reasons.append("mdd") if pnl_mdd < min_pnl_mdd: reasons.append("pnl_mdd") if time_in < min_time_in_range_pct: reasons.append("time_in_range") if p99 > max_p99_share_pct: reasons.append("p99_share") if mx > max_max_share_pct: reasons.append("max_share") return { "pass": not reasons, "reason": "pass" if not reasons else "fail-" + "+".join(reasons), "return_pct": ret, "mdd_pct": mdd, "pnl_mdd": pnl_mdd, "time_in_range_pct": time_in, "p99_share_pct": p99, "max_share_pct": mx, "span_hours": span_hours, } def choose_arm( npz_path: str, dataset_key: str, pool_key: str, price: np.ndarray, input_usd: np.ndarray, active_liq: np.ndarray, sqrt_price: np.ndarray, ts: np.ndarray, idx: int, dec0: int, dec1: int, fee_rate: float, cfg: RouterConfig, ) -> Tuple[RouteArm, Dict[str, float]]: feat = window_features(price, input_usd, ts, idx, cfg.lookback_hours, fee_rate, max(1.0, cfg.total_capital_usd)) macro_feat = window_features(price, input_usd, ts, idx, cfg.bio_macro_lookback_hours, fee_rate, max(1.0, cfg.total_capital_usd)) dema = dema_features(price, ts, idx, cfg) feat.update({f"macro_{k}": float(v) for k, v in macro_feat.items()}) feat.update(dema) if cfg.routing_mode == "locked_candidates": arm = locked_candidate_arm(dataset_key, pool_key, cfg) elif cfg.routing_mode in {"rolling", "rolling_warmup", "rolling_v4"}: arm = rolling_arm(pool_key, feat, macro_feat, dema, cfg) else: raise ValueError(f"Unknown routing mode: {cfg.routing_mode}") if arm.action == "trade": if arm.capital_usd > cfg.total_capital_usd * cfg.max_deploy_fraction + 1e-9: return RouteArm(**{**IDLE_ARM.__dict__, "reason": f"idle-capital-over-max-fraction-{arm.name}"}), feat ok, share_pct, _liq, _lo, _up = current_capacity_ok(arm, float(price[idx]), float(active_liq[idx]), dec0, dec1, cfg) feat["current_share_pct"] = share_pct if not ok: return RouteArm(**{**IDLE_ARM.__dict__, "reason": f"capacity-fail-{arm.name}-share-{share_pct:.4f}%"}), feat if cfg.routing_mode in {"rolling_warmup", "rolling_v4"}: probe = single_arm_probe( arm, price, input_usd, active_liq, sqrt_price, ts, idx, dec0, dec1, fee_rate, cfg, cfg.warmup_probe_hours, cfg.warmup_min_return_pct, cfg.warmup_max_mdd_pct, cfg.warmup_min_pnl_mdd, cfg.warmup_min_time_in_range_pct, cfg.warmup_max_p99_share_pct, cfg.warmup_max_max_share_pct, cfg.prewarm_hours, ) feat.update({ "warmup_probe_pass": 1.0 if probe["pass"] else 0.0, "warmup_probe_return_pct": float(probe["return_pct"]), "warmup_probe_mdd_pct": float(probe["mdd_pct"]), "warmup_probe_pnl_mdd": float(probe["pnl_mdd"]), "warmup_probe_time_in_range_pct": float(probe["time_in_range_pct"]), "warmup_probe_p99_share_pct": float(probe["p99_share_pct"]), "warmup_probe_max_share_pct": float(probe["max_share_pct"]), "warmup_probe_span_hours": float(probe["span_hours"]), }) if not probe["pass"]: return RouteArm(**{**IDLE_ARM.__dict__, "reason": f"idle-warmup-probe-{arm.name}-{probe['reason']}"}), feat return arm, feat def health_check( arm: RouteArm, pool_key: str, price: np.ndarray, input_usd: np.ndarray, active_liq: np.ndarray, sqrt_price: np.ndarray, ts: np.ndarray, idx: int, dec0: int, dec1: int, fee_rate: float, cfg: RouterConfig, ) -> Tuple[bool, str, Dict[str, float]]: if not cfg.health_check_enable or arm.action != "trade": return True, "health-disabled", {} feat = window_features(price, input_usd, ts, idx, cfg.lookback_hours, fee_rate, max(1.0, cfg.total_capital_usd)) macro = window_features(price, input_usd, ts, idx, cfg.bio_macro_lookback_hours, fee_rate, max(1.0, cfg.total_capital_usd)) dema = dema_features(price, ts, idx, cfg) out: Dict[str, float] = {f"health_{k}": float(v) for k, v in feat.items()} out.update({f"health_macro_{k}": float(v) for k, v in macro.items()}) out.update({f"health_{k}": float(v) for k, v in dema.items()}) reasons: List[str] = [] if pool_key == "bio_usdc": if cfg.dema_enable: # Exit DEMA gate is deliberately softer than entry DEMA gate. # A single fast/slow wiggle should not force an LP withdraw; require a real break. hard_dema_break = ( dema.get("dema_slow_slope_pct", 0.0) < -1.0 or ( dema.get("dema_price_above_slow", 1.0) < 0.5 and dema.get("dema_fast_above_slow", 1.0) < 0.5 and feat.get("drift_pct", 0.0) < cfg.health_min_micro_drift_pct ) ) if hard_dema_break: reasons.append("dema_break") if macro.get("lookback_points", 0.0) >= 20.0 and macro.get("drift_pct", 0.0) <= cfg.bio_macro_kill_drift_pct: reasons.append("macro_downtrend") if feat.get("drift_pct", 0.0) < cfg.health_min_micro_drift_pct: reasons.append("micro_drift") ok, share_pct, _liq, _lo, _up = current_capacity_ok(arm, float(price[idx]), float(active_liq[idx]), dec0, dec1, cfg) out["health_current_share_pct"] = share_pct if not ok: reasons.append("capacity") if cfg.health_probe_enable: probe = single_arm_probe( arm, price, input_usd, active_liq, sqrt_price, ts, idx, dec0, dec1, fee_rate, cfg, cfg.health_probe_hours, cfg.health_probe_min_return_pct, cfg.health_probe_max_mdd_pct, cfg.health_probe_min_pnl_mdd, cfg.health_probe_min_time_in_range_pct, cfg.health_probe_max_p99_share_pct, cfg.warmup_max_max_share_pct, min(cfg.health_probe_hours, cfg.prewarm_hours), ) out.update({ "health_probe_pass": 1.0 if probe["pass"] else 0.0, "health_probe_return_pct": float(probe["return_pct"]), "health_probe_mdd_pct": float(probe["mdd_pct"]), "health_probe_pnl_mdd": float(probe["pnl_mdd"]), "health_probe_time_in_range_pct": float(probe["time_in_range_pct"]), "health_probe_p99_share_pct": float(probe["p99_share_pct"]), }) if not probe["pass"]: reasons.append("probe_" + str(probe["reason"])) if reasons: return False, "health-exit-" + "+".join(reasons), out return True, "health-pass", out # ───────────────────────────────────────────────────────────────────────────── # Router replay # ───────────────────────────────────────────────────────────────────────────── def pnl_mdd_from_row(row: Dict[str, Any]) -> float: mdd = abs(float(row.get("mdd_total_pct", row.get("mdd_pct", 0.0)))) ret = float(row.get("return_total_pct", row.get("return_pct", 0.0))) if mdd < 1e-12: return float("inf") if ret > 0 else 0.0 return ret / mdd def strict_pass(row: Dict[str, Any], args: argparse.Namespace) -> bool: return ( float(row["return_total_pct"]) > 0.0 and abs(float(row["mdd_total_pct"])) <= args.strict_mdd_pct and float(row["pnl_mdd_total"]) >= args.min_pnl_mdd and float(row["avg_liquidity_share_pct_when_in_range"]) <= args.max_avg_liquidity_share_pct and float(row["p95_liquidity_share_pct_when_in_range"]) <= args.max_p95_liquidity_share_pct and float(row["p99_liquidity_share_pct_when_in_range"]) <= args.max_p99_liquidity_share_pct and float(row["max_liquidity_share_pct_when_in_range"]) <= args.max_liquidity_share_pct ) def router_score(row: Dict[str, Any], args: argparse.Namespace) -> float: mapped = dict(row) mapped["return_pct"] = row["return_total_pct"] mapped["mdd_pct"] = row["mdd_total_pct"] try: return float(score_row(mapped, args)) except Exception: return float(row["return_total_pct"] - 2.0 * max(0.0, abs(row["mdd_total_pct"]) - args.target_mdd_pct)) def run_router( npz_path: str, dataset_key: str, pool_key: str, price: np.ndarray, input_usd: np.ndarray, active_liq: np.ndarray, ts: np.ndarray, dec0: int, dec1: int, fee_rate: float, fee_scenario: str, cfg: RouterConfig, ) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]: n = len(price) if n < 2: raise SystemExit(f"Too few events: {npz_path}") sqrt_price = sqrt_raw_token1_per_token0_from_price(price, dec0, dec1) decision_interval = max(1, int(cfg.decision_hours * 3600)) deployed = False active_arm = IDLE_ARM unused_capital = cfg.total_capital_usd portfolio_cash = cfg.total_capital_usd position_cash = 0.0 our_liq = 0.0 lower = upper = 0.0 fees_uncollected = 0.0 fees_earned_total = 0.0 fees_reinvested = 0.0 costs_cum = 0.0 next_rebalance_ts = 0 rebalances = 0 entries = 0 exits = 0 health_exits = 0 decisions = 0 idle_decisions = 0 capacity_fail_decisions = 0 route_counts: Dict[str, int] = {} reason_counts: Dict[str, int] = {} action_counts: Dict[str, int] = {} equity_total = np.full(n, cfg.total_capital_usd, dtype=np.float64) deployed_equity = np.zeros(n, dtype=np.float64) pos_arr = np.zeros(n, dtype=np.float64) in_arr = np.zeros(n, dtype=np.int8) share_arr = np.zeros(n, dtype=np.float64) logs: List[Dict[str, Any]] = [] def close_position(idx: int, reason: str) -> None: nonlocal deployed, active_arm, portfolio_cash, position_cash, unused_capital, our_liq, lower, upper nonlocal fees_uncollected, fees_reinvested, costs_cum, rebalances, exits, health_exits, next_rebalance_ts if not deployed or our_liq <= 0.0: return p_now = float(price[idx]) unit_now = unit_value_one_from_sqrt(float(sqrt_price[idx]), p_now, lower, upper, dec0, dec1) pos_val = our_liq * max(0.0, unit_now) position_cash = pos_val + fees_uncollected fees_reinvested += fees_uncollected fees_uncollected = 0.0 cost = cfg.gas_usd + position_cash * (cfg.swap_cost_bps / 10000.0) costs_cum += cost position_cash = max(0.0, position_cash - cost) portfolio_cash = unused_capital + position_cash deployed = False active_arm = IDLE_ARM our_liq = 0.0 lower = upper = 0.0 next_rebalance_ts = 0 rebalances += 1 exits += 1 if reason.startswith("health-exit"): health_exits += 1 def enter_position(idx: int, arm: RouteArm) -> bool: nonlocal deployed, active_arm, portfolio_cash, position_cash, unused_capital, our_liq, lower, upper, next_rebalance_ts, entries if arm.action != "trade" or arm.capital_usd <= 0.0: return False deploy_cap = min(float(arm.capital_usd), portfolio_cash, cfg.total_capital_usd * cfg.max_deploy_fraction) if deploy_cap <= 0.0: return False p = float(price[idx]) lo = p * (1.0 - arm.lower_pct / 100.0) up = p * (1.0 + arm.upper_pct / 100.0) liq = liquidity_for_capital(deploy_cap, p, lo, up, dec0, dec1) if not np.isfinite(liq) or liq <= 0.0: return False position_cash = deploy_cap unused_capital = max(0.0, portfolio_cash - deploy_cap) lower, upper = lo, up our_liq = liq deployed = True active_arm = arm next_rebalance_ts = int(ts[idx]) + int(max(1.0, arm.rebalance_hours) * 3600) entries += 1 return True idx = 0 while idx < n: candidate_arm, feat = choose_arm( npz_path, dataset_key, pool_key, price, input_usd, active_liq, sqrt_price, ts, idx, dec0, dec1, fee_rate, cfg ) decisions += 1 action_taken = "idle" exit_reason = "" health_reason = "" health_feat: Dict[str, float] = {} if deployed and active_arm.action == "trade": health_ok, health_reason, health_feat = health_check( active_arm, pool_key, price, input_usd, active_liq, sqrt_price, ts, idx, dec0, dec1, fee_rate, cfg ) # Additional portfolio DD stop at decision points. This is a live safety gate, # not an optimizer trick: if virtual equity is already bleeding past the allowed # threshold, withdraw and wait for a fresh warmup window. if cfg.health_max_total_dd_pct > 0.0 and our_liq > 0.0: p_now = float(price[idx]) unit_now = unit_value_one_from_sqrt(float(sqrt_price[idx]), p_now, lower, upper, dec0, dec1) current_total_equity = unused_capital + our_liq * max(0.0, unit_now) + fees_uncollected current_total_dd_pct = (current_total_equity / max(cfg.total_capital_usd, 1e-9) - 1.0) * 100.0 health_feat["health_current_total_equity_usd"] = float(current_total_equity) health_feat["health_current_total_dd_pct"] = float(current_total_dd_pct) if current_total_dd_pct <= -cfg.health_max_total_dd_pct: health_ok = False health_reason = f"health-exit-total-dd-{current_total_dd_pct:.2f}%" due_rebalance = cfg.rebalance_when_due and int(ts[idx]) >= int(next_rebalance_ts) route_mismatch = cfg.exit_on_route_mismatch and candidate_arm.name != active_arm.name if not health_ok: exit_reason = health_reason close_position(idx, exit_reason) action_taken = "exit_health" elif route_mismatch: exit_reason = f"route-mismatch-{active_arm.name}-to-{candidate_arm.name}" close_position(idx, exit_reason) action_taken = "exit_route_mismatch" elif due_rebalance: exit_reason = f"periodic-rebalance-due-{active_arm.name}" prev_arm = active_arm close_position(idx, exit_reason) action_taken = "rebalance_due" # Re-enter same valid route if the current candidate still matches it. if candidate_arm.name == prev_arm.name and candidate_arm.action == "trade": if enter_position(idx, candidate_arm): action_taken = "rebalance_reenter" else: action_taken = "hold" if not deployed: if cfg.no_reentry_after_exit and action_taken.startswith("exit"): unused_capital = portfolio_cash position_cash = 0.0 # Force one decision interval of cooldown after any exit. # Without this, the router can health-exit and re-enter on the same stale signal. action_taken = action_taken + "_cooldown" elif candidate_arm.action == "trade": if enter_position(idx, candidate_arm): action_taken = "enter" if action_taken in {"idle", "exit_health", "exit_route_mismatch"} else action_taken else: action_taken = "idle_enter_failed" else: unused_capital = portfolio_cash position_cash = 0.0 action_taken = action_taken if action_taken.startswith("exit") else "idle" route_for_log = active_arm if deployed else candidate_arm route_counts[route_for_log.name] = route_counts.get(route_for_log.name, 0) + 1 reason_counts[route_for_log.reason] = reason_counts.get(route_for_log.reason, 0) + 1 action_counts[action_taken] = action_counts.get(action_taken, 0) + 1 if route_for_log.action == "idle" or action_taken == "idle": idle_decisions += 1 if route_for_log.reason.startswith("capacity-fail"): capacity_fail_decisions += 1 log_row: Dict[str, Any] = { "npz": npz_path, "dataset": dataset_key, "pool_key": pool_key, "fee_scenario": fee_scenario, "ts": int(ts[idx]), "iso_time": ts_to_iso(int(ts[idx])), "event_idx": int(idx), "price": float(price[idx]), "candidate_route": candidate_arm.name, "active_route": active_arm.name if deployed else "idle", "action_taken": action_taken, "exit_reason": exit_reason, "health_reason": health_reason, "capital_usd": float(active_arm.capital_usd if deployed else 0.0), "portfolio_cash_usd": float(portfolio_cash), "unused_capital_usd": float(unused_capital), "position_cash_usd": float(position_cash), "fees_uncollected_usd": float(fees_uncollected), "next_rebalance_ts": int(next_rebalance_ts), **{k: float(v) for k, v in feat.items()}, **{k: float(v) for k, v in health_feat.items()}, } logs.append(log_row) next_ts = int(ts[idx]) + decision_interval next_idx = int(np.searchsorted(ts, next_ts, side="left")) next_idx = min(max(idx + 1, next_idx), n) if deployed and our_liq > 0.0: sl = slice(idx, next_idx) pseg = price[sl] in_range = (pseg >= lower) & (pseg <= upper) share = liquidity_share(our_liq, active_liq[sl]) fee_events = np.zeros_like(pseg, dtype=np.float64) fee_events[in_range] = input_usd[sl][in_range] * fee_rate * share[in_range] fees_cum = np.cumsum(fee_events) unit = unit_value_curve_from_sqrt(sqrt_price[sl], pseg, lower, upper, dec0, dec1) pos_value = our_liq * unit sleeve_equity = pos_value + fees_uncollected + fees_cum deployed_equity[sl] = sleeve_equity pos_arr[sl] = pos_value equity_total[sl] = unused_capital + sleeve_equity in_arr[sl] = in_range.astype(np.int8) share_arr[sl] = share segment_fees = float(fees_cum[-1]) if len(fees_cum) else 0.0 fees_uncollected += segment_fees fees_earned_total += segment_fees else: sl = slice(idx, next_idx) deployed_equity[sl] = 0.0 pos_arr[sl] = 0.0 equity_total[sl] = portfolio_cash in_arr[sl] = 0 share_arr[sl] = 0.0 idx = next_idx share_in = share_arr[in_arr.astype(bool)] avg_s, p95_s, p99_s, max_s = share_stats_pct(share_in) profit_total = float(equity_total[-1] - cfg.total_capital_usd) deployed_denom = max(1e-9, max([a.capital_usd for a in [BIO_BULL_ARM, CHECK_TINY_BEAR_ARM] if route_counts.get(a.name, 0) > 0] or [0.0])) price_return_pct = float((price[-1] / price[0] - 1.0) * 100.0) hodl50 = cfg.total_capital_usd / 2.0 + (cfg.total_capital_usd / 2.0 / float(price[0])) * price row: Dict[str, Any] = { "strategy": "portfolio_router_v4", "routing_mode": cfg.routing_mode, "dataset_key": dataset_key, "pool_key": pool_key, "fee_scenario": fee_scenario, "fee_rate": float(fee_rate), "total_capital_usd": float(cfg.total_capital_usd), "equity_end_usd": float(equity_total[-1]), "profit_usd": profit_total, "return_total_pct": float(profit_total / cfg.total_capital_usd * 100.0), "mdd_total_pct": max_drawdown_pct(equity_total), "return_on_deployed_capital_pct": float(profit_total / deployed_denom * 100.0) if deployed_denom > 0 else 0.0, "fees_earned_total": float(fees_earned_total), "fees_reinvested": float(fees_reinvested), "fees_uncollected_end": float(fees_uncollected), "rebalance_costs": float(costs_cum), "position_value_end_usd": float(pos_arr[-1]), "time_in_range_pct": float(np.asarray(in_arr, dtype=bool).mean() * 100.0), "avg_liquidity_share_pct_when_in_range": avg_s, "p95_liquidity_share_pct_when_in_range": p95_s, "p99_liquidity_share_pct_when_in_range": p99_s, "max_liquidity_share_pct_when_in_range": max_s, "rebalances": int(rebalances), "entries": int(entries), "exits": int(exits), "health_exits": int(health_exits), "decisions": int(decisions), "idle_decisions": int(idle_decisions), "idle_decision_pct": float(idle_decisions / decisions * 100.0) if decisions else 0.0, "capacity_fail_decisions": int(capacity_fail_decisions), "route_counts_json": json.dumps(route_counts, sort_keys=True), "reason_counts_json": json.dumps(reason_counts, sort_keys=True), "action_counts_json": json.dumps(action_counts, sort_keys=True), "route_top": max(route_counts.items(), key=lambda kv: kv[1])[0] if route_counts else "", "reason_top": max(reason_counts.items(), key=lambda kv: kv[1])[0] if reason_counts else "", "action_top": max(action_counts.items(), key=lambda kv: kv[1])[0] if action_counts else "", "hodl50_return_pct": float((hodl50[-1] / cfg.total_capital_usd - 1.0) * 100.0), "vs_hodl50_usd": float(equity_total[-1] - hodl50[-1]), "price_start": float(price[0]), "price_end": float(price[-1]), "price_return_pct": price_return_pct, "script_version": SCRIPT_VERSION, } row["pnl_mdd_total"] = pnl_mdd_from_row(row) return row, logs def run_one_npz(npz_path: str, args: argparse.Namespace, cfg: RouterConfig) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: data = load_npz(npz_path) meta = data.get("meta", {}) or {} time_from, time_to = infer_time_range(data, args.time_from, args.time_to) d = filter_time(data, time_from, time_to) dec0_raw = args.dec0 or int(meta.get("dec0", 6)) dec1_raw = args.dec1 or int(meta.get("dec1", 18)) dec0, dec1 = canonical_decimals(meta, dec0_raw, dec1_raw) pool_key = pool_key_from_meta_path(meta, npz_path) dataset_key = dataset_key_from_path(npz_path, meta) if args.strict_quote_check and pool_key.startswith("unsupported"): raise ValueError(f"Unsupported non-USD quote pool: {pool_key} for {npz_path}") ts = d["ts"].astype(np.int64) price = d["price"].astype(np.float64) input_usd = d["input_usd"].astype(np.float64) active_liq = d["active_liquidity"].astype(np.float64) rows: List[Dict[str, Any]] = [] logs: List[Dict[str, Any]] = [] for fee_name, fee_rate in parse_fee_specs(args.fee_rates, meta.get("fee_rate")): row, decision_logs = run_router( npz_path=npz_path, dataset_key=dataset_key, pool_key=pool_key, price=price, input_usd=input_usd, active_liq=active_liq, ts=ts, dec0=dec0, dec1=dec1, fee_rate=fee_rate, fee_scenario=fee_name, cfg=cfg, ) row.update({ "npz": str(npz_path), "dataset": Path(npz_path).name, "time_from": time_from, "time_to": time_to, "events": int(len(price)), "dec0": int(dec0), "dec1": int(dec1), }) row["pnl_mdd_total"] = pnl_mdd_from_row(row) row["strict_pass"] = strict_pass(row, args) row["score"] = router_score(row, args) rows.append(row) logs.extend(decision_logs) return rows, logs # ───────────────────────────────────────────────────────────────────────────── # CLI # ───────────────────────────────────────────────────────────────────────────── def parse_npzs(args: argparse.Namespace) -> List[str]: out: List[str] = [] if args.npz: out.append(args.npz) if args.npzs: out.extend(args.npzs) if args.npz_glob: for pattern in args.npz_glob: out.extend(str(p) for p in sorted(Path().glob(pattern))) seen = set() unique: List[str] = [] for p in out: if p not in seen: seen.add(p) unique.append(p) if not unique: raise SystemExit("Use --npz, --npzs, or --npz-glob") return unique def build_config(args: argparse.Namespace) -> RouterConfig: return RouterConfig( routing_mode=args.routing_mode, total_capital_usd=args.total_capital_usd, lookback_hours=args.lookback_hours, decision_hours=args.decision_hours, gas_usd=args.gas_usd, swap_cost_bps=args.swap_cost_bps, max_current_share_pct=args.max_current_liquidity_share_pct, max_deploy_fraction=args.max_deploy_fraction, bio_micro_min_drift_pct=args.bio_micro_min_drift_pct, bio_micro_min_trend_ratio=args.bio_micro_min_trend_ratio, bio_macro_lookback_hours=args.bio_macro_lookback_hours, bio_macro_kill_drift_pct=args.bio_macro_kill_drift_pct, bio_bull_min_fee_budget_pct_day=args.bio_bull_min_fee_budget_pct_day, bio_bull_max_toxicity=args.bio_bull_max_toxicity, check_bear_max_drift_pct=args.check_bear_max_drift_pct, check_bear_min_fee_budget_pct_day=args.check_bear_min_fee_budget_pct_day, min_events_per_hour=args.min_events_per_hour, allow_synd=args.allow_synd, no_bootstrap=args.no_bootstrap, strict_quote_check=args.strict_quote_check, prewarm_hours=args.prewarm_hours, warmup_probe_hours=args.warmup_probe_hours, warmup_min_span_fraction=args.warmup_min_span_fraction, warmup_min_return_pct=args.warmup_min_return_pct, warmup_max_mdd_pct=args.warmup_max_mdd_pct, warmup_min_pnl_mdd=args.warmup_min_pnl_mdd, warmup_min_time_in_range_pct=args.warmup_min_time_in_range_pct, warmup_max_p99_share_pct=args.warmup_max_p99_share_pct, warmup_max_max_share_pct=args.warmup_max_max_share_pct, dema_enable=args.dema_enable, dema_fast_hours=args.dema_fast_hours, dema_slow_hours=args.dema_slow_hours, dema_slope_hours=args.dema_slope_hours, dema_min_slow_slope_pct=args.dema_min_slow_slope_pct, dema_require_price_above_slow=not args.dema_no_price_above_slow, dema_require_fast_above_slow=not args.dema_no_fast_above_slow, health_check_enable=args.health_check_enable, health_min_micro_drift_pct=args.health_min_micro_drift_pct, health_probe_enable=args.health_probe_enable, health_probe_hours=args.health_probe_hours, health_probe_min_return_pct=args.health_probe_min_return_pct, health_probe_max_mdd_pct=args.health_probe_max_mdd_pct, health_probe_min_pnl_mdd=args.health_probe_min_pnl_mdd, health_probe_min_time_in_range_pct=args.health_probe_min_time_in_range_pct, health_probe_max_p99_share_pct=args.health_probe_max_p99_share_pct, exit_on_route_mismatch=(args.exit_on_route_mismatch and not args.no_exit_on_route_mismatch), rebalance_when_due=not args.no_rebalance_when_due, no_reentry_after_exit=args.no_reentry_after_exit, health_max_total_dd_pct=args.health_max_total_dd_pct, ) def add_common_args(ap: argparse.ArgumentParser) -> None: ap.add_argument("--npz", default="") ap.add_argument("--npzs", nargs="*", default=[]) ap.add_argument("--npz-glob", nargs="*", default=[]) ap.add_argument("--out-dir", required=True) ap.add_argument("--fee-rates", default="metadata") ap.add_argument("--time-from", default="") ap.add_argument("--time-to", default="") ap.add_argument("--dec0", type=int, default=0) ap.add_argument("--dec1", type=int, default=0) ap.add_argument("--routing-mode", choices=["locked_candidates", "rolling", "rolling_warmup", "rolling_v4"], default="rolling_v4") ap.add_argument("--total-capital-usd", type=float, default=600.0) ap.add_argument("--lookback-hours", type=float, default=48.0) ap.add_argument("--decision-hours", type=float, default=12.0) ap.add_argument("--gas-usd", type=float, default=0.0) ap.add_argument("--swap-cost-bps", type=float, default=0.0) ap.add_argument("--max-current-liquidity-share-pct", type=float, default=15.0) ap.add_argument("--max-deploy-fraction", type=float, default=1.0) ap.add_argument("--bio-micro-min-drift-pct", type=float, default=1.0) ap.add_argument("--bio-micro-min-trend-ratio", type=float, default=0.02) ap.add_argument("--bio-macro-lookback-hours", type=float, default=168.0) ap.add_argument("--bio-macro-kill-drift-pct", type=float, default=-5.0) ap.add_argument("--bio-bull-min-fee-budget-pct-day", type=float, default=0.15) ap.add_argument("--bio-bull-max-toxicity", type=float, default=0.85) ap.add_argument("--check-bear-max-drift-pct", type=float, default=-5.0) ap.add_argument("--check-bear-min-fee-budget-pct-day", type=float, default=0.15) ap.add_argument("--min-events-per-hour", type=float, default=1.0) ap.add_argument("--allow-synd", action="store_true") ap.add_argument("--no-bootstrap", action="store_true") ap.add_argument("--strict-quote-check", action="store_true") ap.add_argument("--prewarm-hours", type=float, default=168.0) ap.add_argument("--warmup-probe-hours", type=float, default=168.0) ap.add_argument("--warmup-min-span-fraction", type=float, default=0.80) ap.add_argument("--warmup-min-return-pct", type=float, default=0.0) ap.add_argument("--warmup-max-mdd-pct", type=float, default=12.0) ap.add_argument("--warmup-min-pnl-mdd", type=float, default=1.5) ap.add_argument("--warmup-min-time-in-range-pct", type=float, default=40.0) ap.add_argument("--warmup-max-p99-share-pct", type=float, default=10.0) ap.add_argument("--warmup-max-max-share-pct", type=float, default=25.0) ap.add_argument("--dema-enable", action="store_true") ap.add_argument("--dema-fast-hours", type=float, default=48.0) ap.add_argument("--dema-slow-hours", type=float, default=168.0) ap.add_argument("--dema-slope-hours", type=float, default=24.0) ap.add_argument("--dema-min-slow-slope-pct", type=float, default=0.0) ap.add_argument("--dema-no-price-above-slow", action="store_true") ap.add_argument("--dema-no-fast-above-slow", action="store_true") ap.add_argument("--health-check-enable", action="store_true") ap.add_argument("--health-min-micro-drift-pct", type=float, default=-5.0) ap.add_argument("--health-probe-enable", action="store_true") ap.add_argument("--health-probe-hours", type=float, default=72.0) ap.add_argument("--health-probe-min-return-pct", type=float, default=-2.0) ap.add_argument("--health-probe-max-mdd-pct", type=float, default=12.0) ap.add_argument("--health-probe-min-pnl-mdd", type=float, default=0.0) ap.add_argument("--health-probe-min-time-in-range-pct", type=float, default=25.0) ap.add_argument("--health-probe-max-p99-share-pct", type=float, default=10.0) ap.add_argument("--exit-on-route-mismatch", action="store_true") ap.add_argument("--no-exit-on-route-mismatch", action="store_true") ap.add_argument("--no-rebalance-when-due", action="store_true") ap.add_argument("--no-reentry-after-exit", action="store_true") ap.add_argument("--health-max-total-dd-pct", type=float, default=0.0) ap.add_argument("--strict-mdd-pct", type=float, default=20.0) ap.add_argument("--min-pnl-mdd", type=float, default=2.0) ap.add_argument("--max-avg-liquidity-share-pct", type=float, default=3.0) ap.add_argument("--max-p95-liquidity-share-pct", type=float, default=5.0) ap.add_argument("--max-p99-liquidity-share-pct", type=float, default=10.0) ap.add_argument("--max-liquidity-share-pct", type=float, default=25.0) ap.add_argument("--target-mdd-pct", type=float, default=20.0) ap.add_argument("--w-mdd", type=float, default=2.0) ap.add_argument("--w-avg-share", type=float, default=5.0) ap.add_argument("--w-p95-share", type=float, default=10.0) ap.add_argument("--w-p99-share", type=float, default=3.0) ap.add_argument("--w-max-share", type=float, default=0.5) ap.add_argument("--w-rebalance", type=float, default=0.02) def write_handoff(out_dir: Path, rows: List[Dict[str, Any]], args: argparse.Namespace) -> None: passed = [r for r in rows if bool(r.get("strict_pass"))] lines = [ "# NEXT_HANDOFF — portfolio_router_v4", "", f"Generated by `{SCRIPT_VERSION}`.", "", "## What changed vs v3", "", "- Added DEMA trend gate for BIO-like entry.", "- Added active-position health-check every decision interval.", "- Does not blindly close/reopen every health check; it holds until health fails or periodic rebalance is due.", "- Writes action counts, entries, exits, and health exit counts.", "", "## Result", "", f"- rows tested: {len(rows)}", f"- strict pass rows: {len(passed)}", "", ] for r in rows: lines.append( f"- `{r.get('dataset','')}`: pass={r.get('strict_pass')} " f"return_total={float(r.get('return_total_pct',0)):.4f}% " f"mdd={float(r.get('mdd_total_pct',0)):.4f}% " f"pnl_mdd={float(r.get('pnl_mdd_total',0)):.4f} " f"entries={int(r.get('entries',0))} health_exits={int(r.get('health_exits',0))} " f"route_top=`{r.get('route_top','')}` action_top=`{r.get('action_top','')}`" ) lines.extend([ "", "## Live status", "", "Still not signed-live ready. Use `paper_live_virtual_lp_v1.py` first.", "", ]) (out_dir / "NEXT_HANDOFF.md").write_text("\n".join(lines), encoding="utf-8") def main() -> None: ap = argparse.ArgumentParser(description="Portfolio router v4: warmup + DEMA + health-check CL LP replay") add_common_args(ap) args = ap.parse_args() cfg = build_config(args) npzs = parse_npzs(args) out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) all_rows: List[Dict[str, Any]] = [] all_logs: List[Dict[str, Any]] = [] errors: List[Dict[str, Any]] = [] for npz_path in npzs: try: rows, logs = run_one_npz(npz_path, args, cfg) all_rows.extend(rows) all_logs.extend(logs) except Exception as exc: errors.append({"npz": str(npz_path), "error": repr(exc)}) all_rows = sorted(all_rows, key=lambda r: float(r.get("score", -1e18)), reverse=True) write_rows_csv(out_dir / "portfolio_router_results.csv", all_rows) write_rows_csv(out_dir / "portfolio_router_decision_log.csv", all_logs) write_rows_csv(out_dir / "portfolio_router_errors.csv", errors) write_handoff(out_dir, all_rows, args) print(json.dumps({ "script_version": SCRIPT_VERSION, "out_dir": str(out_dir), "routing_mode": args.routing_mode, "npzs_requested": len(npzs), "rows": len(all_rows), "decision_log_rows": len(all_logs), "errors": len(errors), "strict_pass_rows": int(sum(1 for r in all_rows if bool(r.get("strict_pass")))), }, indent=2, ensure_ascii=False)) if errors and not all_rows: raise SystemExit(2) if __name__ == "__main__": main()