769 lines
25 KiB
Python
769 lines
25 KiB
Python
"""
|
|
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
|