[Detector] Use graph & add tests

Signed-off-by: Herklos <herklos@drakkar.software>

requirements.txt

@@ -1,4 +1,5 @@
 tqdm
 ccxt
+networkx[default]
 
 OctoBot-Commons>=1.9, <1.10

tests/test_detector.py

@@ -1,14 +1,57 @@
-import triangular_arbitrage.detector
-import octobot_commons.symbols
+import pytest
+import octobot_commons.symbols as symbols
+from triangular_arbitrage.detector import ShortTicker, get_best_opportunity
 
-def test_get_best_opportunity():
+
+@pytest.fixture
+def sample_tickers():
+    return [
+        ShortTicker(symbol=symbols.Symbol('BTC/USDT'), last_price=30000),
+        ShortTicker(symbol=symbols.Symbol('ETH/USDT'), last_price=2000),
+        ShortTicker(symbol=symbols.Symbol('XRP/USDT'), last_price=0.5),
+        ShortTicker(symbol=symbols.Symbol('LTC/USDT'), last_price=100),
+        ShortTicker(symbol=symbols.Symbol('BCH/USDT'), last_price=200),
+    ]
+
+
+def test_get_best_opportunity_handles_empty_tickers():
+    best_opportunity, best_profit = get_best_opportunity([])
+    assert best_profit == 0
+    assert best_opportunity is None
+
+
+def test_get_best_opportunity_handles_no_triplet_opportunity(sample_tickers):
+    sample_tickers.append(ShortTicker(symbol=symbols.Symbol('DOT/USDT'), last_price=0.05))
+    best_opportunity, best_profit = get_best_opportunity(sample_tickers)
+    assert best_profit == 0
+    assert best_opportunity is None
+
+
+def test_get_best_opportunity_returns_correct_triplet_with_correct_tickers():
+    tickers = [
+        ShortTicker(symbol=symbols.Symbol('BTC/USDT'), last_price=30000),
+        ShortTicker(symbol=symbols.Symbol('ETH/BTC'), last_price=0.3),
+        ShortTicker(symbol=symbols.Symbol('ETH/USDT'), last_price=2000),
+    ]
+    best_opportunity, best_profit = get_best_opportunity(tickers)
+    assert len(best_opportunity) == 3
+    assert best_profit == 4.5
+    assert all(isinstance(ticker, ShortTicker) for ticker in best_opportunity)
+
+
+def test_get_best_opportunity_returns_correct_triplet_with_multiple_tickers():
     tickers = [
-        triangular_arbitrage.detector.ShortTicker(octobot_commons.symbols.Symbol('A/B'), 1.2),
-        triangular_arbitrage.detector.ShortTicker(octobot_commons.symbols.Symbol('B/C'), 1.3),
-        triangular_arbitrage.detector.ShortTicker(octobot_commons.symbols.Symbol('C/A'), 0.7),
-        triangular_arbitrage.detector.ShortTicker(octobot_commons.symbols.Symbol('A/C'), 1.4),
-        triangular_arbitrage.detector.ShortTicker(octobot_commons.symbols.Symbol('B/A'), 0.8),
-        triangular_arbitrage.detector.ShortTicker(octobot_commons.symbols.Symbol('C/B'), 0.9)
+        ShortTicker(symbol=symbols.Symbol('BTC/USDT'), last_price=30000),
+        ShortTicker(symbol=symbols.Symbol('ETH/BTC'), last_price=0.3),
+        ShortTicker(symbol=symbols.Symbol('ETH/USDT'), last_price=2000),
+        ShortTicker(symbol=symbols.Symbol('ETH/USDC'), last_price=1900),
+        ShortTicker(symbol=symbols.Symbol('BTC/USDC'), last_price=35000),
+        ShortTicker(symbol=symbols.Symbol('USDC/USDT'), last_price=1.1),
+        ShortTicker(symbol=symbols.Symbol('USDC/TUSD'), last_price=0.95),
+        ShortTicker(symbol=symbols.Symbol('ETH/TUSD'), last_price=1950),
+        ShortTicker(symbol=symbols.Symbol('BTC/TUSD'), last_price=32500),
     ]
-    best_triplet, best_profit = triangular_arbitrage.detector.get_best_opportunity(tickers)
-    assert best_profit > 0
+    best_opportunity, best_profit = get_best_opportunity(tickers)
+    assert len(best_opportunity) == 3
+    assert best_profit == 5.526315789473684
+    assert all(isinstance(ticker, ShortTicker) for ticker in best_opportunity)

