143 lines
5.0 KiB
Python
143 lines
5.0 KiB
Python
"""Tests for MetricCalculator OBI calculation and best bid/ask extraction."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.append(str(Path(__file__).resolve().parents[1]))
|
|
|
|
from models import MetricCalculator, BookSnapshot, OrderbookLevel, Trade
|
|
|
|
|
|
def test_calculate_obi_normal_case():
|
|
"""Test OBI calculation with normal bid and ask volumes."""
|
|
# Create test snapshot with more bid volume than ask volume
|
|
snapshot = BookSnapshot(
|
|
id=1,
|
|
timestamp=1000,
|
|
bids={
|
|
50000.0: OrderbookLevel(price=50000.0, size=10.0, liquidation_count=0, order_count=1),
|
|
49999.0: OrderbookLevel(price=49999.0, size=5.0, liquidation_count=0, order_count=1),
|
|
},
|
|
asks={
|
|
50001.0: OrderbookLevel(price=50001.0, size=3.0, liquidation_count=0, order_count=1),
|
|
50002.0: OrderbookLevel(price=50002.0, size=2.0, liquidation_count=0, order_count=1),
|
|
},
|
|
)
|
|
|
|
# Total bid volume = 15.0, total ask volume = 5.0
|
|
# OBI = (15 - 5) / (15 + 5) = 10 / 20 = 0.5
|
|
obi = MetricCalculator.calculate_obi(snapshot)
|
|
assert obi == 0.5
|
|
|
|
|
|
def test_calculate_obi_zero_volume():
|
|
"""Test OBI calculation when there's no volume."""
|
|
snapshot = BookSnapshot(id=1, timestamp=1000, bids={}, asks={})
|
|
obi = MetricCalculator.calculate_obi(snapshot)
|
|
assert obi == 0.0
|
|
|
|
|
|
def test_calculate_obi_ask_heavy():
|
|
"""Test OBI calculation with more ask volume than bid volume."""
|
|
snapshot = BookSnapshot(
|
|
id=1,
|
|
timestamp=1000,
|
|
bids={
|
|
50000.0: OrderbookLevel(price=50000.0, size=2.0, liquidation_count=0, order_count=1),
|
|
},
|
|
asks={
|
|
50001.0: OrderbookLevel(price=50001.0, size=8.0, liquidation_count=0, order_count=1),
|
|
},
|
|
)
|
|
|
|
# Total bid volume = 2.0, total ask volume = 8.0
|
|
# OBI = (2 - 8) / (2 + 8) = -6 / 10 = -0.6
|
|
obi = MetricCalculator.calculate_obi(snapshot)
|
|
assert obi == -0.6
|
|
|
|
|
|
def test_get_best_bid_ask_normal():
|
|
"""Test best bid/ask extraction with normal orderbook."""
|
|
snapshot = BookSnapshot(
|
|
id=1,
|
|
timestamp=1000,
|
|
bids={
|
|
50000.0: OrderbookLevel(price=50000.0, size=1.0, liquidation_count=0, order_count=1),
|
|
49999.0: OrderbookLevel(price=49999.0, size=1.0, liquidation_count=0, order_count=1),
|
|
49998.0: OrderbookLevel(price=49998.0, size=1.0, liquidation_count=0, order_count=1),
|
|
},
|
|
asks={
|
|
50001.0: OrderbookLevel(price=50001.0, size=1.0, liquidation_count=0, order_count=1),
|
|
50002.0: OrderbookLevel(price=50002.0, size=1.0, liquidation_count=0, order_count=1),
|
|
50003.0: OrderbookLevel(price=50003.0, size=1.0, liquidation_count=0, order_count=1),
|
|
},
|
|
)
|
|
|
|
best_bid, best_ask = MetricCalculator.get_best_bid_ask(snapshot)
|
|
assert best_bid == 50000.0 # Highest bid price
|
|
assert best_ask == 50001.0 # Lowest ask price
|
|
|
|
|
|
def test_get_best_bid_ask_empty():
|
|
"""Test best bid/ask extraction with empty orderbook."""
|
|
snapshot = BookSnapshot(id=1, timestamp=1000, bids={}, asks={})
|
|
best_bid, best_ask = MetricCalculator.get_best_bid_ask(snapshot)
|
|
assert best_bid is None
|
|
assert best_ask is None
|
|
|
|
|
|
def test_calculate_volume_delta_buy_heavy():
|
|
"""Test volume delta calculation with more buy volume than sell volume."""
|
|
trades = [
|
|
Trade(id=1, trade_id=1.0, price=50000.0, size=10.0, side="buy", timestamp=1000),
|
|
Trade(id=2, trade_id=2.0, price=50001.0, size=5.0, side="buy", timestamp=1000),
|
|
Trade(id=3, trade_id=3.0, price=49999.0, size=3.0, side="sell", timestamp=1000),
|
|
]
|
|
|
|
# Buy volume = 15.0, Sell volume = 3.0
|
|
# Volume Delta = 15.0 - 3.0 = 12.0
|
|
vd = MetricCalculator.calculate_volume_delta(trades)
|
|
assert vd == 12.0
|
|
|
|
|
|
def test_calculate_volume_delta_sell_heavy():
|
|
"""Test volume delta calculation with more sell volume than buy volume."""
|
|
trades = [
|
|
Trade(id=1, trade_id=1.0, price=50000.0, size=2.0, side="buy", timestamp=1000),
|
|
Trade(id=2, trade_id=2.0, price=49999.0, size=8.0, side="sell", timestamp=1000),
|
|
]
|
|
|
|
# Buy volume = 2.0, Sell volume = 8.0
|
|
# Volume Delta = 2.0 - 8.0 = -6.0
|
|
vd = MetricCalculator.calculate_volume_delta(trades)
|
|
assert vd == -6.0
|
|
|
|
|
|
def test_calculate_volume_delta_no_trades():
|
|
"""Test volume delta calculation with no trades."""
|
|
trades = []
|
|
vd = MetricCalculator.calculate_volume_delta(trades)
|
|
assert vd == 0.0
|
|
|
|
|
|
def test_calculate_cvd_incremental():
|
|
"""Test incremental CVD calculation."""
|
|
# Start with zero CVD
|
|
cvd1 = MetricCalculator.calculate_cvd(0.0, 10.0)
|
|
assert cvd1 == 10.0
|
|
|
|
# Add more volume delta
|
|
cvd2 = MetricCalculator.calculate_cvd(cvd1, -5.0)
|
|
assert cvd2 == 5.0
|
|
|
|
# Continue accumulating
|
|
cvd3 = MetricCalculator.calculate_cvd(cvd2, 15.0)
|
|
assert cvd3 == 20.0
|
|
|
|
|
|
def test_calculate_cvd_reset_functionality():
|
|
"""Test CVD reset by starting from 0.0."""
|
|
# Simulate reset by passing 0.0 as previous CVD
|
|
cvd_after_reset = MetricCalculator.calculate_cvd(0.0, 25.0)
|
|
assert cvd_after_reset == 25.0
|