Grid Trading Strategy with Ollama: Automated Buy-Low-Sell-High Implementation

# Install core packages
pip install ollama-python ccxt pandas numpy requests

# Install additional trading libraries
pip install python-binance ta-lib plotly

# Verify Ollama installation
ollama serve

import ollama
import json
from datetime import datetime

class OllamaAnalyzer:
    def __init__(self, model_name="llama3.1"):
        self.model = model_name
        self.client = ollama.Client()
    
    def analyze_market_conditions(self, price_data, volume_data):
        """Analyze market conditions using Ollama AI"""
        prompt = f"""
        Analyze this market data for grid trading:
        Current Price: ${price_data['current']}
        24h High: ${price_data['high']}
        24h Low: ${price_data['low']}
        Volume: {volume_data['24h']}
        
        Recommend:
        1. Grid spacing percentage
        2. Number of grid levels
        3. Risk assessment (1-10)
        4. Market volatility rating
        
        Respond in JSON format.
        """
        
        response = self.client.chat(model=self.model, messages=[
            {'role': 'user', 'content': prompt}
        ])
        
        return json.loads(response['message']['content'])

import ccxt
import pandas as pd
import numpy as np
from typing import List, Dict, Tuple

class GridTradingBot:
    def __init__(self, exchange_config: Dict, ollama_analyzer: OllamaAnalyzer):
        self.exchange = ccxt.binance(exchange_config)
        self.analyzer = ollama_analyzer
        self.active_grids = {}
        self.profit_history = []
    
    def calculate_grid_levels(self, current_price: float, 
                            grid_spacing: float, 
                            num_levels: int) -> Tuple[List[float], List[float]]:
        """Calculate buy and sell grid levels"""
        buy_levels = []
        sell_levels = []
        
        # Generate buy levels below current price
        for i in range(1, num_levels + 1):
            buy_price = current_price * (1 - (grid_spacing * i))
            buy_levels.append(round(buy_price, 2))
        
        # Generate sell levels above current price
        for i in range(1, num_levels + 1):
            sell_price = current_price * (1 + (grid_spacing * i))
            sell_levels.append(round(sell_price, 2))
        
        return buy_levels, sell_levels
    
    def place_grid_orders(self, symbol: str, quantity: float):
        """Place initial grid orders"""
        try:
            # Get current market data
            ticker = self.exchange.fetch_ticker(symbol)
            current_price = ticker['last']
            
            # Get AI analysis
            market_data = {
                'current': current_price,
                'high': ticker['high'],
                'low': ticker['low']
            }
            volume_data = {'24h': ticker['quoteVolume']}
            
            ai_analysis = self.analyzer.analyze_market_conditions(
                market_data, volume_data
            )
            
            # Calculate grid levels based on AI recommendation
            grid_spacing = ai_analysis['grid_spacing_percentage'] / 100
            num_levels = ai_analysis['num_grid_levels']
            
            buy_levels, sell_levels = self.calculate_grid_levels(
                current_price, grid_spacing, num_levels
            )
            
            # Place buy orders
            for price in buy_levels:
                order = self.exchange.create_limit_buy_order(
                    symbol, quantity, price
                )
                print(f"Buy order placed: {quantity} at ${price}")
            
            # Place sell orders
            for price in sell_levels:
                order = self.exchange.create_limit_sell_order(
                    symbol, quantity, price
                )
                print(f"Sell order placed: {quantity} at ${price}")
                
            return True
            
        except Exception as e:
            print(f"Error placing grid orders: {e}")
            return False

