Coverage for src/jquantstats/_stats/_periodic.py: 100%

1"""Period-bucketed reporting tables for financial returns data. 

2 

3Tabular, period-grouped views of returns: the monthly-returns pivot, the 

4inlier/outlier distribution across calendar frequencies, the benchmark 

5comparison table, and the worst-N-periods list. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, cast 

11 

12import polars as pl 

13 

14from ..exceptions import NoBenchmarkError 

15 

16if TYPE_CHECKING: 

17 from ..data import Data 

18 

19# ── Periodic reporting mixin ────────────────────────────────────────────────── 

20 

21 

22class _PeriodicReportingMixin: 

23 """Mixin providing period-bucketed reporting tables. 

24 

25 Covers: monthly-returns pivot table, distribution across calendar 

26 frequencies (daily…yearly), benchmark comparison table, and worst-N periods. 

27 """ 

28 

29 _data: Data 

30 all: pl.DataFrame 

31 

32 if TYPE_CHECKING: 

33 from .._protocol import DataLike 

34 

35 data: DataLike 

36 

37 def monthly_returns(self, eoy: bool = True, compounded: bool = True) -> dict[str, pl.DataFrame]: 

38 """Calculate monthly returns in a pivot-table format. 

39 

40 Groups returns by calendar month and year, producing a DataFrame with 

41 years as rows and months (JAN-DEC) as columns, plus an optional EOY 

42 column with the full-year compounded return. 

43 

44 Args: 

45 eoy (bool): Include an EOY column with the annual compounded return. 

46 Defaults to True. 

47 compounded (bool): Compound returns within each period. Defaults to True. 

48 

49 Returns: 

50 dict[str, pl.DataFrame]: Per-asset pivot tables with columns 

51 ``year``, ``JAN`` … ``DEC``, and optionally ``EOY``. 

52 

53 """ 

54 all_df = self.all 

55 date_col_name = self._data.date_col[0] 

56 month_names = { 

57 1: "JAN", 

58 2: "FEB", 

59 3: "MAR", 

60 4: "APR", 

61 5: "MAY", 

62 6: "JUN", 

63 7: "JUL", 

64 8: "AUG", 

65 9: "SEP", 

66 10: "OCT", 

67 11: "NOV", 

68 12: "DEC", 

69 } 

70 month_order = list(month_names.values()) 

71 

72 result: dict[str, pl.DataFrame] = {} 

73 for col, series in self._data.items(): 

74 df = pl.DataFrame({"date": all_df[date_col_name], "ret": series}).drop_nulls() 

75 df = df.with_columns( 

76 [ 

77 pl.col("date").dt.year().alias("year"), 

78 pl.col("date").dt.month().alias("month_num"), 

79 ] 

80 ) 

81 

82 agg_expr = ((1.0 + pl.col("ret")).product() - 1.0) if compounded else pl.col("ret").sum() 

83 monthly = ( 

84 df.group_by(["year", "month_num"]) 

85 .agg(agg_expr.alias("ret")) 

86 .with_columns( 

87 pl.col("month_num") 

88 .replace_strict( 

89 list(month_names.keys()), 

90 list(month_names.values()), 

91 return_dtype=pl.String, 

92 ) 

93 .alias("month_name") 

94 ) 

95 .sort(["year", "month_num"]) 

96 ) 

97 

98 pivoted = monthly.pivot(on="month_name", index="year", values="ret", aggregate_function="first") 

99 for m in month_order: 

100 if m not in pivoted.columns: 

101 pivoted = pivoted.with_columns(pl.lit(0.0).alias(m)) 

102 pivoted = ( 

103 pivoted.select(["year", *month_order]) 

104 .fill_null(0.0) 

105 .with_columns(pl.col("year").cast(pl.Int32)) 

106 .sort("year") 

107 ) 

108 

109 if eoy: 

110 eoy_agg = ( 

111 df.group_by("year") 

112 .agg(agg_expr.alias("EOY")) 

113 .with_columns(pl.col("year").cast(pl.Int32)) 

114 .sort("year") 

115 ) 

116 pivoted = pivoted.join(eoy_agg, on="year").sort("year") 

117 

118 result[col] = pivoted 

119 return result 

120 

121 def distribution(self, compounded: bool = True) -> dict[str, dict[str, dict[str, list[float]]]]: 

122 """Analyse return distributions across daily, weekly, monthly, quarterly, and yearly periods. 

