Real-Time Arbitrage Bot: Ollama Multi-Exchange Cryptocurrency Trading Tutorial

# Install required Python packages
pip install ccxt pandas numpy requests ollama websocket-client

# Additional dependencies for advanced features
pip install asyncio aiohttp ta-lib matplotlib

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull required models
ollama pull llama2
ollama pull codellama

import ccxt
import asyncio
import pandas as pd
from datetime import datetime
import logging

class PriceMonitor:
    def __init__(self, exchanges, symbols):
        self.exchanges = exchanges
        self.symbols = symbols
        self.price_data = {}
        self.logger = self.setup_logger()
    
    def setup_logger(self):
        """Configure logging for trade monitoring"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        return logging.getLogger(__name__)
    
    async def fetch_prices(self, exchange, symbol):
        """Fetch current price from specific exchange"""
        try:
            ticker = await exchange.fetch_ticker(symbol)
            return {
                'exchange': exchange.name,
                'symbol': symbol,
                'bid': ticker['bid'],
                'ask': ticker['ask'],
                'timestamp': datetime.now()
            }
        except Exception as e:
            self.logger.error(f"Error fetching {symbol} from {exchange.name}: {e}")
            return None
    
    async def monitor_all_exchanges(self):
        """Monitor prices across all exchanges simultaneously"""
        while True:
            tasks = []
            for exchange in self.exchanges:
                for symbol in self.symbols:
                    task = self.fetch_prices(exchange, symbol)
                    tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # Process results and update price data
            for result in results:
                if result and not isinstance(result, Exception):
                    key = f"{result['exchange']}_{result['symbol']}"
                    self.price_data[key] = result
            
            await asyncio.sleep(1)  # Update every second

class ArbitrageDetector:
    def __init__(self, min_profit_threshold=0.5):
        self.min_profit_threshold = min_profit_threshold
        self.opportunities = []
    
    def calculate_arbitrage_profit(self, buy_price, sell_price, amount, fees):
        """Calculate net profit after fees"""
        buy_cost = amount * buy_price * (1 + fees['buy'])
        sell_revenue = amount * sell_price * (1 - fees['sell'])
        net_profit = sell_revenue - buy_cost
        profit_percentage = (net_profit / buy_cost) * 100
        
        return {
            'net_profit': net_profit,
            'profit_percentage': profit_percentage,
            'buy_cost': buy_cost,
            'sell_revenue': sell_revenue
        }
    
    def find_opportunities(self, price_data):
        """Identify profitable arbitrage opportunities"""
        opportunities = []
        
        # Group prices by symbol
        by_symbol = {}
        for key, data in price_data.items():
            symbol = data['symbol']
            if symbol not in by_symbol:
                by_symbol[symbol] = []
            by_symbol[symbol].append(data)
        
        # Check each symbol for arbitrage opportunities
        for symbol, prices in by_symbol.items():
            if len(prices) < 2:
                continue
                
            # Find best buy and sell prices
            best_buy = min(prices, key=lambda x: x['ask'])
            best_sell = max(prices, key=lambda x: x['bid'])
            
            if best_buy['exchange'] != best_sell['exchange']:
                profit = self.calculate_arbitrage_profit(
                    best_buy['ask'], 
                    best_sell['bid'], 
                    1000,  # $1000 trade size
                    {'buy': 0.001, 'sell': 0.001}  # 0.1% fees
                )
                
                if profit['profit_percentage'] > self.min_profit_threshold:
                    opportunities.append({
                        'symbol': symbol,
                        'buy_exchange': best_buy['exchange'],
                        'sell_exchange': best_sell['exchange'],
                        'buy_price': best_buy['ask'],
                        'sell_price': best_sell['bid'],
                        'profit_data': profit,
                        'timestamp': datetime.now()
                    })
        
        return opportunities

import ollama
import json

class OllamaMarketAnalyzer:
    def __init__(self, model_name='llama2'):
        self.model_name = model_name
        self.client = ollama.Client()
    
    def analyze_market_conditions(self, price_data, news_data=None):
        """Use Ollama to analyze market conditions"""
        prompt = f"""
        Analyze these cryptocurrency market conditions:
        
        Price Data: {json.dumps(price_data, indent=2)}
        
        Provide analysis on:
        1. Market volatility assessment
        2. Arbitrage risk factors
        3. Recommended position sizing
        4. Exit strategy suggestions
        
        Format response as JSON with clear recommendations.
        """
        
        response = self.client.generate(
            model=self.model_name,
            prompt=prompt,
            options={
                'temperature': 0.1,  # Low temperature for consistent analysis
                'num_ctx': 4096
            }
        )
        
        return self.parse_ollama_response(response['response'])
    
    def parse_ollama_response(self, response):
        """Parse Ollama response into structured data"""
        try:
            # Extract JSON from response
            start_idx = response.find('{')
            end_idx = response.rfind('}') + 1
            json_str = response[start_idx:end_idx]
            return json.loads(json_str)
        except:
            return {'error': 'Failed to parse response', 'raw': response}
    
    def risk_assessment(self, opportunity):
        """Assess risk for specific arbitrage opportunity"""
        prompt = f"""
        Assess the risk for this arbitrage opportunity:
        
        Symbol: {opportunity['symbol']}
        Buy Exchange: {opportunity['buy_exchange']}
        Sell Exchange: {opportunity['sell_exchange']}
        Profit Percentage: {opportunity['profit_data']['profit_percentage']}%
        
        Consider:
        1. Exchange reliability
        2. Liquidity risks
        3. Transfer time risks
        4. Regulatory concerns
        
        Rate risk from 1-10 (1=low, 10=high) and explain reasoning.
        """
        
        response = self.client.generate(
            model=self.model_name,
            prompt=prompt,
            options={'temperature': 0.1}
        )
        
        return response['response']

class AdaptiveTrader:
    def __init__(self, initial_strategy):
        self.strategy = initial_strategy
        self.performance_history = []
        self.ollama_analyzer = OllamaMarketAnalyzer()
    
    def adjust_strategy(self, recent_performance):
        """Use Ollama to adjust trading strategy based on performance"""
        performance_summary = {
            'successful_trades': len([t for t in recent_performance if t['profit'] > 0]),
            'failed_trades': len([t for t in recent_performance if t['profit'] <= 0]),
            'average_profit': sum(t['profit'] for t in recent_performance) / len(recent_performance),
            'max_drawdown': min(t['profit'] for t in recent_performance)
        }
        
        prompt = f"""
        Current trading strategy parameters:
        - Min profit threshold: {self.strategy['min_profit_threshold']}%
        - Max position size: ${self.strategy['max_position_size']}
        - Risk tolerance: {self.strategy['risk_tolerance']}
        
        Recent performance:
        {json.dumps(performance_summary, indent=2)}
        
        Suggest strategy adjustments to improve performance.
        Consider risk management and market conditions.
        """
        
        response = self.ollama_analyzer.client.generate(
            model=self.ollama_analyzer.model_name,
            prompt=prompt,
            options={'temperature': 0.2}
        )
        
        return self.parse_strategy_adjustments(response['response'])
    
    def parse_strategy_adjustments(self, response):
        """Parse strategy recommendations from Ollama"""
        # Implementation to extract specific parameter changes
        # This would include natural language processing to identify
        # numerical adjustments and strategy modifications
        pass

class PortfolioManager:
    def __init__(self, exchanges, initial_balances):
        self.exchanges = exchanges
        self.balances = initial_balances
        self.trade_history = []
        self.risk_limits = {
            'max_position_size': 0.1,  # 10% of portfolio
            'max_daily_trades': 50,
            'max_drawdown': 0.05  # 5% max drawdown
        }
    
    def check_balance_requirements(self, opportunity, trade_size):
        """Verify sufficient balance for arbitrage trade"""
        buy_exchange = opportunity['buy_exchange']
        sell_exchange = opportunity['sell_exchange']
        
        # Check if we have enough balance on both exchanges
        required_buy_balance = trade_size * opportunity['buy_price']
        required_sell_balance = trade_size
        
        buy_balance = self.get_balance(buy_exchange, 'USD')
        sell_balance = self.get_balance(sell_exchange, opportunity['symbol'])
        
        return {
            'can_execute': (buy_balance >= required_buy_balance and 
                           sell_balance >= required_sell_balance),
            'buy_balance': buy_balance,
            'sell_balance': sell_balance,
            'required_buy': required_buy_balance,
            'required_sell': required_sell_balance
        }
    
    def rebalance_exchanges(self):
        """Rebalance funds across exchanges for optimal arbitrage"""
        # Implementation for automatic rebalancing
        # This ensures funds are distributed optimally
        pass

class ArbitrageExecutor:
    def __init__(self, exchanges, portfolio_manager):
        self.exchanges = exchanges
        self.portfolio_manager = portfolio_manager
        self.active_trades = {}
        self.logger = logging.getLogger(__name__)
    
    async def execute_arbitrage(self, opportunity):
        """Execute arbitrage trade with proper error handling"""
        trade_id = f"{opportunity['symbol']}_{int(datetime.now().timestamp())}"
        
        try:
            # Pre-execution checks
            balance_check = self.portfolio_manager.check_balance_requirements(
                opportunity, 1000  # $1000 trade size
            )
            
            if not balance_check['can_execute']:
                self.logger.warning(f"Insufficient balance for {trade_id}")
                return {'status': 'failed', 'reason': 'insufficient_balance'}
            
            # Execute buy order
            buy_exchange = self.get_exchange(opportunity['buy_exchange'])
            buy_order = await buy_exchange.create_market_buy_order(
                opportunity['symbol'],
                1000 / opportunity['buy_price']
            )
            
            # Execute sell order
            sell_exchange = self.get_exchange(opportunity['sell_exchange'])
            sell_order = await sell_exchange.create_market_sell_order(
                opportunity['symbol'],
                1000 / opportunity['buy_price']
            )
            
            # Record trade
            trade_record = {
                'trade_id': trade_id,
                'buy_order': buy_order,
                'sell_order': sell_order,
                'expected_profit': opportunity['profit_data']['net_profit'],
                'timestamp': datetime.now()
            }
            
            self.active_trades[trade_id] = trade_record
            self.logger.info(f"Executed arbitrage trade {trade_id}")
            
            return {'status': 'success', 'trade_id': trade_id}
            
        except Exception as e:
            self.logger.error(f"Error executing arbitrage {trade_id}: {e}")
            return {'status': 'failed', 'reason': str(e)}
    
    def get_exchange(self, exchange_name):
        """Get exchange instance by name"""
        return next(ex for ex in self.exchanges if ex.name == exchange_name)

class OllamaArbitrageBot:
    def __init__(self, config):
        self.config = config
        self.exchanges = self.initialize_exchanges()
        self.price_monitor = PriceMonitor(self.exchanges, config['symbols'])
        self.detector = ArbitrageDetector(config['min_profit_threshold'])
        self.analyzer = OllamaMarketAnalyzer()
        self.portfolio_manager = PortfolioManager(self.exchanges, config['initial_balances'])
        self.executor = ArbitrageExecutor(self.exchanges, self.portfolio_manager)
        self.running = False
    
    def initialize_exchanges(self):
        """Initialize exchange connections"""
        exchanges = []
        for exchange_config in self.config['exchanges']:
            exchange_class = getattr(ccxt, exchange_config['name'])
            exchange = exchange_class({
                'apiKey': exchange_config['api_key'],
                'secret': exchange_config['secret'],
                'sandbox': exchange_config.get('sandbox', False),
                'enableRateLimit': True,
            })
            exchanges.append(exchange)
        return exchanges
    
    async def run(self):
        """Main bot execution loop"""
        self.running = True
        
        # Start price monitoring
        monitor_task = asyncio.create_task(self.price_monitor.monitor_all_exchanges())
        
        while self.running:
            try:
                # Find arbitrage opportunities
                opportunities = self.detector.find_opportunities(
                    self.price_monitor.price_data
                )
                
                for opportunity in opportunities:
                    # Analyze with Ollama
                    risk_assessment = self.analyzer.risk_assessment(opportunity)
                    
                    # Execute if risk is acceptable
                    if self.should_execute_trade(opportunity, risk_assessment):
                        result = await self.executor.execute_arbitrage(opportunity)
                        
                        if result['status'] == 'success':
                            print(f"Executed profitable arbitrage: {result['trade_id']}")
                
                await asyncio.sleep(5)  # Check every 5 seconds
                
            except KeyboardInterrupt:
                print("Shutting down bot...")
                self.running = False
                break
            except Exception as e:
                print(f"Error in main loop: {e}")
                await asyncio.sleep(10)
    
    def should_execute_trade(self, opportunity, risk_assessment):
        """Determine if trade should be executed based on risk analysis"""
        # Parse risk score from Ollama response
        # This would include NLP to extract risk rating
        # For now, simple profit threshold check
        return opportunity['profit_data']['profit_percentage'] > self.config['min_profit_threshold']

# config.py
BOT_CONFIG = {
    'exchanges': [
        {
            'name': 'binance',
            'api_key': 'your_binance_api_key',
            'secret': 'your_binance_secret',
            'sandbox': True  # Use sandbox for testing
        },
        {
            'name': 'coinbasepro',
            'api_key': 'your_coinbase_api_key',
            'secret': 'your_coinbase_secret',
            'sandbox': True
        }
    ],
    'symbols': ['BTC/USD', 'ETH/USD', 'ADA/USD'],
    'min_profit_threshold': 0.5,  # 0.5% minimum profit
    'initial_balances': {
        'binance': {'USD': 5000, 'BTC': 0.1},
        'coinbasepro': {'USD': 5000, 'BTC': 0.1}
    },
    'risk_management': {
        'max_position_size': 1000,  # Maximum $1000 per trade
        'max_daily_trades': 20,
        'stop_loss_percentage': 2.0
    }
}

# main.py
async def main():
    bot = OllamaArbitrageBot(BOT_CONFIG)
    await bot.run()

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

class ArbitrageBacktester:
    def __init__(self, historical_data):
        self.historical_data = historical_data
        self.results = []
    
    def run_backtest(self, strategy_config):
        """Run backtest on historical data"""
        detector = ArbitrageDetector(strategy_config['min_profit_threshold'])
        
        for timestamp, price_data in self.historical_data.items():
            opportunities = detector.find_opportunities(price_data)
            
            for opportunity in opportunities:
                # Simulate trade execution
                trade_result = self.simulate_trade(opportunity)
                self.results.append(trade_result)
        
        return self.calculate_performance_metrics()
    
    def simulate_trade(self, opportunity):
        """Simulate trade execution with realistic slippage"""
        # Add realistic slippage and timing delays
        slippage = 0.05  # 0.05% slippage
        actual_profit = opportunity['profit_data']['profit_percentage'] - slippage
        
        return {
            'timestamp': opportunity['timestamp'],
            'symbol': opportunity['symbol'],
            'expected_profit': opportunity['profit_data']['profit_percentage'],
            'actual_profit': actual_profit,
            'success': actual_profit > 0
        }
    
    def calculate_performance_metrics(self):
        """Calculate backtesting performance metrics"""
        total_trades = len(self.results)
        successful_trades = sum(1 for r in self.results if r['success'])
        
        return {
            'total_trades': total_trades,
            'success_rate': successful_trades / total_trades if total_trades > 0 else 0,
            'average_profit': sum(r['actual_profit'] for r in self.results) / total_trades,
            'max_profit': max(r['actual_profit'] for r in self.results),
            'min_profit': min(r['actual_profit'] for r in self.results)
        }

class PaperTrader:
    def __init__(self, initial_balance=10000):
        self.balance = initial_balance
        self.positions = {}
        self.trade_history = []
        self.start_balance = initial_balance
    
    def execute_paper_trade(self, opportunity):
        """Execute trade without real money"""
        trade_size = min(1000, self.balance * 0.1)  # Use 10% of balance
        
        # Simulate trade execution
        profit = trade_size * (opportunity['profit_data']['profit_percentage'] / 100)
        
        self.balance += profit
        
        trade_record = {
            'timestamp': datetime.now(),
            'symbol': opportunity['symbol'],
            'trade_size': trade_size,
            'profit': profit,
            'balance_after': self.balance
        }
        
        self.trade_history.append(trade_record)
        
        return trade_record
    
    def get_performance_summary(self):
        """Get trading performance summary"""
        total_return = ((self.balance - self.start_balance) / self.start_balance) * 100
        
        return {
            'total_return': total_return,
            'current_balance': self.balance,
            'total_trades': len(self.trade_history),
            'profitable_trades': len([t for t in self.trade_history if t['profit'] > 0])
        }

import matplotlib.pyplot as plt
import pandas as pd

class PerformanceMonitor:
    def __init__(self, bot):
        self.bot = bot
        self.metrics = []
    
    def collect_metrics(self):
        """Collect current performance metrics"""
        current_metrics = {
            'timestamp': datetime.now(),
            'active_opportunities': len(self.bot.detector.opportunities),
            'portfolio_value': self.bot.portfolio_manager.get_total_value(),
            'total_trades': len(self.bot.executor.active_trades),
            'success_rate': self.calculate_success_rate()
        }
        
        self.metrics.append(current_metrics)
        return current_metrics
    
    def generate_performance_report(self):
        """Generate comprehensive performance report"""
        df = pd.DataFrame(self.metrics)
        
        # Create visualizations
        fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 10))
        
        # Portfolio value over time
        ax1.plot(df['timestamp'], df['portfolio_value'])
        ax1.set_title('Portfolio Value Over Time')
        ax1.set_xlabel('Time')
        ax1.set_ylabel('Value ($)')
        
        # Success rate
        ax2.plot(df['timestamp'], df['success_rate'])
        ax2.set_title('Trade Success Rate')
        ax2.set_xlabel('Time')
        ax2.set_ylabel('Success Rate (%)')
        
        # Active opportunities
        ax3.plot(df['timestamp'], df['active_opportunities'])
        ax3.set_title('Active Arbitrage Opportunities')
        ax3.set_xlabel('Time')
        ax3.set_ylabel('Count')
        
        # Total trades
        ax4.plot(df['timestamp'], df['total_trades'])
        ax4.set_title('Cumulative Trades')
        ax4.set_xlabel('Time')
        ax4.set_ylabel('Count')
        
        plt.tight_layout()
        plt.savefig('performance_report.png')
        plt.show()
    
    def calculate_success_rate(self):
        """Calculate current success rate"""
        completed_trades = [t for t in self.bot.executor.active_trades.values() 
                          if t.get('status') == 'completed']
        
        if not completed_trades:
            return 0
        
        successful = len([t for t in completed_trades if t.get('profit', 0) > 0])
        return (successful / len(completed_trades)) * 100

import hashlib
import hmac
import time
from cryptography.fernet import Fernet

class SecurityManager:
    def __init__(self):
        self.encryption_key = Fernet.generate_key()
        self.cipher = Fernet(self.encryption_key)
    
    def encrypt_api_keys(self, api_key, secret):
        """Encrypt API credentials"""
        encrypted_key = self.cipher.encrypt(api_key.encode())
        encrypted_secret = self.cipher.encrypt(secret.encode())
        
        return {
            'encrypted_key': encrypted_key,
            'encrypted_secret': encrypted_secret
        }
    
    def decrypt_api_keys(self, encrypted_data):
        """Decrypt API credentials"""
        api_key = self.cipher.decrypt(encrypted_data['encrypted_key']).decode()
        secret = self.cipher.decrypt(encrypted_data['encrypted_secret']).decode()
        
        return api_key, secret
    
    def validate_trade_signature(self, trade_data, signature):
        """Validate trade execution signature"""
        expected_signature = hmac.new(
            self.encryption_key,
            str(trade_data).encode(),
            hashlib.sha256
        ).hexdigest()
        
        return hmac.compare_digest(signature, expected_signature)

class RiskMonitor:
    def __init__(self, risk_limits):
        self.risk_limits = risk_limits
        self.alerts = []
    
    def check_risk_violations(self, portfolio_manager):
        """Check for risk limit violations"""
        violations = []
        
        # Check position size limits
        for trade in portfolio_manager.active_trades:
            if trade['size'] > self.risk_limits['max_position_size']:
                violations.append({
                    'type': 'position_size_exceeded',
                    'trade_id': trade['id'],
                    'current_size': trade['size'],
                    'limit': self.risk_limits['max_position_size']
                })
        
        # Check daily trade limits
        today_trades = portfolio_manager.get_trades_today()
        if len(today_trades) > self.risk_limits['max_daily_trades']:
            violations.append({
                'type': 'daily_trade_limit_exceeded',
                'current_count': len(today_trades),
                'limit': self.risk_limits['max_daily_trades']
            })
        
        # Check drawdown limits
        current_drawdown = portfolio_manager.calculate_drawdown()
        if current_drawdown > self.risk_limits['max_drawdown']:
            violations.append({
                'type': 'max_drawdown_exceeded',
                'current_drawdown': current_drawdown,
                'limit': self.risk_limits['max_drawdown']
            })
        
        return violations
    
    def handle_risk_violations(self, violations):
        """Handle detected risk violations"""
        for violation in violations:
            if violation['type'] == 'max_drawdown_exceeded':
                # Emergency shutdown
                self.emergency_shutdown()
            elif violation['type'] == 'position_size_exceeded':
                # Reduce position size
                self.reduce_position_size(violation['trade_id'])
            
            self.alerts.append({
                'timestamp': datetime.now(),
                'violation': violation,
                'action_taken': 'handled'
            })
    
    def emergency_shutdown(self):
        """Emergency shutdown procedure"""
        print("EMERGENCY SHUTDOWN: Risk limits exceeded")
        # Implementation for emergency shutdown
        pass

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Install Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

# Copy requirements and install Python dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy application code
COPY . .

# Expose port for monitoring dashboard
EXPOSE 8000

# Start the bot
CMD ["python", "main.py"]

import prometheus_client
from prometheus_client import Counter, Histogram, Gauge

class PrometheusMetrics:
    def __init__(self):
        self.trades_total = Counter('arbitrage_trades_total', 'Total number of trades')
        self.trades_successful = Counter('arbitrage_trades_successful', 'Successful trades')
        self.profit_histogram = Histogram('arbitrage_profit_percentage', 'Profit percentage distribution')
        self.portfolio_value = Gauge('portfolio_value_usd', 'Current portfolio value in USD')
        self.active_opportunities = Gauge('active_opportunities', 'Number of active arbitrage opportunities')
    
    def record_trade(self, trade_result):
        """Record trade metrics"""
        self.trades_total.inc()
        
        if trade_result['success']:
            self.trades_successful.inc()
            self.profit_histogram.observe(trade_result['profit_percentage'])
    
    def update_portfolio_value(self, value):
        """Update portfolio value metric"""
        self.portfolio_value.set(value)
    
    def update_opportunities(self, count):
        """Update active opportunities count"""
        self.active_opportunities.set(count)

class APIHealthChecker:
    def __init__(self, exchanges):
        self.exchanges = exchanges
        self.health_status = {}
    
    async def check_all_exchanges(self):
        """Check health of all exchange connections"""
        for exchange in self.exchanges:
            try:
                # Test basic API call
                balance = await exchange.fetch_balance()
                
                self.health_status[exchange.name] = {
                    'status': 'healthy',
                    'last_check': datetime.now(),
                    'response_time': self.measure_response_time(exchange)
                }
            except Exception as e:
                self.health_status[exchange.name] = {
                    'status': 'unhealthy',
                    'last_check': datetime.now(),
                    'error': str(e)
                }
        
        return self.health_status
    
    def measure_response_time(self, exchange):
        """Measure API response time"""
        start_time = time.time()
        try:
            exchange.fetch_ticker('BTC/USD')
            return time.time() - start_time
        except:
            return None

### Error Recovery Mechanisms

```[python](/python-ai-agent-observability/)
class ErrorRecoveryManager:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
        self.retry_delays = [1, 5, 15]  # Exponential backoff
    
    async def retry_with_backoff(self, func, *args, **kwargs):
        """Retry function with exponential backoff"""
        for attempt in range(self.max_retries):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise e
                
                delay = self.retry_delays[min(attempt, len(self.retry_delays) - 1)]
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                await asyncio.sleep(delay)
    
    def handle_exchange_disconnect(self, exchange_name):
        """Handle exchange disconnection"""
        print(f"Exchange {exchange_name} disconnected. Attempting reconnection...")
        # Implementation for reconnection logic
        pass
    
    def handle_insufficient_balance(self, exchange_name, required_balance):
        """Handle insufficient balance scenarios"""
        print(f"Insufficient balance on {exchange_name}. Required: {required_balance}")
        # Implementation for balance rebalancing
        pass

class MemoryManager:
    def __init__(self, max_history_size=1000):
        self.max_history_size = max_history_size
    
    def cleanup_old_data(self, data_dict, max_age_seconds=300):
        """Clean up old data to prevent memory leaks"""
        current_time = time.time()
        keys_to_remove = []
        
        for key, value in data_dict.items():
            if hasattr(value, 'timestamp'):
                age = current_time - value.timestamp.timestamp()
                if age > max_age_seconds:
                    keys_to_remove.append(key)
        
        for key in keys_to_remove:
            del data_dict[key]
    
    def limit_history_size(self, history_list):
        """Limit history size to prevent memory issues"""
        if len(history_list) > self.max_history_size:
            # Keep only the most recent entries
            history_list[:] = history_list[-self.max_history_size:]

class MultiAssetArbitrage:
    def __init__(self):
        self.correlation_matrix = {}
        self.pair_strategies = {}
    
    def find_triangular_arbitrage(self, exchange, base_currency='USD'):
        """Find triangular arbitrage opportunities"""
        symbols = [s for s in exchange.symbols if base_currency in s]
        opportunities = []
        
        for i, symbol1 in enumerate(symbols):
            for j, symbol2 in enumerate(symbols[i+1:], i+1):
                # Check if there's a common currency for triangular trade
                pair1_currencies = symbol1.split('/')
                pair2_currencies = symbol2.split('/')
                
                common_currency = None
                for curr1 in pair1_currencies:
                    if curr1 in pair2_currencies and curr1 != base_currency:
                        common_currency = curr1
                        break
                
                if common_currency:
                    # Calculate triangular arbitrage opportunity
                    opportunity = self.calculate_triangular_profit(
                        exchange, symbol1, symbol2, common_currency, base_currency
                    )
                    
                    if opportunity and opportunity['profit_percentage'] > 0.5:
                        opportunities.append(opportunity)
        
        return opportunities
    
    def calculate_triangular_profit(self, exchange, symbol1, symbol2, common_currency, base_currency):
        """Calculate profit from triangular arbitrage"""
        try:
            # Get current prices
            ticker1 = exchange.fetch_ticker(symbol1)
            ticker2 = exchange.fetch_ticker(symbol2)
            
            # Calculate the triangular path
            # This is a simplified example
            start_amount = 1000  # $1000 starting amount
            
            # Step 1: Buy common currency with base currency
            step1_amount = start_amount / ticker1['ask']
            
            # Step 2: Buy target currency with common currency
            step2_amount = step1_amount / ticker2['ask']
            
            # Step 3: Sell target currency for base currency
            final_amount = step2_amount * ticker2['bid']
            
            profit = final_amount - start_amount
            profit_percentage = (profit / start_amount) * 100
            
            return {
                'type': 'triangular',
                'path': [symbol1, symbol2],
                'start_amount': start_amount,
                'final_amount': final_amount,
                'profit': profit,
                'profit_percentage': profit_percentage
            }
        except Exception as e:
            return None

class StatisticalArbitrage:
    def __init__(self, lookback_period=100):
        self.lookback_period = lookback_period
        self.price_history = {}
        self.correlation_threshold = 0.8
    
    def update_price_history(self, symbol, price):
        """Update price history for statistical analysis"""
        if symbol not in self.price_history:
            self.price_history[symbol] = []
        
        self.price_history[symbol].append({
            'price': price,
            'timestamp': datetime.now()
        })
        
        # Keep only recent data
        if len(self.price_history[symbol]) > self.lookback_period:
            self.price_history[symbol] = self.price_history[symbol][-self.lookback_period:]
    
    def find_mean_reversion_opportunities(self):
        """Find mean reversion opportunities"""
        opportunities = []
        
        for symbol, history in self.price_history.items():
            if len(history) < 20:  # Need sufficient data
                continue
            
            prices = [h['price'] for h in history]
            mean_price = np.mean(prices)
            std_price = np.std(prices)
            current_price = prices[-1]
            
            # Check for mean reversion signals
            z_score = (current_price - mean_price) / std_price
            
            if abs(z_score) > 2:  # More than 2 standard deviations
                opportunities.append({
                    'symbol': symbol,
                    'type': 'mean_reversion',
                    'z_score': z_score,
                    'current_price': current_price,
                    'mean_price': mean_price,
                    'direction': 'sell' if z_score > 0 else 'buy'
                })
        
        return opportunities

class NewsAnalyzer:
    def __init__(self):
        self.ollama_client = ollama.Client()
        self.news_sources = [
            'https://api.coindesk.com/v1/bpi/currentprice.json',
            'https://newsapi.org/v2/everything?q=cryptocurrency'
        ]
    
    def analyze_news_sentiment(self, news_articles):
        """Analyze news sentiment using Ollama"""
        combined_text = ' '.join([article['title'] + ' ' + article.get('description', '') 
                                for article in news_articles])
        
        prompt = f"""
        Analyze the sentiment of these cryptocurrency news articles:
        
        {combined_text}
        
        Provide:
        1. Overall sentiment score (-1 to 1)
        2. Key themes affecting prices
        3. Specific coins mentioned
        4. Market impact prediction
        
        Format as JSON.
        """
        
        response = self.ollama_client.generate(
            model='llama2',
            prompt=prompt,
            options={'temperature': 0.1}
        )
        
        return self.parse_sentiment_response(response['response'])
    
    def parse_sentiment_response(self, response):
        """Parse sentiment analysis response"""
        try:
            # Extract JSON from response
            start_idx = response.find('{')
            end_idx = response.rfind('}') + 1
            json_str = response[start_idx:end_idx]
            return json.loads(json_str)
        except:
            return {'sentiment_score': 0, 'error': 'Failed to parse sentiment'}

class MarketDataAggregator:
    def __init__(self):
        self.data_sources = {
            'coingecko': 'https://api.coingecko.com/api/v3',
            'coinmarketcap': 'https://pro-api.coinmarketcap.com/v1',
            'messari': 'https://data.messari.io/api/v1'
        }
        self.aggregated_data = {}
    
    async def fetch_market_data(self, symbol):
        """Fetch comprehensive market data"""
        market_data = {}
        
        # Fetch from multiple sources
        for source, base_url in self.data_sources.items():
            try:
                data = await self.fetch_from_source(source, base_url, symbol)
                market_data[source] = data
            except Exception as e:
                print(f"Error fetching from {source}: {e}")
        
        return self.aggregate_market_data(market_data)
    
    def aggregate_market_data(self, data):
        """Aggregate data from multiple sources"""
        aggregated = {
            'volume_24h': [],
            'market_cap': [],
            'price_change_24h': [],
            'social_metrics': {}
        }
        
        # Combine data from different sources
        for source, source_data in data.items():
            if 'volume_24h' in source_data:
                aggregated['volume_24h'].append(source_data['volume_24h'])
            if 'market_cap' in source_data:
                aggregated['market_cap'].append(source_data['market_cap'])
        
        # Calculate averages
        for key in ['volume_24h', 'market_cap']:
            if aggregated[key]:
                aggregated[f'avg_{key}'] = sum(aggregated[key]) / len(aggregated[key])
        
        return aggregated

#!/usr/bin/env python3
"""
Production Ollama Arbitrage Bot
"""

import asyncio
import signal
import sys
from datetime import datetime
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('arbitrage_bot.log'),
        logging.StreamHandler()
    ]
)

class ProductionArbitrageBot:
    def __init__(self, config_file='config.json'):
        self.config = self.load_config(config_file)
        self.components = self.initialize_components()
        self.running = False
        self.setup_signal_handlers()
    
    def load_config(self, config_file):
        """Load configuration from file"""
        with open(config_file, 'r') as f:
            return json.load(f)
    
    def initialize_components(self):
        """Initialize all bot components"""
        return {
            'exchanges': self.setup_exchanges(),
            'price_monitor': PriceMonitor(self.exchanges, self.config['symbols']),
            'detector': ArbitrageDetector(self.config['min_profit_threshold']),
            'analyzer': OllamaMarketAnalyzer(),
            'executor': ArbitrageExecutor(self.exchanges, self.portfolio_manager),
            'risk_monitor': RiskMonitor(self.config['risk_limits']),
            'performance_monitor': PerformanceMonitor(self),
            'health_checker': APIHealthChecker(self.exchanges),
            'metrics': PrometheusMetrics()
        }
    
    def setup_signal_handlers(self):
        """Setup graceful shutdown handlers"""
        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)
    
    def signal_handler(self, signum, frame):
        """Handle shutdown signals"""
        print(f"Received signal {signum}. Shutting down gracefully...")
        self.running = False
    
    async def run(self):
        """Main production bot execution"""
        self.running = True
        
        # Start all background tasks
        tasks = [
            asyncio.create_task(self.price_monitoring_loop()),
            asyncio.create_task(self.arbitrage_detection_loop()),
            asyncio.create_task(self.health_monitoring_loop()),
            asyncio.create_task(self.performance_monitoring_loop()),
            asyncio.create_task(self.risk_monitoring_loop())
        ]
        
        try:
            await asyncio.gather(*tasks)
        except Exception as e:
            logging.error(f"Error in main execution: {e}")
        finally:
            await self.cleanup()
    
    async def price_monitoring_loop(self):
        """Price monitoring background task"""
        while self.running:
            try:
                await self.components['price_monitor'].monitor_all_exchanges()
                await asyncio.sleep(1)
            except Exception as e:
                logging.error(f"Price monitoring error: {e}")
                await asyncio.sleep(5)
    
    async def arbitrage_detection_loop(self):
        """Arbitrage detection and execution loop"""
        while self.running:
            try:
                # Find opportunities
                opportunities = self.components['detector'].find_opportunities(
                    self.components['price_monitor'].price_data
                )
                
                # Process each opportunity
                for opportunity in opportunities:
                    await self.process_opportunity(opportunity)
                
                await asyncio.sleep(2)
            except Exception as e:
                logging.error(f"Arbitrage detection error: {e}")
                await asyncio.sleep(5)
    
    async def process_opportunity(self, opportunity):
        """Process a single arbitrage opportunity"""
        try:
            # Risk assessment with Ollama
            risk_assessment = await self.components['analyzer'].risk_assessment(opportunity)
            
            # Check risk limits
            risk_violations = self.components['risk_monitor'].check_risk_violations(
                self.components['portfolio_manager']
            )
            
            if risk_violations:
                logging.warning(f"Risk violations detected: {risk_violations}")
                return
            
            # Execute trade if profitable and safe
            if self.should_execute_trade(opportunity, risk_assessment):
                result = await self.components['executor'].execute_arbitrage(opportunity)
                
                # Record metrics
                self.components['metrics'].record_trade(result)
                
                logging.info(f"Trade executed: {result}")
        
        except Exception as e:
            logging.error(f"Error processing opportunity: {e}")
    
    async def health_monitoring_loop(self):
        """Monitor system health"""
        while self.running:
            try:
                health_status = await self.components['health_checker'].check_all_exchanges()
                
                # Log unhealthy exchanges
                for exchange, status in health_status.items():
                    if status['status'] != 'healthy':
                        logging.warning(f"Exchange {exchange} unhealthy: {status}")
                
                await asyncio.sleep(30)  # Check every 30 seconds
            except Exception as e:
                logging.error(f"Health monitoring error: {e}")
                await asyncio.sleep(60)
    
    async def performance_monitoring_loop(self):
        """Monitor performance metrics"""
        while self.running:
            try:
                metrics = self.components['performance_monitor'].collect_metrics()
                
                # Update Prometheus metrics
                self.components['metrics'].update_portfolio_value(metrics['portfolio_value'])
                self.components['metrics'].update_opportunities(metrics['active_opportunities'])
                
                # Log performance summary
                if len(self.components['performance_monitor'].metrics) % 100 == 0:
                    self.components['performance_monitor'].generate_performance_report()
                
                await asyncio.sleep(60)  # Update every minute
            except Exception as e:
                logging.error(f"Performance monitoring error: {e}")
                await asyncio.sleep(60)
    
    async def risk_monitoring_loop(self):
        """Monitor risk metrics"""
        while self.running:
            try:
                violations = self.components['risk_monitor'].check_risk_violations(
                    self.components['portfolio_manager']
                )
                
                if violations:
                    await self.components['risk_monitor'].handle_risk_violations(violations)
                
                await asyncio.sleep(10)  # Check every 10 seconds
            except Exception as e:
                logging.error(f"Risk monitoring error: {e}")
                await asyncio.sleep(30)
    
    def should_execute_trade(self, opportunity, risk_assessment):
        """Determine if trade should be executed"""
        # Parse risk assessment and make decision
        profit_threshold = self.config['min_profit_threshold']
        return opportunity['profit_data']['profit_percentage'] > profit_threshold
    
    async def cleanup(self):
        """Cleanup resources on shutdown"""
        logging.info("Cleaning up resources...")
        
        # Close exchange connections
        for exchange in self.components['exchanges']:
            await exchange.close()
        
        # Save final state
        await self.save_state()
        
        logging.info("Cleanup completed")
    
    async def save_state(self):
        """Save bot state for recovery"""
        state = {
            'timestamp': datetime.now().isoformat(),
            'portfolio_value': self.components['portfolio_manager'].get_total_value(),
            'active_trades': len(self.components['executor'].active_trades),
            'performance_metrics': self.components['performance_monitor'].metrics[-10:]  # Last 10 entries
        }
        
        with open('bot_state.json', 'w') as f:
            json.dump(state, f, indent=2)

# Production configuration
PRODUCTION_CONFIG = {
    "exchanges": [
        {
            "name": "binance",
            "api_key": "your_binance_api_key",
            "secret": "your_binance_secret",
            "sandbox": False
        },
        {
            "name": "coinbasepro",
            "api_key": "your_coinbase_api_key",
            "secret": "your_coinbase_secret",
            "sandbox": False
        }
    ],
    "symbols": ["BTC/USD", "ETH/USD", "ADA/USD", "DOT/USD"],
    "min_profit_threshold": 0.3,
    "risk_limits": {
        "max_position_size": 500,
        "max_daily_trades": 50,
        "max_drawdown": 0.03
    },
    "initial_balances": {
        "binance": {"USD": 2500, "BTC": 0.05},
        "coinbasepro": {"USD": 2500, "BTC": 0.05}
    }
}

async def main():
    """Main entry point"""
    # Save config to file
    with open('config.json', 'w') as f:
        json.dump(PRODUCTION_CONFIG, f, indent=2)
    
    # Start bot
    bot = ProductionArbitrageBot()
    await bot.run()

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