How to Build Crypto ETF Arbitrage Strategy with Ollama: Premium/Discount Trading

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

# Pull financial analysis model
ollama pull llama2:13b

# Test Ollama installation
ollama run llama2:13b "Analyze market conditions for arbitrage trading"

import requests
import pandas as pd
import numpy as np
from datetime import datetime
import ollama

class ETFArbitrageDetector:
    def __init__(self):
        self.etf_symbols = ['BITO', 'GBTC', 'ARKB', 'FBTC']
        self.crypto_mappings = {
            'BITO': 'BTC-USD',
            'GBTC': 'BTC-USD', 
            'ARKB': 'BTC-USD',
            'FBTC': 'BTC-USD'
        }
        
    def get_etf_price(self, symbol):
        """Fetch current ETF price from market data API"""
        # Replace with your preferred data provider
        url = f"https://api.marketdata.com/v1/stocks/quotes/{symbol}"
        response = requests.get(url)
        return response.json()['last']
    
    def get_crypto_price(self, symbol):
        """Get underlying crypto asset price"""
        url = f"https://api.coinbase.com/v2/exchange-rates?currency={symbol.split('-')[0]}"
        response = requests.get(url)
        return float(response.json()['data']['rates']['USD'])
    
    def calculate_nav_premium_discount(self, etf_symbol):
        """Calculate ETF premium/discount to NAV"""
        etf_price = self.get_etf_price(etf_symbol)
        crypto_symbol = self.crypto_mappings[etf_symbol]
        crypto_price = self.get_crypto_price(crypto_symbol)
        
        # Simplified NAV calculation (real implementation needs share ratios)
        estimated_nav = crypto_price * 0.001  # Approximate conversion ratio
        premium_discount = ((etf_price - estimated_nav) / estimated_nav) * 100
        
        return {
            'etf_symbol': etf_symbol,
            'etf_price': etf_price,
            'estimated_nav': estimated_nav,
            'premium_discount': premium_discount,
            'timestamp': datetime.now()
        }

class OllamaArbitrageAnalyzer:
    def __init__(self):
        self.client = ollama.Client()
        
    def analyze_arbitrage_opportunity(self, arbitrage_data):
        """Use Ollama to assess arbitrage opportunity quality"""
        
        prompt = f"""
        Analyze this crypto ETF arbitrage opportunity:
        
        ETF: {arbitrage_data['etf_symbol']}
        Current Price: ${arbitrage_data['etf_price']:.2f}
        Estimated NAV: ${arbitrage_data['estimated_nav']:.2f}
        Premium/Discount: {arbitrage_data['premium_discount']:.2f}%
        
        Consider these factors:
        1. Is the premium/discount significant enough for profitable arbitrage?
        2. What market conditions might be causing this mispricing?
        3. What are the execution risks?
        4. Recommend position size (conservative/moderate/aggressive)
        
        Provide a structured analysis with clear recommendations.
        """
        
        response = self.client.chat(model='llama2:13b', messages=[
            {'role': 'user', 'content': prompt}
        ])
        
        return response['message']['content']
    
    def generate_trading_signals(self, arbitrage_opportunities):
        """Generate buy/sell signals based on premium/discount analysis"""
        signals = []
        
        for opportunity in arbitrage_opportunities:
            analysis = self.analyze_arbitrage_opportunity(opportunity)
            
            # Extract signal from Ollama analysis (simplified)
            if "aggressive" in analysis.lower() and abs(opportunity['premium_discount']) > 2.0:
                signal = "STRONG_BUY" if opportunity['premium_discount'] < -1.5 else "STRONG_SELL"
            elif abs(opportunity['premium_discount']) > 1.0:
                signal = "BUY" if opportunity['premium_discount'] < -0.8 else "SELL"
            else:
                signal = "HOLD"
                
            signals.append({
                'symbol': opportunity['etf_symbol'],
                'signal': signal,
                'premium_discount': opportunity['premium_discount'],
                'analysis': analysis,
                'timestamp': opportunity['timestamp']
            })
            
        return signals

