learning_hub/adaptive_hub.py

# ==============================================================================
# 🧠 learning_hub/adaptive_hub.py 
# (V68.1 - GEM-Architect: Realistic Oracle Calibration [0.60])
# ==============================================================================

import json
import asyncio
from typing import Dict, Any

try:
    from ml_engine.processor import SystemLimits
except ImportError:
    SystemLimits = None

class StrategyDNA:
    def __init__(self, name, model_weights, ob_settings, filters, guards):
        self.name = name
        self.model_weights = model_weights
        self.ob_settings = ob_settings
        self.filters = filters
        self.guards = guards
        self.delta = {k: 0.0 for k in filters.keys()}
        self.guard_delta = {k: 0.0 for k in guards.keys()}
        self.trade_count = 0
        self.wins = 0

    def get_final_filters(self):
        final = {}
        for k, v in self.filters.items():
            final[k] = max(0.1, min(0.99, v + self.delta.get(k, 0.0)))
        return final

    def get_final_guards(self):
        final = {}
        for k, v in self.guards.items():
            final[k] = max(0.40, min(0.99, v + self.guard_delta.get(k, 0.0)))
        return final

    def to_dict(self):
        return {
            "name": self.name,
            "model_weights": self.model_weights,
            "ob_settings": self.ob_settings,
            "filters": self.filters,
            "guards": self.guards,
            "delta": self.delta,
            "guard_delta": self.guard_delta,
            "trade_count": self.trade_count,
            "wins": self.wins
        }

