""" Tests for live/monitors/term_structure.py, liq_waterfall.py, and microstructure/spoof_detector.py. """ from live.monitors.term_structure import TermStructureMonitor from live.monitors.liq_waterfall import LiquidationWaterfall from microstructure.spoof_detector import SpoofDetector class TestTermStructure: def test_initial_no_signal(self): tsm = TermStructureMonitor() s = tsm.signal("BTC") assert s["primary_signal"] == "none" def test_basis_computation(self): tsm = TermStructureMonitor(basis_window=10) for i in range(10): tsm.update_perp("BTC", 64500 + i * 10, 0.00001) tsm.update_quarterly("BTC", 64550 + i * 10) basis = tsm.quarterly_perp_basis("BTC") assert basis is not None assert basis["current_bps"] > 0 # quarterly > perp assert basis["n_samples"] >= 2 def test_zscore_signal(self): tsm = TermStructureMonitor(basis_window=20) # Create converging basis (quarterly premium dropping) for i in range(20): tsm.update_perp("BTC", 64500, 0.00001) quarterly_premium = 100 * (1 - i / 20) # declining premium tsm.update_quarterly("BTC", 64500 + quarterly_premium) basis = tsm.quarterly_perp_basis("BTC") assert basis is not None assert basis["z_score"] < 0 # basis declining below mean def test_fair_quarterly_price(self): tsm = TermStructureMonitor(basis_window=10) for _ in range(10): tsm.update_perp("BTC", 64500, 0.00001) fair = tsm.fair_quarterly_price("BTC") assert fair is not None assert fair["perp_price"] == 64500 assert fair["fair_quarterly"] >= 64500 # positive carry def test_signal_on_anomaly(self): tsm = TermStructureMonitor(basis_window=30) for _ in range(15): tsm.update_perp("BTC", 64500, 0.00001) tsm.update_quarterly("BTC", 64550) # Spike: quarterly jumps way above fair (>3 sigma) for _ in range(15): tsm.update_perp("BTC", 64500, 0.00001) tsm.update_quarterly("BTC", 65200) # massive premium ~108 bps s = tsm.signal("BTC") basis = tsm.quarterly_perp_basis("BTC") assert basis is not None assert abs(basis["z_score"]) > 0 # deviation exists class TestLiquidationWaterfall: def test_initial_no_risk(self): lw = LiquidationWaterfall() assert len(lw.at_risk_accounts()) == 0 def test_margin_ratio_computation(self): lw = LiquidationWaterfall() lw.update_account("0xabc123", [ {"coin": "BTC", "szi": "1.0", "entryPx": "64000"}, {"coin": "ETH", "szi": "-10.0", "entryPx": "3100"}, ], margin_balance=2500) # lower balance → at risk risky = lw.at_risk_accounts() assert len(risky) == 1 assert risky[0]["level"] in ("danger", "critical") def test_safe_account_not_flagged(self): lw = LiquidationWaterfall() lw.update_account("0xsafe", [ {"coin": "BTC", "szi": "0.1", "entryPx": "64000"}, ], margin_balance=100000) assert len(lw.at_risk_accounts()) == 0 def test_liquidation_order_prediction(self): lw = LiquidationWaterfall() lw.update_account("0xwhale", [ {"coin": "BTC", "szi": "5.0", "entryPx": "64000"}, {"coin": "ETH", "szi": "-50.0", "entryPx": "3100"}, {"coin": "SOL", "szi": "1000.0", "entryPx": "140"}, ], margin_balance=30000) # Set book depths: BTC very liquid, SOL very thin lw.update_book_depth("BTC", 1000000, 1000000) lw.update_book_depth("ETH", 500000, 500000) lw.update_book_depth("SOL", 10000, 10000) order = lw.predict_liquidation_order("0xwhale") assert len(order) == 3 # SOL should be first (thin book, high mm/book ratio) assert order[0]["predicted_first"] assert order[0]["coin"] == "SOL" def test_signal_when_at_risk(self): lw = LiquidationWaterfall(danger_margin_ratio=5.0) lw.update_account("0xrisk", [ {"coin": "BTC", "szi": "1.0", "entryPx": "64000"}, ], margin_balance=2000) lw.update_book_depth("BTC", 10000, 10000) s = lw.signal("0xrisk") assert s["action"] == "position" assert s["liquidation_target"] == "BTC" def test_signal_ignores_safe_account(self): lw = LiquidationWaterfall() lw.update_account("0xsafe", [ {"coin": "BTC", "szi": "0.1", "entryPx": "64000"}, ], margin_balance=100000) s = lw.signal("0xsafe") assert s["action"] == "none" class TestSpoofDetector: def test_initial_probability_zero(self): sd = SpoofDetector() assert sd.spoof_probability() == 0.0 def test_normal_order_not_spoof(self): sd = SpoofDetector() sd.record_place("o1", "bid", 64400, 0.001, 64500, 100.0) is_spoof = sd.record_cancel("o1", 64500, 105.0) assert not is_spoof # small order, close to mid, reasonable lifetime def test_large_far_quick_cancel_is_spoof(self): sd = SpoofDetector(size_multiple=2.0) for _ in range(10): sd.record_place(f"fill_{_}", "bid", 64400, 0.001, 64500, 0.0) sd.record_fill(f"fill_{_}", 1.0) # Large order far from mid, cancelled immediately sd.record_place("spoof1", "bid", 63000, 10.0, 64500, 200.0) # 1500 bps from mid is_spoof = sd.record_cancel("spoof1", 64500, 200.1) assert is_spoof def test_oversized_short_lived_is_spoof(self): sd = SpoofDetector(size_multiple=2.0) for _ in range(10): sd.record_place(f"n{_}", "bid", 64400, 0.001, 64500, 0.0) sd.record_fill(f"n{_}", 1.0) sd.record_place("big1", "bid", 64400, 50.0, 64500, 300.0) is_spoof = sd.record_cancel("big1", 64500, 301.5) assert is_spoof # 50x typical size, <2s lifetime def test_spoof_probability_increases(self): sd = SpoofDetector(window_seconds=5.0) for _ in range(10): sd.record_place(f"n{_}", "bid", 64400, 0.001, 64500, 0.0) sd.record_fill(f"n{_}", 1.0) # Inject spoofs for i in range(5): sd.record_place(f"s{i}", "bid", 63000, 10.0, 64500, 100.0 + i * 0.1) sd.record_cancel(f"s{i}", 64500, 100.1 + i * 0.1) assert sd.spoof_probability() > 0 def test_summary(self): sd = SpoofDetector() sd.record_place("o1", "bid", 64400, 0.001, 64500, 100.0) sd.record_cancel("o1", 64500, 105.0) s = sd.summary() assert "spoof_probability" in s assert "total_orders" in s assert s["total_orders"] == 1