Spaces:
Sleeping
Sleeping
File size: 8,588 Bytes
77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b 77198ce 2dee41b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 | # 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
|