Building Your First Crypto Trading Bot

import logging from typing import Dict, Tuple, Optional from enum import Enum class SignalType(Enum): STRONG_BUY = 1.0 BUY = 0.75 NEUTRAL = 0.5 SELL = 0.25 STRONG_SELL = 0.0 class SignalDetector: def __init__(self, api_client): self.api = api_client self.logger = logging.getLogger('SignalDetector') self.min_confirmation = 0.70 # 70% confidence threshold def detect_signals(self, symbol: str) -> Tuple[SignalType, Dict]: """ Detect trading signals by combining multiple data sources """ try: # Fetch all relevant data whales = self._get_whale_signals(symbol) derivatives = self._get_derivatives_signals(symbol) onchain = self._get_onchain_signals(symbol) confirmation = self._get_confirmation_score(symbol) # Combine signals with weighted scoring combined_score = ( whales['score'] * 0.35 + derivatives['score'] * 0.35 + onchain['score'] * 0.20 + confirmation['score'] * 0.10 ) # Determine signal type if combined_score >= 0.80: signal = SignalType.STRONG_BUY elif combined_score >= 0.65: signal = SignalType.BUY elif combined_score >= 0.35: signal = SignalType.NEUTRAL elif combined_score >= 0.20: signal = SignalType.SELL else: signal = SignalType.STRONG_SELL return signal, { 'score': combined_score, 'whale_score': whales['score'], 'derivatives_score': derivatives['score'], 'onchain_score': onchain['score'], 'confirmation_score': confirmation['score'], 'reasoning': self._explain_signal(signal, combined_score) } except Exception as e: self.logger.error(f"Error detecting signals for {symbol}: {e}") return SignalType.NEUTRAL, {'error': str(e)} def _get_whale_signals(self, symbol: str) -> Dict: """Analyze whale positioning""" whales = self.api.get_whales(limit=100) # Count whale positions in this symbol long_count = sum(1 for w in whales['data'] if w['largest_position']['asset'] == symbol and w['leverage'] > 1) short_count = sum(1 for w in whales['data'] if w['position_type'] == 'short') whale_score = (long_count - short_count) / len(whales['data']) whale_score = max(0, min(1, whale_score / 2 + 0.5)) # Normalize to 0-1 return {'score': whale_score, 'long_count': long_count, 'short_count': short_count} def _get_derivatives_signals(self, symbol: str) -> Dict: """Analyze derivatives market conditions""" derivatives = self.api.get_derivatives(symbol) # Analyze funding rate funding_rate = float(derivatives['funding_rate']) if funding_rate > 0.005: # High positive funding funding_score = 0.2 # Bearish (longs paying shorts) elif funding_rate < -0.005: # High negative funding funding_score = 0.8 # Bullish (shorts paying longs) else: funding_score = 0.5 # Analyze open interest trend oi_change = derivatives['oi_24h_change'] if oi_change > 0.15: # Large increase in OI oi_score = 0.3 elif oi_change < -0.15: # Large decrease in OI oi_score = 0.7 else: oi_score = 0.5 combined = (funding_score * 0.6 + oi_score * 0.4) return {'score': combined, 'funding_rate': funding_rate, 'oi_change': oi_change} def _get_onchain_signals(self, symbol: str) -> Dict: """Analyze on-chain flows""" flows = self.api.get_onchain_flows() # Analyze exchange flows (deposits indicate selling pressure) net_flow = flows['total_inflows'] - flows['total_outflows'] if net_flow > 0: # More depositing flow_score = 0.3 # Bearish else: flow_score = 0.7 # Bullish return {'score': flow_score, 'net_flow': net_flow} def _get_confirmation_score(self, symbol: str) -> Dict: """Get confirmation score from API""" signals = self.api.get_signals() for signal in signals['data']: if signal['symbol'] == symbol: return { 'score': signal['confirmation_score'], 'signal_type': signal['signal_type'] } return {'score': 0.5} # Default neutral def _explain_signal(self, signal: SignalType, score: float) -> str: """Generate human-readable explanation""" if signal == SignalType.STRONG_BUY: return f"Strong buy signal ({score:.1%}): Multiple bullish indicators aligned" elif signal == SignalType.BUY: return f"Buy signal ({score:.1%}): Mixed bullish indicators" elif signal == SignalType.NEUTRAL: return f"Neutral signal ({score:.1%}): No clear direction" elif signal == SignalType.SELL: return f"Sell signal ({score:.1%}): Mixed bearish indicators" else: return f"Strong sell signal ({score:.1%}): Multiple bearish indicators aligned"

