Coverage for adrs/performance/metric/drawdown.py: 97%

1import numpy as np 

2import polars as pl 

3from typing import Any, cast, override 

4from datetime import datetime, timedelta, timezone 

5 

6from .metric import Metrics 

7 

8 

9class Drawdown(Metrics[dict[str, Any]]): 

10 def __init__( 

11 self, 

12 num_periods: int = 365, 

13 period: timedelta = timedelta(days=1), 

14 interval: timedelta | None = None, 

15 ): 

16 self.num_periods = num_periods # 365 trading days in a year (crypto) 

17 self.period = period # 1 day as a base measurement 

18 self.interval = interval # grid spacing; inferred from df when None 

19 

20 @override 

21 def compute(self, df): 

22 drawdown_df = df.with_columns( 

23 (pl.col("equity") + 1.0).alias("geo_equity"), 

24 (pl.col("equity") + 1.0).cum_max().alias("geo_equity_cummax"), 

25 pl.col("equity").cum_max().alias("equity_cummax"), 

26 ).with_columns((pl.col("equity") - pl.col("equity_cummax")).alias("drawdown")) 

27 

28 mdd_df = drawdown_df.filter(pl.col("drawdown") == pl.col("drawdown").min()) 

29 

30 mdd = cast(float, mdd_df["drawdown"][0]) 

31 mdd_cummax = cast(float, mdd_df["equity_cummax"][0]) 

32 mdd_geo = cast(float, mdd_df["geo_equity"][0]) 

33 mdd_geo_cummax = cast(float, mdd_df["geo_equity_cummax"][0]) 

34 mdd_pct = ( 

35 0.0 if mdd_cummax == 0 else (mdd_geo - mdd_geo_cummax) / mdd_geo_cummax 

36 ) 

37 

38 mdd_start_time_df = drawdown_df.join( 

39 drawdown_df.filter(pl.col("drawdown").shift(1).ge(0)).select( 

40 pl.col("start_time"), pl.col("start_time").alias("mdd_start_time") 

41 ), 

42 how="left", 

43 left_on="start_time", 

44 right_on="start_time", 

45 ).select( 

46 pl.col("start_time"), 

47 pl.col("drawdown"), 

48 pl.col("mdd_start_time").forward_fill().fill_null(pl.col("start_time")), 

49 ) 

50 mdd_recover_df = drawdown_df.filter(pl.col("equity") > mdd_cummax) 

51 mdd_time = cast(datetime, mdd_df["start_time"][0]) 

52 mdd_start_time = cast( 

53 datetime, 

54 mdd_start_time_df.filter(pl.col("start_time") == mdd_time)[ 

55 "mdd_start_time" 

56 ][0], 

57 ) 

58 mdd_recover_time = cast( 

59 datetime, 

60 df["start_time"][-1] 

61 if mdd_recover_df.is_empty() 

62 else mdd_recover_df["start_time"][0], 

63 ) 

64 mdd_duration = ( 

65 df["start_time"][-1] - mdd_time 

66 if mdd_recover_df.is_empty() 

67 else cast(datetime, mdd_recover_df["start_time"][0]) - mdd_start_time 

68 ) 

69 

70 interval = ( 

71 self.interval 

72 if self.interval is not None 

73 else df["start_time"].diff().last() 

74 ) 

75 if not isinstance(interval, timedelta): 

76 raise Exception("performance_df does not have an interval in between data") 

77 mean = cast(float, df["pnl"].mean()) 

78 

79 calmar_ratio = ( 

80 mean * self.num_periods * self.period / interval / abs(mdd) 

81 if mdd != 0 and not np.isnan(mdd) 

82 else np.float64(0.0) 

83 ) 

84 

85 return { 

86 "max_drawdown": mdd, 

87 "max_drawdown_percentage": mdd_pct, 

88 "max_drawdown_start_date": mdd_start_time.replace(tzinfo=timezone.utc), 

89 "max_drawdown_end_date": mdd_time.replace(tzinfo=timezone.utc), 

90 "max_drawdown_recover_date": mdd_recover_time.replace(tzinfo=timezone.utc), 

91 "max_drawdown_max_duration_in_days": mdd_duration.total_seconds() 

92 / (60 * 60 * 24), 

93 "calmar_ratio": calmar_ratio, 

94 }