Create memory_store.py

learning_hub/memory_store.py

@@ -0,0 +1,161 @@
+# learning_hub/memory_store.py
+import json
+import asyncio
+from datetime import datetime
+from typing import List, Dict, Optional
+from .schemas import Delta, ReflectorOutput
+from .policy_engine import PolicyEngine
+
+# للتوافق مع R2Service (نفترض أنه سيتم تمريره)
+# (لا يمكننا استيراده مباشرة لتجنب التبعيات الدائرية)
+
+class MemoryStore:
+    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")
+
+    async def _load_deltas_from_r2(self, domain: str) -> List[Dict]:
+        """تحميل ملف الدلتا المحدد من R2"""
+        key = self.domain_files.get(domain, self.domain_files["general"])
+        try:
+            response = self.r2_service.s3_client.get_object(Bucket="trading", Key=key)
+            deltas_list = json.loads(response['Body'].read())
+            return deltas_list
+        except Exception as e:
+            # إذا فشل التحميل (مثل الملف غير موجود)، أعد قائمة فارغة
+            print(f"ℹ️ [MemoryStore] لم يتم العثور على ملف دلتا لـ {domain}، سيتم إنشاء واحد جديد.")
+            return []
+
+    async def _save_deltas_to_r2(self, domain: str, deltas_list: List[Dict]):
+        """حفظ ملف الدلتا المحدث إلى R2"""
+        key = self.domain_files.get(domain, self.domain_files["general"])
+        try:
+            data_json = json.dumps(deltas_list, 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}")
+
+    async def save_new_delta(self, 
+                             reflector_output: ReflectorOutput, 
+                             trade_object: Dict[str, Any], 
+                             domain: str = "strategy"):
+        """
+        حفظ "دلتا" جديدة بناءً على مخرجات المنعكس وسياسة القبول.
+        (تنفيذ النقطة 5)
+        """
+        try:
+            trade_pnl_percent = trade_object.get('pnl_percent', 0)
+            
+            # 1. التحقق من سياسة القبول (النقطة 5 و 7)
+            is_approved, approval_reason = self.policy_engine.get_delta_acceptance(
+                reflector_output, 
+                trade_pnl_percent
+            )
+            
+            # 2. إنشاء كائن الدلتا (النقطة 3)
+            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())
+            await self._save_deltas_to_r2(domain, deltas_list)
+            
+            print(f"✅ [MemoryStore] تم حفظ دلتا جديدة لـ {domain}. الحالة: {approval_reason}")
+            
+            # 4. تفعيل عملية "التقطير" (Distillation) إذا تم تجاوز الحد (النقطة 6)
+            if len(deltas_list) % self.distill_threshold == 0 and is_approved:
+                print(f"ℹ️ [MemoryStore] تم الوصول إلى حد {self.distill_threshold} دلتا. بدء عملية التقطير...")
+                # (سنقوم باستدعاء عملية التقطير في الخلفية)
+                # asyncio.create_task(self.distill_domain(domain))
+                # (ملاحظة: التقطير سيتم تنفيذه في ملف curator.py لاحقاً)
+                pass
+
+        except Exception as e:
+            print(f"❌ [MemoryStore] فشل فادح في حفظ الدلتا: {e}")
+
+    async def get_active_context(self, domain: str, query: str, top_k: int = 3) -> str:
+        """
+        جلب "السياق النشط" (Active Context) لإرساله إلى النموذج.
+        (تنفيذ النقطة 2 و 5 - خوارزمية الاسترجاع)
+        """
+        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 "No approved deltas (rules) found for this domain."
+            
+            # 2. خوارزمية الاسترجاع (تنفيذ النقطة 5)
+            # (نسخة مبسطة تعتمد على الثقة والأولوية والكلمات المفتاحية)
+            # (لا يمكننا استخدام semantic_sim بدون نموذج تضمين، لذا سنستخدم مطابقة الكلمات)
+            
+            scored_deltas = []
+            for delta in approved_deltas:
+                # أ. حساب الأولوية (priority_score)
+                priority_map = {"high": 1.0, "medium": 0.6, "low": 0.2}
+                priority_score = priority_map.get(delta.priority, 0.6)
+                
+                # ب. حساب الحداثة (freshness_score)
+                try:
+                    age_days = (datetime.now() - datetime.fromisoformat(delta.created_at)).days
+                    freshness_score = max(0, 1.0 - (age_days / 90.0)) # (تتلاشى بعد 90 يوم)
+                except Exception:
+                    freshness_score = 0.5
+
+                # ج. حساب "الملاءمة" (relevance_score) - بديل مبسط لـ semantic_sim
+                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 # (مطابقة للاستراتيجية)
+
+                # د. النتيجة الإجمالية (مستوحاة من النقطة 5)
+                # score_total = 0.7*semantic_sim + 0.2*priority + 0.1*freshness
+                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. تنسيق الموجه (النقطة 3 - ActiveContext)
+            if not top_deltas:
+                return "No relevant deltas (rules) found for this query."
+                
+            playbook_header = f"Playbook (Top {len(top_deltas)} Rules - Domain: {domain}):"
+            delta_lines = [f"• {delta.text} (Ref: {delta.id})" for delta in top_deltas]
+            
+            # (ملاحظة: Distilled rule ستتم إضافتها لاحقاً بواسطة Curator)
+            
+            return "\n".join([playbook_header] + delta_lines)
+
+        except Exception as e:
+            print(f"❌ [MemoryStore] فشل جلب السياق النشط: {e}")
+            return "Error retrieving learning context."