Update data_manager.py

data_manager.py

@@ -54,8 +54,19 @@ class DataManager:
             print(f"❌ فشل تحميل بيانات الأسواق: {e}")
 
     async def close(self):
-        if self.http_client: 
+        # 🔴 --- START OF CHANGE --- 🔴
+        # (إصلاح تسريب الموارد)
+        if self.http_client and not self.http_client.is_closed: 
             await self.http_client.aclose()
+            print("   ✅ DataManager: http_client closed.")
+            
+        if self.exchange:
+            try:
+                await self.exchange.close()
+                print("   ✅ DataManager: ccxt.kucoin exchange closed.")
+            except Exception as e:
+                print(f"   ⚠️ DataManager: Error closing ccxt.kucoin: {e}")
+        # 🔴 --- END OF CHANGE --- 🔴
 
     async def get_market_context_async(self):
         """جلب سياق السوق الأساسي فقط"""
@@ -636,8 +647,7 @@ class DataManager:
         else:                            # أكثر من -10%
             return 0.3
 
-    # 🔴 --- بدء التعديل الجوهري --- 🔴
-    # تم تعديل هذه الدالة لتعمل كـ "منتج" (Producer) يضخ البيانات في طابور
+    # (لا تغيير في دالة التدفق)
     async def stream_ohlcv_data(self, symbols: List[str], queue: asyncio.Queue):
         """
         (معدلة) جلب بيانات OHLCV بشكل متدفق وإرسالها إلى طابور
@@ -692,15 +702,11 @@ class DataManager:
         
         print(f"✅ [المنتج] اكتمل تدفق بيانات OHLCV. تم إرسال {total_successful} عملة للمعالجة.")
         
-        # 🔴 --- START OF CHANGE --- 🔴
-        # (إرسال إشارة "None" لإنهاء المستهلك)
         try:
             await queue.put(None)
             print("      📬 [المنتج] تم إرسال إشارة الإنهاء (None) إلى الطابور.")
         except Exception as q_err:
             print(f"      ❌ [المنتج] فشل إرسال إشارة الإنهاء (None) للطابور: {q_err}")
-        # 🔴 --- END OF CHANGE --- 🔴
-    # 🔴 --- نهاية التعديل الجوهري --- 🔴
 
 
     async def _fetch_complete_ohlcv_parallel(self, symbol: str) -> Dict[str, Any]: