from datetime import UTC, datetime, timedelta import pytest from arbitrade.detection.engine import IncrementalCycleDetector from arbitrade.detection.graph import CurrencyGraph, TriangularCycle from arbitrade.exchange.models import BookLevel from arbitrade.market_data.order_book import OrderBook def _make_book(*, bid: float, ask: float) -> OrderBook: book = OrderBook() book.apply_bids([BookLevel(price=bid, volume=1.0)]) book.apply_asks([BookLevel(price=ask, volume=1.0)]) return book def _make_book_levels( *, bids: list[tuple[float, float]], asks: list[tuple[float, float]] ) -> OrderBook: book = OrderBook() book.apply_bids([BookLevel(price=price, volume=volume) for price, volume in bids]) book.apply_asks([BookLevel(price=price, volume=volume) for price, volume in asks]) return book def test_incremental_detector_scores_only_cycles_touched_by_pair() -> None: cycle_a = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) cycle_b = TriangularCycle( currencies=("USD", "BTC", "LTC"), pairs=("BTC/USD", "LTC/BTC", "LTC/USD"), ) cycle_c = TriangularCycle( currencies=("USD", "SOL", "ADA"), pairs=("SOL/USD", "ADA/SOL", "ADA/USD"), ) cycles = [cycle_a, cycle_b, cycle_c] index = CurrencyGraph.index_cycles_by_pair(cycles) detector = IncrementalCycleDetector(index) books = { "BTC/USD": _make_book(bid=100.0, ask=100.0), "ETH/BTC": _make_book(bid=0.05, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), "LTC/BTC": _make_book(bid=0.01, ask=0.01), "LTC/USD": _make_book(bid=1.02, ask=1.03), "SOL/USD": _make_book(bid=20.0, ask=20.1), "ADA/SOL": _make_book(bid=0.02, ask=0.021), "ADA/USD": _make_book(bid=0.42, ask=0.43), } scores = detector.score_updated_pair("BTC/USD", books) assert len(scores) == 2 assert {score.cycle for score in scores} == {cycle_a, cycle_b} def test_incremental_detector_uses_best_bid_ask_for_gross_rate() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector(CurrencyGraph.index_cycles_by_pair([cycle])) books = { "BTC/USD": _make_book(bid=99.9, ask=100.0), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), } scores = detector.score_updated_pair("ETH/BTC", books) assert len(scores) == 1 assert scores[0].gross_rate == 1.04 assert scores[0].net_rate == 1.04 assert scores[0].is_profitable def test_incremental_detector_applies_fees_to_net_rate() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), fee_rate=0.001, ) books = { "BTC/USD": _make_book(bid=99.9, ask=100.0), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), } scores = detector.score_updated_pair("ETH/BTC", books) assert len(scores) == 1 assert scores[0].gross_rate == 1.04 assert scores[0].net_rate < scores[0].gross_rate def test_incremental_detector_uses_depth_and_slippage() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), max_depth_levels=2, ) books = { "BTC/USD": _make_book_levels( bids=[(99.9, 5.0)], asks=[(100.0, 0.002), (101.0, 0.020)], ), "ETH/BTC": _make_book_levels( bids=[(0.049, 5.0)], asks=[(0.05, 0.5)], ), "ETH/USD": _make_book_levels( bids=[(5.2, 5.0)], asks=[(5.21, 5.0)], ), } scores = detector.score_updated_pair("BTC/USD", books) assert len(scores) == 1 assert scores[0].gross_rate < 1.04 def test_incremental_detector_returns_no_score_on_insufficient_depth() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), max_depth_levels=1, ) books = { "BTC/USD": _make_book_levels( bids=[(99.9, 5.0)], asks=[(100.0, 0.001)], ), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.2, ask=5.21), } scores = detector.score_updated_pair("BTC/USD", books) assert scores == [] def test_incremental_detector_filters_below_profit_threshold() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), min_profit_threshold=0.05, ) books = { "BTC/USD": _make_book(bid=99.9, ask=100.0), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), } scores = detector.score_updated_pair("ETH/BTC", books) assert scores == [] def test_incremental_detector_enforces_min_order_size_by_pair() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), min_order_size_by_pair={"BTC/USD": 0.02}, ) books = { "BTC/USD": _make_book_levels( bids=[(99.9, 5.0)], asks=[(100.0, 0.005), (101.0, 0.005), (102.0, 0.005)], ), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.2, ask=5.21), } scores = detector.score_updated_pair("BTC/USD", books) assert scores == [] def test_incremental_detector_rejects_stale_books() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), max_book_age_seconds=1.0, ) books = { "BTC/USD": _make_book(bid=99.9, ask=100.0), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), } books["ETH/BTC"]._updated_at = datetime.now(UTC) - timedelta(seconds=5) scores = detector.score_updated_pair("ETH/BTC", books) assert scores == [] def test_incremental_detector_accepts_fresh_books_with_staleness_enabled() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), max_book_age_seconds=5.0, ) books = { "BTC/USD": _make_book(bid=99.9, ask=100.0), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), } now = datetime.now(UTC) for book in books.values(): book._updated_at = now - timedelta(seconds=0.2) scores = detector.score_updated_pair("ETH/BTC", books) assert len(scores) == 1 def test_incremental_detector_emits_structured_opportunity_event() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), min_profit_threshold=0.01, ) books = { "BTC/USD": _make_book(bid=99.9, ask=100.0), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), } opportunities = detector.opportunities_for_updated_pair( "ETH/BTC", books, base_capital=500.0, ) assert len(opportunities) == 1 event = opportunities[0] assert event.cycle == "USD->BTC->ETH->USD" assert event.updated_pair == "ETH/BTC" assert event.gross_pct == pytest.approx(4.0) assert event.net_pct == pytest.approx(4.0) assert event.est_profit == pytest.approx(20.0) def test_incremental_detector_estimated_profit_scales_with_capital() -> None: cycle = TriangularCycle( currencies=("USD", "BTC", "ETH"), pairs=("BTC/USD", "ETH/BTC", "ETH/USD"), ) detector = IncrementalCycleDetector( CurrencyGraph.index_cycles_by_pair([cycle]), min_profit_threshold=0.0, ) books = { "BTC/USD": _make_book(bid=99.9, ask=100.0), "ETH/BTC": _make_book(bid=0.049, ask=0.05), "ETH/USD": _make_book(bid=5.20, ask=5.21), } opportunities = detector.opportunities_for_updated_pair( "ETH/BTC", books, base_capital=250.0, ) assert len(opportunities) == 1 assert opportunities[0].est_profit == pytest.approx(10.0)