""" Tests for PaperMarginBroker. Tests isolated margin trading simulation including: - Long/short position opening - Leverage and liquidation calculations - Interest accrual - P&L calculation - SL/TP triggers - Position management (add margin, reduce, close) """ import pytest from datetime import datetime, timezone, timedelta import sys import os # Add src to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) from brokers.paper_margin_broker import ( PaperMarginBroker, MarginPosition, MarginCloseResult, MarginSide ) class TestMarginPosition: """Tests for MarginPosition dataclass.""" def test_create_long_position(self): """Test creating a long position.""" pos = MarginPosition( symbol='BTC/USDT', side='long', size=1.0, entry_price=50000.0, leverage=5.0, collateral=10000.0, borrowed=40000.0, borrowed_asset='USDT', liquidation_price=42000.0, ) assert pos.symbol == 'BTC/USDT' assert pos.side == 'long' assert pos.size == 1.0 assert pos.leverage == 5.0 assert pos.borrowed_asset == 'USDT' def test_create_short_position(self): """Test creating a short position.""" pos = MarginPosition( symbol='BTC/USDT', side='short', size=1.0, entry_price=50000.0, leverage=3.0, collateral=16666.67, borrowed=1.0, # 1 BTC borrowed borrowed_asset='BTC', liquidation_price=65000.0, ) assert pos.side == 'short' assert pos.borrowed_asset == 'BTC' assert pos.borrowed == 1.0 def test_to_dict_from_dict_round_trip(self): """Test serialization round trip.""" original = MarginPosition( symbol='ETH/USDT', side='long', size=2.5, entry_price=3000.0, leverage=3.0, collateral=2500.0, borrowed=5000.0, borrowed_asset='USDT', liquidation_price=2200.0, interest_accrued=1.5, interest_rate_hourly=0.0002, price_source='kucoin', current_price=3100.0, unrealized_pnl=250.0, margin_ratio=85.0, ) data = original.to_dict() restored = MarginPosition.from_dict(data) assert restored.symbol == original.symbol assert restored.side == original.side assert restored.size == original.size assert restored.entry_price == original.entry_price assert restored.leverage == original.leverage assert restored.collateral == original.collateral assert restored.borrowed == original.borrowed assert restored.interest_accrued == original.interest_accrued assert restored.price_source == original.price_source class TestPaperMarginBroker: """Tests for PaperMarginBroker.""" @pytest.fixture def broker(self): """Create a broker with mock price provider.""" prices = {'BTC/USDT': 50000.0, 'ETH/USDT': 3000.0} def price_provider(symbol_key): if ':' in symbol_key: _, symbol = symbol_key.split(':', 1) else: symbol = symbol_key return prices.get(symbol, 0.0) return PaperMarginBroker( price_provider=price_provider, initial_balance=100000.0 ) def test_create_broker(self, broker): """Test broker initialization.""" assert broker.get_balance() == 100000.0 assert broker.get_equity() == 100000.0 assert len(broker.get_all_positions()) == 0 def test_open_long_position(self, broker): """Test opening a long position.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) assert pos.symbol == 'BTC/USDT' assert pos.side == 'long' assert pos.leverage == 5.0 assert pos.collateral == 10000.0 # Position value = 10000 * 5 = 50000 USDT # Size = 50000 / 50000 = 1 BTC assert pos.size == 1.0 # Borrowed = 50000 - 10000 = 40000 USDT assert pos.borrowed == 40000.0 assert pos.borrowed_asset == 'USDT' # Balance should be reduced by collateral assert broker.get_balance() == 90000.0 def test_open_short_position(self, broker): """Test opening a short position.""" pos = broker.open_position( symbol='BTC/USDT', side='short', collateral=10000.0, leverage=3.0, current_price=50000.0 ) assert pos.side == 'short' assert pos.leverage == 3.0 # Position value = 10000 * 3 = 30000 USDT # Size = 30000 / 50000 = 0.6 BTC assert pos.size == 0.6 # Borrowed = 0.6 BTC (base asset) assert pos.borrowed == 0.6 assert pos.borrowed_asset == 'BTC' assert broker.get_balance() == 90000.0 def test_cannot_open_duplicate_position(self, broker): """Test that duplicate positions are rejected.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0 ) with pytest.raises(ValueError, match="Position already open"): broker.open_position( symbol='BTC/USDT', side='short', collateral=5000.0, leverage=2.0, current_price=50000.0 ) def test_insufficient_balance(self, broker): """Test insufficient balance check.""" with pytest.raises(ValueError, match="Insufficient balance"): broker.open_position( symbol='BTC/USDT', side='long', collateral=200000.0, # More than balance leverage=2.0, current_price=50000.0 ) def test_max_leverage_check(self, broker): """Test maximum leverage validation.""" with pytest.raises(ValueError, match="exceeds maximum"): broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=20.0, # Exceeds max of 10x current_price=50000.0 ) def test_liquidation_price_long(self, broker): """Test liquidation price calculation for long.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # For 5x leverage, liquidation should be around 42,105 # (1 - 1/5) / (1 - 0.05) = 0.8 / 0.95 = 0.842 # 50000 * 0.842 = 42,105 assert 42000 < pos.liquidation_price < 43000 def test_liquidation_price_short(self, broker): """Test liquidation price calculation for short.""" pos = broker.open_position( symbol='BTC/USDT', side='short', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # For 5x short, liquidation should be around 57,143 # (1 + 1/5) / (1 + 0.05) = 1.2 / 1.05 = 1.143 # 50000 * 1.143 = 57,143 assert 57000 < pos.liquidation_price < 58000 def test_close_long_position_profit(self, broker): """Test closing a long position with profit.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Price goes up to 55000 result = broker.close_position('BTC/USDT', current_price=55000.0) assert result.success assert result.close_reason == 'manual' # P&L = (55000 - 50000) * 1 BTC = 5000 USDT (minus small interest) assert result.realized_pnl > 4900 # Balance should be initial + profit - interest assert broker.get_balance() > 100000.0 def test_close_long_position_loss(self, broker): """Test closing a long position with loss.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Price drops to 48000 result = broker.close_position('BTC/USDT', current_price=48000.0) assert result.success # P&L = (48000 - 50000) * 1 = -2000 USDT (plus interest cost) assert result.realized_pnl < -2000 def test_close_short_position_profit(self, broker): """Test closing a short position with profit.""" broker.open_position( symbol='BTC/USDT', side='short', collateral=10000.0, leverage=3.0, current_price=50000.0 ) # Price drops to 45000 result = broker.close_position('BTC/USDT', current_price=45000.0) assert result.success # Size = 0.6 BTC # P&L = (50000 - 45000) * 0.6 = 3000 USDT (minus interest) assert result.realized_pnl > 2900 def test_add_margin(self, broker): """Test adding margin to a position.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) old_liq = pos.liquidation_price broker.add_margin('BTC/USDT', 5000.0) pos = broker.get_position('BTC/USDT') assert pos.collateral == 15000.0 # Liquidation price should improve (move lower for long) assert pos.liquidation_price < old_liq # Balance reduced by additional margin assert broker.get_balance() == 85000.0 def test_reduce_position(self, broker): """Test partial position close.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Reduce half the position pos = broker.reduce_position('BTC/USDT', reduce_size=0.5, current_price=52000.0) assert pos is not None assert pos.size == 0.5 assert pos.collateral < 10000.0 # Half returned def test_update_pnl(self, broker): """Test P&L updates.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0 ) # Update price broker.update_price('BTC/USDT', 52000.0) events = broker.update() pos = broker.get_position('BTC/USDT') assert pos.current_price == 52000.0 # Size = 20000 / 50000 = 0.4 BTC # P&L = (52000 - 50000) * 0.4 = 800 (minus interest and entry commission) # Entry commission = 20000 * 0.001 = 20 # Expected ~780 assert pos.unrealized_pnl > 770 def test_stop_loss_trigger(self, broker): """Test SL trigger for long position.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0, stop_loss=48000.0 ) # Price drops below SL broker.update_price('BTC/USDT', 47500.0) events = broker.update() # Should have SL trigger event sl_events = [e for e in events if e.get('type') == 'sltp_triggered'] assert len(sl_events) == 1 assert sl_events[0]['trigger'] == 'stop_loss' assert sl_events[0]['symbol'] == 'BTC/USDT' # Position should be closed assert broker.get_position('BTC/USDT') is None def test_take_profit_trigger(self, broker): """Test TP trigger for long position.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0, take_profit=55000.0 ) # Price rises above TP broker.update_price('BTC/USDT', 56000.0) events = broker.update() # Should have TP trigger event tp_events = [e for e in events if e.get('type') == 'sltp_triggered'] assert len(tp_events) == 1 assert tp_events[0]['trigger'] == 'take_profit' def test_liquidation_long(self, broker): """Test liquidation of long position.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Price drops below liquidation broker.update_price('BTC/USDT', 40000.0) events = broker.update() # Should have liquidation event liq_events = [e for e in events if e.get('type') == 'liquidation'] assert len(liq_events) == 1 assert liq_events[0]['symbol'] == 'BTC/USDT' # Position should be closed assert broker.get_position('BTC/USDT') is None # User loses collateral on liquidation assert broker.get_balance() == 90000.0 # Initial - collateral def test_liquidation_short(self, broker): """Test liquidation of short position.""" broker.open_position( symbol='BTC/USDT', side='short', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Price rises above liquidation broker.update_price('BTC/USDT', 60000.0) events = broker.update() liq_events = [e for e in events if e.get('type') == 'liquidation'] assert len(liq_events) == 1 def test_margin_warning(self, broker): """Test margin ratio calculation and warning deduplication.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Liquidation is at ~42105 # Test that margin ratio decreases as price approaches liquidation broker.update_price('BTC/USDT', 45000.0) broker.update() pos = broker.get_position('BTC/USDT') ratio_at_45k = pos.margin_ratio broker.update_price('BTC/USDT', 43000.0) broker.update() pos = broker.get_position('BTC/USDT') ratio_at_43k = pos.margin_ratio # Margin ratio should decrease as price drops toward liquidation assert ratio_at_43k < ratio_at_45k, "Margin ratio should decrease as price drops" # Verify margin ratio is calculated (not default) assert pos.margin_ratio != 100.0 or pos.current_price == pos.entry_price # Warning deduplication: calling update twice quickly shouldn't emit duplicate warnings broker._warning_state.clear() # Reset for test broker._warning_cooldown = timedelta(minutes=5) events1 = broker.update() events2 = broker.update() # If warnings were emitted, second call should be deduplicated warnings1 = [e for e in events1 if e.get('type') == 'margin_warning'] warnings2 = [e for e in events2 if e.get('type') == 'margin_warning'] if len(warnings1) > 0: assert len(warnings2) == 0, "Duplicate warning should be suppressed by cooldown" def test_price_source_tracking(self, broker): """Test that price source is tracked on positions.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0, exchange='kucoin' ) assert pos.price_source == 'kucoin' # Exchange-qualified price update broker.update_price('BTC/USDT', 51000.0, exchange='kucoin') broker.update() pos = broker.get_position('BTC/USDT') assert pos.current_price == 51000.0 def test_reset(self, broker): """Test broker reset.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0 ) broker.reset() assert broker.get_balance() == 100000.0 assert len(broker.get_all_positions()) == 0 def test_state_persistence(self, broker): """Test state save/restore (without data_cache).""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=3.0, current_price=50000.0, exchange='binance' ) # Serialize state state = broker.to_state_dict() # Create new broker and restore new_broker = PaperMarginBroker(initial_balance=50000.0) new_broker.from_state_dict(state) # Verify restoration assert new_broker.get_balance() == 90000.0 pos = new_broker.get_position('BTC/USDT') assert pos is not None assert pos.side == 'long' assert pos.leverage == 3.0 assert pos.price_source == 'binance' class TestMarginCalculations: """Tests for margin calculation edge cases.""" @pytest.fixture def broker(self): return PaperMarginBroker(initial_balance=100000.0) def test_interest_accrual(self, broker): """Test interest accrual over time.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Manually set last_interest_at to simulate time passing pos.last_interest_at = datetime.now(timezone.utc) - timedelta(hours=24) broker.update_price('BTC/USDT', 50000.0) broker.update() pos = broker.get_position('BTC/USDT') # Interest = 40000 * 0.0001 * 24 = 96 USDT assert pos.interest_accrued > 90 def test_effective_leverage_after_add_margin(self, broker): """Test effective leverage changes after adding margin.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, # Initial leverage current_price=50000.0 ) # Position value = 50000 USDT # Initial effective leverage = 50000 / 10000 = 5x broker.add_margin('BTC/USDT', 10000.0) pos = broker.get_position('BTC/USDT') # New effective leverage = 50000 / 20000 = 2.5x effective_leverage = (pos.size * pos.entry_price) / pos.collateral assert 2.4 < effective_leverage < 2.6 def test_zero_leverage_rejected(self, broker): """Test that zero/negative leverage is rejected.""" with pytest.raises(ValueError, match="at least 1x"): broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=0.5, current_price=50000.0 ) def test_invalid_side_rejected(self, broker): """Test that invalid side is rejected.""" with pytest.raises(ValueError, match="must be 'long' or 'short'"): broker.open_position( symbol='BTC/USDT', side='sideways', collateral=10000.0, leverage=2.0, current_price=50000.0 ) def test_zero_collateral_rejected(self, broker): """Test that zero collateral is rejected.""" with pytest.raises(ValueError, match="must be positive"): broker.open_position( symbol='BTC/USDT', side='long', collateral=0, leverage=2.0, current_price=50000.0 ) def test_negative_collateral_rejected(self, broker): """Test that negative collateral is rejected.""" initial_balance = broker.get_balance() with pytest.raises(ValueError, match="must be positive"): broker.open_position( symbol='BTC/USDT', side='long', collateral=-1000.0, leverage=2.0, current_price=50000.0 ) # Balance should be unchanged assert broker.get_balance() == initial_balance def test_negative_reduce_size_rejected(self, broker): """Test that negative reduce_size is rejected.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0 ) pos = broker.get_position('BTC/USDT') original_size = pos.size with pytest.raises(ValueError, match="must be positive"): broker.reduce_position('BTC/USDT', reduce_size=-0.1, current_price=50000.0) # Position should be unchanged pos = broker.get_position('BTC/USDT') assert pos.size == original_size def test_zero_reduce_size_rejected(self, broker): """Test that zero reduce_size is rejected.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=2.0, current_price=50000.0 ) with pytest.raises(ValueError, match="must be positive"): broker.reduce_position('BTC/USDT', reduce_size=0, current_price=50000.0) def test_reduce_position_accrues_interest_first(self, broker): """Test that reduce_position accrues interest before calculating P&L.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Simulate 24 hours passing pos.last_interest_at = datetime.now(timezone.utc) - timedelta(hours=24) # Reduce half the position balance_before = broker.get_balance() broker.reduce_position('BTC/USDT', reduce_size=0.5, current_price=50000.0) # Interest should have been accrued before P&L calculation # borrowed = 40000, rate = 0.0001/hour, 24 hours = 96 USDT interest # Half of that = 48 USDT deducted from P&L remaining_pos = broker.get_position('BTC/USDT') # The remaining interest should be half (the other half was realized) assert remaining_pos.interest_accrued > 40 # Approximately 48 def test_margin_ratio_at_entry_is_100(self, broker): """Test that margin ratio is ~100% at entry price.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) broker.update_price('BTC/USDT', 50000.0) broker.update() pos = broker.get_position('BTC/USDT') # At entry price, margin ratio should be 100% assert 99 < pos.margin_ratio < 101 def test_margin_ratio_approaches_zero_at_liquidation(self, broker): """Test that margin ratio approaches 0% near liquidation.""" pos = broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Price very close to liquidation (liq is ~42105) broker.update_price('BTC/USDT', 42200.0) broker.update() pos = broker.get_position('BTC/USDT') # Should be very low (close to 0%) assert pos.margin_ratio < 15 def test_margin_ratio_above_100_when_profitable(self, broker): """Test that margin ratio exceeds 100% when price moves favorably.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) # Price moves up significantly broker.update_price('BTC/USDT', 55000.0) broker.update() pos = broker.get_position('BTC/USDT') # Should be above 100% (position is healthier than at entry) assert pos.margin_ratio > 100 def test_margin_ratio_declines_from_interest_even_if_price_is_flat(self, broker): """Accrued interest should erode margin health and move liquidation closer.""" broker.open_position( symbol='BTC/USDT', side='long', collateral=10000.0, leverage=5.0, current_price=50000.0 ) pos = broker.get_position('BTC/USDT') initial_liq = pos.liquidation_price # Simulate a long holding period with flat price. pos.last_interest_at = datetime.now(timezone.utc) - timedelta(hours=24) broker.update_price('BTC/USDT', 50000.0) broker.update() pos = broker.get_position('BTC/USDT') assert pos.margin_ratio < 100 assert pos.liquidation_price > initial_liq