How to Connect Binance API to Ollama: Cryptocurrency Trading Automation Guide

# For macOS
brew install ollama

# For Linux
curl -fsSL https://ollama.ai/install.sh | sh

# For Windows (using PowerShell)
Invoke-WebRequest -Uri https://ollama.ai/install.ps1 -OutFile install.ps1
.\install.ps1

# Install Llama 3.1 for general analysis
ollama pull llama3.1

# Install CodeLlama for Python code generation
ollama pull codellama

# Install Mistral for financial analysis
ollama pull mistral

ollama run llama3.1 "Analyze Bitcoin price trends"

BINANCE_API_KEY=your_api_key_here
BINANCE_SECRET_KEY=your_secret_key_here
OLLAMA_HOST=http://localhost:11434

pip install python-binance requests python-dotenv pandas numpy

python-binance==1.0.19
requests==2.31.0
python-dotenv==1.0.0
pandas==2.0.3
numpy==1.24.3

import os
import json
import requests
import pandas as pd
from binance.client import Client
from binance.exceptions import BinanceAPIException
from dotenv import load_dotenv
from datetime import datetime, timedelta

class BinanceOllamaTrader:
    def __init__(self):
        load_dotenv()
        
        # Initialize Binance client
        self.binance_client = Client(
            api_key=os.getenv('BINANCE_API_KEY'),
            api_secret=os.getenv('BINANCE_SECRET_KEY')
        )
        
        # Ollama configuration
        self.ollama_host = os.getenv('OLLAMA_HOST', 'http://localhost:11434')
        self.model_name = 'llama3.1'
        
    def get_market_data(self, symbol, interval='1h', limit=100):
        """Fetch historical price data from Binance"""
        try:
            klines = self.binance_client.get_historical_klines(
                symbol, interval, f"{limit} hours ago UTC"
            )
            
            # Convert to DataFrame for easier analysis
            df = pd.DataFrame(klines, columns=[
                'timestamp', 'open', 'high', 'low', 'close', 'volume',
                'close_time', 'quote_volume', 'trades', 'taker_buy_base',
                'taker_buy_quote', 'ignored'
            ])
            
            # Convert timestamp to readable format
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            df['close'] = df['close'].astype(float)
            
            return df[['timestamp', 'close', 'volume']]
            
        except BinanceAPIException as e:
            print(f"Binance API Error: {e}")
            return None

