Crypto Trading Bot Development: Binance, Coinbase & Multi-Exchange Integration

Trading Bot Architecture Flow

import asyncio
from binance.client import AsyncClient
from binance import BinanceSocketManager
import pandas as pd

class BinanceTradingBot:
    """Professional Binance Trading Bot Implementation"""
    
    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret
        self.client = None
        
    async def initialize(self):
        """Initialize async Binance client"""
        self.client = await AsyncClient.create(
            api_key=self.api_key,
            api_secret=self.api_secret
        )
        return self
        
    async def get_balance(self, asset: str = 'USDT') -> float:
        """Fetch available balance for asset"""
        account = await self.client.get_account()
        for balance in account['balances']:
            if balance['asset'] == asset:
                return float(balance['free'])
        return 0.0
        
    async def place_order(self, symbol: str, side: str, 
                          quantity: float, price: float):
        """Place limit order with error handling"""
        try:
            order = await self.client.create_order(
                symbol=symbol,
                side=side,
                type='LIMIT',
                timeInForce='GTC',
                quantity=quantity,
                price=str(price)
            )
            return order
        except Exception as e:
            self.logger.error(f"Order failed: {e}")
            raise

# Usage Example
async def main():
    bot = BinanceTradingBot('your_api_key', 'your_secret')
    await bot.initialize()
    balance = await bot.get_balance('USDT')
    print(f"Available USDT: {balance}")

asyncio.run(main())

import hmac
import hashlib
import time
import requests
import json

