Coverage for adrs/report/alpha.py: 0%

1import io 

2import inspect 

3import polars as pl 

4from datetime import datetime, timedelta 

5from pydantic import BaseModel, TypeAdapter 

6from typing import Self, cast, Unpack, NotRequired, TypedDict 

7 

8from adrs import json_utils as json 

9from adrs.alpha import Alpha 

10from adrs.data import Datamap 

11from adrs.types import Performance 

12from adrs.performance import Evaluator 

13from adrs.tests import Sensitivity, SensitivityParameter 

14 

15type Params = dict[str, float | int | timedelta] 

16 

17 

18class SensitivitySharpeRatioSummary(BaseModel): 

19 best_param: float 

20 mean: float 

21 median: float 

22 std: float 

23 min: float 

24 max: float 

25 p25: float 

26 p75: float 

27 num_negative: int 

28 num_positive: int 

29 total_permutations: int 

30 score: float 

31 

32 @classmethod 

33 def compute( 

34 cls, performance: Performance, results: list[tuple[Params, Performance]] 

35 ) -> Self: 

36 df = pl.DataFrame(map(lambda t: t[1], results)) 

37 

38 best_param = performance.sharpe_ratio 

39 mean = cast(float, df["sharpe_ratio"].mean() or 0.0) 

40 std = cast(float, df["sharpe_ratio"].std() or 0.0) 

41 num_positive = df.filter(pl.col("sharpe_ratio") > 0).shape[0] 

42 total_permutations = df.shape[0] 

43 # fmt: off 

44 score = ( 

45 ((1 / std) if std != 0 else 0) * 0.4 

46 + ((mean / best_param) if best_param != 0 else 0) * 0.3 

47 + ((num_positive / total_permutations) if total_permutations != 0 else 0) * 0.3 

48 ) 

49 # fmt: on 

50 

51 return cls( 

52 best_param=best_param, 

53 mean=mean, 

54 median=cast(float, df["sharpe_ratio"].median() or 0.0), 

55 std=std, 

56 min=cast(float, df["sharpe_ratio"].min() or 0.0), 

57 max=cast(float, df["sharpe_ratio"].max() or 0.0), 

58 p25=df["sharpe_ratio"].quantile(0.25) or 0.0, 

59 p75=df["sharpe_ratio"].quantile(0.75) or 0.0, 

60 num_negative=df.filter(pl.col("sharpe_ratio") < 0).shape[0], 

61 num_positive=num_positive, 

62 total_permutations=total_permutations, 

63 score=score, 

64 ) 

65 

66 

67class AlphaReportV1Performance(BaseModel): 

68 performance: Performance 

69 performance_df: pl.DataFrame 

70 sensitivity: list[tuple[Params, Performance]] | None 

71 sensitivity_sr_summary: SensitivitySharpeRatioSummary | None 

72 

73 class Config: 

74 arbitrary_types_allowed = True 

75 

76 

77class AlphaBacktestArgsWithoutDates(TypedDict): 

78 evaluator: Evaluator 

79 base_asset: str 

80 datamap: Datamap 

81 fees: float 

82 data_df: NotRequired[pl.DataFrame] 

83 price_shift: NotRequired[int] 

84 output_columns: NotRequired[list[pl.Expr]] 

85 

86 

87class AlphaReportV1(BaseModel): 

88 alpha_id: str 

89 params: Params 

90 back: AlphaReportV1Performance 

91 forward: AlphaReportV1Performance 

92 sensitivity_params: dict[str, SensitivityParameter] | None 

93 

94 @staticmethod 

95 def version() -> str: 

96 return "alpha-report-v1" 

97 

98 @classmethod 

99 def compute( 

100 cls, 

101 alpha: Alpha, 

102 B_start: datetime, 

103 B_end: datetime, 

104 F_start: datetime, 

105 F_end: datetime, 

106 sensitivity: Sensitivity | None = None, 

107 **kwargs: Unpack[AlphaBacktestArgsWithoutDates], 

108 ) -> Self: 

109 # get alpha params 

110 params = { 

111 name: getattr(alpha, name) 

112 for name in inspect.signature(alpha.__init__).parameters.keys() 

113 } 

114 

115 # run backtest 

116 back, back_df = alpha.backtest(start_time=B_start, end_time=B_end, **kwargs) 

117 back_sensitivity = ( 

118 [ 

119 (x[0], x[1]) 

120 for x in sensitivity.test(start_time=B_start, end_time=B_end, **kwargs) 

121 ] 

122 if sensitivity is not None 

123 else None 

124 ) 

