""" Triangle enumeration for triangular arbitrage. Provides the core data structures (TradingPair, TriangleLeg, Triangle, TriangleIndex) and the enumerate_triangles() function that enumerates all valid triangles from a list of TradingPairs using a fee table. A triangle is a directed cycle of three currencies c1 → c2 → c3 → c1 where each leg corresponds to a tradable pair. The hold currency (primary quote) is the currency that enters and exits the cycle; it must be one of the three currencies and is typically USDT or USDC. """ from dataclasses import dataclass, field from typing import Optional import structlog logger = structlog.get_logger().bind(component="triangle_enum") @dataclass(frozen=True) class TradingPair: """ A single tradable currency pair on KuCoin. Attributes ---------- symbol : str KuCoin symbol e.g. "BTC-USDT". base : str Base currency code e.g. "BTC". quote : str Quote currency code e.g. "USDT". fee_currency : str Currency in which fees are denominated for this pair (from KuCoin's feeCurrency field). Included in signal payloads so the executor can interpret fee amounts correctly. """ symbol: str base: str quote: str fee_currency: str = "" @classmethod def from_api_response(cls, data: dict) -> Optional["TradingPair"]: """ Parse a KuCoin /symbols API response entry into a TradingPair. Returns None if enableTrading is not True or if required fields are missing. """ if data.get("enableTrading") is not True: return None symbol = data.get("symbol", "") base = data.get("baseCurrency", "") quote = data.get("quoteCurrency", "") if not all([symbol, base, quote]): return None return cls(symbol=symbol, base=base, quote=quote, fee_currency=data.get("feeCurrency", "")) @property def currency_pair(self) -> frozenset[str]: """The unordered set of the two currencies in this pair.""" return frozenset([self.base, self.quote]) @dataclass class TriangleLeg: """ One directed hop in a triangle: input_currency → output_currency via a pair. Attributes ---------- pair : TradingPair The trading pair used for this leg. input_currency : str Currency entering this leg. output_currency : str Currency leaving this leg. maker_fee, taker_fee : float Fee rates (fraction) for this leg's base currency. """ pair: TradingPair input_currency: str output_currency: str maker_fee: float taker_fee: float @dataclass class Triangle: """ A complete triangular arbitrage cycle. Attributes ---------- legs : list[TriangleLeg] The three directed hops (must sum to identity: c1 → c2 → c3 → c1). currencies : frozenset[str] The three distinct currency codes in the cycle. pair_symbols : frozenset[str] The three KuCoin symbols involved. primary_quote : str The hold currency that enters and exits the cycle. All minimum order sizes and volumes are expressed in terms of this currency. """ legs: list[TriangleLeg] = field(default_factory=list) currencies: frozenset[str] = field(default_factory=frozenset()) pair_symbols: frozenset[str] = field(default_factory=frozenset()) primary_quote: str = "" @dataclass class TriangleIndex: """ Inverted index of triangles by pair symbol. Allows O(1) lookup of all triangles that involve a given symbol, which is the primary query pattern used by OpportunityEngine on every book update. """ triangles: list[Triangle] = field(default_factory=list) by_pair: dict[str, list[Triangle]] = field(default_factory=dict) def get_triangles_for_pair(self, symbol: str) -> list[Triangle]: """Return all triangles that include the given symbol.""" return self.by_pair.get(symbol, []) def _build_pair_map(pairs: list[TradingPair]) -> dict[frozenset[str], TradingPair]: """ Build a mapping from unordered currency pair (frozenset) to TradingPair. Used to look up pairs by their two currencies regardless of direction. """ pair_map: dict[frozenset[str], TradingPair] = {} for p in pairs: pair_map[p.currency_pair] = p return pair_map def _build_edge_map(pairs: list[TradingPair]) -> dict[str, list[frozenset[str]]]: """ Build an adjacency map: currency → list of currency pairs involving that currency. This is the graph representation used by enumerate_triangles to efficiently find paths of length 2 (c1 → c2 → c3) without enumerating all O(n²) pairs. """ edge_map: dict[str, list[frozenset[str]]] = {} for p in pairs: for c in [p.base, p.quote]: if c not in edge_map: edge_map[c] = [] edge_map[c].append(p.currency_pair) return edge_map def _build_legs( c1: str, c2: str, c3: str, pair_map: dict[frozenset[str], TradingPair], fee_table: dict[str, dict[str, float]], ) -> list[TriangleLeg]: """ Construct the three TriangleLegs for the cycle c1 → c2 → c3 → c1. Looks up each leg's pair via pair_map (keyed by unordered currencies) and resolves fees from fee_table using the base currency of each pair. """ default_fee = {"maker": 0.0010, "taker": 0.0010} def fee_for(base: str, side: str) -> float: return fee_table.get(base, default_fee).get(side, 0.0010) p1 = pair_map[frozenset([c1, c2])] p2 = pair_map[frozenset([c2, c3])] p3 = pair_map[frozenset([c3, c1])] leg1 = TriangleLeg( pair=p1, input_currency=c1, output_currency=c2, maker_fee=fee_for(p1.base, "maker"), taker_fee=fee_for(p1.base, "taker"), ) leg2 = TriangleLeg( pair=p2, input_currency=c2, output_currency=c3, maker_fee=fee_for(p2.base, "maker"), taker_fee=fee_for(p2.base, "taker"), ) leg3 = TriangleLeg( pair=p3, input_currency=c3, output_currency=c1, maker_fee=fee_for(p3.base, "maker"), taker_fee=fee_for(p3.base, "taker"), ) return [leg1, leg2, leg3] def enumerate_triangles( pairs: list[TradingPair], fee_table: dict[str, dict[str, float]], hold_currencies: Optional[list[str]] = None, ) -> TriangleIndex: """ Enumerate all valid triangular arbitrage cycles from a list of TradingPairs. A valid triangle must: - Contain exactly three distinct currencies. - Include at least one hold currency (default: ["USDT"]), which becomes the primary_quote / entry/exit currency. - Have all three legs (c1→c2, c2→c3, c3→c1) represent tradable pairs. Parameters ---------- pairs : list[TradingPair] All known trading pairs (filtered to enableTrading == True by the caller). fee_table : dict[str, dict[str, float]] Per-base-currency fee rates as returned by oe_em.kucoin_api. hold_currencies : list[str] or None Currencies that may serve as the entry/exit point. Defaults to ["USDT"]. Returns ------- TriangleIndex Contains all triangles and an inverted index by symbol for fast lookup. """ if hold_currencies is None: hold_currencies = ["USDT"] hold_set = set(hold_currencies) pair_map = _build_pair_map(pairs) edge_map = _build_edge_map(pairs) triangles: list[Triangle] = [] by_pair: dict[str, list[Triangle]] = {} seen: set[frozenset[str]] = set() all_currencies = sorted(edge_map.keys()) for c1 in all_currencies: for c2_edge in edge_map.get(c1, []): c2 = next(x for x in c2_edge if x != c1) for c3_edge in edge_map.get(c2, []): c3 = next(x for x in c3_edge if x != c2) if c3 == c1: continue if c3_edge == c2_edge: continue if frozenset([c1, c3]) not in pair_map: continue currencies = frozenset([c1, c2, c3]) if currencies in seen: continue seen.add(currencies) in_triangle = hold_set & currencies if not in_triangle: continue for hold_curr in in_triangle: others = [c for c in [c1, c2, c3] if c != hold_curr] for x, y in [(others[0], others[1]), (others[1], others[0])]: legs = _build_legs(hold_curr, x, y, pair_map, fee_table) pair_symbols = frozenset([ legs[0].pair.symbol, legs[1].pair.symbol, legs[2].pair.symbol, ]) triangle = Triangle( legs=legs, currencies=currencies, pair_symbols=pair_symbols, primary_quote=hold_curr, ) triangles.append(triangle) for sym in pair_symbols: if sym not in by_pair: by_pair[sym] = [] by_pair[sym].append(triangle) logger.info("triangles_enumerated", total=len(triangles), indexed_pairs=len(by_pair)) return TriangleIndex(triangles=triangles, by_pair=by_pair)