triangular_arbitrage/detector.py

# 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