class AdaptiveHub:
    def __init__(self, r2_service=None):
        self.r2 = r2_service
        self.dna_file_key = "learning/coin_type_dna_v2.json"
        self.strategies: Dict[str, StrategyDNA] = {}
        self.tuner = None 
        print("🧠 [AdaptiveHub V68.1] Realistic Oracle Calibration (0.60) Loaded.")

    async def initialize(self):
        try:
            # ⚠️ هام: التحديث الإجباري لتطبيق العتبات الجديدة الواقعية
            # سنقوم بإنشاء القيم الافتراضية الجديدة وتجاهل القديمة الصارمة جداً لهذه الجلسة
            if self.r2:
                # محاولة قراءة الملف، لكن إذا كانت العتبات قديمة (0.90) يفضل استبدالها
                # هنا سأقوم بتحميلها، والموالف (Tuner) سيقوم بالتعديل، أو يمكنك عمل Reset من الواجهة
                data = await self.r2.get_file_json_async(self.dna_file_key)
                if data: 
                    self._load_from_dict(data)
                    print("   🧠 Loaded existing DNA from R2.")
                    
                    # 🔥 Auto-Correction: If Oracle thresh is too high, force lower it
                    for k, strat in self.strategies.items():
                        if strat.filters.get("l3_oracle_thresh", 0.90) > 0.70:
                            print(f"   ⚠️ [Correction] Lowering Oracle for {k} to 0.60")
                            strat.filters["l3_oracle_thresh"] = 0.60
                else: 
                    self._create_coin_type_defaults()
                    print("   🧠 Created NEW Realistic Defaults.")
            else:
                self._create_coin_type_defaults()
        except Exception:
            self._create_coin_type_defaults()

    def _create_coin_type_defaults(self):
        """
        Implements the 'Realistic' Profile based on user diagnostics.
        Oracle Max: 0.75 -> Set Threshold: 0.60
        L2 Gates: Lowered to allow flow.
        """
        # 1. SAFE_BOTTOM (Realistic)
        self.strategies["SAFE_BOTTOM"] = StrategyDNA(
            "SAFE_BOTTOM",
            model_weights={"titan": 0.40, "patt": 0.40, "mc": 0.20}, 
            ob_settings={"wall_ratio_limit": 0.40}, 
            filters={
                "l2_gate_titan": 0.60, "l2_gate_pattern": 0.50, "l2_gate_mc": 0.50,
                "l2_min_score": 0.65,
                "l3_oracle_thresh": 0.60, # ✅ Corrected: Realistic (>55, <75)
                "l4_sniper_thresh": 0.40
            },
            guards={
                "hydra_crash": 0.60, "hydra_giveback": 0.80, "hydra_stagnation": 0.60
            }
        )
        
        # 2. MOMENTUM_LAUNCH (Realistic)
        self.strategies["MOMENTUM_LAUNCH"] = StrategyDNA(
            "MOMENTUM_LAUNCH",
            model_weights={"titan": 0.40, "patt": 0.40, "mc": 0.20},
            ob_settings={"wall_ratio_limit": 0.35},
            filters={
                "l2_gate_titan": 0.60, "l2_gate_pattern": 0.50, "l2_gate_mc": 0.50,
                "l2_min_score": 0.65,
                "l3_oracle_thresh": 0.60, # ✅ Corrected
                "l4_sniper_thresh": 0.40
            },
            guards={
                "hydra_crash": 0.60, "hydra_giveback": 0.80, "hydra_stagnation": 0.60
            }
        )
        
        # 3. ACCUMULATION_SQUEEZE (Realistic)
        self.strategies["ACCUMULATION_SQUEEZE"] = StrategyDNA(
            "ACCUMULATION_SQUEEZE",
            model_weights={"titan": 0.40, "patt": 0.40, "mc": 0.20},
            ob_settings={"wall_ratio_limit": 0.30},
            filters={
                "l2_gate_titan": 0.60, "l2_gate_pattern": 0.50, "l2_gate_mc": 0.50,
                "l2_min_score": 0.65,
                "l3_oracle_thresh": 0.60, # ✅ Corrected
                "l4_sniper_thresh": 0.40
            },
            guards={
                "hydra_crash": 0.60, "hydra_giveback": 0.80, "hydra_stagnation": 0.60
            }
        )
        self.strategies["NONE"] = self.strategies["SAFE_BOTTOM"]

    def _load_from_dict(self, data):
        for k, v in data.items():
            self.strategies[k] = StrategyDNA(
                v["name"], v["model_weights"], v["ob_settings"], v["filters"], v["guards"]
            )
            self.strategies[k].delta = v.get("delta", {})
            self.strategies[k].guard_delta = v.get("guard_delta", {})
            self.strategies[k].trade_count = v.get("trade_count", 0)
            self.strategies[k].wins = v.get("wins", 0)

    def get_coin_type_config(self, coin_type: str) -> Dict[str, Any]:
        target = coin_type if coin_type in self.strategies else "SAFE_BOTTOM"
        dna = self.strategies[target]
        f = dna.get_final_filters()
        g = dna.get_final_guards()
        mw = dna.model_weights
        
        return {
            "regime": target, 
            # Weights
            "w_titan": mw.get("titan", 0.40),
            "w_patt": mw.get("patt", 0.40),
            "w_mc": mw.get("mc", 0.20),

            # Gates & Scores
            "l2_gate_titan": f.get("l2_gate_titan", 0.60),
            "l2_gate_pattern": f.get("l2_gate_pattern", 0.50),
            "l2_gate_mc": f.get("l2_gate_mc", 0.50),
            "l2_min_score": f.get("l2_min_score", 0.65),

            # Oracle & Sniper
            "l3_oracle_thresh": f.get("l3_oracle_thresh", 0.60), # ✅ Ensure Value is Passed
            "l4_sniper_thresh": f.get("l4_sniper_thresh", 0.40),
            "l4_ob_wall_ratio": dna.ob_settings.get("wall_ratio_limit", 0.35),
            
            # Hydra
            "hydra_crash": g.get("hydra_crash", 0.60),
            "hydra_giveback": g.get("hydra_giveback", 0.80),
            "hydra_stagnation": g.get("hydra_stagnation", 0.60)
        }

    async def save_state(self):
        if self.r2:
            data = {k: v.to_dict() for k, v in self.strategies.items()}
            await self.r2.upload_json_async(data, self.dna_file_key)

    def get_status(self) -> str:
        dna = self.strategies.get("SAFE_BOTTOM")
        if not dna: return "Initializing..."
        return f"Realistic Mode (Oracle>0.60) | SB: {dna.trade_count}/30 | ML: {self.strategies['MOMENTUM_LAUNCH'].trade_count}/30"