class CoinbaseAdvancedTradeBot:
    """Coinbase Advanced Trade API Implementation"""
    
    BASE_URL = "https://api.coinbase.com/api/v3/brokerage"
    
    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret
        self.session = requests.Session()
        
    def _generate_signature(self, timestamp: str, method: str, 
                              path: str, body: str = "") -> str:
        """Generate HMAC SHA256 signature"""
        message = timestamp + method.upper() + path + body
        signature = hmac.new(
            self.api_secret.encode('utf-8'),
            message.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        return signature
        
    def get_accounts(self):
        """Retrieve all trading accounts"""
        timestamp = str(int(time.time()))
        path = "/api/v3/brokerage/accounts"
        
        headers = {
            "CB-ACCESS-KEY": self.api_key,
            "CB-ACCESS-SIGN": self._generate_signature(
                timestamp, "GET", path
            ),
            "CB-ACCESS-TIMESTAMP": timestamp
        }
        
        response = self.session.get(
            f"{self.BASE_URL}/accounts",
            headers=headers
        )
        return response.json()
        
    def place_market_order(self, product_id: str, 
                           side: str, size: float):
        """Execute market order"""
        order_data = {
            "product_id": product_id,
            "side": side.upper(),
            "order_configuration": {
                "market_market_ioc": {
                    "base_size": str(size)
                }
            }
        }
        # Implementation continues...
        return order_data

from abc import ABC, abstractmethod
from typing import Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum

class Exchange(Enum):
    BINANCE = "binance"
    COINBASE = "coinbase"
    KRAKEN = "kraken"

@dataclass
class UnifiedOrder:
    """Standardized order across exchanges"""
    exchange: Exchange
    symbol: str
    side: str
    quantity: float
    price: float
    status: str = "pending"

class ExchangeInterface(ABC):
    """Abstract base for exchange implementations"""
    
    @abstractmethod
    async def get_orderbook(self, symbol: str) -> Dict:
        pass
        
    @abstractmethod
    async def place_order(self, order: UnifiedOrder):
        pass

class MultiExchangeRouter:
    """Routes orders to optimal exchange"""
    
    def __init__(self):
        self.exchanges: Dict[Exchange, ExchangeInterface] = {}
        
    async def find_best_price(self, symbol: str, 
                               side: str) -> Tuple[Exchange, float]:
        """Find exchange with best execution price"""
        best_exchange = None
        best_price = float('inf') if side == 'buy' else 0
        
        for ex, client in self.exchanges.items():
            orderbook = await client.get_orderbook(symbol)
            price = orderbook['asks'][0][0] if side == 'buy' \
                    else orderbook['bids'][0][0]
            
            if side == 'buy' and price < best_price:
                best_price, best_exchange = price, ex
            elif side == 'sell' and price > best_price:
                best_price, best_exchange = price, ex
                
        return best_exchange, best_price

import pandas as pd
import numpy as np
import ta

class TechnicalAnalysis:
    """Technical indicator calculations"""
    
    def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
        """Add technical indicators to OHLCV data"""
        
        # Trend Indicators
        df['sma_20'] = ta.trend.sma_indicator(df['close'], 20)
        df['sma_50'] = ta.trend.sma_indicator(df['close'], 50)
        df['ema_12'] = ta.trend.ema_indicator(df['close'], 12)
        df['macd'] = ta.trend.macd_diff(df['close'])
        
        # Momentum Indicators
        df['rsi'] = ta.momentum.rsi(df['close'], 14)
        df['stoch'] = ta.momentum.stoch(
            df['high'], df['low'], df['close']
        )
        
        # Volatility Indicators
        df['bb_width'] = ta.volatility.bollinger_wband(df['close'])
        df['atr'] = ta.volatility.average_true_range(
            df['high'], df['low'], df['close']
        )
        
        return df.dropna()
        
    def generate_signal(self, df: pd.DataFrame) -> int:
        """Generate trading signal: 1=Buy, -1=Sell, 0=Hold"""
        latest = df.iloc[-1]
        
        # EMA Crossover + RSI Filter
        if latest['ema_12'] > latest['sma_20'] and latest['rsi'] < 70:
            return 1  # Buy Signal
        elif latest['ema_12'] < latest['sma_20'] and latest['rsi'] > 30:
            return -1  # Sell Signal
        return 0  # Hold

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

class BacktestEngine:
    """Professional backtesting framework"""
    
    def __init__(self, initial_capital: float = 100000):
        self.initial_capital = initial_capital
        self.fee_rate = 0.001  # 0.1% per trade
        self.slippage = 0.0005  # 0.05% slippage
        
    def run_backtest(self, data: pd.DataFrame, 
                      strategy) -> Dict:
        """Execute backtest and return performance metrics"""
        
        capital = self.initial_capital
        position = 0
        trades = []
        equity_curve = []
        
        for i in range(50, len(data)):
            window = data.iloc[:i]
            signal = strategy.generate_signal(window)
            price = data.iloc[i]['close']
            
            # Apply slippage
            exec_price = price * (1 + self.slippage) if signal == 1 \
                         else price * (1 - self.slippage)
            
            if signal == 1 and position == 0:
                # Buy signal
                shares = (capital * 0.95) / exec_price
                cost = shares * exec_price * (1 + self.fee_rate)
                capital -= cost
                position = shares
                trades.append({'type': 'buy', 'price': exec_price})
                
            elif signal == -1 and position > 0:
                # Sell signal
                proceeds = position * exec_price * (1 - self.fee_rate)
                capital += proceeds
                position = 0
                trades.append({'type': 'sell', 'price': exec_price})
            
            # Track equity
            equity = capital + (position * price)
            equity_curve.append(equity)
        
        return self._calculate_metrics(equity_curve, trades)
    
    def _calculate_metrics(self, equity: List, 
                            trades: List) -> Dict:
        """Calculate performance metrics"""
        equity_series = pd.Series(equity)
        returns = equity_series.pct_change().dropna()
        
        return {
            'total_return': (equity[-1] / self.initial_capital - 1) * 100,
            'sharpe_ratio': returns.mean() / returns.std() * np.sqrt(252),
            'max_drawdown': self._max_drawdown(equity_series),
            'total_trades': len(trades),
            'win_rate': self._calculate_win_rate(trades)
        }