Update app.py

app.py

@@ -1,16 +1,15 @@
 import ccxt
 import pandas as pd
-import numpy as np
 import asyncio
 import logging
-from datetime import datetime
+import httpx
+from tenacity import retry, wait_exponential, stop_after_attempt
 from telegram import Bot
 from telegram.error import TelegramError
 from ta.momentum import RSIIndicator
 from ta.trend import EMAIndicator
 import config
 
-# تنظیمات لاگینگ
 logging.basicConfig(
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
     level=logging.INFO
@@ -19,40 +18,36 @@ logger = logging.getLogger(__name__)
 
 class LBankMonitor:
     def __init__(self):
+        self.http_client = httpx.AsyncClient(
+            limits=httpx.Limits(
+                max_connections=config.HTTP_CONFIG['max_connections'],
+                max_keepalive_connections=config.HTTP_CONFIG['max_keepalive']
+            ),
+            timeout=config.HTTP_CONFIG['timeout']
+        )
+        
         self.exchange = ccxt.lbank({
             'apiKey': config.API_CONFIG['apiKey'],
             'secret': config.API_CONFIG['secret'],
-            'enableRateLimit': True
+            'enableRateLimit': True,
+            'session': self.http_client
         })
+        
         self.bot = Bot(token=config.TELEGRAM_CONFIG['bot_token'])
+        self.semaphore = asyncio.Semaphore(config.THROTTLING['max_parallel'])
         self.connected = False
 
-    async def check_connection(self):
-        try:
-            await asyncio.to_thread(self.exchange.fetch_balance)
-            if not self.connected:
-                await self.send_telegram("✅ اتصال به صرافی LBank با موفقیت برقرار شد!")
-                self.connected = True
-            return True
-        except Exception as e:
-            if self.connected:
-                await self.send_telegram(f"⚠️ اتصال قطع شد! خطا: {str(e)}")
-                self.connected = False
-            return False
-
+    @retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(3))
     async def fetch_ohlcv(self, symbol, timeframe):
-        try:
-            ohlcv = await asyncio.to_thread(
-                self.exchange.fetch_ohlcv,
-                symbol,
-                timeframe
-            )
-            df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
-            return df
-        except Exception as e:
-            logger.error(f"Error fetching data for {symbol}: {str(e)}")
-            return None
+        async with self.semaphore:
+            try:
+                ohlcv = await self.exchange.fetch_ohlcv(symbol, timeframe)
+                df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+                df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
+                return df
+            except Exception as e:
+                logger.error(f"Error fetching {symbol} ({timeframe}): {str(e)}")
+                raise
 
     def calculate_indicators(self, df):
         df['rsi'] = RSIIndicator(df['close']).rsi()
@@ -61,38 +56,40 @@ class LBankMonitor:
         return df
 
     async def analyze_symbol(self, symbol):
-        signals = []
-        for tf in config.TIMEFRAMES:
-            df = await self.fetch_ohlcv(symbol, tf)
-            if df is not None and len(df) > 0:
-                df = self.calculate_indicators(df)
-                current_close = df['close'].iloc[-1]
-                rsi = df['rsi'].iloc[-1]
-                
-                # منطق پیشرفته تحلیل (نمونه)
-                if rsi < config.THRESHOLDS['rsi_oversold']:
-                    signals.append({
-                        'symbol': symbol,
-                        'timeframe': tf,
-                        'signal': 'خرید',
-                        'confidence': 95,
-                        'price': current_close
-                    })
-        return signals
+        try:
+            signals = []
+            for tf in config.TIMEFRAMES:
+                df = await self.fetch_ohlcv(symbol, tf)
+                if df is not None and not df.empty:
+                    df = self.calculate_indicators(df)
+                    current_close = df['close'].iloc[-1]
+                    rsi = df['rsi'].iloc[-1]
+                    
+                    # منطق پیشرفته تحلیل (نمونه)
+                    if rsi < config.THRESHOLDS['rsi_oversold']:
+                        signals.append({
+                            'symbol': symbol,
+                            'timeframe': tf,
+                            'signal': 'خرید',
+                            'confidence': 95,
+                            'price': current_close
+                        })
+            return signals
+        except Exception as e:
+            logger.error(f"Analysis failed for {symbol}: {str(e)}")
+            return []
 
     async def send_telegram(self, message):
-        emoji_map = {
-            'خرید': '🟢💰',
-            'فروش': '🔴📉',
-            'اتصال': '🌐',
-            'خطا': '❌'
-        }
-        for key, emoji in emoji_map.items():
-            if key in message:
-                message = f"{emoji} {message}"
-                break
-        
         try:
