Update learning_hub/memory_store.py

learning_hub/memory_store.py

@@ -1,157 +1,68 @@
 # learning_hub/memory_store.py
+# (V13.0 - Samples Storage)
+
 import json
 import asyncio
-from datetime import datetime
-# 🔴 --- START OF CHANGE --- 🔴
-from typing import List, Dict, Optional, Any # (Import Any here)
-# 🔴 --- END OF CHANGE --- 🔴
-from .schemas import Delta, ReflectorOutput
-from .policy_engine import PolicyEngine
-
-# للتوافق مع R2Service (نفترض أنه سيتم تمريره)
-# (لا يمكننا استيراده مباشرة لتجنب التبعيات الدائرية)
+from typing import List, Any
+from .schemas import TrainingSample
 
 class MemoryStore:
-    # (The __init__ signature now correctly uses the imported Any)
-    def __init__(self, r2_service: Any, policy_engine: PolicyEngine, llm_service: Any):
-        self.r2_service = r2_service
-        self.policy_engine = policy_engine
-        self.llm_service = llm_service # نحتاجه لعملية "التقطير" (Distillation)
-        
-        self.domain_files = {
-            "strategy": "learning_deltas_strategy.json",
-            "pattern": "learning_deltas_pattern.json",
-            "indicator": "learning_deltas_indicator.json",
-            "monte_carlo": "learning_deltas_monte_carlo.json",
-            "general": "learning_deltas_general.json"
-        }
-        
-        self.distill_threshold = 50 # (من النقطة 6)
-        print("✅ Learning Hub Module: Memory Store loaded (FIXED: Imported Any)")
+    def __init__(self, r2_service: Any):
+        self.r2 = r2_service
+        self.pending_samples_key = "datasets/pending_training_batch.json"
+        self.archive_dataset_key = "datasets/historical_training_data.jsonl" # JSON Lines for big data
+        print("✅ Learning Hub: Memory Store loaded")
 
-    async def _load_deltas_from_r2(self, domain: str) -> List[Dict]:
-        """تحميل ملف الدلتا المحدد من R2"""
-        key = self.domain_files.get(domain, self.domain_files["general"])
+    async def append_training_sample(self, sample: TrainingSample):
+        """إضافة عينة لقائمة الانتظار في R2"""
         try:
-            response = self.r2_service.s3_client.get_object(Bucket="trading", Key=key)
-            deltas_list = json.loads(response['Body'].read())
-            return deltas_list
+            # 1. جلب القائمة الحالية (مكلف قليلاً لكن ضروري في بيئة stateless)
+            # تحسين: يمكن الاحتفاظ بذاكرة مؤقتة (Cache) وتحديث R2 كل 10 عينات
+            current_batch = await self._load_json_from_r2(self.pending_samples_key) or []
+            current_batch.append(sample.model_dump())
+            
+            await self._save_json_to_r2(self.pending_samples_key, current_batch)
+            
         except Exception as e:
-            # إذا فشل التحميل (مثل الملف غير موجود)، أعد قائمة فارغة
-            print(f"ℹ️ [MemoryStore] لم يتم العثور على ملف دلتا لـ {domain}، سيتم إنشاء واحد جديد.")
-            return []
+            print(f"❌ [MemoryStore] Failed to append sample: {e}")
 
-    async def _save_deltas_to_r2(self, domain: str, deltas_list: List[Dict]):
-        """حفظ ملف الدلتا المحدث إلى R2"""
-        key = self.domain_files.get(domain, self.domain_files["general"])
+    async def get_pending_samples_count(self) -> int:
         try:
-            # (Ensure list contains dicts before saving)
-            deltas_to_save = [d.model_dump() if isinstance(d, Delta) else d for d in deltas_list]
-            data_json = json.dumps(deltas_to_save, indent=2, ensure_ascii=False).encode('utf-8')
-            self.r2_service.s3_client.put_object(
-                Bucket="trading", Key=key, Body=data_json, ContentType="application/json"
-            )
-        except Exception as e:
-            print(f"❌ [MemoryStore] فشل حفظ الدلتا إلى R2: {e}")
+            current_batch = await self._load_json_from_r2(self.pending_samples_key) or []
+            return len(current_batch)
+        except:
+            return 0
 
-    async def save_new_delta(self,
-                             reflector_output: ReflectorOutput,
-                             trade_object: Dict[str, Any],
-                             domain: str = "strategy"):
-        """
-        حفظ "دلتا" جديدة بناءً على مخرجات المنعكس وسياسة القبول.
-        """
+    async def load_and_clear_pending_samples(self) -> List[TrainingSample]:
+        """جلب العينات للتدريب ومسحها من الانتظار وأرشفتها"""
         try:
-            trade_pnl_percent = trade_object.get('pnl_percent', 0)
+            raw_data = await self._load_json_from_r2(self.pending_samples_key)
+            if not raw_data: return []
             
-            # 1. التحقق من سياسة القبول
-            is_approved, approval_reason = self.policy_engine.get_delta_acceptance(
-                reflector_output,
-                trade_pnl_percent
-            )
+            samples = [TrainingSample(**d) for d in raw_data]
             
-            # 2. إنشاء كائن الدلتا
-            new_delta = Delta(
-                text=reflector_output.suggested_rule,
-                domain=domain,
-                score=reflector_output.confidence,
-                evidence_refs=[trade_object.get('id', 'unknown_trade_id')],
-                approved=is_approved,
-                trade_strategy=trade_object.get('strategy', 'unknown'),
-                exit_profile=trade_object.get('decision_data', {}).get('exit_profile', 'unknown')
-            )
-            
-            # 3. تحميل، إضافة، وحفظ الدلتا
-            deltas_list = await self._load_deltas_from_r2(domain)
-            deltas_list.append(new_delta.model_dump()) # Use model_dump() for pydantic models
-            await self._save_deltas_to_r2(domain, deltas_list)
-            
-            print(f"✅ [MemoryStore] تم حفظ دلتا جديدة لـ {domain}. الحالة: {approval_reason}")
-            
-            # 4. تفعيل عملية "التقطير" (لا يتم تنفيذها هنا مباشرة)
-            if len([d for d in deltas_list if d.get('approved')]) % self.distill_threshold == 0 and is_approved:
-                 print(f"ℹ️ [MemoryStore] تم الوصول إلى حد {self.distill_threshold} دلتا لـ {domain}. التقطير سيتم جدولته.")
-                 # (Curator will handle the actual check and distillation)
+            # 1. أرشفة البيانات (Append to History)
+            # لتقليل استهلاك الذاكرة، نرفع البيانات كنص JSONL مباشرة
+            lines = "\n".join([json.dumps(d) for d in raw_data])
+            # (هنا نحتاج دالة append في R2، إذا لم توجد، يمكن تجاوز الأرشفة الآن أو حفظ ملفات مؤرخة)
+            # للتسهيل: سنحفظ ملف أرشيف جديد لكل دفعة
+            archive_key = f"datasets/archive/batch_{samples[0].id}.json"
+            await self._save_json_to_r2(archive_key, raw_data)
 
+            # 2. مسح القائمة الحالية
+            await self._save_json_to_r2(self.pending_samples_key, [])
+            
+            return samples
         except Exception as e:
-            print(f"❌ [MemoryStore] فشل فادح في حفظ الدلتا: {e}")
-            traceback.print_exc()
-
+            print(f"❌ [MemoryStore] Failed to load/clear samples: {e}")
+            return []
 
-    async def get_active_context(self, domain: str, query: str, top_k: int = 3) -> str:
-        """
-        جلب "السياق النشط" (Active Context) لإرساله إلى النموذج.
-        """
+    # --- Helpers ---
+    async def _load_json_from_r2(self, key):
         try:
-            all_deltas_dicts = await self._load_deltas_from_r2(domain)
-            
-            # 1. تصفية الدلتا المعتمدة فقط
-            approved_deltas = [Delta(**d) for d in all_deltas_dicts if d.get('approved', False)]
-            
-            if not approved_deltas:
-                # (Return English text for consistency)
-                return "Playbook: No approved learning rules (Deltas) found for this domain yet."
+            resp = await self.r2.get_file_async(key) # Assuming R2Service has this
+            return json.loads(resp)
+        except: return None
 
-            # 2. خوارزمية الاسترجاع (نسخة مبسطة)
-            scored_deltas = []
-            for delta in approved_deltas:
-                priority_map = {"high": 1.0, "medium": 0.6, "low": 0.2}
-                priority_score = priority_map.get(delta.priority, 0.6)
-                
-                try:
-                    age_days = (datetime.now() - datetime.fromisoformat(delta.created_at)).days
-                    freshness_score = max(0, 1.0 - (age_days / 90.0))
-                except Exception:
-                    freshness_score = 0.5
-
-                relevance_score = 0.5
-                query_words = set(query.lower().split())
-                delta_words = set(delta.text.lower().split())
-                if query_words.intersection(delta_words):
-                    relevance_score = 1.0
-                elif delta.trade_strategy and delta.trade_strategy.lower() in query_words:
-                     relevance_score = 0.8
-
-                final_score = (0.6 * relevance_score) + (0.3 * priority_score) + (0.1 * freshness_score)
-                scored_deltas.append((final_score, delta))
-            
-            # 3. فرز واختيار أفضل K
-            scored_deltas.sort(key=lambda x: x[0], reverse=True)
-            top_deltas = [delta for score, delta in scored_deltas[:top_k]]
-            
-            # 4. تنسيق الموجه (باللغة الإنجليزية)
-            if not top_deltas:
-                 return "Playbook: No relevant learning rules (Deltas) found for this query."
-                
-            playbook_header = f"Playbook (Top {len(top_deltas)} Rules - Domain: {domain}):"
-            # (Added score for context)
-            delta_lines = [f"• {delta.text} (Score: {delta.score:.2f}, Prio: {delta.priority})" for delta in top_deltas]
-            
-            # (Distilled rules are just high-priority deltas in this implementation)
-            
-            return "\n".join([playbook_header] + delta_lines)
-
-        except Exception as e:
-            print(f"❌ [MemoryStore] فشل جلب السياق النشط: {e}")
-            # (Return English text for consistency)
-            return "Playbook: Error retrieving learning context."
+    async def _save_json_to_r2(self, key, data):
+        await self.r2.upload_json_async(data, key) # Assuming R2Service has this