release: v0.16.0

examples/21_fill_at_computed_level.py

@@ -0,0 +1,114 @@
+"""Filling at a computed level — ExecutionPrice.custom(<signal name>).
+
+A mean-reversion band strategy on native 1-minute bars: short at the touch of
+an upper band around an hourly SMA, cover at the lower band. The engine always
+knew how to COMPUTE the band; this example shows the fill landing ON it.
+
+The touch bar itself proves the level traded: it opens below the band and its
+high crosses it, so the band price sits inside [open, high]. Yet with
+``AtClose`` the only reachable fill is the bar's close — on a mean-reverting
+touch, systematically on the wrong side of the level. The same run is done
+both ways so the difference is visible in one place.
+
+Self-contained: generates its own synthetic data in a temp store.
+
+Usage:
+    python examples/21_fill_at_computed_level.py
+"""
+
+import os
+import tempfile
+
+import numpy as np
+import pandas as pd
+
+import manifoldbt as mbt
+from manifoldbt.indicators import close, high, low, open as open_px, sma
+from manifoldbt.helpers import ExecutionPrice, Interval, Slippage
+
+# -- Synthetic 1m data: a mean-reverting walk ---------------------------------
+N = 30 * 1440  # 30 days of 1-minute bars
+rng = np.random.default_rng(7)
+steps = rng.normal(0.0, 0.0010, N)
+level = np.cumsum(steps) * 0.85  # pull the walk back toward its mean
+px = 100.0 * np.exp(level - np.linspace(0, level[-1], N))
+o, c = px, np.roll(px, -1)
+c[-1] = px[-1]
+amp = np.abs(rng.normal(0.0, 0.0012, N))
+ts = pd.date_range("2024-01-01", periods=N, freq="1min", tz="UTC")
+frame = pd.DataFrame(
+    {"timestamp": ts, "open": o,
+     "high": np.maximum(o, c) * (1 + amp), "low": np.minimum(o, c) * (1 - amp),
+     "close": c, "volume": rng.uniform(1_000, 5_000, N)}
+)
+
+# -- Bands around an hourly SMA, evaluated on 1m native bars ------------------
+DEV_UP, DEV_DN = 0.004, 0.003
+
+h1 = mbt.tf("1h")  # hourly columns, as of the last closed hour
+band_up = sma(h1.close, 8) * (1 + DEV_UP)
+band_dn = sma(h1.close, 8) * (1 - DEV_DN)
+
+touch_up = high >= band_up   # entry: short at the touch of the upper band
+touch_dn = low <= band_dn    # exit: cover at the touch of the lower band
+target = mbt.when(touch_dn, 0.0, mbt.when(touch_up, -1.0))
+
+# The level each fill should land on. The nesting mirrors the target's
+# priority, and a bar that opens through a band fills at its open.
+exec_level = mbt.when(
+    touch_dn, mbt.when(open_px <= band_dn, open_px, band_dn),
+    mbt.when(touch_up, mbt.when(open_px >= band_up, open_px, band_up), close),
+)
+
+strategy = (
+    mbt.Strategy.create("band_touch_short")
+    .signal("position", target)
+    .signal("exec_level", exec_level)
+    .size(target)
+    .stop_loss(pct=25.0)
+)
+
+# -- Run the same strategy both ways ------------------------------------------
+if __name__ == "__main__":
+    root = tempfile.mkdtemp(prefix="mbt_example21_")
+    store = mbt.import_dataframe(
+        frame, symbol="SYNTH", symbol_id=1, interval="1m",
+        data_root=os.path.join(root, "data"),
+        metadata_db=os.path.join(root, "meta.sqlite"),
+    )
+
+    def run(execution_price):
+        config = mbt.BacktestConfig(
+            universe=[1],
+            time_range_start=0,
+            time_range_end=int(ts[-1].value) + 86_400_000_000_000,
+            bar_interval=Interval.minutes(1),
+            initial_capital=10_000,
+            execution=mbt.ExecutionConfig(
+                signal_delay=0,
+                execution_price=execution_price,
+                max_position_pct=0.4,
+                allow_short=True,
+                position_sizing_mode="FractionOfEquity",
+            ),
+            fees=mbt.FeeConfig.binance_perps(),
+            slippage=Slippage.fixed_bps(2),
+            warmup_bars=60 * 10,
+            extra_timeframes={"1h": Interval.hours(1)},
+        )
+        return mbt.run(strategy, config, store)
+
+    print(f"{'execution price':<22} {'trades':>7} {'return':>9}   first entry fills")
+    print("-" * 78)
+    for label, price in (("AtClose", "AtClose"),
+                         ("custom('exec_level')", ExecutionPrice.custom("exec_level"))):
+        result = run(price)
+        tr = result.trades_df()
+        entries = tr[tr["fill_price"] > 0].head(3)["fill_price"].round(4).tolist()
+        print(f"{label:<22} {len(tr):>7} {result.metrics['total_return']:>8.2%}   {entries}")
+
+    print(
+        "\nSame signals, same bars: only WHERE the order fills changed. The"
+        "\ncustom fills land on the band level (inside the touch bar's range),"
+        "\nnot on its close. A fill outside [low, high] would be warned about."
+    )