root
0ba5a05219
Add a unified historical endpoint that fans out to every exchange supporting the requested (asset_class, symbol) pair, then merges the results into a single consensus candle series with per-bar divergence metrics: - candles[i].close = median across sources - candles[i].sources = count of contributing exchanges - candles[i].div_pct = (max-min)/median for that bar's close Crypto routes BTC/ETH/SOL across Bybit + Hyperliquid + Deribit; equities route to Alpaca for now (IBKR omitted from MVP because its bars endpoint takes a relative period instead of start/end). Partial failures return a warning envelope (failed_sources) instead of failing the whole request; all sources failing → HTTP 502. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from cerbero_mcp.exchanges.cross.consensus import merge_candles
|
|
|
|
|
|
def _c(ts, o, h, l, c, v):
|
|
return {"timestamp": ts, "open": o, "high": h, "low": l, "close": c, "volume": v}
|
|
|
|
|
|
def test_empty_input():
|
|
assert merge_candles({}) == []
|
|
|
|
|
|
def test_single_source_passthrough():
|
|
out = merge_candles({"bybit": [_c(1, 100, 110, 90, 105, 5)]})
|
|
assert len(out) == 1
|
|
assert out[0]["timestamp"] == 1
|
|
assert out[0]["close"] == 105
|
|
assert out[0]["sources"] == 1
|
|
assert out[0]["div_pct"] == 0.0
|
|
|
|
|
|
def test_three_sources_identical_no_divergence():
|
|
src = {
|
|
"bybit": [_c(1, 100, 110, 90, 105, 5)],
|
|
"hyperliquid": [_c(1, 100, 110, 90, 105, 3)],
|
|
"deribit": [_c(1, 100, 110, 90, 105, 7)],
|
|
}
|
|
out = merge_candles(src)
|
|
assert len(out) == 1
|
|
assert out[0]["close"] == 105.0
|
|
assert out[0]["sources"] == 3
|
|
assert out[0]["div_pct"] == 0.0
|
|
# volume is mean across sources
|
|
assert abs(out[0]["volume"] - 5.0) < 1e-9
|
|
|
|
|
|
def test_three_sources_divergent_close():
|
|
src = {
|
|
"bybit": [_c(1, 100, 110, 90, 100, 1)],
|
|
"hyperliquid": [_c(1, 100, 110, 90, 110, 1)],
|
|
"deribit": [_c(1, 100, 110, 90, 105, 1)],
|
|
}
|
|
out = merge_candles(src)
|
|
# median of [100, 110, 105] = 105
|
|
assert out[0]["close"] == 105.0
|
|
# div_pct = (110 - 100) / 105 ≈ 0.0952
|
|
assert abs(out[0]["div_pct"] - 10 / 105) < 1e-6
|
|
assert out[0]["sources"] == 3
|
|
|
|
|
|
def test_misaligned_timestamps():
|
|
src = {
|
|
"bybit": [_c(1, 100, 110, 90, 105, 1), _c(2, 100, 110, 90, 105, 1)],
|
|
"hyperliquid": [_c(2, 100, 110, 90, 105, 1), _c(3, 100, 110, 90, 105, 1)],
|
|
}
|
|
out = merge_candles(src)
|
|
timestamps = [c["timestamp"] for c in out]
|
|
sources_by_ts = {c["timestamp"]: c["sources"] for c in out}
|
|
assert timestamps == [1, 2, 3]
|
|
assert sources_by_ts == {1: 1, 2: 2, 3: 1}
|
|
|
|
|
|
def test_two_sources_even_median():
|
|
src = {
|
|
"bybit": [_c(1, 100, 110, 90, 100, 1)],
|
|
"hyperliquid": [_c(1, 100, 110, 90, 110, 1)],
|
|
}
|
|
out = merge_candles(src)
|
|
# even median = mean of two = 105
|
|
assert out[0]["close"] == 105.0
|
|
|
|
|
|
def test_empty_source_ignored():
|
|
src = {
|
|
"bybit": [_c(1, 100, 110, 90, 105, 1)],
|
|
"hyperliquid": [],
|
|
}
|
|
out = merge_candles(src)
|
|
assert len(out) == 1
|
|
assert out[0]["sources"] == 1
|
|
|
|
|
|
def test_output_sorted_by_timestamp():
|
|
src = {
|
|
"bybit": [_c(3, 100, 110, 90, 105, 1), _c(1, 100, 110, 90, 105, 1),
|
|
_c(2, 100, 110, 90, 105, 1)],
|
|
}
|
|
out = merge_candles(src)
|
|
assert [c["timestamp"] for c in out] == [1, 2, 3]
|