class ArbitrageRiskManager:
    def __init__(self, max_portfolio_risk=0.02, max_position_size=0.1):
        self.max_portfolio_risk = max_portfolio_risk  # 2% max portfolio risk
        self.max_position_size = max_position_size    # 10% max position size
        
    def calculate_position_size(self, signal_data, portfolio_value, volatility):
        """Calculate optimal position size for arbitrage trade"""
        
        premium_discount = abs(signal_data['premium_discount'])
        signal_strength = signal_data['signal']
        
        # Base position size on arbitrage magnitude and confidence
        if signal_strength in ['STRONG_BUY', 'STRONG_SELL']:
            base_size = min(premium_discount * 0.5, self.max_position_size)
        elif signal_strength in ['BUY', 'SELL']:
            base_size = min(premium_discount * 0.3, self.max_position_size * 0.7)
        else:
            return 0
            
        # Adjust for volatility
        volatility_adjusted_size = base_size * (1 / (1 + volatility))
        
        # Calculate dollar amount
        position_value = portfolio_value * volatility_adjusted_size
        
        return {
            'position_size_pct': volatility_adjusted_size,
            'position_value': position_value,
            'risk_level': 'HIGH' if volatility_adjusted_size > 0.05 else 'MEDIUM'
        }
    
    def set_stop_loss_levels(self, entry_price, premium_discount):
        """Set dynamic stop-loss based on arbitrage convergence expectations"""
        
        # Stop loss when arbitrage opportunity diminishes
        if premium_discount > 0:  # Premium trade (short ETF)
            stop_loss = entry_price * (1 + abs(premium_discount) * 0.005)
        else:  # Discount trade (long ETF)
            stop_loss = entry_price * (1 - abs(premium_discount) * 0.005)
            
        return stop_loss

class ArbitrageExecutionEngine:
    def __init__(self, broker_api_key):
        self.broker_api_key = broker_api_key
        self.active_positions = {}
        
    def execute_arbitrage_trade(self, signal, position_size, risk_manager):
        """Execute arbitrage trade based on signal and position sizing"""
        
        symbol = signal['symbol']
        action = 'BUY' if 'BUY' in signal['signal'] else 'SELL'
        
        # Calculate stop loss
        current_price = self.get_current_price(symbol)
        stop_loss = risk_manager.set_stop_loss_levels(
            current_price, 
            signal['premium_discount']
        )
        
        # Place order (simplified - use your broker's API)
        order = {
            'symbol': symbol,
            'action': action,
            'quantity': position_size['position_value'] // current_price,
            'order_type': 'MARKET',
            'stop_loss': stop_loss,
            'timestamp': datetime.now()
        }
        
        # Execute order through broker API
        order_result = self.place_order(order)
        
        if order_result['status'] == 'FILLED':
            self.active_positions[symbol] = {
                'entry_price': order_result['fill_price'],
                'quantity': order_result['quantity'],
                'action': action,
                'stop_loss': stop_loss,
                'premium_discount_at_entry': signal['premium_discount']
            }
            
        return order_result
    
    def monitor_arbitrage_convergence(self, detector):
        """Monitor open positions for arbitrage convergence"""
        
        for symbol, position in self.active_positions.items():
            current_data = detector.calculate_nav_premium_discount(symbol)
            current_premium_discount = current_data['premium_discount']
            entry_premium_discount = position['premium_discount_at_entry']
            
            # Check if arbitrage has converged (profitability threshold)
            convergence_threshold = abs(entry_premium_discount) * 0.7
            
            if abs(current_premium_discount) < convergence_threshold:
                # Close position - arbitrage opportunity has converged
                self.close_position(symbol, "ARBITRAGE_CONVERGED")
                
    def place_order(self, order):
        """Place order through broker API (implement with your broker)"""
        # Placeholder - implement with your broker's API
        return {
            'status': 'FILLED',
            'fill_price': 50.25,
            'quantity': order['quantity'],
            'order_id': 'ARB_12345'
        }
    
    def close_position(self, symbol, reason):
        """Close arbitrage position"""
        position = self.active_positions.get(symbol)
        if not position:
            return
            
        # Place closing order
        close_action = 'SELL' if position['action'] == 'BUY' else 'BUY'
        close_order = {
            'symbol': symbol,
            'action': close_action,
            'quantity': position['quantity'],
            'order_type': 'MARKET'
        }
        
        result = self.place_order(close_order)
        
        if result['status'] == 'FILLED':
            # Calculate P&L
            if position['action'] == 'BUY':
                pnl = (result['fill_price'] - position['entry_price']) * position['quantity']
            else:
                pnl = (position['entry_price'] - result['fill_price']) * position['quantity']
                
            print(f"Position {symbol} closed. Reason: {reason}. P&L: ${pnl:.2f}")
            del self.active_positions[symbol]

