gavindiaz/manifoldbt
1
0
Fork 0
mirror of https://github.com/manifoldbt/manifoldbt.git synced 2026-08-24 14:38:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

release: v0.12.0

Browse Source
This commit is contained in:
github-actions[bot]
2026-07-15 00:01:41 +00:00
parent 285e858649
commit aece0242af
16 changed files with 1540 additions and 1338 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "manifoldbt"
version = "0.11.0"
version = "0.12.0"
description = "Rust-powered backtesting engine for quantitative research"
requires-python = ">=3.9"
license = { file = "LICENSE" }
python/manifoldbt/__init__.py
+29 -20
View File
@@ -310,7 +310,12 @@ _PREPARED_CFG_CACHE_MAX = 256
def _prepared_config_json(config: BacktestConfig, strategy, store: DataStore) -> str:
"""Content-memoised equivalent of ``_prepare_config(...).to_json()``."""
"""Content-memoised equivalent of ``_prepare_config(...).to_json()``.
The prepared config no longer depends on the strategy (orders travel in the
strategy JSON now), so the memo key is just the config content plus the
metadata DB; the ``strategy`` argument is accepted for call-site symmetry.
"""
try:
meta_db = store.metadata_db()
except Exception:
@@ -318,10 +323,8 @@ def _prepared_config_json(config: BacktestConfig, strategy, store: DataStore) ->
if meta_db is None:
return _prepare_config(config, strategy, store).to_json()
orders = getattr(strategy, "_orders", None) if strategy is not None else None
try:
orders_key = json.dumps(orders, sort_keys=True, default=str) if orders else ""
key = (config.to_json(), orders_key, meta_db)
key = (config.to_json(), meta_db)
except (TypeError, ValueError):
# Unserialisable config content — skip memoisation, never fail.
return _prepare_config(config, strategy, store).to_json()
@@ -397,13 +400,11 @@ def _prepare_config(config: BacktestConfig, strategy, store: DataStore) -> Backt
resolved_sv[int(store.resolve_symbol(key))] = venue
fees.symbol_venue = resolved_sv
# Merge orders from strategy into execution config
if strategy and hasattr(strategy, '_orders') and strategy._orders:
if cfg.execution.orders is None:
cfg.execution.orders = OrderConfig()
for key, val in strategy._orders.items():
setattr(cfg.execution.orders, key, val)
# Per-strategy SL/TP/trailing orders are NOT merged into the config anymore:
# they travel inside the strategy JSON (Strategy.to_json -> StrategyDef.orders)
# so the engine applies them per-strategy. This lets one batch/sweep call run
# strategies carrying different brackets over a single data load. A bracket
# set directly on config.execution.orders still applies as the fallback.
return cfg
@@ -789,6 +790,11 @@ def run_batch(
Loads bars once, aligns timestamps once, then evaluates each strategy
on a separate rayon thread. Much faster than calling ``run()`` in a loop.
Per-strategy ``stop_loss``/``take_profit``/``trailing_stop`` are honored:
each strategy's orders travel inside its JSON and the engine applies them
per-strategy, so a batch of strategies with DIFFERENT brackets still runs
over a single data load.
Args:
strategies: List of Strategy definitions.
config: Shared backtest configuration (same universe/time range).
@@ -800,13 +806,12 @@ def run_batch(
"""
_require_pro_over_combos(len(strategies), "Batch backtesting")
try:
config = _prepare_config(config, None, store)
config = _cap_output_resolution(config)
store = _resolve_store(config, store)
strategy_jsons = [strat.to_json() for strat in strategies]
cfg_json = _prepared_config_json(config, None, store)
raw_results = _run_batch_native(
strategy_jsons,
config.to_json(),
[strat.to_json() for strat in strategies],
cfg_json,
store,
max_parallelism,
)
@@ -828,6 +833,11 @@ def run_batch_lite(
position traces, and Arrow output construction. Ideal for parameter sweeps
where you only need metrics to select the best variant.
Per-strategy ``stop_loss``/``take_profit``/``trailing_stop`` are honored:
each strategy's orders travel inside its JSON and the engine applies them
per-strategy, so a batch of strategies with DIFFERENT brackets still runs
over a single data load.
Args:
strategies: List of Strategy definitions.
config: Shared backtest configuration (same universe/time range).
@@ -839,13 +849,12 @@ def run_batch_lite(
"""
_require_pro_over_combos(len(strategies), "Batch backtesting")
try:
config = _prepare_config(config, None, store)
config = _cap_output_resolution(config)
store = _resolve_store(config, store)
strategy_jsons = [strat.to_json() for strat in strategies]
cfg_json = _prepared_config_json(config, None, store)
return _run_batch_lite_native(
strategy_jsons,
config.to_json(),
[strat.to_json() for strat in strategies],
cfg_json,
store,
max_parallelism,
)
@@ -1372,7 +1381,7 @@ __all__ = [
"__version__",
# Indicators (submodule)
"indicators",
# Plotting (lazy, requires matplotlib)
# Plotting (lazy, requires plotly)
"plot",
# Diagnostics (lazy)
"diagnostics",
python/manifoldbt/plot/__init__.py
+14 -3
View File
@@ -1,4 +1,4 @@
"""Plotting module for manifoldbt (requires matplotlib).
"""Plotting module for manifoldbt (requires plotly).
Install with::
@@ -11,12 +11,18 @@ Quick start::
result = bt.run(strategy, config, store)
bt.plot.tearsheet(result) # full-page dashboard
bt.plot.equity(result, show=True) # single chart
Every chart is interactive (crosshair, hover, zoom). ``show=True`` opens it
in a native window (``pip install manifoldbt[window]``; falls back to a
browser tab, which you can also force with ``show="browser"``).
``save=".html"`` writes a responsive interactive page. Static ``save=".png"``
is optional and needs ``pip install manifoldbt[png]`` (pulls a headless Chromium).
"""
try:
import matplotlib # noqa: F401
import plotly # noqa: F401
except ImportError:
raise ImportError(
"matplotlib is required for the plotting module. "
"plotly is required for the plotting module. "
"Install it with: pip install manifoldbt[plot]"
) from None
@@ -51,6 +57,9 @@ from manifoldbt.plot.research import (
# Composite layouts
from manifoldbt.plot.tearsheet import research_report, tearsheet
# Window display (multi-window, matplotlib-style)
from manifoldbt.plot._window import show
# Theme
from manifoldbt.plot._theme import THEME, apply_theme
@@ -78,6 +87,8 @@ __all__ = [
# Composites
"tearsheet",
"research_report",
# Window display
"show",
# Theme
"THEME",
"apply_theme",
python/manifoldbt/plot/_assets/manifoldbt.ico
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 20 KiB

python/manifoldbt/plot/_decimate.py
+42
View File
@@ -0,0 +1,42 @@
"""Min/max time-series decimation for plotting — pure numpy.
A chart is ~1000-2500 px wide, so plotting 10^5-10^6 samples draws hundreds of
sub-pixel points per column and bloats saved HTML. Per pixel column we keep the
bucket's min and max in time order, which preserves peaks and troughs exactly
(max drawdown survives untouched) at O(n) cost (~1 ms for 1M points).
"""
from __future__ import annotations
import numpy as np
#: Series shorter than this are plotted as-is.
DECIMATE_THRESHOLD = 20_000
def decimate_minmax(x: np.ndarray, y: np.ndarray, n_cols: int = 2500):
"""Per-column min/max envelope. Returns (x, y) unchanged when small."""
n = len(y)
if n <= 2 * n_cols:
return x, y
bucket = n // n_cols
m = bucket * n_cols
yb = y[:m].reshape(n_cols, bucket)
cols = np.arange(n_cols)
idx_min = yb.argmin(axis=1) + cols * bucket
idx_max = yb.argmax(axis=1) + cols * bucket
lo = np.minimum(idx_min, idx_max)
hi = np.maximum(idx_min, idx_max)
idx = np.empty(n_cols * 2, dtype=np.int64)
idx[0::2] = lo
idx[1::2] = hi
if m < n:
idx = np.append(idx, n - 1) # keep the true last sample
idx = np.unique(idx) # dedupe flat buckets (lo == hi)
return x[idx], y[idx]
def maybe_decimate(x: np.ndarray, y: np.ndarray, n_cols: int = 2500):
"""Decimate only when the series is longer than DECIMATE_THRESHOLD."""
if len(y) <= DECIMATE_THRESHOLD:
return x, y
return decimate_minmax(x, y, n_cols)
python/manifoldbt/plot/_theme.py
+87 -76
View File
@@ -1,8 +1,7 @@
"""Clean dark theme — modern, readable, quant-oriented."""
"""Clean dark theme — modern, readable, quant-oriented (plotly template)."""
from __future__ import annotations
from contextlib import contextmanager
from typing import Any, Dict
# ---------------------------------------------------------------------------
# Color palette — neutral dark, no decorative colors
@@ -20,94 +19,106 @@ BG_FIGURE = "#0c0c0f"
BG_AXES = "#111116"
BORDER = "#1e1e24"
GRID_RGBA = (1.0, 1.0, 1.0, 0.04)
GRID_COLOR = "rgba(255,255,255,0.045)"
SERIES_COLORS = [ACCENT, ACCENT_ALT, "#2dd4bf", ORANGE, RED, GREEN, "#f472b6", WHITE]
FONT_FAMILY = "Inter, system-ui, Segoe UI, Arial, sans-serif"
MONO_FAMILY = "SF Mono, Fira Code, Cascadia Code, Consolas, monospace"
# ---------------------------------------------------------------------------
# rcParams
# Colorscales (plotly format) — same stops as the old matplotlib colormaps
# ---------------------------------------------------------------------------
THEME: Dict[str, Any] = {
"figure.facecolor": BG_FIGURE,
"figure.edgecolor": BG_FIGURE,
"figure.dpi": 120,
"axes.facecolor": BG_AXES,
"axes.edgecolor": BORDER,
"axes.labelcolor": GRAY,
"axes.titlecolor": WHITE,
"axes.titlesize": 11,
"axes.titleweight": "medium",
"axes.titlepad": 12,
"axes.labelsize": 9,
"axes.labelpad": 8,
"axes.grid": True,
"grid.color": GRID_RGBA,
"grid.linewidth": 0.5,
"grid.linestyle": "-",
"xtick.color": DARK_GRAY,
"ytick.color": DARK_GRAY,
"xtick.labelsize": 8,
"ytick.labelsize": 8,
"text.color": WHITE,
"font.family": "monospace",
"font.size": 9,
"legend.facecolor": BG_AXES,
"legend.edgecolor": BORDER,
"legend.fontsize": 8,
"legend.labelcolor": GRAY,
"lines.linewidth": 1.3,
"lines.antialiased": True,
"savefig.facecolor": BG_FIGURE,
"savefig.edgecolor": BG_FIGURE,
"savefig.bbox": "tight",
"savefig.dpi": 150,
CS_DIVERGING = [[0.0, "#b91c1c"], [0.5, "#262626"], [1.0, "#15803d"]]
CS_SEQUENTIAL = [[0.0, "#b91c1c"], [0.5, "#d97706"], [1.0, "#15803d"]]
CS_CORRELATION = [[0.0, "#b91c1c"], [0.5, "#262626"], [1.0, "#1d4ed8"]]
# ---------------------------------------------------------------------------
# Layout defaults (also exported as THEME for backward compatibility)
# ---------------------------------------------------------------------------
THEME: dict = {
"paper_bgcolor": BG_FIGURE,
"plot_bgcolor": BG_AXES,
"font": {"family": FONT_FAMILY, "color": GRAY, "size": 12},
"title": {"font": {"color": WHITE, "size": 15}, "x": 0.01, "xanchor": "left"},
"margin": {"l": 64, "r": 24, "t": 48, "b": 36},
"hovermode": "x",
"colorway": SERIES_COLORS,
"hoverlabel": {
"bgcolor": "#1a1a20",
"bordercolor": BORDER,
"font": {"family": MONO_FAMILY, "color": WHITE, "size": 12},
},
"legend": {
"bgcolor": "rgba(17,17,22,0.6)",
"bordercolor": BORDER,
"borderwidth": 1,
"font": {"color": GRAY, "size": 11},
},
}
_AXIS = {
"color": GRAY,
"gridcolor": GRID_COLOR,
"linecolor": BORDER,
"zerolinecolor": GRID_COLOR,
"ticks": "",
"showspikes": True,
"spikemode": "across",
"spikethickness": 1,
"spikedash": "dot",
"spikecolor": GRAY,
}
_SCENE_AXIS = {
"backgroundcolor": BG_AXES,
"gridcolor": "rgba(255,255,255,0.08)",
"color": GRAY,
"showbackground": True,
"zerolinecolor": "rgba(255,255,255,0.08)",
}
def _build_theme() -> Dict[str, Any]:
"""Finalize THEME dict with cycler."""
import matplotlib.pyplot as plt
theme = dict(THEME)
theme["axes.prop_cycle"] = plt.cycler(color=SERIES_COLORS)
return theme
def _build_template():
"""Build the manifoldbt plotly template."""
import plotly.graph_objects as go
# ---------------------------------------------------------------------------
# Colormaps
# ---------------------------------------------------------------------------
def _register_colormaps() -> None:
"""Register custom colormaps (idempotent)."""
from matplotlib.colors import LinearSegmentedColormap
import matplotlib as mpl
_cmaps = {
"bt_diverging": [(0.0, "#b91c1c"), (0.5, "#262626"), (1.0, "#15803d")],
"bt_sequential": [(0.0, "#b91c1c"), (0.5, "#d97706"), (1.0, "#15803d")],
"bt_correlation": [(0.0, "#b91c1c"), (0.5, "#262626"), (1.0, "#1d4ed8")],
layout = dict(THEME)
layout["xaxis"] = dict(_AXIS)
layout["yaxis"] = dict(_AXIS)
layout["scene"] = {
"xaxis": dict(_SCENE_AXIS),
"yaxis": dict(_SCENE_AXIS),
"zaxis": dict(_SCENE_AXIS),
"bgcolor": BG_FIGURE,
}
for name, stops in _cmaps.items():
try:
mpl.colormaps.get_cmap(name)
except ValueError:
positions = [s[0] for s in stops]
colors = [s[1] for s in stops]
cmap = LinearSegmentedColormap.from_list(name, list(zip(positions, colors)), N=256)
mpl.colormaps.register(cmap, name=name)
return go.layout.Template(layout=layout)
_REGISTERED = False
# ---------------------------------------------------------------------------
# Public
# ---------------------------------------------------------------------------
def apply_theme() -> None:
"""Apply the dark theme globally."""
import matplotlib.pyplot as plt
_register_colormaps()
plt.rcParams.update(_build_theme())
"""Register the manifoldbt template and set it as plotly's default."""
global _REGISTERED
import plotly.io as pio
pio.templates["manifoldbt"] = _build_template()
pio.templates.default = "manifoldbt"
_REGISTERED = True
def _ensure_theme() -> None:
if not _REGISTERED:
apply_theme()
@contextmanager
def theme_context():
"""Context manager: apply theme temporarily."""
import matplotlib.pyplot as plt
_register_colormaps()
with plt.rc_context(_build_theme()):
yield
"""Backward-compatible context manager: ensures the theme is registered.
With plotly the theme is a global template rather than a temporary
rc-context, so this simply guarantees registration.
"""
_ensure_theme()
yield
python/manifoldbt/plot/_utils.py
+101 -28
View File
@@ -1,25 +1,32 @@
"""Shared plotting utilities."""
"""Shared plotting utilities (plotly)."""
from __future__ import annotations
from pathlib import Path
from typing import Optional, Tuple, Union
import matplotlib.pyplot as plt
from matplotlib.axes import Axes
from matplotlib.figure import Figure
from manifoldbt.plot._theme import WHITE, _ensure_theme
from manifoldbt.plot._theme import theme_context
_RESPONSIVE_CSS = (
"<style>html,body{height:100%;margin:0;background:#0c0c0f;overflow:hidden}"
".plotly-graph-div{width:100vw!important;height:100vh!important}</style>"
)
_IMAGE_EXTS = {".png", ".svg", ".pdf", ".jpg", ".jpeg", ".webp"}
def get_or_create_ax(
ax: Optional[Axes] = None,
def new_figure(
figsize: Tuple[float, float] = (12, 4),
) -> Tuple[Figure, Axes]:
"""Return (fig, ax). Creates a new themed figure if *ax* is None."""
if ax is not None:
return ax.figure, ax
fig, new_ax = plt.subplots(figsize=figsize)
return fig, new_ax
title: Optional[str] = None,
):
"""Return a themed plotly Figure sized from a matplotlib-style figsize."""
import plotly.graph_objects as go
_ensure_theme()
fig = go.Figure()
fig.update_layout(width=int(figsize[0] * 80), height=int(figsize[1] * 80))
if title:
fig.update_layout(title_text=title)
return fig
def format_pct(value: float, decimals: int = 1) -> str:
@@ -29,32 +36,98 @@ def format_pct(value: float, decimals: int = 1) -> str:
def format_currency(value: float, currency: str = "USD") -> str:
"""Format a number as currency."""
symbol = {"USD": "$", "EUR": "\u20ac", "GBP": "\u00a3"}.get(currency, "")
symbol = {"USD": "$", "EUR": "€", "GBP": "£"}.get(currency, "")
return f"{symbol}{value:,.2f}"
def finalize(
fig: Figure,
fig,
*,
show: bool = False,
show: "bool | str" = False,
save: Optional[Union[str, Path]] = None,
dpi: int = 150,
) -> Figure:
"""Optionally save and/or display the figure, then return it."""
import warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore", UserWarning)
try:
fig.tight_layout()
except Exception:
pass # Skip when axes are incompatible (e.g. inside GridSpec)
window_size: Optional[Tuple[int, int]] = None,
) -> "object":
"""Optionally save and/or display the figure, then return it.
``save`` routes on extension: ``.html`` writes a responsive interactive
page; image extensions (.png/.svg/.pdf/...) go through kaleido.
``show``: ``True`` (or ``"window"``) opens a native window (needs pywebview,
else falls back to a browser tab); ``"browser"`` forces a browser tab.
``dpi`` is kept for backward compatibility and maps to an export scale.
"""
if save is not None:
fig.savefig(str(save), dpi=dpi, bbox_inches="tight")
if show:
plt.show()
path = Path(save)
ext = path.suffix.lower()
if ext in _IMAGE_EXTS:
try:
scale = max(1.0, dpi / 96.0)
fig.write_image(str(path), scale=scale)
except Exception as exc: # kaleido missing or export failure
raise RuntimeError(
f"Static image export to {ext} is optional and needs kaleido. "
"Install it with: pip install manifoldbt[png] "
"(the default is the interactive chart — save to .html)"
) from exc
else:
write_responsive_html(fig, path)
if show == "browser":
fig.show()
elif show: # True or "window" -> native window (browser tab fallback)
from manifoldbt.plot._window import open_in_window
title = "Chart"
try:
t = fig.layout.title.text
if t:
title = t.split("<br>")[0].strip() or title
except Exception:
pass
open_in_window(fig, title=title, size=window_size or (1280, 720))
return fig
def write_responsive_html(fig, path: Union[str, Path]) -> None:
"""Write a self-adjusting full-window HTML page for *fig*.
Strips any fixed width/height so the plot fills (and resizes with) the
window; a clean <title> replaces the browser's filename fallback.
"""
# A fixed layout size would override plotly's responsive resizing.
fig.update_layout(width=None, height=None, autosize=True)
title = "Chart"
try:
t = fig.layout.title.text
if t:
title = t.split("<br>")[0].strip() or title
except Exception:
pass
html = fig.to_html(
include_plotlyjs="cdn",
full_html=True,
default_width="100%",
default_height="100%",
config={"displayModeBar": False, "responsive": True},
)
head = "<head>" + _RESPONSIVE_CSS + f"<title>{title}</title>"
html = html.replace("<head>", head, 1)
Path(path).write_text(html, encoding="utf-8")
def chart_div(fig, *, height: Optional[int] = None) -> str:
"""Return an embeddable div (no plotly.js) for report composition."""
if height is not None:
fig.update_layout(height=height)
fig.update_layout(width=None, autosize=True)
return fig.to_html(
full_html=False,
include_plotlyjs=False,
default_width="100%",
default_height=f"{height}px" if height else "100%",
config={"displayModeBar": False, "responsive": True},
)
def auto_title(result, fallback: str) -> str:
"""Build a title from result manifest strategy_name, or use fallback."""
try:
python/manifoldbt/plot/_window.py
+173
View File
@@ -0,0 +1,173 @@
"""Native frameless windows for charts (pywebview), matplotlib-style.
``show=True`` queues a chart as a borderless native window; a single
``manifoldbt.plot.show()`` (or the automatic one at interpreter exit) opens
ALL queued windows together, so you can have the equity in one window and
the return distribution in another, side by side.
Each window runs in its own child process with a dedicated WebView2 profile:
WebView2 windows sharing one process share one UI thread (several heavy
charts freeze it), and separate processes sharing the default user-data
folder collide on startup. One process + one profile per window avoids both,
keeps every window frameless and responsive, and lets show() be called again
later. show() blocks until all windows are closed (like ``plt.show()``).
Without pywebview each queued chart opens in a browser tab instead. Install
the window backend with ``pip install manifoldbt[window]``.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import webbrowser
from html import escape
from pathlib import Path
from typing import List, Tuple
_SHELL = """<!doctype html><html><head><meta charset="utf-8"><title>{title}</title>
<style>
html,body{{margin:0;height:100%;background:#0c0c0f;overflow:hidden}}
#chart{{height:100vh}} .plotly-graph-div{{width:100%!important;height:100%!important}}
#drag{{position:fixed;top:0;left:0;right:0;height:26px;z-index:5}}
#close{{position:fixed;top:8px;right:10px;z-index:6;width:26px;height:26px;
display:flex;align-items:center;justify-content:center;font-family:Arial,sans-serif;
font-size:17px;color:#8a8a8a;cursor:pointer;border-radius:5px;
background:rgba(17,17,22,0.45);transition:all .12s}}
#close:hover{{background:#ef4444;color:#fff}}
</style></head><body>
<div id="drag" class="pywebview-drag-region"></div>
<div id="close" onclick="pywebview.api.close()" title="Close (Alt+F4)">&times;</div>
<div id="chart">{div}</div>
<script>
function fit(){{var g=document.querySelector('.plotly-graph-div');
if(g&&window.Plotly)Plotly.relayout(g,{{width:window.innerWidth,height:window.innerHeight}});}}
window.addEventListener('resize',fit);
window.addEventListener('load',function(){{fit();requestAnimationFrame(fit);
setTimeout(fit,120);setTimeout(fit,400);}});
</script>
</body></html>"""
# One frameless window; runs inside a dedicated child process.
_CHILD = """
import ctypes, os, sys
try:
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("manifoldbt.plot")
except Exception:
pass
import webview
title, url, x, y, w, h, icon = sys.argv[1:8]
holder = {}
class Api:
def close(self):
win = holder.get("w")
if win is not None:
try:
win.destroy()
except Exception:
pass
holder["w"] = webview.create_window(
title, url, frameless=True, easy_drag=False, js_api=Api(),
width=int(w), height=int(h), x=int(x), y=int(y),
background_color="#0c0c0f",
)
kwargs = dict(storage_path=os.environ.get("MANIFOLDBT_WV_STORAGE"), private_mode=False)
try:
webview.start(icon=icon, **kwargs) if icon and os.path.exists(icon) else webview.start(**kwargs)
except TypeError:
webview.start(**kwargs) # some backends reject the icon kwarg
"""
# Charts registered by show=True, waiting for the next show() / atexit call.
_pending: "List[Tuple[str, str, Tuple[int, int]]]" = []
def _fig_div(fig) -> str:
"""Responsive chart div with plotly.js inlined (instant + offline)."""
fig.update_layout(width=None, height=None, autosize=True)
return fig.to_html(
full_html=False, include_plotlyjs=True,
default_width="100%", default_height="100%",
config={"displayModeBar": False, "responsive": True},
)
def _write_tmp(html: str) -> Path:
tmp = tempfile.NamedTemporaryFile(suffix=".html", delete=False,
mode="w", encoding="utf-8")
tmp.write(html)
tmp.close()
return Path(tmp.name).resolve()
def queue_window(fig, *, title: str = "Chart",
size: Tuple[int, int] = (1280, 720)) -> None:
"""Register *fig* to be shown as a native window on the next show()."""
_pending.append((_fig_div(fig), title, size))
# Backward-compatible alias (was the immediate opener).
open_in_window = queue_window
def show() -> None:
"""Open every chart queued by ``show=True``, each in its own frameless window.
Blocks until all windows are closed (like ``matplotlib.pyplot.show``).
A no-op if nothing is queued; can be called again after more charts are
queued. Falls back to browser tabs when pywebview is not installed.
"""
if not _pending:
return
pending = list(_pending)
_pending.clear()
try:
import webview # noqa: F401 — only to detect the backend
except ImportError:
for div, title, _ in pending:
_open_browser(div, title)
return
icon = Path(__file__).parent / "_assets" / "manifoldbt.ico"
procs = []
for i, (div, title, size) in enumerate(pending):
html = _SHELL.format(title=escape(title), div=div)
path = _write_tmp(html)
env = dict(os.environ)
# Dedicated WebView2 profile: concurrent windows sharing the default
# user-data folder fail to start (window class/profile collision).
storage = tempfile.mkdtemp(prefix="manifoldbt_win_")
env["MANIFOLDBT_WV_STORAGE"] = storage
env["WEBVIEW2_USER_DATA_FOLDER"] = storage
procs.append(subprocess.Popen(
[sys.executable, "-c", _CHILD, title, path.as_uri(),
str(80 + i * 60), str(80 + i * 60),
str(size[0]), str(size[1]), str(icon)],
env=env,
))
for p in procs:
try:
p.wait()
except KeyboardInterrupt:
for q in procs:
if q.poll() is None:
q.terminate()
break
def _open_browser(div: str, title: str) -> None:
# No pywebview: plain browser tab. The OS/browser chrome provides closing.
html = _SHELL.format(title=escape(title),
div=div).replace('pywebview.api.close()', 'window.close()')
webbrowser.open(_write_tmp(html).as_uri())
def _atexit_show() -> None:
# Scripts "just work": show whatever is still queued when the process exits.
show()
import atexit # noqa: E402
atexit.register(_atexit_show)
python/manifoldbt/plot/backtest.py
+270 -198
View File
@@ -1,15 +1,12 @@
"""Charts for BacktestResult visualization."""
"""Charts for BacktestResult visualization (plotly)."""
from __future__ import annotations
from pathlib import Path
from typing import List, Optional, Tuple, Union
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import matplotlib.ticker as mticker
from matplotlib.axes import Axes
from matplotlib.figure import Figure
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from manifoldbt.plot._theme import (
ACCENT,
@@ -29,7 +26,42 @@ from manifoldbt.plot._convert import (
trades_arrays,
_ts_to_int64,
)
from manifoldbt.plot._utils import finalize, format_pct, get_or_create_ax
from manifoldbt.plot._decimate import maybe_decimate
from manifoldbt.plot._utils import finalize, format_pct, new_figure
def _rgba(hex_color: str, alpha: float) -> str:
"""'#rrggbb' -> 'rgba(r,g,b,a)'."""
h = hex_color.lstrip("#")
r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
return f"rgba({r},{g},{b},{alpha})"
def _area_traces(x, y, baseline: float, color: str, *, width: float = 1.5,
name: Optional[str] = None, hovertemplate: Optional[str] = None):
"""Line + fill-to-baseline traces, with a vertical gradient when supported."""
base = go.Scatter(
x=x, y=np.full(len(x), baseline), mode="lines",
line=dict(width=0), hoverinfo="skip", showlegend=False,
)
kwargs = dict(
x=x, y=y, mode="lines",
line=dict(color=color, width=width),
fill="tonexty",
name=name, showlegend=name is not None,
hovertemplate=hovertemplate,
)
try:
line_trace = go.Scatter(
fillgradient=dict(
type="vertical",
colorscale=[[0.0, _rgba(color, 0.0)], [1.0, _rgba(color, 0.22)]],
),
**kwargs,
)
except (ValueError, TypeError): # plotly too old for fillgradient
line_trace = go.Scatter(fillcolor=_rgba(color, 0.07), **kwargs)
return [base, line_trace]
# ── Summary (the essential chart) ────────────────────────────────────────────
@@ -41,29 +73,44 @@ def summary(
figsize: Tuple[float, float] = (14, 8),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""The essential chart: TWR equity + buy-and-hold benchmark, trade activity.
Top panel: TWR-normalized equity curve vs buy-and-hold (close price).
Bottom panel: daily trade count as a bar chart.
Metrics displayed in a clean header line.
Middle panel: daily trade count as a bar chart.
Bottom panel: used margin percentage.
Metrics displayed in the title line.
"""
with theme_context():
fig, (ax_eq, ax_trades, ax_margin) = plt.subplots(
3, 1, figsize=figsize, height_ratios=[3, 1, 1],
sharex=True, gridspec_kw={"hspace": 0.25},
fig = make_subplots(
rows=3, cols=1, shared_xaxes=True, vertical_spacing=0.06,
row_heights=[0.6, 0.2, 0.2],
)
dates, eq_vals = equity_with_dates(result)
metrics = result.metrics if hasattr(result, "metrics") else {}
# ── TWR equity (normalized to 100) ────────────────────────
twr = eq_vals / eq_vals[0] * 100
ax_eq.plot(dates, twr, color=ACCENT, linewidth=0.8, label="Strategy")
ax_eq.fill_between(dates, twr, 100, where=(twr >= 100),
color=GREEN, alpha=0.04, interpolate=True)
ax_eq.fill_between(dates, twr, 100, where=(twr < 100),
color=RED, alpha=0.04, interpolate=True)
twr_full = eq_vals / eq_vals[0] * 100
d_dates, twr = maybe_decimate(dates, twr_full)
fig.add_trace(go.Scatter(
x=d_dates, y=twr, mode="lines", name="Strategy",
line=dict(color=ACCENT, width=1.0),
hovertemplate="%{x|%d %b %Y} %{y:.1f}<extra>Strategy</extra>",
), row=1, col=1)
# Faint green/red fill vs the 100 baseline
for clip_lo, clip_hi, color in ((100.0, None, GREEN), (None, 100.0, RED)):
clipped = np.clip(twr, clip_lo, clip_hi)
fig.add_trace(go.Scatter(
x=d_dates, y=np.full(len(d_dates), 100.0), mode="lines",
line=dict(width=0), hoverinfo="skip", showlegend=False,
), row=1, col=1)
fig.add_trace(go.Scatter(
x=d_dates, y=clipped, mode="lines", line=dict(width=0),
fill="tonexty", fillcolor=_rgba(color, 0.04),
hoverinfo="skip", showlegend=False,
), row=1, col=1)
# ── Benchmark: buy-and-hold from close prices ─────────────
positions = result.positions
@@ -78,7 +125,7 @@ def summary(
benchmark_raw = close_vals / close_vals[0] * 100
# Vol-adjusted benchmark: scale to same volatility as strategy
strat_rets = np.diff(twr) / twr[:-1]
strat_rets = np.diff(twr_full) / twr_full[:-1]
bench_rets = np.diff(benchmark_raw) / benchmark_raw[:-1]
strat_vol = np.nanstd(strat_rets)
bench_vol = np.nanstd(bench_rets)
@@ -90,16 +137,19 @@ def summary(
else:
benchmark = benchmark_raw
ax_eq.plot(dates, benchmark, color=GRAY, linewidth=1.0,
label="Buy & Hold (vol-adj)", alpha=0.7)
b_dates, b_vals = maybe_decimate(dates[: len(benchmark)], benchmark)
fig.add_trace(go.Scatter(
x=b_dates, y=b_vals, mode="lines", name="Buy & Hold (vol-adj)",
line=dict(color=GRAY, width=1.0), opacity=0.7,
hovertemplate="%{x|%d %b %Y} %{y:.1f}<extra>Buy & Hold</extra>",
), row=1, col=1)
ax_eq.axhline(100, color=DARK_GRAY, linewidth=0.4)
# Ensure y-axis zooms to strategy range with some padding
twr_min, twr_max = float(np.nanmin(twr)), float(np.nanmax(twr))
fig.add_hline(y=100, line_color=DARK_GRAY, line_width=0.4, row=1, col=1)
twr_min, twr_max = float(np.nanmin(twr_full)), float(np.nanmax(twr_full))
twr_range = max(twr_max - twr_min, 0.1)
ax_eq.set_ylim(twr_min - twr_range * 0.15, twr_max + twr_range * 0.15)
ax_eq.set_ylabel("TWR (base 100)", fontsize=9)
ax_eq.legend(loc="upper left", framealpha=0.3, fontsize=8)
fig.update_yaxes(title_text="TWR (base 100)",
range=[twr_min - twr_range * 0.15, twr_max + twr_range * 0.15],
row=1, col=1)
# Header metrics
ret = metrics.get("total_return", 0)
@@ -112,10 +162,9 @@ def summary(
f" Max DD {mdd * 100:.1f}%"
f" Trades {n_trades:,}"
)
ax_eq.set_title(title, fontsize=10, loc="left", pad=10)
fig.update_layout(title_text=title)
# ── Adaptive smoothing window ──────────────────────────────
# Scale window: min(7d, max(1d, 5% of total period))
smooth_label = ""
if len(dates) >= 2:
bar_ns = int(dates[1]) - int(dates[0])
@@ -126,22 +175,22 @@ def summary(
smooth_window = min(smooth_window, len(dates))
smooth_days = round(target_ns / day_ns)
smooth_label = f" ({smooth_days}d)" if smooth_days >= 1 else ""
else:
smooth_window = 1
# ── Trade activity (daily trade count) ─────────────────────
try:
ta = trades_arrays(result)
trade_ts = ta.get("execution_timestamp", np.array([], dtype="datetime64[ns]"))
if len(trade_ts) > 0 and len(dates) >= 2:
# Bucket trades into calendar days
trade_days = trade_ts.astype("datetime64[D]")
unique_days, day_counts = np.unique(trade_days, return_counts=True)
day_dates = unique_days.astype("datetime64[ns]")
ax_trades.bar(day_dates, day_counts,
width=np.timedelta64(1, "D"),
color=ACCENT_ALT, alpha=0.4, edgecolor="none")
fig.add_trace(go.Bar(
x=day_dates, y=day_counts, name="Trades/day",
marker_color=_rgba(ACCENT_ALT, 0.4), marker_line_width=0,
showlegend=False,
hovertemplate="%{x|%d %b %Y} %{y} trades<extra></extra>",
), row=2, col=1)
# Rolling 7-day average overlay
eq_days = dates.astype("datetime64[D]")
@@ -154,16 +203,14 @@ def summary(
if win > 1:
kernel = np.ones(win) / win
smoothed = np.convolve(daily_on_grid, kernel, mode="same")
ax_trades.plot(unique_eq_days.astype("datetime64[ns]"), smoothed,
color=ACCENT_ALT, linewidth=1.0, alpha=0.8)
else:
ax_trades.text(0.5, 0.5, "No trade data", transform=ax_trades.transAxes,
ha="center", va="center", color=DARK_GRAY, fontsize=9)
fig.add_trace(go.Scatter(
x=unique_eq_days.astype("datetime64[ns]"), y=smoothed,
mode="lines", line=dict(color=ACCENT_ALT, width=1.0),
opacity=0.8, showlegend=False, hoverinfo="skip",
), row=2, col=1)
except Exception:
ax_trades.text(0.5, 0.5, "No trade data", transform=ax_trades.transAxes,
ha="center", va="center", color=DARK_GRAY, fontsize=9)
ax_trades.set_ylabel("Trades/day", fontsize=8)
pass
fig.update_yaxes(title_text="Trades/day", row=2, col=1)
# ── Used margin % (daily) ──────────────────────────────
try:
@@ -183,29 +230,27 @@ def summary(
# Resample to daily (end-of-day snapshot)
days = used_dates.astype("datetime64[D]")
unique_days, _ = np.unique(days, return_index=True)
# Use last value per day (not first) for end-of-day margin
day_last = np.searchsorted(days, unique_days, side="right") - 1
daily_used = used[day_last]
daily_dates = unique_days.astype("datetime64[ns]")
ax_margin.fill_between(daily_dates, 0, daily_used,
color=GREEN, alpha=0.10, edgecolor="none")
ax_margin.plot(daily_dates, daily_used,
color=GREEN, linewidth=0.7, alpha=0.8)
ax_margin.axhline(0, color=DARK_GRAY, linewidth=0.4)
fig.add_trace(go.Scatter(
x=daily_dates, y=daily_used, mode="lines",
line=dict(color=GREEN, width=0.7), opacity=0.8,
fill="tozeroy", fillcolor=_rgba(GREEN, 0.10),
showlegend=False,
hovertemplate="%{x|%d %b %Y} %{y:.1f}%<extra>Margin</extra>",
), row=3, col=1)
except Exception:
ax_margin.text(
0.5, 0.5, "No position data",
transform=ax_margin.transAxes,
ha="center", va="center", color=DARK_GRAY, fontsize=9,
)
ax_margin.set_ylabel(f"Margin %{smooth_label}", fontsize=8)
ax_margin.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
ax_margin.xaxis.set_major_locator(mdates.AutoDateLocator())
fig.align_ylabels([ax_eq, ax_trades, ax_margin])
fig.autofmt_xdate(rotation=0, ha="center")
pass
fig.update_yaxes(title_text=f"Margin %{smooth_label}", row=3, col=1)
fig.update_layout(
width=int(figsize[0] * 80), height=int(figsize[1] * 80),
legend=dict(orientation="h", yanchor="bottom", y=1.02,
xanchor="right", x=1),
bargap=0.0,
)
return finalize(fig, show=show, save=save)
@@ -215,24 +260,27 @@ def summary(
def equity(
result,
*,
ax: Optional[Axes] = None,
ax=None,
color: str = ACCENT,
title: str = "Equity Curve",
figsize: Tuple[float, float] = (14, 5),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
"""Plot the portfolio equity curve over time."""
) -> go.Figure:
"""Plot the portfolio equity curve over time.
``ax`` is accepted for backward compatibility and ignored (plotly backend).
"""
with theme_context():
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
dates, values = equity_with_dates(result)
ax_.plot(dates, values, color=color, linewidth=1.3)
ax_.fill_between(dates, values, values.min(), color=color, alpha=0.05)
ax_.set_title(title)
ax_.set_ylabel("Equity", fontsize=9)
ax_.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
ax_.xaxis.set_major_locator(mdates.AutoDateLocator())
fig.autofmt_xdate(rotation=0, ha="center")
dates, values = maybe_decimate(dates, values)
fig.add_traces(_area_traces(
dates, values, float(values.min()), color, width=1.5,
hovertemplate="%{x|%d %b %Y} $%{y:,.0f}<extra></extra>",
))
fig.update_yaxes(title_text="Equity")
fig.update_xaxes(tickformat="%b %Y")
return finalize(fig, show=show, save=save)
@@ -243,7 +291,7 @@ def benchmark_equity(
result,
benchmark: np.ndarray,
*,
ax: Optional[Axes] = None,
ax=None,
strategy_color: str = ACCENT,
benchmark_color: str = DARK_GRAY,
normalize: bool = True,
@@ -252,10 +300,10 @@ def benchmark_equity(
figsize: Tuple[float, float] = (14, 5),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Overlay strategy equity and a benchmark, both normalized to 100."""
with theme_context():
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
dates, strat_eq = equity_with_dates(result)
bench = np.asarray(benchmark, dtype=np.float64)
n = min(len(strat_eq), len(bench))
@@ -265,13 +313,19 @@ def benchmark_equity(
strat_eq = strat_eq / strat_eq[0] * 100
bench = bench / bench[0] * 100
ax_.plot(dates, strat_eq, color=strategy_color, linewidth=1.3, label=labels[0])
ax_.plot(dates, bench, color=benchmark_color, linewidth=1.0, label=labels[1])
ax_.set_title(title)
ax_.set_ylabel("Normalized" if normalize else "Equity")
ax_.legend(loc="upper left", framealpha=0.5)
ax_.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
fig.autofmt_xdate(rotation=0, ha="center")
d1, s1 = maybe_decimate(dates, strat_eq)
d2, b1 = maybe_decimate(dates, bench)
fig.add_trace(go.Scatter(
x=d1, y=s1, mode="lines", name=labels[0],
line=dict(color=strategy_color, width=1.5),
))
fig.add_trace(go.Scatter(
x=d2, y=b1, mode="lines", name=labels[1],
line=dict(color=benchmark_color, width=1.0),
))
fig.update_yaxes(title_text="Normalized" if normalize else "Equity")
fig.update_xaxes(tickformat="%b %Y")
fig.update_layout(legend=dict(x=0.01, y=0.99))
return finalize(fig, show=show, save=save)
@@ -281,28 +335,31 @@ def benchmark_equity(
def drawdown(
result,
*,
ax: Optional[Axes] = None,
ax=None,
color: str = RED,
title: str = "Drawdown",
figsize: Tuple[float, float] = (14, 3),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Plot the drawdown as a filled area chart."""
with theme_context():
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
dates, values = equity_with_dates(result)
running_max = np.maximum.accumulate(values)
dd = (values - running_max) / running_max
dates, dd = maybe_decimate(dates, dd)
ax_.fill_between(dates, dd, 0, color=color, alpha=0.25)
ax_.plot(dates, dd, color=color, linewidth=0.8)
ax_.set_title(title)
ax_.set_ylabel("Drawdown")
ax_.yaxis.set_major_formatter(mticker.PercentFormatter(xmax=1.0, decimals=0))
ax_.set_ylim(top=0)
ax_.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
fig.autofmt_xdate(rotation=0, ha="center")
fig.add_trace(go.Scatter(
x=dates, y=dd, mode="lines",
line=dict(color=color, width=0.9),
fill="tozeroy", fillcolor=_rgba(color, 0.25),
hovertemplate="%{x|%d %b %Y} %{y:.1%}<extra></extra>",
))
dd_min = float(dd.min()) if len(dd) else -0.01
fig.update_yaxes(title_text="Drawdown", tickformat=".0%",
range=[dd_min * 1.08, 0])
fig.update_xaxes(tickformat="%b %Y")
return finalize(fig, show=show, save=save)
@@ -312,14 +369,16 @@ def drawdown(
def monthly_returns(
result,
*,
ax: Optional[Axes] = None,
ax=None,
annotate: bool = True,
title: str = "Monthly Returns (%)",
figsize: Tuple[float, float] = (12, 5),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Monthly returns heatmap (year rows x month columns + annual)."""
from manifoldbt.plot._theme import CS_DIVERGING
with theme_context():
dates, values = equity_with_dates(result)
ts = dates.astype("datetime64[M]")
@@ -344,31 +403,27 @@ def monthly_returns(
if len(valid) > 0:
grid[yi, 12] = np.prod(1.0 + valid) - 1.0
fig, ax_ = get_or_create_ax(ax, figsize)
abs_max = max(np.nanmax(np.abs(grid)), 0.01)
cmap = plt.get_cmap("bt_diverging")
im = ax_.imshow(grid, cmap=cmap, aspect="auto", vmin=-abs_max, vmax=abs_max)
month_labels = ["Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec", "YTD"]
ax_.set_xticks(range(13))
ax_.set_xticklabels(month_labels, fontsize=8)
ax_.set_yticks(range(len(years)))
ax_.set_yticklabels([str(y) for y in years], fontsize=9)
if annotate:
for yi in range(len(years)):
for mi in range(13):
val = grid[yi, mi]
if np.isnan(val):
continue
txt = f"{val * 100:+.1f}"
brightness = abs(val) / abs_max
txt_color = WHITE if brightness > 0.4 else GRAY
ax_.text(mi, yi, txt, ha="center", va="center",
fontsize=7, color=txt_color, fontweight="medium")
text = np.where(np.isnan(grid), "", np.vectorize(lambda v: f"{v * 100:+.1f}" if not np.isnan(v) else "")(grid))
ax_.set_title(title)
fig = new_figure(figsize, title)
fig.add_trace(go.Heatmap(
z=grid * 100, x=month_labels, y=[str(y) for y in years],
colorscale=CS_DIVERGING, zmin=-abs_max * 100, zmax=abs_max * 100,
text=text if annotate else None,
texttemplate="%{text}" if annotate else None,
textfont=dict(size=10),
hovertemplate="%{y} %{x}: %{z:+.2f}%<extra></extra>",
colorbar=dict(ticksuffix="%", outlinewidth=0, thickness=12),
hoverongaps=False,
))
fig.update_yaxes(autorange="reversed")
fig.update_xaxes(side="bottom", showspikes=False)
fig.update_yaxes(showspikes=False)
fig.update_layout(hovermode="closest")
return finalize(fig, show=show, save=save)
@@ -378,12 +433,12 @@ def monthly_returns(
def annual_returns(
result,
*,
ax: Optional[Axes] = None,
ax=None,
title: str = "Annual Returns",
figsize: Tuple[float, float] = (10, 4),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Annual returns bar chart with green/red conditional coloring."""
with theme_context():
dates, values = equity_with_dates(result)
@@ -394,19 +449,20 @@ def annual_returns(
idx = np.nonzero(years_arr == y)[0]
ann_rets.append(values[idx[-1]] / values[idx[0]] - 1.0 if len(idx) >= 2 else 0.0)
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
colors = [GREEN if r >= 0 else RED for r in ann_rets]
bars = ax_.bar([str(y) for y in unique_years], ann_rets, color=colors,
width=0.5, alpha=0.85, edgecolor="none")
ax_.axhline(0, color=DARK_GRAY, linewidth=0.5)
ax_.set_title(title)
ax_.yaxis.set_major_formatter(mticker.PercentFormatter(xmax=1.0, decimals=0))
for bar, ret in zip(bars, ann_rets):
ax_.text(bar.get_x() + bar.get_width() / 2, bar.get_height(),
format_pct(ret), ha="center",
va="bottom" if ret >= 0 else "top",
fontsize=8, color=GRAY)
fig.add_trace(go.Bar(
x=[str(y) for y in unique_years], y=ann_rets,
marker_color=colors, opacity=0.85, marker_line_width=0,
width=0.5,
text=[format_pct(r) for r in ann_rets],
textposition="outside", textfont=dict(color=GRAY, size=11),
hovertemplate="%{x}: %{y:.1%}<extra></extra>",
))
fig.add_hline(y=0, line_color=DARK_GRAY, line_width=0.5)
fig.update_yaxes(tickformat=".0%")
fig.update_xaxes(showspikes=False, type="category")
fig.update_layout(hovermode="closest")
return finalize(fig, show=show, save=save)
@@ -416,19 +472,19 @@ def annual_returns(
def returns_histogram(
result,
*,
ax: Optional[Axes] = None,
ax=None,
bins: int = 100,
title: str = "Returns Distribution",
figsize: Tuple[float, float] = (12, 5),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Histogram of daily returns with green/red coloring by sign."""
with theme_context():
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
rets = daily_returns_array(result)
if len(rets) == 0:
ax_.set_title(title + " (no data)")
fig.update_layout(title_text=title + " (no data)")
return finalize(fig, show=show, save=save)
# Clip x-axis to P1-P99 range to avoid empty space from outliers
@@ -436,28 +492,34 @@ def returns_histogram(
margin = (p99 - p1) * 0.3
xlim = (p1 - margin, p99 + margin)
_, bin_edges, patches = ax_.hist(rets, bins=bins, edgecolor="none", alpha=0.7,
range=xlim)
for patch, left in zip(patches, bin_edges[:-1]):
patch.set_facecolor(GREEN if left >= 0 else RED)
counts, bin_edges = np.histogram(rets, bins=bins, range=xlim)
centers = (bin_edges[:-1] + bin_edges[1:]) / 2
bw = bin_edges[1] - bin_edges[0]
colors = [GREEN if left >= 0 else RED for left in bin_edges[:-1]]
ax_.axvline(0, color=DARK_GRAY, linewidth=0.8, linestyle="--")
ax_.set_xlim(xlim)
fig.add_trace(go.Bar(
x=centers, y=counts, width=bw,
marker_color=colors, opacity=0.7, marker_line_width=0,
hovertemplate="%{x:.2%}: %{y}<extra></extra>",
))
fig.add_vline(x=0, line_color=DARK_GRAY, line_width=0.8, line_dash="dash")
# Normal fit (pure numpy)
mu, sigma = rets.mean(), rets.std()
if sigma > 0:
x = np.linspace(xlim[0], xlim[1], 200)
bw = bin_edges[1] - bin_edges[0]
pdf = (1 / (sigma * np.sqrt(2 * np.pi))) * np.exp(-0.5 * ((x - mu) / sigma) ** 2)
ax_.plot(x, pdf * len(rets) * bw, color=ACCENT, linewidth=1.0,
alpha=0.7, label="Normal")
ax_.legend(loc="upper right", framealpha=0.3)
fig.add_trace(go.Scatter(
x=x, y=pdf * len(rets) * bw, mode="lines", name="Normal",
line=dict(color=ACCENT, width=1.0), opacity=0.7,
hoverinfo="skip",
))
fig.update_layout(legend=dict(x=0.99, y=0.99, xanchor="right"))
ax_.set_title(title)
ax_.set_xlabel("Daily Return")
ax_.set_ylabel("Frequency")
ax_.xaxis.set_major_formatter(mticker.PercentFormatter(xmax=1.0, decimals=1))
fig.update_xaxes(title_text="Daily Return", tickformat=".1%",
range=list(xlim))
fig.update_yaxes(title_text="Frequency")
fig.update_layout(hovermode="closest", bargap=0.05)
return finalize(fig, show=show, save=save)
@@ -467,60 +529,66 @@ def returns_histogram(
def var_chart(
result,
*,
ax: Optional[Axes] = None,
ax=None,
confidence: float = 0.05,
bins: int = 120,
title: str = "Value at Risk",
figsize: Tuple[float, float] = (12, 5),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Returns histogram with VaR and CVaR lines at 5% and 1% levels."""
with theme_context():
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
rets = daily_returns_array(result)
if len(rets) == 0:
ax_.set_title(title + " (no data)")
fig.update_layout(title_text=title + " (no data)")
return finalize(fig, show=show, save=save)
rets_pct = rets * 100
# Histogram
n, bin_edges, patches = ax_.hist(
rets_pct, bins=bins, color=ACCENT, alpha=0.5, edgecolor="none",
)
# VaR/CVaR at 5%
# VaR/CVaR at 5% and 1%
var_5 = float(np.percentile(rets, 5))
cvar_5 = float(rets[rets <= var_5].mean()) if np.any(rets <= var_5) else var_5
# VaR/CVaR at 1%
var_1 = float(np.percentile(rets, 1))
cvar_1 = float(rets[rets <= var_1].mean()) if np.any(rets <= var_1) else var_1
# Color tail bins
for b, p in zip(bin_edges, patches):
if b < var_1 * 100:
p.set_facecolor(RED)
p.set_alpha(0.5)
elif b < var_5 * 100:
p.set_facecolor(ORANGE)
p.set_alpha(0.4)
counts, bin_edges = np.histogram(rets_pct, bins=bins)
centers = (bin_edges[:-1] + bin_edges[1:]) / 2
bw = bin_edges[1] - bin_edges[0]
colors = []
for left in bin_edges[:-1]:
if left < var_1 * 100:
colors.append(_rgba(RED, 0.5))
elif left < var_5 * 100:
colors.append(_rgba(ORANGE, 0.4))
else:
colors.append(_rgba(ACCENT, 0.5))
# VaR lines
ax_.axvline(var_5 * 100, color=ORANGE, linewidth=0.8,
label=f"VaR 5%: {format_pct(var_5)}")
ax_.axvline(cvar_5 * 100, color=ORANGE, linewidth=0.6, linestyle="--", alpha=0.5,
label=f"CVaR 5%: {format_pct(cvar_5)}")
ax_.axvline(var_1 * 100, color=RED, linewidth=0.8,
label=f"VaR 1%: {format_pct(var_1)}")
ax_.axvline(cvar_1 * 100, color=RED, linewidth=0.6, linestyle="--", alpha=0.5,
label=f"CVaR 1%: {format_pct(cvar_1)}")
fig.add_trace(go.Bar(
x=centers, y=counts, width=bw, marker_color=colors,
marker_line_width=0, showlegend=False,
hovertemplate="%{x:.2f}%: %{y}<extra></extra>",
))
ax_.set_title(title)
ax_.set_xlabel("Daily Return (%)")
ax_.set_ylabel("Frequency")
ax_.legend(loc="upper right", fontsize=8, framealpha=0.3)
# VaR/CVaR lines with legend proxies
for val, color, dash, label in (
(var_5, ORANGE, None, f"VaR 5%: {format_pct(var_5)}"),
(cvar_5, ORANGE, "dash", f"CVaR 5%: {format_pct(cvar_5)}"),
(var_1, RED, None, f"VaR 1%: {format_pct(var_1)}"),
(cvar_1, RED, "dash", f"CVaR 1%: {format_pct(cvar_1)}"),
):
fig.add_vline(x=val * 100, line_color=color, line_width=0.8,
line_dash=dash, opacity=0.8 if dash is None else 0.5)
fig.add_trace(go.Scatter(
x=[None], y=[None], mode="lines", name=label,
line=dict(color=color, width=1.2, dash=dash),
))
fig.update_xaxes(title_text="Daily Return (%)")
fig.update_yaxes(title_text="Frequency")
fig.update_layout(hovermode="closest", bargap=0.05,
legend=dict(x=0.99, y=0.99, xanchor="right"))
return finalize(fig, show=show, save=save)
@@ -531,20 +599,20 @@ def rolling_sharpe(
result,
*,
windows: Optional[List[int]] = None,
ax: Optional[Axes] = None,
ax=None,
title: str = "Rolling Sharpe",
trading_days_per_year: float = 365.25,
figsize: Tuple[float, float] = (14, 4),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Rolling annualized Sharpe ratio."""
if windows is None:
windows = [126, 252]
colors = [ACCENT, ACCENT_ALT, GREEN, RED]
with theme_context():
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
rets = daily_returns_array(result)
for i, w in enumerate(windows):
@@ -554,13 +622,15 @@ def rolling_sharpe(
rs = _rolling(rets, w, np.std)
with np.errstate(divide="ignore", invalid="ignore"):
sharpe = np.where(rs > 0, rm / rs * np.sqrt(trading_days_per_year), 0.0)
label = f"{w}d"
ax_.plot(sharpe, color=colors[i % len(colors)], linewidth=1.0, label=label)
fig.add_trace(go.Scatter(
y=sharpe, mode="lines", name=f"{w}d",
line=dict(color=colors[i % len(colors)], width=1.0),
hovertemplate="day %{x}: %{y:.2f}<extra>" + f"{w}d" + "</extra>",
))
ax_.axhline(0, color=DARK_GRAY, linewidth=0.5, linestyle="--")
ax_.set_title(title)
ax_.set_ylabel("Sharpe")
ax_.legend(loc="upper left", framealpha=0.3)
fig.add_hline(y=0, line_color=DARK_GRAY, line_width=0.5, line_dash="dash")
fig.update_yaxes(title_text="Sharpe")
fig.update_layout(legend=dict(x=0.01, y=0.99))
return finalize(fig, show=show, save=save)
@@ -571,20 +641,20 @@ def rolling_volatility(
result,
*,
windows: Optional[List[int]] = None,
ax: Optional[Axes] = None,
ax=None,
title: str = "Rolling Volatility",
trading_days_per_year: float = 365.25,
figsize: Tuple[float, float] = (14, 4),
show: bool = False,
save: Optional[Union[str, Path]] = None,
) -> Figure:
) -> go.Figure:
"""Rolling annualized volatility."""
if windows is None:
windows = [126, 252]
colors = [ACCENT, ACCENT_ALT, GREEN, RED]
with theme_context():
fig, ax_ = get_or_create_ax(ax, figsize)
fig = new_figure(figsize, title)
rets = daily_returns_array(result)
for i, w in enumerate(windows):
@@ -592,12 +662,14 @@ def rolling_volatility(
continue
rs = _rolling(rets, w, np.std)
vol = rs * np.sqrt(trading_days_per_year)
ax_.plot(vol, color=colors[i % len(colors)], linewidth=1.0, label=f"{w}d")
fig.add_trace(go.Scatter(
y=vol, mode="lines", name=f"{w}d",
line=dict(color=colors[i % len(colors)], width=1.0),
hovertemplate="day %{x}: %{y:.1%}<extra>" + f"{w}d" + "</extra>",
))
ax_.set_title(title)
ax_.set_ylabel("Volatility")
ax_.yaxis.set_major_formatter(mticker.PercentFormatter(xmax=1.0, decimals=0))
ax_.legend(loc="upper left", framealpha=0.3)
fig.update_yaxes(title_text="Volatility", tickformat=".0%")
fig.update_layout(legend=dict(x=0.01, y=0.99))
return finalize(fig, show=show, save=save)
python/manifoldbt/plot/chart.py
+147 -300
View File
@@ -1,4 +1,4 @@
"""Candlestick chart with indicators and trade markers."""
"""Candlestick chart with indicators and trade markers (plotly)."""
from __future__ import annotations
from pathlib import Path
@@ -12,13 +12,9 @@ from manifoldbt.plot._theme import (
ACCENT_ALT,
BG_AXES,
BG_FIGURE,
BORDER,
DARK_GRAY,
GREEN,
GRID_RGBA,
GRAY,
RED,
WHITE,
theme_context,
)
from manifoldbt.plot._utils import finalize
@@ -143,49 +139,7 @@ def _load_bars(
# ---------------------------------------------------------------------------
# Candlestick drawing
# ---------------------------------------------------------------------------
def _draw_candles(ax, dates, o, h, l, c, width_ratio=0.6):
"""Draw candlestick bodies and wicks on an axes."""
n = len(dates)
if n < 2:
return
# Width in date units
delta = np.median(np.diff(dates)).astype("timedelta64[s]").astype(float)
w = np.timedelta64(int(delta * width_ratio), "s")
bull = c >= o
bear = ~bull
# Wicks (high-low lines)
for i in range(n):
color = GREEN if bull[i] else RED
ax.plot([dates[i], dates[i]], [l[i], h[i]], color=color, linewidth=0.5, alpha=0.7)
# Bodies
for mask, color in [(bull, GREEN), (bear, RED)]:
idx = np.where(mask)[0]
for i in idx:
bottom = min(o[i], c[i])
height = abs(c[i] - o[i])
if height < 1e-10:
height = (h[i] - l[i]) * 0.01
rect = __import__("matplotlib.patches", fromlist=["Rectangle"]).Rectangle(
(dates[i] - w / 2, bottom),
w,
height,
facecolor=color,
edgecolor=color,
alpha=0.85,
linewidth=0.5,
)
ax.add_patch(rect)
# ---------------------------------------------------------------------------
# Public API
# Shared helpers
# ---------------------------------------------------------------------------
INDICATOR_COLORS = [ACCENT, ACCENT_ALT, "#2dd4bf", "#f59e0b", "#f472b6"]
@@ -206,7 +160,7 @@ def _resolve_sym_name(store, symbol_id: int) -> str:
def _prepare_chart_data(result, store, symbol_id, n_bars):
"""Load bars, compute trim offset, extract trades — shared by both renderers."""
"""Load bars, compute trim offset, extract trades."""
manifest = result.manifest
cfg = manifest.get("config", {})
tr = cfg.get("time_range", {})
@@ -244,243 +198,6 @@ def _prepare_chart_data(result, store, symbol_id, n_bars):
}
# ---------------------------------------------------------------------------
# Interactive chart (plotly)
# ---------------------------------------------------------------------------
def _chart_interactive(result, store, symbol_id, *, emas, smas, n_bars, save):
"""Plotly-based interactive candlestick chart."""
import plotly.graph_objects as go
from plotly.subplots import make_subplots
bars, offset, bar_interval_s, filtered_trades = _prepare_chart_data(
result, store, symbol_id, n_bars,
)
close_full = bars["close"]
ts = bars["timestamp"][offset:]
o = bars["open"][offset:]
h = bars["high"][offset:]
l = bars["low"][offset:]
c = bars["close"][offset:]
vol = bars["volume"][offset:]
dates = ts.view("datetime64[ns]")
sym_name = _resolve_sym_name(store, symbol_id)
interval_label = _interval_label(bar_interval_s)
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
row_heights=[0.8, 0.2],
)
# Candlesticks
fig.add_trace(
go.Candlestick(
x=dates, open=o, high=h, low=l, close=c,
increasing_line_color=GREEN, decreasing_line_color=RED,
increasing_fillcolor=GREEN, decreasing_fillcolor=RED,
name="OHLC",
),
row=1, col=1,
)
# Indicators
color_idx = 0
if emas:
for period in emas:
vals = _ema(close_full, period)[offset:]
color = INDICATOR_COLORS[color_idx % len(INDICATOR_COLORS)]
fig.add_trace(
go.Scatter(
x=dates, y=vals, mode="lines",
name=f"EMA({period})",
line=dict(color=color, width=1.5),
),
row=1, col=1,
)
color_idx += 1
if smas:
for period in smas:
vals = _sma(close_full, period)[offset:]
color = INDICATOR_COLORS[color_idx % len(INDICATOR_COLORS)]
fig.add_trace(
go.Scatter(
x=dates, y=vals, mode="lines",
name=f"SMA({period})",
line=dict(color=color, width=1.5, dash="dash"),
),
row=1, col=1,
)
color_idx += 1
# Trade markers
t_ts = filtered_trades["ts"]
t_side = filtered_trades["side"]
t_price = filtered_trades["price"]
t_qty = filtered_trades["qty"]
buy_mask = t_side == 1
sell_mask = t_side == 2
if buy_mask.any():
fig.add_trace(
go.Scatter(
x=t_ts[buy_mask], y=t_price[buy_mask],
mode="markers",
name="BUY",
marker=dict(
symbol="triangle-up", size=12,
color=GREEN, line=dict(color="white", width=1),
),
text=[f"BUY {q:.6f} @ {p:.2f}" for q, p in
zip(t_qty[buy_mask], t_price[buy_mask])],
hoverinfo="text+x",
),
row=1, col=1,
)
if sell_mask.any():
fig.add_trace(
go.Scatter(
x=t_ts[sell_mask], y=t_price[sell_mask],
mode="markers",
name="SELL",
marker=dict(
symbol="triangle-down", size=12,
color=RED, line=dict(color="white", width=1),
),
text=[f"SELL {q:.6f} @ {p:.2f}" for q, p in
zip(t_qty[sell_mask], t_price[sell_mask])],
hoverinfo="text+x",
),
row=1, col=1,
)
# Volume bars
vol_colors = [GREEN if c[i] >= o[i] else RED for i in range(len(c))]
fig.add_trace(
go.Bar(
x=dates, y=vol, name="Volume",
marker_color=vol_colors, opacity=0.5,
showlegend=False,
),
row=2, col=1,
)
# Layout — dark theme
fig.update_layout(
title=f"{sym_name} {interval_label}",
template="plotly_dark",
paper_bgcolor=BG_FIGURE,
plot_bgcolor=BG_AXES,
xaxis_rangeslider_visible=False,
hovermode="x unified",
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0),
height=700,
margin=dict(l=60, r=20, t=60, b=40),
)
fig.update_yaxes(title_text="Price", row=1, col=1)
fig.update_yaxes(title_text="Vol", row=2, col=1)
if save:
fig.write_html(str(save))
fig.show()
return fig
# ---------------------------------------------------------------------------
# Matplotlib (static) chart
# ---------------------------------------------------------------------------
def _chart_matplotlib(result, store, symbol_id, *, emas, smas, n_bars, figsize, show, save):
"""Matplotlib-based static candlestick chart."""
import matplotlib.pyplot as plt
from manifoldbt.plot._theme import theme_context
bars, offset, bar_interval_s, filtered_trades = _prepare_chart_data(
result, store, symbol_id, n_bars,
)
close_full = bars["close"]
ts = bars["timestamp"][offset:]
o = bars["open"][offset:]
h = bars["high"][offset:]
l = bars["low"][offset:]
c = bars["close"][offset:]
dates = ts.view("datetime64[ns]")
sym_name = _resolve_sym_name(store, symbol_id)
interval_label = _interval_label(bar_interval_s)
with theme_context():
fig, (ax_price, ax_vol) = plt.subplots(
2, 1, figsize=figsize, height_ratios=[4, 1],
sharex=True, gridspec_kw={"hspace": 0.05},
)
_draw_candles(ax_price, dates, o, h, l, c)
color_idx = 0
if emas:
for period in emas:
vals = _ema(close_full, period)[offset:]
color = INDICATOR_COLORS[color_idx % len(INDICATOR_COLORS)]
ax_price.plot(dates, vals, color=color, linewidth=1.2,
label=f"EMA({period})", alpha=0.9)
color_idx += 1
if smas:
for period in smas:
vals = _sma(close_full, period)[offset:]
color = INDICATOR_COLORS[color_idx % len(INDICATOR_COLORS)]
ax_price.plot(dates, vals, color=color, linewidth=1.2,
label=f"SMA({period})", linestyle="--", alpha=0.9)
color_idx += 1
# Trade markers
t_ts = filtered_trades["ts"]
t_side = filtered_trades["side"]
t_price = filtered_trades["price"]
buy_mask = t_side == 1
sell_mask = t_side == 2
if buy_mask.any():
ax_price.scatter(
t_ts[buy_mask], t_price[buy_mask],
marker="^", color=GREEN, s=80, zorder=5,
edgecolors=WHITE, linewidths=0.5, label="BUY",
)
if sell_mask.any():
ax_price.scatter(
t_ts[sell_mask], t_price[sell_mask],
marker="v", color=RED, s=80, zorder=5,
edgecolors=WHITE, linewidths=0.5, label="SELL",
)
ax_price.legend(loc="upper left", fontsize=8)
ax_price.set_title(f"{sym_name} {interval_label}", fontsize=11, loc="left")
ax_price.set_ylabel("Price", fontsize=9)
vol = bars["volume"][offset:]
vol_colors = np.where(c >= o, GREEN, RED)
ax_vol.bar(dates, vol, width=np.timedelta64(int(bar_interval_s * 0.6), "s"),
color=vol_colors, alpha=0.5)
ax_vol.set_ylabel("Volume", fontsize=9)
import matplotlib.dates as mdates
if bar_interval_s < 86400:
ax_vol.xaxis.set_major_locator(mdates.AutoDateLocator())
ax_vol.xaxis.set_major_formatter(mdates.DateFormatter("%m-%d %H:%M"))
fig.autofmt_xdate(rotation=30, ha="right")
return finalize(fig, show=show, save=save)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
@@ -507,22 +224,152 @@ def chart(
emas: List of EMA periods to overlay (e.g. [10, 25]).
smas: List of SMA periods to overlay.
n_bars: Number of bars to display (last N).
interactive: Use plotly (True) or matplotlib (False).
figsize: Figure size (matplotlib only).
show: Display the chart (matplotlib only; plotly always shows).
save: Save path (.html for plotly, .png for matplotlib).
interactive: Kept for backward compatibility (plotly renders both
paths; ``save=".png"`` produces a static image via kaleido).
figsize: Figure size in inches, mapped to pixels.
show: Display the chart in the browser.
save: Save path (.html interactive, or .png/.svg via kaleido).
"""
if interactive:
return _chart_interactive(
result, store, symbol_id,
emas=emas, smas=smas, n_bars=n_bars, save=save,
)
return _chart_matplotlib(
result, store, symbol_id,
emas=emas, smas=smas, n_bars=n_bars,
figsize=figsize, show=show, save=save,
_ = interactive # single plotly path
import plotly.graph_objects as go
from plotly.subplots import make_subplots
bars, offset, bar_interval_s, filtered_trades = _prepare_chart_data(
result, store, symbol_id, n_bars,
)
close_full = bars["close"]
ts = bars["timestamp"][offset:]
o = bars["open"][offset:]
h = bars["high"][offset:]
l = bars["low"][offset:]
c = bars["close"][offset:]
vol = bars["volume"][offset:]
dates = ts.view("datetime64[ns]")
sym_name = _resolve_sym_name(store, symbol_id)
interval_label = _interval_label(bar_interval_s)
with theme_context():
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
row_heights=[0.8, 0.2],
)
# Candlesticks
fig.add_trace(
go.Candlestick(
x=dates, open=o, high=h, low=l, close=c,
increasing_line_color=GREEN, decreasing_line_color=RED,
increasing_fillcolor=GREEN, decreasing_fillcolor=RED,
name="OHLC",
),
row=1, col=1,
)
# Indicators
color_idx = 0
if emas:
for period in emas:
vals = _ema(close_full, period)[offset:]
color = INDICATOR_COLORS[color_idx % len(INDICATOR_COLORS)]
fig.add_trace(
go.Scatter(
x=dates, y=vals, mode="lines",
name=f"EMA({period})",
line=dict(color=color, width=1.5),
),
row=1, col=1,
)
color_idx += 1
if smas:
for period in smas:
vals = _sma(close_full, period)[offset:]
color = INDICATOR_COLORS[color_idx % len(INDICATOR_COLORS)]
fig.add_trace(
go.Scatter(
x=dates, y=vals, mode="lines",
name=f"SMA({period})",
line=dict(color=color, width=1.5, dash="dash"),
),
row=1, col=1,
)
color_idx += 1
# Trade markers
t_ts = filtered_trades["ts"]
t_side = filtered_trades["side"]
t_price = filtered_trades["price"]
t_qty = filtered_trades["qty"]
buy_mask = t_side == 1
sell_mask = t_side == 2
if buy_mask.any():
fig.add_trace(
go.Scatter(
x=t_ts[buy_mask], y=t_price[buy_mask],
mode="markers",
name="BUY",
marker=dict(
symbol="triangle-up", size=12,
color=GREEN, line=dict(color="white", width=1),
),
text=[f"BUY {q:.6f} @ {p:.2f}" for q, p in
zip(t_qty[buy_mask], t_price[buy_mask])],
hoverinfo="text+x",
),
row=1, col=1,
)
if sell_mask.any():
fig.add_trace(
go.Scatter(
x=t_ts[sell_mask], y=t_price[sell_mask],
mode="markers",
name="SELL",
marker=dict(
symbol="triangle-down", size=12,
color=RED, line=dict(color="white", width=1),
),
text=[f"SELL {q:.6f} @ {p:.2f}" for q, p in
zip(t_qty[sell_mask], t_price[sell_mask])],
hoverinfo="text+x",
),
row=1, col=1,
)
# Volume bars
vol_colors = [GREEN if c[i] >= o[i] else RED for i in range(len(c))]
fig.add_trace(
go.Bar(
x=dates, y=vol, name="Volume",
marker_color=vol_colors, opacity=0.5,
showlegend=False,
),
row=2, col=1,
)
fig.update_layout(
title=f"{sym_name} {interval_label}",
paper_bgcolor=BG_FIGURE,
plot_bgcolor=BG_AXES,
xaxis_rangeslider_visible=False,
hovermode="x unified",
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0),
width=int(figsize[0] * 80),
height=int(figsize[1] * 80),
margin=dict(l=60, r=20, t=60, b=40),
)
fig.update_yaxes(title_text="Price", row=1, col=1)
fig.update_yaxes(title_text="Vol", row=2, col=1)
return finalize(fig, show=show, save=save)
def _bar_interval_to_seconds(bi: dict) -> int:
"""Convert manifest bar_interval dict to seconds."""
python/manifoldbt/plot/research.py
+418 -400
View File
File diff suppressed because it is too large Load Diff
python/manifoldbt/plot/tearsheet.py
+63 -269
View File
@@ -1,35 +1,25 @@
"""Composite tearsheet — HTML strategy report."""
"""Composite tearsheet — HTML strategy report with interactive plotly charts."""
from __future__ import annotations
import base64
import io
import tempfile
import webbrowser
from html import escape
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.figure import Figure
from manifoldbt.plot._theme import (
BG_AXES,
BG_FIGURE,
DARK_GRAY,
GRAY,
GREEN,
RED,
WHITE,
theme_context,
)
from manifoldbt.plot._convert import equity_with_dates, positions_arrays
from manifoldbt.plot._utils import auto_title, format_pct
from manifoldbt.plot._convert import equity_with_dates
from manifoldbt.plot._utils import auto_title, chart_div, format_pct
from manifoldbt.plot.backtest import (
annual_returns,
drawdown,
equity,
monthly_returns,
returns_histogram,
rolling_sharpe,
@@ -38,44 +28,6 @@ from manifoldbt.plot.backtest import (
var_chart,
)
def _fig_to_base64(fig: Figure, dpi: int = 150) -> str:
"""Render a matplotlib figure to a base64-encoded PNG string."""
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=dpi, bbox_inches="tight",
facecolor=fig.get_facecolor(), edgecolor="none")
plt.close(fig)
buf.seek(0)
return base64.b64encode(buf.read()).decode("ascii")
def _render_chart(chart_fn, result, figsize=(12, 4), dpi=150, **kwargs) -> str:
"""Call a chart function on a fresh figure/axes and return base64 PNG."""
with theme_context():
fig, ax = plt.subplots(figsize=figsize)
chart_fn(result, ax=ax, **kwargs)
fig.tight_layout()
return _fig_to_base64(fig, dpi=dpi)
def _render_summary_b64(result, figsize=(12, 6), dpi=150) -> str:
"""Render the summary chart (equity+benchmark+trades+margin) to base64."""
with theme_context():
fig = summary(result, figsize=figsize)
return _fig_to_base64(fig, dpi=dpi)
def _render_exposure_b64(result, figsize=(12, 4), dpi=150) -> str:
"""Render the exposure chart to base64 PNG."""
with theme_context():
fig, ax = plt.subplots(figsize=figsize)
_render_exposure(ax, result)
_set_title(ax, "Capital Exposure")
_format_dates(ax)
fig.tight_layout()
return _fig_to_base64(fig, dpi=dpi)
_CSS = f"""
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{
@@ -127,12 +79,6 @@ body {{
flex-direction: column;
gap: 12px;
}}
.charts-stack img {{
width: 100%;
display: block;
border-radius: 4px;
border: 1px solid #1e1e24;
}}
.section-label {{
font-size: 10px;
font-weight: 700;
@@ -167,14 +113,11 @@ body {{
font-weight: 500;
white-space: nowrap;
}}
.chart-row {{
margin-bottom: 12px;
}}
.chart-row img {{
width: 100%;
display: block;
border-radius: 4px;
.chart-cell {{
background: {BG_FIGURE};
border: 1px solid #1e1e24;
border-radius: 4px;
overflow: hidden;
}}
.chart-grid {{
display: grid;
@@ -182,27 +125,15 @@ body {{
gap: 12px;
margin-bottom: 12px;
}}
.chart-grid img {{
width: 100%;
display: block;
border-radius: 4px;
border: 1px solid #1e1e24;
}}
.chart-grid-3 {{
display: grid;
grid-template-columns: 1fr 1fr 1fr;
gap: 12px;
margin-bottom: 12px;
}}
.chart-grid-3 img {{
width: 100%;
display: block;
border-radius: 4px;
border: 1px solid #1e1e24;
}}
.plotly-graph-div {{ width: 100% !important; }}
"""
def _div(fig, height: int) -> str:
"""Wrap a plotly figure div in a bordered cell."""
return f'<div class="chart-cell">{chart_div(fig, height=height)}</div>'
def tearsheet(
result,
*,
@@ -211,37 +142,40 @@ def tearsheet(
show: bool = False,
save: Optional[Union[str, Path]] = None,
dpi: int = 150,
plotlyjs: str = "cdn",
) -> str:
"""Strategy report — self-contained HTML page.
"""Strategy report — self-contained HTML page with interactive charts.
Returns the HTML string. Opens in browser when ``show=True``,
writes to disk when ``save`` is given.
Args:
plotlyjs: ``"cdn"`` (small file, needs network on open) or
``"inline"`` (fully offline report, ~4.4 MB heavier).
"""
_ = benchmark # reserved for future benchmark overlay support
_ = dpi # kept for backward compatibility (was the PNG export dpi)
strategy_name = title or auto_title(result, "Backtest")
metrics = result.metrics if hasattr(result, "metrics") else {}
ts = metrics.get("trade_stats", {})
dates, _ = equity_with_dates(result)
dates, _vals = equity_with_dates(result)
date_start = str(dates[0])[:10] if len(dates) > 0 else "?"
date_end = str(dates[-1])[:10] if len(dates) > 0 else "?"
# ── Generate charts as base64 PNGs ────────────────────────────
# Right column: summary chart (equity + benchmark + trades + margin)
chart_summary = _render_summary_b64(result, figsize=(12, 6), dpi=dpi)
chart_dd = _render_chart(drawdown, result, figsize=(12, 2.5), dpi=dpi)
# Left column chart
chart_annual = _render_chart(annual_returns, result, figsize=(5, 4), dpi=dpi)
# Full width grids (2 per row)
chart_monthly = _render_chart(monthly_returns, result, figsize=(8, 4), dpi=dpi)
chart_hist = _render_chart(returns_histogram, result, figsize=(8, 4), dpi=dpi)
chart_sharpe = _render_chart(rolling_sharpe, result, figsize=(8, 3.5), dpi=dpi)
chart_vol = _render_chart(rolling_volatility, result, figsize=(8, 3.5), dpi=dpi)
chart_var = _render_chart(var_chart, result, figsize=(8, 4), dpi=dpi)
# ── Generate interactive chart divs ────────────────────────────
with theme_context():
div_summary = _div(summary(result), height=520)
div_dd = _div(drawdown(result), height=210)
div_annual = _div(annual_returns(result), height=330)
div_monthly = _div(monthly_returns(result), height=340)
div_hist = _div(returns_histogram(result), height=340)
div_sharpe = _div(rolling_sharpe(result), height=300)
div_vol = _div(rolling_volatility(result), height=300)
div_var = _div(var_chart(result), height=340)
# ── Metrics ───────────────────────────────────────────────────
ret = metrics.get("total_return", 0)
_ = ret # used below in metrics_html
def _m(label, value, cls=""):
esc_v = escape(str(value))
@@ -278,6 +212,13 @@ def tearsheet(
+ _m("Fees", f"{ts.get('total_fees', 0):.2f}")
)
# ── plotly.js include ─────────────────────────────────────────
if plotlyjs == "inline":
import plotly.io as pio
plotly_js_tag = f"<script>{pio.get_plotlyjs()}</script>"
else:
plotly_js_tag = '<script src="https://cdn.plot.ly/plotly-2.35.2.min.js" charset="utf-8"></script>'
# ── Assemble HTML ─────────────────────────────────────────────
html = f"""<!DOCTYPE html>
<html lang="en">
@@ -286,6 +227,7 @@ def tearsheet(
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{escape(strategy_name)} — Tearsheet</title>
<style>{_CSS}</style>
{plotly_js_tag}
</head>
<body>
<div class="container">
@@ -298,26 +240,26 @@ def tearsheet(
<div class="main-grid">
<div>
<div class="metrics-panel" style="margin-bottom:12px;">{metrics_html}</div>
<img src="data:image/png;base64,{chart_annual}" alt="Annual Returns" style="width:100%; border-radius:4px; border:1px solid #1e1e24;">
{div_annual}
</div>
<div class="charts-stack">
<img src="data:image/png;base64,{chart_summary}" alt="Equity + Benchmark + Trades + Margin">
<img src="data:image/png;base64,{chart_dd}" alt="Drawdown">
{div_summary}
{div_dd}
</div>
</div>
<div class="chart-grid">
<img src="data:image/png;base64,{chart_monthly}" alt="Monthly Returns">
<img src="data:image/png;base64,{chart_hist}" alt="Returns Distribution">
{div_monthly}
{div_hist}
</div>
<div class="chart-grid">
<img src="data:image/png;base64,{chart_sharpe}" alt="Rolling Sharpe">
<img src="data:image/png;base64,{chart_vol}" alt="Rolling Volatility">
{div_sharpe}
{div_vol}
</div>
<div class="chart-grid">
<img src="data:image/png;base64,{chart_var}" alt="Value at Risk">
{div_var}
</div>
</div>
@@ -329,7 +271,6 @@ def tearsheet(
Path(save).write_text(html, encoding="utf-8")
if show:
# Write the report HTML
if save is not None:
report_path = Path(save).resolve()
else:
@@ -339,22 +280,7 @@ def tearsheet(
tmp.write(html)
tmp.close()
report_path = Path(tmp.name).resolve()
# Create a launcher HTML that opens the report in a 1600x850 window
report_uri = report_path.as_uri()
launcher_html = f"""<!DOCTYPE html><html><head><script>
var w = window.open("{report_uri}", "_blank",
"width=1600,height=850,menubar=no,toolbar=no,location=no,status=no");
if (!w) window.location = "{report_uri}";
else window.close();
</script></head><body></body></html>"""
launcher = tempfile.NamedTemporaryFile(
suffix=".html", delete=False, mode="w", encoding="utf-8"
)
launcher.write(launcher_html)
launcher.close()
webbrowser.open(Path(launcher.name).resolve().as_uri())
webbrowser.open(report_path.as_uri())
return html
@@ -369,40 +295,41 @@ def research_report(
show: bool = False,
save: Optional[Union[str, Path]] = None,
dpi: int = 150,
) -> List[Figure]:
"""Research report — one figure per analysis."""
) -> List[Any]:
"""Research report — one figure per analysis (plotly Figures)."""
from manifoldbt.plot.research import (
heatmap_2d,
stability,
walk_forward,
)
_ = title
figs = []
with theme_context():
if sweep_result is not None:
fig, ax = plt.subplots(figsize=figsize)
heatmap_2d(sweep_result, ax=ax)
figs.append(fig)
figs.append(heatmap_2d(sweep_result, figsize=figsize))
if wf_result is not None:
fig, ax = plt.subplots(figsize=figsize)
walk_forward(wf_result, ax=ax)
figs.append(fig)
figs.append(walk_forward(wf_result, figsize=figsize))
if stability_result is not None:
fig, ax = plt.subplots(figsize=figsize)
stability(stability_result, ax=ax)
figs.append(fig)
figs.append(stability(stability_result, figsize=figsize))
if not figs:
raise ValueError("At least one result (sweep, wf, or stability) required.")
if save is not None:
path = Path(save)
stem, suffix = path.stem, path.suffix or ".png"
stem, suffix = path.stem, path.suffix or ".html"
for i, f in enumerate(figs):
out = path.parent / f"{stem}_{i + 1}{suffix}"
f.savefig(str(out), dpi=dpi, bbox_inches="tight")
if suffix.lower() == ".html":
from manifoldbt.plot._utils import write_responsive_html
write_responsive_html(f, out)
else:
scale = max(1.0, dpi / 96.0)
f.write_image(str(out), scale=scale)
if show:
plt.show()
for f in figs:
f.show()
return figs
@@ -410,113 +337,6 @@ def research_report(
# ── Internal ─────────────────────────────────────────────────────────────────
def _set_title(ax, text):
"""Set title left-aligned, clearing any existing title from sub-functions."""
ax.set_title("", loc="center") # clear default
ax.set_title(text, fontsize=9, loc="left", color=GRAY)
def _format_dates(ax):
try:
ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
ax.xaxis.set_major_locator(mdates.AutoDateLocator())
for lbl in ax.get_xticklabels():
lbl.set_rotation(0)
lbl.set_ha("center")
except Exception:
pass
def _fix_rolling_xaxis(ax, result):
try:
dates, _ = equity_with_dates(result)
from manifoldbt.plot._convert import daily_returns_array
rets = daily_returns_array(result)
n_rets = len(rets)
aligned = dates[len(dates) - n_rets:] if len(dates) > n_rets else dates
for line in ax.get_lines():
xdata = line.get_xdata()
n = len(xdata)
if n <= 1:
continue
if isinstance(xdata[0], (int, float, np.integer, np.floating)):
x0, x1 = float(xdata[0]), float(xdata[1])
if abs(x1 - x0 - 1.0) < 0.01 and n <= len(aligned):
line.set_xdata(aligned[:n])
_format_dates(ax)
ax.relim()
ax.autoscale_view()
except Exception:
pass
def _render_metrics_table(ax, metrics, ts):
"""Render metrics as single-column dotted-leader list with sections."""
ax.set_xticks([])
ax.set_yticks([])
ax.set_facecolor(BG_AXES)
for spine in ax.spines.values():
spine.set_visible(True)
spine.set_color(DARK_GRAY)
spine.set_linewidth(0.5)
ret = metrics.get("total_return", 0)
ret_color = GREEN if ret > 0 else RED if ret < 0 else GRAY
W = 30 # total width for dotted leader alignment
def _line(label, value):
dots = "·" * max(1, W - len(label) - len(str(value)))
return f"{label} {dots} {value}"
# Build sections
sections = [
("RETURNS", GRAY, [
(_line("Total Return", format_pct(ret)), ret_color),
(_line("CAGR", format_pct(metrics.get("cagr", 0))), GRAY),
(_line("Max Drawdown", format_pct(metrics.get("max_drawdown", 0))), RED),
(_line("Volatility", format_pct(metrics.get("volatility", 0))), GRAY),
(_line("Best Day", format_pct(metrics.get("best_day", 0))), GRAY),
(_line("Worst Day", format_pct(metrics.get("worst_day", 0))), GRAY),
]),
("RATIOS", GRAY, [
(_line("Sharpe", f"{metrics.get('sharpe', 0):.2f}"), GRAY),
(_line("Sortino", f"{metrics.get('sortino', 0):.2f}"), GRAY),
(_line("Calmar", f"{metrics.get('calmar', 0):.2f}"), GRAY),
]),
("TRADING", GRAY, [
(_line("Trades", f"{ts.get('total_trades', metrics.get('total_trades', 0))}"), GRAY),
(_line("Win Rate", f"{ts.get('win_rate', metrics.get('win_rate', 0)):.1%}"), GRAY),
(_line("Profit Factor", f"{ts.get('profit_factor', metrics.get('profit_factor', 0)):.2f}"), GRAY),
(_line("Round Trips", f"{ts.get('round_trips', 0)}"), GRAY),
(_line("Avg Hold", _fmt_hold_time(ts.get("avg_holding_seconds", 0))), GRAY),
(_line("Fees", f"{ts.get('total_fees', 0):.2f}"), GRAY),
]),
]
# Count total lines for spacing
total = sum(1 + len(items) + 1 for _, _, items in sections) # header + items + gap
y = 0.97
dy = 0.92 / total
for section_name, section_color, items in sections:
# Section header
ax.text(0.06, y, section_name, fontsize=7, fontweight="bold",
color=DARK_GRAY, transform=ax.transAxes, va="top",
family="monospace")
y -= dy * 1.2
# Items
for text, color in items:
ax.text(0.06, y, text, fontsize=9, color=color,
transform=ax.transAxes, va="top", family="monospace")
y -= dy
# Gap between sections
y -= dy * 0.5
def _fmt_hold_time(seconds):
"""Format holding time in human-readable units."""
if seconds <= 0:
@@ -530,29 +350,3 @@ def _fmt_hold_time(seconds):
return f"{days:.0f}d"
hours = seconds / 3600
return f"{hours:.0f}h"
def _render_exposure(ax, result):
try:
pa = positions_arrays(result)
pos_ts = pa["timestamp"]
pos_cap = pa["capital"]
pos_eq = pa["equity"]
unique_ts, first_idx = np.unique(pos_ts, return_index=True)
first_idx.sort()
cap = pos_cap[first_idx]
eq_arr = pos_eq[first_idx]
used = np.where(eq_arr > 0, (1.0 - cap / eq_arr) * 100, 0.0)
used = np.clip(used, 0, None)
used_dates = unique_ts.astype("datetime64[ns]")
ax.fill_between(used_dates, 0, used,
color=GREEN, alpha=0.10, edgecolor="none")
ax.plot(used_dates, used, color=GREEN, linewidth=0.7, alpha=0.8)
ax.axhline(0, color=DARK_GRAY, linewidth=0.4)
ax.set_ylabel("Exposure %", fontsize=8)
except Exception:
ax.text(0.5, 0.5, "No position data",
transform=ax.transAxes, ha="center", va="center",
color=DARK_GRAY, fontsize=9)
python/manifoldbt/portfolio.py
+9
View File
@@ -37,6 +37,15 @@ class Portfolio:
strategy: A Strategy instance.
weight: Fraction of total capital (0.0 to 1.0).
"""
if getattr(strategy, "_orders", None):
import warnings
warnings.warn(
f"Strategy '{strategy.name}' defines stop_loss/take_profit/"
"trailing_stop orders, but portfolio mode does not support "
"per-strategy orders yet: they are IGNORED in run_portfolio().",
UserWarning,
stacklevel=2,
)
self._strategies.append({
"name": strategy.name,
"strategy_json": strategy.to_json(),
python/manifoldbt/strategy.py
+7 -1
View File
@@ -194,7 +194,7 @@ class Strategy:
spec["range"] = None
params[param_name] = spec
return {
out = {
"name": self.name,
"signals": {
name: expr.to_json() for name, expr in self.signals.items()
@@ -206,6 +206,12 @@ class Strategy:
"description": self._description,
},
}
# Per-strategy SL/TP/trailing orders travel with the strategy so the
# engine applies them per-strategy in a single batch/sweep call (the
# Rust StrategyDef.orders field; omitted when unset for a clean JSON).
if self._orders:
out["orders"] = self._orders
return out
def to_json(self) -> str:
"""Serialize to a JSON string matching Rust ``StrategyDef``.
python/manifoldbt/sweep.py
+42 -42
View File
@@ -93,62 +93,62 @@ class SweepResult:
return best_result
def plot_metric(self, metric: str = "sharpe", **kwargs: Any) -> Any:
"""Plot a metric across sweep results.
"""Plot a metric across sweep results (plotly).
For 2-parameter sweeps, delegates to ``bt.plot.heatmap_2d``.
For 1-parameter sweeps, produces a bar chart.
Args:
metric: Metric to visualize.
**kwargs: Forwarded to the plot function.
**kwargs: ``figsize``, ``show``, ``save`` forwarded to the plot.
"""
import matplotlib.pyplot as plt
import numpy as np
from manifoldbt.plot._theme import ACCENT, theme_context
from manifoldbt.plot._utils import finalize, new_figure
df = self.to_df(backend="pandas")
param_cols = [c for c in df.columns if c.startswith("param_")]
show = kwargs.pop("show", True)
save = kwargs.pop("save", None)
if len(param_cols) == 2:
# 2D heatmap
from manifoldbt.plot.research import heatmap_2d
x_col, y_col = param_cols[0], param_cols[1]
pivot = df.pivot_table(index=y_col, columns=x_col, values=metric)
fig, ax = plt.subplots(figsize=kwargs.pop("figsize", (10, 6)))
im = ax.imshow(pivot.values, aspect="auto", cmap=kwargs.pop("cmap", "RdYlGn"))
ax.set_xticks(range(len(pivot.columns)))
ax.set_xticklabels(pivot.columns, rotation=45)
ax.set_yticks(range(len(pivot.index)))
ax.set_yticklabels(pivot.index)
ax.set_xlabel(x_col.replace("param_", ""))
ax.set_ylabel(y_col.replace("param_", ""))
ax.set_title(f"{metric} heatmap")
plt.colorbar(im, ax=ax, label=metric)
plt.tight_layout()
if kwargs.get("show", True):
plt.show()
return fig
elif len(param_cols) == 1:
# 1D bar chart
p_col = param_cols[0]
fig, ax = plt.subplots(figsize=kwargs.pop("figsize", (10, 5)))
ax.bar(range(len(df)), df[metric].values, tick_label=[str(v) for v in df[p_col].values])
ax.set_xlabel(p_col.replace("param_", ""))
ax.set_ylabel(metric)
ax.set_title(f"{metric} by {p_col.replace('param_', '')}")
plt.tight_layout()
if kwargs.get("show", True):
plt.show()
return fig
else:
# Fallback: simple bar
fig, ax = plt.subplots(figsize=kwargs.pop("figsize", (10, 5)))
ax.bar(range(len(df)), df[metric].values)
ax.set_xlabel("run")
ax.set_ylabel(metric)
ax.set_title(f"{metric} across sweep")
plt.tight_layout()
if kwargs.get("show", True):
plt.show()
return fig
sweep_result = {
"metric_grid": pivot.values.tolist(),
"x_values": list(pivot.columns),
"y_values": list(pivot.index),
"x_param": x_col.replace("param_", ""),
"y_param": y_col.replace("param_", ""),
"metric": metric,
}
return heatmap_2d(sweep_result, show=show, save=save, **kwargs)
import plotly.graph_objects as go
with theme_context():
if len(param_cols) == 1:
p_col = param_cols[0]
fig = new_figure(kwargs.pop("figsize", (10, 5)),
f"{metric} by {p_col.replace('param_', '')}")
fig.add_trace(go.Bar(
x=[str(v) for v in df[p_col].values], y=df[metric].values,
marker_color=ACCENT, marker_line_width=0,
))
fig.update_xaxes(title_text=p_col.replace("param_", ""),
type="category", showspikes=False)
else:
fig = new_figure(kwargs.pop("figsize", (10, 5)),
f"{metric} across sweep")
fig.add_trace(go.Bar(
x=list(range(len(df))), y=df[metric].values,
marker_color=ACCENT, marker_line_width=0,
))
fig.update_xaxes(title_text="run", showspikes=False)
fig.update_yaxes(title_text=metric)
fig.update_layout(hovermode="closest")
return finalize(fig, show=show, save=save)
def __repr__(self) -> str:
params = ", ".join(f"{k}={len(v)} vals" for k, v in self._param_grid.items())
python/tests/test_batch_orders.py
+137
View File
@@ -0,0 +1,137 @@
"""Regression tests for per-strategy orders in batch runs.
History: run_batch/run_batch_lite once dropped SL/TP entirely (they called
_prepare_config(config, None)). They were then fixed by merging orders into a
grouped config. Now orders travel INSIDE the strategy JSON (StrategyDef.orders)
and the engine applies them per-strategy, so a single native call handles a
batch of strategies with DIFFERENT brackets over one data load — the config
carries no orders and there is no per-profile grouping.
Native calls are monkeypatched, so no market data is needed.
"""
import json
import pytest
import manifoldbt as bt
class _DummyStore:
"""Minimal store: no metadata DB, default dataset (all lookups fall back)."""
def dataset(self):
raise RuntimeError("no dataset")
def metadata_db(self):
raise RuntimeError("no metadata db")
def _strategy(name, sl=None, tp=None):
s = bt.Strategy.create(name).signal("sig", bt.lit(1.0)).size(bt.lit(0.1))
if sl is not None:
s = s.stop_loss(pct=sl)
if tp is not None:
s = s.take_profit(pct=tp)
return s
def _config():
return bt.BacktestConfig(
universe=[1],
time_range_start=0,
time_range_end=10_000_000_000,
initial_capital=10_000,
)
@pytest.fixture()
def captured(monkeypatch):
"""Patch both native batch entry points; record (config_dict, [strategy_dict])."""
calls = []
def fake_batch_lite(strategy_jsons, config_json, store, max_parallelism=0):
strats = [json.loads(s) for s in strategy_jsons]
calls.append((json.loads(config_json), strats))
return [f"lite:{s['name']}" for s in strats]
def fake_batch(strategy_jsons, config_json, store, max_parallelism=0):
strats = [json.loads(s) for s in strategy_jsons]
calls.append((json.loads(config_json), strats))
return [object() for _ in strats]
monkeypatch.setattr(bt, "_run_batch_lite_native", fake_batch_lite)
monkeypatch.setattr(bt, "_run_batch_native", fake_batch)
return calls
def _config_orders(cfg_json):
return (cfg_json.get("execution") or {}).get("orders")
def _names(strats):
return [s["name"] for s in strats]
def _sl_of(strat):
orders = strat.get("orders")
return orders["stop_loss"]["stop_pct"] if orders and "stop_loss" in orders else None
def test_batch_lite_carries_sl_tp_in_strategy_json(captured):
strats = [_strategy(f"s{i}", sl=2.0, tp=4.0) for i in range(3)]
out = bt.run_batch_lite(strats, _config(), _DummyStore())
assert len(captured) == 1, "one native call handles the whole batch"
cfg, sent = captured[0]
assert _config_orders(cfg) is None, "orders travel in the strategy JSON, not the config"
for s in sent:
assert s["orders"]["stop_loss"]["stop_pct"] == 2.0
assert s["orders"]["take_profit"]["profit_pct"] == 4.0
assert _names(sent) == ["s0", "s1", "s2"]
assert out == ["lite:s0", "lite:s1", "lite:s2"]
def test_batch_lite_no_orders_absent_from_json(captured):
strats = [_strategy(f"s{i}") for i in range(2)]
bt.run_batch_lite(strats, _config(), _DummyStore())
assert len(captured) == 1
cfg, sent = captured[0]
assert _config_orders(cfg) is None
for s in sent:
assert s.get("orders") is None
def test_batch_lite_mixed_orders_single_call_in_order(captured):
strats = [
_strategy("a", sl=2.0),
_strategy("b"), # no orders
_strategy("c", sl=2.0),
_strategy("d", sl=5.0),
]
out = bt.run_batch_lite(strats, _config(), _DummyStore())
# Heterogeneous brackets now run in ONE native call over one data load,
# each strategy carrying its own orders — no grouping, no reordering.
assert len(captured) == 1
cfg, sent = captured[0]
assert _config_orders(cfg) is None
assert _names(sent) == ["a", "b", "c", "d"]
assert [_sl_of(s) for s in sent] == [2.0, None, 2.0, 5.0]
assert out == ["lite:a", "lite:b", "lite:c", "lite:d"]
def test_run_batch_carries_sl_tp(captured):
strats = [_strategy("x", sl=1.5), _strategy("y", sl=1.5)]
bt.run_batch(strats, _config(), _DummyStore())
assert len(captured) == 1
cfg, sent = captured[0]
assert _config_orders(cfg) is None
assert all(_sl_of(s) == 1.5 for s in sent)
assert _names(sent) == ["x", "y"]
def test_portfolio_warns_on_ignored_orders():
with pytest.warns(UserWarning, match="IGNORED"):
bt.Portfolio().strategy(_strategy("p", sl=2.0), weight=1.0)