""" Technical Analysis Service تحلیل سطوح حمایت و مقاومت پویا + ADX + RSI """ import numpy as np import pandas as pd class TechnicalAnalysis: """سرویس تحلیل تکنیکال برای محاسبه سطوح حمایت/مقاومت و ADX""" def __init__(self): self.timeframe_config = { '15m': {'lookback': 50, 'adx_period': 14, 'cluster_pct': 0.02}, '30m': {'lookback': 60, 'adx_period': 14, 'cluster_pct': 0.018}, '1h': {'lookback': 100, 'adx_period': 14, 'cluster_pct': 0.015}, '4h': {'lookback': 80, 'adx_period': 14, 'cluster_pct': 0.012}, '1d': {'lookback': 50, 'adx_period': 14, 'cluster_pct': 0.01}, '1wk': {'lookback': 30, 'adx_period': 14, 'cluster_pct': 0.008}, '1mo': {'lookback': 20, 'adx_period': 14, 'cluster_pct': 0.007} } def calculate_support_resistance(self, prices, timeframe='1h'): """محاسبه سطوح حمایت و مقاومت پویا""" config = self.timeframe_config.get(timeframe, self.timeframe_config['1h']) lookback = min(config['lookback'], len(prices)) if len(prices) < lookback: lookback = len(prices) recent_prices = prices[-lookback:] current_price = prices[-1] # یافتن قله‌ها (مقاومت) و دره‌ها (حمایت) resistance_levels = [] support_levels = [] for i in range(2, len(recent_prices) - 2): # تشخیص قله (محلی حداکثر) if (recent_prices[i] > recent_prices[i-1] and recent_prices[i] > recent_prices[i-2] and recent_prices[i] > recent_prices[i+1] and recent_prices[i] > recent_prices[i+2]): resistance_levels.append(recent_prices[i]) # تشخیص دره (محلی حداقل) if (recent_prices[i] < recent_prices[i-1] and recent_prices[i] < recent_prices[i-2] and recent_prices[i] < recent_prices[i+1] and recent_prices[i] < recent_prices[i+2]): support_levels.append(recent_prices[i]) # میانگین‌گیری از سطوح نزدیک به هم cluster_pct = config['cluster_pct'] resistance = self._cluster_levels(resistance_levels, current_price, cluster_pct) support = self._cluster_levels(support_levels, current_price, cluster_pct) return { 'resistance_levels': resistance[:5] if len(resistance) > 0 else [], 'support_levels': support[:5] if len(support) > 0 else [], 'current_price': float(current_price) } def _cluster_levels(self, levels, current_price, cluster_pct=0.01): """خوشه‌بندی سطوح نزدیک به هم""" if len(levels) == 0: return [] levels = sorted(levels) clustered = [] current_cluster = [levels[0]] for level in levels[1:]: if level - current_cluster[-1] < current_price * cluster_pct: current_cluster.append(level) else: clustered.append(np.mean(current_cluster)) current_cluster = [level] clustered.append(np.mean(current_cluster)) # مرتب‌سازی بر اساس نزدیکی به قیمت فعلی clustered.sort(key=lambda x: abs(x - current_price)) return clustered def calculate_adx(self, high, low, close, period=14): """محاسبه شاخص ADX برای تشخیص قدرت روند""" high = np.array(high) low = np.array(low) close = np.array(close) if len(high) < period + 1: return {'adx': 20, 'plus_di': 25, 'minus_di': 25, 'trend': 'رنج', 'action': 'HOLD', 'strength': 'ضعیف'} # محاسبه +DM و -DM up_move = high[1:] - high[:-1] down_move = low[:-1] - low[1:] plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0) minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0) # محاسبه TR (True Range) high_low = high[1:] - low[1:] high_close = np.abs(high[1:] - close[:-1]) low_close = np.abs(low[1:] - close[:-1]) tr = np.maximum(high_low, np.maximum(high_close, low_close)) # میانگین‌گیری atr = pd.Series(tr).rolling(period).mean().values plus_di = 100 * pd.Series(plus_dm).rolling(period).mean().values / atr minus_di = 100 * pd.Series(minus_dm).rolling(period).mean().values / atr # محاسبه DX و ADX dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di) adx = pd.Series(dx).rolling(period).mean().values current_adx = adx[-1] if len(adx) > 0 else 20 current_plus_di = plus_di[-1] if len(plus_di) > 0 else 25 current_minus_di = minus_di[-1] if len(minus_di) > 0 else 25 # تعیین وضعیت روند if current_adx > 25: if current_plus_di > current_minus_di: trend = "صعودی قوی" action = "BUY" else: trend = "نزولی قوی" action = "SELL" elif current_adx > 20: if current_plus_di > current_minus_di: trend = "صعودی ضعیف" action = "WATCH" else: trend = "نزولی ضعیف" action = "WATCH" else: trend = "رنج (بدون روند)" action = "HOLD" return { 'adx': round(float(current_adx), 1), 'plus_di': round(float(current_plus_di), 1), 'minus_di': round(float(current_minus_di), 1), 'trend': trend, 'action': action, 'strength': 'قوی' if current_adx > 25 else 'متوسط' if current_adx > 20 else 'ضعیف' } def calculate_rsi(self, prices, period=14): """محاسبه شاخص RSI""" if len(prices) < period + 1: return {'rsi': 50, 'status': 'نرمال', 'signal': 'NEUTRAL'} delta = np.diff(prices) gain = np.where(delta > 0, delta, 0) loss = np.where(delta < 0, -delta, 0) avg_gain = pd.Series(gain).rolling(period).mean().values[-1] avg_loss = pd.Series(loss).rolling(period).mean().values[-1] if avg_loss == 0: rsi = 100 else: rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) # تعیین وضعیت if rsi > 70: status = "اشباع خرید" signal = "SELL" elif rsi < 30: status = "اشباع فروش" signal = "BUY" else: status = "نرمال" signal = "NEUTRAL" return { 'rsi': round(float(rsi), 1), 'status': status, 'signal': signal } def get_full_analysis(self, df, timeframe='1h'): """تحلیل کامل تکنیکال با تایم فریم مشخص""" prices = df['Close'].values high = df['High'].values low = df['Low'].values sr_levels = self.calculate_support_resistance(prices, timeframe) adx_data = self.calculate_adx(high, low, prices) rsi_data = self.calculate_rsi(prices) return { 'support_resistance': sr_levels, 'adx': adx_data, 'rsi': rsi_data, 'summary': self._generate_summary(sr_levels, adx_data, rsi_data) } def _generate_summary(self, sr, adx, rsi): """تولید خلاصه تحلیل""" current_price = sr['current_price'] support = sr['support_levels'][0] if len(sr['support_levels']) > 0 else current_price * 0.95 resistance = sr['resistance_levels'][0] if len(sr['resistance_levels']) > 0 else current_price * 1.05 distance_to_support = ((current_price - support) / current_price) * 100 distance_to_resistance = ((resistance - current_price) / current_price) * 100 if distance_to_support < 2: position = "نزدیک به حمایت (منطقه خرید احتمالی)" elif distance_to_resistance < 2: position = "نزدیک به مقاومت (منطقه فروش احتمالی)" else: position = "در محدوده خنثی" return { 'position': position, 'next_support': round(support, 2), 'next_resistance': round(resistance, 2), 'adx_status': adx['trend'], 'rsi_status': rsi['status'], 'overall_action': adx['action'] if adx['adx'] > 20 else rsi['signal'] }