references/metric_bridge.md
# Metric Bridge Reference
`skills/backtest-expert/scripts/evaluate_backtest.py` takes eight inputs. This
page says where each one comes from and what goes wrong when you source it by
hand.
## The eight
| Evaluator input | Source | Conversion |
|-----------------|--------|------------|
| `--total-trades` | paired round trips | count of completed trips, not fills |
| `--win-rate` | paired round trips | percent of all completed trips that made money net of fees |
| `--avg-win-pct` | paired round trips | mean net return of winners, percent of notional |
| `--avg-loss-pct` | paired round trips | mean net return of losers, absolute value |
| `--max-drawdown-pct` | engine metrics | `abs(max_drawdown) * 100` |
| `--years-tested` | config time range | `(end_ns - start_ns) / 365.25 days` |
| `--num-parameters` | spec | see `strategy_spec.md` |
| `--slippage-tested` | spec | set when fees or slippage are non-zero |
## The four traps
Each one produces a number that looks right and scores the strategy wrongly.
None raises an exception.
### Fills counted as trades
The engine's `trade_stats.total_trades` counts executions. A round trip needs
two, so the raw count runs at about double. The sample-size dimension scores 0
below 30 trades and 20 above 200, so a doubled count can carry a 90-trade
backtest into the top band it has not earned.
Use `round_trips`, which the engine also reports and which `pair_round_trips`
recomputes independently.
### Fills zipped in pairs
Taking rows two at a time assumes buy and sell alternate. That holds for one
shape: single symbol, long only, no scaling. Outside it:
- a sell can **open** a short, so `(exit - entry) / entry` reports the sign
backwards for the whole short side
- scaling in leaves one exit answering several entries, so a pair reads
`(buy, buy)` as a trip and drops the exit
- a universe interleaves fills across symbols, so pairs cross instruments
`pair_round_trips` tracks position per symbol and closes a trip when it crosses
back through flat. It accumulates entry / exit quantities and cash values over
the whole lifecycle, including a scale-out followed by a re-add. The weighted
average prices are for display; gross and net PnL are derived from those cash
flows so earlier exits never use the remaining position's cost basis.
### Gross percentages beside a net win rate
At 7 bps a side, a trade that gains 0.1% on price loses money. Count it as a
win and the win rate rises while the account falls.
The evaluator computes expectancy as `win_rate * avg_win - loss_rate * avg_loss`.
Feed it a gross win rate with net averages and the expectancy is wrong in a
direction that flatters the strategy. All percentages here are net of fees;
`gross_return_pct` stays on each trip for inspection.
`build_evaluation_inputs` compares its win rate against the engine's own and
warns when they differ by more than a point. A gap that large means one of the
two computations is wrong, and the numbers should not leave the machine until
you know which.
### Drawdown passed with its sign
The engine reports `max_drawdown` as a negative fraction: `-0.3846` for a 38.46%
fall. The evaluator wants a positive percent and scores worse as it grows. Pass
`-0.3846` and a near-catastrophic drawdown scores as a perfect one.
## Scratch trades
A trip that returns exactly zero is neither a win nor a loss. `summarize_round_trips`
counts it in `total_trades`, keeps it in the win-rate denominator because it did
not win, and reports it under `scratches`.
The eight-input evaluator cannot represent a third outcome population. It
derives loss rate as `1 - win_rate`, so handing it scratches would make either
expectancy or average loss describe a different population. The bridge therefore
refuses the handoff when scratches are present instead of producing a plausible
but wrong score.
## Warnings the bridge emits
| Warning | Why it matters |
|---------|----------------|
| fewer than 30 round trips | the sample-size dimension scores 0 |
| span under a year | the robustness dimension scores 0 |
| no friction modelled | execution realism scores 0 |
| win rate gap over one point | the pairing and the engine disagree |
Read them before the numbers. Each one changes what the score means.
## Conditions that stop the handoff
| Condition | Why it fails closed |
|-----------|---------------------|
| no completed round trip | win rate and average returns are undefined, while the evaluator can still produce a plausible score |
| `max_drawdown` missing or non-finite | substituting zero would award an almost-perfect risk score |
| one or more scratch trades | the evaluator has no scratch input and cannot derive exact expectancy |
references/strategy_spec.md
# Strategy Spec Reference
A spec is a JSON object. `scripts/spec.py` validates it before any data loads,
so a mistake costs a second instead of a long read.
## Fields
| Field | Type | Default | Meaning |
|-------|------|---------|---------|
| `name` | string | required | Strategy label, carried into the result |
| `indicators` | object | required | Named indicators, at least one |
| `entry` | object | required | One condition; the position is held while it holds |
| `size` | number | `1.0` | Fraction of equity when in position, in `(0, 1]` |
| `stop_loss_pct` | number | none | Stop distance in percent from entry |
| `take_profit_pct` | number | none | Target distance in percent from entry |
| `fees_bps` | number | `0.0` | Fee in basis points, charged both sides |
| `slippage_bps` | number | `0.0` | Fixed slippage in basis points |
| `signal_delay` | integer | `1` | Bars between signal and fill |
## Indicators
```json
"indicators": {
"fast": { "type": "sma", "period": 20 },
"rsi14": { "type": "rsi", "period": 14, "source": "close" }
}
```
`type` is `sma`, `ema` or `rsi`. `period` is a positive integer. `source` is
`open`, `high`, `low` or `close`, defaulting to `close`.
Each name becomes a signal you can reference in `entry`.
## Entry condition
```json
"entry": { "left": "fast", "op": ">", "right": "slow" }
```
`op` is `>`, `<`, `>=` or `<=`. Each side is an indicator name, a price column,
or a number:
```json
"entry": { "left": "rsi14", "op": "<", "right": 30 }
"entry": { "left": "close", "op": ">", "right": "slow" }
```
The strategy holds `size` while the condition is true and goes flat when it
turns false. A crossing is the condition becoming true, so a moving-average
crossover is written as the comparison, not as a separate crossing operator.
## Why `signal_delay` defaults to 1
At 0 the strategy fills on the bar that produced the signal. Unless the fill
price is knowable at signal time, that is look-ahead, and it inflates results in
a way no later stress test recovers. `backtest-expert` lists it among its red
flags. Setting it to 0 is allowed and reported as a warning.
## Parameter count
`count_parameters` counts what the robustness dimension penalises:
- one per indicator `period`
- one per numeric threshold used in `entry`
- one when `size` is explicitly supplied
- one per bracket distance set
Fees, slippage and `signal_delay` are excluded. They are costs and conventions,
not knobs you tune until the equity curve improves.
A spec with more than four parameters triggers a warning before it runs.
## What validation refuses
| Refused | Reason |
|---------|--------|
| Unknown `type` | The three supported kinds are named in the error |
| `period` of 0, negative, or `True` | `True` is an `int` in Python and would build a period of one |
| `entry` naming an undeclared indicator | The error lists what is declared |
| `size` outside `(0, 1]` | Sizing above equity needs leverage the spec does not model |
| Negative `fees_bps` or `slippage_bps` | A negative cost is a rebate the engine does not model |
| Empty `indicators` | A condition needs something to compare |
## Full example
```json
{
"name": "rsi_oversold_costed",
"indicators": {
"rsi14": { "type": "rsi", "period": 14 }
},
"entry": { "left": "rsi14", "op": "<", "right": 30 },
"size": 0.5,
"stop_loss_pct": 2.0,
"take_profit_pct": 4.0,
"fees_bps": 5.0,
"slippage_bps": 2.0,
"signal_delay": 1
}
```
Five parameters: the RSI period, the threshold of 30, the explicit position
size, the stop and the target.
requirements.txt
# Runtime dependencies for standalone .skill installs (issue #330).
# Verified by scripts/check_skill_deps.py.
manifoldbt # unpinned: floor not verified in authoring env (pip install manifoldbt per SKILL.md)
pandas>=2.0 # floor verified pandas==2.3.3 on 2026-09-05
scripts/bridge.py
#!/usr/bin/env python3
"""Translate engine output into the eight inputs ``backtest-expert`` scores.
``skills/backtest-expert/scripts/evaluate_backtest.py`` grades a backtest on
five dimensions and asks for exactly eight numbers:
--total-trades --win-rate --avg-win-pct --avg-loss-pct
--max-drawdown-pct --years-tested --num-parameters --slippage-tested
Its own prerequisites say "metrics are user-provided", which leaves the user to
source those eight by hand. This module produces them from a completed run, so
the two skills chain instead of stopping at each other's edge.
Three conversions are not cosmetic and are the reason this is a module rather
than a dict comprehension:
* **Drawdown sign.** The engine reports ``max_drawdown`` as a negative fraction
(-0.38 for a 38% fall). The evaluator wants a positive percent and scores
worse as the number grows, so passing the raw value scores a deep drawdown as
if it were flawless.
* **Trade count.** The engine's ``total_trades`` counts *fills*, entries and
exits alike, so it runs at roughly twice the number of round trips. The
evaluator's sample-size dimension is about round trips; feeding it fills
doubles the apparent sample and can turn a below-threshold strategy into a
passing one.
* **Win rate basis.** Percentages come from the paired round trips rather than
the engine's own ``win_rate``, so the win rate, the average winner and the
average loser all describe the same population. Mixing sources here is how a
win rate ends up inconsistent with the averages beside it.
"""
from __future__ import annotations
import math
import shlex
from typing import Any
__all__ = ["build_evaluation_inputs", "format_evaluate_command", "NANOS_PER_YEAR"]
# Julian year in nanoseconds: the engine timestamps in ns, and the evaluator's
# robustness dimension counts calendar years of history.
NANOS_PER_YEAR = 365.25 * 24 * 60 * 60 * 1_000_000_000
def build_evaluation_inputs(
metrics: dict[str, Any],
trip_summary: dict[str, Any],
*,
start_ns: int | None = None,
end_ns: int | None = None,
num_parameters: int = 0,
slippage_tested: bool = False,
) -> dict[str, Any]:
"""Assemble the eight evaluator inputs from one completed run.
``metrics`` is the engine's metrics dict, ``trip_summary`` the output of
``round_trips.summarize_round_trips``. Returns the eight values plus a
``warnings`` list naming anything that will make the score meaningless.
"""
warnings_out: list[str] = []
scratches = int(trip_summary.get("scratches", 0))
if scratches:
raise ValueError(
f"cannot hand off {scratches} scratch trade(s): backtest-expert has no "
"scratch-rate input, so its derived expectancy would be wrong"
)
# Drawdown: stored negative, scored positive. Missing or non-finite risk
# cannot be represented honestly; zero would award an almost-perfect score.
dd = metrics.get("max_drawdown")
if dd is None:
raise ValueError("max_drawdown absent from engine metrics; refusing to score risk as 0%")
try:
dd_value = float(dd)
except (TypeError, ValueError) as exc:
raise ValueError(f"max_drawdown is not numeric: {dd!r}") from exc
if not math.isfinite(dd_value):
raise ValueError(f"max_drawdown must be finite, got {dd!r}")
dd_pct = abs(dd_value) * 100.0
total = int(trip_summary.get("total_trades", 0))
if total <= 0:
raise ValueError(
"no completed round trip: win rate and average returns are undefined; "
"refusing to generate an evaluator command"
)
if total < 30:
# Not an error. The evaluator scores this zero on sample size by design,
# and saying so up front beats the user reading a 0 and blaming the run.
warnings_out.append(
f"only {total} round trips; the sample-size dimension scores 0 below 30"
)
years = 0
if start_ns is not None and end_ns is not None and end_ns > start_ns:
years = int((end_ns - start_ns) / NANOS_PER_YEAR)
if years == 0:
warnings_out.append("backtest spans under a year; the robustness dimension scores 0")
if not slippage_tested:
warnings_out.append(
"no slippage or fees modelled; execution realism scores 0 and the "
"result should not be compared against a costed backtest"
)
# Reported so a caller can see the engine and the pairing agree. They are two
# independent computations over the same run, so a gap means one of them is
# wrong and the numbers should not be published until it is explained.
engine_win_rate = metrics.get("trade_stats", {}).get("win_rate")
if engine_win_rate is not None and total:
gap = abs(float(engine_win_rate) * 100.0 - trip_summary["win_rate_pct"])
if gap > 1.0:
warnings_out.append(
f"win rate from pairing ({trip_summary['win_rate_pct']:.1f}%) and from "
f"the engine ({float(engine_win_rate) * 100:.1f}%) differ by {gap:.1f} points; "
"an open final position or a different engine convention usually explains it"
)
return {
"total_trades": total,
"win_rate": round(trip_summary.get("win_rate_pct", 0.0), 4),
"avg_win_pct": round(trip_summary.get("avg_win_pct", 0.0), 4),
"avg_loss_pct": round(trip_summary.get("avg_loss_pct", 0.0), 4),
"max_drawdown_pct": round(dd_pct, 4),
"years_tested": years,
"num_parameters": int(num_parameters),
"slippage_tested": bool(slippage_tested),
"warnings": warnings_out,
}
def format_evaluate_command(
inputs: dict[str, Any],
script: str = "skills/backtest-expert/scripts/evaluate_backtest.py",
) -> str:
"""Render the ready-to-run ``evaluate_backtest.py`` invocation.
Emitted as a string rather than executed here: the two skills stay
independent, and the user sees the numbers being handed over instead of
having them pass behind their back.
"""
parts = [
"python3",
script,
"--total-trades",
str(inputs["total_trades"]),
"--win-rate",
str(inputs["win_rate"]),
"--avg-win-pct",
str(inputs["avg_win_pct"]),
"--avg-loss-pct",
str(inputs["avg_loss_pct"]),
"--max-drawdown-pct",
str(inputs["max_drawdown_pct"]),
"--years-tested",
str(inputs["years_tested"]),
"--num-parameters",
str(inputs["num_parameters"]),
]
if inputs["slippage_tested"]:
parts.append("--slippage-tested")
return " ".join(shlex.quote(p) for p in parts)
scripts/round_trips.py
#!/usr/bin/env python3
"""Pair a fill log into round trips carrying percent returns.
A backtest engine reports *fills*: one row per execution. The quality
framework in ``skills/backtest-expert`` asks for *round trips*: average winner
and average loser expressed in percent. Those are not the same number, and the
gap between them is where most hand-rolled evaluations go wrong.
Zipping consecutive rows and assuming buy and sell alternate holds for one
case: a single-symbol long-only strategy that never scales a position. Three
things break it, and none of them raises. Each produces percentages that look
plausible and are wrong:
* shorting is enabled, so a sell can *open* rather than close;
* the position is scaled in or out, so one exit answers several entries;
* the universe has more than one symbol, so fills interleave.
So this module tracks position per symbol instead of pattern-matching rows. A
fill in the direction of the running position opens or increases it, while a
fill against it reduces or closes it. Entry and exit cash flows accumulate for
the whole flat-to-flat lifecycle, so scaling out and then back in cannot mix the
remaining position's average entry with earlier exits. A round trip is emitted
when the position crosses back through flat, and a fill that flips the sign in
one go is split into the part that closes and the part that opens.
Deliberately dependency-free: plain dicts in, plain dicts out. It runs under the
repository's test job, which installs no dataframe library, and it lets the
pairing be tested without a backtest engine present.
"""
from __future__ import annotations
from collections.abc import Iterable
from typing import Any
__all__ = ["pair_round_trips", "summarize_round_trips", "SIDE_BUY", "SIDE_SELL"]
# Engine convention in the fill log: 1 buys, 2 sells. Which of the two *opens* a
# position depends on the position at that moment, never on the code alone,
# which is the whole reason this module tracks state.
SIDE_BUY = 1
SIDE_SELL = 2
# Positions below this magnitude count as flat. Sizing in fraction-of-equity
# terms leaves float dust behind on a close, and dust must not read as an open
# position or every subsequent round trip is attributed to the wrong entry.
_FLAT_EPS = 1e-9
def _signed_quantity(fill: dict[str, Any]) -> float:
"""Fill quantity, positive for a buy and negative for a sell."""
qty = abs(float(fill.get("quantity", 0.0)))
side = int(fill.get("side", SIDE_BUY))
return qty if side == SIDE_BUY else -qty
def _close_trip(symbol: Any, p: dict[str, Any], ts: Any) -> dict[str, Any]:
"""Emit one completed round trip from a position that has returned to flat.
Fees are charged against the notional put at risk, so the percentage answers
"what did this position return net of costs", which is the only figure a
trader can spend.
"""
entry_avg = p["entry_value"] / p["entry_qty"] if p["entry_qty"] else 0.0
exit_avg = p["exit_value"] / p["exit_qty"] if p["exit_qty"] else 0.0
side_label = "long" if p["entry_sign"] > 0 else "short"
gross_pnl = (
p["exit_value"] - p["entry_value"]
if side_label == "long"
else p["entry_value"] - p["exit_value"]
)
net_pnl = gross_pnl - p["fees"]
gross_return = gross_pnl / p["entry_value"] if p["entry_value"] else 0.0
net_return = net_pnl / p["entry_value"] if p["entry_value"] else 0.0
return {
"symbol_id": symbol,
"direction": side_label,
"quantity": p["entry_qty"],
"entry_qty": p["entry_qty"],
"exit_qty": p["exit_qty"],
"entry_value": p["entry_value"],
"exit_value": p["exit_value"],
"entry_price": entry_avg,
"exit_price": exit_avg,
"gross_pnl": gross_pnl,
"net_pnl": net_pnl,
"gross_return_pct": gross_return * 100.0,
"return_pct": net_return * 100.0,
"fees": p["fees"],
"total_fees": p["fees"],
"entry_timestamp": p["opened_at"],
"exit_timestamp": ts,
}
def pair_round_trips(fills: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
"""Turn a chronological fill log into a list of round trips.
Each fill must carry ``side``, ``quantity`` and ``fill_price``. ``symbol_id``,
``fees`` and ``execution_timestamp`` are used when present. Fills are assumed
to be in execution order, which is how every engine in this repository emits
them; they are not re-sorted, because a timestamp tie would then reorder
entries and exits arbitrarily.
Returns one dict per completed round trip:
``symbol_id``, ``direction`` (``"long"`` / ``"short"``), cumulative entry
and exit quantities / values, their display-average prices, gross and net
PnL / returns, fees, and the entry / exit timestamps.
``return_pct`` is **net of fees**, signed by direction, expressed against the
notional put at risk. Gross would be the easier number to compute and the
wrong one to report: at 7 bps a side, a trade that gains 0.1% on price is a
net loser, so a gross win rate counts as winners a population that lost
money. Since the quality framework derives expectancy from the win rate and
the average winner together, mixing a gross rate with net averages misstates
the edge. ``gross_return_pct`` is kept alongside for inspection.
An open position at the end of the log is dropped. It has no exit, so it has no
return, and counting it inflates the trade count the quality framework
scores.
"""
# Per symbol: running signed quantity, lifecycle entry/exit cash flows,
# accrued fees, and the timestamp the position opened.
state: dict[Any, dict[str, Any]] = {}
closed: list[dict[str, Any]] = []
for fill in fills:
symbol = fill.get("symbol_id", 0)
price = float(fill.get("fill_price", 0.0))
fee = float(fill.get("fees", 0.0) or 0.0)
ts = fill.get("execution_timestamp")
delta = _signed_quantity(fill)
if delta == 0.0:
continue
p = state.setdefault(
symbol,
{
"qty": 0.0,
"entry_qty": 0.0,
"entry_value": 0.0,
"entry_sign": 0.0,
"fees": 0.0,
"opened_at": None,
"exit_qty": 0.0,
"exit_value": 0.0,
},
)
remaining = delta
# A fill that flips the sign closes the old position first and opens the
# new one with what is left. Handling it as one event would book an entry
# price that never existed.
if p["qty"] != 0.0 and (p["qty"] > 0) != (remaining > 0):
closing = min(abs(remaining), abs(p["qty"]))
p["exit_qty"] += closing
p["exit_value"] += closing * price
p["fees"] += fee * (closing / abs(delta))
p["qty"] += closing if p["qty"] < 0 else -closing
remaining += closing if remaining < 0 else -closing
if abs(p["qty"]) <= _FLAT_EPS:
closed.append(_close_trip(symbol, p, ts))
p.update(
{
"qty": 0.0,
"entry_qty": 0.0,
"entry_value": 0.0,
"entry_sign": 0.0,
"fees": 0.0,
"opened_at": None,
"exit_qty": 0.0,
"exit_value": 0.0,
}
)
if remaining != 0.0:
if abs(p["qty"]) <= _FLAT_EPS:
p["entry_qty"] = abs(remaining)
p["entry_value"] = abs(remaining) * price
p["qty"] = remaining
p["entry_sign"] = 1.0 if remaining > 0 else -1.0
p["opened_at"] = ts
p["fees"] += fee * (abs(remaining) / abs(delta))
else:
# Keep the whole lifecycle's entry basis. Re-weighting only the
# remaining inventory here loses the basis of shares already
# sold and makes earlier exits incomparable with the entry.
p["entry_qty"] += abs(remaining)
p["entry_value"] += price * abs(remaining)
p["qty"] += remaining
p["fees"] += fee * (abs(remaining) / abs(delta))
return closed
def summarize_round_trips(trips: list[dict[str, Any]]) -> dict[str, Any]:
"""Aggregate round trips into the shape the quality framework scores.
A zero-return trip counts as neither a win nor a loss: it is a scratch. It
still belongs in the win-rate denominator because it is a completed trade
that did not win. The downstream evaluator has no scratch input, so the
bridge refuses to hand off a population containing scratches rather than
silently treating them as ordinary losses.
"""
gagnants = [t["return_pct"] for t in trips if t["return_pct"] > 0]
perdants = [t["return_pct"] for t in trips if t["return_pct"] < 0]
decides = len(gagnants) + len(perdants)
return {
"total_trades": len(trips),
"wins": len(gagnants),
"losses": len(perdants),
"scratches": len(trips) - decides,
# A scratch is not a win. Keeping it in the denominator makes the
# standalone summary honest even though it is not a loss either.
"win_rate_pct": (100.0 * len(gagnants) / len(trips)) if trips else 0.0,
"avg_win_pct": (sum(gagnants) / len(gagnants)) if gagnants else 0.0,
# Positive by convention: the quality framework asks for the magnitude.
"avg_loss_pct": abs(sum(perdants) / len(perdants)) if perdants else 0.0,
"total_fees": sum(t["fees"] for t in trips),
}
scripts/run_backtest.py
#!/usr/bin/env python3
"""Run a strategy spec through manifoldbt and emit what ``backtest-expert`` scores.
python3 run_backtest.py --spec strategy.json --data bars.csv --symbol BTCUSDT
Reads OHLCV bars, runs the spec, pairs the fill log into round trips, and prints
the eight inputs the quality framework asks for plus the ready-to-run command
that feeds them to it.
The engine is imported inside ``main`` rather than at module scope, so the
pairing and bridging modules stay testable without an engine installed, which is
how this repository's test job runs.
"""
from __future__ import annotations
import argparse
import json
import sys
import tempfile
from pathlib import Path
from typing import Any
SCRIPTS_DIR = Path(__file__).resolve().parent
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
from bridge import build_evaluation_inputs, format_evaluate_command # noqa: E402
from round_trips import pair_round_trips, summarize_round_trips # noqa: E402
from spec import count_parameters, describe_warnings, validate_spec # noqa: E402
REQUIRED_COLUMNS = ("timestamp", "open", "high", "low", "close", "volume")
def load_bars(path: str):
"""Read OHLCV bars from CSV or Parquet, and refuse anything ambiguous."""
import pandas as pd
p = Path(path)
if not p.exists():
raise SystemExit(f"data file not found: {path}")
df = pd.read_parquet(p) if p.suffix.lower() in {".parquet", ".pq"} else pd.read_csv(p)
missing = [c for c in REQUIRED_COLUMNS if c not in df.columns]
if missing:
raise SystemExit(
f"data is missing required column(s): {', '.join(missing)}. "
f"Found: {', '.join(map(str, df.columns))}"
)
df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
# Sorting is not a courtesy: the pairing walks the fill log in order, and an
# out-of-order bar would produce round trips that close before they open.
df = df.sort_values("timestamp").reset_index(drop=True)
if df["timestamp"].duplicated().any():
raise SystemExit(
"duplicate timestamps in the data; deduplicate before backtesting, "
"otherwise one bar is simulated twice"
)
return df
def build_strategy(cfg: dict[str, Any], bt, expr_mod, ind_mod):
"""Turn a validated spec into an engine strategy."""
col, lit, when = expr_mod.col, expr_mod.lit, expr_mod.when
builders = {"sma": ind_mod.sma, "ema": ind_mod.ema, "rsi": ind_mod.rsi}
columns = {
"close": ind_mod.close,
"open": ind_mod.open,
"high": ind_mod.high,
"low": ind_mod.low,
}
strategy = bt.Strategy.create(cfg["name"])
for name, body in cfg["indicators"].items():
source = columns[body["source"]]
strategy = strategy.signal(name, builders[body["type"]](source, body["period"]))
def operand(v):
if isinstance(v, (int, float)) and not isinstance(v, bool):
return lit(float(v))
return col(v) if v in cfg["indicators"] else col(v)
left, right = operand(cfg["entry"]["left"]), operand(cfg["entry"]["right"])
comparators = {
">": lambda a, b: a > b,
"<": lambda a, b: a < b,
">=": lambda a, b: a >= b,
"<=": lambda a, b: a <= b,
}
condition = comparators[cfg["entry"]["op"]](left, right)
strategy = strategy.size(when(condition, lit(cfg["size"]), lit(0.0)))
if cfg["stop_loss_pct"] is not None:
strategy = strategy.stop_loss(pct=cfg["stop_loss_pct"])
if cfg["take_profit_pct"] is not None:
strategy = strategy.take_profit(pct=cfg["take_profit_pct"])
return strategy
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--spec", required=True, help="Strategy spec, JSON")
ap.add_argument("--data", required=True, help="OHLCV bars, CSV or Parquet")
ap.add_argument("--symbol", default="ASSET", help="Symbol label for the store")
ap.add_argument("--interval", default="1m", help="Bar interval (default 1m)")
ap.add_argument("--capital", type=float, default=100_000.0)
ap.add_argument("--json-out", help="Write the full result to this path")
args = ap.parse_args()
raw_spec = json.loads(Path(args.spec).read_text(encoding="utf-8"))
cfg = validate_spec(raw_spec)
num_parameters = count_parameters(raw_spec)
try:
import manifoldbt as bt
from manifoldbt import expr as expr_mod
from manifoldbt import indicators as ind_mod
from manifoldbt.helpers import Interval, Slippage
except ImportError:
raise SystemExit(
"manifoldbt is not installed. Install it with: pip install manifoldbt"
) from None
df = load_bars(args.data)
tmp = tempfile.mkdtemp(prefix="mbt-skill-")
store = bt.import_dataframe(
df,
symbol=args.symbol,
symbol_id=1,
interval=args.interval,
data_root=tmp,
metadata_db=str(Path(tmp) / "meta.sqlite"),
)
start = int(df["timestamp"].iloc[0].value)
end = int(df["timestamp"].iloc[-1].value)
units = {"1m": Interval.minutes(1), "1h": Interval.hours(1), "1d": Interval.days(1)}
if args.interval not in units:
raise SystemExit(f"unsupported interval '{args.interval}'; use one of {list(units)}")
config = bt.BacktestConfig(
universe=[1],
time_range_start=start,
# The engine's range end is exclusive, so the last bar needs one extra
# unit or it is silently dropped from the simulation.
time_range_end=end + 86_400_000_000_000,
bar_interval=units[args.interval],
initial_capital=args.capital,
execution=bt.ExecutionConfig(
signal_delay=cfg["signal_delay"],
execution_price="AtClose",
max_position_pct=1.0,
allow_short=False,
position_sizing_mode="FractionOfEquity",
),
# One rate on both sides of the book. A crossover strategy takes
# liquidity in and out, so charging the taker rate throughout is the
# honest default; splitting maker and taker would need a fill model the
# spec does not carry.
fees=(
bt.FeeConfig.zero()
if cfg["fees_bps"] == 0
else bt.FeeConfig(maker_fee_bps=cfg["fees_bps"], taker_fee_bps=cfg["fees_bps"])
),
slippage=(
Slippage.none() if cfg["slippage_bps"] == 0 else Slippage.fixed_bps(cfg["slippage_bps"])
),
warmup_bars=0,
)
strategy = build_strategy(cfg, bt, expr_mod, ind_mod)
result = bt.run(strategy, config, store)
fills = result.trades_df().to_dict("records")
trips = pair_round_trips(fills)
summary = summarize_round_trips(trips)
inputs = build_evaluation_inputs(
result.metrics,
summary,
start_ns=start,
end_ns=end,
num_parameters=num_parameters,
slippage_tested=bool(cfg["slippage_bps"] or cfg["fees_bps"]),
)
command = format_evaluate_command(inputs)
print(f"Strategy : {cfg['name']}")
print(f"Bars : {len(df):,} Fills: {len(fills):,} Round trips: {len(trips):,}")
print(
f"Win rate : {inputs['win_rate']}% "
f"Avg win: {inputs['avg_win_pct']}% Avg loss: {inputs['avg_loss_pct']}%"
)
print(
f"Max DD : {inputs['max_drawdown_pct']}% "
f"Years: {inputs['years_tested']} Parameters: {inputs['num_parameters']}"
)
for w in describe_warnings(raw_spec) + inputs["warnings"]:
print(f" [warn] {w}")
print("\nHand off to backtest-expert:\n " + command)
if args.json_out:
Path(args.json_out).write_text(
json.dumps(
{
"spec": cfg,
"evaluation_inputs": inputs,
"round_trips": len(trips),
"command": command,
},
indent=2,
),
encoding="utf-8",
)
print(f"\nWrote {args.json_out}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/spec.py
#!/usr/bin/env python3
"""Validate and normalise a strategy spec before anything is run.
The spec is deliberately small and declarative: a strategy is a set of named
indicators plus one entry condition over them. That covers the shapes
``backtest-expert`` tells you to test first: a rule stated in one sentence,
carrying as few tunable parameters as possible. It refuses the rest instead of
half-supporting it.
Validation lives here, separate from execution, for two reasons. It runs under
the repository's test job, which has no backtest engine installed. And a spec
mistake should be reported before minutes of data loading, not after.
The parameter count matters beyond validation: ``backtest-expert`` scores
robustness partly on how many knobs a strategy has, and counting them by hand is
exactly the step a user skips or fudges. ``count_parameters`` does it from the
spec, so the number handed to the evaluator is the number the strategy actually
carries.
"""
from __future__ import annotations
from typing import Any
__all__ = ["validate_spec", "count_parameters", "SpecError", "INDICATORS", "OPERATORS", "SOURCES"]
# Indicator kinds this skill will build. Everything the engine offers is not
# here: the list is what can be expressed in a one-sentence rule, which is the
# scope the methodology skill asks for.
INDICATORS = frozenset({"sma", "ema", "rsi"})
OPERATORS = frozenset({">", "<", ">=", "<="})
SOURCES = frozenset({"open", "high", "low", "close"})
class SpecError(ValueError):
"""A spec that cannot be run, reported before any data is touched."""
def _require(cond: bool, message: str) -> None:
if not cond:
raise SpecError(message)
def _validate_operand(operand: Any, declared: dict[str, Any], side_name: str) -> None:
"""An operand is an indicator name, a raw price column, or a number."""
if isinstance(operand, (int, float)) and not isinstance(operand, bool):
return
_require(
isinstance(operand, str),
f"entry.{side_name} must be an indicator name, a price column, or a number, "
f"got {type(operand).__name__}",
)
_require(
operand in declared or operand in SOURCES,
f"entry.{side_name} references '{operand}', which is neither a declared "
f"indicator ({', '.join(sorted(declared)) or 'none'}) nor a price "
f"column ({', '.join(sorted(SOURCES))})",
)
def validate_spec(spec: dict[str, Any]) -> dict[str, Any]:
"""Check a spec and return it normalised, or raise ``SpecError``.
Defaults applied: ``size`` 1.0, ``fees_bps`` 0.0, ``slippage_bps`` 0.0,
``signal_delay`` 1. The delay defaults to one bar on purpose. Acting on the
bar that produced the signal is look-ahead, and the methodology skill treats
it as a red flag, not a setting.
"""
_require(isinstance(spec, dict), "spec must be a mapping")
_require(bool(spec.get("name")), "spec.name is required")
declared = spec.get("indicators") or {}
_require(isinstance(declared, dict), "spec.indicators must be a mapping")
_require(bool(declared), "spec.indicators must declare at least one indicator")
normalised: dict[str, Any] = {}
for name, body in declared.items():
_require(isinstance(body, dict), f"indicator '{name}' must be a mapping")
kind = body.get("type")
_require(
kind in INDICATORS,
f"indicator '{name}' has type '{kind}'; supported: {', '.join(sorted(INDICATORS))}",
)
period = body.get("period")
_require(
isinstance(period, int) and not isinstance(period, bool) and period > 0,
f"indicator '{name}' needs a positive integer 'period'",
)
source = body.get("source", "close")
_require(
source in SOURCES,
f"indicator '{name}' reads '{source}'; supported: {', '.join(sorted(SOURCES))}",
)
normalised[name] = {"type": kind, "period": period, "source": source}
entry = spec.get("entry")
_require(isinstance(entry, dict), "spec.entry is required and must be a mapping")
op = entry.get("op")
_require(op in OPERATORS, f"entry.op is '{op}'; supported: {', '.join(sorted(OPERATORS))}")
_require("left" in entry and "right" in entry, "entry needs both 'left' and 'right'")
_validate_operand(entry["left"], normalised, "left")
_validate_operand(entry["right"], normalised, "right")
size_pct = float(spec.get("size", 1.0))
_require(0.0 < size_pct <= 1.0, f"spec.size must be in (0, 1], got {size_pct}")
for field in ("stop_loss_pct", "take_profit_pct"):
value = spec.get(field)
if value is not None:
_require(
isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0,
f"spec.{field} must be a positive number when present",
)
for field in ("fees_bps", "slippage_bps"):
value = float(spec.get(field, 0.0))
_require(value >= 0.0, f"spec.{field} cannot be negative")
delay = spec.get("signal_delay", 1)
_require(
isinstance(delay, int) and not isinstance(delay, bool) and delay >= 0,
"spec.signal_delay must be a non-negative integer",
)
return {
"name": spec["name"],
"indicators": normalised,
"entry": {"left": entry["left"], "op": op, "right": entry["right"]},
"size": size_pct,
"stop_loss_pct": spec.get("stop_loss_pct"),
"take_profit_pct": spec.get("take_profit_pct"),
"fees_bps": float(spec.get("fees_bps", 0.0)),
"slippage_bps": float(spec.get("slippage_bps", 0.0)),
"signal_delay": delay,
}
def count_parameters(spec: dict[str, Any]) -> int:
"""Count the strategy's tunable knobs, for the robustness dimension.
Counted: one per indicator period, plus a numeric threshold used directly in
the entry condition, an explicitly supplied position size, plus each bracket
distance. Not counted: fees and slippage, which are costs the market imposes
rather than knobs to fit, and the signal delay, which is an execution
convention.
"""
n = len(spec.get("indicators") or {})
entry = spec.get("entry") or {}
for side_name in ("left", "right"):
v = entry.get(side_name)
if isinstance(v, (int, float)) and not isinstance(v, bool):
n += 1
if "size" in spec:
n += 1
n += sum(1 for c in ("stop_loss_pct", "take_profit_pct") if spec.get(c) is not None)
return n
def describe_warnings(spec: dict[str, Any]) -> list[str]:
"""Spec-level red flags worth saying before the run, not after."""
out: list[str] = []
if spec.get("signal_delay", 1) == 0:
out.append(
"signal_delay is 0: the strategy fills on the bar that produced the "
"signal, which is look-ahead unless the fill price is genuinely "
"knowable at signal time"
)
if not spec.get("fees_bps") and not spec.get("slippage_bps"):
out.append(
"neither fees nor slippage are set: the run is frictionless and the "
"execution-realism dimension will score 0"
)
if count_parameters(spec) > 4:
out.append(
f"{count_parameters(spec)} tunable parameters: the robustness "
"dimension penalises above 4, and every added knob makes an "
"in-sample fit easier to obtain by accident"
)
return out
scripts/tests/conftest.py
"""Shared fixtures for manifoldbt-backtester tests."""
import os
import sys
import pytest
# Ensure scripts/ is on sys.path so round_trips and bridge import cleanly.
SCRIPTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if SCRIPTS_DIR not in sys.path:
sys.path.insert(0, SCRIPTS_DIR)
@pytest.fixture
def fill():
"""Factory for one fill row, matching the engine's trade log columns.
Usage::
fill(side=1, price=100.0, qty=10)
"""
def _factory(side, price, qty=1.0, symbol_id=1, fees=0.0, ts=None):
return {
"side": side,
"quantity": qty,
"fill_price": price,
"symbol_id": symbol_id,
"fees": fees,
"execution_timestamp": ts,
}
return _factory
scripts/tests/test_bridge.py
"""Tests for the hand-off into ``backtest-expert``.
The three conversions tested first are the ones that fail silently: each
produces a plausible number that scores the strategy wrongly rather than
raising.
"""
import pytest
from bridge import NANOS_PER_YEAR, build_evaluation_inputs, format_evaluate_command
def _summary(**kw):
base = {
"total_trades": 200,
"wins": 100,
"losses": 100,
"scratches": 0,
"win_rate_pct": 50.0,
"avg_win_pct": 3.0,
"avg_loss_pct": 2.0,
"total_fees": 0.0,
}
base.update(kw)
return base
# ------------------------------------------------------- the silent-failure trio
def test_negative_drawdown_becomes_positive_percent():
"""Passing the engine's raw -0.38 would score a 38% fall as flawless."""
out = build_evaluation_inputs({"max_drawdown": -0.3846}, _summary())
assert out["max_drawdown_pct"] == 38.46
def test_trade_count_comes_from_round_trips_not_fills():
"""The engine's total_trades counts fills, about twice the round trips.
Handing fills to the sample-size dimension doubles the apparent sample.
"""
metrics = {"max_drawdown": -0.1, "trade_stats": {"total_trades": 400}}
out = build_evaluation_inputs(metrics, _summary(total_trades=200))
assert out["total_trades"] == 200
def test_percentages_all_come_from_the_same_pairing():
"""Win rate must describe the same population as the averages beside it."""
metrics = {"max_drawdown": -0.1, "trade_stats": {"win_rate": 0.50}}
out = build_evaluation_inputs(metrics, _summary(win_rate_pct=50.0))
assert out["win_rate"] == 50.0
assert not any("differ by" in w for w in out["warnings"])
# ------------------------------------------------------------------- disagreement
def test_win_rate_disagreement_is_reported_not_hidden():
metrics = {"max_drawdown": -0.1, "trade_stats": {"win_rate": 0.28}}
out = build_evaluation_inputs(metrics, _summary(win_rate_pct=50.0))
assert any("differ by" in w for w in out["warnings"])
# ------------------------------------------------------------------------ years
def test_years_derived_from_the_tested_span():
start = 1_600_000_000_000_000_000
out = build_evaluation_inputs(
{"max_drawdown": -0.1},
_summary(),
start_ns=start,
end_ns=start + int(NANOS_PER_YEAR * 5),
)
assert out["years_tested"] == 5
def test_span_under_a_year_is_flagged():
start = 1_600_000_000_000_000_000
out = build_evaluation_inputs(
{"max_drawdown": -0.1},
_summary(),
start_ns=start,
end_ns=start + int(NANOS_PER_YEAR * 0.5),
)
assert out["years_tested"] == 0
assert any("under a year" in w for w in out["warnings"])
# ---------------------------------------------------------------------- warnings
def test_no_round_trip_is_rejected_instead_of_scored():
with pytest.raises(ValueError, match="no completed round trip"):
build_evaluation_inputs({"max_drawdown": -0.1}, _summary(total_trades=0))
def test_thin_sample_is_named_before_the_evaluator_scores_it_zero():
out = build_evaluation_inputs({"max_drawdown": -0.1}, _summary(total_trades=12))
assert any("scores 0 below 30" in w for w in out["warnings"])
def test_frictionless_run_is_flagged():
out = build_evaluation_inputs({"max_drawdown": -0.1}, _summary(), slippage_tested=False)
assert any("no slippage" in w for w in out["warnings"])
def test_missing_drawdown_is_rejected_instead_of_scored_as_zero():
with pytest.raises(ValueError, match="max_drawdown absent"):
build_evaluation_inputs({}, _summary())
@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
def test_non_finite_drawdown_is_rejected(value):
with pytest.raises(ValueError, match="must be finite"):
build_evaluation_inputs({"max_drawdown": value}, _summary())
def test_non_numeric_drawdown_is_rejected():
with pytest.raises(ValueError, match="not numeric"):
build_evaluation_inputs({"max_drawdown": "unknown"}, _summary())
def test_scratch_population_is_rejected_before_expectancy_is_distorted():
with pytest.raises(ValueError, match="scratch trade"):
build_evaluation_inputs(
{"max_drawdown": -0.1},
_summary(
total_trades=100,
wins=1,
losses=0,
scratches=99,
win_rate_pct=1.0,
avg_win_pct=10.0,
avg_loss_pct=0.0,
),
)
# ----------------------------------------------------------------- command output
def test_command_carries_every_value():
out = build_evaluation_inputs(
{"max_drawdown": -0.2}, _summary(), num_parameters=3, slippage_tested=True
)
cmd = format_evaluate_command(out)
assert "--total-trades 200" in cmd
assert "--max-drawdown-pct 20.0" in cmd
assert "--num-parameters 3" in cmd
assert cmd.endswith("--slippage-tested")
def test_command_omits_the_flag_when_friction_was_not_modelled():
out = build_evaluation_inputs({"max_drawdown": -0.2}, _summary(), slippage_tested=False)
assert "--slippage-tested" not in format_evaluate_command(out)
scripts/tests/test_round_trips.py
"""Tests for the fill-to-round-trip pairing.
Every case here is one the naive "zip consecutive rows" approach gets wrong.
That is the point of the module, so it is the point of the tests.
"""
import pytest
from round_trips import SIDE_BUY, SIDE_SELL, pair_round_trips, summarize_round_trips
# ---------------------------------------------------------------- the simple case
def test_long_round_trip_returns_price_move(fill):
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10),
fill(SIDE_SELL, 110.0, 10),
]
)
assert len(trips) == 1
assert trips[0]["direction"] == "long"
assert trips[0]["return_pct"] == 10.0
assert trips[0]["entry_price"] == 100.0
assert trips[0]["exit_price"] == 110.0
def test_open_position_at_end_is_dropped(fill):
"""An entry with no exit has no return, and must not inflate the count."""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10),
fill(SIDE_SELL, 110.0, 10),
fill(SIDE_BUY, 105.0, 10),
]
)
assert len(trips) == 1
# --------------------------------------------------------- what naive zipping breaks
def test_short_round_trip_is_signed_by_direction(fill):
"""A sell that OPENS is a short: a price fall is a gain, not a loss.
Zipping rows in pairs and computing (exit - entry) / entry would report
-10% here, sign-flipping the entire short side of a strategy.
"""
trips = pair_round_trips(
[
fill(SIDE_SELL, 100.0, 10),
fill(SIDE_BUY, 90.0, 10),
]
)
assert len(trips) == 1
assert trips[0]["direction"] == "short"
assert trips[0]["return_pct"] == 10.0
def test_scaling_in_uses_weighted_average_entry(fill):
"""Two entries then one exit is ONE round trip, not two.
Naive pairing reads (buy, buy) as a trip and leaves the exit dangling.
"""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10),
fill(SIDE_BUY, 200.0, 10),
fill(SIDE_SELL, 180.0, 20),
]
)
assert len(trips) == 1
assert trips[0]["entry_price"] == 150.0 # weighted, not last, not first
assert trips[0]["return_pct"] == 20.0
assert trips[0]["quantity"] == 20
def test_scaling_out_closes_once_flat(fill):
"""One entry answered by two exits is still one round trip."""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 20),
fill(SIDE_SELL, 110.0, 10),
fill(SIDE_SELL, 130.0, 10),
]
)
assert len(trips) == 1
assert trips[0]["exit_price"] == 120.0 # weighted across both exits
assert trips[0]["return_pct"] == 20.0
def test_scaling_out_then_back_in_uses_whole_lifecycle_cash_flows(fill):
"""A re-add must not mix the remaining-position entry with all exits."""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10),
fill(SIDE_SELL, 110.0, 4),
fill(SIDE_BUY, 120.0, 5),
fill(SIDE_SELL, 130.0, 11),
]
)
assert len(trips) == 1
trip = trips[0]
assert trip["entry_qty"] == pytest.approx(15.0)
assert trip["exit_qty"] == pytest.approx(15.0)
assert trip["entry_value"] == pytest.approx(1_600.0)
assert trip["exit_value"] == pytest.approx(1_870.0)
assert trip["entry_price"] == pytest.approx(1_600.0 / 15.0)
assert trip["exit_price"] == pytest.approx(1_870.0 / 15.0)
assert trip["gross_pnl"] == pytest.approx(270.0)
assert trip["net_pnl"] == pytest.approx(270.0)
assert trip["gross_return_pct"] == pytest.approx(16.875)
assert trip["return_pct"] == pytest.approx(16.875)
def test_scaling_out_then_back_in_cannot_flip_a_loser_to_a_win(fill):
"""Regression for the silent sign flip reported in PR #325 review."""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10),
fill(SIDE_SELL, 50.0, 9),
fill(SIDE_BUY, 50.0, 100),
fill(SIDE_SELL, 51.0, 101),
]
)
assert len(trips) == 1
trip = trips[0]
assert trip["entry_value"] == pytest.approx(6_000.0)
assert trip["exit_value"] == pytest.approx(5_601.0)
assert trip["gross_pnl"] == pytest.approx(-399.0)
assert trip["return_pct"] == pytest.approx(-6.65)
summary = summarize_round_trips(trips)
assert summary["wins"] == 0
assert summary["losses"] == 1
def test_short_scaling_out_then_back_in_uses_whole_lifecycle_cash_flows(fill):
trips = pair_round_trips(
[
fill(SIDE_SELL, 100.0, 10),
fill(SIDE_BUY, 90.0, 4),
fill(SIDE_SELL, 80.0, 5),
fill(SIDE_BUY, 70.0, 11),
]
)
assert len(trips) == 1
trip = trips[0]
assert trip["entry_qty"] == pytest.approx(15.0)
assert trip["exit_qty"] == pytest.approx(15.0)
assert trip["entry_value"] == pytest.approx(1_400.0)
assert trip["exit_value"] == pytest.approx(1_130.0)
assert trip["gross_pnl"] == pytest.approx(270.0)
assert trip["gross_return_pct"] == pytest.approx(100.0 * 270.0 / 1_400.0)
def test_reversal_splits_into_close_then_open(fill):
"""A fill that crosses through flat closes one trip and opens the next.
Booking it as a single event would invent an entry price that never traded.
"""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10),
fill(SIDE_SELL, 110.0, 20), # closes the long, opens a short of 10
fill(SIDE_BUY, 99.0, 10), # closes the short
]
)
assert len(trips) == 2
assert trips[0]["direction"] == "long"
assert trips[0]["return_pct"] == 10.0
assert trips[1]["direction"] == "short"
assert trips[1]["entry_price"] == 110.0
assert round(trips[1]["return_pct"], 6) == 10.0
def test_symbols_do_not_contaminate_each_other(fill):
"""Interleaved fills across a universe must pair per symbol."""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 1, symbol_id=1),
fill(SIDE_BUY, 50.0, 1, symbol_id=2),
fill(SIDE_SELL, 110.0, 1, symbol_id=1),
fill(SIDE_SELL, 45.0, 1, symbol_id=2),
]
)
assert len(trips) == 2
par_sym = {t["symbol_id"]: t["return_pct"] for t in trips}
assert par_sym[1] == 10.0
assert par_sym[2] == -10.0
def test_float_dust_on_close_reads_as_flat(fill):
"""Fraction-of-equity sizing leaves dust; dust must not read as still open."""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10.0),
fill(SIDE_SELL, 110.0, 10.0 - 1e-12),
]
)
assert len(trips) == 1
# ------------------------------------------------------------------ net of fees
def test_return_is_net_of_fees_and_gross_is_kept_alongside(fill):
"""10% on price, 1% of notional in fees, is a 9% trade."""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10, fees=5.0),
fill(SIDE_SELL, 110.0, 10, fees=5.0),
]
)
assert trips[0]["gross_return_pct"] == 10.0
assert round(trips[0]["return_pct"], 6) == 9.0
assert trips[0]["fees"] == 10.0
def test_a_gross_winner_eaten_by_fees_counts_as_a_loser(fill):
"""The case that makes gross the wrong basis.
Price moved in the strategy's favour, the trade still lost money. Counting
it as a win inflates the win rate and, with it, the expectancy the quality
framework computes.
"""
trips = pair_round_trips(
[
fill(SIDE_BUY, 100.0, 10, fees=5.0),
fill(SIDE_SELL, 100.1, 10, fees=5.0),
]
)
assert trips[0]["gross_return_pct"] > 0
assert trips[0]["return_pct"] < 0
s = summarize_round_trips(trips)
assert s["wins"] == 0
assert s["losses"] == 1
def test_short_return_is_also_net(fill):
trips = pair_round_trips(
[
fill(SIDE_SELL, 100.0, 10, fees=5.0),
fill(SIDE_BUY, 90.0, 10, fees=5.0),
]
)
assert trips[0]["gross_return_pct"] == 10.0
assert round(trips[0]["return_pct"], 6) == 9.0
# ------------------------------------------------------------------- the aggregate
def test_summary_reports_loss_magnitude_positive(fill):
"""The quality framework asks for the magnitude of the average loser."""
s = summarize_round_trips(
pair_round_trips(
[
fill(SIDE_BUY, 100.0, 1),
fill(SIDE_SELL, 90.0, 1),
]
)
)
assert s["avg_loss_pct"] == 10.0
assert s["losses"] == 1
assert s["avg_win_pct"] == 0.0
def test_scratch_trades_remain_in_win_rate_denominator(fill):
"""A flat trade is neither a win nor a loss.
It is still a completed trade that did not win, so omitting it from the
denominator would overstate the observed win rate.
"""
s = summarize_round_trips(
pair_round_trips(
[
fill(SIDE_BUY, 100.0, 1),
fill(SIDE_SELL, 110.0, 1), # win
fill(SIDE_BUY, 100.0, 1),
fill(SIDE_SELL, 100.0, 1), # scratch
]
)
)
assert s["total_trades"] == 2
assert s["scratches"] == 1
assert s["win_rate_pct"] == 50.0
def test_empty_log_summarizes_without_dividing_by_zero():
s = summarize_round_trips(pair_round_trips([]))
assert s["total_trades"] == 0
assert s["win_rate_pct"] == 0.0
assert s["avg_win_pct"] == 0.0
assert s["avg_loss_pct"] == 0.0
scripts/tests/test_run_backtest.py
"""Unit tests for the engine adapter that do not require manifoldbt."""
from types import SimpleNamespace
import pandas as pd
import pytest
from run_backtest import build_strategy, load_bars
def _bars(**overrides):
values = {
"timestamp": ["2026-01-02T00:00:00Z", "2026-01-01T00:00:00Z"],
"open": [101.0, 100.0],
"high": [102.0, 101.0],
"low": [100.0, 99.0],
"close": [101.5, 100.5],
"volume": [20.0, 10.0],
}
values.update(overrides)
return pd.DataFrame(values)
def test_load_bars_sorts_csv_by_timestamp(tmp_path):
path = tmp_path / "bars.csv"
_bars().to_csv(path, index=False)
loaded = load_bars(str(path))
assert loaded["timestamp"].is_monotonic_increasing
assert list(loaded["volume"]) == [10.0, 20.0]
def test_load_bars_rejects_a_missing_file(tmp_path):
with pytest.raises(SystemExit, match="data file not found"):
load_bars(str(tmp_path / "missing.csv"))
def test_load_bars_names_missing_columns(tmp_path):
path = tmp_path / "bars.csv"
_bars().drop(columns=["volume", "low"]).to_csv(path, index=False)
with pytest.raises(SystemExit, match="low, volume"):
load_bars(str(path))
def test_load_bars_rejects_duplicate_timestamps(tmp_path):
path = tmp_path / "bars.csv"
_bars(timestamp=["2026-01-01T00:00:00Z"] * 2).to_csv(path, index=False)
with pytest.raises(SystemExit, match="duplicate timestamps"):
load_bars(str(path))
class _Expression:
def __init__(self, value):
self.value = value
def _compare(self, operator, other):
return (operator, self.value, other.value)
def __gt__(self, other):
return self._compare(">", other)
def __lt__(self, other):
return self._compare("<", other)
def __ge__(self, other):
return self._compare(">=", other)
def __le__(self, other):
return self._compare("<=", other)
class _Strategy:
def __init__(self, name):
self.calls = [("create", name)]
def signal(self, name, expression):
self.calls.append(("signal", name, expression))
return self
def size(self, expression):
self.calls.append(("size", expression))
return self
def stop_loss(self, *, pct):
self.calls.append(("stop_loss", pct))
return self
def take_profit(self, *, pct):
self.calls.append(("take_profit", pct))
return self
class _StrategyFactory:
@staticmethod
def create(name):
return _Strategy(name)
def test_build_strategy_maps_indicators_condition_and_brackets():
bt = SimpleNamespace(Strategy=_StrategyFactory)
expr = SimpleNamespace(
col=lambda value: _Expression(("col", value)),
lit=lambda value: _Expression(("lit", value)),
when=lambda condition, yes, no: ("when", condition, yes.value, no.value),
)
indicators = SimpleNamespace(
close="close",
open="open",
high="high",
low="low",
sma=lambda source, period: ("sma", source, period),
ema=lambda source, period: ("ema", source, period),
rsi=lambda source, period: ("rsi", source, period),
)
cfg = {
"name": "adapter_contract",
"indicators": {
"fast": {"type": "sma", "source": "close", "period": 5},
"slow": {"type": "ema", "source": "open", "period": 10},
"momentum": {"type": "rsi", "source": "high", "period": 14},
},
"entry": {"left": "momentum", "op": ">", "right": 30},
"size": 0.5,
"stop_loss_pct": 2.0,
"take_profit_pct": 4.0,
}
strategy = build_strategy(cfg, bt, expr, indicators)
assert strategy.calls[:4] == [
("create", "adapter_contract"),
("signal", "fast", ("sma", "close", 5)),
("signal", "slow", ("ema", "open", 10)),
("signal", "momentum", ("rsi", "high", 14)),
]
assert strategy.calls[4][0] == "size"
assert strategy.calls[-2:] == [("stop_loss", 2.0), ("take_profit", 4.0)]
scripts/tests/test_spec.py
"""Tests for spec validation and the parameter count.
Validation exists to fail before the data is loaded, so the tests are mostly
about rejecting things clearly rather than accepting things quietly.
"""
import pytest
from spec import SpecError, count_parameters, describe_warnings, validate_spec
def _spec(**kw):
base = {
"name": "sma_cross",
"indicators": {
"fast": {"type": "sma", "period": 20},
"slow": {"type": "sma", "period": 60},
},
"entry": {"left": "fast", "op": ">", "right": "slow"},
}
base.update(kw)
return base
# ------------------------------------------------------------------ acceptance
def test_minimal_spec_normalises_with_defaults():
out = validate_spec(_spec())
assert out["size"] == 1.0
assert out["fees_bps"] == 0.0
# One bar, not zero: filling on the signal bar is look-ahead.
assert out["signal_delay"] == 1
assert out["indicators"]["fast"]["source"] == "close"
def test_numeric_threshold_is_accepted_as_an_operand():
out = validate_spec(
_spec(
indicators={"rsi": {"type": "rsi", "period": 14}},
entry={"left": "rsi", "op": "<", "right": 30},
)
)
assert out["entry"]["right"] == 30
def test_price_column_is_accepted_as_an_operand():
out = validate_spec(_spec(entry={"left": "close", "op": ">", "right": "slow"}))
assert out["entry"]["left"] == "close"
# ------------------------------------------------------------------- rejection
def test_unknown_indicator_reference_is_named():
with pytest.raises(SpecError, match="neither a declared indicator"):
validate_spec(_spec(entry={"left": "typo", "op": ">", "right": "slow"}))
def test_unsupported_indicator_type_lists_what_is_supported():
with pytest.raises(SpecError, match="supported: ema, rsi, sma"):
validate_spec(_spec(indicators={"x": {"type": "kalman", "period": 5}}))
def test_non_positive_period_is_rejected():
with pytest.raises(SpecError, match="positive integer"):
validate_spec(_spec(indicators={"x": {"type": "sma", "period": 0}}))
def test_boolean_period_is_not_an_integer():
"""`True` is an int in Python; a spec that says True must not build a period."""
with pytest.raises(SpecError, match="positive integer"):
validate_spec(_spec(indicators={"x": {"type": "sma", "period": True}}))
def test_size_outside_the_unit_interval_is_rejected():
with pytest.raises(SpecError, match="must be in"):
validate_spec(_spec(size=1.5))
def test_empty_indicator_block_is_rejected():
with pytest.raises(SpecError, match="at least one indicator"):
validate_spec(_spec(indicators={}))
def test_negative_cost_is_rejected():
with pytest.raises(SpecError, match="cannot be negative"):
validate_spec(_spec(fees_bps=-1))
# ------------------------------------------------------------- parameter count
def test_parameters_count_indicator_periods():
assert count_parameters(_spec()) == 2
def test_numeric_threshold_counts_as_a_parameter():
s = _spec(
indicators={"rsi": {"type": "rsi", "period": 14}},
entry={"left": "rsi", "op": "<", "right": 30},
)
assert count_parameters(s) == 2 # the period and the threshold
def test_bracket_distances_count_as_parameters():
assert count_parameters(_spec(stop_loss_pct=1.0, take_profit_pct=2.0)) == 4
def test_explicit_position_size_counts_as_a_parameter():
assert count_parameters(_spec(size=0.25)) == 3
def test_default_position_size_is_not_counted_when_omitted():
assert count_parameters(_spec()) == 2
def test_costs_are_not_parameters():
"""Fees are imposed by the market, not fitted; counting them would flatter
nothing and would misreport how much freedom the strategy really has."""
assert count_parameters(_spec(fees_bps=5, slippage_bps=2)) == 2
# ---------------------------------------------------------------- pre-warnings
def test_zero_delay_is_flagged_as_look_ahead():
assert any("look-ahead" in w for w in describe_warnings(_spec(signal_delay=0)))
def test_frictionless_spec_is_flagged_before_running():
assert any("frictionless" in w for w in describe_warnings(_spec()))
def test_knob_count_above_four_is_flagged():
s = _spec(
indicators={f"i{i}": {"type": "sma", "period": 10 + i} for i in range(4)},
stop_loss_pct=1.0,
take_profit_pct=2.0,
)
assert any("tunable parameters" in w for w in describe_warnings(s))
SKILL.md
---
name: manifoldbt-backtester
description: Runs a declarative strategy spec over OHLCV bars with the manifoldbt Rust engine, pairs the fill log into round trips, and emits the eight inputs the backtest-expert skill scores. Use when the user wants to execute a backtest, measure a rule they have described, obtain win rate / average win / average loss / max drawdown from real bars, or feed backtest-expert with measured numbers instead of estimates.
---
# manifoldbt Backtester Skill
## Purpose
Execute what `backtest-expert` teaches. That skill grades a backtest on five
dimensions, and its prerequisites say "metrics are user-provided": it scores
numbers it never produces. This skill produces them. It runs a strategy over
real bars and returns the eight inputs its evaluator asks for.
The two chain in one direction: spec, run, evaluate.
## When to Use This Skill
- A user describes a rule and wants it measured
- `backtest-expert` is about to run and the numbers do not exist yet
- A win rate, average winner, average loser or drawdown must come from bars
- A strategy's parameter count must be established for scoring
Leave the verdict to `backtest-expert`. It owns the thresholds and the red
flags, and this skill does not duplicate them.
## Prerequisites
- Python 3.9+
- `pip install manifoldbt` (the Python package is Apache 2.0, the compiled
engine ships under the free Manifold-BT Engine License; the free tier covers
everything this skill does)
- OHLCV bars as CSV or Parquet with columns `timestamp, open, high, low, close, volume`
- No API key required
## Workflow
### 1. Write the strategy spec
A spec names indicators and one entry condition. Keep it to the smallest rule
that states the hypothesis. Every added knob makes an in-sample fit easier to
reach by accident, and the evaluator penalises the count.
```json
{
"name": "sma_cross_costed",
"indicators": {
"fast": { "type": "sma", "period": 20 },
"slow": { "type": "sma", "period": 60 }
},
"entry": { "left": "fast", "op": ">", "right": "slow" },
"size": 1.0,
"stop_loss_pct": 1.5,
"fees_bps": 5.0,
"slippage_bps": 2.0
}
```
Field reference: `references/strategy_spec.md`.
Set `fees_bps` and `slippage_bps` to realistic values before you read any
result. A frictionless run scores 0 on execution realism, and over short holding
periods costs decide whether an edge survives.
### 2. Run it
```bash
python3 skills/manifoldbt-backtester/scripts/run_backtest.py \
--spec strategy.json \
--data bars.csv \
--symbol BTCUSDT \
--json-out result.json
```
The script validates the spec before it touches the data, so you see a spec
mistake in a second instead of after a long load.
### 3. Read the warnings before the numbers
The run prints warnings that change how you should read the result: a sample
under 30 trades, a span under a year, no friction modelled, or a gap between the
engine's win rate and the paired one. Each one is a reason to fix the setup and
run again.
Three conditions stop the handoff instead of producing a score: no completed
round trips, missing or non-finite maximum drawdown, and scratch trades. The
evaluator has no scratch input, so passing a population that contains them would
make its derived expectancy disagree with the completed trades.
### 4. Hand off to backtest-expert
The run ends with a command you can paste. Run it, or invoke the
`backtest-expert` skill with the same figures:
```bash
python3 skills/backtest-expert/scripts/evaluate_backtest.py \
--total-trades 3854 --win-rate 20.24 \
--avg-win-pct 0.2917 --avg-loss-pct 0.2342 \
--max-drawdown-pct 99.2893 --years-tested 0 \
--num-parameters 3 --slippage-tested
```
## Four conversions that fail without an error
Between an engine's output and the evaluator's inputs sit four conversions. Each
one yields a plausible number and scores the strategy wrongly. None of them
raises.
**A fill is one execution, a round trip is two.** The raw trade count runs at
about twice the number of round trips. Feed fills to the sample-size dimension
and you double the apparent sample, which can lift a thin backtest over a
threshold it should not clear.
**Buy and sell alternate only in the simplest case.** That holds for a
single-symbol long-only strategy that never scales a position. Shorting breaks
it, because a sell can open. Scaling breaks it, because one exit answers several
entries. A universe breaks it, because fills interleave. This skill tracks
position per symbol and closes a trip when it crosses back through flat. Entry
and exit quantities and cash values accumulate across that whole lifecycle;
their weighted-average prices are display values, while PnL comes from the cash
flows themselves.
**Costs decide small trades.** At 7 bps a side, a trade that gains 0.1% on price
loses money. Expectancy comes from the win rate and the average winner together,
so a gross win rate beside net averages misstates the edge. Percentages here are
net of fees, and `gross_return_pct` sits alongside for inspection.
**The engine signs drawdown negative.** The evaluator wants a positive
magnitude. Pass the raw value and a 38% fall scores as a flawless run.
## Scope
Supported: `sma`, `ema`, `rsi` over any OHLC column; one entry condition using
`>`, `<`, `>=`, `<=` against another indicator, a price column or a number;
optional stop-loss and take-profit; fees and slippage in basis points;
long-only.
Refused: multi-condition entries, shorting, multi-asset universes, and
indicators outside the three above. The engine does all of these. This skill
covers the shapes a one-sentence hypothesis produces, and rejects the rest
instead of half-handling it.
## Reference Files
- `references/strategy_spec.md` covers every spec field, its default, and what
validation refuses
- `references/metric_bridge.md` covers the eight inputs, how each is derived,
and the trap in each conversion
## Scripts
- `scripts/run_backtest.py` runs a spec against bars
- `scripts/spec.py` validates a spec and counts its parameters
- `scripts/round_trips.py` pairs fills into round trips with net returns
- `scripts/bridge.py` assembles the evaluator's eight inputs
`spec.py`, `round_trips.py` and `bridge.py` carry no dependencies and import
without the engine, so you can test the logic without running a backtest.