#!/usr/bin/env python # -*- coding: utf-8 -*- import ccxt import pandas as pd import numpy as np import time import asyncio import logging import json import os import math import talib import telegram from telegram.ext import Updater, CommandHandler, MessageHandler, Filters from datetime import datetime, timedelta import threading import schedule import warnings import requests from collections import defaultdict import matplotlib.pyplot as plt import matplotlib import io from config import Config # تنظیم matplotlib برای پشتیبانی از زبان فارسی matplotlib.rc('font', family='DejaVu Sans') # تنظیمات لاگینگ logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("crypto_analyzer.log", encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # اخطارهای غیرضروری را نادیده بگیر warnings.filterwarnings("ignore") class CryptoAnalyzer: def __init__(self, config_path="config.json"): """مقداردهی اولیه کلاس تحلیلگر ارزهای دیجیتال""" try: # بارگیری تنظیمات self.config = Config() # تنظیمات صرافی self.exchange = ccxt.lbank({ 'apiKey': self.config.API_KEY, 'secret': self.config.SECRET_KEY, 'enableRateLimit': True, 'options': {'defaultType': 'spot'} }) # دیکشنری برای ذخیره داده‌های بازار self.market_data = {} # لیست ارزهای تحت نظارت self.symbols = self.config.SYMBOLS # تایم فریم‌ها self.timeframes = self.config.TIMEFRAMES # تنظیمات تلگرام self.bot = telegram.Bot(token=self.config.TELEGRAM_TOKEN) self.chat_id = self.config.TELEGRAM_CHAT_ID # وضعیت اتصال self.is_connected = False # دیکشنری برای ذخیره سیگنال‌های فعال self.active_signals = {} # بررسی اتصال به صرافی self.check_exchange_connection() logger.info("🔄 سیستم تحلیلگر ارزهای دیجیتال با موفقیت راه‌اندازی شد!") self.send_telegram_message("🔄 سیستم تحلیلگر ارزهای دیجیتال با موفقیت راه‌اندازی شد!") except Exception as e: logger.error(f"خطا در راه‌اندازی سیستم: {str(e)}") self.send_telegram_message(f"❌ خطا در راه‌اندازی سیستم: {str(e)}") def check_exchange_connection(self): """بررسی اتصال به صرافی""" try: self.exchange.load_markets() self.is_connected = True logger.info("✅ اتصال به صرافی با موفقیت برقرار شد!") self.send_telegram_message("✅ اتصال به صرافی با موفقیت برقرار شد!") return True except Exception as e: self.is_connected = False logger.error(f"❌ خطا در اتصال به صرافی: {str(e)}") self.send_telegram_message(f"❌ خطا در اتصال به صرافی: {str(e)}") return False async def fetch_ohlcv_data(self, symbol, timeframe): """دریافت داده‌های OHLCV برای یک ارز در یک تایم فریم خاص""" try: limit = 100 # تعداد کندل‌ها ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit) df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') df.set_index('timestamp', inplace=True) if symbol not in self.market_data: self.market_data[symbol] = {} self.market_data[symbol][timeframe] = df logger.debug(f"داده‌های {symbol} در تایم فریم {timeframe} با موفقیت دریافت شد.") return df except Exception as e: logger.error(f"خطا در دریافت داده‌های {symbol} در تایم فریم {timeframe}: {str(e)}") return None async def fetch_order_book(self, symbol): """دریافت اردربوک برای یک ارز""" try: order_book = self.exchange.fetch_order_book(symbol) return order_book except Exception as e: logger.error(f"خطا در دریافت اردربوک {symbol}: {str(e)}") return None def calculate_indicators(self, df): """محاسبه شاخص‌های تکنیکال مختلف""" try: # محاسبه RSI df['rsi'] = talib.RSI(df['close'], timeperiod=14) # محاسبه MACD macd, macd_signal, macd_hist = talib.MACD(df['close']) df['macd'] = macd df['macd_signal'] = macd_signal df['macd_hist'] = macd_hist # محاسبه میانگین متحرک df['sma_20'] = talib.SMA(df['close'], timeperiod=20) df['sma_50'] = talib.SMA(df['close'], timeperiod=50) df['sma_200'] = talib.SMA(df['close'], timeperiod=200) # محاسبه Bollinger Bands df['upper_band'], df['middle_band'], df['lower_band'] = talib.BBANDS(df['close'], timeperiod=20) # محاسبه MFI (Money Flow Index) df['mfi'] = talib.MFI(df['high'], df['low'], df['close'], df['volume'], timeperiod=14) # محاسبه ATR (Average True Range) df['atr'] = talib.ATR(df['high'], df['low'], df['close'], timeperiod=14) return df except Exception as e: logger.error(f"خطا در محاسبه شاخص‌ها: {str(e)}") return df def calculate_volume_profile(self, df): """تحلیل پروفایل حجم معاملات""" try: # تقسیم محدوده قیمت به ۱۰ دسته price_range = df['high'].max() - df['low'].min() bin_size = price_range / 10 price_bins = [] volume_bins = [] for i in range(10): lower = df['low'].min() + i * bin_size upper = lower + bin_size mask = (df['close'] >= lower) & (df['close'] < upper) volume_in_bin = df.loc[mask, 'volume'].sum() price_bins.append((lower + upper) / 2) volume_bins.append(volume_in_bin) return { 'price_bins': price_bins, 'volume_bins': volume_bins } except Exception as e: logger.error(f"خطا در محاسبه پروفایل حجم: {str(e)}") return None def calculate_delta(self, symbol, timeframe): """محاسبه دلتا (تفاوت بین حجم خرید و فروش)""" try: df = self.market_data[symbol][timeframe] # محاسبه دلتا ساده با استفاده از تغییرات قیمت و حجم df['delta'] = np.where(df['close'] > df['open'], df['volume'], -df['volume']) df['cumulative_delta'] = df['delta'].cumsum() # محاسبه دلتا ریورس df['reverse_delta'] = -df['delta'] df['cumulative_reverse_delta'] = df['reverse_delta'].cumsum() return df['cumulative_delta'].iloc[-1], df['cumulative_reverse_delta'].iloc[-1] except Exception as e: logger.error(f"خطا در محاسبه دلتا برای {symbol} در تایم فریم {timeframe}: {str(e)}") return 0, 0 def analyze_bid_ask(self, order_book): """تحلیل بید اسک (خرید و فروش) از اطلاعات اردربوک""" try: # محاسبات بید اسک bids = order_book['bids'] asks = order_book['asks'] # مجموع حجم در سمت خرید و فروش total_bid_volume = sum([bid[1] for bid in bids[:10]]) total_ask_volume = sum([ask[1] for ask in asks[:10]]) # نسبت حجم خرید به فروش bid_ask_ratio = total_bid_volume / total_ask_volume if total_ask_volume > 0 else float('inf') # فشار خرید یا فروش if bid_ask_ratio > 1.5: pressure = "خرید" pressure_strength = min(bid_ask_ratio / 5, 1.0) # نرمال‌سازی قدرت elif bid_ask_ratio < 0.67: pressure = "فروش" pressure_strength = min((1 / bid_ask_ratio) / 5, 1.0) # نرمال سازی قدرت else: pressure = "خنثی" pressure_strength = 0.0 return { 'bid_ask_ratio': bid_ask_ratio, 'pressure': pressure, 'pressure_strength': pressure_strength } except Exception as e: logger.error(f"خطا در تحلیل بید اسک: {str(e)}") return { 'bid_ask_ratio': 1.0, 'pressure': "خنثی", 'pressure_strength': 0.0 } def analyze_order_flow(self, symbol, timeframes): """تحلیل جریان سفارشات (Order Flow) برای یک ارز در تایم فریم‌های مختلف""" order_flow_score = 0 try: order_book = asyncio.run(self.fetch_order_book(symbol)) if not order_book: return 0 bid_ask_analysis = self.analyze_bid_ask(order_book) # جمع آوری دلتا از تمام تایم فریم‌ها total_delta = 0 for tf in timeframes: if symbol in self.market_data and tf in self.market_data[symbol]: delta, _ = self.calculate_delta(symbol, tf) # وزن دهی به دلتا بر اساس تایم فریم weight = {'1m': 0.1, '5m': 0.15, '15m': 0.2, '30m': 0.25, '1h': 0.3}.get(tf, 0.2) total_delta += delta * weight # ترکیب دلتا با تحلیل بید اسک if bid_ask_analysis['pressure'] == "خرید" and total_delta > 0: order_flow_score = 0.5 + bid_ask_analysis['pressure_strength'] * 0.5 elif bid_ask_analysis['pressure'] == "فروش" and total_delta < 0: order_flow_score = -0.5 - bid_ask_analysis['pressure_strength'] * 0.5 else: order_flow_score = total_delta / (abs(total_delta) + 1) * 0.3 # نرمال سازی بین -0.3 و 0.3 return order_flow_score except Exception as e: logger.error(f"خطا در تحلیل جریان سفارشات برای {symbol}: {str(e)}") return order_flow_score def analyze_elliott_waves(self, df): """تحلیل امواج الیوت""" try: # این یک تقریب ساده از تحلیل امواج الیوت است # برای تحلیل دقیق، الگوریتم‌های پیچیده‌تری نیاز است # پیدا کردن نقاط اوج و فرود اخیر prices = df['close'].values window_size = 5 peaks = [] troughs = [] for i in range(window_size, len(prices) - window_size): if all(prices[i] > prices[i-j] for j in range(1, window_size+1)) and all(prices[i] > prices[i+j] for j in range(1, window_size+1)): peaks.append((i, prices[i])) if all(prices[i] < prices[i-j] for j in range(1, window_size+1)) and all(prices[i] < prices[i+j] for j in range(1, window_size+1)): troughs.append((i, prices[i])) # حداقل ۵ نقطه برای تحلیل موج الیوت نیاز است if len(peaks) < 3 or len(troughs) < 2: return { 'wave_pattern': 'نامشخص', 'confidence': 0.0, 'potential_direction': 'خنثی' } # مرتب‌سازی براساس زمان all_points = sorted(peaks + troughs, key=lambda x: x[0]) # تلاش برای تشخیص الگوهای امواج الیوت ۵-۳ if len(all_points) >= 5: # بررسی اینکه آیا ۵ موج رو به بالا داریم if all_points[0][1] < all_points[1][1] > all_points[2][1] < all_points[3][1] > all_points[4][1]: # احتمالا در موج ۵ صعودی هستیم return { 'wave_pattern': 'موج 5 صعودی', 'confidence': 0.7, 'potential_direction': 'نزولی' # پس از موج ۵ صعودی، احتمال نزول وجود دارد } elif all_points[0][1] > all_points[1][1] < all_points[2][1] > all_points[3][1] < all_points[4][1]: # احتمالا در موج ۵ نزولی هستیم return { 'wave_pattern': 'موج 5 نزولی', 'confidence': 0.7, 'potential_direction': 'صعودی' # پس از موج ۵ نزولی، احتمال صعود وجود دارد } # اگر الگوی مشخصی شناسایی نشد return { 'wave_pattern': 'در حال تشکیل', 'confidence': 0.3, 'potential_direction': 'نامشخص' } except Exception as e: logger.error(f"خطا در تحلیل امواج الیوت: {str(e)}") return { 'wave_pattern': 'خطا', 'confidence': 0.0, 'potential_direction': 'خنثی' } def analyze_footprint(self, symbol, timeframe): """تحلیل فوت‌پرینت (نقشه حرارتی معاملات)""" try: df = self.market_data[symbol][timeframe] # تجزیه و تحلیل بر اساس تغییرات قیمت و حجم df['price_change'] = df['close'] - df['open'] df['normalized_volume'] = df['volume'] / df['volume'].mean() # شناسایی کندل‌های مهم با حجم بالا high_volume_candles = df[df['normalized_volume'] > 1.5] # محاسبه نسبت کندل‌های صعودی پرحجم به کل کندل‌های پرحجم if len(high_volume_candles) > 0: bullish_ratio = len(high_volume_candles[high_volume_candles['price_change'] > 0]) / len(high_volume_candles) else: bullish_ratio = 0.5 # نمره دهی به فوت‌پرینت if bullish_ratio > 0.7: footprint_score = (bullish_ratio - 0.5) * 2 # نرمال سازی بین 0 و 1 direction = "صعودی" elif bullish_ratio < 0.3: footprint_score = ((0.5 - bullish_ratio) * 2) * -1 # نرمال سازی بین -1 و 0 direction = "نزولی" else: footprint_score = (bullish_ratio - 0.5) * 2 # بین -0.4 و 0.4 direction = "خنثی" return { 'footprint_score': footprint_score, 'direction': direction, 'high_volume_candles_count': len(high_volume_candles) } except Exception as e: logger.error(f"خطا در تحلیل فوت‌پرینت برای {symbol} در تایم فریم {timeframe}: {str(e)}") return { 'footprint_score': 0, 'direction': "خنثی", 'high_volume_candles_count': 0 } def analyze_poc(self, symbol, timeframe): """تحلیل نقطه کنترل قیمت (Point of Control)""" try: df = self.market_data[symbol][timeframe] # محاسبه پروفایل حجم volume_profile = self.calculate_volume_profile(df) if not volume_profile: return { 'poc_price': df['close'].iloc[-1], 'distance_from_poc': 0, 'poc_significance': 0 } # پیدا کردن قیمت با بیشترین حجم (POC) max_volume_index = np.argmax(volume_profile['volume_bins']) poc_price = volume_profile['price_bins'][max_volume_index] # محاسبه فاصله از POC current_price = df['close'].iloc[-1] distance_from_poc = (current_price - poc_price) / poc_price * 100 # محاسبه اهمیت POC (بر اساس نسبت حجم در POC به میانگین حجم) avg_volume = np.mean(volume_profile['volume_bins']) max_volume = volume_profile['volume_bins'][max_volume_index] poc_significance = max_volume / avg_volume if avg_volume > 0 else 1 return { 'poc_price': poc_price, 'distance_from_poc': distance_from_poc, 'poc_significance': min(poc_significance / 3, 1.0) # نرمال‌سازی بین 0 و 1 } except Exception as e: logger.error(f"خطا در تحلیل POC برای {symbol} در تایم فریم {timeframe}: {str(e)}") return { 'poc_price': 0, 'distance_from_poc': 0, 'poc_significance': 0 } def generate_trading_signal(self, symbol): """تولید سیگنال معاملاتی برای یک ارز بر اساس تحلیل‌های انجام شده""" try: # جمع‌آوری نتایج تحلیل‌ها از تمام تایم فریم‌ها timeframe_analysis = {} signal_strength = 0 signal_direction = "خنثی" for tf in self.timeframes: # بررسی وجود داده برای این تایم فریم if symbol not in self.market_data or tf not in self.market_data[symbol]: continue df = self.market_data[symbol][tf] # وزن تایم فریم در تحلیل نهایی tf_weight = {'1m': 0.05, '5m': 0.1, '15m': 0.2, '30m': 0.25, '1h': 0.4}.get(tf, 0.2) # تحلیل‌های مختلف order_flow_score = self.analyze_order_flow(symbol, [tf]) footprint_analysis = self.analyze_footprint(symbol, tf) poc_analysis = self.analyze_poc(symbol, tf) elliott_analysis = self.analyze_elliott_waves(df) # تحلیل شاخص‌های تکنیکال rsi = df['rsi'].iloc[-1] if 'rsi' in df.columns else 50 macd_hist = df['macd_hist'].iloc[-1] if 'macd_hist' in df.columns else 0 # محاسبه امتیاز نهایی تایم فریم tf_score = 0 # امتیاز بر اساس RSI if rsi > 70: tf_score -= 0.3 # اشباع خرید elif rsi < 30: tf_score += 0.3 # اشباع فروش elif rsi > 60: tf_score += 0.1 # روند صعودی elif rsi < 40: tf_score -= 0.1 # روند نزولی # امتیاز بر اساس MACD if macd_hist > 0 and macd_hist > df['macd_hist'].iloc[-2]: tf_score += 0.2 # سیگنال صعودی قوی elif macd_hist > 0: tf_score += 0.1 # سیگنال صعودی elif macd_hist < 0 and macd_hist < df['macd_hist'].iloc[-2]: tf_score -= 0.2 # سیگنال نزولی قوی elif macd_hist < 0: tf_score -= 0.1 # سیگنال نزولی # امتیاز بر اساس میانگین متحرک if 'sma_20' in df.columns and 'sma_50' in df.columns: if df['close'].iloc[-1] > df['sma_20'].iloc[-1] > df['sma_50'].iloc[-1]: tf_score += 0.2 # روند صعودی قوی elif df['close'].iloc[-1] < df['sma_20'].iloc[-1] < df['sma_50'].iloc[-1]: tf_score -= 0.2 # روند نزولی قوی # امتیاز بر اساس Order Flow tf_score += order_flow_score # امتیاز بر اساس Footprint tf_score += footprint_analysis['footprint_score'] # امتیاز بر اساس POC if poc_analysis['distance_from_poc'] < 0 and poc_analysis['poc_significance'] > 0.5: tf_score += 0.2 # قیمت زیر POC با اهمیت بالا elif poc_analysis['distance_from_poc'] > 0 and poc_analysis['poc_significance'] > 0.5: tf_score -= 0.2 # قیمت بالای POC با اهمیت بالا # امتیاز بر اساس امواج الیوت if elliott_analysis['potential_direction'] == "صعودی": tf_score += 0.3 * elliott_analysis['confidence'] elif elliott_analysis['potential_direction'] == "نزولی": tf_score -= 0.3 * elliott_analysis['confidence'] # ذخیره نتیجه تحلیل تایم فریم timeframe_analysis[tf] = { 'score': tf_score, 'order_flow': order_flow_score, 'footprint': footprint_analysis, 'poc': poc_analysis, 'elliott': elliott_analysis } # افزودن امتیاز تایم فریم به امتیاز کلی با اعمال وزن signal_strength += tf_score * tf_weight # تعیین جهت سیگنال if signal_strength > 0.3: signal_direction = "خرید" elif signal_strength < -0.3: signal_direction = "فروش" # تعیین قدرت سیگنال (بین 90% تا 100%) normalized_strength = min(abs(signal_strength) * 10 / 8, 1.0) # نرمال‌سازی signal_strength_percent = 90 + (normalized_strength * 10) # محاسبه حد سود و حد ضرر current_price = self.market_data[symbol]['1h']['close'].iloc[-1] atr = self.market_data[symbol]['1h']['atr'].iloc[-1] if 'atr' in self.market_data[symbol]['1h'].columns else current_price * 0.01 if signal_direction == "خرید": stop_loss = current_price - (atr * 2) take_profit = current_price + (atr * 4) elif signal_direction == "فروش": stop_loss = current_price + (atr * 2) take_profit = current_price - (atr * 4) else: stop_loss = 0 take_profit = 0 # تعیین مدت زمان ماندن در معامله (حداکثر ۶ ساعت) if signal_strength_percent > 97: duration_hours = 6 elif signal_strength_percent > 95: duration_hours = 4 else: duration_hours = 2 return { 'symbol': symbol, 'direction': signal_direction, 'strength': signal_strength_percent, 'entry_price': current_price, 'stop_loss': stop_loss, 'take_profit': take_profit, 'duration_hours': duration_hours, 'analysis': timeframe_analysis, 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } except Exception as e: logger.error(f"خطا در تولید سیگنال معاملاتی برای {symbol}: {str(e)}") return None def should_send_signal(self, signal): """تصمیم‌گیری برای ارسال سیگنال بر اساس قدرت آن و وضعیت سیگنال‌های قبلی""" try: # اگر سیگنال خنثی است، ارسال نکن if signal['direction'] == "خنثی": return False # بررسی قدرت سیگنال if signal['strength'] < 90: return False # بررسی عدم تکرار سیگنال‌ها در زمان کوتاه symbol = signal['symbol'] if symbol in self.active_signals: last_signal = self.active_signals[symbol] time_diff = datetime.now() - datetime.strptime(last_signal['timestamp'], '%Y-%m-%d %H:%M:%S') # اگر کمتر از ۲ ساعت از سیگنال قبلی گذشته باشد و جهت یکسان باشد if time_diff.total_seconds() < 7200 and last_signal['direction'] == signal['direction']: return False # اگر قدرت سیگنال جدید کمتر از سیگنال قبلی باشد if signal['strength'] <= last_signal['strength'] and time_diff.total_seconds() < 14400: return False return True except Exception as e: logger.error(f"خطا در تصمیم‌گیری برای ارسال سیگنال: {str(e)}") return False def create_signal_chart(self, symbol): """ایجاد نمودار برای سیگنال معاملاتی""" try: if symbol not in self.market_data or '1h' not in self.market_data[symbol]: return None df = self.market_data[symbol]['1h'].copy() # رسم نمودار plt.figure(figsize=(10, 6)) # نمودار قیمت بسته شدن plt.plot(df.index[-30:], df['close'].values[-30:], label='قیمت', color='blue') # اضافه کردن میانگین متحرک if 'sma_20' in df.columns and 'sma_50' in df.columns: plt.plot(df.index[-30:], df['sma_20'].values[-30:], label='SMA 20', color='orange', alpha=0.7) plt.plot(df.index[-30:], df['sma_50'].values[-30:], label='SMA 50', color='red', alpha=0.7) # تنظیمات نمودار plt.title(f"تحلیل ارز {symbol}", fontsize=14) plt.grid(True, alpha=0.3) plt.legend() # ذخیره نمودار در حافظه buf = io.BytesIO() plt.savefig(buf, format='png', dpi=100) plt.close() buf.seek(0) return buf except Exception as e: logger.error(f"خطا در ایجاد نمودار برای {symbol}: {str(e)}") return None def send_signal_to_telegram(self, signal): """ارسال سیگنال معاملاتی به تلگرام""" try: if not signal: return False direction_emoji = "🟢 خرید" if signal['direction'] == "خرید" else "🔴 فروش" # ساخت متن پیام message = f""" ⚡️ *سیگنال معاملاتی جدید* ⚡️ {direction_emoji} *{signal['symbol']}* 💰 قیمت ورود: `{signal['entry_price']}` 🛑 حد ضرر: `{signal['stop_loss']}` ✅ حد سود: `{signal['take_profit']}` ⏱ مدت زمان: *{signal['duration_hours']} ساعت* 💪 قدرت سیگنال: *{signal['strength']:.1f}%* ⏰ تاریخ سیگنال: {signal['timestamp']} 🤖 سیگنال‌دهی خودکار هوشمند """ # ارسال پیام متنی self.bot.send_message( chat_id=self.chat_id, text=message, parse_mode=telegram.ParseMode.MARKDOWN ) # ارسال نمودار chart = self.create_signal_chart(signal['symbol']) if chart: self.bot.send_photo( chat_id=self.chat_id, photo=chart, caption=f"نمودار تحلیلی {signal['symbol']}" ) # ذخیره سیگنال در لیست سیگنال‌های فعال self.active_signals[signal['symbol']] = signal logger.info(f"سیگنال معاملاتی برای {signal['symbol']} با موفقیت ارسال شد.") return True except Exception as e: logger.error(f"خطا در ارسال سیگنال به تلگرام: {str(e)}") return False def send_telegram_message(self, message): """ارسال پیام ساده به تلگرام""" try: self.bot.send_message( chat_id=self.chat_id, text=message ) return True except Exception as e: logger.error(f"خطا در ارسال پیام به تلگرام: {str(e)}") return False async def periodic_analysis(self): """انجام تحلیل دوره‌ای بر روی تمام ارزها""" while True: try: logger.info("شروع تحلیل دوره‌ای بازار...") tasks = [] for symbol in self.symbols: for timeframe in self.timeframes: tasks.append(self.fetch_ohlcv_data(symbol, timeframe)) # اجرای همزمان تمام درخواست‌ها await asyncio.gather(*tasks) # تحلیل و تولید سیگنال برای هر ارز for symbol in self.symbols: signal = self.generate_trading_signal(symbol) if signal and self.should_send_signal(signal): self.send_signal_to_telegram(signal) logger.info("تحلیل دوره‌ای با موفقیت انجام شد. خواب برای 5 دقیقه...") await asyncio.sleep(300) # خواب برای 5 دقیقه except Exception as e: logger.error(f"خطا در تحلیل دوره‌ای: {str(e)}") await asyncio.sleep(60) # در صورت خطا 1 دقیقه صبر کنید def start_telegram_handlers(self): """شروع کردن هندلرهای تلگرام برای دریافت دستورات""" updater = Updater(self.config.TELEGRAM_TOKEN, use_context=True) dp = updater.dispatcher # دستور وضعیت def status(update, context): status_msg = "وضعیت سیستم:\n" status_msg += f"• اتصال به صرافی: {'✅' if self.is_connected else '❌'}\n" status_msg += f"• تعداد ارزهای تحت نظارت: {len(self.symbols)}\n" status_msg += f"• سیگنال‌های فعال: {len(self.active_signals)}" update.message.reply_text(status_msg) # دستور راهنما def help(update, context): help_text = """ دستورات موجود: /status - نمایش وضعیت سیستم /analyze [symbol] - تحلیل فوری یک ارز /list_signals - نمایش سیگنال‌های فعال """ update.message.reply_text(help_text) # ثبت هندلرها dp.add_handler(CommandHandler("status", status)) dp.add_handler(CommandHandler("help", help)) # شروع پولینگ updater.start_polling() logger.info("هندلرهای تلگرام با موفقیت راه‌اندازی شدند.") def run(self): """شروع اجرای اصلی برنامه""" try: # شروع هندلرهای تلگرام در یک ریسمان جداگانه telegram_thread = threading.Thread(target=self.start_telegram_handlers) telegram_thread.daemon = True telegram_thread.start() # شروع تحلیل دوره‌ای در لوپ asyncio loop = asyncio.get_event_loop() loop.run_until_complete(self.periodic_analysis()) except KeyboardInterrupt: logger.info("دریافت سیگنال توقف. خاموش کردن سیستم...") self.send_telegram_message("🛑 سیستم در حال خاموش شدن...") except Exception as e: logger.error(f"خطای غیرمنتظره: {str(e)}") self.send_telegram_message(f"☠️ خطای بحرانی: {str(e)}") if __name__ == "__main__": analyzer = CryptoAnalyzer() analyzer.run()