Spaces:
Running on Zero
Running on Zero
| """Plotly figures in the Bit design language. | |
| Colors come from `PALETTE`, which mirrors the design system tokens. Plotly | |
| cannot read CSS variables, so the token values are resolved here once; if the | |
| design system changes, this table is the single place to update. | |
| Every figure returns a `go.Figure` with the app's dark canvas, square corners, | |
| tight type, and direction encoded by shape as well as color (the design marks | |
| up/down with ▲/▼ so colorblind users are not reading hue alone). | |
| """ | |
| from __future__ import annotations | |
| import numpy as np | |
| import pandas as pd | |
| import plotly.graph_objects as go | |
| from .metrics import drawdown_series, rolling_sharpe | |
| # -------------------------------------------------------------------------- | |
| # Palette (mirrors _ds/tokens/colors.css) | |
| # -------------------------------------------------------------------------- | |
| PALETTE = { | |
| "canvas": "#161512", | |
| "panel": "#1d1c18", | |
| "raised": "#24221d", | |
| "sunken": "#000000", | |
| "border_subtle": "#2c2a24", | |
| "border": "#3d3a32", | |
| "border_strong": "#6f6a56", | |
| "text": "#f7f4ec", | |
| "text_secondary": "#b6b09a", | |
| "text_tertiary": "#6f6a56", | |
| "amber": "#af9209", | |
| "amber_strong": "#cfab0a", | |
| "amber_dim": "#7d6a09", | |
| "moss": "#68781e", | |
| "moss_strong": "#7d901f", | |
| "moss_dim": "#4d5817", | |
| "up": "#19b35a", | |
| "up_strong": "#22c765", | |
| "down": "#e0483c", | |
| "down_strong": "#ee6152", | |
| "up_cvd": "#3f7fd0", | |
| "down_cvd": "#c07a2a", | |
| "mute_red": "#8a5a54", | |
| "mute_teal": "#54898a", | |
| "mute_blue": "#59656e", | |
| "mute_indigo": "#5c5c8a", | |
| "mute_violet": "#75588a", | |
| "mute_green": "#6e8a54", | |
| "mute_yellow": "#8a8154", | |
| } | |
| SERIES_COLORS = [ | |
| PALETTE["amber_strong"], PALETTE["mute_teal"], PALETTE["mute_violet"], | |
| PALETTE["moss_strong"], PALETTE["mute_indigo"], PALETTE["mute_red"], | |
| ] | |
| FONT = "ui-monospace, 'JetBrains Mono', SFMono-Regular, Menlo, monospace" | |
| HEAD_FONT = "'Styrene A', -apple-system, system-ui, sans-serif" | |
| def up_color(cvd: bool = False) -> str: | |
| return PALETTE["up_cvd"] if cvd else PALETTE["up"] | |
| def down_color(cvd: bool = False) -> str: | |
| return PALETTE["down_cvd"] if cvd else PALETTE["down"] | |
| def _base_layout(fig: go.Figure, height: int = 320, *, showlegend: bool = False, | |
| margin: tuple[int, int, int, int] = (8, 8, 8, 8)) -> go.Figure: | |
| l, r, t, b = margin | |
| fig.update_layout( | |
| template="plotly_dark", | |
| paper_bgcolor=PALETTE["panel"], | |
| plot_bgcolor=PALETTE["panel"], | |
| font=dict(family=FONT, size=10, color=PALETTE["text_secondary"]), | |
| height=height, | |
| margin=dict(l=l, r=r, t=t, b=b), | |
| showlegend=showlegend, | |
| legend=dict(bgcolor="rgba(0,0,0,0)", borderwidth=0, | |
| font=dict(size=9, color=PALETTE["text_secondary"]), | |
| orientation="h", yanchor="bottom", y=1.0, x=0), | |
| hoverlabel=dict(bgcolor=PALETTE["raised"], bordercolor=PALETTE["border"], | |
| font=dict(family=FONT, size=10, color=PALETTE["text"])), | |
| xaxis=dict(gridcolor=PALETTE["border_subtle"], zerolinecolor=PALETTE["border"], | |
| linecolor=PALETTE["border"], tickfont=dict(size=9)), | |
| yaxis=dict(gridcolor=PALETTE["border_subtle"], zerolinecolor=PALETTE["border"], | |
| linecolor=PALETTE["border"], tickfont=dict(size=9)), | |
| dragmode="pan", | |
| ) | |
| return fig | |
| def empty_figure(message: str = "No data", height: int = 320) -> go.Figure: | |
| fig = go.Figure() | |
| fig.add_annotation(text=message.upper(), showarrow=False, | |
| font=dict(family=FONT, size=11, color=PALETTE["text_tertiary"]), | |
| x=0.5, y=0.5, xref="paper", yref="paper") | |
| fig.update_xaxes(visible=False) | |
| fig.update_yaxes(visible=False) | |
| return _base_layout(fig, height) | |
| # -------------------------------------------------------------------------- | |
| # Equity curve | |
| # -------------------------------------------------------------------------- | |
| def equity_curve( | |
| equity: pd.Series, | |
| benchmark: pd.Series | None = None, | |
| *, | |
| plan=None, | |
| log_scale: bool = False, | |
| height: int = 340, | |
| drawdown_shading: bool = True, | |
| cvd: bool = False, | |
| ) -> go.Figure: | |
| """Strategy vs buy & hold, with drawdown shading and validation bands.""" | |
| if equity is None or equity.empty: | |
| return empty_figure("no equity curve", height) | |
| fig = go.Figure() | |
| base = float(equity.iloc[0]) | |
| pct = (equity / base - 1.0) * 100.0 | |
| if drawdown_shading: | |
| dd = drawdown_series(equity) | |
| # Shade the stretches spent more than 5% below the running peak. | |
| in_dd = dd < -0.05 | |
| for lo, hi in _true_runs(in_dd): | |
| fig.add_vrect(x0=equity.index[lo], x1=equity.index[hi], | |
| fillcolor=PALETTE["down"], opacity=0.10, | |
| line_width=0, layer="below") | |
| if plan is not None: | |
| for w in getattr(plan, "windows", []): | |
| fig.add_vrect(x0=w.test_start, x1=w.test_end, | |
| fillcolor=PALETTE["moss"], opacity=0.07, | |
| line_width=0, layer="below") | |
| hs = getattr(plan, "holdout_start", None) | |
| if hs is not None: | |
| fig.add_vrect( | |
| x0=hs, x1=equity.index[-1], fillcolor=PALETTE["amber"], opacity=0.10, | |
| line_width=1, line_color=PALETTE["amber_dim"], layer="below", | |
| annotation_text="HOLDOUT", annotation_position="top left", | |
| annotation_font=dict(family=FONT, size=9, color=PALETTE["amber_strong"]), | |
| ) | |
| if benchmark is not None and not benchmark.empty: | |
| bpct = (benchmark / float(benchmark.iloc[0]) - 1.0) * 100.0 | |
| fig.add_trace(go.Scatter( | |
| x=bpct.index, y=bpct.to_numpy(), name="BUY & HOLD", | |
| line=dict(color=PALETTE["text_tertiary"], width=1.2, dash="dash"), | |
| hovertemplate="buy & hold %{y:.1f}%<extra></extra>", | |
| )) | |
| fig.add_trace(go.Scatter( | |
| x=pct.index, y=pct.to_numpy(), name="STRATEGY", | |
| line=dict(color=PALETTE["amber_strong"], width=1.8), | |
| hovertemplate="strategy %{y:.1f}%<extra></extra>", | |
| )) | |
| fig.update_yaxes(ticksuffix="%", title=None) | |
| if log_scale: | |
| # Log scale needs a positive series, so plot the equity multiple. | |
| fig.data = () | |
| mult = equity / base | |
| if benchmark is not None and not benchmark.empty: | |
| fig.add_trace(go.Scatter( | |
| x=benchmark.index, y=(benchmark / float(benchmark.iloc[0])).to_numpy(), | |
| name="BUY & HOLD", | |
| line=dict(color=PALETTE["text_tertiary"], width=1.2, dash="dash"))) | |
| fig.add_trace(go.Scatter(x=mult.index, y=mult.to_numpy(), name="STRATEGY", | |
| line=dict(color=PALETTE["amber_strong"], width=1.8))) | |
| fig.update_yaxes(type="log", ticksuffix="x") | |
| return _base_layout(fig, height, showlegend=True, margin=(8, 8, 24, 8)) | |
| def _true_runs(mask: pd.Series) -> list[tuple[int, int]]: | |
| """Contiguous [start, end] index positions where `mask` is True.""" | |
| arr = mask.to_numpy() | |
| runs, start = [], None | |
| for i, v in enumerate(arr): | |
| if v and start is None: | |
| start = i | |
| elif not v and start is not None: | |
| runs.append((start, i - 1)) | |
| start = None | |
| if start is not None: | |
| runs.append((start, len(arr) - 1)) | |
| return runs | |
| # -------------------------------------------------------------------------- | |
| # Underwater / rolling Sharpe | |
| # -------------------------------------------------------------------------- | |
| def underwater_chart(equity: pd.Series, height: int = 150) -> go.Figure: | |
| if equity is None or equity.empty: | |
| return empty_figure("no drawdown data", height) | |
| dd = drawdown_series(equity) * 100.0 | |
| fig = go.Figure(go.Scatter( | |
| x=dd.index, y=dd.to_numpy(), fill="tozeroy", mode="lines", | |
| line=dict(color=PALETTE["down"], width=1.0), | |
| fillcolor="rgba(224,72,60,0.35)", | |
| hovertemplate="%{y:.1f}%<extra></extra>", | |
| )) | |
| fig.update_yaxes(ticksuffix="%") | |
| return _base_layout(fig, height) | |
| def rolling_sharpe_chart(equity: pd.Series, window: int, bars_per_year: float, | |
| height: int = 150) -> go.Figure: | |
| if equity is None or equity.empty: | |
| return empty_figure("no rolling sharpe", height) | |
| rs = rolling_sharpe(equity, window, bars_per_year) | |
| if rs.empty: | |
| return empty_figure(f"needs > {window} bars", height) | |
| fig = go.Figure(go.Scatter( | |
| x=rs.index, y=rs.to_numpy(), mode="lines", | |
| line=dict(color=PALETTE["mute_teal"], width=1.2), | |
| hovertemplate="sharpe %{y:.2f}<extra></extra>", | |
| )) | |
| fig.add_hline(y=0, line=dict(color=PALETTE["border"], width=1)) | |
| fig.add_hline(y=1, line=dict(color=PALETTE["moss_dim"], width=1, dash="dot")) | |
| return _base_layout(fig, height) | |
| # -------------------------------------------------------------------------- | |
| # Price + trade flags | |
| # -------------------------------------------------------------------------- | |
| def price_with_trades( | |
| prices: pd.DataFrame, trades: pd.DataFrame, *, height: int = 420, | |
| max_bars: int = 600, cvd: bool = False, | |
| ) -> go.Figure: | |
| """Candlesticks with volume, entry/exit flags and win/loss connectors.""" | |
| if prices is None or prices.empty: | |
| return empty_figure("no price data", height) | |
| px = prices.tail(max_bars) | |
| up, down = up_color(cvd), down_color(cvd) | |
| fig = go.Figure() | |
| fig.add_trace(go.Candlestick( | |
| x=px.index, open=px["open"], high=px["high"], low=px["low"], close=px["close"], | |
| increasing=dict(line=dict(color=up, width=1), fillcolor=up), | |
| decreasing=dict(line=dict(color=down, width=1), fillcolor=down), | |
| name="price", yaxis="y", showlegend=False, | |
| )) | |
| if "volume" in px.columns: | |
| vmax = float(px["volume"].max()) or 1.0 | |
| pmin = float(px["low"].min()) | |
| prange = float(px["high"].max()) - pmin or 1.0 | |
| scaled = pmin + (px["volume"] / vmax) * prange * 0.16 | |
| fig.add_trace(go.Bar( | |
| x=px.index, y=scaled - pmin, base=pmin, marker_color=PALETTE["border"], | |
| opacity=0.5, name="volume", showlegend=False, hoverinfo="skip", | |
| )) | |
| if trades is not None and not trades.empty: | |
| window = trades[(trades["entry_ts"] >= px.index[0]) | |
| & (trades["entry_ts"] <= px.index[-1])] | |
| for t in window.itertuples(): | |
| won = t.net_pnl > 0 | |
| color = up if won else down | |
| hollow = str(t.side).lower() == "short" | |
| fig.add_trace(go.Scatter( | |
| x=[t.entry_ts, t.exit_ts], y=[t.entry_px, t.exit_px], | |
| mode="lines", line=dict(color=color, width=1, dash="dot"), | |
| showlegend=False, hoverinfo="skip", | |
| )) | |
| card = ( | |
| f"#{t.id} {str(t.side).upper()}<br>" | |
| f"net {t.net_pnl:+,.2f} ({t.r_multiple:+.2f}R)<br>" | |
| f"costs {t.costs:,.2f}<br>" | |
| f"MAE {t.mae:.2%} · MFE {t.mfe:.2%}<br>" | |
| f"<i>{t.trigger}</i>" | |
| ) | |
| fig.add_trace(go.Scatter( | |
| x=[t.entry_ts], y=[t.entry_px], mode="markers", showlegend=False, | |
| marker=dict(symbol="triangle-up", size=10, color="rgba(0,0,0,0)" if hollow else color, | |
| line=dict(color=color, width=1.5)), | |
| hovertemplate=card + "<extra></extra>", | |
| )) | |
| fig.add_trace(go.Scatter( | |
| x=[t.exit_ts], y=[t.exit_px], mode="markers", showlegend=False, | |
| marker=dict(symbol="triangle-down", size=10, color="rgba(0,0,0,0)" if hollow else color, | |
| line=dict(color=color, width=1.5)), | |
| hovertemplate=card + "<extra></extra>", | |
| )) | |
| fig.update_layout(xaxis_rangeslider_visible=False, barmode="overlay") | |
| return _base_layout(fig, height) | |
| # -------------------------------------------------------------------------- | |
| # Distributions | |
| # -------------------------------------------------------------------------- | |
| def pnl_histogram(trades: pd.DataFrame, height: int = 200, cvd: bool = False) -> go.Figure: | |
| if trades is None or trades.empty: | |
| return empty_figure("no trades", height) | |
| net = trades["net_pnl"].astype(float) | |
| colors = [up_color(cvd) if v > 0 else down_color(cvd) for v in net] | |
| fig = go.Figure(go.Histogram( | |
| x=net, nbinsx=min(40, max(8, len(net) // 3)), | |
| marker=dict(color=PALETTE["mute_blue"], line=dict(color=PALETTE["border"], width=1)), | |
| hovertemplate="%{y} trades in %{x}<extra></extra>", | |
| )) | |
| fig.add_vline(x=0, line=dict(color=PALETTE["text_tertiary"], width=1)) | |
| return _base_layout(fig, height) | |
| def holding_period_histogram(trades: pd.DataFrame, height: int = 200) -> go.Figure: | |
| if trades is None or trades.empty or "duration_bars" in trades.columns is None: | |
| return empty_figure("no trades", height) | |
| dur = trades["duration_bars"].dropna().astype(float) | |
| if dur.empty: | |
| return empty_figure("no durations", height) | |
| fig = go.Figure(go.Histogram( | |
| x=dur, nbinsx=min(30, max(6, len(dur) // 3)), | |
| marker=dict(color=PALETTE["mute_indigo"], line=dict(color=PALETTE["border"], width=1)), | |
| hovertemplate="%{y} trades held %{x} bars<extra></extra>", | |
| )) | |
| return _base_layout(fig, height) | |
| def mae_mfe_scatter(trades: pd.DataFrame, height: int = 200, cvd: bool = False) -> go.Figure: | |
| if trades is None or trades.empty: | |
| return empty_figure("no trades", height) | |
| t = trades.dropna(subset=["mae", "mfe"]) | |
| if t.empty: | |
| return empty_figure("no excursion data", height) | |
| won = t["net_pnl"] > 0 | |
| fig = go.Figure() | |
| for label, mask, color, sym in ( | |
| ("wins", won, up_color(cvd), "triangle-up"), | |
| ("losses", ~won, down_color(cvd), "triangle-down"), | |
| ): | |
| sub = t[mask] | |
| if sub.empty: | |
| continue | |
| fig.add_trace(go.Scatter( | |
| x=(sub["mae"] * 100).to_numpy(), y=(sub["mfe"] * 100).to_numpy(), | |
| mode="markers", name=label.upper(), | |
| marker=dict(color=color, size=6, symbol=sym, opacity=0.75), | |
| customdata=sub[["id", "net_pnl"]].to_numpy(), | |
| hovertemplate="#%{customdata[0]} net %{customdata[1]:+,.0f}<br>" | |
| "MAE %{x:.1f}% · MFE %{y:.1f}%<extra></extra>", | |
| )) | |
| fig.update_xaxes(title=dict(text="MAE %", font=dict(size=9)), ticksuffix="%") | |
| fig.update_yaxes(title=dict(text="MFE %", font=dict(size=9)), ticksuffix="%") | |
| return _base_layout(fig, height, showlegend=True, margin=(8, 8, 22, 28)) | |
| # -------------------------------------------------------------------------- | |
| # Comparison | |
| # -------------------------------------------------------------------------- | |
| def strategy_timeframe_heatmap(df: pd.DataFrame, *, height: int = 320, | |
| value_col: str = "oos_sharpe") -> go.Figure: | |
| """Strategy x timeframe OOS-Sharpe matrix. Scale fixed at -0.5 -> 2.0.""" | |
| if df is None or df.empty: | |
| return empty_figure("no comparison coverage", height) | |
| pivot = df.pivot_table(index="strategy", columns="timeframe", | |
| values=value_col, aggfunc="mean") | |
| order = [tf for tf in ("15m", "1h", "1d") if tf in pivot.columns] | |
| pivot = pivot.reindex(columns=order or list(pivot.columns)) | |
| fig = go.Figure(go.Heatmap( | |
| z=pivot.to_numpy(), x=list(pivot.columns), y=list(pivot.index), | |
| zmin=-0.5, zmax=2.0, | |
| colorscale=[[0.0, PALETTE["down"]], [0.2, PALETTE["panel"]], | |
| [0.5, PALETTE["moss_dim"]], [1.0, PALETTE["amber_strong"]]], | |
| hovertemplate="%{y} · %{x}<br>OOS Sharpe %{z:.2f}<extra></extra>", | |
| colorbar=dict(thickness=8, len=0.8, tickfont=dict(size=9), | |
| outlinewidth=0, title=dict(text="SHARPE", font=dict(size=9))), | |
| xgap=2, ygap=2, | |
| )) | |
| return _base_layout(fig, height, margin=(8, 8, 8, 8)) | |
| def overlaid_returns(curves: dict[str, pd.Series], *, height: int = 300, | |
| oos_start=None) -> go.Figure: | |
| """Cumulative return of several runs on one shared scale.""" | |
| if not curves: | |
| return empty_figure("select runs to compare", height) | |
| fig = go.Figure() | |
| for i, (name, eq) in enumerate(curves.items()): | |
| if eq is None or eq.empty: | |
| continue | |
| pct = (eq / float(eq.iloc[0]) - 1.0) * 100.0 | |
| fig.add_trace(go.Scatter( | |
| x=pct.index, y=pct.to_numpy(), name=name[:34], | |
| line=dict(color=SERIES_COLORS[i % len(SERIES_COLORS)], width=1.4), | |
| hovertemplate=f"{name}: %{{y:.1f}}%<extra></extra>", | |
| )) | |
| if oos_start is not None: | |
| fig.add_vrect(x0=oos_start, x1=max(s.index[-1] for s in curves.values() if len(s)), | |
| fillcolor=PALETTE["moss"], opacity=0.07, line_width=0, layer="below") | |
| fig.update_yaxes(ticksuffix="%") | |
| return _base_layout(fig, height, showlegend=True, margin=(8, 8, 26, 8)) | |
| def small_multiples(curves: dict[str, pd.Series], *, height: int = 260) -> go.Figure: | |
| """Grid of equity curves, one per selected run, on a shared y-scale.""" | |
| from plotly.subplots import make_subplots | |
| if not curves: | |
| return empty_figure("select runs to compare", height) | |
| n = len(curves) | |
| cols = min(3, n) | |
| rows = (n + cols - 1) // cols | |
| fig = make_subplots(rows=rows, cols=cols, subplot_titles=[k[:26] for k in curves], | |
| vertical_spacing=0.18, horizontal_spacing=0.06) | |
| for i, (name, eq) in enumerate(curves.items()): | |
| r, c = divmod(i, cols) | |
| pct = (eq / float(eq.iloc[0]) - 1.0) * 100.0 if len(eq) else eq | |
| fig.add_trace(go.Scatter( | |
| x=pct.index, y=pct.to_numpy(), showlegend=False, | |
| line=dict(color=SERIES_COLORS[i % len(SERIES_COLORS)], width=1.2), | |
| ), row=r + 1, col=c + 1) | |
| fig.update_annotations(font=dict(family=FONT, size=9, color=PALETTE["text_secondary"])) | |
| return _base_layout(fig, max(height, 130 * rows), margin=(8, 8, 22, 8)) | |
| def correlation_matrix(returns: dict[str, pd.Series], height: int = 280) -> go.Figure: | |
| """Return correlation between selected runs -- 'are these the same bet?'""" | |
| if len(returns) < 2: | |
| return empty_figure("select at least two runs", height) | |
| df = pd.DataFrame({k: v for k, v in returns.items()}).dropna() | |
| if df.empty or df.shape[1] < 2: | |
| return empty_figure("no overlapping period", height) | |
| corr = df.corr() | |
| fig = go.Figure(go.Heatmap( | |
| z=corr.to_numpy(), x=[c[:18] for c in corr.columns], y=[c[:18] for c in corr.index], | |
| zmin=-1, zmax=1, | |
| colorscale=[[0.0, PALETTE["mute_blue"]], [0.5, PALETTE["panel"]], | |
| [1.0, PALETTE["amber_strong"]]], | |
| hovertemplate="%{y} vs %{x}<br>r = %{z:.2f}<extra></extra>", | |
| colorbar=dict(thickness=8, len=0.8, tickfont=dict(size=9), outlinewidth=0), | |
| xgap=2, ygap=2, | |
| )) | |
| return _base_layout(fig, height) | |
| def regime_bars(by_regime: pd.DataFrame, height: int = 260, cvd: bool = False) -> go.Figure: | |
| """Return grouped by market regime (bull / bear / chop).""" | |
| if by_regime is None or by_regime.empty: | |
| return empty_figure("no regime breakdown", height) | |
| fig = go.Figure() | |
| for i, col in enumerate([c for c in by_regime.columns if c != "regime"]): | |
| fig.add_trace(go.Bar( | |
| x=by_regime["regime"], y=by_regime[col] * 100.0, name=col[:24], | |
| marker_color=SERIES_COLORS[i % len(SERIES_COLORS)], | |
| hovertemplate="%{x}: %{y:.1f}%<extra></extra>", | |
| )) | |
| fig.add_hline(y=0, line=dict(color=PALETTE["border"], width=1)) | |
| fig.update_yaxes(ticksuffix="%") | |
| return _base_layout(fig, height, showlegend=True, margin=(8, 8, 26, 8)) | |
| # -------------------------------------------------------------------------- | |
| # Robustness | |
| # -------------------------------------------------------------------------- | |
| def parameter_sensitivity(grid: pd.DataFrame, *, x: str, y: str, z: str = "oos_sharpe", | |
| chosen: tuple | None = None, height: int = 300) -> go.Figure: | |
| """Heatmap of OOS Sharpe across a two-parameter sweep.""" | |
| if grid is None or grid.empty: | |
| return empty_figure("run a sweep to see sensitivity", height) | |
| pivot = grid.pivot_table(index=y, columns=x, values=z, aggfunc="mean") | |
| fig = go.Figure(go.Heatmap( | |
| z=pivot.to_numpy(), x=list(pivot.columns), y=list(pivot.index), | |
| colorscale=[[0.0, PALETTE["down"]], [0.35, PALETTE["panel"]], | |
| [0.7, PALETTE["moss_dim"]], [1.0, PALETTE["amber_strong"]]], | |
| hovertemplate=f"{x} %{{x}} · {y} %{{y}}<br>Sharpe %{{z:.2f}}<extra></extra>", | |
| colorbar=dict(thickness=8, len=0.8, tickfont=dict(size=9), outlinewidth=0), | |
| xgap=1, ygap=1, | |
| )) | |
| if chosen is not None: | |
| fig.add_shape(type="rect", | |
| x0=chosen[0] - 0.5, x1=chosen[0] + 0.5, | |
| y0=chosen[1] - 0.5, y1=chosen[1] + 0.5, | |
| line=dict(color=PALETTE["text"], width=2)) | |
| fig.update_xaxes(title=dict(text=x.upper(), font=dict(size=9))) | |
| fig.update_yaxes(title=dict(text=y.upper(), font=dict(size=9))) | |
| return _base_layout(fig, height, margin=(8, 8, 8, 28)) | |
| def monte_carlo_cone(paths: np.ndarray, index=None, *, height: int = 300) -> go.Figure: | |
| """P5 / P50 / P95 cone over reshuffled trade sequences.""" | |
| if paths is None or len(paths) == 0: | |
| return empty_figure("needs trades to reshuffle", height) | |
| p5 = np.percentile(paths, 5, axis=0) * 100.0 | |
| p50 = np.percentile(paths, 50, axis=0) * 100.0 | |
| p95 = np.percentile(paths, 95, axis=0) * 100.0 | |
| x = list(index) if index is not None else list(range(len(p50))) | |
| fig = go.Figure() | |
| fig.add_trace(go.Scatter(x=x + x[::-1], y=list(p95) + list(p5)[::-1], | |
| fill="toself", fillcolor="rgba(175,146,9,0.14)", | |
| line=dict(width=0), hoverinfo="skip", showlegend=False)) | |
| fig.add_trace(go.Scatter(x=x, y=p50, line=dict(color=PALETTE["amber_strong"], width=1.6), | |
| name="P50", hovertemplate="P50 %{y:.1f}%<extra></extra>")) | |
| fig.add_hline(y=0, line=dict(color=PALETTE["border"], width=1)) | |
| fig.update_yaxes(ticksuffix="%") | |
| return _base_layout(fig, height) | |
| def walk_forward_bars(windows, height: int = 240, cvd: bool = False) -> go.Figure: | |
| """Per-window OOS return -- the consistency check.""" | |
| if not windows: | |
| return empty_figure("no walk-forward windows", height) | |
| labels = [f"W{w.window.idx + 1}" for w in windows] | |
| vals = [w.metrics.total_return * 100.0 for w in windows] | |
| colors = [up_color(cvd) if v > 0 else down_color(cvd) for v in vals] | |
| fig = go.Figure(go.Bar( | |
| x=labels, y=vals, marker_color=colors, | |
| hovertemplate="%{x}: %{y:+.1f}% OOS<extra></extra>", | |
| )) | |
| fig.add_hline(y=0, line=dict(color=PALETTE["border"], width=1)) | |
| fig.update_yaxes(ticksuffix="%") | |
| return _base_layout(fig, height) | |
| def slippage_stress(points: list[tuple[float, float]], height: int = 240) -> go.Figure: | |
| """Sharpe as modelled slippage rises -- where does the edge die?""" | |
| if not points: | |
| return empty_figure("no stress run", height) | |
| xs = [p[0] for p in points] | |
| ys = [p[1] for p in points] | |
| fig = go.Figure(go.Scatter( | |
| x=xs, y=ys, mode="lines+markers", | |
| line=dict(color=PALETTE["amber_strong"], width=1.6), | |
| marker=dict(size=7, color=PALETTE["amber_strong"]), | |
| hovertemplate="%{x} bps → Sharpe %{y:.2f}<extra></extra>", | |
| )) | |
| fig.add_hline(y=0, line=dict(color=PALETTE["down"], width=1, dash="dot")) | |
| fig.update_xaxes(title=dict(text="SLIPPAGE (BPS)", font=dict(size=9))) | |
| return _base_layout(fig, height, margin=(8, 8, 8, 28)) | |
| def regime_strip(prices: pd.DataFrame, height: int = 42) -> go.Figure: | |
| """Thin bull/bear/chop band shown under the equity curve.""" | |
| if prices is None or prices.empty: | |
| return empty_figure("", height) | |
| reg = classify_regime(prices) | |
| color_of = {"bull": PALETTE["moss_strong"], "bear": PALETTE["down"], | |
| "chop": PALETTE["mute_yellow"]} | |
| fig = go.Figure() | |
| for lo, hi, label in _segments(reg): | |
| fig.add_vrect(x0=reg.index[lo], x1=reg.index[hi], | |
| fillcolor=color_of.get(label, PALETTE["border"]), | |
| opacity=0.8, line_width=0) | |
| fig.update_xaxes(visible=False) | |
| fig.update_yaxes(visible=False, range=[0, 1]) | |
| fig.update_layout(margin=dict(l=0, r=0, t=0, b=0)) | |
| return _base_layout(fig, height, margin=(0, 0, 0, 0)) | |
| def classify_regime(prices: pd.DataFrame, window: int = 60) -> pd.Series: | |
| """Bull / bear / chop from trailing trend and volatility. Causal.""" | |
| close = prices["close"] | |
| trend = close.pct_change(window) | |
| vol = close.pct_change().rolling(window).std() | |
| med_vol = vol.rolling(window * 3, min_periods=window).median() | |
| out = pd.Series("chop", index=close.index, dtype="object") | |
| out[(trend > 0.05) & (vol <= med_vol * 1.5)] = "bull" | |
| out[trend < -0.05] = "bear" | |
| return out.fillna("chop") | |
| def _segments(series: pd.Series) -> list[tuple[int, int, str]]: | |
| vals = series.to_numpy() | |
| out, start = [], 0 | |
| for i in range(1, len(vals)): | |
| if vals[i] != vals[start]: | |
| out.append((start, i - 1, vals[start])) | |
| start = i | |
| if len(vals): | |
| out.append((start, len(vals) - 1, vals[start])) | |
| return out | |
| def monte_carlo_paths(trades: pd.DataFrame, n_paths: int = 1000, seed: int = 0) -> np.ndarray: | |
| """Reshuffle the realised trade sequence `n_paths` times. | |
| Seeded, so the cone the UI shows is reproducible run to run. | |
| """ | |
| if trades is None or trades.empty: | |
| return np.empty((0, 0)) | |
| rets = (trades["net_pnl"] / trades["entry_px"].abs().clip(lower=1e-9) | |
| / trades["size"].abs().clip(lower=1e-9)).to_numpy() | |
| rets = rets[np.isfinite(rets)] | |
| if len(rets) == 0: | |
| return np.empty((0, 0)) | |
| rng = np.random.default_rng(seed) | |
| out = np.empty((n_paths, len(rets))) | |
| for i in range(n_paths): | |
| out[i] = np.cumprod(1.0 + rng.permutation(rets)) - 1.0 | |
| return out | |
| # -------------------------------------------------------------------------- | |
| # Catalog / global comparison | |
| # -------------------------------------------------------------------------- | |
| def multi_return_overlay(curves: dict[str, pd.Series], *, height: int = 420, | |
| highlight: str | None = None, | |
| max_series: int = 24) -> go.Figure: | |
| """Cumulative return of many algorithms on one shared axis. | |
| Series arrive already normalised to cumulative return by the catalog, so | |
| nothing is re-based here and every line is directly comparable. Beyond | |
| `max_series` the chart stops being readable, so extras are dropped and the | |
| caller is expected to say so rather than silently truncating. | |
| """ | |
| if not curves: | |
| return empty_figure("select rows to plot", height) | |
| fig = go.Figure() | |
| items = list(curves.items())[:max_series] | |
| for i, (name, series) in enumerate(items): | |
| if series is None or len(series) == 0: | |
| continue | |
| is_hl = highlight is not None and name == highlight | |
| color = SERIES_COLORS[i % len(SERIES_COLORS)] | |
| fig.add_trace(go.Scatter( | |
| x=series.index, y=(series * 100.0).to_numpy(), name=name[:40], | |
| line=dict(color=PALETTE["amber_strong"] if is_hl else color, | |
| width=2.4 if is_hl else 1.3), | |
| opacity=1.0 if (is_hl or highlight is None) else 0.45, | |
| hovertemplate=f"{name}<br>%{{y:.1f}}%<extra></extra>", | |
| )) | |
| fig.add_hline(y=0, line=dict(color=PALETTE["border"], width=1)) | |
| fig.update_yaxes(ticksuffix="%") | |
| return _base_layout(fig, height, showlegend=True, margin=(8, 8, 30, 8)) | |
| def risk_return_scatter(df: pd.DataFrame, *, height: int = 380, | |
| x: str = "max_drawdown", y: str = "total_return", | |
| size: str = "trades", color_by: str = "strategy") -> go.Figure: | |
| """Where every catalog row sits in risk/return space. | |
| Marker area encodes trade count, so a spectacular result built on four | |
| trades looks as small as it deserves to. | |
| """ | |
| if df is None or df.empty: | |
| return empty_figure("no catalog rows", height) | |
| d = df.dropna(subset=[x, y]).copy() | |
| if d.empty: | |
| return empty_figure("no plottable rows", height) | |
| d["_size"] = d[size].fillna(0).clip(lower=1) ** 0.5 if size in d.columns else 4 | |
| groups = list(dict.fromkeys(d[color_by])) if color_by in d.columns else ["all"] | |
| fig = go.Figure() | |
| for i, g in enumerate(groups): | |
| sub = d[d[color_by] == g] if color_by in d.columns else d | |
| if sub.empty: | |
| continue | |
| label = sub.get("model_display", pd.Series([""] * len(sub), index=sub.index)) | |
| fig.add_trace(go.Scatter( | |
| x=(sub[x] * 100).to_numpy(), y=(sub[y] * 100).to_numpy(), | |
| mode="markers", name=str(g)[:26], | |
| marker=dict(color=SERIES_COLORS[i % len(SERIES_COLORS)], | |
| size=sub["_size"].to_numpy(), sizemode="area", | |
| sizeref=max(d["_size"].max() ** 2 / 900, 1e-9), sizemin=4, | |
| line=dict(width=0.5, color=PALETTE["border"])), | |
| customdata=np.stack([sub["asset"], sub["timeframe"], label, | |
| sub.get("trades", pd.Series(0, index=sub.index))], axis=-1), | |
| hovertemplate=("%{customdata[0]} · %{customdata[1]} · %{customdata[2]}<br>" | |
| "drawdown %{x:.1f}% · return %{y:.1f}%<br>" | |
| "%{customdata[3]} trades<extra></extra>"), | |
| )) | |
| fig.add_hline(y=0, line=dict(color=PALETTE["border"], width=1)) | |
| fig.update_xaxes(title=dict(text="MAX DRAWDOWN", font=dict(size=9)), ticksuffix="%") | |
| fig.update_yaxes(title=dict(text="TOTAL RETURN", font=dict(size=9)), ticksuffix="%") | |
| return _base_layout(fig, height, showlegend=True, margin=(8, 8, 30, 34)) | |
| def model_accuracy_bars(scorecard: pd.DataFrame, *, height: int = 320, | |
| timeframe: str | None = None) -> go.Figure: | |
| """Directional accuracy per model, with the coin-flip line drawn in. | |
| Baselines are coloured differently on purpose: the interesting question is | |
| not which model scores highest, it is whether any learned model clears the | |
| naive ones at all. | |
| """ | |
| if scorecard is None or scorecard.empty: | |
| return empty_figure("no scorecard rows", height) | |
| d = scorecard | |
| if timeframe: | |
| d = d[d["timeframe"] == timeframe] | |
| d = d.dropna(subset=["directional_accuracy"]) | |
| if d.empty: | |
| return empty_figure("no directional calls recorded", height) | |
| agg = (d.groupby(["model_display", "is_baseline"])["directional_accuracy"] | |
| .mean().reset_index().sort_values("directional_accuracy")) | |
| colors = [PALETTE["mute_blue"] if b else PALETTE["amber_strong"] | |
| for b in agg["is_baseline"]] | |
| fig = go.Figure(go.Bar( | |
| x=(agg["directional_accuracy"] * 100).to_numpy(), | |
| y=agg["model_display"].to_numpy(), orientation="h", | |
| marker_color=colors, | |
| hovertemplate="%{y}: %{x:.1f}% of directional calls correct<extra></extra>", | |
| )) | |
| fig.add_vline(x=50, line=dict(color=PALETTE["down"], width=1.5, dash="dot"), | |
| annotation_text="COIN FLIP", annotation_position="top", | |
| annotation_font=dict(family=FONT, size=9, color=PALETTE["down"])) | |
| fig.update_xaxes(ticksuffix="%", range=[ | |
| max(0, float((agg["directional_accuracy"] * 100).min()) - 4), | |
| float((agg["directional_accuracy"] * 100).max()) + 4]) | |
| return _base_layout(fig, height, margin=(8, 8, 20, 8)) | |
| def calibration_scatter(scorecard: pd.DataFrame, *, height: int = 320) -> go.Figure: | |
| """Band coverage against the 80% nominal line -- who is actually calibrated.""" | |
| if scorecard is None or scorecard.empty: | |
| return empty_figure("no scorecard rows", height) | |
| d = scorecard.dropna(subset=["coverage_q10_q90"]) | |
| if d.empty: | |
| return empty_figure("no calibration data", height) | |
| fig = go.Figure() | |
| for i, (name, sub) in enumerate(d.groupby("model_display")): | |
| fig.add_trace(go.Scatter( | |
| x=sub["timeframe"], y=(sub["coverage_q10_q90"] * 100).to_numpy(), | |
| mode="markers", name=str(name)[:24], | |
| marker=dict(size=9, color=SERIES_COLORS[i % len(SERIES_COLORS)], | |
| symbol="diamond" if sub["is_baseline"].iloc[0] else "circle"), | |
| customdata=sub[["asset"]].to_numpy(), | |
| hovertemplate="%{customdata[0]}<br>coverage %{y:.1f}%<extra></extra>", | |
| )) | |
| fig.add_hline(y=80, line=dict(color=PALETTE["moss_strong"], width=1.5, dash="dot"), | |
| annotation_text="NOMINAL 80%", annotation_position="top left", | |
| annotation_font=dict(family=FONT, size=9, color=PALETTE["moss_strong"])) | |
| fig.update_yaxes(ticksuffix="%", title=dict(text="q10–q90 COVERAGE", font=dict(size=9))) | |
| return _base_layout(fig, height, showlegend=True, margin=(8, 8, 30, 34)) | |
| def model_leaderboard_bars(df: pd.DataFrame, *, height: int = 320, | |
| metric: str = "oos_sharpe") -> go.Figure: | |
| """Best result each model achieved, side by side.""" | |
| if df is None or df.empty or metric not in df.columns: | |
| return empty_figure("no catalog rows", height) | |
| d = df[df.get("model_slug", "") != ""].dropna(subset=[metric]) | |
| if d.empty: | |
| return empty_figure("no model-driven rows", height) | |
| agg = (d.groupby(["model_display", "is_baseline_model"])[metric] | |
| .max().reset_index().sort_values(metric)) | |
| colors = [PALETTE["mute_blue"] if b else PALETTE["amber_strong"] | |
| for b in agg["is_baseline_model"]] | |
| fig = go.Figure(go.Bar( | |
| x=agg[metric].to_numpy(), y=agg["model_display"].to_numpy(), | |
| orientation="h", marker_color=colors, | |
| hovertemplate="%{y}: best " + metric.replace("_", " ") + " %{x:.2f}<extra></extra>", | |
| )) | |
| fig.add_vline(x=0, line=dict(color=PALETTE["border"], width=1)) | |
| return _base_layout(fig, height, margin=(8, 8, 8, 8)) | |