Upload Quasar_axrvi_ranker.py

Quasar_axrvi_ranker.py

@@ -3939,10 +3939,6 @@ class QCSAMCrossAssetLayer(nn.Module):
     """
     Live QCSAM/FABLE cross-asset core — integrates the full quantum pipeline
     into the AXRVINet forward pass.
-    """
-    """
-    Live QCSAM/FABLE cross-asset core — integrates the full quantum pipeline
-    into the AXRVINet forward pass.
 
     This is the component that turns dead QCSAM/FABLE code into the active
     cross-asset interaction engine of the ranker.
@@ -3973,6 +3969,14 @@ class QCSAMCrossAssetLayer(nn.Module):
         qffn_output    : (B, hilbert_dim) cdouble
         readout_input  : (B, N, hilbert_dim) float32 (real part broadcast)
         layer_output   : (B, N, d_model) float32
+
+    Checkpoint notes:
+        All submodules (adapter_proj, qmha, qffn, readout, residual_gate) are
+        registered nn.Module / nn.Parameter attributes and are therefore fully
+        captured by AXRVINet.state_dict().  FABLECLCU inside qmha is eagerly
+        initialised via num_assets so its weights are in state_dict from the
+        very first save.  _clcu_n_patches_buf (persistent buffer) guards
+        against accidental re-initialisation on load_state_dict.
     """
 
     # Registry flag — set True after first successful forward.
@@ -4170,7 +4174,7 @@ class AXRVINet(nn.Module):
         # directly (e.g. test_components) still get the right architecture.
         if feature_dim is not None or seq_len is not None:
             import copy
-            config = cssopy.copy(config)
+            config = copy.copy(config)   # FIX 8: was cssopy.copy (NameError typo)
             if feature_dim is not None:
                 config.feature_dim = feature_dim
             if seq_len is not None:
@@ -5637,7 +5641,9 @@ class HybridTrainer:
             self.lambda_align= ckpt.get("lambda_align", self.lambda_align)  # QCSAM
             self.rank_margin = ckpt.get("rank_margin",  self.rank_margin)
             if "loss_history" in ckpt:
-                self.loss_history.extend(ckpt["loss_history"])
+                # FIX 4c: replace (not extend) to avoid doubling on resume
+                _maxlen = self.loss_history.maxlen
+                self.loss_history = deque(ckpt["loss_history"], maxlen=_maxlen)
             logger.info(f"✅ Model loaded ← {path} | resumed from train_step={self.train_step}")
         except FileNotFoundError:
             logger.info(f"ℹ️  No checkpoint at {path} — starting fresh")
@@ -6677,7 +6683,7 @@ class QuasarAXRVIBridge:
         hub_ws_url:      str  = os.environ.get("QUASAR_HUB_URL", "ws://localhost:7860/ws/subscribe"),
         enable_logging:  bool = True,
         checkpoint_dir:  str  = "./Ranker6",   # folder for full-state checkpoints
-        resume:          bool = False,         # start afresh by default; set True to resume
+        resume:          bool = True,          # FIX 1c: default True — always resume from latest checkpoint
         hf_repo_id:      Optional[str] = "KarlQuant/k1rl-checkpoints",  # HF Dataset repo
     ):
         self.config          = config or AssetRankerConfig()
@@ -8242,22 +8248,22 @@ class QuasarAXRVIBridge:
                         f"ce={train_result.get('ce', 0):.4f} | "
                         f"batch_size={len(batch)}"
                     )
