Coverage for adrs/performance/metric/trade.py: 35%

1import polars as pl 

2from datetime import timedelta 

3from typing import override, cast 

4 

5from .metric import Metrics 

6 

7 

8class Trade(Metrics[dict[str, float | int]]): 

9 @override 

10 def compute(self, df): 

11 # Make the dataframe that will be used to compute the metrics 

12 trades_df = ( 

13 df.select( 

14 pl.col("start_time"), 

15 pl.col("signal"), 

16 pl.col("prev_signal"), 

17 pl.col("price"), 

18 pl.col("trade"), 

19 ) 

20 .filter(pl.col("signal") != pl.col("prev_signal")) 

21 .with_columns( 

22 pl.col("price").shift(1).alias("prev_price"), 

23 pl.col("start_time").diff().alias("holding_time"), 

24 ) 

25 .filter(pl.col("prev_signal") != 0) 

26 ) 

27 

28 # Reusable polars expressions for filtering 

29 close_long_cond = ((pl.col("prev_signal") > 0) & (pl.col("signal") == 0)) | ( 

30 (pl.col("prev_signal") > 0) & (pl.col("signal") < 0) 

31 ) 

32 close_short_cond = ((pl.col("prev_signal") < 0) & (pl.col("signal") == 0)) | ( 

33 (pl.col("prev_signal") < 0) & (pl.col("signal") > 0) 

34 ) 

35 close_long_pnl_expr = pl.col("price") - pl.col("prev_price") 

36 close_short_pnl_expr = pl.col("prev_price") - pl.col("price") 

37 

38 # Calculate number of win & lose trades 

39 win_trades = ( 

40 trades_df.filter(close_long_cond & (close_long_pnl_expr > 0))["trade"] 

41 .abs() 

42 .sum() 

43 + trades_df.filter(close_short_cond & (close_short_pnl_expr > 0))["trade"] 

44 .abs() 

45 .sum() 

46 ) 

47 lose_trades = ( 

48 trades_df.filter(close_long_cond & (close_long_pnl_expr <= 0))["trade"] 

49 .abs() 

50 .sum() 

51 + trades_df.filter(close_short_cond & (close_short_pnl_expr <= 0))["trade"] 

52 .abs() 

53 .sum() 

54 ) 

55 

56 avg_holding_time = cast(timedelta | None, trades_df["holding_time"].mean()) 

57 max_holding_time = cast(timedelta | None, trades_df["holding_time"].max()) 

58 

59 # Calculate win & loss streaks 

60 streak_df = ( 

61 trades_df.with_columns( 

62 pl.when( 

63 (close_long_cond & (close_long_pnl_expr > 0)) 

64 | (close_short_cond & (close_short_pnl_expr > 0)) 

65 ) 

66 .then(pl.lit("win")) 

67 .otherwise(pl.lit("loss")) 

68 .alias("result") 

69 ) 

70 .with_columns(pl.col("result").rle_id().alias("streak_id")) 

71 .group_by(["streak_id", "result"]) 

72 .agg(pl.count().alias("streak_length")) 

73 ) 

74 

75 win_streak = ( 

76 streak_df.filter(pl.col("result") == "win") 

77 .select(pl.col("streak_length").max()) 

78 .item() 

79 ) or 0 

80 lose_streak = ( 

81 streak_df.filter(pl.col("result") == "loss") 

82 .select(pl.col("streak_length").max()) 

83 .item() 

84 ) or 0 

85 

86 return { 

87 "largest_win": cast(float, df["pnl"].max()), 

88 "largest_loss": cast(float, df["pnl"].min()), 

89 "num_datapoints": df.shape[0], 

90 "num_trades": round(df["trade"].abs().sum()), 

91 "avg_holding_time_in_seconds": avg_holding_time.total_seconds() 

92 if avg_holding_time is not None 

93 else 0.0, 

94 "max_holding_time_in_seconds": max_holding_time.total_seconds() 

95 if max_holding_time is not None 

96 else 0.0, 

97 "long_trades": round(df.filter(pl.col("trade") > 0)["trade"].sum()), 

98 "short_trades": round(df.filter(pl.col("trade") < 0)["trade"].abs().sum()), 

99 "win_trades": round(win_trades), 

100 "lose_trades": round(lose_trades), 

101 "win_streak": win_streak, 

102 "lose_streak": lose_streak, 

103 "win_rate": win_trades / (win_trades + lose_trades) 

104 if (win_trades + lose_trades) > 0 

105 else 0.0, 

106 }