Arbitrage Algorithms

Rate(A→B) × Rate(B→C) × Rate(C→A) ≠ 1.0

Rate(A→B) × Rate(B→C) × Rate(C→A) > 1.0 + fees

import networkx as nx
from typing import List, Tuple

class TriangularArbitrage:
    def __init__(self, currencies: List[str]):
        self.graph = nx.DiGraph()
        self.currencies = currencies

    def update_rate(self, from_curr: str, to_curr: str, rate: float):
        """Update exchange rate in graph"""
        # Store negative log for Bellman-Ford
        self.graph.add_edge(
            from_curr,
            to_curr,
            weight=-math.log(rate)
        )

    def find_arbitrage(self) -> List[Tuple[List[str], float]]:
        """Find all arbitrage opportunities using Bellman-Ford"""
        opportunities = []

        for source in self.currencies:
            # Run Bellman-Ford from each currency
            distances, predecessors = self._bellman_ford(source)

            # Check for negative cycles
            for u, v, data in self.graph.edges(data=True):
                if distances[u] + data['weight'] < distances[v]:
                    # Negative cycle found = arbitrage opportunity
                    cycle = self._extract_cycle(v, predecessors)
                    profit = self._calculate_profit(cycle)
                    if profit > 0:
                        opportunities.append((cycle, profit))

        return opportunities

    def _calculate_profit(self, cycle: List[str]) -> float:
        """Calculate profit percentage for a cycle"""
        rate = 1.0
        for i in range(len(cycle) - 1):
            edge_data = self.graph[cycle[i]][cycle[i+1]]
            rate *= math.exp(-edge_data['weight'])

        # Return profit as percentage
        return (rate - 1.0) * 100

         EUR
        /   \
       /     \
   1.08      0.85
     /         \
    /           \
  USD ────────── GBP
       1.27

If USD→EUR→GBP→USD > 1.0, arbitrage exists

from dataclasses import dataclass
from typing import Dict

@dataclass
class ExchangeQuote:
    exchange: str
    symbol: str
    bid: float
    ask: float
    bid_size: float
    ask_size: float
    timestamp: float

class InterExchangeArbitrage:
    def __init__(self, exchanges: List[str]):
        self.quotes: Dict[str, Dict[str, ExchangeQuote]] = {}
        self.fee_rates = {
            'binance': 0.001,
            'coinbase': 0.006,
            'kraken': 0.0026,
            'ftx': 0.0007,  # RIP
        }

    def update_quote(self, quote: ExchangeQuote):
        """Update quote for exchange/symbol pair"""
        if quote.symbol not in self.quotes:
            self.quotes[quote.symbol] = {}
        self.quotes[quote.symbol][quote.exchange] = quote

    def find_opportunities(self, symbol: str) -> List[Dict]:
        """Find arbitrage opportunities for a symbol"""
        quotes = self.quotes.get(symbol, {})
        opportunities = []

        exchanges = list(quotes.keys())
        for i, buy_ex in enumerate(exchanges):
            for sell_ex in exchanges[i+1:]:
                buy_quote = quotes[buy_ex]
                sell_quote = quotes[sell_ex]

                # Buy on exchange with lower ask, sell where bid is higher
                profit = self._calculate_profit(
                    buy_quote, sell_quote,
                    self.fee_rates[buy_ex], self.fee_rates[sell_ex]
                )

                if profit > 0:
                    opportunities.append({
                        'buy_exchange': buy_ex,
                        'sell_exchange': sell_ex,
                        'buy_price': buy_quote.ask,
                        'sell_price': sell_quote.bid,
                        'profit_pct': profit,
                        'max_size': min(buy_quote.ask_size, sell_quote.bid_size),
                    })

        return opportunities

    def _calculate_profit(
        self,
        buy: ExchangeQuote,
        sell: ExchangeQuote,
        buy_fee: float,
        sell_fee: float
    ) -> float:
        """Calculate net profit after fees"""
        buy_cost = buy.ask * (1 + buy_fee)
        sell_revenue = sell.bid * (1 - sell_fee)
        return ((sell_revenue / buy_cost) - 1) * 100

import random
from typing import Callable

class GeneticOptimizer:
    """Genetic algorithm for optimal arbitrage route finding"""

    def __init__(
        self,
        fitness_fn: Callable,
        population_size: int = 100,
        mutation_rate: float = 0.1,
        generations: int = 50
    ):
        self.fitness_fn = fitness_fn
        self.pop_size = population_size
        self.mutation_rate = mutation_rate
        self.generations = generations

    def optimize(self, routes: List[List[str]]) -> List[str]:
        """Find optimal route using genetic algorithm"""
        # Initialize population
        population = self._initialize_population(routes)

        for generation in range(self.generations):
            # Evaluate fitness
            fitness_scores = [
                (route, self.fitness_fn(route))
                for route in population
            ]

            # Selection (tournament)
            selected = self._tournament_selection(fitness_scores)

            # Crossover
            offspring = self._crossover(selected)

            # Mutation
            population = self._mutate(offspring)

        # Return best route
        best = max(population, key=self.fitness_fn)
        return best

    def _tournament_selection(self, scored: List, k: int = 3):
        """Tournament selection for genetic diversity"""
        selected = []
        for _ in range(self.pop_size):
            tournament = random.sample(scored, k)
            winner = max(tournament, key=lambda x: x[1])
            selected.append(winner[0])
        return selected

import heapq
from typing import Dict, Set

def find_best_route(
    graph: Dict[str, Dict[str, float]],
    start: str,
    end: str
) -> Tuple[List[str], float]:
    """
    A* search for optimal arbitrage route

    Args:
        graph: Currency graph with edge weights as -log(rate)
        start: Starting currency
        end: Target currency

    Returns:
        (route, profit) tuple
    """
    # Priority queue: (estimated_cost, actual_cost, node, path)
    pq = [(0, 0, start, [start])]
    visited: Set[str] = set()

    while pq:
        est_cost, actual_cost, current, path = heapq.heappop(pq)

        if current == end and len(path) > 1:
            # Calculate actual profit
            profit = math.exp(-actual_cost) - 1
            return (path, profit * 100)

        if current in visited:
            continue
        visited.add(current)

        for neighbor, weight in graph.get(current, {}).items():
            if neighbor not in visited:
                new_cost = actual_cost + weight
                # Heuristic: estimate remaining cost
                heuristic = self._estimate_remaining(neighbor, end)
                heapq.heappush(pq, (
                    new_cost + heuristic,
                    new_cost,
                    neighbor,
                    path + [neighbor]
                ))

    return ([], 0)

// Florin Fixie arbitrage detection
const arbitrageOpportunities = await florin.detectArbitrage({
  currencies: ['USD', 'EUR', 'GBP', 'SOL'],
  minProfit: 0.1,  // 0.1% minimum
  maxHops: 4,
});

archive/fxyz-knowledge-project/old-projects/lagrange-repos/Triangular-Arbitrage-Data-master/
├── data/
│   ├── forex_rates_2021.csv
│   ├── forex_rates_2022.csv
│   ├── dex_prices/
│   └── cex_prices/
├── models/
│   ├── bellman_ford.py
│   ├── genetic.py
│   └── astar.py
└── analysis/
    └── backtest_results.ipynb