From 71156abe18f68d5ce5894b89eb976a2fbdab4b4b Mon Sep 17 00:00:00 2001 From: Jimmy7892 Date: Sat, 22 Aug 2026 04:17:34 +0200 Subject: [PATCH] preview: 0.19.0rc1 - higher-timeframe indicators and expression choices 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. --- pyproject.toml | 2 +- python/manifoldbt/__init__.py | 3 +- python/manifoldbt/expr.py | 83 ++++++++++++ python/tests/test_choice.py | 140 ++++++++++++++++++++ python/tests/test_on_timeframe.py | 205 ++++++++++++++++++++++++++++++ 5 files changed, 431 insertions(+), 2 deletions(-) create mode 100644 python/tests/test_choice.py create mode 100644 python/tests/test_on_timeframe.py diff --git a/pyproject.toml b/pyproject.toml index 25e2146..e29afbb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "manifoldbt" -version = "0.18.0" +version = "0.19.0rc1" description = "Rust-powered backtesting engine for quantitative research" requires-python = ">=3.9" license = { file = "LICENSE" } diff --git a/python/manifoldbt/__init__.py b/python/manifoldbt/__init__.py index 480ac07..79da9a7 100644 --- a/python/manifoldbt/__init__.py +++ b/python/manifoldbt/__init__.py @@ -54,7 +54,7 @@ from manifoldbt.exceptions import ( LicenseError, StrategyError, ) -from manifoldbt.expr import AssetRef, Expr, TimeframeRef, asset, col, exo, hold, lit, param, s, scan, symbol_ref, tf, when +from manifoldbt.expr import AssetRef, Expr, TimeframeRef, asset, choice, col, exo, hold, lit, param, s, scan, symbol_ref, tf, when from manifoldbt.helpers import ( ExecutionPrice, FillModel, @@ -1642,6 +1642,7 @@ __all__ = [ "s", "scan", "symbol_ref", + "choice", "tf", "when", # Strategy & config diff --git a/python/manifoldbt/expr.py b/python/manifoldbt/expr.py index db0203b..c72009d 100644 --- a/python/manifoldbt/expr.py +++ b/python/manifoldbt/expr.py @@ -182,6 +182,17 @@ class Expr: if v == "IfElse": return {v: [args[0].to_json(), args[1].to_json(), args[2].to_json()]} + if v == "Choice": + # Choice(String, Vec<(String, Expr)>) -- serde attend une liste de + # paires, pas un dict : l'ORDRE des branches est porteur (la + # premiere sert de defaut a la compilation initiale). + return {"Choice": [args[0], [[k, e.to_json()] for k, e in args[1]]]} + + if v == "OnTimeframe": + # OnTimeframe(String, Box) -- l'expression est evaluee sur la + # grille de la timeframe nommee puis etalee en escalier. + return {"OnTimeframe": [args[0], args[1].to_json()]} + if v == "Column": return {"Column": args[0]} if v == "Literal": @@ -518,6 +529,56 @@ def when(condition: Expr, true_value: Any = 1.0, false_value: Any = float("nan") return Expr("IfElse", condition, _coerce(true_value), _coerce(false_value)) +def choice(name: str, branches: "dict[str, Expr]", *, description: str = "") -> Expr: + """Balayer un CHOIX d'expression, pas seulement un nombre. + + Un ``param()`` ordinaire porte une valeur numerique. ``choice()`` porte un + NOM, et chaque nom designe une sous-expression differente. Le moteur + remplace le noeud entier par la branche choisie AVANT de simuler, donc une + combinaison n'evalue que sa propre branche : les autres n'existent plus. + + C'est ce qui le distingue d'un ``when()`` imbrique, qui construit toutes + les variantes et tranche barre par barre. + + Usage (balayer la timeframe d'une bande, sim en 1m):: + + bande = mbt.choice("band", { + "30m": sma(mbt.tf("30m").close, mbt.param("len")), + "1h": sma(mbt.tf("1h").close, mbt.param("len")), + "2h": sma(mbt.tf("2h").close, mbt.param("len")), + }) + # grille : {"len": [10, 20, 30], "band": ["30m", "1h", "2h"]} + + Les branches acceptent n'importe quelle expression, donc le meme mecanisme + balaie une colonne exogene, un actif ou un type d'indicateur. + + Args: + name: nom du parametre selecteur, a mettre dans la grille. + branches: nom de branche -> expression. L'ordre compte : la premiere + sert de defaut quand le parametre est absent. + description: metadonnee libre. + + Raises: + ValueError: si ``branches`` est vide. + """ + if not branches: + raise ValueError( + f"choice({name!r}) needs at least one branch; an empty choice has " + f"nothing to resolve to." + ) + items = [(str(k), _coerce(v)) for k, v in branches.items()] + expr = Expr("Choice", name, items) + # Declare le selecteur comme un parametre a part entiere, sans quoi le + # balayer serait refuse par la validation ("parameter not declared"). + expr._param_meta = { + "name": name, + "default": items[0][0], + "range": None, + "description": description, + } + return expr + + def exo(name: str, column: Optional[str] = None) -> Expr: """Reference an exogenous data column. @@ -626,6 +687,28 @@ class TimeframeRef: """Reference any column from this timeframe.""" return col(f"{self._tf}.{name}") + def apply(self, expr: "Expr") -> Expr: + """Evaluate *expr* ON this timeframe's own grid, then step-hold the + result back onto the simulation grid (forward-filled, no lookahead: + a completed bar's value becomes readable from the next bar on). + + This is what makes higher-timeframe INDICATORS correct. Periods + inside *expr* count in THIS timeframe's bars:: + + h1 = bt.tf("1h") + band = h1.apply(sma(close, mbt.param("len"))) # len = HOURS + + is a true SMA of ``len`` hourly closes, sweepable like any param. + By contrast ``sma(h1.close, 20)`` counts 20 SIMULATION bars over a + step-held hourly series -- on a 1m simulation that is a 20-MINUTE + smoothing of a staircase, not a 20-hour average. + + Inside *expr*, ``close``/``open``/... refer to this timeframe's own + resampled columns. Requires ``extra_timeframes`` to declare the + timeframe. Nesting ``apply`` inside another ``apply`` is rejected. + """ + return Expr("OnTimeframe", self._tf, _coerce(expr)) + def __repr__(self) -> str: return f"TimeframeRef({self._tf!r})" diff --git a/python/tests/test_choice.py b/python/tests/test_choice.py new file mode 100644 index 0000000..d765b70 --- /dev/null +++ b/python/tests/test_choice.py @@ -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" + ) diff --git a/python/tests/test_on_timeframe.py b/python/tests/test_on_timeframe.py new file mode 100644 index 0000000..924d756 --- /dev/null +++ b/python/tests/test_on_timeframe.py @@ -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)