class CryptoETFArbitrageStrategy:
    def __init__(self, broker_api_key, portfolio_value=100000):
        self.detector = ETFArbitrageDetector()
        self.analyzer = OllamaArbitrageAnalyzer()
        self.risk_manager = ArbitrageRiskManager()
        self.execution_engine = ArbitrageExecutionEngine(broker_api_key)
        self.portfolio_value = portfolio_value
        
    def run_arbitrage_cycle(self):
        """Execute one complete arbitrage detection and trading cycle"""
        
        print("🔍 Scanning for ETF arbitrage opportunities...")
        
        # 1. Detect premium/discount opportunities
        opportunities = []
        for symbol in self.detector.etf_symbols:
            try:
                arb_data = self.detector.calculate_nav_premium_discount(symbol)
                opportunities.append(arb_data)
                print(f"{symbol}: {arb_data['premium_discount']:.2f}% premium/discount")
            except Exception as e:
                print(f"Error analyzing {symbol}: {e}")
                
        # 2. Generate trading signals with Ollama analysis
        signals = self.analyzer.generate_trading_signals(opportunities)
        
        # 3. Execute trades for strong signals
        for signal in signals:
            if signal['signal'] in ['STRONG_BUY', 'STRONG_SELL', 'BUY', 'SELL']:
                
                # Calculate position size
                volatility = self.estimate_volatility(signal['symbol'])
                position_size = self.risk_manager.calculate_position_size(
                    signal, self.portfolio_value, volatility
                )
                
                if position_size['position_value'] > 1000:  # Minimum trade size
                    print(f"🎯 Executing {signal['signal']} for {signal['symbol']}")
                    print(f"   Premium/Discount: {signal['premium_discount']:.2f}%")
                    print(f"   Position Size: ${position_size['position_value']:.2f}")
                    
                    # Execute trade
                    self.execution_engine.execute_arbitrage_trade(
                        signal, position_size, self.risk_manager
                    )
                    
        # 4. Monitor existing positions for convergence
        self.execution_engine.monitor_arbitrage_convergence(self.detector)
        
    def estimate_volatility(self, symbol):
        """Estimate symbol volatility for risk adjustment"""
        # Simplified volatility estimation
        # In production, use historical price data
        volatility_map = {
            'BITO': 0.6,
            'GBTC': 0.8,
            'ARKB': 0.7,
            'FBTC': 0.6
        }
        return volatility_map.get(symbol, 0.5)
    
    def run_continuous_strategy(self, interval_minutes=15):
        """Run arbitrage strategy continuously"""
        import time
        
        print("🚀 Starting Crypto ETF Arbitrage Strategy with Ollama")
        print(f"📊 Portfolio Value: ${self.portfolio_value:,.2f}")
        print(f"⏰ Scan Interval: {interval_minutes} minutes")
        
        while True:
            try:
                self.run_arbitrage_cycle()
                print(f"💤 Waiting {interval_minutes} minutes until next scan...\n")
                time.sleep(interval_minutes * 60)
                
            except KeyboardInterrupt:
                print("🛑 Strategy stopped by user")
                break
            except Exception as e:
                print(f"❌ Error in strategy cycle: {e}")
                time.sleep(60)  # Wait 1 minute on error

# Example usage
if __name__ == "__main__":
    # Initialize strategy
    strategy = CryptoETFArbitrageStrategy(
        broker_api_key="your_broker_api_key",
        portfolio_value=50000
    )
    
    # Run single cycle for testing
    strategy.run_arbitrage_cycle()
    
    # Run continuously (uncomment for live trading)
    # strategy.run_continuous_strategy(interval_minutes=10)