class DynamicGridManager:
    def __init__(self, bot: GridTradingBot):
        self.bot = bot
        self.last_adjustment = datetime.now()
        self.adjustment_interval = 3600  # 1 hour in seconds
    
    def monitor_and_adjust(self, symbol: str):
        """Monitor market conditions and adjust grid dynamically"""
        current_time = datetime.now()
        
        # Check if adjustment interval has passed
        if (current_time - self.last_adjustment).seconds < self.adjustment_interval:
            return
        
        try:
            # Get current market conditions
            ticker = self.bot.exchange.fetch_ticker(symbol)
            
            # Analyze with Ollama
            market_analysis = self.bot.analyzer.analyze_market_conditions(
                {
                    'current': ticker['last'],
                    'high': ticker['high'],
                    'low': ticker['low']
                },
                {'24h': ticker['quoteVolume']}
            )
            
            # Check if grid adjustment is needed
            volatility_rating = market_analysis['market_volatility_rating']
            
            if volatility_rating > 7:  # High volatility
                self.expand_grid_spacing(symbol, 1.5)
            elif volatility_rating < 3:  # Low volatility
                self.tighten_grid_spacing(symbol, 0.7)
            
            self.last_adjustment = current_time
            
        except Exception as e:
            print(f"Error in dynamic adjustment: {e}")
    
    def expand_grid_spacing(self, symbol: str, multiplier: float):
        """Expand grid spacing for high volatility"""
        print(f"Expanding grid spacing by {multiplier}x for {symbol}")
        # Cancel existing orders and recreate with wider spacing
        self.recreate_grid_with_adjustment(symbol, multiplier)
    
    def tighten_grid_spacing(self, symbol: str, multiplier: float):
        """Tighten grid spacing for low volatility"""
        print(f"Tightening grid spacing by {multiplier}x for {symbol}")
        # Cancel existing orders and recreate with tighter spacing
        self.recreate_grid_with_adjustment(symbol, multiplier)

class RiskManager:
    def __init__(self, max_drawdown: float = 0.1, 
                 max_position_size: float = 0.05):
        self.max_drawdown = max_drawdown
        self.max_position_size = max_position_size
        self.initial_balance = None
        self.current_balance = None
    
    def check_risk_limits(self, current_balance: float, 
                         open_positions: Dict) -> bool:
        """Check if current positions exceed risk limits"""
        if self.initial_balance is None:
            self.initial_balance = current_balance
        
        self.current_balance = current_balance
        
        # Calculate current drawdown
        drawdown = (self.initial_balance - current_balance) / self.initial_balance
        
        if drawdown > self.max_drawdown:
            print(f"WARNING: Drawdown {drawdown:.2%} exceeds limit {self.max_drawdown:.2%}")
            return False
        
        # Check position size limits
        total_position_value = sum(pos['value'] for pos in open_positions.values())
        position_ratio = total_position_value / current_balance
        
        if position_ratio > self.max_position_size:
            print(f"WARNING: Position size {position_ratio:.2%} exceeds limit")
            return False
        
        return True
    
    def calculate_optimal_quantity(self, price: float, 
                                 balance: float, 
                                 risk_per_trade: float = 0.01) -> float:
        """Calculate optimal quantity based on risk management"""
        max_trade_value = balance * risk_per_trade
        quantity = max_trade_value / price
        return round(quantity, 8)  # Round to 8 decimal places for crypto

def create_arithmetic_grid(base_price: float, spacing: float, levels: int):
    """Create grid with equal dollar spacing"""
    buy_levels = [base_price - (spacing * i) for i in range(1, levels + 1)]
    sell_levels = [base_price + (spacing * i) for i in range(1, levels + 1)]
    return buy_levels, sell_levels

def create_geometric_grid(base_price: float, spacing_pct: float, levels: int):
    """Create grid with equal percentage spacing"""
    buy_levels = [base_price * (1 - spacing_pct) ** i for i in range(1, levels + 1)]
    sell_levels = [base_price * (1 + spacing_pct) ** i for i in range(1, levels + 1)]
    return buy_levels, sell_levels

def adapt_strategy_to_market(market_trend: str, volatility: float):
    """Adapt grid strategy based on market conditions"""
    strategies = {
        'bullish': {
            'bias': 'buy_heavy',  # More buy orders than sell
            'spacing_multiplier': 1.2,
            'levels': 8
        },
        'bearish': {
            'bias': 'sell_heavy',  # More sell orders than buy
            'spacing_multiplier': 1.5,
            'levels': 6
        },
        'sideways': {
            'bias': 'balanced',
            'spacing_multiplier': 1.0,
            'levels': 10
        }
    }
    
    if volatility > 0.05:  # High volatility
        strategies[market_trend]['spacing_multiplier'] *= 1.5
    
    return strategies[market_trend]