123 

124 For each period, splits values into inliers and outliers using the 

125 IQR method (1.5 * IQR beyond Q1/Q3). 

126 

127 Args: 

128 compounded (bool): Compound returns within each period. Defaults to True. 

129 

130 Returns: 

131 dict: Nested dict ``{asset: {period: {"values": [...], "outliers": [...]}}}`` 

132 where period is one of ``"Daily"``, ``"Weekly"``, ``"Monthly"``, 

133 ``"Quarterly"``, ``"Yearly"``. 

134 

135 """ 

136 all_df = self.all 

137 date_col_name = self._data.date_col[0] 

138 

139 def _agg(df: pl.DataFrame, group_col: str) -> pl.Series: 

140 """Aggregate returns within each group using product or sum.""" 

141 expr = ((1.0 + pl.col("ret")).product() - 1.0) if compounded else pl.col("ret").sum() 

142 return df.group_by(group_col).agg(expr.alias("ret"))["ret"] 

143 

144 def _iqr_split(s: pl.Series) -> dict[str, list[float]]: 

145 """Split series into inliers and outliers using the IQR method.""" 

146 q1 = cast(float, s.quantile(0.25)) 

147 q3 = cast(float, s.quantile(0.75)) 

148 iqr = q3 - q1 

149 mask = (s >= q1 - 1.5 * iqr) & (s <= q3 + 1.5 * iqr) 

150 return {"values": s.filter(mask).to_list(), "outliers": s.filter(~mask).to_list()} 

151 

152 result: dict[str, dict[str, dict[str, list[float]]]] = {} 

153 for col, series in self._data.items(): 

154 df = pl.DataFrame({"date": all_df[date_col_name], "ret": series}).drop_nulls() 

155 df = df.with_columns( 

156 [ 

157 pl.col("date").dt.truncate("1w").alias("week"), 

158 pl.col("date").dt.truncate("1mo").alias("month"), 

159 pl.col("date").dt.truncate("3mo").alias("quarter"), 

160 pl.col("date").dt.truncate("1y").alias("year"), 

161 ] 

162 ) 

163 result[col] = { 

164 "Daily": _iqr_split(df["ret"]), 

165 "Weekly": _iqr_split(_agg(df, "week")), 

166 "Monthly": _iqr_split(_agg(df, "month")), 

167 "Quarterly": _iqr_split(_agg(df, "quarter")), 

168 "Yearly": _iqr_split(_agg(df, "year")), 

169 } 

170 return result 

171 

172 def compare( 

173 self, 

174 aggregate: str | None = None, 

175 compounded: bool = True, 

176 round_vals: int | None = None, 

177 ) -> dict[str, pl.DataFrame]: 

178 """Compare each asset's returns against the benchmark. 

179 

180 Aligns returns and benchmark by date, multiplies by 100 (percentage), 

181 then computes a ``Multiplier`` (Returns / Benchmark) and ``Won`` 

182 indicator (``"+"`` when the asset outperformed, ``"-"`` otherwise). 

183 

184 Args: 

185 aggregate (str | None): Pandas-style resample frequency for 

186 period aggregation (e.g. ``"ME"``, ``"QE"``, ``"YE"``). 

187 ``None`` returns daily rows. Defaults to None. 

188 compounded (bool): Compound returns when aggregating. Defaults to True. 

189 round_vals (int | None): Decimal places to round. Defaults to None. 

190 

191 Returns: 

192 dict[str, pl.DataFrame]: Per-asset DataFrames with columns 

193 ``Benchmark``, ``Returns``, ``Multiplier``, ``Won``. 

194 

195 Raises: 

196 AttributeError: If no benchmark data is attached. 

197 

198 """ 

199 if self._data.benchmark is None: 

200 raise NoBenchmarkError 

201 

202 all_df = self.all 

203 date_col_name = self._data.date_col[0] 

204 bench_col = self._data.benchmark.columns[0] 

205 

206 _freq_map = {"ME": "1mo", "QE": "3mo", "YE": "1y", "W": "1w"} 

207 

208 def _agg_series(df: pl.DataFrame, period_col: str, val_col: str) -> pl.DataFrame: 

209 """Aggregate a value column grouped by period using product or sum.""" 

210 expr = ((1.0 + pl.col(val_col)).product() - 1.0) if compounded else pl.col(val_col).sum() 

211 return df.group_by(period_col).agg(expr.alias(val_col)).sort(period_col) 

212 

213 result: dict[str, pl.DataFrame] = {} 

214 for col in self._data.returns.columns: 

215 df = all_df.select( 

216 [ 

217 pl.col(date_col_name), 

218 pl.col(col).alias("ret"), 

219 pl.col(bench_col).alias("bench"), 

220 ] 

221 ) 

222 

223 if aggregate is not None and aggregate in _freq_map: 

224 trunc = _freq_map[aggregate] 

225 df = df.with_columns(pl.col(date_col_name).dt.truncate(trunc).alias("period")) 

226 ret_agg = _agg_series(df.drop_nulls(subset=["ret"]), "period", "ret") 

227 bench_agg = _agg_series(df.drop_nulls(subset=["bench"]), "period", "bench") 

228 df = ret_agg.join(bench_agg, on="period", how="full", coalesce=True).sort("period") 

229 ret_col, bench_col_name, _date_alias = "ret", "bench", "period" 

230 else: 

231 ret_col, bench_col_name, _date_alias = "ret", "bench", date_col_name 

232 

233 ret_pct = (df[ret_col] * 100).alias("Returns") 

234 bench_pct = (df[bench_col_name] * 100).alias("Benchmark") 

235 out = pl.DataFrame( 

236 { 

237 "Benchmark": bench_pct, 

238 "Returns": ret_pct, 

239 } 

240 ) 

241 out = out.with_columns( 

242 [ 

243 (pl.col("Returns") / pl.col("Benchmark").replace(0.0, None)).alias("Multiplier"), 

244 pl.when(pl.col("Returns") >= pl.col("Benchmark")) 

245 .then(pl.lit("+")) 

246 .otherwise(pl.lit("-")) 

247 .alias("Won"), 

248 ] 

249 ) 

250 

251 if round_vals is not None: 

252 out = out.with_columns( 

253 [ 

254 pl.col("Benchmark").round(round_vals), 

255 pl.col("Returns").round(round_vals), 

256 pl.col("Multiplier").round(round_vals), 

257 ] 

258 ) 

259 

260 result[col] = out 

261 return result 

262 

263 def worst_n_periods(self, n: int = 5) -> dict[str, list[float | None]]: 

264 """Return the N worst return periods per asset. 

265 

266 If a series has fewer than ``n`` non-null observations the list is 

267 padded with ``None`` on the right. 

268 

269 Args: 

270 n: Number of worst periods to return. Defaults to 5. 

271 

272 Returns: 

273 dict[str, list[float | None]]: Sorted worst returns per asset. 

274 """ 

275 result: dict[str, list[float | None]] = {} 

276 for col, series in self._data.items(): 

277 nonnull = series.drop_nulls() 

278 worst: list[float | None] = nonnull.sort(descending=False).head(n).to_list() 

279 while len(worst) < n: 

280 worst.append(None) 

281 result[col] = worst 

282 return result