1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
"""Tests for RiskManager."""
from decimal import Decimal
from shared.models import OrderSide, Signal
from order_executor.risk_manager import RiskManager
def make_signal(side: OrderSide, price: str, quantity: str, symbol: str = "BTC/USDT") -> Signal:
return Signal(
strategy="test",
symbol=symbol,
side=side,
price=Decimal(price),
quantity=Decimal(quantity),
reason="test signal",
)
def make_risk_manager(
max_position_size: str = "0.1",
stop_loss_pct: str = "5.0",
daily_loss_limit_pct: str = "10.0",
) -> RiskManager:
return RiskManager(
max_position_size=Decimal(max_position_size),
stop_loss_pct=Decimal(stop_loss_pct),
daily_loss_limit_pct=Decimal(daily_loss_limit_pct),
)
def test_risk_check_passes_normal_order():
"""Small BUY order with enough balance should be allowed."""
rm = make_risk_manager()
signal = make_signal(side=OrderSide.BUY, price="100", quantity="0.5")
# cost = 50, balance = 10000, position_value = 0 => (0+50)/10000 = 0.5% < 10%
result = rm.check(signal, balance=Decimal("10000"), positions={}, daily_pnl=Decimal("0"))
assert result.allowed is True
assert result.reason == "OK"
def test_risk_check_rejects_exceeding_position_size():
"""5 BTC at $50,000 = $250,000 order cost on $10,000,000 balance exceeds 10% limit."""
rm = make_risk_manager(max_position_size="0.1")
signal = make_signal(side=OrderSide.BUY, price="50000", quantity="5")
# cost = 250000, balance = 1000000 => 250000/1000000 = 25% > 10%
# balance is sufficient (250000 < 1000000) but position size is exceeded
result = rm.check(signal, balance=Decimal("1000000"), positions={}, daily_pnl=Decimal("0"))
assert result.allowed is False
assert result.reason == "Position size exceeded"
def test_risk_check_rejects_daily_loss_exceeded():
"""Daily PnL of -1100 on 10000 balance = -11%, exceeding -10% limit."""
rm = make_risk_manager(daily_loss_limit_pct="10.0")
signal = make_signal(side=OrderSide.BUY, price="100", quantity="0.1")
result = rm.check(signal, balance=Decimal("10000"), positions={}, daily_pnl=Decimal("-1100"))
assert result.allowed is False
assert result.reason == "Daily loss limit exceeded"
def test_risk_check_rejects_insufficient_balance():
"""Order cost of 500 exceeds available balance of 100."""
rm = make_risk_manager()
signal = make_signal(side=OrderSide.BUY, price="100", quantity="5")
# cost = 500, balance = 100
result = rm.check(signal, balance=Decimal("100"), positions={}, daily_pnl=Decimal("0"))
assert result.allowed is False
assert result.reason == "Insufficient balance"
|
