#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ SmartBot AI Endpoint Example ============================ Contoh endpoint AI sederhana untuk SmartBot MT5 Menggunakan Flask dan beberapa library machine learning Installation: pip install flask pandas numpy scikit-learn requests Usage: python ai_endpoint_example.py Endpoint akan berjalan di: http://localhost:5000/ai/trade """ from flask import Flask, request, jsonify import json import pandas as pd import numpy as np from datetime import datetime import logging # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) class SmartBotAI: def __init__(self): """Initialize AI model and parameters""" self.confidence_threshold = 0.7 self.min_signal_strength = 60 self.risk_levels = { 'low': {'max_confidence': 0.8, 'min_confirmations': 4}, 'medium': {'max_confidence': 0.9, 'min_confirmations': 3}, 'high': {'max_confidence': 0.95, 'min_confirmations': 5} } def analyze_market_conditions(self, data): """Analyze overall market conditions""" try: # Extract indicators rsi = data['indicators']['rsi'] adx = data['indicators']['adx'] ema_fast = data['indicators']['ema_fast'] ema_slow = data['indicators']['ema_slow'] stoch_k = data['indicators']['stoch_k'] stoch_d = data['indicators']['stoch_d'] volume = data['indicators']['volume'] atr = data['atr'] spread = data['spread'] # Market condition analysis conditions = { 'trend_strength': 'weak', 'volatility': 'low', 'momentum': 'neutral', 'volume_profile': 'normal', 'overall_sentiment': 'neutral' } # Trend strength if adx >= 35: conditions['trend_strength'] = 'strong' elif adx >= 25: conditions['trend_strength'] = 'moderate' # Volatility if atr > 0.0020: # High volatility conditions['volatility'] = 'high' elif atr > 0.0010: # Medium volatility conditions['volatility'] = 'medium' # Momentum if rsi > 70: conditions['momentum'] = 'overbought' elif rsi < 30: conditions['momentum'] = 'oversold' elif rsi > 60: conditions['momentum'] = 'bullish' elif rsi < 40: conditions['momentum'] = 'bearish' # Volume profile if volume > 1000: # High volume conditions['volume_profile'] = 'high' elif volume < 100: # Low volume conditions['volume_profile'] = 'low' return conditions except Exception as e: logger.error(f"Error analyzing market conditions: {e}") return None def calculate_signal_confidence(self, data, candidate): """Calculate confidence level for trading signal""" try: # Base confidence from signal strength signal_strength = data.get('signal_strength', 0) confirmations = data.get('confirmations', 0) # Normalize signal strength (0-100 to 0-1) base_confidence = min(signal_strength / 100.0, 1.0) # Adjust based on confirmations confirmation_bonus = min(confirmations * 0.1, 0.3) # Market condition adjustments conditions = self.analyze_market_conditions(data) if conditions: # Trend strength adjustment if conditions['trend_strength'] == 'strong': base_confidence += 0.1 elif conditions['trend_strength'] == 'weak': base_confidence -= 0.1 # Volatility adjustment if conditions['volatility'] == 'high': base_confidence -= 0.05 # Reduce confidence in high volatility # Momentum alignment if candidate == 'BUY' and conditions['momentum'] == 'bullish': base_confidence += 0.05 elif candidate == 'SELL' and conditions['momentum'] == 'bearish': base_confidence += 0.05 elif candidate == 'BUY' and conditions['momentum'] == 'overbought': base_confidence -= 0.1 elif candidate == 'SELL' and conditions['momentum'] == 'oversold': base_confidence -= 0.1 # Spread penalty spread = data.get('spread', 0) if spread > 500: # High spread penalty base_confidence -= 0.1 elif spread > 300: # Medium spread penalty base_confidence -= 0.05 # Mode adjustment mode = data.get('mode', 'intraday') if mode == 'scalping': # Scalping requires higher precision base_confidence -= 0.05 elif mode == 'swing': # Swing trading can be more lenient base_confidence += 0.05 # Final confidence (clamp between 0 and 1) final_confidence = max(0.0, min(1.0, base_confidence + confirmation_bonus)) return final_confidence except Exception as e: logger.error(f"Error calculating confidence: {e}") return 0.5 # Default neutral confidence def generate_trading_decision(self, data): """Generate trading decision based on analysis""" try: candidate = data.get('candidate', '') if not candidate: return self._create_response('reject', 0.0, 'No trading candidate specified') # Calculate confidence confidence = self.calculate_signal_confidence(data, candidate) # Decision logic if confidence < self.confidence_threshold: return self._create_response('reject', confidence, f'Confidence too low ({confidence:.2f} < {self.confidence_threshold})') # Check signal strength signal_strength = data.get('signal_strength', 0) if signal_strength < self.min_signal_strength: return self._create_response('reject', confidence, f'Signal strength too low ({signal_strength} < {self.min_signal_strength})') # Check confirmations confirmations = data.get('confirmations', 0) if confirmations < 3: return self._create_response('reject', confidence, f'Insufficient confirmations ({confirmations} < 3)') # Market conditions check conditions = self.analyze_market_conditions(data) if conditions: if conditions['trend_strength'] == 'weak' and data.get('mode') == 'swing': return self._create_response('reject', confidence, 'Weak trend for swing trading') if conditions['volatility'] == 'high' and data.get('mode') == 'scalping': return self._create_response('reject', confidence, 'High volatility unsuitable for scalping') # Generate TP/SL suggestions tp_sl = self._calculate_optimal_tp_sl(data, candidate) # Decision if candidate == 'BUY': verdict = 'confirm_buy' reason = f'Strong buy signal with {confidence:.2f} confidence' elif candidate == 'SELL': verdict = 'confirm_sell' reason = f'Strong sell signal with {confidence:.2f} confidence' else: return self._create_response('reject', confidence, 'Invalid candidate') return self._create_response(verdict, confidence, reason, tp_sl) except Exception as e: logger.error(f"Error generating decision: {e}") return self._create_response('reject', 0.0, f'Error in analysis: {str(e)}') def _calculate_optimal_tp_sl(self, data, candidate): """Calculate optimal TP/SL levels""" try: # This is a simplified calculation # In real implementation, you might use more sophisticated methods # Get current price (approximate from indicators) ema_fast = data['indicators']['ema_fast'] ema_slow = data['indicators']['ema_slow'] atr = data['atr'] # Use EMA crossover as reference price current_price = (ema_fast + ema_slow) / 2 # Calculate TP/SL based on ATR if candidate == 'BUY': suggested_sl = current_price - (atr * 1.5) suggested_tp = current_price + (atr * 2.0) else: # SELL suggested_sl = current_price + (atr * 1.5) suggested_tp = current_price - (atr * 2.0) return { 'suggested_sl': round(suggested_sl, 5), 'suggested_tp': round(suggested_tp, 5) } except Exception as e: logger.error(f"Error calculating TP/SL: {e}") return None def _create_response(self, verdict, confidence, reason, tp_sl=None): """Create standardized response""" response = { 'verdict': verdict, 'confidence': round(confidence, 3), 'reason': reason, 'timestamp': datetime.now().isoformat() } if tp_sl: response.update(tp_sl) return response # Initialize AI ai_model = SmartBotAI() @app.route('/ai/trade', methods=['POST']) def trade_analysis(): """Main endpoint for trading analysis""" try: # Get request data data = request.get_json() if not data: return jsonify({ 'error': 'No data provided', 'verdict': 'reject' }), 400 logger.info(f"Received trade analysis request: {data.get('pair', 'Unknown')}") # Generate AI decision decision = ai_model.generate_trading_decision(data) logger.info(f"AI Decision: {decision['verdict']} (confidence: {decision['confidence']})") return jsonify(decision) except Exception as e: logger.error(f"Error in trade analysis: {e}") return jsonify({ 'error': str(e), 'verdict': 'reject', 'confidence': 0.0, 'reason': f'Server error: {str(e)}' }), 500 @app.route('/health', methods=['GET']) def health_check(): """Health check endpoint""" return jsonify({ 'status': 'healthy', 'service': 'SmartBot AI', 'timestamp': datetime.now().isoformat() }) @app.route('/', methods=['GET']) def home(): """Home page with API documentation""" return jsonify({ 'service': 'SmartBot AI Trading Assistant', 'version': '1.0.0', 'endpoints': { '/ai/trade': 'POST - Trading analysis', '/health': 'GET - Health check', '/': 'GET - This documentation' }, 'usage': { 'method': 'POST', 'url': '/ai/trade', 'content_type': 'application/json', 'body_format': { 'pair': 'string (e.g., "EURUSD")', 'tf': 'string (e.g., "PERIOD_M5")', 'spread': 'integer', 'atr': 'float', 'indicators': { 'ema_fast': 'float', 'ema_slow': 'float', 'rsi': 'float', 'adx': 'float', 'stoch_k': 'float', 'stoch_d': 'float', 'volume': 'float' }, 'candidate': 'string ("BUY" or "SELL")', 'mode': 'string ("scalping", "intraday", or "swing")', 'confirmations': 'integer', 'signal_strength': 'float' } } }) if __name__ == '__main__': print("🤖 SmartBot AI Endpoint Starting...") print("📍 Endpoint: http://localhost:5000/ai/trade") print("📊 Health Check: http://localhost:5000/health") print("📚 Documentation: http://localhost:5000/") print("=" * 50) # Run the Flask app app.run(host='0.0.0.0', port=5000, debug=True)