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

preview: 0.19.0rc1 - higher-timeframe indicators and expression choices

Browse Source 
Synced from the private engine at 1e39bfa (branch feat/sweep-choice-expr),
tracked files only. Two additions to the expression API:

- tf("1h").apply(expr): evaluate an expression ON the higher timeframe's
  grid, then step-hold it onto the simulation grid without lookahead.
  Periods inside count in that timeframe's bars, so
  tf("1h").apply(sma(close, param("len"))) is a true SMA of len hourly
  closes, sweepable like any param.
- choice(name, {branch: expr}): sweep a CHOICE of expression. The selector
  becomes a grid axis; each combination resolves to its branch before
  simulation.

Wheel for this preview is attached to the v0.19.0rc1 pre-release; built
locally, not by the release pipeline, not on PyPI.
This commit is contained in:
Jimmy7892
2026-08-22 04:17:34 +02:00
parent 0b93aed93b
commit 71156abe18
5 changed files with 431 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/tests/test_choice.py
+140
View File
@@ -0,0 +1,140 @@
"""Tests for bt.choice() — sweeping a CHOICE of expression, not just a number.
The contract under test: `choice("sel", {...})` resolves to exactly one branch
per combo BEFORE simulation, so a sweep over the selector must be
bit-identical to running each branch inlined by hand. The selector must count
as a declared parameter (otherwise `_validate_swept_params` would reject the
sweep), and an unknown branch must fail with a message naming the known ones.
"""
import json
import os
import pytest
import manifoldbt as bt
from manifoldbt.indicators import close, sma
pd = pytest.importorskip("pandas")
np = pytest.importorskip("numpy")
N_BARS = 3_000
def _store(tmp_path):
ts = pd.date_range("2022-01-01", periods=N_BARS, freq="1min", tz="UTC")
rng = np.random.default_rng(11)
px = 100.0 + np.cumsum(np.sin(np.arange(N_BARS) / 90.0) * 0.3 + rng.normal(0, 0.2, N_BARS)) * 0.05
px = np.maximum(px, 1.0)
df = pd.DataFrame(
{
"timestamp": ts,
"open": px,
"high": px * 1.0005,
"low": px * 0.9995,
"close": px,
"volume": [1000.0] * N_BARS,
}
)
root = tmp_path / "choice_store"
return bt.import_dataframe(
df,
symbol="ZC",
symbol_id=1,
interval="1m",
asset_class="equity",
exchange="TEST",
data_root=str(root / "data"),
metadata_db=str(root / "metadata.sqlite"),
)
def _config():
start, end = bt.time_range("2022-01-01", "2022-01-03")
cfg = bt.BacktestConfig(
universe=[1],
time_range_start=start,
time_range_end=end,
initial_capital=10_000.0,
provider="TEST",
bar_interval=bt.Interval.minutes(1),
symbol_names={"ZC": 1},
)
cfg.warmup_bars = 0
return cfg
def _strategy_with_choice():
band = bt.choice(
"pick",
{
"fast": sma(close, 5),
"slow": sma(close, 20),
},
)
return (
bt.Strategy.create("choice_e2e")
.signal("band", band)
.size(bt.when(close > bt.col("band"), 1.0, 0.0))
)
def _strategy_inlined(period):
return (
bt.Strategy.create(f"inline_{period}")
.signal("band", sma(close, period))
.size(bt.when(close > bt.col("band"), 1.0, 0.0))
)
def test_serializes_as_ordered_pairs():
"""serde expects Choice(String, Vec<(String, Expr)>): a list of pairs,
order preserved — the first branch is the compile-time default."""
e = bt.choice("pick", {"a": close, "b": sma(close, 3)})
payload = json.loads(json.dumps(e.to_json()))
assert list(payload) == ["Choice"]
name, branches = payload["Choice"]
assert name == "pick"
assert [k for k, _ in branches] == ["a", "b"]
def test_empty_branches_rejected():
with pytest.raises(ValueError, match="at least one branch"):
bt.choice("pick", {})
def test_selector_counts_as_declared_parameter():
"""Sweeping the selector must pass strategy-side validation: choice()
declares it via _param_meta exactly like param() does."""
strat = _strategy_with_choice()
assert "pick" in (strat.to_json_dict().get("parameters") or {})
def test_sweep_over_choice_matches_inlined_branches(tmp_path):
"""The money test: each combo of the selector sweep is bit-identical to
the strategy with that branch written directly."""
store = _store(tmp_path)
cfg = _config()
sweep = bt.run_sweep_lite(
_strategy_with_choice(), {"pick": ["fast", "slow"]}, cfg, store, device="cpu"
)
assert len(sweep) == 2
by_branch = dict(zip(["fast", "slow"], sweep))
for name, period in (("fast", 5), ("slow", 20)):
ref = bt.run_sweep_lite(
_strategy_inlined(period), {}, cfg, store, device="cpu"
)[0]
got, want = by_branch[name].metrics, ref.metrics
for key in ("total_return", "sharpe", "max_drawdown"):
assert got.get(key) == want.get(key), (
f"branch {name!r}: {key} diverged ({got.get(key)} vs {want.get(key)})"
)
def test_unknown_branch_names_the_known_ones(tmp_path):
store = _store(tmp_path)
with pytest.raises(Exception, match="fast"):
bt.run_sweep_lite(
_strategy_with_choice(), {"pick": ["nope"]}, _config(), store, device="cpu"
)
python/tests/test_on_timeframe.py
+205
View File
@@ -0,0 +1,205 @@
"""Tests for tf(..).apply(..) — indicators evaluated ON the higher timeframe.
The defect this feature fixes, pinned by `test_apply_differs_from_staircase`:
an indicator over a step-held `tf()` column counts its period in SIMULATION
bars, so `sma(tf("1h").close, 20)` on a 1m simulation is a 20-MINUTE smoothing
of an hourly staircase — mid-hour it equals the previous hourly close exactly.
`tf("1h").apply(sma(close, 20))` is the true 20-HOUR average.
The reference implementation (`_hand_band`) is the exo-column recipe users had
to build by hand before this feature: resample to 1h in pandas, indicator on
the 1h grid, shift(1) (a closed bar is readable from the next bar on), ffill
onto the 1m grid. `test_apply_matches_hand_rolled_exo` demands bit-identical
metrics against it.
"""
import os
import pytest
import manifoldbt as bt
from manifoldbt.indicators import close, sma
pd = pytest.importorskip("pandas")
np = pytest.importorskip("numpy")
N_BARS = 20_000 # ~13.9 days of 1m bars -> ~333 hourly bars
PERIOD = 20
def _bars_df():
ts = pd.date_range("2022-01-01", periods=N_BARS, freq="1min", tz="UTC")
rng = np.random.default_rng(3)
px = 100.0 + np.cumsum(np.sin(np.arange(N_BARS) / 700.0) * 0.5 + rng.normal(0, 0.4, N_BARS)) * 0.01
px = np.maximum(px, 1.0)
return pd.DataFrame(
{
"timestamp": ts,
"open": px,
"high": px * 1.0005,
"low": px * 0.9995,
"close": px,
"volume": [1000.0] * N_BARS,
}
)
def _store(tmp_path, df):
root = tmp_path / "otf_store"
return bt.import_dataframe(
df,
symbol="ZT",
symbol_id=1,
interval="1m",
asset_class="equity",
exchange="TEST",
data_root=str(root / "data"),
metadata_db=str(root / "metadata.sqlite"),
), str(root / "data")
def _hand_band(df, period):
"""The pre-feature recipe: hourly SMA built by hand, no lookahead."""
h1 = df.set_index("timestamp")["close"].resample("1h").last()
band = h1.rolling(period).mean().shift(1)
return band.reindex(df["timestamp"], method="ffill")
def _config(tmp_path=None, exo=False):
start, end = bt.time_range("2022-01-01", "2022-01-14")
cfg = bt.BacktestConfig(
universe=[1],
time_range_start=start,
time_range_end=end,
initial_capital=10_000.0,
provider="TEST",
bar_interval=bt.Interval.minutes(1),
symbol_names={"ZT": 1},
extra_timeframes={} if exo else {"1h": bt.Interval.hours(1)},
exo_data=["hand_band"] if exo else [],
)
cfg.warmup_bars = 0
return cfg
def _strategy(band_expr, name):
return (
bt.Strategy.create(name)
.signal("band", band_expr)
.size(bt.when(close > bt.col("band"), 1.0, 0.0))
)
def test_serializes_as_on_timeframe():
e = bt.tf("1h").apply(sma(close, PERIOD))
payload = e.to_json()
assert list(payload) == ["OnTimeframe"]
label, inner = payload["OnTimeframe"]
assert label == "1h"
assert list(inner) == ["RollingMean"]
def test_apply_matches_hand_rolled_exo(tmp_path):
"""The money test: tf("1h").apply(sma(close, 20)) must be bit-identical to
the hand-precomputed hourly-SMA exo column it replaces."""
df = _bars_df()
store, data_root = _store(tmp_path, df)
band = _hand_band(df, PERIOD)
bt.register_exo(
"hand_band",
pd.DataFrame({"timestamp": df["timestamp"], "value": band.values}),
data_root=data_root,
)
native = bt.run(
_strategy(bt.tf("1h").apply(sma(close, PERIOD)), "native"),
_config(),
store,
)
hand = bt.run(
_strategy(bt.exo("hand_band", "value"), "hand"),
_config(exo=True),
store,
)
for key in ("total_return", "sharpe", "max_drawdown", "volatility"):
assert native.metrics.get(key) == hand.metrics.get(key), (
f"{key}: native {native.metrics.get(key)} != hand {hand.metrics.get(key)}"
)
assert len(native.trades_df()) == len(hand.trades_df())
def test_apply_differs_from_staircase(tmp_path):
"""Guard against regressing to the old semantics: the staircase version
(indicator over the step-held tf() column) must NOT equal apply()."""
df = _bars_df()
store, _ = _store(tmp_path, df)
applied = bt.run(
_strategy(bt.tf("1h").apply(sma(close, PERIOD)), "applied"),
_config(),
store,
)
staircase = bt.run(
_strategy(sma(bt.tf("1h").close, PERIOD), "staircase"),
_config(),
store,
)
assert applied.metrics.get("total_return") != staircase.metrics.get("total_return"), (
"apply() and the staircase smoothing agreed; the coarse-grid evaluation "
"is not actually happening"
)
def test_swept_period_matches_fixed_runs(tmp_path):
"""param() INSIDE apply(): each combo must equal the fixed-period run."""
df = _bars_df()
store, _ = _store(tmp_path, df)
periods = [10, 20, 40]
sweep = bt.run_sweep_lite(
_strategy(bt.tf("1h").apply(sma(close, bt.param("len"))), "swept"),
{"len": periods},
_config(),
store,
device="cpu",
)
assert len(sweep) == len(periods)
for got, period in zip(sweep, periods):
ref = bt.run_sweep_lite(
_strategy(bt.tf("1h").apply(sma(close, period)), f"fixed_{period}"),
{},
_config(),
store,
device="cpu",
)[0]
for key in ("total_return", "sharpe", "max_drawdown"):
assert got.metrics.get(key) == ref.metrics.get(key), (
f"len={period}: {key} diverged"
)
def test_lite_matches_run(tmp_path):
df = _bars_df()
store, _ = _store(tmp_path, df)
strat = _strategy(bt.tf("1h").apply(sma(close, PERIOD)), "parity")
full = bt.run(strat, _config(), store)
lite = bt.run_sweep_lite(strat, {}, _config(), store, device="cpu")[0]
for key in ("total_return", "sharpe"):
assert full.metrics.get(key) == lite.metrics.get(key), f"{key} diverged"
# max_drawdown carries a pre-existing ~1e-16 run-vs-lite float-noise gap
# (measured on a plain sma(close, 20) strategy with no OnTimeframe on this
# same data), so exact equality would pin the wrong thing here.
a, b = full.metrics["max_drawdown"], lite.metrics["max_drawdown"]
assert a == pytest.approx(b, rel=1e-12), f"max_drawdown diverged: {a} vs {b}"
def test_missing_extra_timeframe_is_a_clear_error(tmp_path):
df = _bars_df()
store, _ = _store(tmp_path, df)
cfg = _config()
cfg.extra_timeframes = {}
with pytest.raises(Exception, match="extra_timeframes"):
bt.run(_strategy(bt.tf("1h").apply(sma(close, PERIOD)), "no_tf"), cfg, store)