Upload 2 files

Quasar_axrvi_ranker.py

@@ -1344,6 +1344,21 @@ class AdaptiveNormalizer:
 
     def update_and_normalize(self, x: np.ndarray) -> np.ndarray:
         x = np.asarray(x, dtype=np.float32)
+
+        # FIX: this is an EMA (mean = (1-m)*mean + m*x). A single NaN/Inf
+        # in x poisons `mean`/`var` PERMANENTLY from that point on — every
+        # future update is (1-m)*NaN + m*(clean value) = NaN forever,
+        # even once the input goes back to being perfectly clean. That
+        # state is also what gets checkpointed, so one bad tick anywhere
+        # upstream (e.g. a degenerate correlation calc dividing by a
+        # near-zero stddev) permanently corrupts this feature dimension
+        # across every future restart. Replace non-finite inputs with the
+        # current running mean (best available "neutral" estimate, and
+        # exactly zero-effect on the mean update) so a transient bad
+        # value never enters the EMA at all.
+        if not np.all(np.isfinite(x)):
+            x = np.where(np.isfinite(x), x, self.mean if self.n > 0 else 0.0)
+
         if self.n == 0:
             self.mean = x.copy()
             self.var  = np.ones_like(x)
@@ -1354,6 +1369,17 @@ class AdaptiveNormalizer:
         # Use old_mean so the variance delta is computed before the mean shifts
         self.var  = (1 - self.momentum) * self.var  + self.momentum * (x - old_mean) ** 2
         self.n   += 1
+
+        # Self-heal: if mean/var are *already* NaN (e.g. restored from a
+        # checkpoint saved during an earlier NaN episode, before this fix
+        # existed), don't let that dead state haunt every future call —
+        # reset the poisoned dims back to a neutral prior so training can
+        # actually recover instead of being stuck at NaN forever.
+        bad = ~np.isfinite(self.mean) | ~np.isfinite(self.var)
+        if np.any(bad):
+            self.mean = np.where(bad, 0.0, self.mean)
+            self.var  = np.where(bad, 1.0, self.var)
+
         return np.clip((x - self.mean) / (np.sqrt(self.var) + self.eps), -5.0, 5.0)
 
     def state_dict(self) -> dict:
@@ -1376,6 +1402,24 @@ class AdaptiveNormalizer:
         self.var      = np.array(state["var"],  dtype=np.float32)
         self.n        = state["n"]
 
+        # FIX: existing checkpoints (including the one currently deployed)
+        # were saved *while* a NaN was already loose in the EMA state —
+        # see update_and_normalize() for how a single bad tick poisons
+        # mean/var permanently. Restoring that NaN state verbatim would
+        # keep the corruption alive forever across every future restart.
+        # Reset only the poisoned dims to a neutral prior; everything
+        # else the checkpoint learned is kept intact.
+        bad = ~np.isfinite(self.mean) | ~np.isfinite(self.var)
+        if np.any(bad):
+            n_bad = int(bad.sum())
+            self.mean = np.where(bad, 0.0, self.mean)
+            self.var  = np.where(bad, 1.0, self.var)
+            logger.warning(
+                f"[AdaptiveNormalizer] Healed {n_bad} NaN/Inf dimension(s) "
+                f"in restored normalizer state (checkpoint was saved during "
+                f"a prior NaN training episode) — reset to neutral prior."
+            )
+
 
 # ══════════════════════════════════════════════════════════════════════════════════════
 # SECTION 6 — UNIFIED FEATURE ENGINE (26-dim)
@@ -1676,8 +1720,26 @@ class UnifiedFeatureEngine:
             rets = np.array(list(self._returns)[-20:])
             vols = np.abs(rets)
             if len(rets) > 1:
-                corr    = np.corrcoef(rets[:-1], vols[1:])[0, 1]
-                raw[22] = float(np.clip(corr, -1.0, 1.0))
+                # FIX: np.corrcoef divides internally by each series' std
+                # dev. Early after a restart (or any low-tick-history
+                # window) rets[:-1] or vols[1:] can be near-constant
+                # (stddev≈0) — corrcoef then returns NaN silently, and
+                # np.clip does NOT sanitize NaN, it passes it straight
+                # through. That single NaN feature poisons the whole
+                # forward pass (every downstream tensor becomes NaN),
+                # which is exactly what was showing up as
+                # "gate=nan | align_loss=nan" on the very first forward
+                # after every restart — the checkpoint kept re-saving
+                # the corruption. Skip the correlation (leave the
+                # feature at its neutral default) when either series
+                # has ~zero variance instead of computing a meaningless
+                # ratio.
+                r_std = np.std(rets[:-1])
+                v_std = np.std(vols[1:])
+                if r_std > 1e-12 and v_std > 1e-12:
+                    corr = np.corrcoef(rets[:-1], vols[1:])[0, 1]
+                    if np.isfinite(corr):
+                        raw[22] = float(np.clip(corr, -1.0, 1.0))
 
         # [23] Martingale deviation (variance-ratio test)
         if (self.stoch_config.use_martingale_deviation
@@ -5881,6 +5943,39 @@ class HybridTrainer:
                     + self.lambda_align * l_align)
 
         # ── Backward pass (AMP-aware) ─────────────────────────────────────────
+        # FIX: if loss is NaN/Inf, backward() propagates NaN gradients to
+        # EVERY parameter that participated in the forward pass, and the
+        # very next optimizer.step() then sets all of those parameters to
+        # NaN in one shot — permanently, since NaN arithmetic never
+        # recovers on its own. That NaN state then gets checkpointed and
+        # reloaded on every future restart, which is exactly the
+        # "gate=nan | align_loss=nan" seen from fwd#1 immediately after
+        # every restore. Skip the step entirely when this happens so a
+        # single bad batch (e.g. degenerate features at startup) can't
+        # take down the whole model — the checkpoint stays as it was and
+        # training just tries again on the next cycle.
+        if not torch.isfinite(loss):
+            logger.error(
+                f"[HybridTrainer] ⚠️  Skipping optimizer step at train_step="
+                f"{self.train_step} — loss is {loss.item()} (non-finite). "
+                f"This would otherwise permanently poison model weights "
+                f"with NaN. Model/optimizer state left untouched; "
+                f"training will retry on the next batch."
+            )
+            self.optimizer.zero_grad()
+            return {
+                "total": float("nan"), "rl": float("nan"), "ce": float("nan"),
+                "rank": float("nan"), "risk": float("nan"), "ql": float("nan"),
+                "rl_norm": float("nan"), "ce_norm": float("nan"),
+                "rank_norm": float("nan"), "risk_norm": float("nan"),
+                "ql_norm": float("nan"), "div": float("nan"), "moe": float("nan"),
+                "gate": float("nan"), "crps": float("nan"), "rent": float("nan"),
+                "align": float("nan"), "grad_norm": 0.0,
+                "reward_mean": 0.0, "reward_std": 0.0, "reward_absmax": 0.0,
+                "step": self.train_step,
+                "skipped_nan_step": True,
+            }
+
         self.optimizer.zero_grad()
         self.scaler.scale(loss).backward()
         # Unscale before clipping so grad norms are in the original fp32 scale
@@ -10198,6 +10293,33 @@ class RankerCheckpointManager:
                 if incompatible.unexpected_keys:
                     logger.warning(f"[Restore] axrvi_net unexpected keys: {incompatible.unexpected_keys}")
                 logger.info("   ✅ axrvi_net restored")