+            emojis = {
+                'خرید': '🟢💰', 'فروش': '🔴📉',
+                'اتصال': '🌐', 'خطا': '❌'
+            }
+            for k, v in emojis.items():
+                if k in message:
+                    message = f"{v} {message}"
+                    break
+            
             await self.bot.send_message(
                 chat_id=config.TELEGRAM_CONFIG['chat_id'],
                 text=message,
@@ -101,34 +98,67 @@ class LBankMonitor:
         except TelegramError as e:
             logger.error(f"Telegram error: {str(e)}")
 
+    async def check_connection(self):
+        try:
+            await self.exchange.fetch_balance()
+            if not self.connected:
+                await self.send_telegram("✅ اتصال به صرافی با موفقیت برقرار شد")
+                self.connected = True
+            return True
+        except Exception as e:
+            if self.connected:
+                await self.send_telegram(f"⚠️ اتصال قطع شد! خطا: {str(e)}")
+                self.connected = False
+            return False
+
     async def run_analysis(self):
         if await self.check_connection():
-            tasks = []
-            for symbol in config.SYMBOLS:
-                tasks.append(self.analyze_symbol(symbol))
+            symbols = config.SYMBOLS
+            batch_size = config.THROTTLING['batch_size']
             
-            results = await asyncio.gather(*tasks)
-            for signals in results:
-                for signal in signals:
-                    if signal['confidence'] >= config.THRESHOLDS['min_confidence']:
-                        message = (
-                            f"🚨 <b>سیگنال معاملاتی شناسایی شد!</b> 🚨\n\n"
-                            f"🔹 ارز: {signal['symbol']}\n"
-                            f"🔹 نوع سیگنال: {signal['signal']}\n"
-                            f"🔹 اعتبار: {signal['confidence']}%\n"
-                            f"🔹 قیمت فعلی: {signal['price']:.4f}\n"
-                            f"🔹 حد سود: {signal['price'] * 1.03:.4f}\n"
-                            f"🔹 حد ضرر: {signal['price'] * 0.97:.4f}\n"
-                            f"⏳ اعتبار سیگنال: 6 ساعت"
-                        )
-                        await self.send_telegram(message)
+            for i in range(0, len(symbols), batch_size):
+                batch = symbols[i:i+batch_size]
+                tasks = [self.analyze_symbol(symbol) for symbol in batch]
+                results = await asyncio.gather(*tasks)
+                
+                for signals in results:
+                    for signal in signals:
+                        if signal['confidence'] >= config.THRESHOLDS['min_confidence']:
+                            await self.generate_signal_message(signal)
+                
+                await asyncio.sleep(config.THROTTLING['delay_between_batches'])
+
+    async def generate_signal_message(self, signal):
+        message = (
+            f"🚨 <b>سیگنال معاملاتی!</b> 🚨\n\n"
+            f"🏷 ارز: {signal['symbol']}\n"
+            f"📊 تایم فریم: {signal['timeframe']}\n"
+            f"🔢 نوع سیگنال: {signal['signal']}\n"
+            f"📈 قیمت فعلی: {signal['price']:.4f}\n"
+            f"✅ اعتبار: {signal['confidence']}%\n"
+            f"🎯 حد سود: {signal['price'] * 1.03:.4f}\n"
+            f"🛑 حد ضرر: {signal['price'] * 0.97:.4f}\n"
+            f"⏳ مدت اعتبار: ۶ ساعت"
+        )
+        await self.send_telegram(message)
 
     async def main(self):
-        await self.send_telegram("🤖 ربات مانیتورینگ LBank راه‌اندازی شد!")
+        await self.send_telegram("🤖 ربات مانیتورینگ LBank فعال شد!")
         while True:
-            await self.run_analysis()
-            await asyncio.sleep(1800)  # هر 30 دقیقه
+            try:
+                await self.run_analysis()
+                await asyncio.sleep(1800)
+            except Exception as e:
+                logger.error(f"Critical error: {str(e)}")
+                await self.send_telegram(f"🔥 خطای سیستمی: {str(e)}")
+                await asyncio.sleep(60)
+
+    async def close(self):
+        await self.http_client.aclose()
 
 if __name__ == "__main__":
     monitor = LBankMonitor()
-    asyncio.run(monitor.main())
+    try:
+        asyncio.run(monitor.main())
+    except KeyboardInterrupt:
+        asyncio.run(monitor.close())