# pylint: disable=W0702, C0325 import ccxt.async_support as ccxt from typing import List, Tuple, Optional, Dict from dataclasses import dataclass import networkx as nx import octobot_commons.symbols as symbols import octobot_commons.constants as constants from triangular_arbitrage.fees import ( get_exchange_fees, calculate_net_profit, get_fee_breakdown, ExchangeFees, fetch_user_fee_tier ) @dataclass class ShortTicker: symbol: symbols.Symbol last_price: float reversed: bool = False def __repr__(self): return f"ShortTicker(symbol={str(self.symbol)}, last_price={self.last_price}, reversed={self.reversed})" async def fetch_tickers(exchange): return await exchange.fetch_tickers() if exchange.has['fetchTickers'] else {} def get_symbol_from_key(key_symbol: str) -> symbols.Symbol: try: return symbols.parse_symbol(key_symbol) except: return None def is_delisted_symbols(exchange_time, ticker, threshold=1 * constants.DAYS_TO_SECONDS * constants.MSECONDS_TO_SECONDS) -> bool: ticker_time = ticker['timestamp'] return ticker_time is not None and not (exchange_time - ticker_time <= threshold) def get_last_prices(exchange_time, tickers, ignored_symbols, whitelisted_symbols=None): return [ ShortTicker(symbol=get_symbol_from_key(key), last_price=tickers[key]['close']) for key, _ in tickers.items() if tickers[key]['close'] is not None and not is_delisted_symbols(exchange_time, tickers[key]) and str(get_symbol_from_key(key)) not in ignored_symbols and get_symbol_from_key(key).is_spot() and (whitelisted_symbols is None or str(get_symbol_from_key(key)) in whitelisted_symbols) ] def get_best_triangular_opportunity(tickers: List[ShortTicker]) -> Tuple[List[ShortTicker], float]: # Build a directed graph of currencies return get_best_opportunity(tickers, 3) def get_best_opportunity( tickers: List[ShortTicker], max_cycle: int = 10, exchange_fees: Optional[ExchangeFees] = None, use_taker: bool = True, trade_size_usd: float = 1000.0, slippage_factor: float = 0.0005, include_fees: bool = True ) -> Tuple[List[ShortTicker], float, Optional[Dict[str, float]]]: """ Find the best arbitrage opportunity. Args: tickers: List of tickers with prices max_cycle: Maximum cycle length to consider exchange_fees: Exchange fee structure (if None, fees won't be applied) use_taker: Whether to use taker fees (True) or maker fees (False) trade_size_usd: Estimated trade size for slippage calculation slippage_factor: Slippage factor per $1000 traded include_fees: Whether to calculate net profit with fees and slippage Returns: Tuple of (best_cycle, best_profit, fee_breakdown) - best_cycle: List of tickers in the best cycle - best_profit: Best profit multiplier (gross if include_fees=False, net if include_fees=True) - fee_breakdown: Dictionary with fee details (None if include_fees=False) """ # Build a directed graph of currencies graph = nx.DiGraph() for ticker in tickers: if ticker.symbol is not None: graph.add_edge(ticker.symbol.base, ticker.symbol.quote, ticker=ticker) graph.add_edge(ticker.symbol.quote, ticker.symbol.base, ticker=ShortTicker(symbols.Symbol(f"{ticker.symbol.quote}/{ticker.symbol.base}"), 1 / ticker.last_price, reversed=True)) best_profit = 1 best_cycle = None best_gross_profit = 1 best_fee_breakdown = None # Find all cycles in the graph with a length <= max_cycle for cycle in nx.simple_cycles(graph): if len(cycle) > max_cycle: continue # Skip cycles longer than max_cycle gross_profit = 1 tickers_in_cycle = [] # Calculate the gross profits along the cycle for i, base in enumerate(cycle): quote = cycle[(i + 1) % len(cycle)] # Wrap around to complete the cycle ticker = graph[base][quote]['ticker'] tickers_in_cycle.append(ticker) gross_profit *= ticker.last_price # Calculate net profit if fees should be included if include_fees and exchange_fees is not None: net_profit = calculate_net_profit( gross_profit, exchange_fees, num_trades=len(cycle), use_taker=use_taker, trade_size_usd=trade_size_usd, slippage_factor=slippage_factor ) profit_to_compare = net_profit else: profit_to_compare = gross_profit if profit_to_compare > best_profit: best_profit = profit_to_compare best_gross_profit = gross_profit best_cycle = tickers_in_cycle # Calculate fee breakdown if fees are included if include_fees and exchange_fees is not None: best_fee_breakdown = get_fee_breakdown( gross_profit, exchange_fees, num_trades=len(cycle), use_taker=use_taker, trade_size_usd=trade_size_usd, slippage_factor=slippage_factor ) if best_cycle is not None: best_cycle = [ ShortTicker(symbols.Symbol(f"{ticker.symbol.quote}/{ticker.symbol.base}"), ticker.last_price, reversed=True) if ticker.reversed else ticker for ticker in best_cycle ] return best_cycle, best_profit, best_fee_breakdown async def get_exchange_data(exchange_name): exchange_class = getattr(ccxt, exchange_name) exchange = exchange_class() tickers = await fetch_tickers(exchange) filtered_tickers = { symbol: ticker for symbol, ticker in tickers.items() if exchange.markets.get(symbol, {}).get( "active", True ) is True } exchange_time = exchange.milliseconds() await exchange.close() return filtered_tickers, exchange_time async def get_exchange_last_prices(exchange_name, ignored_symbols, whitelisted_symbols=None): tickers, exchange_time = await get_exchange_data(exchange_name) last_prices = get_last_prices(exchange_time, tickers, ignored_symbols, whitelisted_symbols) return last_prices async def run_detection( exchange_name, ignored_symbols=None, whitelisted_symbols=None, max_cycle=10, include_fees=True, use_taker=True, trade_size_usd=1000.0, slippage_factor=0.0005, api_key=None, api_secret=None ): """ Run arbitrage detection with optional fee calculation. Args: exchange_name: Name of the exchange ignored_symbols: List of symbols to ignore whitelisted_symbols: List of symbols to include (None = all) max_cycle: Maximum cycle length include_fees: Whether to include fees and slippage in profit calculation use_taker: Whether to use taker fees (True) or maker fees (False) trade_size_usd: Estimated trade size for slippage calculation slippage_factor: Slippage factor per $1000 traded api_key: Optional API key for fetching actual fee tiers api_secret: Optional API secret for fetching actual fee tiers Returns: Tuple of (best_opportunity, best_profit, fee_breakdown) """ last_prices = await get_exchange_last_prices(exchange_name, ignored_symbols or [], whitelisted_symbols) # Get exchange fees exchange_fees = None if include_fees: # Try to fetch user-specific fees if API keys are provided if api_key and api_secret: user_fees = await fetch_user_fee_tier(exchange_name, api_key, api_secret) if user_fees: exchange_fees = user_fees # Fall back to standard fees if exchange_fees is None: exchange_fees = get_exchange_fees(exchange_name) # Find best opportunity with fees best_opportunity, best_profit, fee_breakdown = get_best_opportunity( last_prices, max_cycle=max_cycle, exchange_fees=exchange_fees, use_taker=use_taker, trade_size_usd=trade_size_usd, slippage_factor=slippage_factor, include_fees=include_fees ) return best_opportunity, best_profit, fee_breakdown