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.15.0

Browse Source
This commit is contained in:
github-actions[bot]
2026-08-16 12:02:58 +00:00
parent 5f0321b189
commit f02d462480
9 changed files with 1096 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/strategy-authoring.md
+87 -7
View File
@@ -15,13 +15,14 @@ This guide describes how to define trading strategies using the manifoldbt Pytho
5. [Backtest Configuration](#backtest-configuration)
6. [Execution Model](#execution-model)
7. [Fee & Slippage Models](#fee--slippage-models)
8. [Orders (SL/TP/Trailing)](#orders-sltp-trailing)
9. [Cross-Asset References](#cross-asset-references)
10. [Dataset Auto-Resolution](#dataset-auto-resolution)
11. [Diagnostics](#diagnostics)
12. [Profiling](#profiling)
13. [Complete Examples](#complete-examples)
14. [Indicator Reference](#indicator-reference)
8. [Orders (SL/TP/Trailing)](#orders-sltptrailing)
9. [Entry Orders](#entry-orders)
10. [Cross-Asset References](#cross-asset-references)
11. [Dataset Auto-Resolution](#dataset-auto-resolution)
12. [Diagnostics](#diagnostics)
13. [Profiling](#profiling)
14. [Complete Examples](#complete-examples)
15. [Indicator Reference](#indicator-reference)
---
@@ -195,6 +196,11 @@ best = sweep.best("sharpe")
batch = mbt.run_sweep_lite(strategy, {"fast": range(5, 100), "slow": range(10, 500)}, config, store)
```
Grids this size need Pro. Community is capped at 256 backtests cumulatively per
Python session across all sweep/batch calls, and each sweep call waits 5 s
before starting; single `bt.run()` calls are never gated. See
`docs/sweep-combo-limit-plan.md`.
`run_sweep_lite` is optimized for large parameter grids (100k+ combos):
- Cartesian product expansion in Rust (no Python loop)
- Shared indicator cache (EMA(12) computed once, reused across combos)
@@ -312,6 +318,80 @@ strategy = (
---
## Entry Orders
By default an entry takes a market fill on the execution bar (see
[Execution Model](#execution-model)). Four order types let the entry rest at a
price instead:
| Builder method | Fills when | Fill price | Costs |
|---|---|---|---|
| `.limit_entry(...)` | price comes **to** the level | the level exactly | maker, no slippage |
| `.stop_entry(...)` | price breaks **through** the level | the level, or the open if the bar gapped through it | taker + slippage |
| `.market_if_touched(...)` | price comes **to** the level | the level | taker + slippage |
| `.stop_limit_entry(...)` | breaks through `stop`, then rests at `limit` | the limit | maker, no slippage |
### Where the level comes from
Every method takes exactly one of three price forms:
```python
.limit_entry(offset_bps=25) # 25 bps below the signal close (above, for a sell)
.limit_entry(price=60_000) # a fixed level
.limit_entry(signal="entry_px") # a level this strategy computes
```
`signal=` is the general form: name any signal the strategy defines and the
order rests on that series, read on the signal bar.
```python
from manifoldbt.indicators import atr, close, ema
trend = ema(close, 50)
entry_px = close - atr(14) # rest one ATR below the close
strategy = (
mbt.Strategy.create("pullback_entry")
.signal("trend", trend)
.signal("entry_px", entry_px) # named so the order can reference it
.size(mbt.when(close > trend, 1.0, 0.0))
.limit_entry(signal="entry_px", time_in_force={"GTB": 5})
.stop_loss(pct=3.0)
)
```
### Time in force
`"GTC"` (default, rests until filled or the signal changes), `{"GTB": n}`
(cancel after n bars), `"IOC"` (fill on the arrival bar or cancel).
### Two things to watch
**A resting entry can simply never fill.** A strategy whose entries never
trigger produces a flat equity curve with no drawdown, which reads as a clean
backtest. The engine counts unfilled entries and reports them:
```python
result = mbt.run_backtest(strategy, config)
for w in result.warnings:
print(w) # "N entry order(s) expired unfilled and M were still resting ..."
```
**Sizing uses the close, not the level.** In `FractionOfEquity` mode a target of
`1.0` is converted to units at the signal-bar close, so an entry resting 2% away
buys ~2% too much notional. `size_at_fill_price=True` sizes off the order's own
level instead. It is off by default because turning it on changes the results of
strategies written against the old behaviour.
### Cost
A conditional entry runs on the general simulation loop rather than the fast
kernel, so parameter sweeps over one are slower than sweeps over a market entry
and cannot use the GPU. `run_sweep` reports which setting took you off the fast
path.
---
## Cross-Asset References
Use `mbt.symbol_ref()` to reference another symbol's data in multi-asset strategies:
examples/20_entry_orders.py
+98
View File
@@ -0,0 +1,98 @@
"""Entry orders — resting an entry at a price instead of taking the close.
By default an entry takes a market fill on the execution bar. This example runs
the same signal four ways so the difference is visible in one place:
market fill at the execution bar's close
limit wait for a pullback, fill passively (maker, no slippage)
stop wait for a breakout, fill through the level (taker + gap)
limit on a signal rest on a level the DSL computes (here: 1 ATR below close)
Usage:
python examples/20_entry_orders.py
"""
import os
from time import perf_counter
import manifoldbt as mbt
from manifoldbt.indicators import atr, close, ema
from manifoldbt.helpers import Interval, Slippage, time_range
# -- Signal -------------------------------------------------------------------
fast = ema(close, 12)
slow = ema(close, 50)
trend = mbt.when(fast > slow, 1.0, 0.0)
# The level a signal-priced entry rests on: one ATR below the close.
pullback = close - atr(14)
def build(name: str, entry) -> "mbt.Strategy":
"""The same strategy every time; only the entry order changes."""
s = (
mbt.Strategy.create(name)
.signal("fast", fast)
.signal("slow", slow)
.signal("pullback", pullback)
.size(trend)
.stop_loss(pct=3.0)
)
return entry(s) if entry else s
VARIANTS = {
# Market: no entry order at all. The fast kernel stays available.
"market": None,
# Passive: 25 bps below the signal close, cancelled if unfilled after 5 bars.
"limit -25bps": lambda s: s.limit_entry(offset_bps=25, time_in_force={"GTB": 5}),
# Breakout: 25 bps above. Crosses the book, and a gap through it fills at the open.
"stop +25bps": lambda s: s.stop_entry(offset_bps=-25, time_in_force={"GTB": 5}),
# Signal-priced: rest on whatever the DSL computed, here close - atr(14).
"limit @ close-ATR": lambda s: s.limit_entry(signal="pullback", time_in_force={"GTB": 5}),
}
# -- Config -------------------------------------------------------------------
start, end = time_range("2022-01-01", "2025-01-01")
config = mbt.BacktestConfig(
universe={"binance": ["BTC-USDT:perp"]},
time_range_start=start,
time_range_end=end,
bar_interval=Interval.hours(4),
initial_capital=10_000,
fees=mbt.FeeConfig.binance_perps(),
slippage=Slippage.fixed_bps(2),
warmup_bars=60,
)
# -- Run ----------------------------------------------------------------------
if __name__ == "__main__":
root = os.path.join(os.path.dirname(__file__), "..")
data_root = os.path.abspath(os.path.join(root, "data"))
store = mbt.DataStore(
data_root=data_root,
metadata_db=os.path.abspath(os.path.join(root, "metadata", "metadata.sqlite")),
arrow_dir=os.path.join(data_root, "mega"),
)
print(f"{'entry':<20} {'trades':>7} {'return':>9} {'sharpe':>8} {'elapsed':>9}")
print("-" * 56)
for label, entry in VARIANTS.items():
strategy = build(label.replace(" ", "_"), entry)
t0 = perf_counter()
result = mbt.run(strategy, config, store)
elapsed = perf_counter() - t0
m = result.metrics
print(
f"{label:<20} {result.trades.num_rows:>7} "
f"{m['total_return']:>8.1%} {m['sharpe']:>8.2f} {elapsed:>8.2f}s"
)
# A resting entry can simply never fill. That failure mode looks like a
# clean backtest, so the engine reports it rather than staying silent.
for w in result.warnings:
if "unfilled" in w:
print(f"{'':<20} ! {w}")
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "manifoldbt"
version = "0.14.1"
version = "0.15.0"
description = "Rust-powered backtesting engine for quantitative research"
requires-python = ">=3.9"
license = { file = "LICENSE" }
python/manifoldbt/__init__.py
+12 -6
View File
@@ -38,6 +38,7 @@ from manifoldbt.config import (
FeeConfig,
OrderConfig,
VenueFees,
entry_price,
resolve_universe,
)
from manifoldbt.exceptions import (
@@ -157,9 +158,10 @@ def _require_pro_for_gpu(device, feature: str) -> None:
# Community fan-out budget: sweeps and batches may run up to this many backtests
# per call for free; beyond it requires Pro. Single run() is never affected.
# Keep in sync with the native bt_license::COMMUNITY_MAX_SWEEP_COMBOS.
_COMMUNITY_MAX_COMBOS = 500
# cumulatively per process for free; beyond it requires Pro. Single run() is
# never affected. Keep in sync with the native
# bt_license::COMMUNITY_MAX_SWEEP_COMBOS.
_COMMUNITY_MAX_COMBOS = 256
def _grid_combos(param_grid) -> int:
@@ -173,14 +175,17 @@ def _grid_combos(param_grid) -> int:
def _require_pro_over_combos(n_combos: int, what: str) -> None:
"""Raise LicenseError if a fan-out exceeds the Community combination limit.
No-op at or below the limit, or for Pro users. Mirrors the native
``require_combo_limit`` so Community and Pro see identical behaviour.
Fast-fail UX layer only: catches a single call that could never fit the
budget. The authoritative gate is the native ``require_combo_limit``,
which enforces the limit **cumulatively per session** — small calls also
consume budget there, and this mirror cannot (and must not) track that.
"""
if n_combos <= _COMMUNITY_MAX_COMBOS or _is_pro():
return
raise LicenseError(
f"{what} with {n_combos} runs exceeds the Community limit of "
f"{_COMMUNITY_MAX_COMBOS}. Upgrade to Pro at www.manifoldbt.com"
f"{_COMMUNITY_MAX_COMBOS} combinations per session. "
f"Upgrade to Pro at www.manifoldbt.com"
)
@@ -1603,6 +1608,7 @@ __all__ = [
"FeeConfig",
"VenueFees",
"OrderConfig",
"entry_price",
# Helpers
"date_to_ns",
"time_range",
python/manifoldbt/config.py
+83 -5
View File
@@ -6,19 +6,51 @@ from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union
def entry_price(
*,
offset_bps: Optional[float] = None,
price: Optional[float] = None,
signal: Optional[str] = None,
) -> dict:
"""Build the price spec for an entry order. Pass exactly one of:
- ``offset_bps``: distance from the signal-bar close, in bps. Positive is
more passive (buy lower / sell higher).
- ``price``: a fixed level, the same on every bar.
- ``signal``: the name of a strategy signal to read the level from, so the
order can rest on ``ema(close, 20)``, ``close - 2 * atr(close, 14)``, a
prior swing low, or anything else the DSL can express.
"""
given = [x for x in (offset_bps, price, signal) if x is not None]
if len(given) != 1:
raise ValueError("entry_price takes exactly one of offset_bps, price, signal")
if offset_bps is not None:
return {"OffsetBps": offset_bps}
if price is not None:
return {"Absolute": price}
return {"Signal": signal}
@dataclass
class OrderConfig:
"""Order management configuration for limit entries, stop-loss, take-profit,
and trailing stops. All fields are optional — when nothing is set the engine
uses the legacy market-order path with zero overhead.
"""Order management configuration for conditional entries, stop-loss,
take-profit, and trailing stops. All fields are optional — when nothing is
set the engine uses the legacy market-order path with zero overhead.
Sub-config dicts:
limit_entry: {"offset_bps": 10.0, "time_in_force": "GTC"}
offset_bps: distance from close in bps (buy: close*(1-offset/10000))
limit_entry: where an entry rests instead of taking a market fill.
price: {"OffsetBps": 10.0} | {"Absolute": 60000.0} | {"Signal": "entry_px"}
(omit and set offset_bps for the legacy shape)
trigger: "Limit" (default), "Stop", "StopLimit", "MarketIfTouched"
limit_price: same shape as price; required by "StopLimit"
time_in_force: "GTC" (default), {"GTB": 5}, or "IOC"
size_at_fill_price: size off the order's own level instead of the close
stop_loss: {"stop_pct": 2.0} — % from entry price
take_profit: {"profit_pct": 5.0} — % from entry price
trailing_stop: {"trail_pct": 3.0, "use_high": true}
Note that a conditional entry runs on the general simulation loop, not the
fast kernel, so sweeps over one are slower than sweeps over a market entry.
"""
limit_entry: Optional[dict] = None
@@ -44,6 +76,52 @@ class OrderConfig:
"""Convenience: trailing stop only."""
return cls(trailing_stop={"trail_pct": trail_pct, "use_high": use_high})
@classmethod
def limit_entry_at(
cls,
*,
offset_bps: Optional[float] = None,
price: Optional[float] = None,
signal: Optional[str] = None,
time_in_force: Union[str, dict] = "GTC",
size_at_fill_price: bool = False,
) -> "OrderConfig":
"""Convenience: a passive limit entry resting at the given level."""
return cls(
limit_entry={
"price": entry_price(
offset_bps=offset_bps, price=price, signal=signal
),
"trigger": "Limit",
"time_in_force": time_in_force,
"size_at_fill_price": size_at_fill_price,
}
)
@classmethod
def stop_entry_at(
cls,
*,
offset_bps: Optional[float] = None,
price: Optional[float] = None,
signal: Optional[str] = None,
time_in_force: Union[str, dict] = "GTC",
size_at_fill_price: bool = False,
) -> "OrderConfig":
"""Convenience: a breakout entry that fills once price trades through
the level. Crosses the book, so it pays taker fees and slippage, and a
bar that gaps through the level fills at the open."""
return cls(
limit_entry={
"price": entry_price(
offset_bps=offset_bps, price=price, signal=signal
),
"trigger": "Stop",
"time_in_force": time_in_force,
"size_at_fill_price": size_at_fill_price,
}
)
def to_json_dict(self) -> dict:
d: dict = {}
if self.limit_entry is not None:
python/manifoldbt/strategy.py
+125
View File
@@ -158,6 +158,131 @@ class Strategy:
self._json_cache = None
return self
def _entry(
self,
trigger: str,
offset_bps: Optional[float],
price: Optional[float],
signal: Optional[str],
time_in_force: Any,
size_at_fill_price: bool,
limit_price: Optional[Dict[str, Any]] = None,
) -> "Strategy":
from .config import entry_price
if self._orders is None:
self._orders = {}
entry: Dict[str, Any] = {
"price": entry_price(offset_bps=offset_bps, price=price, signal=signal),
"trigger": trigger,
"time_in_force": time_in_force,
"size_at_fill_price": size_at_fill_price,
}
if limit_price is not None:
entry["limit_price"] = limit_price
self._orders["limit_entry"] = entry
self._json_cache = None
return self
def limit_entry(
self,
*,
offset_bps: Optional[float] = None,
price: Optional[float] = None,
signal: Optional[str] = None,
time_in_force: Any = "GTC",
size_at_fill_price: bool = False,
) -> "Strategy":
"""Rest the entry passively at a level instead of taking a market fill.
Pass exactly one of ``offset_bps`` (distance from the signal-bar close),
``price`` (a fixed level), or ``signal`` (the name of a signal this
strategy defines, so the level can be any series the DSL computes).
A passive fill pays maker fees, takes no slippage, and lands on the
level exactly. It can also never fill: check ``result.warnings``.
Args:
offset_bps: Distance from the signal close in bps (positive = more passive).
price: A fixed price level.
signal: Name of a signal to read the level from.
time_in_force: ``"GTC"`` (default), ``{"GTB": 5}``, or ``"IOC"``.
size_at_fill_price: Size off the order's level instead of the close.
"""
return self._entry(
"Limit", offset_bps, price, signal, time_in_force, size_at_fill_price
)
def stop_entry(
self,
*,
offset_bps: Optional[float] = None,
price: Optional[float] = None,
signal: Optional[str] = None,
time_in_force: Any = "GTC",
size_at_fill_price: bool = False,
) -> "Strategy":
"""Enter on a breakout: fill once price trades **through** the level.
The mirror of :meth:`limit_entry`. It crosses the book, so it pays taker
fees and slippage, and a bar that gaps through the level fills at the
open rather than at the level.
"""
return self._entry(
"Stop", offset_bps, price, signal, time_in_force, size_at_fill_price
)
def market_if_touched(
self,
*,
offset_bps: Optional[float] = None,
price: Optional[float] = None,
signal: Optional[str] = None,
time_in_force: Any = "GTC",
size_at_fill_price: bool = False,
) -> "Strategy":
"""Wait for price to come to the level, then take a market fill.
Same trigger as :meth:`limit_entry`, but the fill crosses the book:
taker fees and slippage apply.
"""
return self._entry(
"MarketIfTouched",
offset_bps,
price,
signal,
time_in_force,
size_at_fill_price,
)
def stop_limit_entry(
self,
*,
stop: Optional[float] = None,
stop_signal: Optional[str] = None,
limit: Optional[float] = None,
limit_signal: Optional[str] = None,
time_in_force: Any = "GTC",
size_at_fill_price: bool = False,
) -> "Strategy":
"""Breakout that arms a resting limit.
The ``stop`` level arms the order; it then rests at ``limit`` and fills
there with maker fees. Pass each level either as a number or as the name
of a signal.
"""
from .config import entry_price
return self._entry(
"StopLimit",
None,
stop,
stop_signal,
time_in_force,
size_at_fill_price,
limit_price=entry_price(price=limit, signal=limit_signal),
)
def describe(self, text: str) -> "Strategy":
"""Set strategy description (returns self for chaining)."""
self._description = text
python/tests/test_combo_limit.py
+243
View File
@@ -0,0 +1,243 @@
"""Community sweep gating: cumulative combo budget + throughput penalty.
See docs/sweep-combo-limit-plan.md. Two mechanisms, tested here through real
sweep calls on a tiny dataset:
* the 500-combo cap is enforced on the running total **per process**, not per
call — otherwise slicing a grid into small calls bypasses it at a measured
+0.5% cost;
* every accepted Community call waits `SWEEP_MIN_INTERVAL`, held under a
machine-wide file lock, so the remaining bypass (a fresh interpreter per
slice) costs 5 s each and cannot be parallelised away.
The rate-gate tests run in subprocesses on purpose: "serialised across
processes" is only observable between processes, and an in-process test would
also depend on which test happened to run first.
"""
import subprocess
import sys
import textwrap
import time
import numpy as np
import pandas as pd
import pytest
import manifoldbt as bt
from manifoldbt._native import _combo_budget
IS_PRO = bt.license_info()[0] == "Pro"
community_only = pytest.mark.skipif(
IS_PRO, reason="Community-only; deactivate Pro/BT_UNLOCKED to test"
)
pro_only = pytest.mark.skipif(not IS_PRO, reason="requires an active Pro license")
@pytest.fixture(scope="module")
def store_paths(tmp_path_factory):
"""A minimal store on disk; returns (data_root, metadata_db, arrow_dir)."""
root = tmp_path_factory.mktemp("combo_limit")
idx = pd.date_range("2024-01-01", periods=120, freq="1min", tz="UTC")
close = 100.0 + np.arange(120, dtype=float)
df = pd.DataFrame({
"timestamp": idx,
"open": close, "high": close * 1.01, "low": close * 0.99,
"close": close, "volume": np.full(120, 1_000.0),
})
data_root, metadata_db = str(root / "data"), str(root / "metadata.sqlite")
bt.import_dataframe(
df, symbol="CL", symbol_id=1, interval="1m",
data_root=data_root, metadata_db=metadata_db,
)
return data_root, metadata_db, f"{data_root}/mega"
@pytest.fixture(scope="module")
def daily_store(store_paths):
data_root, metadata_db, arrow_dir = store_paths
return bt.DataStore(data_root, metadata_db, "bars_1m", None, arrow_dir)
# --- shared snippet: build strategy + config, run one sweep of n combos ------
_HARNESS = '''
import sys, time
import manifoldbt as bt
store = bt.DataStore({data_root!r}, {metadata_db!r}, "bars_1m", None, {arrow_dir!r})
strat = bt.Strategy(
name="budget_probe",
signals={{"signal": bt.lit(1.0)}},
position_sizing=bt.lit(1.0) * bt.param("size", default=1.0),
parameters={{"size": bt.param("size", default=1.0)}},
)
t0, t1 = bt.time_range("2024-01-01", "2024-01-02")
cfg = bt.BacktestConfig(universe=[1], time_range_start=t0, time_range_end=t1,
bar_interval={{"Minutes": 1}})
def sweep(n):
grid = {{"size": [1.0 + 0.001 * i for i in range(n)]}}
return bt.run_sweep_lite(strat, grid, cfg, store)
def timed(n):
# PermissionError comes from the native gate (cumulative cap), LicenseError
# from the Python mirror (a single call larger than the cap). Both are
# "refused", and neither should have waited out the rate limit.
t = time.perf_counter()
try:
sweep(n)
ok = True
except (PermissionError, bt.LicenseError):
ok = False
return time.perf_counter() - t, ok
'''
def _run(store_paths, body):
"""Run `body` in a fresh interpreter; return its stdout floats/flags."""
data_root, metadata_db, arrow_dir = store_paths
code = _HARNESS.format(
data_root=data_root, metadata_db=metadata_db, arrow_dir=arrow_dir
) + textwrap.dedent(body)
out = subprocess.run(
[sys.executable, "-c", code], capture_output=True, text=True, timeout=300
)
assert out.returncode == 0, f"subprocess failed:\n{out.stdout}\n{out.stderr}"
return out.stdout.strip().splitlines()[-1].split()
def _sweep(store, n_combos):
"""One in-process run_sweep_lite call with exactly n_combos combinations."""
strat = bt.Strategy(
name="budget_probe",
signals={"signal": bt.lit(1.0)},
position_sizing=bt.lit(1.0) * bt.param("size", default=1.0),
parameters={"size": bt.param("size", default=1.0)},
)
# Minutes(1) is silently coarsened to daily on Community; the runs then
# produce zero trades, which is irrelevant here — only the gate matters.
t0, t1 = bt.time_range("2024-01-01", "2024-01-02")
cfg = bt.BacktestConfig(
universe=[1], time_range_start=t0, time_range_end=t1,
bar_interval={"Minutes": 1},
)
grid = {"size": [1.0 + 0.001 * i for i in range(n_combos)]}
return bt.run_sweep_lite(strat, grid, cfg, store)
# --------------------------------------------------------------- the budget --
@community_only
def test_cumulative_budget(daily_store):
used0, limit, is_pro = _combo_budget()
assert not is_pro
remaining = limit - used0
if remaining < 8:
pytest.skip(f"only {remaining} combos left in this process")
# Two calls of just over half the remaining budget: the first fits,
# the second would cross the cap even though it is individually small.
n = int(remaining) // 2 + 1
_sweep(daily_store, n)
with pytest.raises(PermissionError, match="already used this session"):
_sweep(daily_store, n)
# The rejected call consumed nothing: what actually remains still fits.
leftover = int(limit - _combo_budget()[0])
assert leftover == int(remaining) - n
if leftover >= 1:
_sweep(daily_store, leftover)
assert _combo_budget()[0] == limit
# Budget now exhausted: even a single combo is refused.
with pytest.raises(PermissionError, match="0 remaining"):
_sweep(daily_store, 1)
# ------------------------------------------------------------ the rate gate --
#
# SWEEP_MIN_INTERVAL is 5 s. Thresholds leave generous slack: a call that
# waited is asserted above 4 s, one that did not below 2 s. Nothing here
# depends on machine speed — that is the point of a wall-clock gate.
_INTERVAL = 5.0
_WAITED = 4.0
_DID_NOT_WAIT = 2.0
@community_only
def test_rate_gate_applies_to_every_call(store_paths):
"""Each accepted call waits the interval — it is a rate limit, not a toll."""
first, second = (
float(x) for x in _run(store_paths, """
t1, _ = timed(2)
t2, _ = timed(2)
print(t1, t2)
""")
)
assert first > _WAITED, f"first call took {first:.2f}s, expected a ~5 s wait"
assert second > _WAITED, (
f"second call took {second:.2f}s — the gate is behaving like a one-off "
f"charge instead of a rate limit"
)
@community_only
def test_rate_gate_serialises_across_processes(store_paths):
"""The lock is the mechanism: concurrent waits must queue, not overlap.
Without the file lock two processes would sleep through the same 5 s and
both proceed — the failure mode of every sleep-based limiter. With it, two
concurrent sweeps cost two intervals.
"""
data_root, metadata_db, arrow_dir = store_paths
code = _HARNESS.format(
data_root=data_root, metadata_db=metadata_db, arrow_dir=arrow_dir
) + "timed(2)\n"
t = time.perf_counter()
procs = [
subprocess.Popen([sys.executable, "-c", code],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
for _ in range(2)
]
for proc in procs:
assert proc.wait(timeout=120) == 0
elapsed = time.perf_counter() - t
assert elapsed > 2 * _WAITED, (
f"two concurrent sweeps took {elapsed:.2f}s — under two intervals, so "
f"their waits overlapped and the lock is not serialising them"
)
@community_only
def test_refused_call_does_not_wait(store_paths):
"""Refusing is instant: no 5 s wait before being told no.
The refusal comes first in a fresh process, so a later accepted call still
waits — the refusal neither charged nor exempted anything.
"""
refused, ok, accepted = _run(store_paths, """
t_refused, ok = timed(1000) # larger than the cap
t_accepted, _ = timed(2) # first *accepted* call: waits
print(t_refused, ok, t_accepted)
""")
assert ok == "False", "expected the over-cap call to be refused"
assert float(refused) < _DID_NOT_WAIT, (
f"refused call took {float(refused):.2f}s — it should not wait"
)
assert float(accepted) > _WAITED, (
"the accepted call after a refusal did not wait out the rate limit"
)
@pro_only
def test_pro_is_not_rate_limited(store_paths):
"""Pro skips the gate entirely: no counter, no wait, on any call."""
first, second = (
float(x) for x in _run(store_paths, """
t1, _ = timed(2)
t2, _ = timed(2)
print(t1, t2)
""")
)
assert first < _DID_NOT_WAIT and second < _DID_NOT_WAIT, (
f"Pro waited ({first:.2f}s, {second:.2f}s) — the rate gate leaked"
)
python/tests/test_import_dataframe.py
+48
View File
@@ -162,3 +162,51 @@ def test_import_dataframe_integer_timestamp_raises(tmp_path):
def test_import_dataframe_empty_raises(tmp_path):
with pytest.raises(bt.DataError, match="no data rows"):
_import_df(_bars_df(0), tmp_path)
def test_import_dataframe_daily_interval_runs(tmp_path):
"""Daily bars import AND backtest.
Regression: the resolution table listed only 1m/1h, so a daily store
resolved to the (empty) 1m directory and the run died with "empty bar
dataset for symbol". A ``1d`` entry in the table lets the daily provider
layout be found. 1m/1h were unaffected, which is exactly why this slipped.
"""
n = 30
ts = pd.date_range("2021-01-01", periods=n, freq="1D", tz="UTC")
close = [100.0 + i for i in range(n)] # strictly rising → buy & hold profits
df = pd.DataFrame(
{
"timestamp": ts,
"open": close,
"high": [c + 1.0 for c in close],
"low": [c - 1.0 for c in close],
"close": close,
"volume": [10.0] * n,
}
)
store = _import_df(df, tmp_path, name="daily", interval="1d")
assert store.resolve_symbol("BTCUSDT") == 1
strategy = bt.Strategy(
name="bh",
signals={"signal": bt.lit(1.0)},
position_sizing=bt.col("signal"),
)
config = bt.BacktestConfig(
universe=[1],
time_range_start=0,
time_range_end=int(ts[-1].value) + 5 * 86_400_000_000_000,
bar_interval={"Days": 1},
initial_capital=1000.0,
execution=bt.ExecutionConfig(
signal_delay=1, execution_price="AtClose",
position_sizing_mode="Units",
),
fees=bt.FeeConfig(),
slippage={"FixedBps": {"bps": 0.0}},
)
result = bt.run(strategy, config, store)
equity = result.equity_curve.to_pylist()
assert len(equity) > 0
assert equity[-1] > 1000.0
python/tests/test_parity_vectorbt.py
+399
View File
@@ -0,0 +1,399 @@
"""Cross-engine parity: manifoldbt vs vectorbt on brackets, shorts, fees.
This suite pins manifoldbt's fill semantics against an independent engine
(vectorbt) on controlled synthetic bars, so a refactor that silently changes a
fill price, a stop level, or PnL booking is caught here rather than in the wild.
Coverage — only what vectorbt can legitimately model apples-to-apples:
* market entry + take-profit (test_market_take_profit_parity)
* market entry + stop-loss (test_market_stop_loss_parity)
* combined SL+TP bracket (test_bracket_sl_tp_parity)
* short entry + take-profit (test_short_take_profit_parity)
* trailing stop (test_trailing_stop_parity)
* fees over multiple round-trips (test_fees_multi_trade_parity)
Out of scope for vectorbt (validated separately, NOT against vectorbt):
* determined-price / resting limit entry — vectorbt has no resting order, so
``test_limit_entry_matches_independent_reference`` pins it against a NumPy
model instead.
* sizing under fees — with ``FractionOfEquity`` the engines size differently
once fees exist (manifoldbt charges the fee on top of a full-equity notional;
vectorbt reserves it out of cash). Both are legitimate; the fee test sizes in
fixed units to compare the fee arithmetic without that policy difference.
What is compared, and why only this:
* Trade fills (entry price, exit price, exit reason) and final ``total_return``.
These are computed at full internal resolution and are exact. The *equity
curve* is deliberately NOT compared: on a Community build the output series is
capped to daily resolution, so its shape is not apples-to-apples with
vectorbt. The realised trades and the final equity are unaffected by that cap.
Convention alignment (measured against manifoldbt 0.14.1, not assumed):
* ``signal_delay=0`` + ``AtClose`` → a market entry fills at the *close* of the
signal bar. vectorbt ``from_signals`` fills the entry bar at close by default,
so entries line up with no shift.
* ``FractionOfEquity`` sizing is taken at the *signal-bar close*
(``size_at_fill_price=False``). For a market entry that equals the fill price,
so vectorbt ``size_type="percent"`` matches. For a resting limit entry the
signal close and the fill price differ, so vectorbt is fed an explicit unit
size to reproduce manifoldbt's "size at signal close" rule.
* Take-profit is a passive target: it fills at the level even if the bar gaps
through it. Stop-loss fills at the level (or worse on a gap). vectorbt's
``stop_exit_price=StopMarket`` reproduces the level fill on these
no-gap-at-open scenarios.
vectorbt has no resting entry order, so the limit-entry scenario also carries an
independent NumPy reference for *where* the order fills; vectorbt only checks the
downstream take-profit off that fill.
"""
import os
import pytest
pd = pytest.importorskip("pandas")
vbt = pytest.importorskip("vectorbt")
import manifoldbt as bt # noqa: E402
from manifoldbt.expr import col, lit, when # noqa: E402
from manifoldbt.helpers import Interval, Slippage # noqa: E402
from vectorbt.portfolio.enums import StopExitPrice, Direction # noqa: E402
CAPITAL = 10_000.0
REL_TOL = 1e-6
# Exit-reason codes emitted in trades_df (measured):
REASON_NONE, REASON_SL, REASON_TP, REASON_TRAIL = 0, 1, 2, 3
# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def _bars(o, h, l, c, start="2023-01-01"):
ts = pd.date_range(start, periods=len(c), freq="1h", tz="UTC")
return pd.DataFrame(
{"timestamp": ts, "open": list(map(float, o)), "high": list(map(float, h)),
"low": list(map(float, l)), "close": list(map(float, c)),
"volume": [1000.0] * len(c)}
)
def _mbt_run(df, strat, tmp_path, name, *, delay=0, allow_short=False,
sizing="FractionOfEquity", fees=None):
"""Run manifoldbt on an in-memory OHLC frame; return the Result."""
root = str(tmp_path / name)
os.makedirs(root, exist_ok=True)
store = bt.import_dataframe(
df, symbol="TEST", symbol_id=1, interval="1h",
data_root=os.path.join(root, "data"),
metadata_db=os.path.join(root, "meta.sqlite"),
)
ts = df["timestamp"]
cfg = bt.BacktestConfig(
universe=[1],
time_range_start=0,
time_range_end=int(ts.iloc[-1].value) + 30 * 86_400_000_000_000,
bar_interval=Interval.hours(1),
initial_capital=CAPITAL,
execution=bt.ExecutionConfig(
signal_delay=delay, execution_price="AtClose",
max_position_pct=1.0, allow_short=allow_short,
position_sizing_mode=sizing,
),
fees=fees if fees is not None else bt.FeeConfig.zero(),
slippage=Slippage.none(),
warmup_bars=0,
)
return bt.run(strat, cfg, store)
def _mbt_trades(res):
"""(entry_fill, exit_fill, exit_reason) from a two-row round-trip."""
tr = res.trades_df()
assert len(tr) == 2, f"expected one round-trip, got {len(tr)} rows:\n{tr}"
entry = tr.iloc[0]
exit_ = tr.iloc[1]
return float(entry["fill_price"]), float(exit_["fill_price"]), int(exit_["exit_reason"])
def _vbt_from_signals(df, entries, *, exits=None, tp=None, sl=None,
sl_trail=False, size=1.0, size_type="percent",
direction=None, fees=0.0):
idx = pd.DatetimeIndex(df["timestamp"])
close = pd.Series(df["close"].values, index=idx, dtype=float)
ent = pd.Series(entries, index=idx)
ex = pd.Series(exits if exits is not None else False, index=idx)
kwargs = dict(
open=pd.Series(df["open"].values, index=idx, dtype=float),
high=pd.Series(df["high"].values, index=idx, dtype=float),
low=pd.Series(df["low"].values, index=idx, dtype=float),
init_cash=CAPITAL, size=size, size_type=size_type,
fees=fees, slippage=0.0, sl_stop=sl, tp_stop=tp, sl_trail=sl_trail,
stop_exit_price=StopExitPrice.StopMarket,
freq="1h", accumulate=False,
)
if direction is not None:
kwargs["direction"] = direction
return vbt.Portfolio.from_signals(close, ent, ex, **kwargs)
def _assert_close(a, b, msg):
assert abs(a - b) <= REL_TOL * max(1.0, abs(b)), f"{msg}: {a} != {b}"
# --------------------------------------------------------------------------- #
# Scenario A — market entry + take-profit
# --------------------------------------------------------------------------- #
def test_market_take_profit_parity(tmp_path):
# Enter long at bar 0 close (100). TP +10% (110) is crossed at bar 3
# (open 108 < 110 < high 115): both engines fill the target at 110.
df = _bars(
o=[100, 100, 104, 108, 111, 113],
h=[101, 102, 106, 115, 112, 114],
l=[99, 99, 103, 107, 110, 112],
c=[100, 100, 105, 112, 111, 113],
)
# Long only while close in (99.5, 106): true on bars 0-2, false after, so
# the position is a single clean round-trip closed by the TP.
entry = when((col("close") > lit(99.5)) & (col("close") < lit(106.0)),
lit(1.0), lit(0.0))
strat = (bt.Strategy.create("mkt_tp")
.signal("d", col("close")).size(entry).take_profit(pct=10.0))
res = _mbt_run(df, strat, tmp_path, "A")
m_entry, m_exit, reason = _mbt_trades(res)
assert reason == REASON_TP
_assert_close(m_entry, 100.0, "mbt entry")
_assert_close(m_exit, 110.0, "mbt tp exit")
pf = _vbt_from_signals(df, [True, False, False, False, False, False], tp=0.10)
v_tr = pf.trades.records_readable.iloc[0]
_assert_close(float(v_tr["Avg Entry Price"]), m_entry, "entry price")
_assert_close(float(v_tr["Avg Exit Price"]), m_exit, "exit price")
_assert_close(pf.total_return(), res.metrics["total_return"], "total_return")
# --------------------------------------------------------------------------- #
# Scenario B — market entry + stop-loss
# --------------------------------------------------------------------------- #
def test_market_stop_loss_parity(tmp_path):
# Enter long at bar 0 close (100). SL -5% (95) is hit at bar 3
# (open 97 > 95, low 94 <= 95): both engines fill the stop at 95.
df = _bars(
o=[100, 100, 99, 97, 96, 95],
h=[101, 101, 100, 98, 97, 96],
l=[99, 99, 96, 94, 95, 94],
c=[100, 100, 98, 96, 96, 95],
)
entry = when(col("close") >= lit(97.0), lit(1.0), lit(0.0))
strat = (bt.Strategy.create("mkt_sl")
.signal("d", col("close")).size(entry).stop_loss(pct=5.0))
res = _mbt_run(df, strat, tmp_path, "B")
m_entry, m_exit, reason = _mbt_trades(res)
assert reason == REASON_SL
_assert_close(m_entry, 100.0, "mbt entry")
_assert_close(m_exit, 95.0, "mbt sl exit")
pf = _vbt_from_signals(df, [True, False, False, False, False, False], sl=0.05)
v_tr = pf.trades.records_readable.iloc[0]
_assert_close(float(v_tr["Avg Entry Price"]), m_entry, "entry price")
_assert_close(float(v_tr["Avg Exit Price"]), m_exit, "exit price")
_assert_close(pf.total_return(), res.metrics["total_return"], "total_return")
# --------------------------------------------------------------------------- #
# Scenario C — resting limit entry at a determined price + take-profit
# --------------------------------------------------------------------------- #
def _resting_limit_reference(df, signal_bar, offset_frac, tp_frac, capital):
"""Independent NumPy model of a resting buy-limit + take-profit.
Mirrors the measured manifoldbt rule: the limit rests at
``signal_close * (1 - offset_frac)``, fills on the first bar AFTER the
signal bar whose low touches it (fill AT the level), sizes at the signal
close, then a passive TP at ``fill * (1 + tp_frac)`` closes it on the first
later bar whose high reaches it.
"""
close = df["close"].to_numpy(float)
high = df["high"].to_numpy(float)
low = df["low"].to_numpy(float)
signal_close = close[signal_bar]
limit = signal_close * (1.0 - offset_frac)
qty = capital / signal_close # size_at_fill_price=False
fill_bar = next((i for i in range(signal_bar + 1, len(low)) if low[i] <= limit), None)
assert fill_bar is not None, "limit never filled in reference"
tp = limit * (1.0 + tp_frac)
exit_bar = next((i for i in range(fill_bar, len(high)) if high[i] >= tp), None)
assert exit_bar is not None, "TP never reached in reference"
total_return = qty * (tp - limit) / capital
return dict(limit=limit, qty=qty, fill_bar=fill_bar, tp=tp,
exit_bar=exit_bar, total_return=total_return)
def test_limit_entry_matches_independent_reference(tmp_path):
"""Determined-price (resting limit) entry — validated WITHOUT vectorbt.
vectorbt has no resting entry order: it cannot wait across bars for price to
trade down to a level, so a "vs vectorbt" check would not be apples-to-apples
and is deliberately not attempted. This manifoldbt-only feature is pinned
against an independent NumPy model of the resting fill instead. The vectorbt
suite above covers what both engines share (market entry, SL, TP).
Signal at bar 0 (close 100). Limit rests 2% below (98). Bar 1 low 97 <= 98
fills at 98. TP +5% off the fill (102.9) is reached at bar 3 (open 102 < the
target, so it fills the passive target at the level, not on a gap).
"""
df = _bars(
o=[100, 99, 101, 102, 104, 105],
h=[100.5, 100, 102, 104, 105, 106],
l=[99.5, 97, 100, 101.5, 103, 104],
c=[100, 99, 101, 103, 104, 105],
start="2023-01-02",
)
# Signal fires only on bar 0 so exactly one resting order is placed.
entry = when((col("close") >= lit(99.5)) & (col("close") <= lit(100.5)),
lit(1.0), lit(0.0))
strat = (bt.Strategy.create("lim_tp")
.signal("d", col("close"))
.size(entry)
.limit_entry(offset_bps=200, time_in_force="GTC") # 200 bps = 2%
.take_profit(pct=5.0))
res = _mbt_run(df, strat, tmp_path, "C")
m_entry, m_exit, reason = _mbt_trades(res)
ref = _resting_limit_reference(df, signal_bar=0, offset_frac=0.02,
tp_frac=0.05, capital=CAPITAL)
_assert_close(m_entry, ref["limit"], "limit fill price") # 98.0
_assert_close(m_exit, ref["tp"], "tp exit price") # 102.9
assert reason == REASON_TP
_assert_close(res.metrics["total_return"], ref["total_return"], "total_return")
# --------------------------------------------------------------------------- #
# Scenario D — combined SL+TP bracket (both armed, the right one fires)
# --------------------------------------------------------------------------- #
def test_bracket_sl_tp_parity(tmp_path):
# SL -5% (95) AND TP +10% (110) armed together. Price rises, so the TP fires
# at bar 3 and the stop never triggers — the bracket must not misfire.
df = _bars(
o=[100, 100, 104, 108, 111, 113],
h=[101, 102, 106, 115, 112, 114],
l=[99, 99, 103, 107, 110, 112],
c=[100, 100, 105, 112, 111, 113],
)
entry = when((col("close") > lit(99.5)) & (col("close") < lit(106.0)),
lit(1.0), lit(0.0))
strat = (bt.Strategy.create("bracket")
.signal("d", col("close")).size(entry)
.stop_loss(pct=5.0).take_profit(pct=10.0))
res = _mbt_run(df, strat, tmp_path, "D")
m_entry, m_exit, reason = _mbt_trades(res)
assert reason == REASON_TP
_assert_close(m_exit, 110.0, "mbt tp exit")
pf = _vbt_from_signals(df, [True, False, False, False, False, False],
sl=0.05, tp=0.10)
v_tr = pf.trades.records_readable.iloc[0]
_assert_close(float(v_tr["Avg Exit Price"]), m_exit, "exit price")
_assert_close(pf.total_return(), res.metrics["total_return"], "total_return")
# --------------------------------------------------------------------------- #
# Scenario E — short entry + take-profit
# --------------------------------------------------------------------------- #
def test_short_take_profit_parity(tmp_path):
# Short at bar 0 close (100). TP -5% (95, profit for a short) is hit at bar 3
# (open 96 > 95, low 94 <= 95): both engines cover at 95 for a +5% return.
# Signal is short on bars 0-2 and flat from bar 3, so the TP closes it with
# no re-entry.
df = _bars(
o=[100, 99, 98, 96, 95, 94],
h=[100.5, 100, 99, 97, 96, 95],
l=[99.5, 98, 97, 94, 94, 93],
c=[100, 98, 97, 95, 94, 93],
)
entry = when(col("close") >= lit(96.0), lit(-1.0), lit(0.0))
strat = (bt.Strategy.create("short_tp")
.signal("d", col("close")).size(entry).take_profit(pct=5.0))
res = _mbt_run(df, strat, tmp_path, "E", allow_short=True)
m_entry, m_exit, reason = _mbt_trades(res)
assert reason == REASON_TP
_assert_close(m_entry, 100.0, "short entry")
_assert_close(m_exit, 95.0, "short cover")
pf = _vbt_from_signals(df, [True, False, False, False, False, False],
tp=0.05, direction=Direction.ShortOnly)
v_tr = pf.trades.records_readable.iloc[0]
_assert_close(float(v_tr["Avg Entry Price"]), m_entry, "entry price")
_assert_close(float(v_tr["Avg Exit Price"]), m_exit, "cover price")
_assert_close(pf.total_return(), res.metrics["total_return"], "total_return")
# --------------------------------------------------------------------------- #
# Scenario F — trailing stop
# --------------------------------------------------------------------------- #
def test_trailing_stop_parity(tmp_path):
# Always long. The high peaks at 112 (bar 3-4), so a 5% trailing stop rests
# at 112 * 0.95 = 106.4. Bar 5 low (104) trades through it: both engines exit
# at 106.4. (vectorbt's sl_trail also trails off the high when high is given.)
df = _bars(
o=[100, 101, 106, 110, 111, 108],
h=[100, 102, 108, 112, 112, 109],
l=[100, 100, 105, 109, 109, 104],
c=[100, 102, 107, 111, 110, 105],
)
strat = (bt.Strategy.create("trail")
.signal("d", col("close")).size(lit(1.0))
.trailing_stop(pct=5.0, use_high=True))
res = _mbt_run(df, strat, tmp_path, "F")
tr = res.trades_df()
# Always-long re-enters at the exit bar's close (a mark-flat no-op on the
# last bar), so the round-trip is the first two rows; assert on those.
assert float(tr.iloc[1]["fill_price"]) == pytest.approx(106.4)
assert int(tr.iloc[1]["exit_reason"]) == REASON_TRAIL
pf = _vbt_from_signals(df, [True, False, False, False, False, False],
sl=0.05, sl_trail=True)
v_tr = pf.trades.records_readable.iloc[0]
_assert_close(float(v_tr["Avg Exit Price"]), 106.4, "trailing exit")
_assert_close(pf.total_return(), res.metrics["total_return"], "total_return")
# --------------------------------------------------------------------------- #
# Scenario G — fees over multiple round-trips (cumulative accounting)
# --------------------------------------------------------------------------- #
def test_fees_multi_trade_parity(tmp_path):
# Two round-trips with a 20 bps taker fee, sized in FIXED UNITS. Fixed units
# are deliberate: under FractionOfEquity the engines size differently once
# fees exist (manifoldbt charges the fee on top of a full-equity notional,
# vectorbt reserves the fee out of cash), which is a legitimate design choice
# rather than a parity bug. Fixing the unit count isolates the thing both
# engines must agree on — the fee arithmetic and its cumulative effect.
units = 50.0
close = [100, 101, 102, 99, 98, 103, 99]
df = _bars(
o=close, h=[c + 0.5 for c in close], l=[c - 0.5 for c in close], c=close,
start="2023-06-01",
)
# Long while close > 100: enters bar 1, exits bar 3, re-enters bar 5, exits
# bar 6 → two clean round-trips.
entry = when(col("close") > lit(100.0), lit(units), lit(0.0))
strat = bt.Strategy.create("fees").signal("d", col("close")).size(entry)
fees = bt.FeeConfig(maker_fee_bps=10.0, taker_fee_bps=20.0)
res = _mbt_run(df, strat, tmp_path, "G", sizing="Units", fees=fees)
sig = pd.Series(close, dtype=float) > 100
entries = sig & ~sig.shift(1, fill_value=False)
exits = ~sig & sig.shift(1, fill_value=False)
pf = _vbt_from_signals(df, entries.tolist(), exits=exits.tolist(),
size=units, size_type="amount", fees=0.002)
_assert_close(pf.total_return(), res.metrics["total_return"], "total_return")