Coverage for src/jquantstats/_stats/_summary.py: 100%
60 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04:52 +0000
1"""The tidy `summary` table and its calendar-year breakdown.
3Split out of `_reporting.py`. These are the aggregating metrics: they call
4across every other mixin rather than computing anything themselves, which is
5why they carry the largest block of cross-mixin type stubs in the package.
6"""
8from __future__ import annotations
10from typing import TYPE_CHECKING, Any, cast
12import polars as pl
14if TYPE_CHECKING:
15 from ..data import Data
18class _SummaryStatsMixin:
19 """Mixin providing the `summary` table and `annual_breakdown`.
21 Cross-mixin dependencies:
22 - _BasicStatsMixin: avg_return, avg_win, avg_loss, win_rate, profit_factor,
23 payoff_ratio, best, worst, volatility, skew, kurtosis, value_at_risk,
24 conditional_value_at_risk
25 - _RiskStatsMixin: sharpe
26 - _DrawdownMixin: max_drawdown
27 - _ReportingStatsMixin: monthly_win_rate, avg_drawdown,
28 max_drawdown_duration, calmar, recovery_factor
29 """
31 _data: Data
32 all: pl.DataFrame
34 if TYPE_CHECKING:
35 from .._protocol import DataLike
37 data: DataLike
39 def avg_return(self) -> dict[str, float]:
40 """Defined on _BasicStatsMixin."""
42 def avg_win(self) -> dict[str, float]:
43 """Defined on _BasicStatsMixin."""
45 def avg_loss(self) -> dict[str, float]:
46 """Defined on _BasicStatsMixin."""
48 def win_rate(self) -> dict[str, float]:
49 """Defined on _BasicStatsMixin."""
51 def profit_factor(self) -> dict[str, float]:
52 """Defined on _BasicStatsMixin."""
54 def payoff_ratio(self) -> dict[str, float]:
55 """Defined on _BasicStatsMixin."""
57 def best(self) -> dict[str, float | None]:
58 """Defined on _BasicStatsMixin."""
60 def worst(self) -> dict[str, float | None]:
61 """Defined on _BasicStatsMixin."""
63 def volatility(self) -> dict[str, float]:
64 """Defined on _BasicStatsMixin."""
66 def sharpe(self) -> dict[str, float]:
67 """Defined on _RiskStatsMixin."""
69 def skew(self) -> dict[str, int | float | None]:
70 """Defined on _BasicStatsMixin."""
72 def kurtosis(self) -> dict[str, int | float | None]:
73 """Defined on _BasicStatsMixin."""
75 def value_at_risk(self) -> dict[str, float]:
76 """Defined on _BasicStatsMixin."""
78 def conditional_value_at_risk(self) -> dict[str, float]:
79 """Defined on _BasicStatsMixin."""
81 def max_drawdown(self) -> dict[str, float]:
82 """Defined on _DrawdownMixin."""
84 def monthly_win_rate(self) -> dict[str, float]:
85 """Defined on _ReportingStatsMixin."""
87 def avg_drawdown(self) -> dict[str, float]:
88 """Defined on _ReportingStatsMixin."""
90 def max_drawdown_duration(self) -> dict[str, float | int | None]:
91 """Defined on _ReportingStatsMixin."""
93 def calmar(self) -> dict[str, float]:
94 """Defined on _ReportingStatsMixin."""
96 def recovery_factor(self) -> dict[str, float]:
97 """Defined on _ReportingStatsMixin."""
99 def annual_breakdown(self) -> pl.DataFrame:
100 """Summary statistics broken down by calendar year.
102 Groups the data by calendar year using the date index, computes a
103 full `summary` for each year, and stacks the results with an
104 additional ``year`` column.
106 Returns:
107 pl.DataFrame: Columns ``year``, ``metric``, one per asset, sorted
108 by ``year``.
110 Raises:
111 ValueError: If the data has no date index.
112 """
113 all_df = self.all
114 date_col_name = self._data.date_col[0] if self._data.date_col else None
115 has_temporal = date_col_name is not None and all_df[date_col_name].dtype.is_temporal()
117 if not has_temporal:
118 return self._annual_breakdown_integer(all_df)
119 if date_col_name is None: # unreachable: has_temporal guarantees non-None # pragma: no cover
120 return pl.DataFrame() # pragma: no cover
121 return self._annual_breakdown_temporal(all_df, date_col_name)
123 def _summary_frame(self, sub_all: pl.DataFrame, index_cols: list[str], label: int) -> pl.DataFrame:
124 """Compute a `summary` for one sub-period and tag it with a ``year`` label.
126 Args:
127 sub_all: The combined (index + returns + benchmark) rows for the period.
128 index_cols: Column name(s) to use as the sub-period's date index.
129 label: Value written to the ``year`` column (calendar year or chunk ordinal).
131 Returns:
132 The summary DataFrame with an added ``year`` column.
133 """
134 # Construct the sub-period Data via type(self._data) rather than importing
135 # the concrete class: a lazy `from ..data import Data` would put the upper
136 # layer back into this subpackage's import graph, which is exactly the
137 # coupling _protocol.py exists to prevent. Mirrors the type(self) call below.
138 data_factory = cast(Any, type(self._data))
139 sub_returns = sub_all.select(self._data.returns.columns)
140 sub_benchmark = sub_all.select(self._data.benchmark.columns) if self._data.benchmark is not None else None
141 sub_data = data_factory(returns=sub_returns, index=sub_all.select(index_cols), benchmark=sub_benchmark)
142 summary: pl.DataFrame = cast(Any, type(self))(sub_data).summary()
143 return summary.with_columns(pl.lit(label).alias("year"))
145 @staticmethod
146 def _order_breakdown(result: pl.DataFrame) -> pl.DataFrame:
147 """Reorder breakdown columns so ``year`` and ``metric`` lead."""
148 ordered = ["year", "metric", *[c for c in result.columns if c not in ("year", "metric")]]
149 return result.select(ordered)
151 def _annual_breakdown_integer(self, all_df: pl.DataFrame) -> pl.DataFrame:
152 """Break down by fixed row chunks (~one year each) for an integer index."""
153 chunk = round(self._data._periods_per_year)
154 total = all_df.height
155 frames: list[pl.DataFrame] = []
156 for i, start in enumerate(range(0, total, chunk), start=1):
157 chunk_all = all_df.slice(start, chunk)
158 if chunk_all.height < max(5, chunk // 4):
159 continue
160 frames.append(self._summary_frame(chunk_all, self._data.date_col, i))
161 if not frames:
162 return pl.DataFrame()
163 return self._order_breakdown(pl.concat(frames))
165 def _annual_breakdown_temporal(self, all_df: pl.DataFrame, date_col_name: str) -> pl.DataFrame:
166 """Break down by calendar year for a temporal index."""
167 years = all_df[date_col_name].dt.year().unique().sort().to_list()
168 frames: list[pl.DataFrame] = []
169 for year in years:
170 year_all = all_df.filter(pl.col(date_col_name).dt.year() == year)
171 if year_all.height < 2:
172 continue
173 frames.append(self._summary_frame(year_all, [date_col_name], year))
174 if not frames:
175 asset_cols = list(self._data.returns.columns)
176 schema: dict[str, type[pl.DataType]] = {
177 "year": pl.Int32,
178 "metric": pl.String,
179 **dict.fromkeys(asset_cols, pl.Float64),
180 }
181 return pl.DataFrame(schema=schema)
182 return self._order_breakdown(pl.concat(frames))
184 def summary(self) -> pl.DataFrame:
185 """Summary statistics for each asset as a tidy DataFrame.
187 Each row is one metric; each column beyond ``metric`` is one asset.
189 Returns:
190 pl.DataFrame: A DataFrame with a ``metric`` column followed by one
191 column per asset.
193 Returns NaN when:
194 Cells are ``float("nan")`` when the underlying metric is unavailable
195 for the data (e.g. no temporal index or no benchmark).
196 """
197 assets = [col for col, _ in self._data.items()]
199 def _safe(fn: Any) -> dict[str, Any]:
200 """Call *fn()* and return its result; return NaN for each asset on any exception."""
201 try:
202 result: dict[str, Any] = fn()
203 except Exception:
204 return dict.fromkeys(assets, float("nan"))
205 return result
207 metrics: dict[str, dict[str, Any]] = {
208 "avg_return": _safe(self.avg_return),
209 "avg_win": _safe(self.avg_win),
210 "avg_loss": _safe(self.avg_loss),
211 "win_rate": _safe(self.win_rate),
212 "profit_factor": _safe(self.profit_factor),
213 "payoff_ratio": _safe(self.payoff_ratio),
214 "monthly_win_rate": _safe(self.monthly_win_rate),
215 "best": _safe(self.best),
216 "worst": _safe(self.worst),
217 "volatility": _safe(self.volatility),
218 "sharpe": _safe(self.sharpe),
219 "skew": _safe(self.skew),
220 "kurtosis": _safe(self.kurtosis),
221 "value_at_risk": _safe(self.value_at_risk),
222 "conditional_value_at_risk": _safe(self.conditional_value_at_risk),
223 "max_drawdown": _safe(self.max_drawdown),
224 "avg_drawdown": _safe(self.avg_drawdown),
225 "max_drawdown_duration": _safe(self.max_drawdown_duration),
226 "calmar": _safe(self.calmar),
227 "recovery_factor": _safe(self.recovery_factor),
228 }
230 rows: list[dict[str, Any]] = [
231 {"metric": name, **{asset: values.get(asset) for asset in assets}} for name, values in metrics.items()
232 ]
233 return pl.DataFrame(rows)