Create patterns.py

ml_engine/patterns.py

@@ -0,0 +1,294 @@
+# ml_engine/patterns.py
+import pandas as pd
+import numpy as np
+
+class ChartPatternAnalyzer:
+    def __init__(self):
+        self.pattern_cache = {}
+    
+    async def detect_chart_patterns(self, ohlcv_data):
+        """اكتشاف الأنماط البيانية لجميع الأطر الزمنية"""
+        patterns = {
+            'pattern_detected': 'no_clear_pattern',
+            'pattern_confidence': 0,
+            'predicted_direction': 'neutral',
+            'timeframe_analysis': {},
+            'all_patterns': []
+        }
+        
+        try:
+            for timeframe, candles in ohlcv_data.items():
+                if candles and len(candles) >= 20:
+                    dataframe = self._create_dataframe(candles)
+                    timeframe_pattern = await self._analyze_timeframe_patterns(dataframe, timeframe)
+                    patterns['timeframe_analysis'][timeframe] = timeframe_pattern
+                    patterns['all_patterns'].append(timeframe_pattern)
+                    
+                    if timeframe_pattern['confidence'] > patterns['pattern_confidence']:
+                        patterns.update({
+                            'pattern_detected': timeframe_pattern['pattern'],
+                            'pattern_confidence': timeframe_pattern['confidence'],
+                            'predicted_direction': timeframe_pattern['direction']
+                        })
+            
+            return patterns
+            
+        except Exception as e:
+            print(f"❌ خطأ في اكتشاف الأنماط: {e}")
+            return patterns
+    
+    def _create_dataframe(self, candles):
+        """إنشاء DataFrame من بيانات الشموع"""
+        try:
+            df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+            df[['open', 'high', 'low', 'close', 'volume']] = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
+            return df
+        except Exception as e:
+            print(f"❌ خطأ في إنشاء DataFrame: {e}")
+            return pd.DataFrame()
+    
+    async def _analyze_timeframe_patterns(self, dataframe, timeframe):
+        """تحليل الأنماط لإطار زمني محدد"""
+        pattern_info = {
+            'pattern': 'no_clear_pattern',
+            'confidence': 0,
+            'direction': 'neutral',
+            'timeframe': timeframe,
+            'details': {}
+        }
+        
+        try:
+            if dataframe is None or dataframe.empty or len(dataframe) < 20:
+                return pattern_info
+            
+            closes = dataframe['close'].values
+            highs = dataframe['high'].values
+            lows = dataframe['low'].values
+            current_price = closes[-1]
+            
+            patterns_detected = []
+            
+            double_pattern = self._detect_double_pattern(highs, lows, closes)
+            if double_pattern['detected']:
+                patterns_detected.append(double_pattern)
+            
+            breakout_pattern = self._detect_breakout_pattern(highs, lows, closes)
+            if breakout_pattern['detected']:
+                patterns_detected.append(breakout_pattern)
+            
+            trend_pattern = self._detect_trend_pattern(dataframe)
+            if trend_pattern['detected']:
+                patterns_detected.append(trend_pattern)
+            
+            support_resistance_pattern = self._detect_support_resistance(highs, lows, closes)
+            if support_resistance_pattern['detected']:
+                patterns_detected.append(support_resistance_pattern)
+            
+            if patterns_detected:
+                best_pattern = max(patterns_detected, key=lambda x: x['confidence'])
+                pattern_info.update({
+                    'pattern': best_pattern['pattern'],
+                    'confidence': best_pattern['confidence'],
+                    'direction': best_pattern.get('direction', 'neutral'),
+                    'details': best_pattern.get('details', {})
+                })
+            
+            return pattern_info
+            
+        except Exception as e:
+            print(f"❌ خطأ في تحليل الأنماط للإطار {timeframe}: {e}")
+            return pattern_info
+    
+    def _detect_double_pattern(self, highs, lows, closes):
+        """كشف نمط القمة المزدوجة أو القاع المزدوج"""
+        try:
+            if len(highs) < 15:
+                return {'detected': False}
+            
+            recent_highs = highs[-15:]
+            recent_lows = lows[-15:]
+            
+            high_indices = np.argsort(recent_highs)[-2:]
+            high_indices.sort()
+            
+            low_indices = np.argsort(recent_lows)[:2]
+            low_indices.sort()
+            
+            double_top = False
+            double_bottom = False
+            
+            if len(high_indices) == 2:
+                high1 = recent_highs[high_indices[0]]
+                high2 = recent_highs[high_indices[1]]
+                time_diff = high_indices[1] - high_indices[0]
+                
+                if (abs(high1 - high2) / high1 < 0.02 and
+                    time_diff >= 3 and time_diff <= 10 and
+                    closes[-1] < min(high1, high2)):
+                    double_top = True
+            
+            if len(low_indices) == 2:
+                low1 = recent_lows[low_indices[0]]
+                low2 = recent_lows[low_indices[1]]
+                time_diff = low_indices[1] - low_indices[0]
+                
+                if (abs(low1 - low2) / low1 < 0.02 and
+                    time_diff >= 3 and time_diff <= 10 and
+                    closes[-1] > max(low1, low2)):
+                    double_bottom = True
+            
+            if double_top:
+                return {
+                    'detected': True,
+                    'pattern': 'Double Top',
+                    'confidence': 0.75,
+                    'direction': 'down',
+                    'details': {
+                        'resistance_level': np.mean([high1, high2]),
+                        'breakdown_level': min(lows[-5:])
+                    }
+                }
+            elif double_bottom:
+                return {
+                    'detected': True,
+                    'pattern': 'Double Bottom', 
+                    'confidence': 0.75,
+                    'direction': 'up',
+                    'details': {
+                        'support_level': np.mean([low1, low2]),
+                        'breakout_level': max(highs[-5:])
+                    }
+                }
+            
+            return {'detected': False}
+            
+        except Exception as e:
+            return {'detected': False}
+    
+    def _detect_breakout_pattern(self, highs, lows, closes):
+        """كشف نمط الاختراق"""
+        try:
+            if len(highs) < 25:
+                return {'detected': False}
+            
+            current_price = closes[-1]
+            
+            resistance = np.max(highs[-25:-5])
+            support = np.min(lows[-25:-5])
+            
+            if current_price > resistance * 1.01:
+                return {
+                    'detected': True,
+                    'pattern': 'Breakout Up',
+                    'confidence': 0.8,
+                    'direction': 'up',
+                    'details': {
+                        'breakout_level': resistance,
+                        'target_level': resistance * 1.05
+                    }
+                }
+            elif current_price < support * 0.99:
+                return {
+                    'detected': True,
+                    'pattern': 'Breakout Down',
+                    'confidence': 0.8,
+                    'direction': 'down',
+                    'details': {
+                        'breakdown_level': support,
+                        'target_level': support * 0.95
+                    }
+                }
+            
+            return {'detected': False}
+            
+        except Exception as e:
+            return {'detected': False}
+    
+    def _detect_trend_pattern(self, dataframe):
+        """كشف نمط الاتجاه"""
+        try:
+            if dataframe is None or dataframe.empty or len(dataframe) < 20:
+                return {'detected': False}
+            
+            closes = dataframe['close'].values
+            
+            ma_short = np.mean(closes[-5:])
+            ma_medium = np.mean(closes[-13:])
+            ma_long = np.mean(closes[-21:])
+            
+            if ma_short > ma_medium > ma_long and closes[-1] > ma_short:
+                trend_strength = (ma_short - ma_long) / ma_long
+                confidence = min(0.3 + trend_strength * 10, 0.8)
+                return {
+                    'detected': True,
+                    'pattern': 'Uptrend',
+                    'confidence': confidence,
+                    'direction': 'up',
+                    'details': {
+                        'trend_strength': trend_strength,
+                        'support_level': ma_medium
+                    }
+                }
+            elif ma_short < ma_medium < ma_long and closes[-1] < ma_short:
+                trend_strength = (ma_long - ma_short) / ma_long
+                confidence = min(0.3 + trend_strength * 10, 0.8)
+                return {
+                    'detected': True,
+                    'pattern': 'Downtrend',
+                    'confidence': confidence,
+                    'direction': 'down',
+                    'details': {
+                        'trend_strength': trend_strength,
+                        'resistance_level': ma_medium
+                    }
+                }
+            
+            return {'detected': False}
+            
+        except Exception as e:
+            return {'detected': False}
+    
+    def _detect_support_resistance(self, highs, lows, closes):
+        """كشف مستويات الدعم والمقاومة"""
+        try:
+            if len(highs) < 20:
+                return {'detected': False}
+            
+            current_price = closes[-1]
+            
+            resistance_level = np.max(highs[-20:])
+            support_level = np.min(lows[-20:])
+            
+            position = (current_price - support_level) / (resistance_level - support_level)
+            
+            if position < 0.2:
+                return {
+                    'detected': True,
+                    'pattern': 'Near Support',
+                    'confidence': 0.6,
+                    'direction': 'up',
+                    'details': {
+                        'support_level': support_level,
+                        'resistance_level': resistance_level,
+                        'position': position
+                    }
+                }
+            elif position > 0.8:
+                return {
+                    'detected': True,
+                    'pattern': 'Near Resistance',
+                    'confidence': 0.6,
+                    'direction': 'down',
+                    'details': {
+                        'support_level': support_level,
+                        'resistance_level': resistance_level,
+                        'position': position
+                    }
+                }
+            
+            return {'detected': False}
+            
+        except Exception as e:
+            return {'detected': False}
+
+print("✅ ML Module: Chart Pattern Analyzer loaded")