diff options
Diffstat (limited to 'services/order-executor/tests/test_risk_manager.py')
| -rw-r--r-- | services/order-executor/tests/test_risk_manager.py | 198 |
1 files changed, 183 insertions, 15 deletions
diff --git a/services/order-executor/tests/test_risk_manager.py b/services/order-executor/tests/test_risk_manager.py index efabe73..3d5175b 100644 --- a/services/order-executor/tests/test_risk_manager.py +++ b/services/order-executor/tests/test_risk_manager.py @@ -7,7 +7,7 @@ from shared.models import OrderSide, Position, Signal from order_executor.risk_manager import RiskManager -def make_signal(side: OrderSide, price: str, quantity: str, symbol: str = "BTC/USDT") -> Signal: +def make_signal(side: OrderSide, price: str, quantity: str, symbol: str = "AAPL") -> Signal: return Signal( strategy="test", symbol=symbol, @@ -93,7 +93,7 @@ def test_risk_check_rejects_insufficient_balance(): def test_trailing_stop_set_and_trigger(): """Trailing stop should trigger when price drops below stop level.""" rm = make_risk_manager(trailing_stop_pct="5") - rm.set_trailing_stop("BTC/USDT", Decimal("100")) + rm.set_trailing_stop("AAPL", Decimal("100")) signal = make_signal(side=OrderSide.BUY, price="94", quantity="0.01") result = rm.check(signal, balance=Decimal("10000"), positions={}, daily_pnl=Decimal("0")) @@ -104,10 +104,10 @@ def test_trailing_stop_set_and_trigger(): def test_trailing_stop_updates_highest_price(): """Trailing stop should track the highest price seen.""" rm = make_risk_manager(trailing_stop_pct="5") - rm.set_trailing_stop("BTC/USDT", Decimal("100")) + rm.set_trailing_stop("AAPL", Decimal("100")) # Price rises to 120 => stop at 114 - rm.update_price("BTC/USDT", Decimal("120")) + rm.update_price("AAPL", Decimal("120")) # Price at 115 is above stop (114), should be allowed signal = make_signal(side=OrderSide.BUY, price="115", quantity="0.01") @@ -124,7 +124,7 @@ def test_trailing_stop_updates_highest_price(): def test_trailing_stop_not_triggered_above_stop(): """Trailing stop should not trigger when price is above stop level.""" rm = make_risk_manager(trailing_stop_pct="5") - rm.set_trailing_stop("BTC/USDT", Decimal("100")) + rm.set_trailing_stop("AAPL", Decimal("100")) # Price at 96 is above stop (95), should be allowed signal = make_signal(side=OrderSide.BUY, price="96", quantity="0.01") @@ -140,11 +140,11 @@ def test_max_open_positions_check(): rm = make_risk_manager(max_open_positions=2) positions = { - "BTC/USDT": make_position("BTC/USDT", "1", "100", "100"), - "ETH/USDT": make_position("ETH/USDT", "10", "50", "50"), + "AAPL": make_position("AAPL", "1", "100", "100"), + "MSFT": make_position("MSFT", "10", "50", "50"), } - signal = make_signal(side=OrderSide.BUY, price="10", quantity="1", symbol="SOL/USDT") + signal = make_signal(side=OrderSide.BUY, price="10", quantity="1", symbol="TSLA") result = rm.check(signal, balance=Decimal("10000"), positions=positions, daily_pnl=Decimal("0")) assert result.allowed is False assert result.reason == "Max open positions reached" @@ -158,14 +158,14 @@ def test_volatility_calculation(): rm = make_risk_manager(volatility_lookback=5) # No history yet - assert rm.get_volatility("BTC/USDT") is None + assert rm.get_volatility("AAPL") is None # Feed prices prices = [100, 102, 98, 105, 101] for p in prices: - rm.update_price("BTC/USDT", Decimal(str(p))) + rm.update_price("AAPL", Decimal(str(p))) - vol = rm.get_volatility("BTC/USDT") + vol = rm.get_volatility("AAPL") assert vol is not None assert vol > 0 @@ -177,9 +177,9 @@ def test_position_size_with_volatility_scaling(): # Feed volatile prices prices = [100, 120, 80, 130, 70] for p in prices: - rm.update_price("BTC/USDT", Decimal(str(p))) + rm.update_price("AAPL", Decimal(str(p))) - size = rm.calculate_position_size("BTC/USDT", Decimal("10000")) + size = rm.calculate_position_size("AAPL", Decimal("10000")) base = Decimal("10000") * Decimal("0.1") # High volatility should reduce size below base @@ -192,9 +192,177 @@ def test_position_size_without_scaling(): prices = [100, 120, 80, 130, 70] for p in prices: - rm.update_price("BTC/USDT", Decimal(str(p))) + rm.update_price("AAPL", Decimal(str(p))) - size = rm.calculate_position_size("BTC/USDT", Decimal("10000")) + size = rm.calculate_position_size("AAPL", Decimal("10000")) base = Decimal("10000") * Decimal("0.1") assert size == base + + +# --- Portfolio exposure tests --- + + +def test_portfolio_exposure_check_passes(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + max_portfolio_exposure=0.8, + ) + positions = { + "AAPL": Position( + symbol="AAPL", + quantity=Decimal("0.01"), + avg_entry_price=Decimal("50000"), + current_price=Decimal("50000"), + ) + } + result = rm.check_portfolio_exposure(positions, Decimal("10000")) + assert result.allowed # 500/10000 = 5% < 80% + + +def test_portfolio_exposure_check_rejects(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + max_portfolio_exposure=0.3, + ) + positions = { + "AAPL": Position( + symbol="AAPL", + quantity=Decimal("1"), + avg_entry_price=Decimal("50000"), + current_price=Decimal("50000"), + ) + } + result = rm.check_portfolio_exposure(positions, Decimal("10000")) + assert not result.allowed # 50000/10000 = 500% > 30% + + +def test_correlation_calculation(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + ) + # Feed identical price histories — correlation should be ~1.0 + for i in range(20): + rm.update_price("A", Decimal(str(100 + i))) + rm.update_price("B", Decimal(str(100 + i))) + corr = rm.calculate_correlation("A", "B") + assert corr is not None + assert corr > 0.9 + + +def test_var_calculation(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + ) + for i in range(30): + rm.update_price("AAPL", Decimal(str(100 + (i % 5) - 2))) + positions = { + "AAPL": Position( + symbol="AAPL", + quantity=Decimal("1"), + avg_entry_price=Decimal("100"), + current_price=Decimal("100"), + ) + } + var = rm.calculate_portfolio_var(positions, Decimal("10000")) + assert var >= 0 # Non-negative + + +# --- Drawdown position scaling tests --- + + +def test_drawdown_position_scale_full(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + drawdown_reduction_threshold=0.1, + drawdown_halt_threshold=0.2, + ) + rm.update_balance(Decimal("10000")) + scale = rm.get_position_scale(Decimal("10000")) + assert scale == 1.0 # No drawdown + + +def test_drawdown_position_scale_reduced(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + drawdown_reduction_threshold=0.1, + drawdown_halt_threshold=0.2, + ) + rm.update_balance(Decimal("10000")) + scale = rm.get_position_scale(Decimal("8500")) # 15% drawdown (between 10% and 20%) + assert 0.25 < scale < 1.0 + + +def test_drawdown_halt(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + drawdown_reduction_threshold=0.1, + drawdown_halt_threshold=0.2, + ) + rm.update_balance(Decimal("10000")) + scale = rm.get_position_scale(Decimal("7500")) # 25% drawdown + assert scale == 0.0 + + +def test_consecutive_losses_pause(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + max_consecutive_losses=3, + loss_pause_minutes=60, + ) + rm.record_trade_result(False) + rm.record_trade_result(False) + assert not rm.is_paused() + rm.record_trade_result(False) # 3rd loss + assert rm.is_paused() + + +def test_consecutive_losses_reset_on_win(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + max_consecutive_losses=3, + ) + rm.record_trade_result(False) + rm.record_trade_result(False) + rm.record_trade_result(True) # Win resets counter + rm.record_trade_result(False) + assert not rm.is_paused() # Only 1 consecutive loss + + +def test_drawdown_check_rejects_in_check(): + rm = RiskManager( + max_position_size=Decimal("0.5"), + stop_loss_pct=Decimal("5"), + daily_loss_limit_pct=Decimal("10"), + drawdown_halt_threshold=0.15, + ) + rm.update_balance(Decimal("10000")) + signal = Signal( + strategy="test", + symbol="AAPL", + side=OrderSide.BUY, + price=Decimal("50000"), + quantity=Decimal("0.01"), + reason="test", + ) + result = rm.check(signal, Decimal("8000"), {}, Decimal("0")) # 20% dd > 15% + assert not result.allowed + assert "halted" in result.reason.lower() |