Coverage for src/jquantstats/_plots/_specs/_diagnostics.py: 100%

1"""Spec builders for the portfolio diagnostic charts. 

2 

3These ask questions about the strategy rather than reporting its returns: how 

4sensitive is it to execution delay, how correlated are its holdings, which 

5months carried it, and how much trading cost it can absorb. 

6""" 

7 

8from __future__ import annotations 

9 

10import math 

11from typing import TYPE_CHECKING 

12 

13import polars as pl 

14 

15from .._spec import Axis, BarSeries, FigureSpec, HeatmapGrid, LineSeries, Panel, RefLine 

16 

17if TYPE_CHECKING: 

18 from .._protocol import PortfolioLike 

19 

20__all__ = [ 

21 "correlation_heatmap_spec", 

22 "lead_lag_ir_spec", 

23 "monthly_returns_heatmap_spec", 

24 "trading_cost_impact_spec", 

25] 

26 

27_MONTH_NAMES = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") 

28 

29# The undelayed portfolio is the one being judged, so it is picked out from the 

30# lagged variants around it. 

31_FOCUS_COLOR = "red" 

32_NEUTRAL_COLOR = "#1f77b4" 

33 

34# A correlation runs from -1 to 1 whatever this particular data happens to span, 

35# so the ramp is pinned rather than fitted. 

36_CORRELATION_RANGE = (-1, 1) 

37_CORRELATION_SIZE = (700, 600) 

38 

39_MARKER_SIZE = 6 

40 

41 

42def lead_lag_ir_spec(portfolio: PortfolioLike, start: int, end: int) -> FigureSpec: 

43 """Describe the Sharpe-by-execution-delay chart. 

44 

45 Shifting the positions forward and backward shows how much of the 

46 strategy's edge depends on trading precisely when it says to. A peak away 

47 from zero suggests the signal is mistimed. 

48 

49 Args: 

50 portfolio: The portfolio to plot. 

51 start: First lag to include. 

52 end: Last lag to include. 

53 

54 Returns: 

55 FigureSpec: One bar per lag, with the undelayed one picked out. 

56 

57 Raises: 

58 TypeError: If *start* or *end* is not an integer. 

59 

60 """ 

61 if not isinstance(start, int) or not isinstance(end, int): 

62 raise TypeError 

63 if start > end: 

64 start, end = end, start 

65 

66 lags = list(range(start, end + 1)) 

67 sharpes: list[float] = [] 

68 for n in lags: 

69 lagged = portfolio if n == 0 else portfolio.lag(n) 

70 value = lagged.stats.sharpe().get("returns", float("nan")) 

71 sharpes.append(float(value) if value is not None else float("nan")) 

72 

73 panel = Panel( 

74 bars=( 

75 BarSeries( 

76 name="Sharpe by lag", 

77 x=lags, 

78 y=sharpes, 

79 colors=tuple(_FOCUS_COLOR if lag == 0 else _NEUTRAL_COLOR for lag in lags), 

80 ), 

81 ), 

82 xaxis=Axis(title="Lag (steps)"), 

83 yaxis=Axis(title="Sharpe ratio"), 

84 ) 

85 return FigureSpec( 

86 title="Lead/Lag Information Ratio (Sharpe) by Lag", 

87 panels=(panel,), 

88 height=None, 

89 chrome="plain", 

90 hover_mode="x", 

91 ) 

92 

93 

94def correlation_heatmap_spec(portfolio: PortfolioLike, frame: pl.DataFrame | None, name: str, title: str) -> FigureSpec: 

95 """Describe the correlation matrix of the holdings and the portfolio. 

96 

97 Args: 

98 portfolio: The portfolio to plot. 

99 frame: Series to correlate against, or None for the portfolio's prices. 

100 name: Column name to give the portfolio's own profit series. 

101 title: Chart title. 

102 

103 Returns: 

104 FigureSpec: A square matrix pinned to the full [-1, 1] range. 

105 

106 """ 

107 frame = portfolio.prices if frame is None else frame 

108 corr = portfolio.correlation(frame, name=name) 

109 labels = tuple(corr.columns) 

110 values = tuple(tuple(row) for row in corr.rows()) 

111 low, high = _CORRELATION_RANGE 

112 width, height = _CORRELATION_SIZE 

113 