-                    # ── HF SYNC: mirror trainer .pt to HF on every training step ──
-                    # trainer.save() (via step_scheduler) writes quasar_axrvi_v6.pt
-                    # using a plain torch.save() with no HF awareness.  Queue the
-                    # upload here so HF always reflects the latest trained weights.
-                    # Non-blocking — handed off to the background HFSyncWorker thread.
-                    self.checkpoint_mgr._hf.queue_upload(
-                        local_path=self.config.model_path,
-                        step=self.trainer.train_step,
-                        metadata={
-                            "reason":     "train_step",
-                            "loss":       train_result.get("total", 0.0),
-                            "rl":         train_result.get("rl",    0.0),
-                            "ce":         train_result.get("ce",    0.0),
-                            "train_step": self.trainer.train_step,
-                        },
-                    )
+                    # ── HF SYNC: mirror trainer .pt to HF periodically ────────────
+                    # FIX 6: Upload every 10 training steps (not every step) to avoid
+                    # flooding HF with commits.  The full-state autosave (maybe_save)
+                    # handles persistence every 5 min regardless.
+                    if self.trainer.train_step % 10 == 0:
+                        self.checkpoint_mgr._hf.queue_upload(
+                            local_path=self.config.model_path,
+                            step=self.trainer.train_step,
+                            metadata={
+                                "reason":     "train_step",
+                                "loss":       train_result.get("total", 0.0),
+                                "rl":         train_result.get("rl",    0.0),
+                                "ce":         train_result.get("ce",    0.0),
+                                "train_step": self.trainer.train_step,
+                            },
+                        )
                 else:
                     logger.warning(
                         "⚠️  [Training] train_on_batch returned empty result — "
@@ -8576,6 +8582,11 @@ class HFDatasetCheckpointManager:
         """
         Upload *local_path* to HF Dataset as step_XXXXXXX.pt.
         Updates ranker_index.json and latest sentinel on HF.
+
+        FIX 3: All three files (.pt, index, sentinel) are batched into a
+        SINGLE commit via create_commit + CommitOperationAdd, reducing HF
+        commit usage from 3 → 1 per save.  This prevents the 256/hr rate
+        limit from being blown by the 11-space fleet.
         Returns True on success, False on any error (non-raising).
         """
         if not self._ensure_hf():
@@ -8586,95 +8597,95 @@ class HFDatasetCheckpointManager:
             logger.warning(f"☁️  HF upload skipped — file not found: {local_path}")
             return False
 
+        import tempfile
+        import json as _json
+
         hf_path = self._hf_ckpt_path(step)
         try:
             if self.verbose:
-                logger.info(f"☁️  Uploading {hf_path} → {self.repo_id} …")
+                logger.info(f"☁️  Uploading {hf_path} → {self.repo_id} (batched commit) …")
 
-            with open(local_path, "rb") as fh:
-                self._hfapi.upload_file(
-                    path_or_fileobj=fh,
-                    path_in_repo=hf_path,
-                    repo_id=self.repo_id,
-                    repo_type="dataset",
-                    commit_message=f"checkpoint step={step:07d}",
-                )
+            from huggingface_hub import CommitOperationAdd
+
+            # ── Build updated index payload ────────────────────────────────────
+            existing: Dict[str, Any] = {}
+            try:
+                with tempfile.TemporaryDirectory() as tmpdir:
+                    idx_file = self._hf_hub_dl(
+                        repo_id=self.repo_id,
+                        filename=self.INDEX_FILENAME,
+                        repo_type="dataset",
+                        token=self.token,
+                        local_dir=tmpdir,
+                        local_dir_use_symlinks=False,
+                    )
+                    with open(idx_file) as fh:
+                        existing = _json.load(fh)
+            except Exception:
+                pass   # no existing index — start fresh
+
+            entry: Dict[str, Any] = {
+                "step":      step,
+                "filename":  hf_path,
+                "size_mb":   round(local_path.stat().st_size / 1_048_576, 2),
+                "timestamp": datetime.now().isoformat(),
+            }
+            if metadata:
+                entry.update(metadata)
 
-            # Update index and latest sentinel
-            self._update_hf_index(step, local_path, metadata)
-            self._update_hf_latest(step)
+            if "checkpoints" not in existing:
+                existing["checkpoints"] = []
+            existing["checkpoints"] = [
+                cp for cp in existing["checkpoints"] if cp.get("step") != step
+            ]
+            existing["checkpoints"].append(entry)
+            existing["checkpoints"].sort(key=lambda x: x.get("step", 0))
+            existing["latest_step"]       = step
+            existing["last_updated"]      = datetime.now().isoformat()
+            existing["total_checkpoints"] = len(existing["checkpoints"])
+
+            index_bytes   = _json.dumps(existing, indent=2).encode()
+            sentinel_bytes = str(step).encode()
+
+            # ── Single batched commit: .pt + index + sentinel ──────────────────
+            with open(local_path, "rb") as pt_fh:
+                pt_bytes = pt_fh.read()
+
+            self._hfapi.create_commit(
+                repo_id=self.repo_id,
+                repo_type="dataset",
+                operations=[
+                    CommitOperationAdd(
+                        path_in_repo=hf_path,
+                        path_or_fileobj=pt_bytes,
+                    ),
+                    CommitOperationAdd(
+                        path_in_repo=self.INDEX_FILENAME,
+                        path_or_fileobj=index_bytes,
+                    ),
+                    CommitOperationAdd(
+                        path_in_repo=self.LATEST_SENTINEL,
+                        path_or_fileobj=sentinel_bytes,
+                    ),
+                ],
+                commit_message=f"checkpoint step={step:07d}",
+            )
 
             self._upload_count    += 1
             self._last_upload_step = step
-            logger.info(f"☁️  ✅ HF upload complete | step={step} | repo={self.repo_id}")
+            logger.info(
+                f"☁️  ✅ HF upload complete (1 commit) | step={step} | repo={self.repo_id}"
+            )
             return True
 
         except Exception as exc:
             logger.warning(f"☁️  ⚠️  HF upload failed (non-fatal): {exc}")
             return False
 
-    def _update_hf_index(
-        self,
-        step: int,
-        local_path: Path,
-        metadata: Optional[Dict],
-    ) -> None:
-        """Download existing HF index, append entry, re-upload."""
-        import tempfile, json as _json
-
-        existing: Dict[str, Any] = {}
-        try:
-            with tempfile.TemporaryDirectory() as tmpdir:
-                idx_file = self._hf_hub_dl(
-                    repo_id=self.repo_id,
-                    filename=self.INDEX_FILENAME,
-                    repo_type="dataset",
-                    token=self.token,
-                    local_dir=tmpdir,
-                    local_dir_use_symlinks=False,
-                )
-                with open(idx_file) as fh:
-                    existing = _json.load(fh)
-        except Exception:
-            pass   # no existing index — start fresh
-
-        entry: Dict[str, Any] = {
-            "step":      step,
-            "filename":  self._hf_ckpt_path(step),
-            "size_mb":   round(local_path.stat().st_size / 1_048_576, 2),
-            "timestamp": datetime.now().isoformat(),
-        }
-        if metadata:
-            entry.update(metadata)
-
-        # Preserve local-index schema
-        if "checkpoints" not in existing:
-            existing["checkpoints"] = []
-        existing["checkpoints"] = [
-            cp for cp in existing["checkpoints"] if cp.get("step") != step
-        ]
-        existing["checkpoints"].append(entry)
-        existing["checkpoints"].sort(key=lambda x: x.get("step", 0))
-        existing["latest_step"]    = step
-        existing["last_updated"]   = datetime.now().isoformat()
-        existing["total_checkpoints"] = len(existing["checkpoints"])
-
-        self._hfapi.upload_file(
-            path_or_fileobj=_json.dumps(existing, indent=2).encode(),
-            path_in_repo=self.INDEX_FILENAME,
-            repo_id=self.repo_id,
-            repo_type="dataset",
-            commit_message=f"update index step={step}",
-        )
-
-    def _update_hf_latest(self, step: int) -> None:
-        self._hfapi.upload_file(
-            path_or_fileobj=str(step).encode(),
-            path_in_repo=self.LATEST_SENTINEL,
-            repo_id=self.repo_id,
-            repo_type="dataset",
-            commit_message=f"update latest → step={step}",
-        )
+    # _update_hf_index and _update_hf_latest removed in FIX 3.
+    # All three files (.pt, index, sentinel) are now batched into a single
+    # create_commit call inside upload() to avoid the 3x commit-per-save
+    # rate-limit explosion.
 
     # ── Download ───────────────────────────────────────────────────────────────
 
@@ -9157,7 +9168,9 @@ class RankerCheckpointManager:
             tr.lambda_align = ckpt.get("lambda_align", tr.lambda_align)
             tr.rank_margin  = ckpt.get("rank_margin",  tr.rank_margin)
             if "loss_history" in ckpt:
-                tr.loss_history.extend(ckpt["loss_history"])
+                # FIX 4: replace (not extend) so history isn't doubled on restore
+                _maxlen = tr.loss_history.maxlen
+                tr.loss_history = deque(ckpt["loss_history"], maxlen=_maxlen)
             logger.info(f"   ✅ trainer restored | train_step={tr.train_step}")
 
         # ── Replay buffer ─────────────────────────────────────────────────────
@@ -9190,16 +9203,27 @@ class RankerCheckpointManager:
             else:
                 logger.warning(f"   ⚠️  feature_engine[{asset_id}] not in bridge.asset_buffers — skipped")
 
-        # ── Runtime counters & episode tracking ───────────────────────────────
+        # ── Runtime counters (safe to restore) ───────────────────────────────
         bridge.rank_count = ckpt.get("rank_count", bridge.rank_count)
-        restored_ep  = ckpt.get("pending_episodes",  {})
-        restored_tc  = ckpt.get("trade_tick_counts", {})
-        bridge._pending_episodes.update(restored_ep)
-        bridge._trade_tick_counts.update(restored_tc)
-        if restored_ep:
-            logger.info(f"   ✅ pending_episodes restored | count={len(restored_ep)}")
-        if restored_tc:
-            logger.info(f"   ✅ trade_tick_counts restored | count={len(restored_tc)}")
+
+        # FIX 5: Do NOT restore pending_episodes or trade_tick_counts.
+        # These reference Deriv multiplier contracts that are gone after a restart.
+        # Restoring them causes phantom open_count > 0, which blocks the top-4
+        # floor enforcer from opening any new trades and corrupts RL episode pairing.
+        # rank_count is the only runtime counter safe to restore across restarts.
+        stale_ep = ckpt.get("pending_episodes", {})
+        stale_tc = ckpt.get("trade_tick_counts", {})
+        if stale_ep:
+            logger.warning(
+                f"   ⚠️  Discarded {len(stale_ep)} stale pending_episodes from checkpoint "
+                "(Deriv contracts expired across restart — resetting to avoid phantom positions)"
+            )
+        if stale_tc:
+            logger.info(
+                f"   ℹ️  Discarded {len(stale_tc)} stale trade_tick_counts (reset on restart)"
+            )
+        # bridge._pending_episodes and bridge._trade_tick_counts stay at their
+        # freshly-initialised empty-dict state — no update needed.
 
         # Post-load validation
         assert bridge.rank_count >= 0, "rank_count must be non-negative after restore"
@@ -9740,7 +9764,7 @@ class QuasarCheckpointManager:
         self,
         ranker,                     # QUASAR AXRVI Ranker instance
         training_steps: Optional[int] = None,
-        device: str = 'cuda',
+        device: str = 'cpu',        # FIX 7: was 'cuda' — HF Spaces are CPU-only, 'cuda' crashes
         strict: bool = True,
         load_optimizer: bool = True,
         load_replay_buffer: bool = True,
@@ -9876,7 +9900,9 @@ class QuasarCheckpointManager:
             tr.lambda_align = checkpoint.get("lambda_align", tr.lambda_align)
             tr.rank_margin  = checkpoint.get("rank_margin",  tr.rank_margin)
             if "loss_history" in checkpoint:
-                tr.loss_history.extend(checkpoint["loss_history"])
+                # FIX 4b: replace (not extend) to avoid doubling history on restore
+                _maxlen = tr.loss_history.maxlen
+                tr.loss_history = deque(checkpoint["loss_history"], maxlen=_maxlen)
             print(f"   ✅ trainer scalars (train_step={tr.train_step})")
         
         # ====================================================================
@@ -9959,17 +9985,24 @@ class QuasarCheckpointManager:
                     except Exception as e:
                         print(f"   ⚠️  feature_engine[{asset_id}]: {str(e)[:50]}")
         
-        # Pending episodes: bridge._pending_episodes
-        if 'pending_episodes' in checkpoint and hasattr(ranker, '_pending_episodes'):
-            ranker._pending_episodes.update(checkpoint['pending_episodes'])
-            print(f"   ✅ pending_episodes ({len(checkpoint['pending_episodes'])})")
-        
-        # Tick counts: bridge._trade_tick_counts
-        if 'trade_tick_counts' in checkpoint and hasattr(ranker, '_trade_tick_counts'):
-            ranker._trade_tick_counts.update(checkpoint['trade_tick_counts'])
-            print(f"   ✅ trade_tick_counts ({len(checkpoint['trade_tick_counts'])})")
-        
-        # Runtime counters
+        # FIX 5b: Do NOT restore pending_episodes or trade_tick_counts.
+        # These reference Deriv multiplier contracts that are gone after a restart.
+        # Restoring them causes phantom open_count > 0, blocking the top-4 floor
+        # enforcer and corrupting RL episode pairing with dead contract IDs.
+        stale_ep = checkpoint.get('pending_episodes', {})
+        stale_tc = checkpoint.get('trade_tick_counts', {})
+        if stale_ep:
+            print(
+                f"   ⚠️  Discarded {len(stale_ep)} stale pending_episodes "
+                "(Deriv contracts expired across restart — phantom positions prevented)"
+            )
+        if stale_tc:
+            print(
+                f"   ℹ️  Discarded {len(stale_tc)} stale trade_tick_counts (reset on restart)"
+            )
+        # _pending_episodes and _trade_tick_counts stay at fresh empty-dict state.
+
+        # Runtime counters (safe to restore)
         if 'rank_count' in checkpoint and hasattr(ranker, 'rank_count'):
             ranker.rank_count = checkpoint['rank_count']
             print(f"   ✅ rank_count={ranker.rank_count}")
@@ -10189,7 +10222,7 @@ async def run_live_trading_system(
     enable_logging:  bool = True,
     shreve_config:   Optional[ShreveConfig] = None,
     checkpoint_dir:  str  = "./Ranker6",
-    resume:          bool = False,   # start fresh by default
+    resume:          bool = True,    # FIX 2: default True — always resume from checkpoint
     hf_repo_id:      Optional[str] = "KarlQuant/k1rl-checkpoints",  # HF Dataset repo
 ) -> None:
     config = AssetRankerConfig(
@@ -10504,8 +10537,8 @@ def _parse_args():
                         help="[S7] Gate E martingale deviation threshold (default 0.05)")
     parser.add_argument("--checkpoint-dir", default="./Ranker6",
                         help="Directory for full-state checkpoints (default ./Ranker6)")
-    parser.add_argument("--resume", action="store_true",
-                        help="Resume training from the latest saved checkpoint (default: start fresh)")
+    parser.add_argument("--no-resume", dest="no_resume", action="store_true",
+                        help="Disable checkpoint resume on startup (default: always resume from latest checkpoint)")
     parser.add_argument("--hf-repo", default=None,
                         metavar="OWNER/REPO",
                         help="Hugging Face Dataset repo for checkpoint sync "
@@ -10562,7 +10595,7 @@ if __name__ == "__main__":
         hub_ws_url      = args.hub,
         enable_logging  = not args.no_logs,
         checkpoint_dir  = args.checkpoint_dir,
-        resume          = args.resume,
+        resume          = not args.no_resume,   # FIX 1: default True (always resume)
         hf_repo_id      = args.hf_repo or "KarlQuant/k1rl-checkpoints",
     )
 
@@ -10585,7 +10618,7 @@ if __name__ == "__main__":
                 hub_ws_url      = args.hub,
                 enable_logging  = not args.no_logs,
                 checkpoint_dir  = args.checkpoint_dir,   # FIX 1: was silently ignored
-                resume          = args.resume,            # FIX 1: was silently ignored
+                resume          = not args.no_resume,      # FIX 2: default True (always resume)
                 hf_repo_id      = args.hf_repo or "KarlQuant/k1rl-checkpoints",
             ))
     except KeyboardInterrupt: