allucanget/arbitrade
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
arbitrade/tests/unit/test_incremental_detector.py
T

307 lines
9.0 KiB
Python
Raw Permalink Blame History

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)
Reference in New Issue View Git Blame Copy Permalink