114 panel = Panel( 

115 heatmap=HeatmapGrid( 

116 x_labels=labels, 

117 y_labels=labels, 

118 z=values, 

119 text=tuple(tuple(f"{value:.2f}" for value in row) for row in values), 

120 colorscale="rdbu_r", 

121 # A diverging ramp centred on zero: uncorrelated reads as neutral. 

122 zmid=0, 

123 zmin=low, 

124 zmax=high, 

125 colorbar_title="Correlation", 

126 # px.imshow generated a tooltip automatically; going to a plain 

127 # heatmap means asking for one, or the chart silently loses it. 

128 hover_label="Correlation", 

129 ), 

130 # The labels are the series names; repeating them as axis titles would 

131 # say nothing. 

132 xaxis=Axis(title=""), 

133 yaxis=Axis(title=""), 

134 ) 

135 return FigureSpec(title=title, panels=(panel,), height=height, width=width, chrome="bare") 

136 

137 

138def monthly_returns_heatmap_spec(portfolio: PortfolioLike) -> FigureSpec: 

139 """Describe the portfolio's monthly-returns calendar. 

140 

141 Args: 

142 portfolio: The portfolio to plot. 

143 

144 Returns: 

145 FigureSpec: A year-by-month grid. 

146 

147 """ 

148 monthly = portfolio.monthly 

149 years = monthly["year"].unique().sort().to_list() 

150 

151 z: list[tuple[float | None, ...]] = [] 

152 text: list[tuple[str, ...]] = [] 

153 for year in years: 

154 rows = monthly.filter(pl.col("year") == year) 

155 by_month = {int(row["month"]): float(row["returns"]) for row in rows.iter_rows(named=True)} 

156 percents = [by_month[m] * 100.0 if m in by_month else None for m in range(1, 13)] 

157 z.append(tuple(percents)) 

158 text.append(tuple("" if value is None else f"{value:.1f}%" for value in percents)) 

159 

160 panel = Panel( 

161 heatmap=HeatmapGrid( 

162 x_labels=_MONTH_NAMES, 

163 y_labels=tuple(str(year) for year in years), 

164 z=tuple(z), 

165 text=tuple(text), 

166 colorscale="rdylgn", 

167 colorbar_title="Return (%)", 

168 hover_label="Return", 

169 ), 

170 xaxis=Axis(title="Month"), 

171 # Years are labels, not a continuous scale: 2023 and 2024 are adjacent 

172 # rows, not a year apart on a number line. 

173 yaxis=Axis(title="Year", kind="category"), 

174 ) 

175 return FigureSpec(title="Monthly Returns Heatmap", panels=(panel,), height=None, chrome="bare") 

176 

177 

178def trading_cost_impact_spec(portfolio: PortfolioLike, max_bps: int) -> FigureSpec: 

179 """Describe how trading cost erodes the Sharpe ratio. 

180 

181 Args: 

182 portfolio: The portfolio to plot. 

183 max_bps: Highest one-way cost to evaluate, in basis points. 

184 

185 Returns: 

186 FigureSpec: Sharpe against cost, with the zero-cost level marked when 

187 it is finite. 

188 

189 Raises: 

190 ValueError: If *max_bps* is not a positive integer. 

191 

192 """ 

193 impact = portfolio.trading_cost_impact(max_bps=max_bps) 

194 costs = impact["cost_bps"].to_list() 

195 sharpes = impact["sharpe"].to_list() 

196 

197 baseline = float(sharpes[0]) if sharpes and sharpes[0] is not None else float("nan") 

198 # An unmeasurable baseline gets no line rather than one drawn at NaN. 

199 ref_lines = () if math.isnan(baseline) else (RefLine(value=baseline, dash="dash", label="0 bps baseline"),) 

200 

201 panel = Panel( 

202 lines=( 

203 LineSeries( 

204 name="Sharpe (cost-adjusted)", 

205 x=costs, 

206 y=sharpes, 

207 color=_NEUTRAL_COLOR, 

208 markers=True, 

209 marker_size=_MARKER_SIZE, 

210 ), 

211 ), 

212 ref_lines=ref_lines, 

213 # One tick per basis point: the axis spans a couple of dozen integers. 

214 xaxis=Axis(title="One-way cost (basis points)", dtick=1), 

215 yaxis=Axis(title="Annualised Sharpe ratio"), 

216 ) 

217 return FigureSpec( 

218 # Escaped rather than written literally: the title has always used an en 

219 # dash, and ruff flags the bare character as visually ambiguous. 

220 title=f"Trading Cost Impact on Sharpe Ratio (0\u2013{max_bps} bps)", 

221 panels=(panel,), 

222 height=None, 

223 chrome="plain", 

224 hover_mode="x unified", 

225 )