class PerformanceTracker:
    def __init__(self):
        self.trades = []
        self.daily_pnl = {}
        self.metrics = {}
    
    def record_trade(self, trade_data: Dict):
        """Record completed trade"""
        self.trades.append({
            'timestamp': datetime.now(),
            'symbol': trade_data['symbol'],
            'side': trade_data['side'],
            'quantity': trade_data['quantity'],
            'price': trade_data['price'],
            'profit': trade_data.get('profit', 0)
        })
    
    def calculate_metrics(self) -> Dict:
        """Calculate performance metrics"""
        if not self.trades:
            return {}
        
        profits = [trade['profit'] for trade in self.trades if trade['profit'] != 0]
        
        metrics = {
            'total_trades': len(self.trades),
            'profitable_trades': len([p for p in profits if p > 0]),
            'total_profit': sum(profits),
            'average_profit_per_trade': np.mean(profits) if profits else 0,
            'win_rate': len([p for p in profits if p > 0]) / len(profits) if profits else 0,
            'sharpe_ratio': self.calculate_sharpe_ratio(profits),
            'max_drawdown': self.calculate_max_drawdown()
        }
        
        return metrics
    
    def generate_performance_report(self) -> str:
        """Generate detailed performance report"""
        metrics = self.calculate_metrics()
        
        report = f"""
        📊 Grid Trading Performance Report
        ================================
        
        Total Trades: {metrics.get('total_trades', 0)}
        Win Rate: {metrics.get('win_rate', 0):.2%}
        Total Profit: ${metrics.get('total_profit', 0):.2f}
        Average Profit/Trade: ${metrics.get('average_profit_per_trade', 0):.2f}
        Sharpe Ratio: {metrics.get('sharpe_ratio', 0):.2f}
        Max Drawdown: {metrics.get('max_drawdown', 0):.2%}
        """
        
        return report

# production_config.py
PRODUCTION_CONFIG = {
    'exchange': {
        'apiKey': 'your_api_key',
        'secret': 'your_secret_key',
        'sandbox': False,  # Set to True for testing
        'enableRateLimit': True
    },
    'trading': {
        'default_quantity': 0.001,  # BTC
        'max_active_grids': 5,
        'rebalance_interval': 3600,  # 1 hour
    },
    'risk': {
        'max_drawdown': 0.15,
        'max_position_size': 0.1,
        'stop_loss_threshold': 0.2
    },
    'logging': {
        'level': 'INFO',
        'file': 'grid_trading.log',
        'rotation': 'daily'
    }
}

import smtplib
from email.mime.text import MIMEText

class AlertSystem:
    def __init__(self, smtp_config: Dict):
        self.smtp_config = smtp_config
    
    def send_alert(self, subject: str, message: str, urgency: str = 'medium'):
        """Send email alert for important events"""
        try:
            msg = MIMEText(message)
            msg['Subject'] = f"[{urgency.upper()}] {subject}"
            msg['From'] = self.smtp_config['from_email']
            msg['To'] = self.smtp_config['to_email']
            
            with smtplib.SMTP(self.smtp_config['server'], 
                            self.smtp_config['port']) as server:
                server.starttls()
                server.login(self.smtp_config['username'], 
                           self.smtp_config['password'])
                server.send_message(msg)
                
        except Exception as e:
            print(f"Failed to send alert: {e}")
    
    def check_system_health(self, bot: GridTradingBot):
        """Monitor system health and send alerts"""
        # Check API connectivity
        try:
            bot.exchange.fetch_balance()
        except Exception as e:
            self.send_alert("API Connection Failed", 
                          f"Exchange API error: {e}", "high")
        
        # Check for unusual losses
        recent_trades = bot.performance_tracker.trades[-10:]
        if len(recent_trades) >= 5:
            recent_profits = [t['profit'] for t in recent_trades]
            if sum(recent_profits) < -100:  # $100 loss threshold
                self.send_alert("Unusual Losses Detected", 
                              f"Recent losses: ${sum(recent_profits):.2f}", "high")

