manifoldbt/examples/21_fill_at_computed_level.py

"""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, cover at the lower one. The engine always knew how to COMPUTE
the band; this example shows the fill landing ON it.

**This is a feature demo, not a claim about markets.** The data is synthetic and
detrended by construction, so mean reversion cannot lose on it whatever the
parameters — the returns printed below describe the fixture, nothing else. The
same run is done both ways only so you can see that the setting takes effect.

Which of the two fills is the better one depends entirely on what the market
does after the touch, and this fixture reverts by construction. Do not read a
rule into the sign of the gap.

The touch bar proves the level traded: it opens below the band and its high
crosses it, so the band price sits inside [open, high]. On this data all 248
intended levels land inside their bar. A level that did not would be clamped
back into [low, high] and reported in ``result.warnings`` — the band comes from
the previous closed hour, so it can sit outside a bar that never reached it.

Look-ahead was tested for and ruled out: the fill price does not move when the
triggering bar's high is inflated by 3%.

Self-contained: generates its own synthetic data in a temp store.

Demonstrates:
  - ExecutionPrice.custom(<signal name>): the fill price as a strategy signal
  - tf("1h").apply(...): an indicator whose period counts hourly candles
  - a level computed from a higher timeframe, known before the bar opens

Data: synthetic (seed 7) — generated by this file, reproducible

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 walk with its drift REMOVED -------------------------
# `- np.linspace(0, level[-1], N)` subtracts the entire trend, forcing the
# series to end exactly where it started: the raw walk loses 25.6%, this one
# returns -0.0%. Mean reversion is therefore GUARANTEED here, by construction.
#
# That is deliberate, and it is why the absolute returns printed below mean
# nothing. Measured over nine parameter sets (bands 0.002 to 0.008, periods 4h
# to 20h), the custom fill returns +2.2% to +9.1% and AtClose -25.5% to +4.2%:
# neither range is a property of the strategy, and AtClose even turns positive
# on the widest band. What holds across all nine is only the ORDER — the custom
# fill is ahead every time — and that gap is the whole subject.
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
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 8-hour mean, on 1m native bars ---------------------------
# `apply()` evaluates the SMA on the hourly grid, so its period counts hourly
# candles. Written `sma(h1.close, 8)` the period would count 8 SIMULATION bars
# over a step-held column, which is 8 minutes, not 8 hours. See `bt.tf`.
DEV_UP, DEV_DN = 0.004, 0.003

h1 = mbt.tf("1h")
hourly_mean = h1.apply(sma(close, 8))     # mean of the last 8 hourly closes
band_up = hourly_mean * (1 + DEV_UP)
band_dn = hourly_mean * (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("Synthetic detrended data: the LEVELS below are an artifact of the")
    print("fixture, not a result. Only the gap between the two rows is real.\n")
    print(f"{'execution price':<22} {'trades':>7} {'return*':>9}   first entry fills")
    print("-" * 78)

    returns = {}
    n_trades = 0
    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()
        returns[label] = result.metrics["total_return"]
        n_trades = len(tr)
        print(f"{label:<22} {len(tr):>7} {returns[label]:>8.2%}   {entries}")

    gap = returns["custom('exec_level')"] - returns["AtClose"]
    print(f"\n  GAP: {gap:.1%} on identical signals and identical bars.")
    print(
        "\n* Both figures describe the fixture, not a market: the series is"
        "\n  detrended, so reversion wins on it by construction. What the two"
        "\n  rows show is that the setting takes effect -- the custom fills land"
        "\n  on the band, computed from the previous closed hour, where AtClose"
        "\n  can only reach the bar's close. Which of the two is the better"
        "\n  price depends on what the market does next, and this series was"
        "\n  built to revert."
        "\n"
        f"\n  All {n_trades} intended levels land inside their own bar here. One"
        f"\n  that did not would be clamped into [low, high] and reported in"
        f"\n  result.warnings."
    )