sigitholic
352 lines
13 KiB
Python
352 lines
13 KiB
Python
#!/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)
|