def detect_trend(price_data: List[float], period: int = 20) -> str:
    """Detect market trend using moving averages"""
    if len(price_data) < period:
        return 'neutral'
    
    ma_short = np.mean(price_data[-10:])
    ma_long = np.mean(price_data[-period:])
    
    if ma_short > ma_long * 1.02:
        return 'bullish'
    elif ma_short < ma_long * 0.98:
        return 'bearish'
    else:
        return 'sideways'

def calculate_net_profit(buy_price: float, sell_price: float, 
                        quantity: float, fee_rate: float = 0.001) -> float:
    """Calculate profit after fees"""
    gross_profit = (sell_price - buy_price) * quantity
    total_fees = (buy_price + sell_price) * quantity * fee_rate
    return gross_profit - total_fees

class AdvancedOllamaAnalysis:
    def __init__(self):
        self.client = ollama.Client()
    
    def technical_analysis(self, price_data: pd.DataFrame) -> Dict:
        """Perform technical analysis using Ollama"""
        # Prepare technical indicators
        rsi = self.calculate_rsi(price_data['close'])
        macd = self.calculate_macd(price_data['close'])
        bb = self.calculate_bollinger_bands(price_data['close'])
        
        analysis_prompt = f"""
        Analyze these technical indicators for grid trading:
        
        RSI: {rsi:.2f}
        MACD: {macd['signal']:.4f}
        Bollinger Bands: Upper {bb['upper']:.2f}, Lower {bb['lower']:.2f}
        Current Price: {price_data['close'].iloc[-1]:.2f}
        
        Provide:
        1. Market condition assessment
        2. Optimal grid spacing recommendation
        3. Risk level (1-10)
        4. Entry/exit signals
        
        Format response as JSON.
        """
        
        response = self.client.chat(
            model='llama3.1',
            messages=[{'role': 'user', 'content': analysis_prompt}]
        )
        
        return json.loads(response['message']['content'])

def analyze_market_sentiment(news_data: List[str]) -> float:
    """Analyze market sentiment using Ollama"""
    combined_news = " ".join(news_data[:5])  # Use latest 5 news items
    
    sentiment_prompt = f"""
    Analyze the market sentiment from this news:
    {combined_news}
    
    Rate sentiment from -1 (very bearish) to +1 (very bullish).
    Consider impact on cryptocurrency/trading markets.
    Respond with only a decimal number.
    """
    
    response = ollama.Client().chat(
        model='llama3.1',
        messages=[{'role': 'user', 'content': sentiment_prompt}]
    )
    
    return float(response['message']['content'].strip())

class MultiAssetGridBot:
    def __init__(self):
        self.asset_bots = {}
        self.correlation_matrix = {}
    
    def add_trading_pair(self, symbol: str, config: Dict):
        """Add new trading pair to the grid system"""
        self.asset_bots[symbol] = GridTradingBot(config['exchange'], 
                                               config['analyzer'])
        
    def calculate_portfolio_risk(self) -> float:
        """Calculate overall portfolio risk"""
        # Implement correlation-based risk calculation
        total_exposure = sum(bot.get_total_exposure() 
                           for bot in self.asset_bots.values())
        
        # Apply correlation adjustments
        adjusted_risk = total_exposure * self.get_correlation_adjustment()
        
        return adjusted_risk
    
    def rebalance_portfolio(self):
        """Rebalance grid allocations based on performance"""
        performances = {symbol: bot.get_performance_metrics() 
                       for symbol, bot in self.asset_bots.items()}
        
        # Allocate more capital to better-performing pairs
        self.redistribute_capital_based_on_performance(performances)

# docker-compose.yml for production deployment
version: '3.8'
services:
  trading-engine:
    build: .
    environment:
      - ENV=production
      - REDIS_URL=redis://redis:6379
      - DATABASE_URL=postgresql://user:pass@db:5432/trading
    depends_on:
      - redis
      - db
    restart: always
  
  redis:
    image: redis:alpine
    restart: always
  
  db:
    image: postgres:13
    environment:
      POSTGRES_DB: trading
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: always
  
  monitoring:
    image: grafana/grafana
    ports:
      - "3000:3000"
    restart: always

volumes:
  postgres_data: