diff --git a/Makefile b/Makefile index e224f855..f1fb8526 100644 --- a/Makefile +++ b/Makefile @@ -9,6 +9,12 @@ MKDOCS_EXTRA_PACKAGES = --with 'mkdocstrings[python]' # Always include the Rhiza API (template-managed) include .rhiza/rhiza.mk +# The optional FastAPI service under api/ (the [web] extra) is repo-owned and not +# contributed by any template bundle, so deptry never saw it and reported its +# dependencies as unused. Append it to the shared scan; DEPTRY_FOLDERS is the +# accumulating variable python.mk builds up, so this must come after the include. +DEPTRY_FOLDERS += api + # Architectural import contracts (import-linter): the analytics subpackages # (_stats, _plots, _reports, _utils) annotate against the structural Protocols in # _protocol.py and must never import the concrete Data / Portfolio at runtime. diff --git a/docs/STABILITY.md b/docs/STABILITY.md index 83745a24..395406b5 100644 --- a/docs/STABILITY.md +++ b/docs/STABILITY.md @@ -7,6 +7,13 @@ icon: material/shield-check This document defines the public API surface of **jquantstats** and the stability guarantees that apply from **v1.0.0** onwards. +!!! info "Current status: 0.x — the guarantees below are not yet in force" + + jquantstats has not reached `v1.0.0`. The surface described here is what + *will* be frozen at 1.0, and is already treated as settled in practice, but + the formal Semantic Versioning contract begins at that tag. See + [Before v1.0.0](#before-v100) for the policy that applies today. + ## Stable public exports The following names are exported from the top-level `jquantstats` package and @@ -49,8 +56,9 @@ From **v1.0.0** onwards jquantstats follows [Semantic Versioning](https://semver Anything that is **not** in the table above is considered internal and may change or be removed in any release: -- Private modules: `_stats.py`, `_plots.py`, `_reports.py`, - `_types.py`, `_portfolio_data.py`. +- Private modules and subpackages: `_stats/`, `_plots/`, `_reports/`, + `_utils/`, the `_portfolio_*.py` mixins, `_types.py`, `_protocol.py`, + `_cost_model.py`, `_cache.py`, `_data_reshape.py`. - Private classes, functions, or attributes whose names begin with an underscore (e.g. `Data._raw_returns`, `Stats._df`). - Sub-module paths such as `jquantstats.portfolio` @@ -80,8 +88,38 @@ Example timeline: | `1.3.0` | `old_name` deprecated; `DeprecationWarning` raised on use; `new_name` available | | `1.4.0` | `old_name` removed | -## Pre-release versions - -Releases tagged `0.x.y` carry **no stability guarantee**. The API may -change in any release. Once `v1.0.0` is tagged the guarantees above -apply. +## Before v1.0.0 + +Releases tagged `0.x.y` carry **no formal stability guarantee** — Semantic +Versioning, as described above, begins at `v1.0.0`. That is the contractual +position. In practice the project is more conservative than that, and the +paragraphs below describe what a caller can actually rely on today. + +**What is already settled.** The exports in the table above — `Portfolio`, +`Data`, `Stats`, `Plots` and the two type aliases — are not expected to be +renamed or removed before 1.0. The constructors (`Portfolio.from_position`, +`from_cash_position`, `from_risk_position`, `Data.from_returns`) and the +`.stats` / `.plots` / `.report` accessors are likewise treated as fixed +points. Changes here would be disruptive enough that they are held for the +1.0 boundary. + +**What may still move.** Individual metric methods on `Stats` may gain +keyword arguments, change default values, or be renamed for consistency as +the QuantStats-parity work settles; chart signatures on `Plots` may change +as the Plotly builders are refactored. Anything private — see +[What is *not* stable](#what-is-not-stable) — may change in any release, +including the internal module layout, which has been reorganised more than +once during 0.x. + +**How changes are communicated.** Every release notes its changes in the +[changelog](changelog.md). A minor bump (`0.10 → 0.11`) is where a breaking +change to a public name will appear; patch releases (`0.10.0 → 0.10.1`) are +bug fixes only. Where a rename is avoidable, the old name is kept for one +minor version with a `DeprecationWarning`, following the same courtesy as the +post-1.0 [deprecation policy](#deprecation-policy) — but before 1.0 this is a +practice, not a promise. + +**Pinning advice.** Pin to a minor version (`jquantstats>=0.10,<0.11`) if you +need the surface to hold still; a bare `jquantstats>=0.10` may pick up a +breaking change at the next minor. Once `v1.0.0` is tagged, the guarantees in +the sections above replace everything here. diff --git a/pyproject.toml b/pyproject.toml index df8cc109..a9e152bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -95,9 +95,20 @@ include = [ [tool.bandit] exclude_dirs = ["src/tests"] # Exclude test directories from security scanning -# Ignore specific dependency issues +# Ignore specific dependency issues. +# +# DEP002 flags a declared dependency with no import site. The three below have +# none by design, so the ignore records a fact rather than hiding a gap: +# kaleido - Plotly's static image-export backend. Selected at runtime +# by plotly.io, never imported by us. +# uvicorn - ASGI server for api/app.py. Invoked as a command +# (`uvicorn api.app:app`), not imported. +# python-multipart - parses multipart/form-data for FastAPI's Form/UploadFile. +# Imported by Starlette, never by us. +# fastapi is deliberately absent: api/ is now part of the deptry scan (see the +# DEPTRY_FOLDERS line in the Makefile), so its import there is observed directly. [tool.deptry.per_rule_ignores] -DEP002 = ["kaleido", "fastapi", "uvicorn", "python-multipart"] # DEP002: Unused direct dependencies +DEP002 = ["kaleido", "uvicorn", "python-multipart"] # Package to module name mapping [tool.deptry.package_module_name_map] diff --git a/src/jquantstats/_plots/_data/_periodic.py b/src/jquantstats/_plots/_data/_periodic.py index 755bbb01..cbd7aeea 100644 --- a/src/jquantstats/_plots/_data/_periodic.py +++ b/src/jquantstats/_plots/_data/_periodic.py @@ -7,12 +7,28 @@ import plotly.graph_objects as go import polars as pl -from ._styling import _apply_base_layout, _bar_colors, _hex_to_rgba, _ticker_colors +from ._styling import _apply_base_layout, _bar_colors, _ticker_colors, _yearly_bar_colors if TYPE_CHECKING: from jquantstats._protocol import DataLike +def _period_agg_exprs(tickers: list[str], compounded: bool) -> list[pl.Expr]: + """Per-ticker aggregation expressions for a period bucket. + + Args: + tickers: Asset column names to aggregate. + compounded: Compound returns within the bucket when True, sum them + when False. + + Returns: + One aliased expression per ticker. + """ + if compounded: + return [((1.0 + pl.col(t)).product() - 1.0).alias(t) for t in tickers] + return [pl.col(t).sum().alias(t) for t in tickers] + + def _monthly_heatmap_matrix( monthly: pl.DataFrame, years: list[int] ) -> tuple[list[list[float | None]], list[list[str]]]: @@ -103,21 +119,14 @@ def yearly_returns(self, title: str = "Yearly Returns", compounded: bool = True) tickers = [c for c in df.columns if c != date_col] colors = _ticker_colors(tickers) - agg_exprs = ( - [((1.0 + pl.col(t)).product() - 1.0).alias(t) for t in tickers] - if compounded - else [pl.col(t).sum().alias(t) for t in tickers] - ) + agg_exprs = _period_agg_exprs(tickers, compounded) yearly = ( df.with_columns(pl.col(date_col).dt.year().alias("_year")).group_by("_year").agg(agg_exprs).sort("_year") ) fig = go.Figure() for ticker in tickers: - values = yearly[ticker].to_list() - bar_colors = [ - colors[ticker] if v is not None and v >= 0 else _hex_to_rgba(colors[ticker], 0.5) for v in values - ] + bar_colors = _yearly_bar_colors(yearly[ticker].to_list(), colors[ticker]) fig.add_trace( go.Bar( x=yearly["_year"], @@ -153,9 +162,7 @@ def monthly_returns(self, title: str = "Monthly Returns", compounded: bool = Tru monthly = df.group_by_dynamic( index_column=date_col, every="1mo", period="1mo", closed="right", label="right" - ).agg( - [((1.0 + pl.col(t)).product() - 1.0).alias(t) if compounded else pl.col(t).sum().alias(t) for t in tickers] - ) + ).agg(_period_agg_exprs(tickers, compounded)) fig = go.Figure() for ticker in tickers: diff --git a/src/jquantstats/_plots/_data/_rolling.py b/src/jquantstats/_plots/_data/_rolling.py index c4db3ad2..64cc9fe1 100644 --- a/src/jquantstats/_plots/_data/_rolling.py +++ b/src/jquantstats/_plots/_data/_rolling.py @@ -16,6 +16,27 @@ from jquantstats._protocol import DataLike +def _rolling_beta_expr(asset: str, bench_col: str, window: int) -> pl.Expr: + """Trailing-window OLS beta of *asset* against *bench_col*. + + Beta is ``cov(asset, bench) / var(bench)``, expanded into rolling means so + the whole estimate is a single Polars expression. + + Args: + asset: Asset column name. + bench_col: Benchmark column name. + window: Trailing window size in rows. + + Returns: + An expression aliased ``beta``. + """ + mean_x = pl.col(asset).rolling_mean(window_size=window) + mean_y = pl.col(bench_col).rolling_mean(window_size=window) + mean_xy = (pl.col(asset) * pl.col(bench_col)).rolling_mean(window_size=window) + mean_y2 = (pl.col(bench_col) ** 2).rolling_mean(window_size=window) + return ((mean_xy - mean_x * mean_y) / (mean_y2 - mean_y**2)).alias("beta") + + class _RollingPlotsMixin: """Rolling-window metric plots for :class:`DataPlots`.""" @@ -23,6 +44,26 @@ class _RollingPlotsMixin: _data: DataLike + def _beta_assets(self, df: pl.DataFrame, date_col: str, bench_col: str) -> list[str]: + """Asset columns to plot beta for. + + Prefers the explicit ``returns`` frame when the data exposes one, and + otherwise falls back to every column of *df* that is neither the date + nor the benchmark. + + Args: + df: The combined index/returns/benchmark frame. + date_col: Name of the date column. + bench_col: Name of the benchmark column. + + Returns: + The asset column names. + """ + returns_df = getattr(self._data, "returns", None) + if returns_df is not None: + return list(returns_df.columns) + return [c for c in df.columns if c != date_col and c != bench_col] + def rolling_sharpe( self, rolling_period: int = 126, @@ -219,12 +260,7 @@ def rolling_beta( raise NoBenchmarkError bench_col = benchmark_df.columns[0] - returns_df = getattr(self._data, "returns", None) - assets = ( - list(returns_df.columns) - if returns_df is not None - else [c for c in df.columns if c != date_col and c != bench_col] - ) + assets = self._beta_assets(df, date_col, bench_col) colors = _ticker_colors(assets) windows = [w for w in (rolling_period, rolling_period2) if w is not None] line_styles = ["solid", "dash"] @@ -232,13 +268,7 @@ def rolling_beta( fig = go.Figure() for asset in assets: for w, dash in zip(windows, line_styles, strict=False): - mean_x = pl.col(asset).rolling_mean(window_size=w) - mean_y = pl.col(bench_col).rolling_mean(window_size=w) - mean_xy = (pl.col(asset) * pl.col(bench_col)).rolling_mean(window_size=w) - mean_y2 = (pl.col(bench_col) ** 2).rolling_mean(window_size=w) - beta_expr = ((mean_xy - mean_x * mean_y) / (mean_y2 - mean_y**2)).alias("beta") - - beta_df = df.with_columns(beta_expr) + beta_df = df.with_columns(_rolling_beta_expr(asset, bench_col, w)) label = f"{asset} ({w}d)" fig.add_trace( go.Scatter( diff --git a/src/jquantstats/_plots/_data/_styling.py b/src/jquantstats/_plots/_data/_styling.py index b5b5b86f..98bcee4c 100644 --- a/src/jquantstats/_plots/_data/_styling.py +++ b/src/jquantstats/_plots/_data/_styling.py @@ -112,6 +112,25 @@ def _bar_colors(values: list[float | None], positive_color: str, single_asset: b return [positive_color if v is not None and v > 0 else negative_color for v in values] +def _yearly_bar_colors(values: list[float | None], positive_color: str) -> list[str]: + """Bar colors for the yearly-returns chart. + + Deliberately distinct from `_bar_colors`: the yearly chart treats a flat + zero year as positive (``>= 0``) and fades negatives to alpha 0.5 rather + than 0.4, so the two cannot share an implementation without changing what + is rendered. + + Args: + values: The per-year return values; ``None`` counts as negative. + positive_color: The asset's base color. + + Returns: + One color string per value. + """ + negative_color = _hex_to_rgba(positive_color, 0.5) + return [positive_color if v is not None and v >= 0 else negative_color for v in values] + + def _compute_drawdown_periods(prices: list[float], n: int) -> list[dict[str, Any]]: """Identify the top *n* drawdown periods from a cumulative price series. diff --git a/src/jquantstats/_plots/_portfolio.py b/src/jquantstats/_plots/_portfolio.py deleted file mode 100644 index e1a6fb48..00000000 --- a/src/jquantstats/_plots/_portfolio.py +++ /dev/null @@ -1,588 +0,0 @@ -"""Plotting utilities for portfolio analytics using Plotly. - -This module defines the PortfolioPlots facade which renders common portfolio visuals -such as snapshots, lagged performance curves, smoothed-holdings curves, and -lead/lag information ratio bar charts. Designed for notebook use. -""" - -from __future__ import annotations - -from typing import TYPE_CHECKING - -import plotly.express as px -import plotly.graph_objects as go -import plotly.io as pio -import polars as pl -from plotly.subplots import make_subplots - -from ._data._styling import _apply_base_layout - -if TYPE_CHECKING: - from ._protocol import PortfolioLike - -# Ensure Plotly works with Marimo (set after imports to satisfy linters) -pio.renderers.default = "plotly_mimetype" - - -class PortfolioPlots: - """Facade for portfolio plots built with Plotly. - - Provides convenience methods to visualize portfolio performance and - diagnostics directly from a Portfolio instance (e.g., snapshot charts, - lagged performance, smoothed holdings, and lead/lag IR). - """ - - __slots__ = ("_portfolio",) - - def __init__(self, portfolio: PortfolioLike) -> None: - self._portfolio = portfolio - - def lead_lag_ir_plot(self, start: int = -10, end: int = 19) -> go.Figure: - """Plot Sharpe ratio (IR) across lead/lag variants of the portfolio. - - Builds portfolios with cash positions lagged from ``start`` to ``end`` - (inclusive) and plots a bar chart of the Sharpe ratio for each lag. - Positive lags delay weights; negative lags lead them. - - Args: - start: First lag to include (default: -10). - end: Last lag to include (default: +19). - - Returns: - A Plotly Figure with one bar per lag labeled by the lag value. - """ - if not isinstance(start, int) or not isinstance(end, int): - raise TypeError - if start > end: - start, end = end, start - - lags = list(range(start, end + 1)) - - x_vals: list[int] = [] - y_vals: list[float] = [] - - for n in lags: - pf = self._portfolio if n == 0 else self._portfolio.lag(n) - # Compute Sharpe on the portfolio's returns series - sharpe_val = pf.stats.sharpe().get("returns", float("nan")) - # Ensure a float (Stats returns mapping asset->value) - y_vals.append(float(sharpe_val) if sharpe_val is not None else float("nan")) - x_vals.append(n) - - colors = ["red" if x == 0 else "#1f77b4" for x in x_vals] - fig = go.Figure( - data=[ - go.Bar(x=x_vals, y=y_vals, name="Sharpe by lag", marker_color=colors), - ] - ) - fig.update_layout( - title="Lead/Lag Information Ratio (Sharpe) by Lag", - xaxis_title="Lag (steps)", - yaxis_title="Sharpe ratio", - plot_bgcolor="white", - hovermode="x", - ) - fig.update_xaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey") - fig.update_yaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey") - return fig - - def snapshot(self, log_scale: bool = False) -> go.Figure: - """Return a snapshot dashboard of NAV and drawdown. - - When the portfolio has a non-zero ``cost_model.cost_per_unit``, an additional - ``"Net-of-Cost NAV"`` trace is overlaid on the NAV panel showing the - realised NAV path after deducting position-delta trading costs. - - Args: - log_scale (bool, optional): If True, display NAV on a log scale. Defaults to False. - - Returns: - plotly.graph_objects.Figure: A Figure with accumulated NAV (including tilt/timing) - and drawdown shaded area, equipped with a range selector. - """ - # Create subplot grid with domain for stats table - fig = make_subplots( - rows=2, - cols=1, - shared_xaxes=True, - row_heights=[0.66, 0.33], - subplot_titles=["Accumulated Profit", "Drawdown"], - vertical_spacing=0.05, - ) - - # --- Row 1: Cumulative Returns - fig.add_trace( - go.Scatter( - x=self._portfolio.nav_accumulated["date"], - y=self._portfolio.nav_accumulated["NAV_accumulated"], - mode="lines", - name="NAV", - showlegend=False, - ), - row=1, - col=1, - ) - - fig.add_trace( - go.Scatter( - x=self._portfolio.tilt.nav_accumulated["date"], - y=self._portfolio.tilt.nav_accumulated["NAV_accumulated"], - mode="lines", - name="Tilt", - showlegend=False, - ), - row=1, - col=1, - ) - - fig.add_trace( - go.Scatter( - x=self._portfolio.timing.nav_accumulated["date"], - y=self._portfolio.timing.nav_accumulated["NAV_accumulated"], - mode="lines", - name="Timing", - showlegend=False, - ), - row=1, - col=1, - ) - - # Net-of-cost NAV overlay (only when a cost model is active) - if self._portfolio.cost_model.cost_per_unit > 0: - net_nav_df = self._portfolio.net_cost_nav - x_dates = net_nav_df["date"] if "date" in net_nav_df.columns else None - fig.add_trace( - go.Scatter( - x=x_dates, - y=net_nav_df["NAV_accumulated_net"], - mode="lines", - name="Net-of-Cost NAV", - line={"dash": "dash"}, - showlegend=True, - ), - row=1, - col=1, - ) - - fig.add_trace( - go.Scatter( - x=self._portfolio.drawdown["date"], - y=self._portfolio.drawdown["drawdown_pct"], - mode="lines", - fill="tozeroy", - name="Drawdown", - showlegend=False, - ), - row=2, - col=1, - ) - - fig.add_hline(y=0, line_width=1, line_color="gray", row=2, col=1) - - _apply_base_layout(fig, "Performance Dashboard", height=1200) - - fig.update_yaxes(title_text="NAV (accumulated)", row=1, col=1, tickformat=".2s") - fig.update_yaxes(title_text="Drawdown", row=2, col=1, tickformat=".0%") - - if log_scale: - fig.update_yaxes(type="log", row=1, col=1) - # Ensure the first y-axis is explicitly set for environments - # where subplot updates may not propagate to layout alias. - if hasattr(fig.layout, "yaxis"): # pragma: no branch — plotly figures always have .yaxis - fig.layout.yaxis.type = "log" - - return fig - - @staticmethod - def _apply_nav_layout(fig: go.Figure, title: str, log_scale: bool = False) -> None: - """Apply common NAV-accumulated layout to *fig* in-place. - - Configures the plot background, legend, hover mode, x-axis date range - selector, y-axis label, grid lines, and optional logarithmic y-scale. - Shared by `lagged_performance_plot` and - `smoothed_holdings_performance_plot`. - - Args: - fig: The Plotly Figure to configure. - title: Chart title text. - log_scale: If True, set the primary y-axis to logarithmic scale. - """ - _apply_base_layout(fig, title) - fig.update_yaxes(title_text="NAV (accumulated)") - - if log_scale: - fig.update_yaxes(type="log") - if hasattr(fig.layout, "yaxis"): # pragma: no branch — plotly figures always have .yaxis - fig.layout.yaxis.type = "log" - - def lagged_performance_plot(self, lags: list[int] | None = None, log_scale: bool = False) -> go.Figure: - """Plot NAV_accumulated for multiple lagged portfolios. - - Creates a Plotly figure with one line per lag value showing the - accumulated NAV series for the portfolio with cash positions - shifted by that lag. By default, lags [0, 1, 2, 3, 4] are used. - - Args: - lags: A list of integer lags to apply; defaults to [0, 1, 2, 3, 4]. - log_scale: If True, set the primary y-axis to logarithmic scale. - - Returns: - A Plotly Figure containing one trace per requested lag. - """ - if lags is None: - lags = [0, 1, 2, 3, 4] - if not isinstance(lags, list) or not all(isinstance(x, int) for x in lags): - raise TypeError - - fig = go.Figure() - for lag in lags: - pf = self._portfolio if lag == 0 else self._portfolio.lag(lag) - nav = pf.nav_accumulated - fig.add_trace( - go.Scatter( - x=nav["date"], - y=nav["NAV_accumulated"], - mode="lines", - name=f"lag {lag}", - line={"width": 1}, - ) - ) - - self._apply_nav_layout(fig, title="NAV accumulated by lag", log_scale=log_scale) - return fig - - def rolling_sharpe_plot(self, window: int = 63) -> go.Figure: - """Plot rolling annualised Sharpe ratio over time. - - Computes the rolling Sharpe for each asset column using the given - window and renders one line per asset. - - Args: - window: Rolling-window size in periods. Defaults to 63. - - Returns: - A Plotly Figure with one trace per asset. - - Raises: - ValueError: If ``window`` is not a positive integer. - """ - if not isinstance(window, int) or window <= 0: - raise ValueError(f"window must be a positive integer, got {window!r}") # noqa: TRY003 - - rolling = self._portfolio.stats.rolling_sharpe(rolling_period=window) - - fig = go.Figure() - date_col = rolling["date"] if "date" in rolling.columns else None - for col in rolling.columns: - if col == "date": - continue - fig.add_trace( - go.Scatter( - x=date_col, - y=rolling[col], - mode="lines", - name=col, - line={"width": 1}, - ) - ) - - fig.add_hline(y=0, line_width=1, line_dash="dash", line_color="gray") - - _apply_base_layout(fig, f"Rolling Sharpe Ratio ({window}-period window)") - fig.update_yaxes(title_text="Sharpe ratio") - return fig - - def rolling_volatility_plot(self, window: int = 63) -> go.Figure: - """Plot rolling annualised volatility over time. - - Computes the rolling volatility for each asset column using the given - window and renders one line per asset. - - Args: - window: Rolling-window size in periods. Defaults to 63. - - Returns: - A Plotly Figure with one trace per asset. - - Raises: - ValueError: If ``window`` is not a positive integer. - """ - if not isinstance(window, int) or window <= 0: - raise ValueError(f"window must be a positive integer, got {window!r}") # noqa: TRY003 - - rolling = self._portfolio.stats.rolling_volatility(rolling_period=window) - - fig = go.Figure() - date_col = rolling["date"] if "date" in rolling.columns else None - for col in rolling.columns: - if col == "date": - continue - fig.add_trace( - go.Scatter( - x=date_col, - y=rolling[col], - mode="lines", - name=col, - line={"width": 1}, - ) - ) - - _apply_base_layout(fig, f"Rolling Volatility ({window}-period window)") - fig.update_yaxes(title_text="Annualised volatility") - return fig - - def annual_sharpe_plot(self) -> go.Figure: - """Plot annualised Sharpe ratio broken down by calendar year. - - Computes the Sharpe ratio for each calendar year from the portfolio - returns and renders a grouped bar chart with one bar per year per - asset. - - Returns: - A Plotly Figure with one bar group per asset. - """ - breakdown = self._portfolio.stats.annual_breakdown() - - # Extract the sharpe row for each year - sharpe_rows = breakdown.filter(pl.col("metric") == "sharpe") - asset_cols = [c for c in sharpe_rows.columns if c not in ("year", "metric")] - - fig = go.Figure() - for asset in asset_cols: - fig.add_trace( - go.Bar( - x=sharpe_rows["year"], - y=sharpe_rows[asset], - name=asset, - ) - ) - - fig.add_hline(y=0, line_width=1, line_color="gray") - - fig.update_layout( - title="Annual Sharpe Ratio by Year", - barmode="group", - hovermode="x unified", - plot_bgcolor="white", - legend={"orientation": "h", "yanchor": "bottom", "y": 1.02, "xanchor": "right", "x": 1}, - ) - fig.update_yaxes(title_text="Sharpe ratio") - fig.update_xaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey", title_text="Year") - fig.update_yaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey") - return fig - - def correlation_heatmap( - self, - frame: pl.DataFrame | None = None, - name: str = "portfolio", - title: str = "Correlation heatmap", - ) -> go.Figure: - """Plot a correlation heatmap for assets and the portfolio. - - If ``frame`` is None, uses the portfolio's prices. The portfolio's - profit series is appended under ``name`` before computing the - correlation matrix. - - Args: - frame: Optional Polars DataFrame with at least the asset price - columns. If omitted, uses ``self._portfolio.prices``. - name: Column name under which to include the portfolio profit. - title: Plot title. - - Returns: - A Plotly Figure rendering the correlation matrix as a heatmap. - """ - if frame is None: - frame = self._portfolio.prices - - corr = self._portfolio.correlation(frame, name=name) - - # Create an interactive heatmap - fig = px.imshow( - corr, - x=corr.columns, - y=corr.columns, - text_auto=".2f", # show correlation values - color_continuous_scale="RdBu_r", # red-blue diverging colormap - zmin=-1, - zmax=1, # correlation range - title=title, - ) - - # Adjust layout - fig.update_layout( - xaxis_title="", yaxis_title="", width=700, height=600, coloraxis_colorbar={"title": "Correlation"} - ) - - return fig - - def monthly_returns_heatmap(self) -> go.Figure: - """Plot a monthly returns calendar heatmap. - - Groups portfolio returns by calendar year and month, then renders a - Plotly heatmap with months on the x-axis and years on the y-axis. - Green cells indicate positive months; red cells indicate negative - months. Cell text shows the percentage return for that month. - - Returns: - A Plotly Figure with a calendar heatmap of monthly returns. - - Raises: - ValueError: If the portfolio has no ``date`` column. - """ - monthly = self._portfolio.monthly - - years = monthly["year"].unique().sort().to_list() - month_names = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] - - z: list[list[float | None]] = [] - text: list[list[str]] = [] - for year in years: - year_data = monthly.filter(pl.col("year") == year) - year_row: list[float | None] = [] - year_text: list[str] = [] - for m in range(1, 13): - month_data = year_data.filter(pl.col("month") == m) - if month_data.is_empty(): - year_row.append(None) - year_text.append("") - else: - ret = float(month_data["returns"][0]) - year_row.append(ret * 100.0) - year_text.append(f"{ret * 100.0:.1f}%") - z.append(year_row) - text.append(year_text) - - fig = go.Figure( - data=go.Heatmap( - z=z, - x=month_names, - y=[str(y) for y in years], - text=text, - texttemplate="%{text}", - colorscale="RdYlGn", - zmid=0, - colorbar={"title": "Return (%)"}, - hovertemplate="%{y} %{x}
Return: %{text}", - ) - ) - - fig.update_layout( - title="Monthly Returns Heatmap", - xaxis_title="Month", - yaxis_title="Year", - plot_bgcolor="white", - yaxis={"type": "category"}, - ) - - return fig - - def smoothed_holdings_performance_plot( - self, - windows: list[int] | None = None, - log_scale: bool = False, - ) -> go.Figure: - """Plot NAV_accumulated for smoothed-holding portfolios. - - Builds portfolios with cash positions smoothed by a trailing rolling - mean over the previous ``n`` steps (window size n+1) for n in - ``windows`` (defaults to [0, 1, 2, 3, 4]) and plots their - accumulated NAV curves. - - Args: - windows: List of non-negative integers specifying smoothing steps - to include; defaults to [0, 1, 2, 3, 4]. - log_scale: If True, set the primary y-axis to logarithmic scale. - - Returns: - A Plotly Figure containing one line per requested smoothing level. - """ - if windows is None: - windows = [0, 1, 2, 3, 4] - if not isinstance(windows, list) or not all(isinstance(x, int) and x >= 0 for x in windows): - raise TypeError - - fig = go.Figure() - for n in windows: - pf = self._portfolio if n == 0 else self._portfolio.smoothed_holding(n) - nav = pf.nav_accumulated - fig.add_trace( - go.Scatter( - x=nav["date"], - y=nav["NAV_accumulated"], - mode="lines", - name=f"smooth {n}", - line={"width": 1}, - ) - ) - - self._apply_nav_layout(fig, title="NAV accumulated by smoothed holdings", log_scale=log_scale) - return fig - - def trading_cost_impact_plot(self, max_bps: int = 20) -> go.Figure: - """Plot the Sharpe ratio as a function of one-way trading costs. - - Evaluates the portfolio's annualised Sharpe ratio at each integer - cost level from 0 up to ``max_bps`` basis points and renders the - result as a line chart. The zero-cost Sharpe is shown as a - reference horizontal line so that the reader can quickly gauge - at what cost level the strategy's edge is eroded. - - Args: - max_bps: Maximum one-way trading cost to evaluate, in basis - points. Defaults to 20. - - Returns: - A Plotly Figure with one line trace showing Sharpe vs. cost. - - Raises: - ValueError: If ``max_bps`` is not a positive integer. - """ - impact = self._portfolio.trading_cost_impact(max_bps=max_bps) - - cost_vals = impact["cost_bps"].to_list() - sharpe_vals = impact["sharpe"].to_list() - - # Baseline Sharpe at zero cost - baseline = float(sharpe_vals[0]) if sharpe_vals and sharpe_vals[0] is not None else float("nan") - - fig = go.Figure() - fig.add_trace( - go.Scatter( - x=cost_vals, - y=sharpe_vals, - mode="lines+markers", - name="Sharpe (cost-adjusted)", - marker={"size": 6}, - line={"width": 2, "color": "#1f77b4"}, - ) - ) - if baseline == baseline: # only add when baseline is finite (NaN != NaN) - fig.add_hline( - y=baseline, - line_width=1, - line_dash="dash", - line_color="gray", - annotation_text="0 bps baseline", - annotation_position="top right", - ) - - fig.update_layout( - title=f"Trading Cost Impact on Sharpe Ratio (0\u2013{max_bps} bps)", - hovermode="x unified", - plot_bgcolor="white", - ) - fig.update_xaxes( - title_text="One-way cost (basis points)", - showgrid=True, - gridwidth=0.5, - gridcolor="lightgrey", - dtick=1, - ) - fig.update_yaxes( - title_text="Annualised Sharpe ratio", - showgrid=True, - gridwidth=0.5, - gridcolor="lightgrey", - ) - return fig diff --git a/src/jquantstats/_plots/_portfolio/__init__.py b/src/jquantstats/_plots/_portfolio/__init__.py new file mode 100644 index 00000000..c1f4bc33 --- /dev/null +++ b/src/jquantstats/_plots/_portfolio/__init__.py @@ -0,0 +1,12 @@ +"""Plotting utilities for portfolio analytics using Plotly. + +Renders common portfolio visuals — snapshots, lagged performance curves, +smoothed-holdings curves, rolling risk metrics and lead/lag information ratio +bar charts. Designed for notebook use. +""" + +from __future__ import annotations + +from ._core import PortfolioPlots + +__all__ = ["PortfolioPlots"] diff --git a/src/jquantstats/_plots/_portfolio/_core.py b/src/jquantstats/_plots/_portfolio/_core.py new file mode 100644 index 00000000..d21b43e8 --- /dev/null +++ b/src/jquantstats/_plots/_portfolio/_core.py @@ -0,0 +1,44 @@ +"""The :class:`PortfolioPlots` facade combining the portfolio plot-family mixins.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import plotly.io as pio + +from ._diagnostics import _DiagnosticPlotsMixin +from ._nav import _NavPlotsMixin +from ._rolling import _RollingPortfolioPlotsMixin + +if TYPE_CHECKING: + from .._protocol import PortfolioLike + +# Ensure Plotly works with Marimo (set after imports to satisfy linters) +pio.renderers.default = "plotly_mimetype" + + +class PortfolioPlots( + _NavPlotsMixin, + _RollingPortfolioPlotsMixin, + _DiagnosticPlotsMixin, +): + """Facade for portfolio plots built with Plotly. + + Provides convenience methods to visualize portfolio performance and + diagnostics directly from a Portfolio instance (e.g., snapshot charts, + lagged performance, smoothed holdings, and lead/lag IR). + + Charts are organised into focused mixins: + + - `_NavPlotsMixin` — accumulated-NAV curves: snapshot, lag sweep, + smoothed holdings. + - `_RollingPortfolioPlotsMixin` — rolling Sharpe/volatility and the + per-year Sharpe breakdown. + - `_DiagnosticPlotsMixin` — lead/lag IR, correlation heatmap, monthly + returns calendar, trading-cost impact. + """ + + __slots__ = ("_portfolio",) + + def __init__(self, portfolio: PortfolioLike) -> None: + self._portfolio = portfolio diff --git a/src/jquantstats/_plots/_portfolio/_diagnostics.py b/src/jquantstats/_plots/_portfolio/_diagnostics.py new file mode 100644 index 00000000..44cfa261 --- /dev/null +++ b/src/jquantstats/_plots/_portfolio/_diagnostics.py @@ -0,0 +1,247 @@ +"""Diagnostic charts: lead/lag IR, correlation, monthly calendar, cost impact. + +Split out of the former single-module `_plots/_portfolio.py`; composed into +:class:`PortfolioPlots` by `_core.py`. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import plotly.express as px +import plotly.graph_objects as go +import polars as pl + +if TYPE_CHECKING: + from .._protocol import PortfolioLike + + +class _DiagnosticPlotsMixin: + """Diagnostic charts for :class:`PortfolioPlots`.""" + + __slots__ = () + + _portfolio: PortfolioLike + + def lead_lag_ir_plot(self, start: int = -10, end: int = 19) -> go.Figure: + """Plot Sharpe ratio (IR) across lead/lag variants of the portfolio. + + Builds portfolios with cash positions lagged from ``start`` to ``end`` + (inclusive) and plots a bar chart of the Sharpe ratio for each lag. + Positive lags delay weights; negative lags lead them. + + Args: + start: First lag to include (default: -10). + end: Last lag to include (default: +19). + + Returns: + A Plotly Figure with one bar per lag labeled by the lag value. + """ + if not isinstance(start, int) or not isinstance(end, int): + raise TypeError + if start > end: + start, end = end, start + + lags = list(range(start, end + 1)) + + x_vals: list[int] = [] + y_vals: list[float] = [] + + for n in lags: + pf = self._portfolio if n == 0 else self._portfolio.lag(n) + # Compute Sharpe on the portfolio's returns series + sharpe_val = pf.stats.sharpe().get("returns", float("nan")) + # Ensure a float (Stats returns mapping asset->value) + y_vals.append(float(sharpe_val) if sharpe_val is not None else float("nan")) + x_vals.append(n) + + colors = ["red" if x == 0 else "#1f77b4" for x in x_vals] + fig = go.Figure( + data=[ + go.Bar(x=x_vals, y=y_vals, name="Sharpe by lag", marker_color=colors), + ] + ) + fig.update_layout( + title="Lead/Lag Information Ratio (Sharpe) by Lag", + xaxis_title="Lag (steps)", + yaxis_title="Sharpe ratio", + plot_bgcolor="white", + hovermode="x", + ) + fig.update_xaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey") + fig.update_yaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey") + return fig + + def correlation_heatmap( + self, + frame: pl.DataFrame | None = None, + name: str = "portfolio", + title: str = "Correlation heatmap", + ) -> go.Figure: + """Plot a correlation heatmap for assets and the portfolio. + + If ``frame`` is None, uses the portfolio's prices. The portfolio's + profit series is appended under ``name`` before computing the + correlation matrix. + + Args: + frame: Optional Polars DataFrame with at least the asset price + columns. If omitted, uses ``self._portfolio.prices``. + name: Column name under which to include the portfolio profit. + title: Plot title. + + Returns: + A Plotly Figure rendering the correlation matrix as a heatmap. + """ + if frame is None: + frame = self._portfolio.prices + + corr = self._portfolio.correlation(frame, name=name) + + # Create an interactive heatmap + fig = px.imshow( + corr, + x=corr.columns, + y=corr.columns, + text_auto=".2f", # show correlation values + color_continuous_scale="RdBu_r", # red-blue diverging colormap + zmin=-1, + zmax=1, # correlation range + title=title, + ) + + # Adjust layout + fig.update_layout( + xaxis_title="", yaxis_title="", width=700, height=600, coloraxis_colorbar={"title": "Correlation"} + ) + + return fig + + def monthly_returns_heatmap(self) -> go.Figure: + """Plot a monthly returns calendar heatmap. + + Groups portfolio returns by calendar year and month, then renders a + Plotly heatmap with months on the x-axis and years on the y-axis. + Green cells indicate positive months; red cells indicate negative + months. Cell text shows the percentage return for that month. + + Returns: + A Plotly Figure with a calendar heatmap of monthly returns. + + Raises: + ValueError: If the portfolio has no ``date`` column. + """ + monthly = self._portfolio.monthly + + years = monthly["year"].unique().sort().to_list() + month_names = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] + + z: list[list[float | None]] = [] + text: list[list[str]] = [] + for year in years: + year_data = monthly.filter(pl.col("year") == year) + year_row: list[float | None] = [] + year_text: list[str] = [] + for m in range(1, 13): + month_data = year_data.filter(pl.col("month") == m) + if month_data.is_empty(): + year_row.append(None) + year_text.append("") + else: + ret = float(month_data["returns"][0]) + year_row.append(ret * 100.0) + year_text.append(f"{ret * 100.0:.1f}%") + z.append(year_row) + text.append(year_text) + + fig = go.Figure( + data=go.Heatmap( + z=z, + x=month_names, + y=[str(y) for y in years], + text=text, + texttemplate="%{text}", + colorscale="RdYlGn", + zmid=0, + colorbar={"title": "Return (%)"}, + hovertemplate="%{y} %{x}
Return: %{text}", + ) + ) + + fig.update_layout( + title="Monthly Returns Heatmap", + xaxis_title="Month", + yaxis_title="Year", + plot_bgcolor="white", + yaxis={"type": "category"}, + ) + + return fig + + def trading_cost_impact_plot(self, max_bps: int = 20) -> go.Figure: + """Plot the Sharpe ratio as a function of one-way trading costs. + + Evaluates the portfolio's annualised Sharpe ratio at each integer + cost level from 0 up to ``max_bps`` basis points and renders the + result as a line chart. The zero-cost Sharpe is shown as a + reference horizontal line so that the reader can quickly gauge + at what cost level the strategy's edge is eroded. + + Args: + max_bps: Maximum one-way trading cost to evaluate, in basis + points. Defaults to 20. + + Returns: + A Plotly Figure with one line trace showing Sharpe vs. cost. + + Raises: + ValueError: If ``max_bps`` is not a positive integer. + """ + impact = self._portfolio.trading_cost_impact(max_bps=max_bps) + + cost_vals = impact["cost_bps"].to_list() + sharpe_vals = impact["sharpe"].to_list() + + # Baseline Sharpe at zero cost + baseline = float(sharpe_vals[0]) if sharpe_vals and sharpe_vals[0] is not None else float("nan") + + fig = go.Figure() + fig.add_trace( + go.Scatter( + x=cost_vals, + y=sharpe_vals, + mode="lines+markers", + name="Sharpe (cost-adjusted)", + marker={"size": 6}, + line={"width": 2, "color": "#1f77b4"}, + ) + ) + if baseline == baseline: # only add when baseline is finite (NaN != NaN) + fig.add_hline( + y=baseline, + line_width=1, + line_dash="dash", + line_color="gray", + annotation_text="0 bps baseline", + annotation_position="top right", + ) + + fig.update_layout( + title=f"Trading Cost Impact on Sharpe Ratio (0\u2013{max_bps} bps)", + hovermode="x unified", + plot_bgcolor="white", + ) + fig.update_xaxes( + title_text="One-way cost (basis points)", + showgrid=True, + gridwidth=0.5, + gridcolor="lightgrey", + dtick=1, + ) + fig.update_yaxes( + title_text="Annualised Sharpe ratio", + showgrid=True, + gridwidth=0.5, + gridcolor="lightgrey", + ) + return fig diff --git a/src/jquantstats/_plots/_portfolio/_nav.py b/src/jquantstats/_plots/_portfolio/_nav.py new file mode 100644 index 00000000..5a987a02 --- /dev/null +++ b/src/jquantstats/_plots/_portfolio/_nav.py @@ -0,0 +1,232 @@ +"""NAV-accumulated performance charts: snapshot, lag sweep, holdings smoothing. + +Split out of the former single-module `_plots/_portfolio.py`; composed into +:class:`PortfolioPlots` by `_core.py`. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import plotly.graph_objects as go +from plotly.subplots import make_subplots + +from .._data._styling import _apply_base_layout + +if TYPE_CHECKING: + from .._protocol import PortfolioLike + + +class _NavPlotsMixin: + """Accumulated-NAV charts for :class:`PortfolioPlots`.""" + + __slots__ = () + + _portfolio: PortfolioLike + + def snapshot(self, log_scale: bool = False) -> go.Figure: + """Return a snapshot dashboard of NAV and drawdown. + + When the portfolio has a non-zero ``cost_model.cost_per_unit``, an additional + ``"Net-of-Cost NAV"`` trace is overlaid on the NAV panel showing the + realised NAV path after deducting position-delta trading costs. + + Args: + log_scale (bool, optional): If True, display NAV on a log scale. Defaults to False. + + Returns: + plotly.graph_objects.Figure: A Figure with accumulated NAV (including tilt/timing) + and drawdown shaded area, equipped with a range selector. + """ + # Create subplot grid with domain for stats table + fig = make_subplots( + rows=2, + cols=1, + shared_xaxes=True, + row_heights=[0.66, 0.33], + subplot_titles=["Accumulated Profit", "Drawdown"], + vertical_spacing=0.05, + ) + + # --- Row 1: Cumulative Returns + fig.add_trace( + go.Scatter( + x=self._portfolio.nav_accumulated["date"], + y=self._portfolio.nav_accumulated["NAV_accumulated"], + mode="lines", + name="NAV", + showlegend=False, + ), + row=1, + col=1, + ) + + fig.add_trace( + go.Scatter( + x=self._portfolio.tilt.nav_accumulated["date"], + y=self._portfolio.tilt.nav_accumulated["NAV_accumulated"], + mode="lines", + name="Tilt", + showlegend=False, + ), + row=1, + col=1, + ) + + fig.add_trace( + go.Scatter( + x=self._portfolio.timing.nav_accumulated["date"], + y=self._portfolio.timing.nav_accumulated["NAV_accumulated"], + mode="lines", + name="Timing", + showlegend=False, + ), + row=1, + col=1, + ) + + # Net-of-cost NAV overlay (only when a cost model is active) + if self._portfolio.cost_model.cost_per_unit > 0: + net_nav_df = self._portfolio.net_cost_nav + x_dates = net_nav_df["date"] if "date" in net_nav_df.columns else None + fig.add_trace( + go.Scatter( + x=x_dates, + y=net_nav_df["NAV_accumulated_net"], + mode="lines", + name="Net-of-Cost NAV", + line={"dash": "dash"}, + showlegend=True, + ), + row=1, + col=1, + ) + + fig.add_trace( + go.Scatter( + x=self._portfolio.drawdown["date"], + y=self._portfolio.drawdown["drawdown_pct"], + mode="lines", + fill="tozeroy", + name="Drawdown", + showlegend=False, + ), + row=2, + col=1, + ) + + fig.add_hline(y=0, line_width=1, line_color="gray", row=2, col=1) + + _apply_base_layout(fig, "Performance Dashboard", height=1200) + + fig.update_yaxes(title_text="NAV (accumulated)", row=1, col=1, tickformat=".2s") + fig.update_yaxes(title_text="Drawdown", row=2, col=1, tickformat=".0%") + + if log_scale: + fig.update_yaxes(type="log", row=1, col=1) + # Ensure the first y-axis is explicitly set for environments + # where subplot updates may not propagate to layout alias. + if hasattr(fig.layout, "yaxis"): # pragma: no branch — plotly figures always have .yaxis + fig.layout.yaxis.type = "log" + + return fig + + @staticmethod + def _apply_nav_layout(fig: go.Figure, title: str, log_scale: bool = False) -> None: + """Apply common NAV-accumulated layout to *fig* in-place. + + Configures the plot background, legend, hover mode, x-axis date range + selector, y-axis label, grid lines, and optional logarithmic y-scale. + Shared by `lagged_performance_plot` and + `smoothed_holdings_performance_plot`. + + Args: + fig: The Plotly Figure to configure. + title: Chart title text. + log_scale: If True, set the primary y-axis to logarithmic scale. + """ + _apply_base_layout(fig, title) + fig.update_yaxes(title_text="NAV (accumulated)") + + if log_scale: + fig.update_yaxes(type="log") + if hasattr(fig.layout, "yaxis"): # pragma: no branch — plotly figures always have .yaxis + fig.layout.yaxis.type = "log" + + def lagged_performance_plot(self, lags: list[int] | None = None, log_scale: bool = False) -> go.Figure: + """Plot NAV_accumulated for multiple lagged portfolios. + + Creates a Plotly figure with one line per lag value showing the + accumulated NAV series for the portfolio with cash positions + shifted by that lag. By default, lags [0, 1, 2, 3, 4] are used. + + Args: + lags: A list of integer lags to apply; defaults to [0, 1, 2, 3, 4]. + log_scale: If True, set the primary y-axis to logarithmic scale. + + Returns: + A Plotly Figure containing one trace per requested lag. + """ + if lags is None: + lags = [0, 1, 2, 3, 4] + if not isinstance(lags, list) or not all(isinstance(x, int) for x in lags): + raise TypeError + + fig = go.Figure() + for lag in lags: + pf = self._portfolio if lag == 0 else self._portfolio.lag(lag) + nav = pf.nav_accumulated + fig.add_trace( + go.Scatter( + x=nav["date"], + y=nav["NAV_accumulated"], + mode="lines", + name=f"lag {lag}", + line={"width": 1}, + ) + ) + + self._apply_nav_layout(fig, title="NAV accumulated by lag", log_scale=log_scale) + return fig + + def smoothed_holdings_performance_plot( + self, + windows: list[int] | None = None, + log_scale: bool = False, + ) -> go.Figure: + """Plot NAV_accumulated for smoothed-holding portfolios. + + Builds portfolios with cash positions smoothed by a trailing rolling + mean over the previous ``n`` steps (window size n+1) for n in + ``windows`` (defaults to [0, 1, 2, 3, 4]) and plots their + accumulated NAV curves. + + Args: + windows: List of non-negative integers specifying smoothing steps + to include; defaults to [0, 1, 2, 3, 4]. + log_scale: If True, set the primary y-axis to logarithmic scale. + + Returns: + A Plotly Figure containing one line per requested smoothing level. + """ + if windows is None: + windows = [0, 1, 2, 3, 4] + if not isinstance(windows, list) or not all(isinstance(x, int) and x >= 0 for x in windows): + raise TypeError + + fig = go.Figure() + for n in windows: + pf = self._portfolio if n == 0 else self._portfolio.smoothed_holding(n) + nav = pf.nav_accumulated + fig.add_trace( + go.Scatter( + x=nav["date"], + y=nav["NAV_accumulated"], + mode="lines", + name=f"smooth {n}", + line={"width": 1}, + ) + ) + + self._apply_nav_layout(fig, title="NAV accumulated by smoothed holdings", log_scale=log_scale) + return fig diff --git a/src/jquantstats/_plots/_portfolio/_rolling.py b/src/jquantstats/_plots/_portfolio/_rolling.py new file mode 100644 index 00000000..1592e41c --- /dev/null +++ b/src/jquantstats/_plots/_portfolio/_rolling.py @@ -0,0 +1,155 @@ +"""Rolling-window and per-year risk charts for a portfolio. + +Split out of the former single-module `_plots/_portfolio.py`; composed into +:class:`PortfolioPlots` by `_core.py`. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import plotly.graph_objects as go +import polars as pl + +from .._data._styling import _apply_base_layout + +if TYPE_CHECKING: + from .._protocol import PortfolioLike + + +class _RollingPortfolioPlotsMixin: + """Rolling-window and annual risk charts for :class:`PortfolioPlots`.""" + + __slots__ = () + + _portfolio: PortfolioLike + + @staticmethod + def _validate_window(window: int) -> None: + """Reject a non-positive or non-integer rolling window. + + Args: + window: The candidate rolling-window size. + + Raises: + ValueError: If ``window`` is not a positive integer. + """ + if not isinstance(window, int) or window <= 0: + raise ValueError(f"window must be a positive integer, got {window!r}") # noqa: TRY003 + + @staticmethod + def _line_per_column(rolling: pl.DataFrame) -> go.Figure: + """Render one line trace per non-date column of *rolling*. + + Shared by `rolling_sharpe_plot` and `rolling_volatility_plot`, which + differ only in the metric they fetch and the labels they apply. + + Args: + rolling: A frame with an optional ``date`` column and one column + per asset. + + Returns: + A Figure carrying the traces, with no layout applied yet. + """ + fig = go.Figure() + date_col = rolling["date"] if "date" in rolling.columns else None + for col in rolling.columns: + if col == "date": + continue + fig.add_trace( + go.Scatter( + x=date_col, + y=rolling[col], + mode="lines", + name=col, + line={"width": 1}, + ) + ) + return fig + + def rolling_sharpe_plot(self, window: int = 63) -> go.Figure: + """Plot rolling annualised Sharpe ratio over time. + + Computes the rolling Sharpe for each asset column using the given + window and renders one line per asset. + + Args: + window: Rolling-window size in periods. Defaults to 63. + + Returns: + A Plotly Figure with one trace per asset. + + Raises: + ValueError: If ``window`` is not a positive integer. + """ + self._validate_window(window) + + fig = self._line_per_column(self._portfolio.stats.rolling_sharpe(rolling_period=window)) + fig.add_hline(y=0, line_width=1, line_dash="dash", line_color="gray") + + _apply_base_layout(fig, f"Rolling Sharpe Ratio ({window}-period window)") + fig.update_yaxes(title_text="Sharpe ratio") + return fig + + def rolling_volatility_plot(self, window: int = 63) -> go.Figure: + """Plot rolling annualised volatility over time. + + Computes the rolling volatility for each asset column using the given + window and renders one line per asset. + + Args: + window: Rolling-window size in periods. Defaults to 63. + + Returns: + A Plotly Figure with one trace per asset. + + Raises: + ValueError: If ``window`` is not a positive integer. + """ + self._validate_window(window) + + fig = self._line_per_column(self._portfolio.stats.rolling_volatility(rolling_period=window)) + + _apply_base_layout(fig, f"Rolling Volatility ({window}-period window)") + fig.update_yaxes(title_text="Annualised volatility") + return fig + + def annual_sharpe_plot(self) -> go.Figure: + """Plot annualised Sharpe ratio broken down by calendar year. + + Computes the Sharpe ratio for each calendar year from the portfolio + returns and renders a grouped bar chart with one bar per year per + asset. + + Returns: + A Plotly Figure with one bar group per asset. + """ + breakdown = self._portfolio.stats.annual_breakdown() + + # Extract the sharpe row for each year + sharpe_rows = breakdown.filter(pl.col("metric") == "sharpe") + asset_cols = [c for c in sharpe_rows.columns if c not in ("year", "metric")] + + fig = go.Figure() + for asset in asset_cols: + fig.add_trace( + go.Bar( + x=sharpe_rows["year"], + y=sharpe_rows[asset], + name=asset, + ) + ) + + fig.add_hline(y=0, line_width=1, line_color="gray") + + fig.update_layout( + title="Annual Sharpe Ratio by Year", + barmode="group", + hovermode="x unified", + plot_bgcolor="white", + legend={"orientation": "h", "yanchor": "bottom", "y": 1.02, "xanchor": "right", "x": 1}, + ) + fig.update_yaxes(title_text="Sharpe ratio") + fig.update_xaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey", title_text="Year") + fig.update_yaxes(showgrid=True, gridwidth=0.5, gridcolor="lightgrey") + return fig diff --git a/src/jquantstats/_portfolio_transform.py b/src/jquantstats/_portfolio_transform.py index 1b8efbaa..256c3e68 100644 --- a/src/jquantstats/_portfolio_transform.py +++ b/src/jquantstats/_portfolio_transform.py @@ -36,6 +36,100 @@ def from_cash_position( """Create a Portfolio directly from cash positions aligned with prices.""" ... + # ── Shared construction helpers ──────────────────────────────────────────── + + def _rebuild(self, cash_position: pl.DataFrame, prices: pl.DataFrame | None = None) -> Self: + """Build a new Portfolio of the same concrete type from derived frames. + + Every transform in this mixin ends the same way: hand the derived + frames back to ``from_cash_position`` while carrying ``aum`` and both + cost parameters across unchanged. Centralising that here keeps a new + construction parameter from having to be threaded through each + transform individually. + + Args: + cash_position: The derived cash-position frame. + prices: The derived price frame; defaults to the current prices, + which the lag and smoothing transforms leave untouched. + + Returns: + A new Portfolio of the same concrete type. + """ + return type(self).from_cash_position( + prices=self.prices if prices is None else prices, + cash_position=cash_position, + aum=self.aum, + cost_per_unit=self.cost_per_unit, + cost_bps=self.cost_bps, + ) + + @property + def _numeric_assets(self) -> list[str]: + """Names of the numeric asset columns in the cash-position frame. + + Excludes ``'date'`` and any non-numeric column, so column-wise + transforms touch only the asset series. + + Returns: + The numeric asset column names, in frame order. + """ + return [c for c in self.cashposition.columns if c != "date" and self.cashposition[c].dtype.is_numeric()] + + @staticmethod + def _date_range_mask( + start: date | datetime | str | int | None, + end: date | datetime | str | int | None, + ) -> pl.Expr: + """Build the inclusive ``[start, end]`` filter over the ``'date'`` column. + + Args: + start: Optional inclusive lower bound; no lower bound when None. + end: Optional inclusive upper bound; no upper bound when None. + + Returns: + A boolean Polars expression, ``lit(True)`` when both bounds are None. + """ + cond = pl.lit(True) + if start is not None: + cond = cond & (pl.col("date") >= pl.lit(start)) + if end is not None: + cond = cond & (pl.col("date") <= pl.lit(end)) + return cond + + @staticmethod + def _row_slice_bounds( + start: date | datetime | str | int | None, + end: date | datetime | str | int | None, + height: int, + ) -> tuple[int, int]: + """Resolve integer row bounds into a ``(offset, length)`` slice. + + Used when the portfolio has no ``'date'`` column and truncation falls + back to 0-based row indexing. + + Args: + start: Optional inclusive lower row index; 0 when None. + end: Optional inclusive upper row index; the last row when None. + height: Row count of the frame being sliced. + + Returns: + The ``(offset, length)`` pair to pass to ``DataFrame.slice``. + ``length`` is clamped at 0 so an inverted range yields an empty + frame rather than a negative slice. + + Raises: + IntegerIndexBoundError: When a supplied bound is not an integer. + """ + if start is not None and not isinstance(start, int): + raise IntegerIndexBoundError("start", type(start).__name__) + if end is not None and not isinstance(end, int): + raise IntegerIndexBoundError("end", type(end).__name__) + row_start = int(start) if start is not None else 0 + row_end = int(end) + 1 if end is not None else height + return row_start, max(0, row_end - row_start) + + # ── Transforms ───────────────────────────────────────────────────────────── + def truncate( self, start: date | datetime | str | int | None = None, @@ -71,32 +165,15 @@ def truncate( TypeError: When the portfolio has no ``'date'`` column and a non-integer bound is supplied. """ - has_date = "date" in self.prices.columns - if has_date: - cond = pl.lit(True) - if start is not None: - cond = cond & (pl.col("date") >= pl.lit(start)) - if end is not None: - cond = cond & (pl.col("date") <= pl.lit(end)) + if "date" in self.prices.columns: + cond = self._date_range_mask(start, end) pr = self.prices.filter(cond) cp = self.cashposition.filter(cond) else: - if start is not None and not isinstance(start, int): - raise IntegerIndexBoundError("start", type(start).__name__) - if end is not None and not isinstance(end, int): - raise IntegerIndexBoundError("end", type(end).__name__) - row_start = int(start) if start is not None else 0 - row_end = int(end) + 1 if end is not None else self.prices.height - length = max(0, row_end - row_start) - pr = self.prices.slice(row_start, length) - cp = self.cashposition.slice(row_start, length) - return type(self).from_cash_position( - prices=pr, - cash_position=cp, - aum=self.aum, - cost_per_unit=self.cost_per_unit, - cost_bps=self.cost_bps, - ) + offset, length = self._row_slice_bounds(start, end, self.prices.height) + pr = self.prices.slice(offset, length) + cp = self.cashposition.slice(offset, length) + return self._rebuild(prices=pr, cash_position=cp) def lag(self, n: int) -> Self: """Return a new Portfolio with cash positions lagged by ``n`` steps. @@ -124,15 +201,8 @@ def lag(self, n: int) -> Self: if n == 0: return self - assets = [c for c in self.cashposition.columns if c != "date" and self.cashposition[c].dtype.is_numeric()] - cp_lagged = self.cashposition.with_columns(pl.col(c).shift(n) for c in assets) - return type(self).from_cash_position( - prices=self.prices, - cash_position=cp_lagged, - aum=self.aum, - cost_per_unit=self.cost_per_unit, - cost_bps=self.cost_bps, - ) + cp_lagged = self.cashposition.with_columns(pl.col(c).shift(n) for c in self._numeric_assets) + return self._rebuild(cash_position=cp_lagged) def smoothed_holding(self, n: int) -> Self: """Return a new Portfolio with cash positions smoothed by a rolling mean. @@ -160,18 +230,11 @@ def smoothed_holding(self, n: int) -> Self: if n == 0: return self - assets = [c for c in self.cashposition.columns if c != "date" and self.cashposition[c].dtype.is_numeric()] window = n + 1 cp_smoothed = self.cashposition.with_columns( - pl.col(c).rolling_mean(window_size=window, min_samples=1).alias(c) for c in assets - ) - return type(self).from_cash_position( - prices=self.prices, - cash_position=cp_smoothed, - aum=self.aum, - cost_per_unit=self.cost_per_unit, - cost_bps=self.cost_bps, + pl.col(c).rolling_mean(window_size=window, min_samples=1).alias(c) for c in self._numeric_assets ) + return self._rebuild(cash_position=cp_smoothed) # ── Utility ──────────────────────────────────────────────────────────────── diff --git a/src/jquantstats/_stats/_capture.py b/src/jquantstats/_stats/_capture.py new file mode 100644 index 00000000..dd0dbad2 --- /dev/null +++ b/src/jquantstats/_stats/_capture.py @@ -0,0 +1,112 @@ +"""Up- and down-market capture ratios. + +Split out of `_reporting.py`: capture ratios are the only metrics in that +module that take an explicit benchmark series as an argument rather than +reading the benchmark off `Data`, and they share a computation that is worth +stating once. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import polars as pl + +if TYPE_CHECKING: + from ..data import Data + + +class _CaptureStatsMixin: + """Mixin providing up-market and down-market capture ratios.""" + + _data: Data + all: pl.DataFrame + + if TYPE_CHECKING: + from .._protocol import DataLike + + data: DataLike + + @staticmethod + def _geometric_mean(series: pl.Series) -> float: + """Geometric mean return of *series*: ``prod(1 + r)^(1/n) - 1``. + + Args: + series: A non-empty return series. + + Returns: + The per-period geometric mean return. + """ + return float(float((series + 1.0).product()) ** (1.0 / len(series)) - 1.0) + + def _capture_ratio(self, benchmark: pl.Series, mask: pl.Series) -> dict[str, float]: + """Ratio of each asset's geometric mean to the benchmark's, over *mask*. + + Shared by `up_capture` and `down_capture`, which differ only in the + sign of the benchmark periods they select. + + Args: + benchmark: Benchmark return series aligned row-by-row with the data. + mask: Boolean series selecting the periods to measure over. + + Returns: + dict[str, float]: Capture ratio per asset; ``float("nan")`` where + the benchmark or the asset has nothing usable in the selected + periods. + """ + bench_selected = benchmark.filter(mask).drop_nulls() + # A benchmark with no periods of this sign makes capture undefined for every asset. + if bench_selected.is_empty(): + return {col: float("nan") for col, _ in self._data.items()} + bench_geom = self._geometric_mean(bench_selected) + if bench_geom == 0.0: # pragma: no cover + return {col: float("nan") for col, _ in self._data.items()} + + result: dict[str, float] = {} + for col, series in self._data.items(): + strat_selected = series.filter(mask).drop_nulls() + # An asset may have no usable returns during the selected periods after null filtering. + if strat_selected.is_empty(): + result[col] = float("nan") + else: + result[col] = self._geometric_mean(strat_selected) / bench_geom + return result + + def up_capture(self, benchmark: pl.Series) -> dict[str, float]: + """Up-market capture ratio relative to an explicit benchmark series. + + Measures the fraction of the benchmark's upside that the strategy + captures. A value greater than 1.0 means the strategy outperformed + the benchmark in rising markets. + + Args: + benchmark: Benchmark return series aligned row-by-row with the data. + + Returns: + dict[str, float]: Up capture ratio per asset. + + Returns NaN when: + Entries are ``float("nan")`` when the benchmark has no positive + periods, its up-market geometric mean is zero, or an asset has no + usable returns during those periods. + """ + return self._capture_ratio(benchmark, benchmark > 0) + + def down_capture(self, benchmark: pl.Series) -> dict[str, float]: + """Down-market capture ratio relative to an explicit benchmark series. + + A value less than 1.0 means the strategy lost less than the benchmark + in falling markets (a desirable property). + + Args: + benchmark: Benchmark return series aligned row-by-row with the data. + + Returns: + dict[str, float]: Down capture ratio per asset. + + Returns NaN when: + Entries are ``float("nan")`` when the benchmark has no negative + periods, its down-market geometric mean is zero, or an asset has no + usable returns during those periods. + """ + return self._capture_ratio(benchmark, benchmark < 0) diff --git a/src/jquantstats/_stats/_reporting.py b/src/jquantstats/_stats/_reporting.py index c2442d2e..75456b69 100644 --- a/src/jquantstats/_stats/_reporting.py +++ b/src/jquantstats/_stats/_reporting.py @@ -1,8 +1,12 @@ -"""Temporal reporting, capture ratios, and summary statistics.""" +"""Temporal reporting metrics. + +Capture ratios live in `_capture.py` and the aggregating `summary` / +`annual_breakdown` pair in `_summary.py`; all three are composed into `Stats`. +""" from __future__ import annotations -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING import polars as pl @@ -16,18 +20,13 @@ class _ReportingStatsMixin: - """Mixin providing temporal, capture, and summary reporting metrics. + """Mixin providing temporal reporting metrics. - Covers: periods per year, average drawdown, Calmar ratio, recovery factor, - max drawdown duration, monthly win rate, up/down capture ratios, annual - breakdown, and summary statistics table. + Covers: periods per year, average drawdown, CAGR, expected return, RAR, + Calmar ratio, recovery factor, max drawdown duration, and monthly win rate. Cross-mixin dependencies: - - _BasicStatsMixin: avg_return, avg_win, avg_loss, win_rate, profit_factor, - payoff_ratio, best, worst, volatility, skew, kurtosis, value_at_risk, - conditional_value_at_risk, exposure - - _RiskStatsMixin: sharpe - - _DrawdownMixin: max_drawdown + - _BasicStatsMixin: exposure """ _data: Data @@ -38,56 +37,9 @@ class _ReportingStatsMixin: data: DataLike - def avg_return(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def avg_win(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def avg_loss(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def win_rate(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def profit_factor(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def payoff_ratio(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def best(self) -> dict[str, float | None]: - """Defined on _BasicStatsMixin.""" - - def worst(self) -> dict[str, float | None]: - """Defined on _BasicStatsMixin.""" - - def volatility(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def sharpe(self) -> dict[str, float]: - """Defined on _RiskStatsMixin.""" - - def skew(self) -> dict[str, int | float | None]: - """Defined on _BasicStatsMixin.""" - - def kurtosis(self) -> dict[str, int | float | None]: - """Defined on _BasicStatsMixin.""" - - def value_at_risk(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def conditional_value_at_risk(self) -> dict[str, float]: - """Defined on _BasicStatsMixin.""" - - def max_drawdown(self) -> dict[str, float]: - """Defined on _DrawdownMixin.""" - def exposure(self) -> dict[str, float]: """Defined on _BasicStatsMixin.""" - # ── Temporal & reporting ────────────────────────────────────────────────── - @property def periods_per_year(self) -> float: """Estimate the number of periods per year from the data index spacing. @@ -377,216 +329,3 @@ def monthly_win_rate(self) -> dict[str, float]: n_positive = int((monthly["monthly_return"] > 0).sum()) result[col] = n_positive / n_total return result - - # ── Capture ratios ──────────────────────────────────────────────────────── - - def up_capture(self, benchmark: pl.Series) -> dict[str, float]: - """Up-market capture ratio relative to an explicit benchmark series. - - Measures the fraction of the benchmark's upside that the strategy - captures. A value greater than 1.0 means the strategy outperformed - the benchmark in rising markets. - - Args: - benchmark: Benchmark return series aligned row-by-row with the data. - - Returns: - dict[str, float]: Up capture ratio per asset. - - Returns NaN when: - Entries are ``float("nan")`` when the benchmark has no positive - periods, its up-market geometric mean is zero, or an asset has no - usable returns during those periods. - """ - up_mask = benchmark > 0 - bench_up = benchmark.filter(up_mask).drop_nulls() - # A benchmark with no positive periods makes up-capture undefined for every asset. - if bench_up.is_empty(): - return {col: float("nan") for col, _ in self._data.items()} - bench_geom = float((bench_up + 1.0).product()) ** (1.0 / len(bench_up)) - 1.0 - if bench_geom == 0.0: # pragma: no cover - return {col: float("nan") for col, _ in self._data.items()} - result: dict[str, float] = {} - for col, series in self._data.items(): - strat_up = series.filter(up_mask).drop_nulls() - # An asset may have no usable returns during the benchmark's up periods after null filtering. - if strat_up.is_empty(): - result[col] = float("nan") - else: - strat_geom = float((strat_up + 1.0).product()) ** (1.0 / len(strat_up)) - 1.0 - result[col] = strat_geom / bench_geom - return result - - def down_capture(self, benchmark: pl.Series) -> dict[str, float]: - """Down-market capture ratio relative to an explicit benchmark series. - - A value less than 1.0 means the strategy lost less than the benchmark - in falling markets (a desirable property). - - Args: - benchmark: Benchmark return series aligned row-by-row with the data. - - Returns: - dict[str, float]: Down capture ratio per asset. - - Returns NaN when: - Entries are ``float("nan")`` when the benchmark has no negative - periods, its down-market geometric mean is zero, or an asset has no - usable returns during those periods. - """ - down_mask = benchmark < 0 - bench_down = benchmark.filter(down_mask).drop_nulls() - # A benchmark with no negative periods makes down-capture undefined for every asset. - if bench_down.is_empty(): - return {col: float("nan") for col, _ in self._data.items()} - bench_geom = float((bench_down + 1.0).product()) ** (1.0 / len(bench_down)) - 1.0 - if bench_geom == 0.0: # pragma: no cover - return {col: float("nan") for col, _ in self._data.items()} - result: dict[str, float] = {} - for col, series in self._data.items(): - strat_down = series.filter(down_mask).drop_nulls() - # An asset may have no usable returns during the benchmark's down periods after null filtering. - if strat_down.is_empty(): - result[col] = float("nan") - else: - strat_geom = float((strat_down + 1.0).product()) ** (1.0 / len(strat_down)) - 1.0 - result[col] = strat_geom / bench_geom - return result - - # ── Summary & breakdown ──────────────────────────────────────────────────── - - def annual_breakdown(self) -> pl.DataFrame: - """Summary statistics broken down by calendar year. - - Groups the data by calendar year using the date index, computes a - full `summary` for each year, and stacks the results with an - additional ``year`` column. - - Returns: - pl.DataFrame: Columns ``year``, ``metric``, one per asset, sorted - by ``year``. - - Raises: - ValueError: If the data has no date index. - """ - all_df = self.all - date_col_name = self._data.date_col[0] if self._data.date_col else None - has_temporal = date_col_name is not None and all_df[date_col_name].dtype.is_temporal() - - if not has_temporal: - return self._annual_breakdown_integer(all_df) - if date_col_name is None: # unreachable: has_temporal guarantees non-None # pragma: no cover - return pl.DataFrame() # pragma: no cover - return self._annual_breakdown_temporal(all_df, date_col_name) - - def _summary_frame(self, sub_all: pl.DataFrame, index_cols: list[str], label: int) -> pl.DataFrame: - """Compute a `summary` for one sub-period and tag it with a ``year`` label. - - Args: - sub_all: The combined (index + returns + benchmark) rows for the period. - index_cols: Column name(s) to use as the sub-period's date index. - label: Value written to the ``year`` column (calendar year or chunk ordinal). - - Returns: - The summary DataFrame with an added ``year`` column. - """ - # Construct the sub-period Data via type(self._data) rather than importing - # the concrete class: a lazy `from ..data import Data` would put the upper - # layer back into this subpackage's import graph, which is exactly the - # coupling _protocol.py exists to prevent. Mirrors the type(self) call below. - data_factory = cast(Any, type(self._data)) - sub_returns = sub_all.select(self._data.returns.columns) - sub_benchmark = sub_all.select(self._data.benchmark.columns) if self._data.benchmark is not None else None - sub_data = data_factory(returns=sub_returns, index=sub_all.select(index_cols), benchmark=sub_benchmark) - summary: pl.DataFrame = cast(Any, type(self))(sub_data).summary() - return summary.with_columns(pl.lit(label).alias("year")) - - @staticmethod - def _order_breakdown(result: pl.DataFrame) -> pl.DataFrame: - """Reorder breakdown columns so ``year`` and ``metric`` lead.""" - ordered = ["year", "metric", *[c for c in result.columns if c not in ("year", "metric")]] - return result.select(ordered) - - def _annual_breakdown_integer(self, all_df: pl.DataFrame) -> pl.DataFrame: - """Break down by fixed row chunks (~one year each) for an integer index.""" - chunk = round(self._data._periods_per_year) - total = all_df.height - frames: list[pl.DataFrame] = [] - for i, start in enumerate(range(0, total, chunk), start=1): - chunk_all = all_df.slice(start, chunk) - if chunk_all.height < max(5, chunk // 4): - continue - frames.append(self._summary_frame(chunk_all, self._data.date_col, i)) - if not frames: - return pl.DataFrame() - return self._order_breakdown(pl.concat(frames)) - - def _annual_breakdown_temporal(self, all_df: pl.DataFrame, date_col_name: str) -> pl.DataFrame: - """Break down by calendar year for a temporal index.""" - years = all_df[date_col_name].dt.year().unique().sort().to_list() - frames: list[pl.DataFrame] = [] - for year in years: - year_all = all_df.filter(pl.col(date_col_name).dt.year() == year) - if year_all.height < 2: - continue - frames.append(self._summary_frame(year_all, [date_col_name], year)) - if not frames: - asset_cols = list(self._data.returns.columns) - schema: dict[str, type[pl.DataType]] = { - "year": pl.Int32, - "metric": pl.String, - **dict.fromkeys(asset_cols, pl.Float64), - } - return pl.DataFrame(schema=schema) - return self._order_breakdown(pl.concat(frames)) - - def summary(self) -> pl.DataFrame: - """Summary statistics for each asset as a tidy DataFrame. - - Each row is one metric; each column beyond ``metric`` is one asset. - - Returns: - pl.DataFrame: A DataFrame with a ``metric`` column followed by one - column per asset. - - Returns NaN when: - Cells are ``float("nan")`` when the underlying metric is unavailable - for the data (e.g. no temporal index or no benchmark). - """ - assets = [col for col, _ in self._data.items()] - - def _safe(fn: Any) -> dict[str, Any]: - """Call *fn()* and return its result; return NaN for each asset on any exception.""" - try: - result: dict[str, Any] = fn() - except Exception: - return dict.fromkeys(assets, float("nan")) - return result - - metrics: dict[str, dict[str, Any]] = { - "avg_return": _safe(self.avg_return), - "avg_win": _safe(self.avg_win), - "avg_loss": _safe(self.avg_loss), - "win_rate": _safe(self.win_rate), - "profit_factor": _safe(self.profit_factor), - "payoff_ratio": _safe(self.payoff_ratio), - "monthly_win_rate": _safe(self.monthly_win_rate), - "best": _safe(self.best), - "worst": _safe(self.worst), - "volatility": _safe(self.volatility), - "sharpe": _safe(self.sharpe), - "skew": _safe(self.skew), - "kurtosis": _safe(self.kurtosis), - "value_at_risk": _safe(self.value_at_risk), - "conditional_value_at_risk": _safe(self.conditional_value_at_risk), - "max_drawdown": _safe(self.max_drawdown), - "avg_drawdown": _safe(self.avg_drawdown), - "max_drawdown_duration": _safe(self.max_drawdown_duration), - "calmar": _safe(self.calmar), - "recovery_factor": _safe(self.recovery_factor), - } - - rows: list[dict[str, Any]] = [ - {"metric": name, **{asset: values.get(asset) for asset in assets}} for name, values in metrics.items() - ] - return pl.DataFrame(rows) diff --git a/src/jquantstats/_stats/_stats.py b/src/jquantstats/_stats/_stats.py index 286f7a46..7d74c1fe 100644 --- a/src/jquantstats/_stats/_stats.py +++ b/src/jquantstats/_stats/_stats.py @@ -13,9 +13,10 @@ class that combines five mixin classes: beta, information ratio, Treynor). - `_DrawdownMixin` — cumulative returns, drawdown series, max drawdown, and per-episode drawdown details. -- `_ReportingStatsMixin` — temporal - reporting, Calmar, recovery factor, capture ratios, annual breakdown, and - summary. +- `_ReportingStatsMixin` — temporal reporting: CAGR, expected return, RAR, + Calmar, recovery factor, drawdown duration, monthly win rate. +- `_CaptureStatsMixin` — up- and down-market capture ratios. +- `_SummaryStatsMixin` — the tidy `summary` table and its `annual_breakdown`. - `_PeriodicReportingMixin` — period-bucketed tables: monthly-returns pivot, distribution across calendar frequencies, benchmark comparison, worst-N periods. - `_RollingStatsMixin` — rolling-window @@ -36,6 +37,7 @@ class that combines five mixin classes: from ._basic import _BasicStatsMixin from ._benchmark import _BenchmarkStatsMixin +from ._capture import _CaptureStatsMixin from ._concentration import _ConcentrationStatsMixin from ._core import ( _drawdown_series, @@ -56,6 +58,7 @@ class that combines five mixin classes: from ._periodic import _PeriodicReportingMixin from ._reporting import _ReportingStatsMixin from ._rolling import _RollingStatsMixin +from ._summary import _SummaryStatsMixin if TYPE_CHECKING: from ..data import Data @@ -81,6 +84,8 @@ class Stats( _BenchmarkStatsMixin, _DrawdownMixin, _ReportingStatsMixin, + _CaptureStatsMixin, + _SummaryStatsMixin, _PeriodicReportingMixin, _RollingStatsMixin, _MonteCarloStatsMixin, @@ -109,6 +114,8 @@ class Stats( - `_BenchmarkStatsMixin` - `_DrawdownMixin` - `_ReportingStatsMixin` + - `_CaptureStatsMixin` + - `_SummaryStatsMixin` - `_PeriodicReportingMixin` - `_RollingStatsMixin` - `_MonteCarloStatsMixin` diff --git a/src/jquantstats/_stats/_summary.py b/src/jquantstats/_stats/_summary.py new file mode 100644 index 00000000..8f499ac0 --- /dev/null +++ b/src/jquantstats/_stats/_summary.py @@ -0,0 +1,233 @@ +"""The tidy `summary` table and its calendar-year breakdown. + +Split out of `_reporting.py`. These are the aggregating metrics: they call +across every other mixin rather than computing anything themselves, which is +why they carry the largest block of cross-mixin type stubs in the package. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, cast + +import polars as pl + +if TYPE_CHECKING: + from ..data import Data + + +class _SummaryStatsMixin: + """Mixin providing the `summary` table and `annual_breakdown`. + + Cross-mixin dependencies: + - _BasicStatsMixin: avg_return, avg_win, avg_loss, win_rate, profit_factor, + payoff_ratio, best, worst, volatility, skew, kurtosis, value_at_risk, + conditional_value_at_risk + - _RiskStatsMixin: sharpe + - _DrawdownMixin: max_drawdown + - _ReportingStatsMixin: monthly_win_rate, avg_drawdown, + max_drawdown_duration, calmar, recovery_factor + """ + + _data: Data + all: pl.DataFrame + + if TYPE_CHECKING: + from .._protocol import DataLike + + data: DataLike + + def avg_return(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def avg_win(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def avg_loss(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def win_rate(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def profit_factor(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def payoff_ratio(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def best(self) -> dict[str, float | None]: + """Defined on _BasicStatsMixin.""" + + def worst(self) -> dict[str, float | None]: + """Defined on _BasicStatsMixin.""" + + def volatility(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def sharpe(self) -> dict[str, float]: + """Defined on _RiskStatsMixin.""" + + def skew(self) -> dict[str, int | float | None]: + """Defined on _BasicStatsMixin.""" + + def kurtosis(self) -> dict[str, int | float | None]: + """Defined on _BasicStatsMixin.""" + + def value_at_risk(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def conditional_value_at_risk(self) -> dict[str, float]: + """Defined on _BasicStatsMixin.""" + + def max_drawdown(self) -> dict[str, float]: + """Defined on _DrawdownMixin.""" + + def monthly_win_rate(self) -> dict[str, float]: + """Defined on _ReportingStatsMixin.""" + + def avg_drawdown(self) -> dict[str, float]: + """Defined on _ReportingStatsMixin.""" + + def max_drawdown_duration(self) -> dict[str, float | int | None]: + """Defined on _ReportingStatsMixin.""" + + def calmar(self) -> dict[str, float]: + """Defined on _ReportingStatsMixin.""" + + def recovery_factor(self) -> dict[str, float]: + """Defined on _ReportingStatsMixin.""" + + def annual_breakdown(self) -> pl.DataFrame: + """Summary statistics broken down by calendar year. + + Groups the data by calendar year using the date index, computes a + full `summary` for each year, and stacks the results with an + additional ``year`` column. + + Returns: + pl.DataFrame: Columns ``year``, ``metric``, one per asset, sorted + by ``year``. + + Raises: + ValueError: If the data has no date index. + """ + all_df = self.all + date_col_name = self._data.date_col[0] if self._data.date_col else None + has_temporal = date_col_name is not None and all_df[date_col_name].dtype.is_temporal() + + if not has_temporal: + return self._annual_breakdown_integer(all_df) + if date_col_name is None: # unreachable: has_temporal guarantees non-None # pragma: no cover + return pl.DataFrame() # pragma: no cover + return self._annual_breakdown_temporal(all_df, date_col_name) + + def _summary_frame(self, sub_all: pl.DataFrame, index_cols: list[str], label: int) -> pl.DataFrame: + """Compute a `summary` for one sub-period and tag it with a ``year`` label. + + Args: + sub_all: The combined (index + returns + benchmark) rows for the period. + index_cols: Column name(s) to use as the sub-period's date index. + label: Value written to the ``year`` column (calendar year or chunk ordinal). + + Returns: + The summary DataFrame with an added ``year`` column. + """ + # Construct the sub-period Data via type(self._data) rather than importing + # the concrete class: a lazy `from ..data import Data` would put the upper + # layer back into this subpackage's import graph, which is exactly the + # coupling _protocol.py exists to prevent. Mirrors the type(self) call below. + data_factory = cast(Any, type(self._data)) + sub_returns = sub_all.select(self._data.returns.columns) + sub_benchmark = sub_all.select(self._data.benchmark.columns) if self._data.benchmark is not None else None + sub_data = data_factory(returns=sub_returns, index=sub_all.select(index_cols), benchmark=sub_benchmark) + summary: pl.DataFrame = cast(Any, type(self))(sub_data).summary() + return summary.with_columns(pl.lit(label).alias("year")) + + @staticmethod + def _order_breakdown(result: pl.DataFrame) -> pl.DataFrame: + """Reorder breakdown columns so ``year`` and ``metric`` lead.""" + ordered = ["year", "metric", *[c for c in result.columns if c not in ("year", "metric")]] + return result.select(ordered) + + def _annual_breakdown_integer(self, all_df: pl.DataFrame) -> pl.DataFrame: + """Break down by fixed row chunks (~one year each) for an integer index.""" + chunk = round(self._data._periods_per_year) + total = all_df.height + frames: list[pl.DataFrame] = [] + for i, start in enumerate(range(0, total, chunk), start=1): + chunk_all = all_df.slice(start, chunk) + if chunk_all.height < max(5, chunk // 4): + continue + frames.append(self._summary_frame(chunk_all, self._data.date_col, i)) + if not frames: + return pl.DataFrame() + return self._order_breakdown(pl.concat(frames)) + + def _annual_breakdown_temporal(self, all_df: pl.DataFrame, date_col_name: str) -> pl.DataFrame: + """Break down by calendar year for a temporal index.""" + years = all_df[date_col_name].dt.year().unique().sort().to_list() + frames: list[pl.DataFrame] = [] + for year in years: + year_all = all_df.filter(pl.col(date_col_name).dt.year() == year) + if year_all.height < 2: + continue + frames.append(self._summary_frame(year_all, [date_col_name], year)) + if not frames: + asset_cols = list(self._data.returns.columns) + schema: dict[str, type[pl.DataType]] = { + "year": pl.Int32, + "metric": pl.String, + **dict.fromkeys(asset_cols, pl.Float64), + } + return pl.DataFrame(schema=schema) + return self._order_breakdown(pl.concat(frames)) + + def summary(self) -> pl.DataFrame: + """Summary statistics for each asset as a tidy DataFrame. + + Each row is one metric; each column beyond ``metric`` is one asset. + + Returns: + pl.DataFrame: A DataFrame with a ``metric`` column followed by one + column per asset. + + Returns NaN when: + Cells are ``float("nan")`` when the underlying metric is unavailable + for the data (e.g. no temporal index or no benchmark). + """ + assets = [col for col, _ in self._data.items()] + + def _safe(fn: Any) -> dict[str, Any]: + """Call *fn()* and return its result; return NaN for each asset on any exception.""" + try: + result: dict[str, Any] = fn() + except Exception: + return dict.fromkeys(assets, float("nan")) + return result + + metrics: dict[str, dict[str, Any]] = { + "avg_return": _safe(self.avg_return), + "avg_win": _safe(self.avg_win), + "avg_loss": _safe(self.avg_loss), + "win_rate": _safe(self.win_rate), + "profit_factor": _safe(self.profit_factor), + "payoff_ratio": _safe(self.payoff_ratio), + "monthly_win_rate": _safe(self.monthly_win_rate), + "best": _safe(self.best), + "worst": _safe(self.worst), + "volatility": _safe(self.volatility), + "sharpe": _safe(self.sharpe), + "skew": _safe(self.skew), + "kurtosis": _safe(self.kurtosis), + "value_at_risk": _safe(self.value_at_risk), + "conditional_value_at_risk": _safe(self.conditional_value_at_risk), + "max_drawdown": _safe(self.max_drawdown), + "avg_drawdown": _safe(self.avg_drawdown), + "max_drawdown_duration": _safe(self.max_drawdown_duration), + "calmar": _safe(self.calmar), + "recovery_factor": _safe(self.recovery_factor), + } + + rows: list[dict[str, Any]] = [ + {"metric": name, **{asset: values.get(asset) for asset in assets}} for name, values in metrics.items() + ] + return pl.DataFrame(rows) diff --git a/tests/test_jquantstats/test_migration/test_plot.py b/tests/test_jquantstats/test_migration/test_plot.py index 3f4ec6ff..44b0db98 100644 --- a/tests/test_jquantstats/test_migration/test_plot.py +++ b/tests/test_jquantstats/test_migration/test_plot.py @@ -128,6 +128,38 @@ def test_plot_rolling_beta_no_benchmark_raises(data_no_benchmark): data_no_benchmark.plots.rolling_beta() +def test_plot_rolling_beta_falls_back_to_frame_columns_without_returns(data): + """rolling_beta derives assets from `all` when the data exposes no `returns`. + + `_beta_assets` prefers an explicit ``returns`` frame and otherwise takes + every column of ``all`` that is neither the date nor the benchmark. That + fallback is reachable through the `DataLike` protocol — an implementation + need not carry a ``returns`` attribute — so it is exercised here with a + proxy that hides it. + """ + + class _NoReturnsData: + """DataLike proxy exposing everything except ``returns``.""" + + def __init__(self, inner): + self._inner = inner + + def __getattr__(self, item): + """Delegate every attribute except ``returns`` to the wrapped data.""" + if item == "returns": + raise AttributeError(item) + return getattr(self._inner, item) + + plots = type(data.plots)(_NoReturnsData(data)) + fig = plots.rolling_beta(rolling_period=63, rolling_period2=None) + + date_col = data.all.columns[0] + bench_col = data.benchmark.columns[0] + expected = [c for c in data.all.columns if c not in (date_col, bench_col)] + assert len(fig.data) == len(expected) + assert [trace.name for trace in fig.data] == [f"{asset} (63d)" for asset in expected] + + def test_plot_compare_no_benchmark_raises(data_no_benchmark): """Compare raises AttributeError when no benchmark is attached.""" with pytest.raises(AttributeError): diff --git a/tests/test_jquantstats/test_properties.py b/tests/test_jquantstats/test_properties.py index 360ac13e..b4e23737 100644 --- a/tests/test_jquantstats/test_properties.py +++ b/tests/test_jquantstats/test_properties.py @@ -17,6 +17,11 @@ - Average return is positive when every return is strictly positive - CAGR is non-negative when all returns are non-negative - lag(0) is a no-op on Portfolio: cashposition is unchanged +- Ordering invariants: worst ≤ best, CVaR ≤ VaR, avg_drawdown ≥ max_drawdown +- Bounded-ratio invariants: exposure and monthly_win_rate lie in [0, 1] +- Capture ratios against the series itself are exactly 1 (self-capture identity) +- summary() agrees cell-for-cell with the metric methods it tabulates +- truncate(None, None) and smoothed_holding(0) are no-ops on Portfolio """ import math @@ -370,3 +375,184 @@ def test_lag_zero_is_identity_property(data_tuple: tuple[list[float], list[float pt.assert_frame_equal(lagged.cashposition, portfolio.cashposition) pt.assert_frame_equal(lagged.prices, portfolio.prices) assert lagged.aum == portfolio.aum + + +# ── Ordering invariants ────────────────────────────────────────────────────── + + +@pytest.mark.property +@given(returns=st.lists(_general_returns, min_size=10, max_size=50)) +@settings(max_examples=100) +def test_worst_never_exceeds_best(returns: list[float]) -> None: + """The worst single-period return is never greater than the best. + + Both are order statistics of the same series, so ``min ≤ max`` holds by + definition for any non-empty series. + """ + data = Data.from_returns(returns=_make_returns_df(returns)) + best = data.stats.best()["Asset"] + worst = data.stats.worst()["Asset"] + assert worst <= best, f"Expected worst <= best, got worst={worst}, best={best}" + + +@pytest.mark.property +@given(returns=st.lists(_general_returns, min_size=10, max_size=50)) +@settings(max_examples=100) +def test_conditional_var_at_least_as_extreme_as_var(returns: list[float]) -> None: + """CVaR ≤ VaR: the mean of the tail is at least as bad as the tail's edge. + + Conditional value-at-risk averages the losses beyond the VaR threshold. + Every observation in that average is ≤ the threshold, so the mean cannot + exceed it. Both are expressed as signed returns, so "worse" means smaller. + """ + data = Data.from_returns(returns=_make_returns_df(returns)) + var = data.stats.value_at_risk()["Asset"] + cvar = data.stats.conditional_value_at_risk()["Asset"] + assume(not math.isnan(var) and not math.isnan(cvar)) + assert cvar <= var + TOL_PINNED, f"Expected cvar <= var, got cvar={cvar}, var={var}" + + +@pytest.mark.property +@given(returns=st.lists(_general_returns, min_size=10, max_size=50)) +@settings(max_examples=100) +def test_avg_drawdown_is_shallower_than_max_drawdown(returns: list[float]) -> None: + """The average drawdown is never deeper than the maximum drawdown. + + Both are reported as negative fractions, so "shallower" means greater: + the mean of the underwater series cannot be below its own minimum. + """ + data = Data.from_returns(returns=_make_returns_df(returns)) + avg_dd = data.stats.avg_drawdown()["Asset"] + max_dd = data.stats.max_drawdown()["Asset"] + assume(not math.isnan(avg_dd) and not math.isnan(max_dd)) + assert avg_dd >= max_dd - TOL_PINNED, f"Expected avg_drawdown >= max_drawdown, got {avg_dd} < {max_dd}" + + +# ── Bounded ratios ─────────────────────────────────────────────────────────── + + +@pytest.mark.property +@given(returns=st.lists(_general_returns, min_size=10, max_size=50)) +@settings(max_examples=100) +def test_exposure_in_unit_interval(returns: list[float]) -> None: + """Exposure is a fraction of observed periods, so it lies in [0, 1].""" + data = Data.from_returns(returns=_make_returns_df(returns)) + exposure = data.stats.exposure()["Asset"] + assert 0.0 <= exposure <= 1.0, f"Expected exposure in [0, 1], got {exposure}" + + +@pytest.mark.property +@given(returns=st.lists(_general_returns, min_size=40, max_size=120)) +@settings(max_examples=50) +def test_monthly_win_rate_in_unit_interval(returns: list[float]) -> None: + """Monthly win rate is a fraction of calendar months, so it lies in [0, 1]. + + The series is generated with one observation per day from a fixed start, + so a 40–120 element run always spans at least two calendar months. + """ + data = Data.from_returns(returns=_make_returns_df(returns)) + rate = data.stats.monthly_win_rate()["Asset"] + assume(not math.isnan(rate)) + assert 0.0 <= rate <= 1.0, f"Expected monthly_win_rate in [0, 1], got {rate}" + + +# ── Capture ratios ─────────────────────────────────────────────────────────── + + +@pytest.mark.property +@given(returns=st.lists(_general_returns, min_size=10, max_size=50)) +@settings(max_examples=100) +def test_self_capture_is_unity(returns: list[float]) -> None: + """Capturing against oneself gives exactly 1, in both up and down markets. + + Up- and down-capture divide the asset's geometric mean over the selected + periods by the benchmark's over the same periods. When the benchmark *is* + the asset, numerator and denominator are the same number. + """ + df = _make_returns_df(returns) + data = Data.from_returns(returns=df) + own = df["Asset"] + + for name, ratio in (("up", data.stats.up_capture(own)), ("down", data.stats.down_capture(own))): + value = ratio["Asset"] + # NaN is the documented result when the benchmark has no period of that + # sign — nothing to capture, so there is no ratio to check. + if not math.isnan(value): + assert value == pytest.approx(1.0, rel=TOL_PARITY), f"Expected {name}_capture == 1, got {value}" + + +# ── The summary table agrees with its sources ──────────────────────────────── + + +@pytest.mark.property +@given(returns=st.lists(_general_returns, min_size=40, max_size=120)) +@settings(max_examples=25) +def test_summary_matches_the_metric_methods_it_tabulates(returns: list[float]) -> None: + """Every summary row reproduces the metric method of the same name. + + `summary` is a facade over the other mixins; nothing stops a row from + drifting to a stale or wrongly-named source. Comparing the table against + a direct call per metric is what makes that drift fail the suite. + + Where the direct call raises — an all-zero series divides by zero in + `win_rate`, for instance — the documented contract is that `summary` + substitutes NaN rather than propagating, so that is asserted instead. + """ + data = Data.from_returns(returns=_make_returns_df(returns)) + stats = data.stats + table = stats.summary() + tabulated = dict(zip(table["metric"].to_list(), table["Asset"].to_list(), strict=True)) + + for metric in ("volatility", "win_rate", "best", "worst", "max_drawdown", "avg_drawdown", "sharpe"): + reported = tabulated[metric] + try: + direct = getattr(stats, metric)()["Asset"] + except Exception: + assert reported is None or math.isnan(reported), ( + f"{metric}() raised, so summary must report NaN, got {reported}" + ) + continue + + if direct is None or (isinstance(direct, float) and math.isnan(direct)): + assert reported is None or math.isnan(reported), f"{metric}: expected NaN/None, got {reported}" + else: + assert reported == pytest.approx(direct, rel=TOL_PARITY), ( + f"summary row {metric!r} = {reported} but {metric}() = {direct}" + ) + + +# ── Portfolio transform identities ─────────────────────────────────────────── + + +@pytest.mark.property +@given( + data_tuple=st.integers(min_value=5, max_value=20).flatmap( + lambda n: st.tuples( + st.lists(_pos_price, min_size=n, max_size=n), + st.lists(_any_position, min_size=n, max_size=n), + ) + ) +) +@settings(max_examples=50) +def test_unbounded_truncate_and_zero_smoothing_are_identities( + data_tuple: tuple[list[float], list[float]], +) -> None: + """`truncate(None, None)` and `smoothed_holding(0)` leave the portfolio alone. + + An unbounded truncation selects every row, and smoothing over a window of + one averages each weight with nothing. Both must therefore round-trip to a + data-identical Portfolio. + """ + prices_list, positions_list = data_tuple + n = len(prices_list) + start = date(2020, 1, 1) + dates = [start + timedelta(days=i) for i in range(n)] + + prices_df = pl.DataFrame({"date": dates, "Asset": pl.Series(prices_list, dtype=pl.Float64)}) + positions_df = pl.DataFrame({"date": dates, "Asset": pl.Series(positions_list, dtype=pl.Float64)}) + portfolio = Portfolio(prices=prices_df, cashposition=positions_df, aum=1e5) + + for derived in (portfolio.truncate(), portfolio.smoothed_holding(0)): + pt.assert_frame_equal(derived.cashposition, portfolio.cashposition) + pt.assert_frame_equal(derived.prices, portfolio.prices) + assert derived.aum == portfolio.aum