291 lines
9.4 KiB
Python
291 lines
9.4 KiB
Python
"""
|
|
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) |