+
+                # FIX: if a prior session's loss went NaN and its
+                # optimizer.step() ran anyway, every parameter that
+                # received a gradient that step becomes NaN permanently
+                # (NaN arithmetic never self-corrects) — and that NaN
+                # state is exactly what gets checkpointed and re-restored
+                # here on every future run. Detect it and reinitialise
+                # only the poisoned tensors (leaving every healthy,
+                # actually-trained parameter untouched) so a single past
+                # bad step doesn't permanently brick the model.
+                poisoned = []
+                for name, param in bridge.axrvi_net.named_parameters():
+                    if not torch.isfinite(param.data).all():
+                        with torch.no_grad():
+                            if param.dim() >= 2:
+                                torch.nn.init.xavier_uniform_(param.data, gain=0.5)
+                            else:
+                                torch.nn.init.zeros_(param.data)
+                        poisoned.append(name)
+                if poisoned:
+                    logger.warning(
+                        f"[Restore] ⚠️  Healed {len(poisoned)} NaN/Inf parameter "
+                        f"tensor(s) from a previously-corrupted checkpoint "
+                        f"(reinitialised fresh — these had already permanently "
+                        f"lost their trained values to a prior NaN training "
+                        f"step): {poisoned[:8]}{' …' if len(poisoned) > 8 else ''}"
+                    )
             else:
                 logger.info("   ⏭️  axrvi_net skipped (asset count mismatch — fresh weights kept)")
 

websocket_hub.py

@@ -53,7 +53,39 @@ from typing import Dict, List, Optional, Set
 import uvicorn
 from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse, JSONResponse
+from fastapi.responses import FileResponse, JSONResponse as _RawJSONResponse
+
+
+def _json_safe(obj):
+    """
+    Recursively replace NaN / +Inf / -Inf floats with None. These values
+    leak in from live model outputs (e.g. a confidence/profit computed
+    during a NaN training step, which is exactly what's happening
+    upstream in the ranker right now) and crash json.dumps with
+    "Out of range float values are not JSON compliant" — which previously
+    took down whichever endpoint happened to serialize them (api/state,
+    api/ranker/logs/recent, etc. — this hit each one separately since
+    every route calls JSONResponse independently).
+    """
+    if isinstance(obj, float):
+        if math.isnan(obj) or math.isinf(obj):
+            return None
+        return obj
+    if isinstance(obj, dict):
+        return {k: _json_safe(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [_json_safe(v) for v in obj]
+    return obj
+
+
+def JSONResponse(content=None, *args, **kwargs):
+    """
+    Drop-in replacement for starlette's JSONResponse that sanitizes
+    NaN/Infinity before serialization. Shadowing the name here (rather
+    than patching each call site) means every existing `JSONResponse(...)`
+    call below — and any added later — is covered automatically.
+    """
+    return _RawJSONResponse(_json_safe(content), *args, **kwargs)
 
 # ─── Logging ────────────────────────────────────────────────────────────────────────
 logging.basicConfig(
@@ -1982,40 +2014,11 @@ async def api_flips_asset(asset: str):
     return JSONResponse({"ok": True, "flip": flip, "hub_timestamp": time.time()})
 
 
-def _json_safe(obj):
-    """
-    Recursively replace NaN / +Inf / -Inf floats with None so the
-    standard-library json encoder (used by Starlette's JSONResponse,
-    which calls json.dumps with allow_nan defaulting to True but still
-    raises on Inf/NaN found *within nested dict/list structures* under
-    strict encoders) never raises "Out of range float values are not
-    JSON compliant". These values leak in from live model outputs
-    (e.g. a confidence/profit computed from a divide-by-zero or an
-    exploded loss during a bad training step) and previously took the
-    whole /api/state endpoint down with a 500 until the offending
-    snapshot aged out.
-    """
-    if isinstance(obj, float):
-        if math.isnan(obj) or math.isinf(obj):
-            return None
-        return obj
-    if isinstance(obj, dict):
-        return {k: _json_safe(v) for k, v in obj.items()}
-    if isinstance(obj, (list, tuple)):
-        return [_json_safe(v) for v in obj]
-    return obj
-
-
-def SafeJSONResponse(content: dict, **kwargs) -> JSONResponse:
-    """Drop-in replacement for JSONResponse that sanitizes NaN/Inf first."""
-    return JSONResponse(_json_safe(content), **kwargs)
-
-
 @app.get("/api/state")
 async def api_state():
     """Full dashboard state polled by hub_dashboard.html every 2 s."""
     rankings = _compute_rankings()
-    return SafeJSONResponse({
+    return JSONResponse({
         "rankings":       rankings,
         "metric_history": manager.get_metric_history(),
         "health": {