""" Trading safety limits and validations. This module provides safety checks and limits for crypto trading operations with reasonable defaults that won't interfere with normal operations. """ from decimal import Decimal from typing import Dict, NamedTuple, Optional, Pattern, Set import re import logging # Common patterns for crypto trading pairs SYMBOL_PATTERN = re.compile(r'^[A-Z0-9]{2,10}[-/][A-Z0-9]{2,10}$') MAX_SYMBOL_LENGTH = 20 # Longest known pair + margin for future class TradeLimits(NamedTuple): """Trading limits for a symbol.""" min_size: Decimal # Minimum trade size in base currency max_size: Decimal # Maximum trade size in base currency min_notional: Decimal # Minimum trade value in quote currency max_notional: Decimal # Maximum trade value in quote currency price_precision: int # Number of decimal places for price size_precision: int # Number of decimal places for size max_price_deviation: Decimal # Maximum allowed deviation from market price (in percent) # Default limits that are generous but still protect against extreme errors DEFAULT_LIMITS = TradeLimits( min_size=Decimal('0.00000001'), # 1 satoshi equivalent max_size=Decimal('10000.0'), # Large enough for most trades min_notional=Decimal('1.0'), # Minimum $1 equivalent max_notional=Decimal('10000000.0'), # $10M per trade limit price_precision=8, # Standard for most exchanges size_precision=8, # Standard for most exchanges max_price_deviation=Decimal('30.0') # 30% max deviation ) # Common stablecoin pairs can have higher limits STABLECOIN_LIMITS = DEFAULT_LIMITS._replace( max_size=Decimal('1000000.0'), # $1M equivalent max_notional=Decimal('50000000.0'), # $50M per trade max_price_deviation=Decimal('5.0') # 5% max deviation for stables ) # More restrictive limits for volatile/illiquid pairs VOLATILE_LIMITS = DEFAULT_LIMITS._replace( max_size=Decimal('1000.0'), # Smaller position size max_notional=Decimal('1000000.0'), # $1M per trade max_price_deviation=Decimal('50.0') # 50% for very volatile markets ) # Known stablecoin symbols STABLECOINS = {'USDT', 'USDC', 'DAI', 'BUSD', 'UST', 'TUSD'} def is_stablecoin_pair(symbol: str) -> bool: """Check if the trading pair involves a stablecoin.""" parts = re.split('[-/]', symbol.upper()) return any(coin in STABLECOINS for coin in parts) def get_trade_limits(symbol: str) -> TradeLimits: """ Get appropriate trade limits for a symbol. Args: symbol: Trading pair symbol Returns: TradeLimits with appropriate limits for the symbol """ if is_stablecoin_pair(symbol): return STABLECOIN_LIMITS return VOLATILE_LIMITS def validate_trade_size( size: Decimal, price: Decimal, symbol: str, logger: Optional[logging.Logger] = None ) -> None: """ Validate trade size against limits. Args: size: Trade size in base currency price: Trade price symbol: Trading pair symbol logger: Optional logger for warnings Raises: ValueError: If size violates limits """ limits = get_trade_limits(symbol) notional = size * price # Check minimum size if size < limits.min_size: raise ValueError( f"Trade size {size} below minimum {limits.min_size} for {symbol}" ) # Check maximum size with warning at 90% if size > limits.max_size * Decimal('0.9') and logger: logger.warning( f"Large trade size {size} approaching maximum {limits.max_size} for {symbol}" ) if size > limits.max_size: raise ValueError( f"Trade size {size} exceeds maximum {limits.max_size} for {symbol}" ) # Check minimum notional if notional < limits.min_notional: raise ValueError( f"Trade value ${notional} below minimum ${limits.min_notional} for {symbol}" ) # Check maximum notional with warning at 90% if notional > limits.max_notional * Decimal('0.9') and logger: logger.warning( f"Large trade value ${notional} approaching maximum ${limits.max_notional} for {symbol}" ) if notional > limits.max_notional: raise ValueError( f"Trade value ${notional} exceeds maximum ${limits.max_notional} for {symbol}" ) def validate_trade_price( price: Decimal, market_price: Optional[Decimal], symbol: str, logger: Optional[logging.Logger] = None ) -> None: """ Validate trade price against limits and market price. Args: price: Trade price market_price: Current market price (if available) symbol: Trading pair symbol logger: Optional logger for warnings Raises: ValueError: If price violates limits """ limits = get_trade_limits(symbol) # Skip market price check if not available if market_price is None: return # Calculate price deviation deviation = abs(price - market_price) / market_price * 100 # Warn at 80% of maximum deviation if deviation > limits.max_price_deviation * Decimal('0.8') and logger: logger.warning( f"Price deviation {deviation}% approaching maximum {limits.max_price_deviation}% for {symbol}" ) # Error at maximum deviation if deviation > limits.max_price_deviation: raise ValueError( f"Price deviation {deviation}% exceeds maximum {limits.max_price_deviation}% for {symbol}" ) def validate_symbol_format( symbol: str, logger: Optional[logging.Logger] = None ) -> None: """ Validate trading symbol format. Args: symbol: Trading pair symbol logger: Optional logger for warnings Raises: ValueError: If symbol format is invalid """ if not symbol or not isinstance(symbol, str): raise ValueError(f"Invalid symbol: {symbol}") # Check length if len(symbol) > MAX_SYMBOL_LENGTH: raise ValueError(f"Symbol too long: {symbol}") # Check format if not SYMBOL_PATTERN.match(symbol.upper()): raise ValueError( f"Invalid symbol format: {symbol}. Expected format: 'XXX-YYY' or 'XXX/YYY'" )