def query_ollama(self, prompt, model=None):
    """Send analysis request to Ollama"""
    if not model:
        model = self.model_name
        
    payload = {
        "model": model,
        "prompt": prompt,
        "stream": False
    }
    
    try:
        response = requests.post(
            f"{self.ollama_host}/api/generate",
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            return response.json()['response']
        else:
            print(f"Ollama Error: {response.status_code}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"Connection Error: {e}")
        return None

def analyze_market_sentiment(self, symbol, market_data):
    """Use Ollama to analyze market trends"""
    
    # Prepare market data summary
    latest_price = market_data['close'].iloc[-1]
    price_change = ((latest_price - market_data['close'].iloc[0]) / 
                   market_data['close'].iloc[0]) * 100
    
    prompt = f"""
    Analyze the cryptocurrency market data for {symbol}:
    
    Current Price: ${latest_price:.2f}
    Price Change: {price_change:.2f}%
    Data Points: {len(market_data)} hours
    
    Recent price movements:
    {market_data.tail(10).to_string()}
    
    Provide a trading recommendation with:
    1. Market sentiment (bullish/bearish/neutral)
    2. Key support/resistance levels
    3. Risk assessment
    4. Suggested action (buy/sell/hold)
    
    Keep response concise and actionable.
    """
    
    return self.query_ollama(prompt)

def trend_following_strategy(self, symbol, investment_amount=100):
    """Execute trend-based trading using AI analysis"""
    
    # Get market data
    market_data = self.get_market_data(symbol)
    if market_data is None:
        return {"error": "Failed to fetch market data"}
    
    # Get AI analysis
    analysis = self.analyze_market_sentiment(symbol, market_data)
    if not analysis:
        return {"error": "AI analysis failed"}
    
    # Parse AI recommendation
    recommendation = self.parse_trading_signal(analysis)
    
    # Execute trade based on AI recommendation
    trade_result = self.execute_trade(
        symbol, 
        recommendation['action'], 
        investment_amount
    )
    
    return {
        "symbol": symbol,
        "analysis": analysis,
        "recommendation": recommendation,
        "trade_result": trade_result,
        "timestamp": datetime.now().isoformat()
    }

def parse_trading_signal(self, analysis):
    """Extract trading signals from AI analysis"""
    analysis_lower = analysis.lower()
    
    # Determine action based on AI response
    if any(word in analysis_lower for word in ['buy', 'bullish', 'uptrend']):
        action = 'BUY'
        confidence = 'high' if 'strong' in analysis_lower else 'medium'
    elif any(word in analysis_lower for word in ['sell', 'bearish', 'downtrend']):
        action = 'SELL'
        confidence = 'high' if 'strong' in analysis_lower else 'medium'
    else:
        action = 'HOLD'
        confidence = 'low'
    
    return {
        "action": action,
        "confidence": confidence,
        "reasoning": analysis[:200] + "..." if len(analysis) > 200 else analysis
    }

def execute_trade(self, symbol, action, amount):
    """Execute trade with risk management"""
    
    try:
        # Get current account balance
        account_info = self.binance_client.get_account()
        
        if action == 'BUY':
            # Calculate quantity based on current price
            ticker = self.binance_client.get_symbol_ticker(symbol=symbol)
            current_price = float(ticker['price'])
            quantity = amount / current_price
            
            # Place buy order
            order = self.binance_client.order_market_buy(
                symbol=symbol,
                quoteOrderQty=amount
            )
            
            return {
                "status": "success",
                "action": "BUY",
                "quantity": quantity,
                "price": current_price,
                "order_id": order['orderId']
            }
            
        elif action == 'SELL':
            # Get current holdings
            balance = self.get_asset_balance(symbol.replace('USDT', ''))
            
            if balance and float(balance['free']) > 0:
                # Place sell order
                order = self.binance_client.order_market_sell(
                    symbol=symbol,
                    quantity=balance['free']
                )
                
                return {
                    "status": "success",
                    "action": "SELL",
                    "quantity": balance['free'],
                    "order_id": order['orderId']
                }
            else:
                return {"status": "error", "message": "No assets to sell"}
                
        else:  # HOLD
            return {"status": "hold", "message": "No action taken"}
            
    except BinanceAPIException as e:
        return {"status": "error", "message": str(e)}

def get_asset_balance(self, asset):
    """Get balance for specific cryptocurrency"""
    try:
        account_info = self.binance_client.get_account()
        for balance in account_info['balances']:
            if balance['asset'] == asset:
                return balance
        return None
    except BinanceAPIException as e:
        print(f"Error fetching balance: {e}")
        return None

def manage_portfolio(self, symbols, total_investment):
    """Manage multiple cryptocurrency positions"""
    
    portfolio_analysis = {}
    allocation_per_asset = total_investment / len(symbols)
    
    for symbol in symbols:
        # Analyze each asset
        market_data = self.get_market_data(symbol)
        analysis = self.analyze_market_sentiment(symbol, market_data)
        
        portfolio_analysis[symbol] = {
            "analysis": analysis,
            "allocation": allocation_per_asset,
            "recommendation": self.parse_trading_signal(analysis)
        }
    
    # Execute trades based on analysis
    results = []
    for symbol, data in portfolio_analysis.items():
        if data['recommendation']['confidence'] == 'high':
            trade_result = self.execute_trade(
                symbol, 
                data['recommendation']['action'], 
                data['allocation']
            )
            results.append({
                "symbol": symbol,
                "trade": trade_result,
                "analysis": data['analysis'][:100] + "..."
            })
    
    return results

def set_stop_loss(self, symbol, stop_price, quantity):
    """Set stop-loss order for risk management"""
    try:
        order = self.binance_client.create_order(
            symbol=symbol,
            side='SELL',
            type='STOP_LOSS_LIMIT',
            timeInForce='GTC',
            quantity=quantity,
            price=stop_price * 0.99,  # Slightly below stop price
            stopPrice=stop_price
        )
        
        return {"status": "success", "order_id": order['orderId']}
        
    except BinanceAPIException as e:
        return {"status": "error", "message": str(e)}

def monitor_positions(self):
    """Monitor active positions and adjust stop-losses"""
    try:
        # Get all open orders
        open_orders = self.binance_client.get_open_orders()
        
        for order in open_orders:
            symbol = order['symbol']
            current_price = float(
                self.binance_client.get_symbol_ticker(symbol=symbol)['price']
            )
            
            # Update trailing stop-loss if price moved favorably
            if order['type'] == 'STOP_LOSS_LIMIT':
                self.update_trailing_stop(order, current_price)
                
    except BinanceAPIException as e:
        print(f"Error monitoring positions: {e}")

def run_trading_bot(self):
    """Main trading loop with error handling"""
    
    # Define trading pairs
    trading_pairs = ['BTCUSDT', 'ETHUSDT', 'ADAUSDT']
    
    print("Starting Binance-Ollama Trading Bot...")
    print(f"Monitoring pairs: {', '.join(trading_pairs)}")
    
    while True:
        try:
            for symbol in trading_pairs:
                print(f"\n--- Analyzing {symbol} ---")
                
                # Run strategy
                result = self.trend_following_strategy(symbol, 50)
                
                # Log results
                if 'error' not in result:
                    print(f"Action: {result['recommendation']['action']}")
                    print(f"Confidence: {result['recommendation']['confidence']}")
                    
                    # Execute trade if confidence is high
                    if result['recommendation']['confidence'] == 'high':
                        trade_result = result['trade_result']
                        if trade_result['status'] == 'success':
                            print(f"✅ Trade executed: {trade_result['action']}")
                        else:
                            print(f"❌ Trade failed: {trade_result['message']}")
                else:
                    print(f"❌ Analysis failed: {result['error']}")
                
                time.sleep(10)  # Wait between analyses
                
            # Wait before next trading cycle
            time.sleep(300)  # 5 minutes between cycles
            
        except KeyboardInterrupt:
            print("\n🛑 Trading bot stopped by user")
            break
        except Exception as e:
            print(f"❌ Unexpected error: {e}")
            time.sleep(60)  # Wait 1 minute before retrying

# Initialize and run the bot
if __name__ == "__main__":
    trader = BinanceOllamaTrader()
    trader.run_trading_bot()

def test_connections(self):
    """Test both Binance and Ollama connections"""
    
    print("Testing Binance API connection...")
    try:
        account_info = self.binance_client.get_account()
        print("✅ Binance API connected successfully")
        print(f"Account type: {account_info['accountType']}")
    except Exception as e:
        print(f"❌ Binance API connection failed: {e}")
        return False
    
    print("\nTesting Ollama connection...")
    try:
        response = self.query_ollama("Test connection")
        if response:
            print("✅ Ollama connected successfully")
            print(f"Response: {response[:100]}...")
        else:
            print("❌ Ollama connection failed")
            return False
    except Exception as e:
        print(f"❌ Ollama connection error: {e}")
        return False
    
    return True

# Run connection test
trader = BinanceOllamaTrader()
if trader.test_connections():
    print("\n🎉 All systems ready for trading!")
else:
    print("\n🚨 Fix connection issues before trading")

def backtest_strategy(self, symbol, days=30):
    """Test strategy performance on historical data"""
    
    # Get historical data
    historical_data = self.get_market_data(symbol, '1h', days * 24)
    
    if historical_data is None:
        return {"error": "Failed to fetch historical data"}
    
    # Simulate trades
    portfolio_value = 1000  # Starting with $1000
    trade_log = []
    
    for i in range(24, len(historical_data), 24):  # Daily analysis
        day_data = historical_data.iloc[i-24:i]
        
        # Get AI analysis for this day
        analysis = self.analyze_market_sentiment(symbol, day_data)
        recommendation = self.parse_trading_signal(analysis)
        
        # Simulate trade execution
        current_price = day_data['close'].iloc[-1]
        
        if recommendation['action'] == 'BUY':
            shares = portfolio_value / current_price
            trade_log.append({
                'day': i//24,
                'action': 'BUY',
                'price': current_price,
                'shares': shares,
                'portfolio_value': portfolio_value
            })
        elif recommendation['action'] == 'SELL' and trade_log:
            last_buy = next((t for t in reversed(trade_log) if t['action'] == 'BUY'), None)
            if last_buy:
                profit = (current_price - last_buy['price']) * last_buy['shares']
                portfolio_value += profit
                trade_log.append({
                    'day': i//24,
                    'action': 'SELL',
                    'price': current_price,
                    'profit': profit,
                    'portfolio_value': portfolio_value
                })
    
    return {
        'symbol': symbol,
        'starting_value': 1000,
        'ending_value': portfolio_value,
        'total_return': ((portfolio_value - 1000) / 1000) * 100,
        'trades': len(trade_log),
        'trade_log': trade_log
    }

# .env file structure
BINANCE_API_KEY=your_api_key_here
BINANCE_SECRET_KEY=your_secret_key_here
OLLAMA_HOST=http://localhost:11434

# Additional security settings
MAX_TRADE_AMOUNT=100
STOP_LOSS_PERCENTAGE=5
TAKE_PROFIT_PERCENTAGE=10

import time
from functools import wraps

def rate_limit(calls_per_minute=60):
    """Decorator to limit API calls"""
    def decorator(func):
        last_called = [0.0]
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = 60.0 / calls_per_minute - elapsed
            
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        
        return wrapper
    return decorator

# Apply rate limiting to API calls
@rate_limit(calls_per_minute=30)
def get_market_data_limited(self, symbol):
    return self.get_market_data(symbol)

# Common error handling
try:
    result = self.binance_client.get_account()
except BinanceAPIException as e:
    if e.code == -1021:
        print("Timestamp error - check system time")
    elif e.code == -2010:
        print("Account has insufficient balance")
    elif e.code == -1022:
        print("Invalid signature - check API keys")
    else:
        print(f"Binance API Error: {e}")

def diagnose_ollama(self):
    """Diagnose Ollama connection problems"""
    
    # Check if Ollama service is running
    try:
        response = requests.get(f"{self.ollama_host}/api/tags", timeout=5)
        if response.status_code == 200:
            models = response.json()
            print(f"✅ Ollama running with {len(models.get('models', []))} models")
        else:
            print(f"❌ Ollama returned status {response.status_code}")
    except requests.exceptions.ConnectionError:
        print("❌ Cannot connect to Ollama - is it running?")
        print("Start Ollama with: ollama serve")

# Optimize API calls with caching
import functools
import time

@functools.lru_cache(maxsize=128)
def cached_market_data(symbol, timestamp):
    """Cache market data for 5 minutes"""
    return self.get_market_data(symbol)

def get_cached_data(self, symbol):
    """Get cached market data"""
    current_time = int(time.time() // 300)  # 5-minute buckets
    return cached_market_data(symbol, current_time)

import logging
import json
from datetime import datetime

class ProductionTrader(BinanceOllamaTrader):
    def __init__(self):
        super().__init__()
        self.setup_logging()
        self.setup_monitoring()
    
    def setup_logging(self):
        """Configure detailed logging"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('trading_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def log_trade(self, symbol, action, result):
        """Log trading activity"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'symbol': symbol,
            'action': action,
            'result': result,
            'portfolio_value': self.get_portfolio_value()
        }
        
        self.logger.info(f"Trade executed: {json.dumps(log_entry)}")
        
        # Save to trading history
        with open('trade_history.json', 'a') as f:
            f.write(json.dumps(log_entry) + '\n')

def generate_performance_report(self):
    """Generate trading performance report"""
    
    # Read trade history
    trades = []
    try:
        with open('trade_history.json', 'r') as f:
            trades = [json.loads(line) for line in f]
    except FileNotFoundError:
        return {"error": "No trade history found"}
    
    # Calculate performance metrics
    total_trades = len(trades)
    profitable_trades = sum(1 for t in trades if t['result'].get('profit', 0) > 0)
    
    performance_report = {
        'total_trades': total_trades,
        'win_rate': (profitable_trades / total_trades) * 100 if total_trades > 0 else 0,
        'current_portfolio_value': self.get_portfolio_value(),
        'last_trade': trades[-1] if trades else None,
        'generated_at': datetime.now().isoformat()
    }
    
    return performance_report

def analyze_market_news(self, symbol):
    """Analyze cryptocurrency news sentiment"""
    
    # This would integrate with news APIs in production
    news_prompt = f"""
    Analyze the current market sentiment for {symbol} based on:
    1. Recent price movements
    2. Trading volume changes
    3. Market trends
    4. General cryptocurrency market conditions
    
    Provide a sentiment score (1-10) and explanation.
    Focus on actionable insights for trading decisions.
    """
    
    sentiment_analysis = self.query_ollama(news_prompt)
    return sentiment_analysis

def multi_factor_analysis(self, symbol):
    """Combine technical and sentiment analysis"""
    
    # Get market data
    market_data = self.get_market_data(symbol)
    technical_analysis = self.analyze_market_sentiment(symbol, market_data)
    
    # Get news sentiment
    news_sentiment = self.analyze_market_news(symbol)
    
    # Combine analyses
    combined_prompt = f"""
    Based on the following analyses for {symbol}:
    
    Technical Analysis:
    {technical_analysis}
    
    Market Sentiment:
    {news_sentiment}
    
    Provide a unified trading recommendation with:
    1. Overall confidence level (1-10)
    2. Risk assessment
    3. Suggested position size
    4. Entry and exit points
    """
    
    return self.query_ollama(combined_prompt)