zwitschi
652b20274a
- Implement IncrementalCycleDetector for scoring based on updated market data. - Introduce CycleScore class to encapsulate cycle scoring details. - Update CurrencyGraph and MarketDataFeed to integrate cycle detection. - Add unit tests for IncrementalCycleDetector functionality.
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
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 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].is_profitable
|