triangular_arbitrage/detector.py

@@ -5,10 +5,12 @@
 from tqdm.auto import tqdm
 from itertools import combinations
 from dataclasses import dataclass
+import networkx as nx
 
 import octobot_commons.symbols as symbols
 import octobot_commons.constants as constants
 
+
 @dataclass
 class ShortTicker:
     symbol: symbols.Symbol
@@ -19,69 +21,58 @@ class ShortTicker:
 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:
+
+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']) 
+        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 (whitelisted_symbols is None or str(get_symbol_from_key(key)) in whitelisted_symbols)
+        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 (whitelisted_symbols is None or str(get_symbol_from_key(key)) in whitelisted_symbols)
     ]
 
+
 def get_best_opportunity(tickers: List[ShortTicker]) -> Tuple[List[ShortTicker], float]:
     # pylint: disable=W1114
     ticker_dict = {str(ticker.symbol): ticker for ticker in tickers if ticker.symbol is not None}
 
-    currencies = set()
+    # Build a directed graph of currencies
+    graph = nx.DiGraph()
+
     for ticker in tickers:
         if ticker.symbol is not None:
-            currencies.add(ticker.symbol.base)
-            currencies.add(ticker.symbol.quote)
+            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 = 0
     best_triplet = None
 
-    def get_opportunity_symbol(a, b):
-        return f"{a}/{b}"
-
-    # Try all combinations of three currencies.
-    for a, b, c in tqdm(combinations(currencies, 3)):
-        # Look up the tickers in the dictionary instead of searching through the list.
-        a_to_b = ticker_dict.get(get_opportunity_symbol(a,b))
-        b_to_c = ticker_dict.get(get_opportunity_symbol(b,c))
-        c_to_a = ticker_dict.get(get_opportunity_symbol(c,a))
-
-        # If the ticker does not exist, try the inverse
-        if not a_to_b:
-            b_to_a = ticker_dict.get(get_opportunity_symbol(b,a))
-            if b_to_a:
-                a_to_b = ShortTicker(symbols.Symbol(get_opportunity_symbol(a,b)), 1/b_to_a.last_price, reversed=True)
-
-        if not b_to_c:
-            c_to_b = ticker_dict.get(get_opportunity_symbol(c,b))
-            if c_to_b:
-                b_to_c = ShortTicker(symbols.Symbol(get_opportunity_symbol(b,c)), 1/c_to_b.last_price, reversed=True)
-
-        if not c_to_a:
-            a_to_c = ticker_dict.get(get_opportunity_symbol(a,c))
-            if a_to_c:
-                c_to_a = ShortTicker(symbols.Symbol(get_opportunity_symbol(c,a)), 1/a_to_c.last_price, reversed=True)
-
-        if not all([a_to_b, b_to_c, c_to_a]):
+    for cycle in nx.simple_cycles(graph):
+        if len(cycle) != 3:
             continue
-
+
+        a, b, c = cycle
+        a_to_b = graph[a][b]['ticker']
+        b_to_c = graph[b][c]['ticker']
+        c_to_a = graph[c][a]['ticker']
+
         profit = a_to_b.last_price * b_to_c.last_price * c_to_a.last_price
 
         if profit > best_profit:
@@ -91,12 +82,15 @@ def get_opportunity_symbol(a, b):
     if best_triplet is not None:
         # restore original symbols for reversed pairs
         best_triplet = [
-            ShortTicker(symbols.Symbol(f"{triplet.symbol.quote}/{triplet.symbol.base}"), triplet.last_price, reversed=True) 
-            if triplet.reversed else triplet 
-            for triplet in best_triplet]
+            ShortTicker(symbols.Symbol(f"{triplet.symbol.quote}/{triplet.symbol.base}"), triplet.last_price,
+                        reversed=True)
+            if triplet.reversed else triplet
+            for triplet in best_triplet
+        ]
 
     return best_triplet, best_profit
 
+
 async def get_exchange_data(exchange_name):
     exchange_class = getattr(ccxt, exchange_name)
     exchange = exchange_class()
@@ -105,12 +99,14 @@ async def get_exchange_data(exchange_name):
     await exchange.close()
     return 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):
     last_prices = await get_exchange_last_prices(exchange_name, ignored_symbols or [], whitelisted_symbols)
-    best_opportunity, best_profit = get_best_opportunity(last_prices)    
+    best_opportunity, best_profit = get_best_opportunity(last_prices)
     return best_opportunity, best_profit