Coverage for src/finbot/risk/risk_manager.py: 95%

1import math 

2 

3from finbot.portfolio.portfolio import Portfolio 

4from finbot.risk.models import RiskDecision, RiskLimits 

5from finbot.strategy.core import Signal, SignalDirection 

6 

7 

8class RiskManager: 

9 def __init__(self, limits: RiskLimits): 

10 self.limits = limits 

11 

12 def assess( 

13 self, 

14 signal: Signal, 

15 portfolio: Portfolio, 

16 market_prices: dict[str, float], 

17 ) -> RiskDecision: 

18 self._validate_market_prices( 

19 signal=signal, 

20 portfolio=portfolio, 

21 market_prices=market_prices, 

22 ) 

23 

24 if signal.direction == SignalDirection.LONG: 

25 return self._assess_long( 

26 signal=signal, 

27 portfolio=portfolio, 

28 market_prices=market_prices, 

29 ) 

30 

31 if signal.direction == SignalDirection.SHORT: 

32 return self._assess_short( 

33 signal=signal, 

34 portfolio=portfolio, 

35 market_prices=market_prices, 

36 ) 

37 

38 if signal.direction == SignalDirection.FLAT: 

39 return RiskDecision.reject("FLAT signal does not require an order.") 

40 

41 return RiskDecision.reject(f"Unsupported signal direction: {signal.direction}") 

42 

43 def _assess_long( 

44 self, 

45 *, 

46 signal: Signal, 

47 portfolio: Portfolio, 

48 market_prices: dict[str, float], 

49 ) -> RiskDecision: 

50 nav = portfolio.total_value(market_prices) 

51 

52 if nav <= 0: 

53 return RiskDecision.reject("Portfolio NAV must be greater than zero.") 

54 

55 market_price = market_prices[signal.symbol] 

56 

57 position = portfolio.positions.get(signal.symbol) 

58 current_symbol_value = (0.0 if position is None else position.market_value(market_price)) 

59 

60 is_new_position = position is None or math.isclose(position.quantity, 0.0, abs_tol=1e-12) 

61 

62 if is_new_position: 

63 open_positions = sum( 

64 (1 

65 for current_position in portfolio.positions.values() 

66 if current_position.quantity > 0 

67 ), 

68 ) 

69 

70 if open_positions >= self.limits.max_open_positions: 

71 return RiskDecision.reject("Maximum number of open positions reached.") 

72 

73 gross_exposure = portfolio.positions_value(market_prices) 

74 

75 max_position_value = nav * self.limits.max_position_pct_nav 

76 max_symbol_value = nav * self.limits.max_symbol_exposure 

77 max_gross_value = nav * self.limits.max_gross_exposure 

78 

79 position_capacity = max_position_value - current_symbol_value 

80 symbol_capacity = max_symbol_value - current_symbol_value 

81 gross_capacity = max_gross_value - gross_exposure 

82 cash_capacity = portfolio.cash - self.limits.minimum_cash_reserve 

83 

84 capacities = { 

85 "max position % NAV": position_capacity, 

86 "max symbol exposure": symbol_capacity, 

87 "max gross exposure": gross_capacity, 

88 "minimum cash reserve": cash_capacity, 

89 "max order value": self.limits.max_order_value, 

90 } 

91 

92 order_value_budget = min(capacities.values()) 

93 

94 if order_value_budget <= 0: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 limiting_reason = min( 

96 (reason for reason, capacity in capacities.items() if capacity <= 0), 

97 key=lambda reason: capacities[reason], 

98 ) 

99 return RiskDecision.reject(f"Order rejected by {limiting_reason}.") 

100 

101 if signal.strength is not None: 

102 order_value_budget *= signal.strength 

103 

104 if order_value_budget <= 0: 

105 return RiskDecision.reject("Signal strength produced a zero order value.") 

106 

107 return RiskDecision.accept(order_value_budget=order_value_budget) 

108 

109 def _assess_short( 

110 self, 

111 *, 

112 signal: Signal, 

113 portfolio: Portfolio, 

114 market_prices: dict[str, float], 

115 ): 

116 position = portfolio.positions.get(signal.symbol) 

117 

118 if position is None or position.quantity <= 0: 

119 return RiskDecision.reject(f"No open long position for {signal.symbol} to reduce.") 

120 

121 position_value = position.market_value(market_prices[signal.symbol]) 

122 

123 order_value_budget = min(position_value, self.limits.max_order_value) 

124 

125 if signal.strength is not None: 

126 order_value_budget *= signal.strength 

127 

128 if order_value_budget <= 0: 

129 return RiskDecision.reject("No sellable position value remains.") 

130 

131 return RiskDecision.accept(order_value_budget=order_value_budget) 

132 

133 @staticmethod 

134 def _validate_market_prices( 

135 *, 

136 signal: Signal, 

137 portfolio: Portfolio, 

138 market_prices: dict[str, float], 

139 ) -> None: 

140 required_symbols = { 

141 symbol 

142 for symbol, position in portfolio.positions.items() 

143 if position.quantity > 0 

144 } 

145 

146 required_symbols.add(signal.symbol) 

147 

148 for symbol in required_symbols: 

149 if symbol not in market_prices: 

150 raise ValueError(f"Missing market price for {symbol}.") 

151 

152 price = market_prices[symbol] 

153 

154 if not math.isfinite(price) or price <= 0: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 raise ValueError(f"Market price for {symbol} must be a positive finite number.")