125 

126 # run forward test 

127 forward, forward_df = alpha.backtest( 

128 start_time=F_start, end_time=F_end, **kwargs 

129 ) 

130 forward_sensitivity = ( 

131 [ 

132 (x[0], x[1]) 

133 for x in sensitivity.test(start_time=F_start, end_time=F_end, **kwargs) 

134 ] 

135 if sensitivity is not None 

136 else None 

137 ) 

138 

139 return cls( 

140 alpha_id=alpha.id, 

141 params=params, 

142 sensitivity_params=sensitivity.sensitivity_parameters 

143 if sensitivity is not None 

144 else None, 

145 back=AlphaReportV1Performance( 

146 performance=back, 

147 performance_df=back_df, 

148 sensitivity=back_sensitivity, 

149 sensitivity_sr_summary=SensitivitySharpeRatioSummary.compute( 

150 back, back_sensitivity 

151 ) 

152 if back_sensitivity is not None 

153 else None, 

154 ), 

155 forward=AlphaReportV1Performance( 

156 performance=forward, 

157 performance_df=forward_df, 

158 sensitivity=forward_sensitivity, 

159 sensitivity_sr_summary=SensitivitySharpeRatioSummary.compute( 

160 forward, forward_sensitivity 

161 ) 

162 if forward_sensitivity is not None 

163 else None, 

164 ), 

165 ) 

166 

167 def write_parquet(self, path: str): 

168 """Write the report to a parquet file.""" 

169 with open(path, "wb+") as f: 

170 f.write(self.serialize()) 

171 

172 def serialize(self) -> bytes: 

173 back_pdf_buf = io.BytesIO() 

174 self.back.performance_df.write_parquet(back_pdf_buf) 

175 forward_pdf_buf = io.BytesIO() 

176 self.forward.performance_df.write_parquet(forward_pdf_buf) 

177 

178 buf = io.BytesIO() 

179 df = pl.DataFrame( 

180 { 

181 "alpha_id": [self.alpha_id], 

182 "params": [json.dumps(self.params)], 

183 "sensitivity_params": [ 

184 TypeAdapter(dict[str, SensitivityParameter] | None).dump_json( 

185 self.sensitivity_params 

186 ) 

187 ], 

188 "back": [self.back.model_dump_json(exclude={"performance_df"})], 

189 "back_pdf_buf": [back_pdf_buf.getvalue()], 

190 "forward": [self.forward.model_dump_json(exclude={"performance_df"})], 

191 "forward_pdf_buf": [forward_pdf_buf.getvalue()], 

192 } 

193 ) 

194 df.write_parquet(buf) 

195 

196 return buf.getvalue() 

197 

198 @classmethod 

199 def deserialize(cls, buf: bytes) -> Self: 

200 df = pl.read_parquet(buf) 

201 

202 back_pdf = pl.read_parquet(df["back_pdf_buf"][0]) 

203 forward_pdf = pl.read_parquet(df["forward_pdf_buf"][0]) 

204 

205 return cls( 

206 alpha_id=df["alpha_id"][0], 

207 params=json.loads(df["params"][0]), 

208 sensitivity_params=TypeAdapter( 

209 dict[str, SensitivityParameter] | None 

210 ).validate_json(df["sensitivity_params"][0]), 

211 back=AlphaReportV1Performance( 

212 **json.loads(df["back"][0]), 

213 performance_df=back_pdf, 

214 ), 

215 forward=AlphaReportV1Performance( 

216 **json.loads(df["forward"][0]), 

217 performance_df=forward_pdf, 

218 ), 

219 ) 

220 

221 def __eq__(self, other: object) -> bool: 

222 if not isinstance(other, AlphaReportV1): 

223 return False 

224 return ( 

225 self.alpha_id == other.alpha_id 

226 and self.params == other.params 

227 and self.sensitivity_params == other.sensitivity_params 

228 and self.back.performance == other.back.performance 

229 and self.back.sensitivity == other.back.sensitivity 

230 and self.back.sensitivity_sr_summary == other.back.sensitivity_sr_summary 

231 and self.back.performance_df.equals(other.back.performance_df) 

232 and self.forward.performance == other.forward.performance 

233 and self.forward.sensitivity == other.forward.sensitivity 

234 and self.forward.sensitivity_sr_summary 

235 == other.forward.sensitivity_sr_summary 

236 and self.forward.performance_df.equals(other.forward.performance_df) 

237 )