golden_maverick_bot.py

python
#!/usr/bin/env python3
import os
import time
import json
import asyncio
import numpy as np
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv
from tradingview_ta import TA_Handler, Interval
from ta.trend import MACD, EMAIndicator
from ta.momentum import RSIIndicator
from ta.volatility import BollingerBands
from ta.volatility import AverageTrueRange
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import load_model

# Load environment variables
load_dotenv()

class GoldenMaverick:
    """
    AI-powered Gold (XAU/USD) Trading Bot with automated technical analysis and trade execution
    """
    
    def __init__(self):
        self.symbol = "XAUUSD"
        self.exchange = "OANDA"
        self.timeframe = Interval.INTERVAL_5_MINUTES
        self.model_path = "models/gold_predictor.h5"
        self.risk_per_trade = 0.02  # 2% risk per trade
        self.take_profit_ratio = 2.0  # 1:2 risk-reward
        self.trailing_stop_enabled = True
        self.trailing_stop_distance = 0.0020  # 20 pips for XAU/USD
        self.last_signal = None
        self.account_balance = 10000  # Default, should be fetched from broker
        self.load_ai_model()
        
    def load_ai_model(self):
        """Load pre-trained AI model for market prediction"""
        try:
            self.model = load_model(self.model_path)
            print("AI Model loaded successfully")
        except Exception as e:
            print(f"Error loading AI model: {e}")
            self.model = None

    async def fetch_market_data(self):
        """Fetch real-time market data from TradingView"""
        while True:
            try:
                handler = TA_Handler(
                    symbol=self.symbol,
                    exchange=self.exchange,
                    screener="forex",
                    interval=self.timeframe,
                    timeout=10
                )
                analysis = handler.get_analysis()
                self.current_price = analysis.indicators['close']
                self.ohlc_data = {
                    'open': analysis.indicators['open'],
                    'high': analysis.indicators['high'],
                    'low': analysis.indicators['low'],
                    'close': self.current_price,
                    'volume': analysis.indicators['volume']
                }
                print(f"\n[{datetime.now()}] XAU/USD Price: {self.current_price:.2f}")
                await self.analyze_market()
                await asyncio.sleep(30)  # Update every 30 seconds
            except Exception as e:
                print(f"Error fetching data: {e}")
                await asyncio.sleep(60)

    async def analyze_market(self):
        """Perform comprehensive technical analysis"""
        # Load data into DataFrame
        df = pd.DataFrame([self.ohlc_data])
        
        # Calculate indicators
        df = self.calculate_technical_indicators(df)
        
        # AI-based prediction
        ai_signal = await self.generate_ai_signal(df)
        
        # Traditional TA signals
        ta_signals = self.generate_ta_signals(df)
        
        # Combine signals
        final_signal = self.combine_signals(ai_signal, ta_signals)
        
        # Execute trading logic
        if final_signal['signal'] != 'WAIT':
            await self.execute_trade(final_signal)

    def calculate_technical_indicators(self, df):
        """Calculate all technical indicators"""
        # Moving Averages
        df['ema_20'] = EMAIndicator(df['close'], window=20).ema_indicator()
        df['ema_50'] = EMAIndicator(df['close'], window=50).ema_indicator()
        df['ema_200'] = EMAIndicator(df['close'], window=200).ema_indicator()
        
        # Momentum Indicators
        df['rsi'] = RSIIndicator(df['close'], window=14).rsi()
        macd = MACD(df['close'], window_slow=26, window_fast=12, window_sign=9)
        df['macd'] = macd.macd()
        df['macd_signal'] = macd.macd_signal()
        df['macd_diff'] = macd.macd_diff()
        
        # Volatility Indicators
        bb = BollingerBands(df['close'], window=20, window_dev=2)
        df['bb_upper'] = bb.bollinger_hband()
        df['bb_middle'] = bb.bollinger_mavg()
        df['bb_lower'] = bb.bollinger_lband()
        df['atr'] = AverageTrueRange(
            high=df['high'], 
            low=df['low'], 
            close=df['close'], 
            window=14
        ).average_true_range()
        
        return df

    async def generate_ai_signal(self, df):
        """Generate signal using AI model"""
        if not self.model:
            return {'signal': 'WAIT', 'confidence': 0, 'reason': 'AI Model not available'}
        
        try:
            # Prepare data for AI model
            features = df[['close', 'ema_20', 'ema_50', 'rsi', 'macd', 'atr']].values
            scaler = MinMaxScaler()
            features_scaled = scaler.fit_transform(features)
            features_scaled = np.reshape(features_scaled, (1, features_scaled.shape[0], features_scaled.shape[1]))
            
            # Make prediction
            prediction = self.model.predict(features_scaled)
            confidence = np.max(prediction)
            signal = np.argmax(prediction)
            
            return {
                'signal': 'BUY' if signal == 0 else 'SELL' if signal == 1 else 'WAIT',
                'confidence': float(confidence),
                'reason': 'AI Market Prediction'
            }
        except Exception as e:
            print(f"AI Prediction Error: {e}")
            return {'signal': 'WAIT', 'confidence': 0, 'reason': 'Prediction Error'}

    def generate_ta_signals(self, df):
        """Generate signals based on traditional technical analysis"""
        signals = []
        confidence = 0
        reasons = []
        
        # EMA Crossover Strategy
        if df['ema_20'].iloc[-1] > df['ema_50'].iloc[-1] and df['ema_50'].iloc[-1] > df['ema_200'].iloc[-1]:
            signals.append('BUY')
            confidence += 0.3
            reasons.append("Golden Cross (EMA 20 > 50 > 200)")
        elif df['ema_20'].iloc[-1] < df['ema_50'].iloc[-1] and df['ema_50'].iloc[-1] < df['ema_200'].iloc[-1]:
            signals.append('SELL')
            confidence += 0.3
            reasons.append("Death Cross (EMA 20 < 50 < 200)")
        
        # RSI Analysis
        if df['rsi'].iloc[-1] < 30:
            signals.append('BUY')
            confidence += 0.2
            reasons.append("Oversold (RSI < 30)")
        elif df['rsi'].iloc[-1] > 70:
            signals.append('SELL')
            confidence += 0.2
            reasons.append("Overbought (RSI > 70)")
        
        # MACD Analysis
        if df['macd'].iloc[-1] > df['macd_signal'].iloc[-1] and df['macd'].iloc[-2] <= df['macd_signal'].iloc[-2]:
            signals.append('BUY')
            confidence += 0.2
            reasons.append("MACD Bullish Crossover")
        elif df['macd'].iloc[-1] < df['macd_signal'].iloc[-1] and df['macd'].iloc[-2] >= df['macd_signal'].iloc[-2]:
            signals.append('SELL')
            confidence += 0.2
            reasons.append("MACD Bearish Crossover")
        
        # Bollinger Bands Analysis
        if df['close'].iloc[-1] < df['bb_lower'].iloc[-1]:
            signals.append('BUY')
            confidence += 0.1
            reasons.append("Price at Lower Bollinger Band")
        elif df['close'].iloc[-1] > df['bb_upper'].iloc[-1]:
            signals.append('SELL')
            confidence += 0.1
            reasons.append("Price at Upper Bollinger Band")
        
        # Determine final signal
        if not signals:
            return {'signal': 'WAIT', 'confidence': 0, 'reason': 'No strong signals detected'}
        
        # Count votes and determine primary signal
        buy_count = signals.count('BUY')
        sell_count = signals.count('SELL')
        
        final_signal = 'BUY' if buy_count > sell_count else 'SELL' if sell_count > buy_count else 'WAIT'
        
        return {
            'signal': final_signal,
            'confidence': min(confidence, 0.9),  # Cap at 90% for TA alone
            'reason': " | ".join(reasons)
        }

    def combine_signals(self, ai_signal, ta_signal):
        """Combine AI and traditional TA signals with weighted confidence"""
        if ai_signal['signal'] == 'WAIT' and ta_signal['signal'] == 'WAIT':
            return {'signal': 'WAIT', 'confidence': 0, 'reason': 'No consensus'}
        
        # Weighted combination (60% AI, 40% TA)
        ai_weight = 0.6
        ta_weight = 0.4
        
        if ai_signal['signal'] == ta_signal['signal']:
            combined_confidence = (ai_signal['confidence'] * ai_weight + 
                                 ta_signal['confidence'] * ta_weight)
            return {
                'signal': ai_signal['signal'],
                'confidence': combined_confidence,
                'reason': f"AI+TA Consensus: {ai_signal['reason']} & {ta_signal['reason']}"
            }
        else:
            # When signals conflict, prefer AI with higher confidence
            if ai_signal['confidence'] >= 0.7:
                return ai_signal
            elif ta_signal['confidence'] >= 0.7:
                return ta_signal
            else:
                return {'signal': 'WAIT', 'confidence': 0, 'reason': 'Conflicting signals'}

    async def execute_trade(self, signal):
        """Execute trade based on generated signal"""
        # Check if this is a new signal
        if self.last_signal == signal['signal']:
            print(f"Maintaining existing {signal['signal']} position")
            return
            
        print(f"\n🚀 Executing {signal['signal']} Signal 🚀")
        print(f"Confidence: {signal['confidence']*100:.1f}%")
        print(f"Reason: {signal['reason']}")
        print(f"Current Price: {self.current_price:.2f}")
        
        # Calculate position size and risk parameters
        stop_loss = self.calculate_stop_loss(signal['signal'])
        take_profit = self.calculate_take_profit(signal['signal'], stop_loss)
        position_size = self.calculate_position_size(stop_loss)
        
        # Execute trade through broker API (pseudo-code)
        trade_success = await self.send_to_broker(
            symbol=self.symbol,
            action=signal['signal'],
            size=position_size,
            stop_loss=stop_loss,
            take_profit=take_profit
        )
        
        if trade_success:
            self.last_signal = signal['signal']
            await self.send_notification(
                action=signal['signal'],
                price=self.current_price,
                stop_loss=stop_loss,
                take_profit=take_profit,
                confidence=signal['confidence'],
                reason=signal['reason']
            )

    def calculate_stop_loss(self, signal_type):
        """Calculate stop loss based on ATR"""
        if signal_type == 'BUY':
            return self.current_price - (self.ohlc_data['atr'] * 1.5)
        elif signal_type == 'SELL':
            return self.current_price + (self.ohlc_data['atr'] * 1.5)
        return None

    def calculate_take_profit(self, signal_type, stop_loss):
        """Calculate take profit based on risk-reward ratio"""
        if signal_type == 'BUY':
            return self.current_price + abs(self.current_price - stop_loss) * self.take_profit_ratio
        elif signal_type == 'SELL':
            return self.current_price - abs(stop_loss - self.current_price) * self.take_profit_ratio
        return None

    def calculate_position_size(self, stop_loss):
        """Calculate position size based on account balance and risk"""
        risk_amount = self.account_balance * self.risk_per_trade
        risk_per_unit = abs(self.current_price - stop_loss)
        
        if risk_per_unit == 0:
            return 0
            
        # For XAU/USD, 1 lot = 100 oz, price per pip is $0.10 for mini lots
        position_size = (risk_amount / risk_per_unit) / 100
        return round(position_size, 2)  # Round to 2 decimal places

    async def send_to_broker(self, **trade_params):
        """Placeholder for broker API integration"""
        print(f"\n📊 Would execute trade with params:")
        for k, v in trade_params.items():
            print(f"{k}: {v}")
        
        # In production, implement actual broker API calls here
        # Example for MT5:
        # mt5.initialize()
        # request = {
        #     "action": mt5.TRADE_ACTION_DEAL,
        #     "symbol": trade_params['symbol'],
        #     "volume": trade_params['size'],
        #     "type": mt5.ORDER_TYPE_BUY if trade_params['action'] == 'BUY' else mt5.ORDER_TYPE_SELL,
        #     "price": trade_params['price'],
        #     "sl": trade_params['stop_loss'],
        #     "tp": trade_params['take_profit'],
        #     "deviation": 10,
        #     "magic": 123456,
        #     "comment": "Golden Maverick Trade",
        #     "type_time": mt5.ORDER_TIME_GTC,
        #     "type_filling": mt5.ORDER_FILLING_IOC,
        # }
        # result = mt5.order_send(request)
        
        # Simulate success for demo
        return True

    async def send_notification(self, **kwargs):
        """Send trade notification via Telegram or other channels"""
        message = (
            f"🟢 New Trade Alert 🟢\n\n"
            f"Symbol: {self.symbol}\n"
            f"Action: {kwargs['action']}\n"
            f"Price: {kwargs['price']:.2f}\n"
            f"Stop Loss: {kwargs['stop_loss']:.2f}\n"
            f"Take Profit: {kwargs['take_profit']:.2f}\n"
            f"Confidence: {kwargs['confidence']*100:.1f}%\n"
            f"Reason: {kwargs['reason']}\n\n"
            f"Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        )
        
        print(f"\n📩 Notification:\n{message}")
        
        # Actual Telegram implementation would look like:
        # if os.getenv('TELEGRAM_BOT_TOKEN'):
        #     bot = telegram.Bot(token=os.getenv('TELEGRAM_BOT_TOKEN'))
        #     await bot.send_message(
        #         chat_id=os.getenv('TELEGRAM_CHAT_ID'),
        #         text=message
        #     )

async def main():
    bot = GoldenMaverick()
    print("🔥 Golden Maverick AI Gold Trading Bot Activated 🔥")
    print(f"Tracking: {bot.symbol} | Timeframe: {bot.timeframe}")
    await bot.fetch_market_data()

if __name__ == "__main__":
    asyncio.run(main())

</html>