Coverage for src / monte_neo / metrics / drawdown.py: 66%

1"""Drawdown metrics. 

2 

3Calculates maximum drawdown and related metrics. 

4""" 

5 

6from __future__ import annotations 

7 

8import numpy as np 

9 

10from monte_neo.utils.logger import get_logger 

11 

12logger = get_logger(__name__) 

13 

14 

15class DrawdownMetric: 

16 """Drawdown calculator.""" 

17 

18 def calculate_max(self, equity: list[float] | np.ndarray) -> float: 

19 """Calculate maximum drawdown. 

20 

21 Max Drawdown = (Peak - Trough) / Peak 

22 

23 Args: 

24 equity: Equity curve values. 

25 

26 Returns: 

27 Maximum drawdown as decimal (e.g., 0.20 = 20%). 

28 """ 

29 if not len(equity): 

30 return 0.0 

31 

32 equity = np.array(equity) 

33 

34 # Running maximum 

35 running_max = np.maximum.accumulate(equity) 

36 

37 # Drawdown at each point 

38 drawdowns = (running_max - equity) / running_max 

39 

40 return float(np.max(drawdowns)) 

41 

42 def calculate_avg(self, equity: list[float] | np.ndarray) -> float: 

43 """Calculate average drawdown. 

44 

45 Args: 

46 equity: Equity curve values. 

47 

48 Returns: 

49 Average drawdown as decimal. 

50 """ 

51 if not len(equity): 

52 return 0.0 

53 

54 equity = np.array(equity) 

55 running_max = np.maximum.accumulate(equity) 

56 drawdowns = (running_max - equity) / running_max 

57 

58 # Only count non-zero drawdowns 

59 dd_values = drawdowns[drawdowns > 0] 

60 

61 if len(dd_values) == 0: 

62 return 0.0 

63 

64 return float(np.mean(dd_values)) 

65 

66 def calculate_duration(self, equity: list[float] | np.ndarray) -> int: 

67 """Calculate maximum drawdown duration. 

68 

69 Args: 

70 equity: Equity curve values. 

71 

72 Returns: 

73 Maximum number of periods in drawdown. 

74 """ 

75 if not len(equity): 

76 return 0 

77 

78 equity = np.array(equity) 

79 running_max = np.maximum.accumulate(equity) 

80 

81 # Find where we're in drawdown 

82 in_drawdown = equity < running_max 

83 

84 # Count consecutive drawdown periods 

85 max_duration = 0 

86 current_duration = 0 

87 

88 for is_dd in in_drawdown: 

89 if is_dd: 

90 current_duration += 1 

91 max_duration = max(max_duration, current_duration) 

92 else: 

93 current_duration = 0 

94 

95 return max_duration 

96 

97 def get_drawdown_curve( 

98 self, 

99 equity: list[float] | np.ndarray, 

100 ) -> np.ndarray: 

101 """Get the full drawdown curve. 

102 

103 Args: 

104 equity: Equity curve values. 

105 

106 Returns: 

107 Array of drawdown values at each point. 

108 """ 

109 if not len(equity): 

110 return np.array([]) 

111 

112 equity = np.array(equity) 

113 running_max = np.maximum.accumulate(equity) 

114 drawdowns = (running_max - equity) / running_max 

115 

116 return drawdowns 

117 

118 def get_underwater_curve( 

119 self, 

120 equity: list[float] | np.ndarray, 

121 ) -> np.ndarray: 

122 """Get underwater curve (negative drawdowns). 

123 

124 Args: 

125 equity: Equity curve values. 

126 

127 Returns: 

128 Array of negative drawdown values. 

129 """ 

130 return -self.get_drawdown_curve(equity) 

131 

132 def analyze_drawdowns( 

133 self, 

134 equity: list[float] | np.ndarray, 

135 n_worst: int = 5, 

136 ) -> list[dict]: 

137 """Analyze worst drawdown periods. 

138 

139 Args: 

140 equity: Equity curve values. 

141 n_worst: Number of worst drawdowns to return. 

142 

143 Returns: 

144 List of drawdown info dictionaries. 

145 """ 

146 if not len(equity): 

147 return [] 

148 

149 equity = np.array(equity) 

150 running_max = np.maximum.accumulate(equity) 

151 drawdowns = (running_max - equity) / running_max 

152 

153 # Find drawdown periods 

154 periods = [] 

155 in_dd = False 

156 start_idx = 0 

157 peak_val = 0 

158 

159 for i, (dd, eq, peak) in enumerate(zip(drawdowns, equity, running_max)): 

160 if dd > 0 and not in_dd: 

161 # Start of drawdown 

162 in_dd = True 

163 start_idx = i 

164 peak_val = peak 

165 elif dd == 0 and in_dd: 

166 # End of drawdown 

167 in_dd = False 

168 periods.append( 

169 { 

170 "start_idx": start_idx, 

171 "end_idx": i, 

172 "duration": i - start_idx, 

173 "max_drawdown": float(np.max(drawdowns[start_idx:i])), 

174 "peak_value": float(peak_val), 

175 "trough_value": float(np.min(equity[start_idx:i])), 

176 } 

177 ) 

178 

179 # Handle ongoing drawdown 

180 if in_dd: 

181 periods.append( 

182 { 

183 "start_idx": start_idx, 

184 "end_idx": len(equity) - 1, 

185 "duration": len(equity) - start_idx, 

186 "max_drawdown": float(np.max(drawdowns[start_idx:])), 

187 "peak_value": float(peak_val), 

188 "trough_value": float(np.min(equity[start_idx:])), 

189 } 

190 ) 

191 

192 # Sort by max_drawdown and return worst N 

193 periods.sort(key=lambda x: x["max_drawdown"], reverse=True) 

194 return periods[:n_worst]