from dataclasses import dataclass from enum import Enum from datetime import datetime class OrderStatus(Enum): PENDING = 'pending' FILLED = 'filled' PARTIAL = 'partial' CANCELLED = 'cancelled' FAILED = 'failed' @dataclass class Trade: id: str symbol: str side: str # 'buy' or 'sell' size: float entry_price: float entry_time: datetime stop_loss: Optional[float] = None take_profit: Optional[float] = None status: OrderStatus = OrderStatus.PENDING exit_price: Optional[float] = None exit_time: Optional[datetime] = None class TradeExecutor: def __init__(self, exchange_client, portfolio_manager, risk_manager): self.exchange = exchange_client self.portfolio = portfolio_manager self.risk = risk_manager self.logger = logging.getLogger('TradeExecutor') self.open_trades = {} def execute_signal(self, signal: SignalType, symbol: str, signal_data: Dict) -> Optional[Trade]: """Execute a trade based on signal""" # Validate signal strength if signal_data['score'] < 0.65: self.logger.info(f"Signal score {signal_data['score']:.2%} below threshold") return None # Check if we should trade if not self._should_trade(symbol, signal): return None # Calculate position size size = self.risk.calculate_position_size(symbol, signal) if size <= 0: self.logger.warning(f"Position size calculation returned {size}") return None try: # Determine trade direction if signal in [SignalType.STRONG_BUY, SignalType.BUY]: side = 'buy' else: side = 'sell' # Get current price current_price = self._get_current_price(symbol) # Calculate stop loss and take profit stop_loss = self.risk.calculate_stop_loss(symbol, side, current_price) take_profit = self.risk.calculate_take_profit(symbol, side, current_price) # Place order order = self.exchange.place_order( symbol=symbol, side=side, size=size, price=current_price, stop_loss=stop_loss, take_profit=take_profit ) # Create trade record trade = Trade( id=order['id'], symbol=symbol, side=side, size=size, entry_price=current_price, entry_time=datetime.now(), stop_loss=stop_loss, take_profit=take_profit ) self.open_trades[trade.id] = trade self.logger.info(f"Trade {trade.id}: {side} {size} {symbol} @ {current_price}") return trade except Exception as e: self.logger.error(f"Trade execution failed: {e}") return None def _should_trade(self, symbol: str, signal: SignalType) -> bool: """Check if we should trade this signal""" # Check if we already have position in this symbol if self.portfolio.has_open_position(symbol): return False # Check if we have available capital if not self.portfolio.has_available_capital(): return False return True def _get_current_price(self, symbol: str) -> float: """Get current price from exchange""" return self.exchange.get_latest_price(symbol) def check_exit_conditions(self): """Periodically check if trades should be closed""" for trade_id, trade in list(self.open_trades.items()): current_price = self._get_current_price(trade.symbol) # Check stop loss if trade.stop_loss and self._check_stop_loss(trade, current_price): self.close_trade(trade_id, current_price, 'stop_loss') # Check take profit elif trade.take_profit and self._check_take_profit(trade, current_price): self.close_trade(trade_id, current_price, 'take_profit') def close_trade(self, trade_id: str, exit_price: float, reason: str): """Close an open trade""" trade = self.open_trades.get(trade_id) if not trade: return try: self.exchange.close_position(trade.id) trade.exit_price = exit_price trade.exit_time = datetime.now() trade.status = OrderStatus.FILLED profit = (exit_price - trade.entry_price) * trade.size profit_pct = (exit_price - trade.entry_price) / trade.entry_price self.logger.info( f"Trade {trade_id} closed ({reason}): " f"${profit:.2f} ({profit_pct:.2%})" ) del self.open_trades[trade_id] except Exception as e: self.logger.error(f"Failed to close trade {trade_id}: {e}")

from collections import defaultdict class PortfolioManager: def __init__(self, initial_capital: float, max_position_size: float = 0.1): self.initial_capital = initial_capital self.current_capital = initial_capital self.max_position_size = max_position_size # Max 10% per position self.positions = defaultdict(float) # Symbol -> size self.trade_history = [] self.logger = logging.getLogger('Portfolio') def add_position(self, symbol: str, size: float, price: float): """Add a position to portfolio""" cost = size * price if cost > self.current_capital: raise ValueError(f"Insufficient capital. Need ${cost}, have ${self.current_capital}") self.positions[symbol] += size self.current_capital -= cost self.logger.info(f"Added {size} {symbol} @ ${price}") def close_position(self, symbol: str, size: float, price: float) -> float: """Close a position and return profit/loss""" if self.positions[symbol] < size: raise ValueError(f"Insufficient {symbol} to close") revenue = size * price self.positions[symbol] -= size self.current_capital += revenue return revenue def get_portfolio_value(self, current_prices: Dict[str, float]) -> float: """Calculate total portfolio value""" holdings_value = sum( self.positions[symbol] * current_prices.get(symbol, 0) for symbol in self.positions ) return self.current_capital + holdings_value def get_returns(self, current_prices: Dict[str, float]) -> Tuple[float, float]: """Get absolute and percentage returns""" portfolio_value = self.get_portfolio_value(current_prices) absolute_return = portfolio_value - self.initial_capital percentage_return = absolute_return / self.initial_capital return absolute_return, percentage_return def has_available_capital(self) -> bool: """Check if we have capital for new trades""" return self.current_capital > 0 def has_open_position(self, symbol: str) -> bool: """Check if we have position in symbol""" return self.positions[symbol] > 0 def get_position_percentage(self, symbol: str, current_price: float) -> float: """Get position size as % of portfolio""" position_value = self.positions[symbol] * current_price portfolio_value = sum( self.positions[s] * current_price for s in self.positions ) if portfolio_value == 0: return 0 return position_value / portfolio_value