import logging from typing import Dict logger = logging.getLogger(__name__) class FeeCalculator: """Calculates fees and validates net profit after fees""" # Typical fee structures (can be customized per exchange) DEFAULT_MAKER_FEE = 0.0002 # 0.02% maker fee DEFAULT_TAKER_FEE = 0.0004 # 0.04% taker fee def __init__(self): # Exchange-specific fees (can be expanded) self.exchange_fees = { 'binance': {'maker': 0.0002, 'taker': 0.0004}, 'bybit': {'maker': 0.0001, 'taker': 0.0006}, 'okx': {'maker': 0.0002, 'taker': 0.0005}, # Default for others } def calculate_total_fees(self, opportunity: Dict, position_size: float, use_maker: bool = True) -> float: """Calculate total fees for opening and closing a position""" long_exchange = opportunity['long_exchange'] short_exchange = opportunity['short_exchange'] fee_rate = self.DEFAULT_MAKER_FEE if use_maker else self.DEFAULT_TAKER_FEE # Get exchange-specific fees long_fee = self.exchange_fees.get(long_exchange, {}).get('maker' if use_maker else 'taker', fee_rate) short_fee = self.exchange_fees.get(short_exchange, {}).get('maker' if use_maker else 'taker', fee_rate) # Fees for opening (long + short) open_fees = (long_fee + short_fee) * position_size # Fees for closing (long + short) close_fees = (long_fee + short_fee) * position_size total_fees = open_fees + close_fees return total_fees def calculate_net_profit(self, opportunity: Dict, position_size: float, use_maker: bool = True) -> Dict: """Calculate net profit after fees""" net_funding_rate = opportunity.get('net_funding_rate', 0) # Expected profit from funding rate (per funding period) expected_profit = net_funding_rate * position_size # Total fees total_fees = self.calculate_total_fees(opportunity, position_size, use_maker) # Net profit net_profit = expected_profit - total_fees # Annualized (3 funding periods per day) annualized_profit = net_profit * 3 * 365 return { 'gross_profit': expected_profit, 'total_fees': total_fees, 'net_profit': net_profit, 'net_profit_rate': net_profit / position_size if position_size > 0 else 0, 'annualized_profit': annualized_profit, 'fee_percentage': (total_fees / position_size) * 100 if position_size > 0 else 0, } def is_profitable(self, opportunity: Dict, position_size: float, min_net_rate: float = 0.0002) -> bool: """Check if opportunity is profitable after fees""" profit_data = self.calculate_net_profit(opportunity, position_size) return profit_data['net_profit_rate'] >= min_net_rate def get_fee_for_exchange(self, exchange_id: str, order_type: str = 'maker') -> float: """Get fee rate for a specific exchange""" return self.exchange_fees.get(exchange_id, {}).get(order_type, self.DEFAULT_MAKER_FEE if order_type == 'maker' else self.DEFAULT_TAKER_FEE)