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

1import math 

2 

3from finbot.portfolio.portfolio import Portfolio 

4from finbot.risk import RiskDecision, RiskLimits 

5from finbot.strategy.signal import Signal, SignalDirection 

6 

7 

8class RiskManager: 

9 def __init__(self, limits: RiskLimits): 

10 self.limits = limits 

11 

12 def assess( 

13 self, 

14 *, 

15 signal: Signal, 

16 portfolio: Portfolio, 

17 market_prices: dict[str, float], 

18 ) -> RiskDecision: 

19 self._validate_market_prices( 

20 signal=signal, 

21 portfolio=portfolio, 

22 market_prices=market_prices, 

23 ) 

24 

25 if signal.direction == SignalDirection.LONG: 

26 return self._assess_long( 

27 signal=signal, 

28 portfolio=portfolio, 

29 market_prices=market_prices, 

30 ) 

31 

32 if signal.direction == SignalDirection.SHORT: 

33 return self._assess_short( 

34 signal=signal, 

35 portfolio=portfolio, 

36 market_prices=market_prices, 

37 ) 

38 

39 if signal.direction == SignalDirection.FLAT: 

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

41 

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

43 

44 def _assess_long( 

45 self, 

46 *, 

47 signal: Signal, 

48 portfolio: Portfolio, 

49 market_prices: dict[str, float], 

50 ) -> RiskDecision: 

51 nav = portfolio.total_value(market_prices) 

52 

53 if nav <= 0: 

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

55 

56 market_price = market_prices[signal.symbol] 

57 

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

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

60 

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

62 

63 if is_new_position: 

64 open_positions = sum( 

65 (1 

66 for current_position in portfolio.positions.values() 

67 if current_position.quantity > 0 

68 ), 

69 ) 

70 

71 if open_positions >= self.limits.max_open_positions: 

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

73 

74 gross_exposure = portfolio.positions_value(market_prices) 

75 

76 max_position_value = nav * self.limits.max_position_pct_nav 

77 max_symbol_value = nav * self.limits.max_symbol_exposure 

78 max_gross_value = nav * self.limits.max_gross_exposure 

79 

80 position_capacity = max_position_value - current_symbol_value 

81 symbol_capacity = max_symbol_value - current_symbol_value 

82 gross_capacity = max_gross_value - gross_exposure 

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

84 

85 capacities = { 

86 "max position % NAV": position_capacity, 

87 "max symbol exposure": symbol_capacity, 

88 "max gross exposure": gross_capacity, 

89 "minimum cash reserve": cash_capacity, 

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

91 } 

92 

93 order_value_budget = min(capacities.values()) 

94 

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

96 limiting_reason = min( 

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

98 key=lambda reason: capacities[reason], 

99 ) 

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

101 

102 if signal.strength is not None: 

103 order_value_budget *= signal.strength 

104 

105 if order_value_budget <= 0: 

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

107 

108 return RiskDecision.accept(order_value_budget=order_value_budget) 

109 

110 def _assess_short( 

111 self, 

112 *, 

113 signal: Signal, 

114 portfolio: Portfolio, 

115 market_prices: dict[str, float], 

116 ): 

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

118 

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

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

121 

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

123 

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

125 

126 if signal.strength is not None: 

127 order_value_budget *= signal.strength 

128 

129 if order_value_budget <= 0: 

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

131 

132 return RiskDecision.accept(order_value_budget=order_value_budget) 

133 

134 @staticmethod 

135 def _validate_market_prices( 

136 *, 

137 signal: Signal, 

138 portfolio: Portfolio, 

139 market_prices: dict[str, float], 

140 ) -> None: 

141 required_symbols = { 

142 symbol 

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

144 if position.quantity > 0 

145 } 

146 

147 required_symbols.add(signal.symbol) 

148 

149 for symbol in required_symbols: 

150 if symbol not in market_prices: 

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

152 

153 price = market_prices[symbol] 

154 

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

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