sigitholic
@@ -0,0 +1,351 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user