Update ml_engine/data_manager.py

ml_engine/data_manager.py

@@ -1,6 +1,6 @@
 # ============================================================
 # 📂 ml_engine/data_manager.py 
-# (V58.0 - GEM-Architect: Asset-Context Edition)
+# (V58.1 - GEM-Architect: Asset-Context Edition + Fixed)
 # ============================================================
 
 import asyncio
@@ -28,8 +28,7 @@ class DataManager:
         self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
         self.r2_service = r2_service
-        # Pass the hub instance later or via global context if strictly necessary, 
-        # but better to handle regime logic internally or pass config.
+        # Pass the hub instance later via method arguments
         self.adaptive_hub_ref = None 
 
         self.exchange = ccxt.kucoin({
@@ -47,7 +46,7 @@ class DataManager:
             'UP', 'DOWN', 'BEAR', 'BULL', '3S', '3L', 'USDD', 'USDP', 'HT', 'KCS'
         ]
         
-        print(f"📦 [DataManager V58.0] Quality Gate & Context Engine Active.")
+        print(f"📦 [DataManager V58.1] Quality Gate & Context Engine Active.")
 
     async def initialize(self):
         self.http_client = httpx.AsyncClient(timeout=30.0)
@@ -64,6 +63,18 @@ class DataManager:
         if self.http_client: await self.http_client.aclose()
         if self.exchange: await self.exchange.close()
 
+    # ✅ FIXED: Restored Missing Method for Startup Sequence
+    async def load_contracts_from_r2(self):
+        if not self.r2_service: return
+        try:
+            self.contracts_db = await self.r2_service.load_contracts_db_async()
+            print(f"   📂 [DataManager] Contracts DB loaded: {len(self.contracts_db)} items.")
+        except Exception as e: 
+            print(f"   ⚠️ [DataManager] Failed to load contracts: {e}")
+            self.contracts_db = {}
+            
+    def get_contracts_db(self): return self.contracts_db
+
     # ==================================================================
     # 🛡️ Stage 0: The "Anti-Junk" Gate (Quality Control)
     # ==================================================================
@@ -204,10 +215,7 @@ class DataManager:
         print(f"   🧬 [Stage 1] Diagnosing 4H Regime for {len(candidates)} assets...")
         
         # 2. Parallel Diagnosis
-        # We need to fetch 4H data for all of them. 
-        # To avoid rate limits, we might batch them or use asyncio.gather with care.
-        # Assuming KuCoin handles ~80 requests quickly enough or ccxt throttles automatically.
-        
+        # Fetching 4H data for all candidates
         tasks = [self._determine_4h_regime(c['symbol']) for c in candidates]
         regime_results = await asyncio.gather(*tasks, return_exceptions=True)
         
@@ -238,4 +246,6 @@ class DataManager:
     async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
         try: return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
         except: return []
-    def get_contracts_db(self): return self.contracts_db
+    async def get_order_book_snapshot(self, symbol: str, limit: int = 20) -> Dict[str, Any]:
+        try: return await self.exchange.fetch_order_book(symbol, limit)
+        except: return {}