class MultiAssetArbitrageDetector(ETFArbitrageDetector):
    def __init__(self):
        super().__init__()
        self.correlation_pairs = [
            ('BITO', 'GBTC'),  # Both Bitcoin exposure
            ('ARKB', 'FBTC'),  # Alternative Bitcoin ETFs
        ]
    
    def detect_cross_etf_arbitrage(self):
        """Detect arbitrage between related ETFs"""
        arbitrage_opportunities = []
        
        for etf1, etf2 in self.correlation_pairs:
            data1 = self.calculate_nav_premium_discount(etf1)
            data2 = self.calculate_nav_premium_discount(etf2)
            
            # Look for significant premium/discount divergence
            spread = data1['premium_discount'] - data2['premium_discount']
            
            if abs(spread) > 1.5:  # 1.5% spread threshold
                arbitrage_opportunities.append({
                    'type': 'CROSS_ETF_ARBITRAGE',
                    'long_etf': etf1 if spread < 0 else etf2,
                    'short_etf': etf2 if spread < 0 else etf1,
                    'spread': abs(spread),
                    'confidence': 'HIGH' if abs(spread) > 2.5 else 'MEDIUM'
                })
                
        return arbitrage_opportunities

class ArbitragePerformanceTracker:
    def __init__(self):
        self.trades = []
        self.daily_pnl = {}
        
    def record_trade(self, trade_data):
        """Record completed arbitrage trade for analysis"""
        self.trades.append({
            'timestamp': trade_data['timestamp'],
            'symbol': trade_data['symbol'],
            'entry_premium_discount': trade_data['entry_premium_discount'],
            'exit_premium_discount': trade_data['exit_premium_discount'],
            'pnl': trade_data['pnl'],
            'holding_period': trade_data['holding_period'],
            'trade_type': trade_data['trade_type']
        })
        
    def calculate_strategy_metrics(self):
        """Calculate key performance metrics"""
        if not self.trades:
            return {}
            
        total_pnl = sum(trade['pnl'] for trade in self.trades)
        winning_trades = [t for t in self.trades if t['pnl'] > 0]
        losing_trades = [t for t in self.trades if t['pnl'] < 0]
        
        metrics = {
            'total_trades': len(self.trades),
            'total_pnl': total_pnl,
            'win_rate': len(winning_trades) / len(self.trades) * 100,
            'avg_win': np.mean([t['pnl'] for t in winning_trades]) if winning_trades else 0,
            'avg_loss': np.mean([t['pnl'] for t in losing_trades]) if losing_trades else 0,
            'avg_holding_period': np.mean([t['holding_period'] for t in self.trades]),
            'sharpe_ratio': self.calculate_sharpe_ratio()
        }
        
        return metrics
    
    def calculate_sharpe_ratio(self):
        """Calculate Sharpe ratio for strategy performance"""
        daily_returns = list(self.daily_pnl.values())
        if len(daily_returns) < 2:
            return 0
            
        avg_return = np.mean(daily_returns)
        std_return = np.std(daily_returns)
        
        return (avg_return / std_return) * np.sqrt(252) if std_return > 0 else 0

class DynamicRiskManager(ArbitrageRiskManager):
    def __init__(self):
        super().__init__()
        self.market_volatility_threshold = 0.3
        
    def adjust_risk_for_market_conditions(self, current_vix, crypto_volatility):
        """Adjust risk parameters based on market conditions"""
        
        if current_vix > 30 or crypto_volatility > self.market_volatility_threshold:
            # High volatility - reduce position sizes
            self.max_position_size *= 0.6
            self.max_portfolio_risk *= 0.7
            return "DEFENSIVE"
        elif current_vix < 15 and crypto_volatility < 0.15:
            # Low volatility - can increase position sizes
            self.max_position_size *= 1.2
            self.max_portfolio_risk *= 1.1
            return "AGGRESSIVE"
        else:
            return "NORMAL"

class ArbitrageSystemMonitor:
    def __init__(self, strategy):
        self.strategy = strategy
        self.alerts = []
        
    def health_check(self):
        """Perform system health checks"""
        checks = {
            'ollama_connection': self.check_ollama_status(),
            'broker_connection': self.check_broker_status(),
            'data_feeds': self.check_data_feed_status(),
            'active_positions': len(self.strategy.execution_engine.active_positions),
            'system_memory': self.get_system_memory_usage()
        }
        
        return checks
    
    def check_ollama_status(self):
        """Verify Ollama is responding"""
        try:
            response = ollama.list()
            return "HEALTHY" if response else "ERROR"
        except:
            return "OFFLINE"
    
    def send_alert_if_needed(self, checks):
        """Send alerts for system issues"""
        if checks['ollama_connection'] != "HEALTHY":
            self.send_alert("Ollama connection issues detected")
        
        if checks['active_positions'] > 10:
            self.send_alert(f"High position count: {checks['active_positions']}")