Home / Libraries / ML4T Backtest / Docs
ML4T Backtest
ML4T Backtest Documentation
Event-driven backtesting with realistic execution
Source on GitHub
Skip to content

API Reference

Auto-generated from source docstrings.

Core

Engine 

Engine(
    feed,
    strategy,
    config=None,
    *,
    contract_specs=None,
    market_impact_model=None,
    execution_limits=None,
    funding_df=None,
    lifecycle_version=V1,
    execution_policy=None,
    target_intent_state=None,
)

Event-driven backtesting engine.

The Engine orchestrates the backtest by iterating through market data, managing the broker, and calling the strategy on each bar. Engine instances are single-use; create a new instance for each run.

Execution Flow
  1. Call on_start before any market bar is registered.
  2. Call on_prepare with the resolved config and no future feed data.
  3. For each accepted session bar, register data, process eligible deferred orders and risk, call the per-bar strategy callbacks, process configured current-bar orders, and record marked portfolio state.
  4. Call on_end after the final timestamp.
  5. Return closed trades plus marked open positions. The engine does not submit automatic end-of-data liquidation orders.

Attributes:

Name Type Description
feed

DataFeed providing price and signal data
strategy

Strategy implementing trading logic
broker

Broker handling order execution and positions
config

BacktestConfig with all behavioral settings
equity_curve list[tuple[datetime, float]]

List of (timestamp, equity) tuples
Example

from ml4t.backtest import Engine, DataFeed, Strategy, BacktestConfig

class MyStrategy(Strategy): ... def on_data(self, timestamp, data, context, broker): ... for asset, bar in data.items(): ... if bar.get('signal', 0) > 0.5: ... broker.submit_order(asset, 100)

feed = DataFeed(prices_df=df) engine = Engine(feed=feed, strategy=MyStrategy()) result = engine.run() print(result['total_return'])

Source code in src/ml4t/backtest/engine.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def __init__(
    self,
    feed: DataFeed,
    strategy: Strategy,
    config: BacktestConfig | None = None,
    *,
    contract_specs: dict[str, Any] | None = None,
    market_impact_model: Any | None = None,
    execution_limits: Any | None = None,
    funding_df: pl.DataFrame | None = None,
    lifecycle_version: LifecycleVersion | str = LifecycleVersion.V1,
    execution_policy: ExecutionPolicy | None = None,
    target_intent_state: dict[str, Any] | None = None,
):
    from .config import BacktestConfig as ConfigCls

    negotiated_version = negotiate_lifecycle_version(lifecycle_version)
    self._validate_strategy_lifecycle(strategy)
    if config is None:
        config = ConfigCls()

    self.feed = feed
    self.strategy = strategy
    self.config = config.merge_feed_spec(getattr(feed, "feed_spec", None))
    self.execution_mode = self.config.execution_mode
    if (
        getattr(feed, "session_col", None) is not None
        and self.execution_mode != ExecutionMode.NEXT_BAR
    ):
        raise ValueError("session decisions require NEXT_BAR execution")
    self.lifecycle_version = negotiated_version
    self.execution_policy = execution_policy or default_execution_policy(self.config)
    self.broker = Broker.from_config(
        self.config,
        contract_specs=contract_specs,
        market_impact_model=market_impact_model,
        execution_limits=execution_limits,
    )
    if funding_df is None:
        self._funding_events = {}
    else:
        prices_frame = feed.prices
        assert prices_frame is not None
        self._funding_events = index_funding_events(
            funding_df,
            list(feed.timestamps),
            set(prices_frame[feed._entity_col].unique().to_list()),
        )
    self.funding_payments: list[FundingPayment] = []
    if self.broker.execution_price is ExecutionPrice.VWAP and not getattr(
        self.feed.feed_spec, "vwap_col", None
    ):
        raise ValueError(
            "execution_price is VWAP but the feed declares no VWAP column. Set "
            "FeedSpec.vwap_col to the column holding the volume-weighted average "
            "price. This is checked here, once, rather than at fill time: a missing "
            "column is a configuration error, while an individual bar with no VWAP is "
            "an ordinary no-trade bar that leaves its order unfilled."
        )
    self.equity_curve: list[tuple[datetime, float]] = []
    self.portfolio_state: list[tuple[datetime, float, float, float, float, int]] = []
    self.lifecycle_dispatcher = LifecycleDispatcher(
        strategy,
        LIFECYCLE_V1,
        retain_invocations=self.config.retain_lifecycle_history,
    )
    self.preopen_target_manager = self.broker._create_preopen_target_manager(
        self.execution_policy,
        self.lifecycle_version,
        calendar=self.config.resolved_calendar,
        timezone=self.config.resolved_timezone,
        session_start_time=self.config.resolved_session_start_time,
        data_frequency=self.config.resolved_data_frequency,
        timestamp_semantics=self.config.resolved_timestamp_semantics,
    )
    if target_intent_state is not None:
        self.preopen_target_manager.restore_state(target_intent_state)
    self._strategy_finalized = False
    self._accepted_market_event_count = 0

    # Calendar session enforcement (lazy initialized in run())
    self._calendar = None
    self._skipped_bars = 0
    self._has_run = False

run 

run()

Run backtest and return structured results.

Returns:

Type Description
BacktestResult

BacktestResult with trades, equity curve, metrics, and export methods.
BacktestResult

Call .to_dict() for backward-compatible dictionary output.

Raises:

Type Description
RuntimeError

If a run was already started on this Engine instance.
Source code in src/ml4t/backtest/engine.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def run(self) -> BacktestResult:
    """Run backtest and return structured results.

    Returns:
        BacktestResult with trades, equity curve, metrics, and export methods.
        Call .to_dict() for backward-compatible dictionary output.

    Raises:
        RuntimeError: If a run was already started on this Engine instance.
    """
    if self._has_run:
        raise RuntimeError("Engine.run() was already started; create a new Engine for each run")
    self._has_run = True
    try:
        return self._run_once()
    except BaseException as failure:
        try:
            self._finalize_strategy()
        except BaseException as finalization_failure:
            failure.add_note(
                "on_end also failed during cleanup: "
                f"{type(finalization_failure).__name__}: {finalization_failure}"
            )
        raise

run_dict 

run_dict()

Run backtest and return dictionary (backward compatible).

This is equivalent to run().to_dict() but more explicit for code that requires dictionary output.

Returns:

Type Description
dict[str, Any]

Dictionary with metrics, trades, and equity curve.

Raises:

Type Description
RuntimeError

If a run was already started on this Engine instance.
Source code in src/ml4t/backtest/engine.py 
584
585
586
587
588
589
590
591
592
593
594
595
596
def run_dict(self) -> dict[str, Any]:
    """Run backtest and return dictionary (backward compatible).

    This is equivalent to run().to_dict() but more explicit for code
    that requires dictionary output.

    Returns:
        Dictionary with metrics, trades, and equity curve.

    Raises:
        RuntimeError: If a run was already started on this Engine instance.
    """
    return self.run().to_dict()

from_config classmethod 

from_config(
    feed,
    strategy,
    config,
    *,
    contract_specs=None,
    market_impact_model=None,
    execution_limits=None,
    funding_df=None,
    lifecycle_version=V1,
    execution_policy=None,
    target_intent_state=None,
)

Create an Engine instance from a BacktestConfig.

Equivalent to Engine(feed, strategy, config). Kept as a convenience for code that reads more clearly with a named constructor.

Parameters:

Name Type Description Default
feed DataFeed

DataFeed with price data

 required
strategy Strategy

Strategy to execute

 required
config BacktestConfig

BacktestConfig with all behavioral settings

 required
contract_specs dict[str, Any] | None

Per-asset contract specifications (futures multipliers, etc.)

 None
market_impact_model Any | None

Market impact model for fill simulation

 None
execution_limits Any | None

Execution limits (max order size, etc.)

 None
funding_df DataFrame | None

Timestamped funding rates or amounts for named assets

 None

Returns:

Type Description
Engine

Configured Engine instance
Source code in src/ml4t/backtest/engine.py 
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
@classmethod
def from_config(
    cls,
    feed: DataFeed,
    strategy: Strategy,
    config: BacktestConfig,
    *,
    contract_specs: dict[str, Any] | None = None,
    market_impact_model: Any | None = None,
    execution_limits: Any | None = None,
    funding_df: pl.DataFrame | None = None,
    lifecycle_version: LifecycleVersion | str = LifecycleVersion.V1,
    execution_policy: ExecutionPolicy | None = None,
    target_intent_state: dict[str, Any] | None = None,
) -> Engine:
    """Create an Engine instance from a BacktestConfig.

    Equivalent to ``Engine(feed, strategy, config)``. Kept as a convenience
    for code that reads more clearly with a named constructor.

    Args:
        feed: DataFeed with price data
        strategy: Strategy to execute
        config: BacktestConfig with all behavioral settings
        contract_specs: Per-asset contract specifications (futures multipliers, etc.)
        market_impact_model: Market impact model for fill simulation
        execution_limits: Execution limits (max order size, etc.)
        funding_df: Timestamped funding rates or amounts for named assets

    Returns:
        Configured Engine instance
    """
    return cls(
        feed,
        strategy,
        config,
        contract_specs=contract_specs,
        market_impact_model=market_impact_model,
        execution_limits=execution_limits,
        funding_df=funding_df,
        lifecycle_version=lifecycle_version,
        execution_policy=execution_policy,
        target_intent_state=target_intent_state,
    )

run_backtest 

run_backtest(
    prices,
    strategy,
    signals=None,
    context=None,
    config=None,
    *,
    feed_spec=None,
    contract=None,
    contract_specs=None,
    market_impact_model=None,
    execution_limits=None,
    funding_df=None,
    lifecycle_version=V1,
    execution_policy=None,
    target_intent_state=None,
)

Run a backtest with minimal setup.

Parameters:

Name Type Description Default
prices DataFrame | str

Price DataFrame or path to parquet file

 required
strategy Strategy

Strategy instance to execute

 required
signals DataFrame | str | None

Optional signals DataFrame or path

 None
context DataFrame | str | None

Optional context DataFrame or path

 None
config BacktestConfig | str | None

BacktestConfig instance, preset name (str), or None for defaults

 None
feed_spec Any | None

Optional shared dataset contract for schema and temporal metadata

 None
contract Any | None

Alias for feed_spec

 None
contract_specs dict[str, Any] | None

Per-asset contract specifications (futures multipliers, etc.)

 None
market_impact_model Any | None

Market impact model for fill simulation

 None
execution_limits Any | None

Execution limits (max order size, etc.)

 None
funding_df DataFrame | None

Timestamped funding rates or amounts for named assets

 None

Returns:

Type Description
BacktestResult

BacktestResult with metrics, trades, equity curve, and export methods.
Example

Using config preset

result = run_backtest(prices_df, strategy, config="backtrader") print(result.metrics["sharpe"])

Using custom config

config = BacktestConfig.from_preset("backtrader") config.commission_rate = 0.002 result = run_backtest(prices_df, strategy, config=config)

Futures with contract specs

from ml4t.backtest import ContractSpec, AssetClass specs = {"ES": ContractSpec(symbol="ES", asset_class=AssetClass.FUTURE, multiplier=50.0)} result = run_backtest(prices_df, strategy, config=config, contract_specs=specs)

Source code in src/ml4t/backtest/engine.py 
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
def run_backtest(
    prices: pl.DataFrame | str,
    strategy: Strategy,
    signals: pl.DataFrame | str | None = None,
    context: pl.DataFrame | str | None = None,
    config: BacktestConfig | str | None = None,
    *,
    feed_spec: Any | None = None,
    contract: Any | None = None,
    contract_specs: dict[str, Any] | None = None,
    market_impact_model: Any | None = None,
    execution_limits: Any | None = None,
    funding_df: pl.DataFrame | None = None,
    lifecycle_version: LifecycleVersion | str = LifecycleVersion.V1,
    execution_policy: ExecutionPolicy | None = None,
    target_intent_state: dict[str, Any] | None = None,
) -> BacktestResult:
    """Run a backtest with minimal setup.

    Args:
        prices: Price DataFrame or path to parquet file
        strategy: Strategy instance to execute
        signals: Optional signals DataFrame or path
        context: Optional context DataFrame or path
        config: BacktestConfig instance, preset name (str), or None for defaults
        feed_spec: Optional shared dataset contract for schema and temporal metadata
        contract: Alias for feed_spec
        contract_specs: Per-asset contract specifications (futures multipliers, etc.)
        market_impact_model: Market impact model for fill simulation
        execution_limits: Execution limits (max order size, etc.)
        funding_df: Timestamped funding rates or amounts for named assets

    Returns:
        BacktestResult with metrics, trades, equity curve, and export methods.

    Example:
        # Using config preset
        result = run_backtest(prices_df, strategy, config="backtrader")
        print(result.metrics["sharpe"])

        # Using custom config
        config = BacktestConfig.from_preset("backtrader")
        config.commission_rate = 0.002
        result = run_backtest(prices_df, strategy, config=config)

        # Futures with contract specs
        from ml4t.backtest import ContractSpec, AssetClass
        specs = {"ES": ContractSpec(symbol="ES", asset_class=AssetClass.FUTURE, multiplier=50.0)}
        result = run_backtest(prices_df, strategy, config=config, contract_specs=specs)
    """
    feed = DataFeed(
        prices_path=prices if isinstance(prices, str) else None,
        signals_path=signals if isinstance(signals, str) else None,
        context_path=context if isinstance(context, str) else None,
        prices_df=prices if isinstance(prices, pl.DataFrame) else None,
        signals_df=signals if isinstance(signals, pl.DataFrame) else None,
        context_df=context if isinstance(context, pl.DataFrame) else None,
        feed_spec=feed_spec,
        contract=contract,
    )

    if isinstance(config, str):
        from .config import BacktestConfig as ConfigCls

        config = ConfigCls.from_preset(config)

    return Engine(
        feed,
        strategy,
        config,
        contract_specs=contract_specs,
        market_impact_model=market_impact_model,
        execution_limits=execution_limits,
        funding_df=funding_df,
        lifecycle_version=lifecycle_version,
        execution_policy=execution_policy,
        target_intent_state=target_intent_state,
    ).run()

Strategy

Bases: ABC

Base strategy class.

on_prepare 

on_prepare(broker, config=None)

Called after on_start with causal configuration and no future feed data.

Source code in src/ml4t/backtest/strategy.py 
30
31
32
33
34
35
36
def on_prepare(
    self,
    broker: Any,
    config: Any | None = None,
) -> None:
    """Called after on_start with causal configuration and no future feed data."""
    return None

on_start 

on_start(broker)

Initialize strategy state before the first bar.

The broker is configured, but no market bar has been registered. Use this callback for position rules and state that does not require prices.

Source code in src/ml4t/backtest/strategy.py 
22
23
24
25
26
27
28
def on_start(self, broker: Any) -> None:  # noqa: B027
    """Initialize strategy state before the first bar.

    The broker is configured, but no market bar has been registered. Use
    this callback for position rules and state that does not require prices.
    """
    pass

on_data abstractmethod 

on_data(timestamp, data, context, broker)

Called for each timestamp with all available data.

Source code in src/ml4t/backtest/strategy.py 
11
12
13
14
15
16
17
18
19
20
@abstractmethod
def on_data(
    self,
    timestamp: datetime,
    data: dict[str, dict],
    context: dict[str, Any],
    broker: Any,  # Avoid circular import, use Any for broker type
) -> None:
    """Called for each timestamp with all available data."""
    pass

on_end 

on_end(broker)

Finalize strategy state after the final bar.

Open positions remain marked in the returned result; the engine does not submit automatic end-of-data closing orders before this callback.

Source code in src/ml4t/backtest/strategy.py 
38
39
40
41
42
43
44
def on_end(self, broker: Any) -> None:  # noqa: B027
    """Finalize strategy state after the final bar.

    Open positions remain marked in the returned result; the engine does not
    submit automatic end-of-data closing orders before this callback.
    """
    pass

DataFeed 

DataFeed(
    prices_path=None,
    signals_path=None,
    context_path=None,
    prices_df=None,
    signals_df=None,
    context_df=None,
    *,
    feed_spec=None,
    contract=None,
    entity_col=None,
    timestamp_col=None,
    price_col=None,
    open_col=None,
    high_col=None,
    low_col=None,
    close_col=None,
    volume_col=None,
    vwap_col=None,
    bid_col=None,
    ask_col=None,
    mid_col=None,
    bid_size_col=None,
    ask_size_col=None,
    session_col=None,
)

Polars-based multi-asset data feed with signals and context.

Pre-indexes data by timestamp at initialization for O(1) lookups during iteration. Public prices, signals, and context attributes are normalized copies sorted by timestamp when needed. DataFrames are kept in their native format and converted to dicts only for the active bar, avoiding the large memory overhead of materializing one child DataFrame per timestamp.

The resolved timestamp, entity, and reference-price columns are required. Open, high, low, and volume columns are optional. Missing OHLC values fall back to the configured close and missing volume becomes zero.

Usage

feed = DataFeed(prices_df=prices, signals_df=signals) for timestamp, assets_data, context in feed: # assets_data: {"AAPL": {"close": 150.0, "signals": {...}}, ...} process(timestamp, assets_data)

Source code in src/ml4t/backtest/datafeed.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def __init__(
    self,
    prices_path: str | None = None,
    signals_path: str | None = None,
    context_path: str | None = None,
    prices_df: pl.DataFrame | None = None,
    signals_df: pl.DataFrame | None = None,
    context_df: pl.DataFrame | None = None,
    *,
    feed_spec: FeedSpec | Any | None = None,
    contract: FeedSpec | Any | None = None,
    entity_col: str | None = None,
    timestamp_col: str | None = None,
    price_col: str | None = None,
    open_col: str | None = None,
    high_col: str | None = None,
    low_col: str | None = None,
    close_col: str | None = None,
    volume_col: str | None = None,
    vwap_col: str | None = None,
    bid_col: str | None = None,
    ask_col: str | None = None,
    mid_col: str | None = None,
    bid_size_col: str | None = None,
    ask_size_col: str | None = None,
    session_col: str | None = None,
):
    if feed_spec is not None and contract is not None:
        raise ValueError("Pass either feed_spec or contract, not both")
    for label, path, frame in (
        ("prices", prices_path, prices_df),
        ("signals", signals_path, signals_df),
        ("context", context_path, context_df),
    ):
        if path is not None and frame is not None:
            raise ValueError(f"Pass either {label}_path or {label}_df, not both")

    self.prices = (
        prices_df
        if prices_df is not None
        else (_read_parquet(prices_path) if prices_path else None)
    )
    self.signals = (
        signals_df
        if signals_df is not None
        else (_read_parquet(signals_path) if signals_path else None)
    )
    self.context = (
        context_df
        if context_df is not None
        else (_read_parquet(context_path) if context_path else None)
    )

    if self.prices is None:
        raise ValueError("prices_path or prices_df required")

    raw_spec = FeedSpec.from_any(feed_spec if feed_spec is not None else contract)
    default_price_col = FeedSpec().price_col
    price_col_is_explicit = price_col is not None or raw_spec.price_col != default_price_col
    self.feed_spec = raw_spec.with_overrides(
        entity_col=entity_col,
        timestamp_col=timestamp_col,
        price_col=price_col,
        open_col=open_col,
        high_col=high_col,
        low_col=low_col,
        close_col=close_col,
        volume_col=volume_col,
        vwap_col=vwap_col,
        bid_col=bid_col,
        ask_col=ask_col,
        mid_col=mid_col,
        bid_size_col=bid_size_col,
        ask_size_col=ask_size_col,
    ).resolve(self.prices.columns, self.ENTITY_COL_CANDIDATES)
    self.contract = self.feed_spec
    self._timestamp_col = self.feed_spec.timestamp_col
    resolved_entity_col = self.feed_spec.entity_col
    if not isinstance(resolved_entity_col, str):
        raise ValueError("DataFeed requires one resolved string entity column")
    self._entity_col = resolved_entity_col
    self._price_col = self.feed_spec.price_col
    if self._price_col not in self.prices.columns:
        if price_col_is_explicit:
            raise ValueError(
                f"price_col={self._price_col!r} not found in price columns "
                f"{self.prices.columns}"
            )
        if self.feed_spec.close_col not in self.prices.columns:
            raise ValueError(
                f"price_col={self._price_col!r} and close_col={self.feed_spec.close_col!r} "
                f"not found in price columns {self.prices.columns}"
            )
        self._price_col = self.feed_spec.close_col
        self.feed_spec = self.feed_spec.with_overrides(price_col=self._price_col)
        self.contract = self.feed_spec
    self._open_col = self.feed_spec.open_col
    self._high_col = self.feed_spec.high_col
    self._low_col = self.feed_spec.low_col
    self._close_col = self.feed_spec.close_col
    self._volume_col = self.feed_spec.volume_col
    self._vwap_col = self.feed_spec.vwap_col
    self._bid_col = self.feed_spec.bid_col
    self._ask_col = self.feed_spec.ask_col
    self._mid_col = self.feed_spec.mid_col
    self._bid_size_col = self.feed_spec.bid_size_col
    self._ask_size_col = self.feed_spec.ask_size_col

    self.prices, self._price_ranges_by_ts = self._index_by_timestamp(self.prices)
    if self.signals is not None:
        if self._entity_col not in self.signals.columns:
            raise ValueError(
                f"entity_col={self._entity_col!r} not found in signal columns "
                f"{self.signals.columns}"
            )
        self.signals, self._signal_ranges_by_ts = self._index_by_timestamp(self.signals)
    else:
        self._signal_ranges_by_ts = {}
    if self.context is not None:
        self.context, self._context_ranges_by_ts = self._index_by_timestamp(self.context)
    else:
        self._context_ranges_by_ts = {}

    self._timestamps = self._get_timestamps()
    self.session_col = session_col
    self._session_by_timestamp: dict[datetime, date] = {}
    self._session_decision_timestamps: set[datetime] = set()
    if session_col is not None:
        self._index_decision_sessions(session_col)
    self._idx = 0
    self._signal_columns = (
        [c for c in self.signals.columns if c not in (self._timestamp_col, self._entity_col)]
        if self.signals is not None
        else []
    )
    self._context_columns = (
        [c for c in self.context.columns if c != self._timestamp_col]
        if self.context is not None
        else []
    )

    price_cols = self.prices.columns
    self._price_entity_idx = price_cols.index(self._entity_col)
    self._price_open_idx = (
        price_cols.index(self._open_col) if self._open_col in price_cols else -1
    )
    self._price_high_idx = (
        price_cols.index(self._high_col) if self._high_col in price_cols else -1
    )
    self._price_low_idx = price_cols.index(self._low_col) if self._low_col in price_cols else -1
    self._price_close_idx = (
        price_cols.index(self._close_col) if self._close_col in price_cols else -1
    )
    self._price_price_idx = (
        price_cols.index(self._price_col)
        if self._price_col in price_cols
        else self._price_close_idx
    )
    self._price_volume_idx = (
        price_cols.index(self._volume_col) if self._volume_col in price_cols else -1
    )
    self._price_vwap_idx = (
        price_cols.index(self._vwap_col) if self._vwap_col in price_cols else -1
    )
    self._price_bid_idx = price_cols.index(self._bid_col) if self._bid_col in price_cols else -1
    self._price_ask_idx = price_cols.index(self._ask_col) if self._ask_col in price_cols else -1
    self._price_mid_idx = price_cols.index(self._mid_col) if self._mid_col in price_cols else -1
    self._price_bid_size_idx = (
        price_cols.index(self._bid_size_col) if self._bid_size_col in price_cols else -1
    )
    self._price_ask_size_idx = (
        price_cols.index(self._ask_size_col) if self._ask_size_col in price_cols else -1
    )

    if self.signals is not None:
        signal_cols = self.signals.columns
        if self._timestamp_col not in signal_cols:
            raise ValueError(
                f"timestamp_col={self._timestamp_col!r} not found in signal columns {signal_cols}"
            )
        self._signal_entity_idx = signal_cols.index(self._entity_col)
        self._signal_col_indices = [signal_cols.index(c) for c in self._signal_columns]
    else:
        self._signal_entity_idx = -1
        self._signal_col_indices = []

    if self.context is not None:
        context_cols = self.context.columns
        if self._timestamp_col not in context_cols:
            raise ValueError(
                f"timestamp_col={self._timestamp_col!r} not found in context columns {context_cols}"
            )
        self._context_col_indices = [context_cols.index(c) for c in self._context_columns]
    else:
        self._context_col_indices = []

n_bars property 

n_bars

Number of unique timestamps/bars.

timestamps property 

timestamps

Unique feed timestamps in iteration order.

Configuration

BacktestConfig dataclass 

BacktestConfig(
    allow_short_selling=False,
    allow_leverage=False,
    initial_margin=0.5,
    long_maintenance_margin=0.25,
    short_maintenance_margin=0.3,
    fixed_margin_schedule=None,
    margin_pct_schedule=None,
    short_cash_policy=CREDIT,
    lock_notional_update_mode=POSITION_LEGS,
    execution_price=OPEN,
    mark_price=PRICE,
    execution_mode=NEXT_BAR,
    stop_fill_mode=STOP_PRICE,
    stop_level_basis=FILL_PRICE,
    trail_hwm_source=CLOSE,
    trail_include_entry_bar_extremes=False,
    initial_hwm_source=FILL_PRICE,
    trail_stop_timing=LAGGED,
    share_type=INTEGER,
    share_rounding=NEAREST,
    commission_type=NONE,
    commission_rate=0.0,
    commission_per_share=0.0,
    commission_per_trade=0.0,
    commission_minimum=0.0,
    slippage_type=NONE,
    slippage_rate=0.0,
    slippage_fixed=0.0,
    slippage_spread=0.0,
    slippage_spread_by_asset=dict(),
    slippage_spread_convention=FULL_SPREAD,
    stop_slippage_rate=0.0,
    initial_cash=100000.0,
    cash_buffer_pct=0.0,
    settlement_delay=0,
    settlement_reduces_buying_power=True,
    reject_on_insufficient_cash=True,
    skip_cash_validation=False,
    partial_fills_allowed=False,
    fill_ordering=EXIT_FIRST,
    entry_order_priority=SUBMISSION,
    next_bar_submission_precheck=False,
    next_bar_simple_cash_check=False,
    buying_power_reservation=False,
    next_bar_queue_shadow_validation=False,
    immediate_fill=False,
    rebalance_mode=INCREMENTAL,
    rebalance_headroom_pct=1.0,
    missing_price_policy=SKIP,
    late_asset_policy=ALLOW,
    late_asset_min_bars=1,
    calendar=None,
    timezone="UTC",
    data_frequency=DAILY,
    enforce_sessions=False,
    preset_name=None,
    feed_spec=None,
    metadata=dict(),
    retain_intent_history=False,
    retain_lifecycle_history=False,
)

Complete configuration for backtesting behavior.

All behavioral differences between frameworks are captured here. Load presets to match specific frameworks exactly.

This is the single source of truth for all backtest settings. Broker and Engine are configured entirely from this dataclass.

from_preset classmethod 

from_preset(preset)

Load a predefined configuration preset.

Available presets: - "default": Sensible defaults for general use - "backtrader": Match Backtrader's default behavior - "vectorbt": Match VectorBT's default behavior - "zipline": Match Zipline's default behavior - "lean": Match the frozen LEAN daily US-equity comparison protocol - "realistic": Conservative settings for realistic simulation - "ibkr_us_stocks_fixed": Interactive Brokers US stocks fixed pricing

Source code in src/ml4t/backtest/config.py 
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
@classmethod
def from_preset(cls, preset: str) -> BacktestConfig:
    """
    Load a predefined configuration preset.

    Available presets:
    - "default": Sensible defaults for general use
    - "backtrader": Match Backtrader's default behavior
    - "vectorbt": Match VectorBT's default behavior
    - "zipline": Match Zipline's default behavior
    - "lean": Match the frozen LEAN daily US-equity comparison protocol
    - "realistic": Conservative settings for realistic simulation
    - "ibkr_us_stocks_fixed": Interactive Brokers US stocks fixed pricing
    """
    from .profiles import get_profile_config

    profile_data = get_profile_config(preset)
    return cls.from_dict(profile_data, preset_name=preset, strict=True)

from_yaml classmethod 

from_yaml(path)

Load config from YAML file.

Source code in src/ml4t/backtest/config.py 
1252
1253
1254
1255
1256
1257
1258
@classmethod
def from_yaml(cls, path: str | Path) -> BacktestConfig:
    """Load config from YAML file."""
    path = Path(path)
    with open(path) as f:
        data = yaml.safe_load(f)
    return cls.from_dict(data, preset_name=path.stem, strict=True)

from_dict classmethod 

from_dict(data, preset_name=None, strict=True)

Create config from dictionary.

Parameters:

Name Type Description Default
data dict

Nested config dictionary

 required
preset_name str | None

Optional metadata label

 None
strict bool

If True, reject unknown sections/keys

 True
Source code in src/ml4t/backtest/config.py 
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
@classmethod
def from_dict(
    cls, data: dict, preset_name: str | None = None, strict: bool = True
) -> BacktestConfig:
    """Create config from dictionary.

    Args:
        data: Nested config dictionary
        preset_name: Optional metadata label
        strict: If True, reject unknown sections/keys
    """
    if not isinstance(data, dict):
        raise TypeError(f"Config data must be a dict, got {type(data).__name__}")

    if strict:
        allowed_sections = {
            "account",
            "execution",
            "stops",
            "position_sizing",
            "commission",
            "slippage",
            "cash",
            "settlement",
            "orders",
            "calendar",
            "feed",
            "result",
            "metadata",
        }
        unknown_sections = set(data) - allowed_sections
        if unknown_sections:
            raise ValueError(f"Unknown config section(s): {sorted(unknown_sections)}")

        allowed_keys_by_section = {
            "account": {
                "allow_short_selling",
                "allow_leverage",
                "initial_margin",
                "long_maintenance_margin",
                "short_maintenance_margin",
                "fixed_margin_schedule",
                "margin_pct_schedule",
                "short_cash_policy",
                "lock_notional_update_mode",
            },
            "execution": {"execution_price", "mark_price", "execution_mode"},
            "stops": {
                "stop_fill_mode",
                "stop_level_basis",
                "trail_hwm_source",
                "trail_include_entry_bar_extremes",
                "initial_hwm_source",
                "trail_stop_timing",
            },
            "position_sizing": {"share_type", "share_rounding"},
            "commission": {"model", "rate", "per_share", "per_trade", "minimum"},
            "slippage": {
                "model",
                "rate",
                "fixed",
                "spread",
                "spread_by_asset",
                "spread_convention",
                "stop_rate",
            },
            "cash": {"initial", "buffer_pct"},
            "settlement": {"delay", "reduces_buying_power"},
            "orders": {
                "reject_on_insufficient_cash",
                "skip_cash_validation",
                "partial_fills_allowed",
                "fill_ordering",
                "entry_order_priority",
                "next_bar_submission_precheck",
                "next_bar_simple_cash_check",
                "buying_power_reservation",
                "next_bar_queue_shadow_validation",
                "immediate_fill",
                "rebalance_mode",
                "rebalance_headroom_pct",
                "missing_price_policy",
                "late_asset_policy",
                "late_asset_min_bars",
            },
            "calendar": {
                "calendar",
                "timezone",
                "data_frequency",
                "enforce_sessions",
            },
            "feed": {
                "timestamp_col",
                "entity_col",
                "price_col",
                "open_col",
                "high_col",
                "low_col",
                "close_col",
                "volume_col",
                "vwap_col",
                "bid_col",
                "ask_col",
                "mid_col",
                "bid_size_col",
                "ask_size_col",
                "calendar",
                "timezone",
                "data_frequency",
                "bar_type",
                "timestamp_semantics",
                "session_start_time",
            },
            "result": {"retain_intent_history", "retain_lifecycle_history"},
        }
        for section, cfg in data.items():
            if section == "metadata":
                if not isinstance(cfg, dict):
                    raise TypeError(
                        f"Section 'metadata' must be a dict, got {type(cfg).__name__}"
                    )
                continue
            if not isinstance(cfg, dict):
                raise TypeError(f"Section '{section}' must be a dict, got {type(cfg).__name__}")
            unknown_keys = set(cfg) - allowed_keys_by_section[section]
            if unknown_keys:
                raise ValueError(
                    f"Unknown key(s) in section '{section}': {sorted(unknown_keys)}"
                )

    acct_cfg = data.get("account", {})
    exec_cfg = data.get("execution", {})
    stops_cfg = data.get("stops", {})
    sizing_cfg = data.get("position_sizing", {})
    comm_cfg = data.get("commission", {})
    slip_cfg = data.get("slippage", {})
    cash_cfg = data.get("cash", {})
    settle_cfg = data.get("settlement", {})
    order_cfg = data.get("orders", {})
    cal_cfg = data.get("calendar", {})
    feed_cfg = data.get("feed", {})
    result_cfg = data.get("result", {})
    metadata = data.get("metadata", {})

    if metadata is None:
        metadata = {}
    if not isinstance(metadata, dict):
        raise TypeError(f"Section 'metadata' must be a dict, got {type(metadata).__name__}")
    spread_by_asset = slip_cfg.get("spread_by_asset", {})
    if spread_by_asset is None:
        spread_by_asset = {}
    if not isinstance(spread_by_asset, dict):
        raise TypeError(
            f"Section 'slippage.spread_by_asset' must be a dict, got "
            f"{type(spread_by_asset).__name__}"
        )

    allow_short_selling = acct_cfg.get("allow_short_selling", False)
    allow_leverage = acct_cfg.get("allow_leverage", False)

    return cls(
        # Account
        allow_short_selling=allow_short_selling,
        allow_leverage=allow_leverage,
        initial_margin=acct_cfg.get("initial_margin", 0.5),
        long_maintenance_margin=acct_cfg.get("long_maintenance_margin", 0.25),
        short_maintenance_margin=acct_cfg.get("short_maintenance_margin", 0.30),
        fixed_margin_schedule=acct_cfg.get("fixed_margin_schedule"),
        margin_pct_schedule=acct_cfg.get("margin_pct_schedule"),
        short_cash_policy=ShortCashPolicy(acct_cfg.get("short_cash_policy", "credit")),
        lock_notional_update_mode=LockNotionalUpdateMode(
            acct_cfg.get("lock_notional_update_mode", "position_legs")
        ),
        # Execution
        execution_price=ExecutionPrice(
            exec_cfg.get("execution_price", ExecutionPrice.OPEN.value)
        ),
        mark_price=ExecutionPrice(exec_cfg.get("mark_price", ExecutionPrice.PRICE.value)),
        execution_mode=ExecutionMode(
            exec_cfg.get("execution_mode", ExecutionMode.NEXT_BAR.value)
        ),
        # Stops
        stop_fill_mode=StopFillMode(stops_cfg.get("stop_fill_mode", "stop_price")),
        stop_level_basis=StopLevelBasis(stops_cfg.get("stop_level_basis", "fill_price")),
        trail_hwm_source=WaterMarkSource(stops_cfg.get("trail_hwm_source", "close")),
        trail_include_entry_bar_extremes=stops_cfg.get(
            "trail_include_entry_bar_extremes", False
        ),
        initial_hwm_source=InitialHwmSource(stops_cfg.get("initial_hwm_source", "fill_price")),
        trail_stop_timing=TrailStopTiming(stops_cfg.get("trail_stop_timing", "lagged")),
        # Sizing
        share_type=ShareType(sizing_cfg.get("share_type", "integer")),
        share_rounding=ShareRounding(sizing_cfg.get("share_rounding", "nearest")),
        # Commission
        commission_type=CommissionType(comm_cfg.get("model", "none")),
        commission_rate=comm_cfg.get("rate", 0.0),
        commission_per_share=comm_cfg.get("per_share", 0.0),
        commission_per_trade=comm_cfg.get("per_trade", 0.0),
        commission_minimum=comm_cfg.get("minimum", 0.0),
        # Slippage
        slippage_type=SlippageType(slip_cfg.get("model", "none")),
        slippage_rate=slip_cfg.get("rate", 0.0),
        slippage_fixed=slip_cfg.get("fixed", 0.0),
        slippage_spread=slip_cfg.get("spread", 0.0),
        slippage_spread_by_asset={
            str(asset): float(spread) for asset, spread in spread_by_asset.items()
        },
        slippage_spread_convention=SpreadConvention(
            slip_cfg.get("spread_convention", "full_spread")
        ),
        stop_slippage_rate=slip_cfg.get("stop_rate", 0.0),
        # Cash
        initial_cash=cash_cfg.get("initial", 100000.0),
        cash_buffer_pct=cash_cfg.get("buffer_pct", 0.0),
        # Settlement
        settlement_delay=settle_cfg.get("delay", 0),
        settlement_reduces_buying_power=settle_cfg.get("reduces_buying_power", True),
        # Orders
        reject_on_insufficient_cash=order_cfg.get("reject_on_insufficient_cash", True),
        skip_cash_validation=order_cfg.get("skip_cash_validation", False),
        partial_fills_allowed=order_cfg.get("partial_fills_allowed", False),
        fill_ordering=FillOrdering(order_cfg.get("fill_ordering", "exit_first")),
        entry_order_priority=EntryOrderPriority(
            order_cfg.get("entry_order_priority", "submission")
        ),
        next_bar_submission_precheck=order_cfg.get("next_bar_submission_precheck", False),
        next_bar_simple_cash_check=order_cfg.get("next_bar_simple_cash_check", False),
        buying_power_reservation=order_cfg.get("buying_power_reservation", False),
        next_bar_queue_shadow_validation=order_cfg.get(
            "next_bar_queue_shadow_validation", False
        ),
        immediate_fill=order_cfg.get("immediate_fill", False),
        rebalance_mode=RebalanceMode(
            order_cfg.get("rebalance_mode", RebalanceMode.INCREMENTAL.value)
        ),
        rebalance_headroom_pct=order_cfg.get("rebalance_headroom_pct", 1.0),
        missing_price_policy=MissingPricePolicy(order_cfg.get("missing_price_policy", "skip")),
        late_asset_policy=LateAssetPolicy(order_cfg.get("late_asset_policy", "allow")),
        late_asset_min_bars=order_cfg.get("late_asset_min_bars", 1),
        # Calendar
        calendar=cal_cfg.get("calendar"),
        timezone=cal_cfg.get("timezone", "UTC"),
        data_frequency=DataFrequency(cal_cfg.get("data_frequency", "daily")),
        enforce_sessions=cal_cfg.get("enforce_sessions", False),
        # Metadata
        preset_name=preset_name,
        feed_spec=FeedSpec.from_any(feed_cfg) if feed_cfg else None,
        retain_intent_history=result_cfg.get("retain_intent_history", False),
        retain_lifecycle_history=result_cfg.get("retain_lifecycle_history", False),
        metadata=dict(metadata),
    )

to_yaml 

to_yaml(path)

Save config to YAML file.

Source code in src/ml4t/backtest/config.py 
1246
1247
1248
1249
1250
def to_yaml(self, path: str | Path) -> None:
    """Save config to YAML file."""
    path = Path(path)
    with open(path, "w") as f:
        yaml.safe_dump(self.to_dict(), f, default_flow_style=False, sort_keys=False)

to_dict 

to_dict()

Convert config to dictionary for serialization.

Source code in src/ml4t/backtest/config.py 
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
def to_dict(self) -> dict:
    """Convert config to dictionary for serialization."""
    return {
        "account": {
            "allow_short_selling": self.allow_short_selling,
            "allow_leverage": self.allow_leverage,
            "initial_margin": self.initial_margin,
            "long_maintenance_margin": self.long_maintenance_margin,
            "short_maintenance_margin": self.short_maintenance_margin,
            "fixed_margin_schedule": _margin_schedule_to_dict(self.fixed_margin_schedule),
            "margin_pct_schedule": _margin_schedule_to_dict(self.margin_pct_schedule),
            "short_cash_policy": self.short_cash_policy.value,
            "lock_notional_update_mode": self.lock_notional_update_mode.value,
        },
        "execution": {
            "execution_price": self.execution_price.value,
            "mark_price": self.mark_price.value,
            "execution_mode": self.execution_mode.value,
        },
        "stops": {
            "stop_fill_mode": self.stop_fill_mode.value,
            "stop_level_basis": self.stop_level_basis.value,
            "trail_hwm_source": self.trail_hwm_source.value,
            "trail_include_entry_bar_extremes": self.trail_include_entry_bar_extremes,
            "initial_hwm_source": self.initial_hwm_source.value,
            "trail_stop_timing": self.trail_stop_timing.value,
        },
        "position_sizing": {
            "share_type": self.share_type.value,
            "share_rounding": self.share_rounding.value,
        },
        "commission": {
            "model": self.commission_type.value,
            "rate": self.commission_rate,
            "per_share": self.commission_per_share,
            "per_trade": self.commission_per_trade,
            "minimum": self.commission_minimum,
        },
        "slippage": {
            "model": self.slippage_type.value,
            "rate": self.slippage_rate,
            "fixed": self.slippage_fixed,
            "spread": self.slippage_spread,
            "spread_by_asset": self.slippage_spread_by_asset,
            "spread_convention": self.slippage_spread_convention.value,
            "stop_rate": self.stop_slippage_rate,
        },
        "cash": {
            "initial": self.initial_cash,
            "buffer_pct": self.cash_buffer_pct,
        },
        "settlement": {
            "delay": self.settlement_delay,
            "reduces_buying_power": self.settlement_reduces_buying_power,
        },
        "orders": {
            "reject_on_insufficient_cash": self.reject_on_insufficient_cash,
            "skip_cash_validation": self.skip_cash_validation,
            "partial_fills_allowed": self.partial_fills_allowed,
            "fill_ordering": self.fill_ordering.value,
            "entry_order_priority": self.entry_order_priority.value,
            "next_bar_submission_precheck": self.next_bar_submission_precheck,
            "next_bar_simple_cash_check": self.next_bar_simple_cash_check,
            "buying_power_reservation": self.buying_power_reservation,
            "next_bar_queue_shadow_validation": self.next_bar_queue_shadow_validation,
            "immediate_fill": self.immediate_fill,
            "rebalance_mode": self.rebalance_mode.value,
            "rebalance_headroom_pct": self.rebalance_headroom_pct,
            "missing_price_policy": self.missing_price_policy.value,
            "late_asset_policy": self.late_asset_policy.value,
            "late_asset_min_bars": self.late_asset_min_bars,
        },
        "calendar": {
            "calendar": self.calendar,
            "timezone": self.timezone,
            "data_frequency": self.data_frequency.value,
            "enforce_sessions": self.enforce_sessions,
        },
        "feed": _feed_spec_to_dict(self.resolved_feed_spec),
        "result": {
            "retain_intent_history": self.retain_intent_history,
            "retain_lifecycle_history": self.retain_lifecycle_history,
        },
        "metadata": serialize_artifact_value(self.metadata),
    }

validate 

validate(warn=True)

Validate configuration and return warnings for edge cases.

Checks for configurations that may produce unexpected results or indicate potential issues. Returns a list of warning messages.

Parameters:

Name Type Description Default
warn bool

If True, emit warnings via warnings.warn(). Default True.

 True

Returns:

Type Description
list[str]

List of warning message strings (empty if no issues found).
Example

config = BacktestConfig(execution_mode=ExecutionMode.SAME_BAR) warnings = config.validate()

["SAME_BAR execution has look-ahead bias risk..."]

Source code in src/ml4t/backtest/config.py 
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
def validate(self, warn: bool = True) -> list[str]:
    """Validate configuration and return warnings for edge cases.

    Checks for configurations that may produce unexpected results or
    indicate potential issues. Returns a list of warning messages.

    Args:
        warn: If True, emit warnings via warnings.warn(). Default True.

    Returns:
        List of warning message strings (empty if no issues found).

    Example:
        config = BacktestConfig(execution_mode=ExecutionMode.SAME_BAR)
        warnings = config.validate()
        # ["SAME_BAR execution has look-ahead bias risk..."]
    """
    import warnings as _warnings

    issues: list[str] = self._execution_validation_errors()

    # Look-ahead bias warning
    if self.execution_mode == ExecutionMode.SAME_BAR:
        issues.append(
            "SAME_BAR execution has look-ahead bias risk. "
            "Use NEXT_BAR execution mode for realistic backtesting."
        )

    # Zero cost warning
    if self.commission_type == CommissionType.NONE and self.slippage_type == SlippageType.NONE:
        issues.append(
            "Both commission and slippage are disabled. Results may be overly optimistic."
        )

    # Volume-based slippage without partial fills
    if self.slippage_type == SlippageType.VOLUME_BASED and not self.partial_fills_allowed:
        issues.append(
            "Volume-based slippage without partial_fills_allowed may cause "
            "orders to be rejected in low-volume conditions."
        )

    # High slippage + high commission
    total_cost = self.slippage_rate + self.commission_rate
    if total_cost > 0.01:  # > 1% round-trip
        issues.append(
            f"Total transaction cost ({total_cost:.2%}) is high. "
            "Verify this matches your broker's actual costs."
        )

    if (
        self.slippage_type == SlippageType.SPREAD
        and self.slippage_spread == 0.0
        and not self.slippage_spread_by_asset
    ):
        issues.append(
            "slippage_type='spread' requires slippage_spread > 0 or "
            "slippage_spread_by_asset entries."
        )

    # Fractional shares warning for production
    if self.share_type == ShareType.FRACTIONAL and self.preset_name == "realistic":
        issues.append(
            "REALISTIC preset with fractional shares may not match all brokers. "
            "Set share_type=INTEGER for most accurate simulation."
        )

    # Margin parameter validation
    if self.allow_leverage:
        if not 0.0 < self.initial_margin <= 1.0:
            issues.append(f"initial_margin ({self.initial_margin}) must be in (0.0, 1.0]")
        if not 0.0 < self.long_maintenance_margin <= 1.0:
            issues.append(
                f"long_maintenance_margin ({self.long_maintenance_margin}) must be in (0.0, 1.0]"
            )
        if not 0.0 < self.short_maintenance_margin <= 1.0:
            issues.append(
                f"short_maintenance_margin ({self.short_maintenance_margin}) must be in (0.0, 1.0]"
            )
        if self.long_maintenance_margin >= self.initial_margin:
            issues.append(
                f"long_maintenance_margin ({self.long_maintenance_margin}) must be < "
                f"initial_margin ({self.initial_margin})"
            )
        if self.short_maintenance_margin >= self.initial_margin:
            issues.append(
                f"short_maintenance_margin ({self.short_maintenance_margin}) must be < "
                f"initial_margin ({self.initial_margin})"
            )

    fixed_assets = set(self.fixed_margin_schedule or {})
    pct_assets = set(self.margin_pct_schedule or {})
    overlapping_margin_assets = sorted(fixed_assets & pct_assets)
    if overlapping_margin_assets:
        issues.append(
            "fixed_margin_schedule and margin_pct_schedule cannot both define: "
            f"{overlapping_margin_assets}"
        )
    issues.extend(_validate_margin_pct_schedule(self.margin_pct_schedule))

    if self.settlement_delay < 0 or self.settlement_delay > 5:
        issues.append(
            f"settlement_delay ({self.settlement_delay}) should be 0-5. "
            "Values count processed bars: 0 is instant, 1 delays one bar."
        )

    if not 0.0 < self.rebalance_headroom_pct <= 1.0:
        issues.append(
            f"rebalance_headroom_pct ({self.rebalance_headroom_pct}) must be in (0.0, 1.0]"
        )
    if self.late_asset_min_bars < 1:
        issues.append(f"late_asset_min_bars ({self.late_asset_min_bars}) must be >= 1")

    # Emit warnings if requested
    if warn and issues:
        for msg in issues:
            _warnings.warn(msg, UserWarning, stacklevel=2)

    return issues

describe 

describe()

Return human-readable description of configuration.

Source code in src/ml4t/backtest/config.py 
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
def describe(self) -> str:
    """Return human-readable description of configuration."""
    allow_shorts, allow_leverage = self.get_effective_account_settings()
    account_str = self.get_effective_account_type()

    lines = [
        f"BacktestConfig (preset: {self.preset_name or 'custom'})",
        "=" * 50,
        "",
        "Account:",
        f"  Type: {account_str}",
        f"  Short selling: {'allowed' if allow_shorts else 'disabled'}",
        f"  Leverage: {'enabled' if allow_leverage else 'disabled'}",
    ]

    if allow_leverage:
        lines.extend(
            [
                f"  Initial margin: {self.initial_margin:.0%}",
                f"  Long maintenance: {self.long_maintenance_margin:.0%}",
                f"  Short maintenance: {self.short_maintenance_margin:.0%}",
            ]
        )

    lines.extend(
        [
            "",
            "Execution:",
            f"  Execution mode: {self.execution_mode.value}",
            f"  Execution price: {self.execution_price.value}",
            f"  Mark price: {self.mark_price.value}",
            "",
            "Stops:",
            f"  Fill mode: {self.stop_fill_mode.value}",
            f"  Level basis: {self.stop_level_basis.value}",
            f"  Trail HWM source: {self.trail_hwm_source.value}",
            f"  Include entry-bar extremes: {self.trail_include_entry_bar_extremes}",
            f"  Trail timing: {self.trail_stop_timing.value}",
            "",
            "Position Sizing:",
            f"  Share type: {self.share_type.value}",
            "",
            "Costs:",
            f"  Commission: {self.commission_type.value} @ {self.commission_rate:.2%}",
            (
                "  Slippage: "
                f"{self.slippage_type.value} default_spread={self.slippage_spread:.6f} "
                f"convention={self.slippage_spread_convention.value}"
                if self.slippage_type == SlippageType.SPREAD
                else f"  Slippage: {self.slippage_type.value} @ {self.slippage_rate:.2%}"
            ),
        ]
    )

    if self.stop_slippage_rate > 0:
        lines.append(f"  Stop slippage: +{self.stop_slippage_rate:.2%}")

    lines.extend(
        [
            "",
            "Orders:",
            f"  Fill ordering: {self.fill_ordering.value}",
            f"  Entry priority: {self.entry_order_priority.value}",
            f"  Next-bar precheck: {self.next_bar_submission_precheck}",
            f"  Next-bar cash check: {self.next_bar_simple_cash_check}",
            f"  Next-bar queue shadow validation: {self.next_bar_queue_shadow_validation}",
            f"  Rebalance mode: {self.rebalance_mode.value}",
            f"  Rebalance headroom: {self.rebalance_headroom_pct:.3f}",
            f"  Missing price policy: {self.missing_price_policy.value}",
            f"  Late asset policy: {self.late_asset_policy.value}",
            f"  Late asset min bars: {self.late_asset_min_bars}",
            f"  Reject insufficient: {self.reject_on_insufficient_cash}",
            f"  Skip cash validation: {self.skip_cash_validation}",
            f"  Partial fills: {self.partial_fills_allowed}",
            "",
            "Cash:",
            f"  Initial: ${self.initial_cash:,.0f}",
            f"  Buffer: {self.cash_buffer_pct:.1%}",
        ]
    )

    if self.settlement_delay > 0:
        lines.extend(
            [
                "",
                "Settlement:",
                f"  Delay: T+{self.settlement_delay}",
            ]
        )

    return "\n".join(line for line in lines if line is not None)

profiles

Centralized profile definitions for framework-aligned behavior.

get_profile_config 

get_profile_config(name)

Return a deep copy of nested config data for the named profile.

Source code in src/ml4t/backtest/profiles.py 
463
464
465
466
467
468
469
def get_profile_config(name: str) -> dict:
    """Return a deep copy of nested config data for the named profile."""
    key = _ALIASES.get(name, name)
    if key not in _PROFILES:
        available = ", ".join(sorted(_PROFILES.keys()))
        raise ValueError(f"Unknown preset '{name}'. Available: {available}")
    return deepcopy(_PROFILES[key])

list_profiles 

list_profiles()

List canonical preset names.

Source code in src/ml4t/backtest/profiles.py 
472
473
474
def list_profiles() -> list[str]:
    """List canonical preset names."""
    return _CORE_PROFILE_NAMES.copy()

Broker

Broker 

Broker(
    initial_cash=100000.0,
    commission_model=None,
    slippage_model=None,
    stop_slippage_rate=0.0,
    execution_mode=SAME_BAR,
    execution_price=CLOSE,
    mark_price=PRICE,
    stop_fill_mode=STOP_PRICE,
    stop_level_basis=FILL_PRICE,
    trail_hwm_source=CLOSE,
    trail_include_entry_bar_extremes=False,
    initial_hwm_source=FILL_PRICE,
    trail_stop_timing=LAGGED,
    allow_short_selling=False,
    allow_leverage=False,
    initial_margin=0.5,
    long_maintenance_margin=0.25,
    short_maintenance_margin=0.3,
    fixed_margin_schedule=None,
    margin_pct_schedule=None,
    short_cash_policy=CREDIT,
    lock_notional_update_mode=POSITION_LEGS,
    execution_limits=None,
    market_impact_model=None,
    contract_specs=None,
    share_type=INTEGER,
    fill_ordering=EXIT_FIRST,
    entry_order_priority=SUBMISSION,
    next_bar_submission_precheck=False,
    next_bar_simple_cash_check=False,
    buying_power_reservation=False,
    next_bar_queue_shadow_validation=False,
    immediate_fill=False,
    reject_on_insufficient_cash=True,
    skip_cash_validation=False,
    cash_buffer_pct=0.0,
    partial_fills_allowed=False,
    rebalance_headroom_pct=1.0,
    missing_price_policy=SKIP,
    late_asset_policy=ALLOW,
    late_asset_min_bars=1,
    settlement_delay=0,
    settlement_reduces_buying_power=True,
)

Broker interface - same for backtest and live trading.

Source code in src/ml4t/backtest/broker.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def __init__(
    self,
    initial_cash: float = 100000.0,
    commission_model: CommissionModel | None = None,
    slippage_model: SlippageModel | None = None,
    stop_slippage_rate: float = 0.0,
    execution_mode: ExecutionMode = ExecutionMode.SAME_BAR,
    execution_price: ExecutionPrice = ExecutionPrice.CLOSE,
    mark_price: ExecutionPrice = ExecutionPrice.PRICE,
    stop_fill_mode: StopFillMode = StopFillMode.STOP_PRICE,
    stop_level_basis: StopLevelBasis = StopLevelBasis.FILL_PRICE,
    trail_hwm_source: WaterMarkSource = WaterMarkSource.CLOSE,
    trail_include_entry_bar_extremes: bool = False,
    initial_hwm_source: InitialHwmSource = InitialHwmSource.FILL_PRICE,
    trail_stop_timing: TrailStopTiming = TrailStopTiming.LAGGED,
    allow_short_selling: bool = False,
    allow_leverage: bool = False,
    initial_margin: float = 0.5,
    long_maintenance_margin: float = 0.25,
    short_maintenance_margin: float = 0.30,
    fixed_margin_schedule: dict[str, tuple[float, float]] | None = None,
    margin_pct_schedule: dict[str, tuple[float, float]] | None = None,
    short_cash_policy: ShortCashPolicy = ShortCashPolicy.CREDIT,
    lock_notional_update_mode: LockNotionalUpdateMode = LockNotionalUpdateMode.POSITION_LEGS,
    execution_limits: ExecutionLimits | None = None,
    market_impact_model: MarketImpactModel | None = None,
    contract_specs: dict[str, ContractSpec] | None = None,
    share_type: ShareType = ShareType.INTEGER,
    fill_ordering: FillOrdering = FillOrdering.EXIT_FIRST,
    entry_order_priority: EntryOrderPriority = EntryOrderPriority.SUBMISSION,
    next_bar_submission_precheck: bool = False,
    next_bar_simple_cash_check: bool = False,
    buying_power_reservation: bool = False,
    next_bar_queue_shadow_validation: bool = False,
    immediate_fill: bool = False,
    reject_on_insufficient_cash: bool = True,
    skip_cash_validation: bool = False,
    cash_buffer_pct: float = 0.0,
    partial_fills_allowed: bool = False,
    rebalance_headroom_pct: float = 1.0,
    missing_price_policy: MissingPricePolicy = MissingPricePolicy.SKIP,
    late_asset_policy: LateAssetPolicy = LateAssetPolicy.ALLOW,
    late_asset_min_bars: int = 1,
    settlement_delay: int = 0,
    settlement_reduces_buying_power: bool = True,
):
    # Runtime imports for accounting classes.
    # These are imported here rather than at module level because:
    # 1. The package __init__.py imports Broker, creating a potential import order issue
    # 2. TYPE_CHECKING block above provides type hints for static analysis
    # 3. This pattern allows mypy/pyright to validate types without runtime circular import
    from .accounting import (
        AccountState,
        Gatekeeper,
        UnifiedAccountPolicy,
    )

    self.initial_cash = initial_cash
    # Note: self.cash is now a property delegating to self.account.cash (Bug #5 fix)
    self.commission_model = commission_model or NoCommission()
    self.slippage_model = slippage_model or NoSlippage()
    self.stop_slippage_rate = stop_slippage_rate
    self.execution_mode = execution_mode
    self.execution_price = execution_price
    self.mark_price = mark_price
    self.stop_fill_mode = stop_fill_mode
    self.stop_level_basis = stop_level_basis
    self.trail_hwm_source = trail_hwm_source
    self.trail_include_entry_bar_extremes = trail_include_entry_bar_extremes
    self.initial_hwm_source = initial_hwm_source
    self.trail_stop_timing = trail_stop_timing
    self.share_type = share_type
    self.fill_ordering = fill_ordering
    self.entry_order_priority = entry_order_priority
    self.next_bar_submission_precheck = next_bar_submission_precheck
    self.next_bar_simple_cash_check = next_bar_simple_cash_check
    self.buying_power_reservation = buying_power_reservation
    self.next_bar_queue_shadow_validation = next_bar_queue_shadow_validation
    self.immediate_fill = immediate_fill
    self.reject_on_insufficient_cash = reject_on_insufficient_cash
    self.skip_cash_validation = skip_cash_validation
    self.cash_buffer_pct = cash_buffer_pct
    self.partial_fills_allowed = partial_fills_allowed
    self.rebalance_headroom_pct = rebalance_headroom_pct
    self.missing_price_policy = missing_price_policy
    self.late_asset_policy = late_asset_policy
    self.late_asset_min_bars = late_asset_min_bars
    self.settlement_delay = settlement_delay
    self.settlement_reduces_buying_power = settlement_reduces_buying_power
    self._market_state = MarketState()
    self._order_state = OrderState()
    self._risk_state = RiskState()
    self._execution_journal = ExecutionJournal()

    # Auto-populate margin schedules from ContractSpec settings
    # This lets users specify margin once on ContractSpec rather than duplicating
    # it in both ContractSpec and BacktestConfig.fixed_margin_schedule.
    effective_margin_schedule = dict(fixed_margin_schedule or {})
    effective_margin_pct_schedule = dict(margin_pct_schedule or {})
    if contract_specs:
        for symbol, spec in contract_specs.items():
            if spec.margin is not None and symbol not in effective_margin_schedule:
                # Use spec.margin as initial margin, 50% as maintenance (industry standard)
                effective_margin_schedule[symbol] = (spec.margin, spec.margin * 0.5)
            if spec.margin_pct is not None and symbol not in effective_margin_pct_schedule:
                effective_margin_pct_schedule[symbol] = spec.margin_pct

    # Create AccountState with UnifiedAccountPolicy
    policy: AccountPolicy = UnifiedAccountPolicy(
        allow_short_selling=allow_short_selling,
        allow_leverage=allow_leverage,
        initial_margin=initial_margin,
        long_maintenance_margin=long_maintenance_margin,
        short_maintenance_margin=short_maintenance_margin,
        fixed_margin_schedule=effective_margin_schedule or None,
        margin_pct_schedule=effective_margin_pct_schedule or None,
        short_cash_policy=short_cash_policy.value,
    )

    self.account = AccountState(initial_cash=initial_cash, policy=policy)
    # Derive account_type string from flags for backward compat
    if allow_leverage:
        self.account_type = "margin"
    elif allow_short_selling:
        self.account_type = "crypto"
    else:
        self.account_type = "cash"
    self.allow_short_selling = allow_short_selling
    self.allow_leverage = allow_leverage
    self.initial_margin = initial_margin
    self.long_maintenance_margin = long_maintenance_margin
    self.short_maintenance_margin = short_maintenance_margin
    self.fixed_margin_schedule = effective_margin_schedule
    self.margin_pct_schedule = effective_margin_pct_schedule
    self.short_cash_policy = short_cash_policy
    self.lock_notional_update_mode = lock_notional_update_mode

    # Create Gatekeeper for order validation
    self.gatekeeper = Gatekeeper(
        self.account,
        self.commission_model,
        cash_buffer_pct=self.cash_buffer_pct,
        settlement_reduces_buying_power=self.settlement_reduces_buying_power,
        multiplier_resolver=self.get_multiplier,
    )

    self._rebalance_counter = 0
    self._lifecycle_transaction: Any | None = None
    self._active_lifecycle_phase: LifecyclePhase | None = None
    self._preopen_target_manager: PreOpenTargetManager | None = None
    self._completion_validators: dict[int, Callable[[int], None]] = {}

    # Execution model (volume limits and market impact)
    self.execution_limits = execution_limits  # ExecutionLimits instance
    self.market_impact_model = market_impact_model  # MarketImpactModel instance

    # Contract specifications (for futures and other derivatives)
    self._contract_specs: dict[str, ContractSpec] = contract_specs or {}

    # Fill execution (extracted from _execute_fill)
    self._fill_executor = FillExecutor(
        self,
        account=self.account,
        market=self._market_state,
        orders=self._order_state,
        risk=self._risk_state,
        journal=self._execution_journal,
        record_pnl=self._record_pnl_event,
    )

    # Per-asset trading statistics for stateful decision-making
    self._asset_stats: dict[str, AssetTradingStats] = {}
    self._stats_config = StatsConfig()
    self._session_config = None  # Optional SessionConfig for session boundary detection
    self._session_boundary: tuple[ZoneInfo, ZoneInfo, int, int] | None = None
    self._last_session_id: int | None = None  # Track current session for boundary detection

    # Extracted orchestration components (Phase B1 alpha-reset)
    self._fill_engine = FillEngine(
        self,
        market=self._market_state,
        orders=self._order_state,
        executor=self._fill_executor,
    )
    self._fill_executor.fill_engine = self._fill_engine
    self._order_book = OrderBook(
        self,
        account=self.account,
        market=self._market_state,
        orders=self._order_state,
        risk=self._risk_state,
        fill_engine=self._fill_engine,
    )
    self._risk_engine = RiskEngine(
        self,
        account=self.account,
        market=self._market_state,
        risk=self._risk_state,
        fill_engine=self._fill_engine,
    )
    self._execution_engine = ExecutionEngine(
        self,
        account=self.account,
        market=self._market_state,
        orders=self._order_state,
        fill_engine=self._fill_engine,
    )
    self._portfolio_ledger = PortfolioLedger(
        self,
        account=self.account,
        market=self._market_state,
        orders=self._order_state,
    )

orders property writable 

orders

Return the complete order history owned by OrderState.

pending_orders property writable 

pending_orders

Return the pending-order collection owned by OrderState.

positions property writable 

positions

Return the open-position ledger owned by AccountState.

cash property writable 

cash

Return the cash balance owned by the account ledger.

The value can be negative for a configured margin account. Assignment is retained for compatibility and updates the same account ledger.

trades property writable 

trades

Return the trade journal owned by ExecutionJournal.

fills property writable 

fills

Return the fill journal owned by ExecutionJournal.

last_rejection_reason property 

last_rejection_reason

Get reason for most recent order rejection.

Returns:

Type Description
str | None

Rejection reason string, or None if no orders have been rejected

from_config classmethod 

from_config(
    config,
    execution_limits=None,
    market_impact_model=None,
    contract_specs=None,
)

Create Broker from BacktestConfig.

This is the recommended way to create a Broker. All settings come from the BacktestConfig, ensuring consistency with Engine and other components.

Parameters:

Name Type Description Default
config BacktestConfig

BacktestConfig with all behavioral settings

 required
execution_limits ExecutionLimits | None

Optional execution limits (not in config)

 None
market_impact_model MarketImpactModel | None

Optional market impact model (not in config)

 None
contract_specs dict[str, ContractSpec] | None

Optional contract specifications (not in config)

 None

Returns:

Type Description
Broker

Configured Broker instance
Example

config = BacktestConfig.from_preset("backtrader") broker = Broker.from_config(config)

Source code in src/ml4t/backtest/broker.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
@classmethod
def from_config(
    cls,
    config: BacktestConfig,
    execution_limits: ExecutionLimits | None = None,
    market_impact_model: MarketImpactModel | None = None,
    contract_specs: dict[str, ContractSpec] | None = None,
) -> Broker:
    """Create Broker from BacktestConfig.

    This is the recommended way to create a Broker. All settings come from
    the BacktestConfig, ensuring consistency with Engine and other components.

    Args:
        config: BacktestConfig with all behavioral settings
        execution_limits: Optional execution limits (not in config)
        market_impact_model: Optional market impact model (not in config)
        contract_specs: Optional contract specifications (not in config)

    Returns:
        Configured Broker instance

    Example:
        config = BacktestConfig.from_preset("backtrader")
        broker = Broker.from_config(config)
    """
    from .config import CommissionType, SlippageType
    from .models import (
        CombinedCommission,
        FixedSlippage,
        NoCommission,
        NoSlippage,
        PercentageCommission,
        PercentageSlippage,
        PerShareCommission,
        SpreadSlippage,
        TieredCommission,
        VolumeShareSlippage,
    )

    config._validate_for_execution()

    effective_commission_type = config.commission_type
    if effective_commission_type == CommissionType.NONE:
        if config.commission_per_share > 0:
            effective_commission_type = CommissionType.PER_SHARE
        elif config.commission_per_trade > 0:
            effective_commission_type = CommissionType.PER_TRADE
        elif config.commission_rate > 0:
            effective_commission_type = CommissionType.PERCENTAGE

    effective_slippage_type = config.slippage_type
    if effective_slippage_type == SlippageType.NONE:
        if config.slippage_spread > 0 or config.slippage_spread_by_asset:
            effective_slippage_type = SlippageType.SPREAD
        elif config.slippage_fixed > 0:
            effective_slippage_type = SlippageType.FIXED
        elif config.slippage_rate > 0:
            effective_slippage_type = SlippageType.PERCENTAGE

    # Build commission model from config
    commission_model = None
    if effective_commission_type == CommissionType.PERCENTAGE:
        commission_model = PercentageCommission(rate=config.commission_rate)
    elif effective_commission_type == CommissionType.PER_SHARE:
        commission_model = PerShareCommission(
            per_share=config.commission_per_share,
            minimum=config.commission_minimum,
        )
    elif effective_commission_type == CommissionType.PER_TRADE:
        commission_model = CombinedCommission(fixed=config.commission_per_trade)
    elif effective_commission_type == CommissionType.TIERED:
        commission_model = TieredCommission(
            tiers=[(float("inf"), config.commission_rate)],
        )
    elif effective_commission_type == CommissionType.NONE:
        commission_model = NoCommission()

    # Build slippage model from config
    slippage_model = None
    if effective_slippage_type == SlippageType.PERCENTAGE:
        slippage_model = PercentageSlippage(rate=config.slippage_rate)
    elif effective_slippage_type == SlippageType.FIXED:
        slippage_model = FixedSlippage(amount=config.slippage_fixed)
    elif effective_slippage_type == SlippageType.SPREAD:
        slippage_model = SpreadSlippage(
            spread=config.slippage_spread,
            asset_spreads=config.slippage_spread_by_asset,
            convention=config.slippage_spread_convention.value,
        )
    elif effective_slippage_type == SlippageType.VOLUME_BASED:
        slippage_model = VolumeShareSlippage(impact_factor=config.slippage_rate)
    elif effective_slippage_type == SlippageType.NONE:
        slippage_model = NoSlippage()

    return cls(
        initial_cash=config.initial_cash,
        commission_model=commission_model,
        slippage_model=slippage_model,
        stop_slippage_rate=config.stop_slippage_rate,
        execution_mode=config.execution_mode,
        execution_price=config.execution_price,
        mark_price=config.mark_price,
        stop_fill_mode=config.stop_fill_mode,
        stop_level_basis=config.stop_level_basis,
        trail_hwm_source=config.trail_hwm_source,
        trail_include_entry_bar_extremes=config.trail_include_entry_bar_extremes,
        initial_hwm_source=config.initial_hwm_source,
        trail_stop_timing=config.trail_stop_timing,
        allow_short_selling=config.allow_short_selling,
        allow_leverage=config.allow_leverage,
        initial_margin=config.initial_margin,
        long_maintenance_margin=config.long_maintenance_margin,
        short_maintenance_margin=config.short_maintenance_margin,
        fixed_margin_schedule=config.fixed_margin_schedule,
        margin_pct_schedule=config.margin_pct_schedule,
        short_cash_policy=config.short_cash_policy,
        lock_notional_update_mode=config.lock_notional_update_mode,
        execution_limits=execution_limits,
        market_impact_model=market_impact_model,
        contract_specs=contract_specs,
        share_type=config.share_type,
        fill_ordering=config.fill_ordering,
        entry_order_priority=config.entry_order_priority,
        next_bar_submission_precheck=config.next_bar_submission_precheck,
        next_bar_simple_cash_check=config.next_bar_simple_cash_check,
        buying_power_reservation=config.buying_power_reservation,
        next_bar_queue_shadow_validation=config.next_bar_queue_shadow_validation,
        immediate_fill=config.immediate_fill,
        reject_on_insufficient_cash=config.reject_on_insufficient_cash,
        skip_cash_validation=config.skip_cash_validation,
        cash_buffer_pct=config.cash_buffer_pct,
        partial_fills_allowed=config.partial_fills_allowed,
        rebalance_headroom_pct=config.rebalance_headroom_pct,
        missing_price_policy=config.missing_price_policy,
        late_asset_policy=config.late_asset_policy,
        late_asset_min_bars=config.late_asset_min_bars,
        settlement_delay=config.settlement_delay,
        settlement_reduces_buying_power=config.settlement_reduces_buying_power,
    )

submit_order 

submit_order(
    asset,
    quantity,
    side=None,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    trail_amount=None,
    _options=None,
)

Submit a new order to the broker.

Creates and queues an order for execution. Orders are validated by the Gatekeeper before fills to ensure account constraints are met.

Parameters:

Name Type Description Default
asset str

Asset symbol (e.g., "AAPL", "BTC-USD")

 required
quantity float

Number of shares/units. Positive = buy, negative = sell (if side is not specified)

 required
side OrderSide | None

OrderSide.BUY or OrderSide.SELL. If None, inferred from quantity sign

 None
order_type OrderType

Order type (MARKET, LIMIT, STOP, TRAILING_STOP)

 MARKET
limit_price float | None

Limit price for LIMIT orders

 None
stop_price float | None

Stop/trigger price for STOP orders

 None
trail_amount float | None

Trail distance for TRAILING_STOP orders

 None

Returns:

Type Description
Order | None

Order object if submitted successfully, None if rejected
Order | None

(e.g., same-bar re-entry after stop exit in VBT Pro mode)

Examples:

Market buy

order = broker.submit_order("AAPL", 100)

Market sell (using negative quantity)

order = broker.submit_order("AAPL", -100)

Limit buy

order = broker.submit_order("AAPL", 100, order_type=OrderType.LIMIT, limit_price=150.0)

Stop sell (stop-loss)

order = broker.submit_order("AAPL", -100, order_type=OrderType.STOP, stop_price=145.0)

Source code in src/ml4t/backtest/broker.py 
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
def submit_order(
    self,
    asset: str,
    quantity: float,
    side: OrderSide | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
    stop_price: float | None = None,
    trail_amount: float | None = None,
    _options: SubmitOrderOptions | None = None,
) -> Order | None:
    """Submit a new order to the broker.

    Creates and queues an order for execution. Orders are validated by the
    Gatekeeper before fills to ensure account constraints are met.

    Args:
        asset: Asset symbol (e.g., "AAPL", "BTC-USD")
        quantity: Number of shares/units. Positive = buy, negative = sell
                 (if side is not specified)
        side: OrderSide.BUY or OrderSide.SELL. If None, inferred from quantity sign
        order_type: Order type (MARKET, LIMIT, STOP, TRAILING_STOP)
        limit_price: Limit price for LIMIT orders
        stop_price: Stop/trigger price for STOP orders
        trail_amount: Trail distance for TRAILING_STOP orders

    Returns:
        Order object if submitted successfully, None if rejected
        (e.g., same-bar re-entry after stop exit in VBT Pro mode)

    Examples:
        # Market buy
        order = broker.submit_order("AAPL", 100)

        # Market sell (using negative quantity)
        order = broker.submit_order("AAPL", -100)

        # Limit buy
        order = broker.submit_order("AAPL", 100, order_type=OrderType.LIMIT,
                                    limit_price=150.0)

        # Stop sell (stop-loss)
        order = broker.submit_order("AAPL", -100, order_type=OrderType.STOP,
                                    stop_price=145.0)
    """
    self._capture_lifecycle_mutation(asset=asset)
    return self._order_book.submit_order(
        asset=asset,
        quantity=quantity,
        side=side,
        order_type=order_type,
        limit_price=limit_price,
        stop_price=stop_price,
        trail_amount=trail_amount,
        options=_options,
    )

submit_bracket 

submit_bracket(
    asset,
    quantity,
    take_profit,
    stop_loss,
    entry_type=MARKET,
    entry_limit=None,
    validate_prices=True,
)

Submit entry with take-profit and stop-loss.

Creates a bracket order with entry, take-profit limit, and stop-loss orders. The exit side is automatically determined from the entry direction.

Parameters:

Name Type Description Default
asset str

Asset symbol to trade

 required
quantity float

Position size (positive for long, negative for short)

 required
take_profit float

Take-profit price level (LIMIT order)

 required
stop_loss float

Stop-loss price level (STOP order)

 required
entry_type OrderType

Entry order type (default MARKET)

 MARKET
entry_limit float | None

Entry limit price (if entry_type is LIMIT)

 None
validate_prices bool

If True, validate that TP/SL prices are sensible for the position direction (default True)

 True

Returns:

Type Description
tuple[Order, Order, Order] | None

Tuple of (entry_order, take_profit_order, stop_loss_order) or None if any fails.

Raises:

Type Description
ValueError

If validate_prices=True and prices are inverted for direction.
Notes

For LONG entries (quantity > 0): - take_profit should be > reference_price (profit on up move) - stop_loss should be < reference_price (exit on down move)

For SHORT entries (quantity < 0): - take_profit should be < reference_price (profit on down move) - stop_loss should be > reference_price (exit on up move)

Reference price is entry_limit (if LIMIT order) or current market price.

Source code in src/ml4t/backtest/broker.py 
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
def submit_bracket(
    self,
    asset: str,
    quantity: float,
    take_profit: float,
    stop_loss: float,
    entry_type: OrderType = OrderType.MARKET,
    entry_limit: float | None = None,
    validate_prices: bool = True,
) -> tuple[Order, Order, Order] | None:
    """Submit entry with take-profit and stop-loss.

    Creates a bracket order with entry, take-profit limit, and stop-loss orders.
    The exit side is automatically determined from the entry direction.

    Args:
        asset: Asset symbol to trade
        quantity: Position size (positive for long, negative for short)
        take_profit: Take-profit price level (LIMIT order)
        stop_loss: Stop-loss price level (STOP order)
        entry_type: Entry order type (default MARKET)
        entry_limit: Entry limit price (if entry_type is LIMIT)
        validate_prices: If True, validate that TP/SL prices are sensible
                        for the position direction (default True)

    Returns:
        Tuple of (entry_order, take_profit_order, stop_loss_order) or None if any fails.

    Raises:
        ValueError: If validate_prices=True and prices are inverted for direction.

    Notes:
        For LONG entries (quantity > 0):
            - take_profit should be > reference_price (profit on up move)
            - stop_loss should be < reference_price (exit on down move)

        For SHORT entries (quantity < 0):
            - take_profit should be < reference_price (profit on down move)
            - stop_loss should be > reference_price (exit on up move)

        Reference price is entry_limit (if LIMIT order) or current market price.
    """
    import warnings

    entry = self.submit_order(asset, quantity, order_type=entry_type, limit_price=entry_limit)
    if entry is None:
        return None

    # Derive exit side from entry direction (Bug #4 fix)
    # Long entry (BUY) -> SELL to exit; Short entry (SELL) -> BUY to cover
    exit_side = OrderSide.SELL if entry.side == OrderSide.BUY else OrderSide.BUY
    exit_qty = abs(quantity)

    # Validate bracket prices if requested
    if validate_prices:
        ref_price = entry_limit if entry_limit is not None else self._current_prices.get(asset)
        if ref_price is not None:
            is_long = entry.side == OrderSide.BUY

            if is_long:
                # Long: TP should be above entry, SL should be below
                if take_profit <= ref_price:
                    warnings.warn(
                        f"Bracket order for LONG {asset}: take_profit ({take_profit}) <= "
                        f"entry ({ref_price}). TP should be above entry for longs.",
                        UserWarning,
                        stacklevel=2,
                    )
                if stop_loss >= ref_price:
                    warnings.warn(
                        f"Bracket order for LONG {asset}: stop_loss ({stop_loss}) >= "
                        f"entry ({ref_price}). SL should be below entry for longs.",
                        UserWarning,
                        stacklevel=2,
                    )
            else:
                # Short: TP should be below entry, SL should be above
                if take_profit >= ref_price:
                    warnings.warn(
                        f"Bracket order for SHORT {asset}: take_profit ({take_profit}) >= "
                        f"entry ({ref_price}). TP should be below entry for shorts.",
                        UserWarning,
                        stacklevel=2,
                    )
                if stop_loss <= ref_price:
                    warnings.warn(
                        f"Bracket order for SHORT {asset}: stop_loss ({stop_loss}) <= "
                        f"entry ({ref_price}). SL should be above entry for shorts.",
                        UserWarning,
                        stacklevel=2,
                    )

    tp = self.submit_order(asset, exit_qty, exit_side, OrderType.LIMIT, limit_price=take_profit)
    if tp is None:
        return None
    tp.parent_id = entry.order_id

    sl = self.submit_order(asset, exit_qty, exit_side, OrderType.STOP, stop_price=stop_loss)
    if sl is None:
        return None
    sl.parent_id = entry.order_id

    return entry, tp, sl

buy 

buy(
    asset,
    shares=None,
    contracts=None,
    amount=None,
    dollars=None,
    order_type=MARKET,
    limit_price=None,
)

Buy an asset with explicit quantity specification.

Provides a clean API for buying with different quantity types: - shares: Number of shares (equities) - contracts: Number of contracts (futures) - amount: Base currency amount (crypto) - dollars: Dollar value to buy (any asset)

Exactly one quantity parameter must be provided.

Parameters:

Name Type Description Default
asset str

Asset symbol

 required
shares float | None

Number of shares to buy (equities)

 None
contracts int | None

Number of contracts to buy (futures)

 None
amount float | None

Base currency amount to buy (crypto)

 None
dollars float | None

Dollar value to buy (converted to quantity at current price)

 None
order_type OrderType

Order type (default MARKET)

 MARKET
limit_price float | None

Limit price for LIMIT orders

 None

Returns:

Type Description
Order | None

Order object if submitted, None if no current price or invalid params

Raises:

Type Description
ValueError

If zero or more than one quantity parameter is provided
Example

Buy 100 shares of AAPL

broker.buy("AAPL", shares=100)

Buy 2 ES futures contracts

broker.buy("ES", contracts=2)

Buy $5000 worth of BTC

broker.buy("BTC", dollars=5000)

Buy 0.5 BTC

broker.buy("BTC", amount=0.5)

Source code in src/ml4t/backtest/broker.py 
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
def buy(
    self,
    asset: str,
    shares: float | None = None,
    contracts: int | None = None,
    amount: float | None = None,
    dollars: float | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
) -> Order | None:
    """Buy an asset with explicit quantity specification.

    Provides a clean API for buying with different quantity types:
    - shares: Number of shares (equities)
    - contracts: Number of contracts (futures)
    - amount: Base currency amount (crypto)
    - dollars: Dollar value to buy (any asset)

    Exactly one quantity parameter must be provided.

    Args:
        asset: Asset symbol
        shares: Number of shares to buy (equities)
        contracts: Number of contracts to buy (futures)
        amount: Base currency amount to buy (crypto)
        dollars: Dollar value to buy (converted to quantity at current price)
        order_type: Order type (default MARKET)
        limit_price: Limit price for LIMIT orders

    Returns:
        Order object if submitted, None if no current price or invalid params

    Raises:
        ValueError: If zero or more than one quantity parameter is provided

    Example:
        # Buy 100 shares of AAPL
        broker.buy("AAPL", shares=100)

        # Buy 2 ES futures contracts
        broker.buy("ES", contracts=2)

        # Buy $5000 worth of BTC
        broker.buy("BTC", dollars=5000)

        # Buy 0.5 BTC
        broker.buy("BTC", amount=0.5)
    """
    return self._submit_side_order(
        OrderSide.BUY,
        asset,
        shares,
        contracts,
        amount,
        dollars,
        order_type,
        limit_price,
    )

sell 

sell(
    asset,
    shares=None,
    contracts=None,
    amount=None,
    dollars=None,
    order_type=MARKET,
    limit_price=None,
)

Sell an asset with explicit quantity specification.

Provides a clean API for selling with different quantity types: - shares: Number of shares (equities) - contracts: Number of contracts (futures) - amount: Base currency amount (crypto) - dollars: Dollar value to sell (any asset)

Exactly one quantity parameter must be provided.

Parameters:

Name Type Description Default
asset str

Asset symbol

 required
shares float | None

Number of shares to sell (equities)

 None
contracts int | None

Number of contracts to sell (futures)

 None
amount float | None

Base currency amount to sell (crypto)

 None
dollars float | None

Dollar value to sell (converted to quantity at current price)

 None
order_type OrderType

Order type (default MARKET)

 MARKET
limit_price float | None

Limit price for LIMIT orders

 None

Returns:

Type Description
Order | None

Order object if submitted, None if no current price or invalid params

Raises:

Type Description
ValueError

If zero or more than one quantity parameter is provided
Example

Sell 50 shares of AAPL

broker.sell("AAPL", shares=50)

Sell 1 ES futures contract

broker.sell("ES", contracts=1)

Sell $2500 worth of position

broker.sell("BTC", dollars=2500)

Source code in src/ml4t/backtest/broker.py 
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
def sell(
    self,
    asset: str,
    shares: float | None = None,
    contracts: int | None = None,
    amount: float | None = None,
    dollars: float | None = None,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
) -> Order | None:
    """Sell an asset with explicit quantity specification.

    Provides a clean API for selling with different quantity types:
    - shares: Number of shares (equities)
    - contracts: Number of contracts (futures)
    - amount: Base currency amount (crypto)
    - dollars: Dollar value to sell (any asset)

    Exactly one quantity parameter must be provided.

    Args:
        asset: Asset symbol
        shares: Number of shares to sell (equities)
        contracts: Number of contracts to sell (futures)
        amount: Base currency amount to sell (crypto)
        dollars: Dollar value to sell (converted to quantity at current price)
        order_type: Order type (default MARKET)
        limit_price: Limit price for LIMIT orders

    Returns:
        Order object if submitted, None if no current price or invalid params

    Raises:
        ValueError: If zero or more than one quantity parameter is provided

    Example:
        # Sell 50 shares of AAPL
        broker.sell("AAPL", shares=50)

        # Sell 1 ES futures contract
        broker.sell("ES", contracts=1)

        # Sell $2500 worth of position
        broker.sell("BTC", dollars=2500)
    """
    return self._submit_side_order(
        OrderSide.SELL,
        asset,
        shares,
        contracts,
        amount,
        dollars,
        order_type,
        limit_price,
    )

close_position 

close_position(asset, order_type=MARKET, _options=None)

Close an open position for the given asset.

Submits an order to fully close the position.

Parameters:

Name Type Description Default
asset str

Asset symbol to close

 required
order_type OrderType

Exit order type (default MARKET). Use OrderType.MOC for market-on-close flattening.

 MARKET

Returns:

Type Description
Order | None

Order object if position exists and order submitted, None otherwise
Example

Close AAPL position

order = broker.close_position("AAPL")

Flatten at the bar close

order = broker.close_position("AAPL", order_type=OrderType.MOC)

Source code in src/ml4t/backtest/broker.py 
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
def close_position(
    self,
    asset: str,
    order_type: OrderType = OrderType.MARKET,
    _options: SubmitOrderOptions | None = None,
) -> Order | None:
    """Close an open position for the given asset.

    Submits an order to fully close the position.

    Args:
        asset: Asset symbol to close
        order_type: Exit order type (default MARKET). Use `OrderType.MOC`
            for market-on-close flattening.

    Returns:
        Order object if position exists and order submitted, None otherwise

    Example:
        # Close AAPL position
        order = broker.close_position("AAPL")

        # Flatten at the bar close
        order = broker.close_position("AAPL", order_type=OrderType.MOC)
    """
    pos = self.positions.get(asset)
    if pos and pos.quantity != 0:
        side = OrderSide.SELL if pos.quantity > 0 else OrderSide.BUY
        return self.submit_order(
            asset,
            abs(pos.quantity),
            side,
            order_type=order_type,
            _options=_options,
        )
    return None

reduce_position 

reduce_position(
    asset, fraction, order_type=MARKET, limit_price=None
)

Reduce an existing position by a fraction.

Sells (for long) or covers (for short) a percentage of the current position.

Parameters:

Name Type Description Default
asset str

Asset symbol to reduce

 required
fraction float

Fraction to exit (0.5 = sell half, 0.25 = sell quarter) Must be between 0 and 1 (exclusive of 0, inclusive of 1)

 required
order_type OrderType

Order type (default MARKET)

 MARKET
limit_price float | None

Limit price for LIMIT orders

 None

Returns:

Type Description
Order | None

Order object if position exists and order submitted, None otherwise

Raises:

Type Description
ValueError

If fraction is not in (0, 1]
Example

Sell half of AAPL position

order = broker.reduce_position("AAPL", fraction=0.5)

Sell 25% of position with limit

order = broker.reduce_position("AAPL", fraction=0.25, order_type=OrderType.LIMIT, limit_price=155.0)

Source code in src/ml4t/backtest/broker.py 
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
def reduce_position(
    self,
    asset: str,
    fraction: float,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
) -> Order | None:
    """Reduce an existing position by a fraction.

    Sells (for long) or covers (for short) a percentage of the current position.

    Args:
        asset: Asset symbol to reduce
        fraction: Fraction to exit (0.5 = sell half, 0.25 = sell quarter)
                 Must be between 0 and 1 (exclusive of 0, inclusive of 1)
        order_type: Order type (default MARKET)
        limit_price: Limit price for LIMIT orders

    Returns:
        Order object if position exists and order submitted, None otherwise

    Raises:
        ValueError: If fraction is not in (0, 1]

    Example:
        # Sell half of AAPL position
        order = broker.reduce_position("AAPL", fraction=0.5)

        # Sell 25% of position with limit
        order = broker.reduce_position("AAPL", fraction=0.25,
                                       order_type=OrderType.LIMIT,
                                       limit_price=155.0)
    """
    if fraction <= 0 or fraction > 1:
        raise ValueError(f"fraction must be in (0, 1], got {fraction}")

    pos = self.positions.get(asset)
    if pos is None or pos.quantity == 0:
        return None

    # Calculate quantity to exit
    exit_qty = abs(pos.quantity) * fraction

    # Determine exit side (opposite of position direction)
    side = OrderSide.SELL if pos.quantity > 0 else OrderSide.BUY

    return self.submit_order(asset, exit_qty, side, order_type, limit_price)

reduce_all_positions 

reduce_all_positions(fraction, reason, order_type=MARKET)

Cancel pending orders and reduce each open position atomically.

Source code in src/ml4t/backtest/broker.py 
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
def reduce_all_positions(
    self,
    fraction: float,
    reason: str,
    order_type: OrderType = OrderType.MARKET,
) -> list[Order]:
    """Cancel pending orders and reduce each open position atomically."""
    if not isfinite(fraction) or not 0 < fraction <= 1:
        raise ValueError(f"fraction must be a finite value in (0, 1], got {fraction!r}")

    active = [(asset, pos.quantity) for asset, pos in self.positions.items() if pos.quantity]
    if not active:
        return []
    for asset, quantity in active:
        exit_qty = abs(quantity) * fraction
        if not isfinite(exit_qty) or exit_qty <= self._order_book._MIN_ORDER_SIZE:
            raise ValueError(f"reduction quantity for {asset} is not executable")
        if self.share_type == ShareType.INTEGER and int(exit_qty) == 0:
            raise ValueError(f"reduction quantity for {asset} rounds to zero")

    state = self._snapshot_lifecycle_state(
        all_positions=True, all_pending_orders=True, all_asset_stats=True
    )
    try:
        for order in list(self.pending_orders):
            self.cancel_order(order.order_id)

        reductions: list[Order] = []
        for asset, quantity in active:
            exit_side = OrderSide.SELL if quantity > 0 else OrderSide.BUY
            order = self.submit_order(
                asset,
                abs(quantity) * fraction,
                exit_side,
                order_type,
                _options=SubmitOrderOptions(
                    risk_exit_reason=f"risk reduction: {reason}",
                    exit_reason=ExitReason.RISK_LIQUIDATION,
                ),
            )
            if order is None or order.status is OrderStatus.REJECTED:
                raise RuntimeError(f"risk reduction order was not accepted for {asset}")
            reductions.append(order)
        return reductions
    except Exception:
        self._restore_lifecycle_state(state)
        raise

flatten_all_positions 

flatten_all_positions(reason, order_type=MARKET)

Cancel pending orders and submit exits for every open position.

Parameters:

Name Type Description Default
reason str

Human-readable risk reason for the liquidation

 required
order_type OrderType

Exit order type to use for the flattening orders

 MARKET

Returns:

Type Description
list[Order]

List of submitted exit orders
Source code in src/ml4t/backtest/broker.py 
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
def flatten_all_positions(
    self,
    reason: str,
    order_type: OrderType = OrderType.MARKET,
) -> list[Order]:
    """Cancel pending orders and submit exits for every open position.

    Args:
        reason: Human-readable risk reason for the liquidation
        order_type: Exit order type to use for the flattening orders

    Returns:
        List of submitted exit orders
    """
    for order in list(self.pending_orders):
        self.cancel_order(order.order_id)

    liquidations: list[Order] = []
    for asset in list(self.positions):
        order = self.close_position(
            asset,
            order_type=order_type,
            _options=SubmitOrderOptions(
                risk_exit_reason=reason,
                exit_reason=ExitReason.RISK_LIQUIDATION,
            ),
        )
        if order is None:
            continue
        liquidations.append(order)

    return liquidations

order_target_percent 

order_target_percent(
    asset,
    target_percent,
    order_type=MARKET,
    limit_price=None,
)

Order to achieve target portfolio weight.

Calculates the order quantity needed to reach the target percentage of total portfolio value for this asset. Weights can exceed 1.0 for leveraged portfolios (e.g., futures, margin accounts). The gatekeeper validates whether the account has sufficient buying power.

Parameters:

Name Type Description Default
asset str

Asset symbol

 required
target_percent float

Target weight as decimal (0.10 = 10% of portfolio). Can exceed 1.0 for leveraged positions if allow_leverage=True.

 required
order_type OrderType

Order type (default MARKET)

 MARKET
limit_price float | None

Limit price for LIMIT orders

 None

Returns:

Type Description
Order | None

Submitted order, or None if no order needed or rejected
Example

Target 10% of portfolio in AAPL

broker.order_target_percent("AAPL", 0.10)

Target 0% (close position)

broker.order_target_percent("AAPL", 0.0)

Leveraged: target 150% in ES futures (requires allow_leverage=True)

broker.order_target_percent("ES", 1.50)

Source code in src/ml4t/backtest/broker.py 
2113
2114
2115
2116
2117
2118
2119
2120
2121
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
def order_target_percent(
    self,
    asset: str,
    target_percent: float,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
) -> Order | None:
    """Order to achieve target portfolio weight.

    Calculates the order quantity needed to reach the target percentage
    of total portfolio value for this asset. Weights can exceed 1.0 for
    leveraged portfolios (e.g., futures, margin accounts). The gatekeeper
    validates whether the account has sufficient buying power.

    Args:
        asset: Asset symbol
        target_percent: Target weight as decimal (0.10 = 10% of portfolio).
            Can exceed 1.0 for leveraged positions if allow_leverage=True.
        order_type: Order type (default MARKET)
        limit_price: Limit price for LIMIT orders

    Returns:
        Submitted order, or None if no order needed or rejected

    Example:
        # Target 10% of portfolio in AAPL
        broker.order_target_percent("AAPL", 0.10)

        # Target 0% (close position)
        broker.order_target_percent("AAPL", 0.0)

        # Leveraged: target 150% in ES futures (requires allow_leverage=True)
        broker.order_target_percent("ES", 1.50)
    """

    portfolio_value = self.get_account_value()
    if portfolio_value <= 0:
        return None

    price = self._current_prices.get(asset)
    if price is None or price <= 0:
        return None

    target_value = portfolio_value * target_percent
    return self._order_to_target_value(asset, target_value, price, order_type, limit_price)

order_target_value 

order_target_value(
    asset, target_value, order_type=MARKET, limit_price=None
)

Order to achieve target position value.

Calculates the order quantity needed to reach the target dollar value for this position.

Parameters:

Name Type Description Default
asset str

Asset symbol

 required
target_value float

Target position value in dollars (negative for short)

 required
order_type OrderType

Order type (default MARKET)

 MARKET
limit_price float | None

Limit price for LIMIT orders

 None

Returns:

Type Description
Order | None

Submitted order, or None if no order needed or rejected
Example

Target $10,000 position in AAPL

broker.order_target_value("AAPL", 10000)

Target short $5,000

broker.order_target_value("AAPL", -5000)

Source code in src/ml4t/backtest/broker.py 
2159
2160
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
2171
2172
2173
2174
2175
2176
2177
2178
2179
2180
2181
2182
2183
2184
2185
2186
2187
2188
2189
2190
2191
def order_target_value(
    self,
    asset: str,
    target_value: float,
    order_type: OrderType = OrderType.MARKET,
    limit_price: float | None = None,
) -> Order | None:
    """Order to achieve target position value.

    Calculates the order quantity needed to reach the target dollar value
    for this position.

    Args:
        asset: Asset symbol
        target_value: Target position value in dollars (negative for short)
        order_type: Order type (default MARKET)
        limit_price: Limit price for LIMIT orders

    Returns:
        Submitted order, or None if no order needed or rejected

    Example:
        # Target $10,000 position in AAPL
        broker.order_target_value("AAPL", 10000)

        # Target short $5,000
        broker.order_target_value("AAPL", -5000)
    """
    price = self._current_prices.get(asset)
    if price is None or price <= 0:
        return None

    return self._order_to_target_value(asset, target_value, price, order_type, limit_price)

rebalance_to_weights 

rebalance_to_weights(target_weights, order_type=MARKET)

Rebalance portfolio to target weights.

Calculates orders needed to achieve target portfolio allocation. Processes sells before buys to free up capital.

Parameters:

Name Type Description Default
target_weights dict[str, float]

Dict of {asset: weight} where weights are decimals (0.10 = 10%). Weights should sum to <= 1.0.

 required
order_type OrderType

Order type for all orders (default MARKET)

 MARKET

Returns:

Type Description
list[Order]

List of submitted orders (may include None for rejected orders)
Example

Equal weight three stocks

broker.rebalance_to_weights({ "AAPL": 0.33, "GOOGL": 0.33, "MSFT": 0.34, })

Source code in src/ml4t/backtest/broker.py 
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
2305
2306
2307
2308
2309
2310
2311
2312
2313
2314
2315
2316
2317
2318
2319
2320
2321
2322
2323
2324
2325
2326
2327
2328
2329
2330
2331
2332
2333
2334
2335
2336
2337
2338
2339
2340
2341
2342
2343
2344
2345
2346
2347
2348
2349
2350
2351
2352
2353
2354
2355
2356
2357
2358
2359
2360
2361
2362
2363
2364
def rebalance_to_weights(
    self,
    target_weights: dict[str, float],
    order_type: OrderType = OrderType.MARKET,
) -> list[Order]:
    """Rebalance portfolio to target weights.

    Calculates orders needed to achieve target portfolio allocation.
    Processes sells before buys to free up capital.

    Args:
        target_weights: Dict of {asset: weight} where weights are decimals
                       (0.10 = 10%). Weights should sum to <= 1.0.
        order_type: Order type for all orders (default MARKET)

    Returns:
        List of submitted orders (may include None for rejected orders)

    Example:
        # Equal weight three stocks
        broker.rebalance_to_weights({
            "AAPL": 0.33,
            "GOOGL": 0.33,
            "MSFT": 0.34,
        })
    """
    portfolio_value = self.get_account_value()
    if portfolio_value <= 0:
        return []

    orders: list[Order] = []
    sells: list[tuple[str, float]] = []  # (asset, target_value)
    buys: list[tuple[str, float]] = []  # (asset, target_value)
    rebalance_id: str | None = None

    scaled_weights = {
        asset: weight * self.rebalance_headroom_pct for asset, weight in target_weights.items()
    }

    def resolve_price(asset: str) -> float | None:
        price = self._current_prices.get(asset)
        if price is not None and price > 0:
            return price
        if self.missing_price_policy == MissingPricePolicy.USE_LAST:
            last = self._last_prices.get(asset)
            if last is not None and last > 0:
                return last
        return None

    def allows_trading(asset: str) -> bool:
        if self.late_asset_policy != LateAssetPolicy.REQUIRE_HISTORY:
            return True
        return self._asset_bars_seen.get(asset, 0) >= self.late_asset_min_bars

    def rebalance_options() -> SubmitOrderOptions:
        nonlocal rebalance_id
        if rebalance_id is None:
            rebalance_id = self._next_rebalance_id()
        return SubmitOrderOptions(rebalance_id=rebalance_id)

    # Calculate target values and categorize as buys or sells
    for asset, weight in scaled_weights.items():
        if not allows_trading(asset):
            continue
        price = resolve_price(asset)
        if price is None:
            continue

        target_value = portfolio_value * weight

        pos = self.positions.get(asset)
        # Bug #2 fix: Include contract multiplier in value calculations
        multiplier = self.get_multiplier(asset)
        current_value = pos.quantity * price * multiplier if pos and pos.quantity != 0 else 0.0

        delta = target_value - current_value
        if abs(delta) < 0.01:  # Less than 1 cent
            continue

        if delta < 0:
            sells.append((asset, target_value))
        else:
            buys.append((asset, target_value))

    # Also close positions not in target weights
    for asset, pos in self.positions.items():
        if pos.quantity != 0 and asset not in scaled_weights:
            sells.append((asset, 0.0))

    # Process sells first (frees capital for buys)
    for asset, target_value in sells:
        price = resolve_price(asset)
        if price is not None:
            order = self._order_to_target_value(
                asset,
                target_value,
                price,
                order_type,
                None,
                rebalance_options(),
            )
            if order:
                orders.append(order)

    # Then process buys
    for asset, target_value in buys:
        price = resolve_price(asset)
        if price is not None:
            order = self._order_to_target_value(
                asset,
                target_value,
                price,
                order_type,
                None,
                rebalance_options(),
            )
            if order:
                orders.append(order)

    return orders

register_target_intent 

register_target_intent(intent, *, position_rules=None)

Register an idempotent target for a causal opening phase.

Source code in src/ml4t/backtest/broker.py 
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
def register_target_intent(
    self,
    intent: CanonicalTargetIntent,
    *,
    position_rules: PositionRule | Mapping[str, PositionRule] | None = None,
) -> CanonicalTargetIntent:
    """Register an idempotent target for a causal opening phase."""
    self._capture_lifecycle_mutation(target_intents=True)
    if self._preopen_target_manager is None:
        raise RuntimeError("target intents require an Engine-configured broker")
    return self._preopen_target_manager.register(
        intent,
        position_rules=position_rules,
        active_phase=self._active_lifecycle_phase,
    )

register_position_rule_policy 

register_position_rule_policy(policy_id, rules)

Bind a position-rule implementation to a portable policy identity.

Source code in src/ml4t/backtest/broker.py 
1441
1442
1443
1444
1445
1446
def register_position_rule_policy(self, policy_id: str, rules: PositionRule) -> None:
    """Bind a position-rule implementation to a portable policy identity."""
    self._capture_lifecycle_mutation(target_intents=True)
    if self._preopen_target_manager is None:
        raise RuntimeError("position rule policies require an Engine-configured broker")
    self._preopen_target_manager.register_position_rule_policy(policy_id, rules)

get_target_intents 

get_target_intents()

Return accepted canonical target intents.

Source code in src/ml4t/backtest/broker.py 
1448
1449
1450
1451
1452
def get_target_intents(self) -> tuple[CanonicalTargetIntent, ...]:
    """Return accepted canonical target intents."""
    if self._preopen_target_manager is None:
        return ()
    return self._preopen_target_manager.targets

get_child_order_intents 

get_child_order_intents()

Return canonical child intents derived from accepted targets.

Source code in src/ml4t/backtest/broker.py 
1454
1455
1456
1457
1458
def get_child_order_intents(self) -> tuple[CanonicalChildOrderIntent, ...]:
    """Return canonical child intents derived from accepted targets."""
    if self._preopen_target_manager is None:
        return ()
    return self._preopen_target_manager.children

get_intent_reconciliations 

get_intent_reconciliations()

Return retained child fill and remainder evidence.

Source code in src/ml4t/backtest/broker.py 
1460
1461
1462
1463
1464
def get_intent_reconciliations(self) -> tuple[IntentReconciliation, ...]:
    """Return retained child fill and remainder evidence."""
    if self._preopen_target_manager is None:
        return ()
    return self._preopen_target_manager.reconciliations

get_target_rule_reconciliations 

get_target_rule_reconciliations()

Return retained per-target rule transition evidence.

Source code in src/ml4t/backtest/broker.py 
1466
1467
1468
1469
1470
def get_target_rule_reconciliations(self) -> tuple[TargetRuleReconciliation, ...]:
    """Return retained per-target rule transition evidence."""
    if self._preopen_target_manager is None:
        return ()
    return self._preopen_target_manager.target_rule_reconciliations

export_target_intent_state 

export_target_intent_state()

Serialize accepted target, child, and reconciliation state.

Source code in src/ml4t/backtest/broker.py 
1472
1473
1474
1475
1476
def export_target_intent_state(self) -> dict[str, Any]:
    """Serialize accepted target, child, and reconciliation state."""
    if self._preopen_target_manager is None:
        return {}
    return self._preopen_target_manager.to_state()

restore_target_intent_state 

restore_target_intent_state(state)

Restore target intent state before strategy initialization.

Source code in src/ml4t/backtest/broker.py 
1478
1479
1480
1481
1482
1483
1484
def restore_target_intent_state(self, state: dict[str, Any]) -> None:
    """Restore target intent state before strategy initialization."""
    if self._active_lifecycle_phase is not None:
        raise RuntimeError("target intent state cannot be restored during a lifecycle callback")
    if self._preopen_target_manager is None:
        raise RuntimeError("target intents require an Engine-configured broker")
    self._preopen_target_manager.restore_state(state)

update_order 

update_order(order_id, **kwargs)

Update pending order parameters.

Only the following fields can be updated: - quantity: Order size - limit_price: Limit price for LIMIT orders - stop_price: Stop/trigger price for STOP orders - trail_amount: Trail distance for TRAILING_STOP orders

Parameters:

Name Type Description Default
order_id str

ID of the order to update

 required
**kwargs Unpack[OrderUpdate]

Fields to update

 {}

Returns:

Type Description
bool

True if order was found and updated, False otherwise

Raises:

Type Description
ValueError

If attempting to update non-updatable fields
Source code in src/ml4t/backtest/broker.py 
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
def update_order(self, order_id: str, **kwargs: Unpack[OrderUpdate]) -> bool:
    """Update pending order parameters.

    Only the following fields can be updated:
    - quantity: Order size
    - limit_price: Limit price for LIMIT orders
    - stop_price: Stop/trigger price for STOP orders
    - trail_amount: Trail distance for TRAILING_STOP orders

    Args:
        order_id: ID of the order to update
        **kwargs: Fields to update

    Returns:
        True if order was found and updated, False otherwise

    Raises:
        ValueError: If attempting to update non-updatable fields
    """
    self._capture_lifecycle_mutation(order_id=order_id)
    return self._order_book.update_order(order_id, **kwargs)

cancel_order 

cancel_order(order_id)

Cancel a pending order.

Returns True only when the identifier names a pending order. Filled, rejected, cancelled, and unknown orders return False.

Source code in src/ml4t/backtest/broker.py 
1683
1684
1685
1686
1687
1688
1689
1690
def cancel_order(self, order_id: str) -> bool:
    """Cancel a pending order.

    Returns True only when the identifier names a pending order. Filled,
    rejected, cancelled, and unknown orders return False.
    """
    self._capture_lifecycle_mutation(order_id=order_id)
    return self._order_book.cancel_order(order_id)

get_order 

get_order(order_id)

Return an order from the complete order history, or None when unknown.

Source code in src/ml4t/backtest/broker.py 
2366
2367
2368
def get_order(self, order_id: str) -> Order | None:
    """Return an order from the complete order history, or None when unknown."""
    return self._order_book.get_order(order_id)

get_pending_orders 

get_pending_orders(asset=None)

Return a copy of pending orders, optionally filtered by asset.

Source code in src/ml4t/backtest/broker.py 
2370
2371
2372
def get_pending_orders(self, asset: str | None = None) -> list[Order]:
    """Return a copy of pending orders, optionally filtered by asset."""
    return self._order_book.get_pending_orders(asset=asset)

get_rejected_orders 

get_rejected_orders(asset=None)

Get all rejected orders, optionally filtered by asset.

Parameters:

Name Type Description Default
asset str | None

If provided, filter to only this asset's rejected orders

 None

Returns:

Type Description
list[Order]

List of rejected Order objects with rejection_reason populated
Source code in src/ml4t/backtest/broker.py 
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
def get_rejected_orders(self, asset: str | None = None) -> list[Order]:
    """Get all rejected orders, optionally filtered by asset.

    Args:
        asset: If provided, filter to only this asset's rejected orders

    Returns:
        List of rejected Order objects with rejection_reason populated
    """
    return self._portfolio_ledger.get_rejected_orders(asset=asset)

get_position 

get_position(asset)

Get the current position for an asset.

Parameters:

Name Type Description Default
asset str

Asset symbol

 required

Returns:

Type Description
Position | None

Position object if position exists, None otherwise
Source code in src/ml4t/backtest/broker.py 
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
def get_position(self, asset: str) -> Position | None:
    """Get the current position for an asset.

    Args:
        asset: Asset symbol

    Returns:
        Position object if position exists, None otherwise
    """
    return self.positions.get(asset)

get_positions 

get_positions()

Get all current positions.

Returns:

Type Description
dict[str, Position]

Dictionary mapping asset symbols to Position objects
Source code in src/ml4t/backtest/broker.py 
1336
1337
1338
1339
1340
1341
1342
def get_positions(self) -> dict[str, Position]:
    """Get all current positions.

    Returns:
        Dictionary mapping asset symbols to Position objects
    """
    return self.positions

get_cash 

get_cash()

Get current cash balance.

Returns:

Type Description
float

Current cash balance (can be negative for margin accounts)
Source code in src/ml4t/backtest/broker.py 
1344
1345
1346
1347
1348
1349
1350
def get_cash(self) -> float:
    """Get current cash balance.

    Returns:
        Current cash balance (can be negative for margin accounts)
    """
    return self.cash

equity 

equity()

Return cash plus positions marked from the configured price source.

Source code in src/ml4t/backtest/broker.py 
1352
1353
1354
def equity(self) -> float:
    """Return cash plus positions marked from the configured price source."""
    return self._portfolio_ledger.get_account_value()

get_account_value 

get_account_value()

Return marked account equity using the compatibility method name.

Source code in src/ml4t/backtest/broker.py 
1356
1357
1358
def get_account_value(self) -> float:
    """Return marked account equity using the compatibility method name."""
    return self.equity()

get_buying_power 

get_buying_power()

Get current buying power.

Returns:

Type Description
float

Available buying power based on account policy:
float
  • Cash account: max(0, cash)
float
  • Margin account: (NLV - maintenance_margin) / initial_margin_rate
Source code in src/ml4t/backtest/broker.py 
2103
2104
2105
2106
2107
2108
2109
2110
2111
def get_buying_power(self) -> float:
    """Get current buying power.

    Returns:
        Available buying power based on account policy:
        - Cash account: max(0, cash)
        - Margin account: (NLV - maintenance_margin) / initial_margin_rate
    """
    return self.account.buying_power

get_trades 

get_trades(asset=None, last_n=None)

Get completed trades, optionally filtered.

Provides access to trade history during the backtest for stateful decision-making (e.g., adjusting position sizing based on recent wins/losses, implementing cooldown logic after stop-outs).

Parameters:

Name Type Description Default
asset str | None

Filter to only this asset's trades. If None, returns all trades.

 None
last_n int | None

Return only the last N trades (after other filters)

 None

Returns:

Type Description
list[Trade]

List of Trade objects matching the filters, ordered by exit time
Example

Get all trades for BTC

btc_trades = broker.get_trades(asset="BTC")

Get last 5 trades overall

recent = broker.get_trades(last_n=5)

Get recent trades for cooldown logic

last_trade = broker.get_last_trade("AAPL") if last_trade and last_trade.exit_reason == "stop_loss": # Implement cooldown after stop-out pass

Source code in src/ml4t/backtest/broker.py 
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
def get_trades(
    self,
    asset: str | None = None,
    last_n: int | None = None,
) -> list[Trade]:
    """Get completed trades, optionally filtered.

    Provides access to trade history during the backtest for stateful
    decision-making (e.g., adjusting position sizing based on recent
    wins/losses, implementing cooldown logic after stop-outs).

    Args:
        asset: Filter to only this asset's trades. If None, returns all trades.
        last_n: Return only the last N trades (after other filters)

    Returns:
        List of Trade objects matching the filters, ordered by exit time

    Example:
        # Get all trades for BTC
        btc_trades = broker.get_trades(asset="BTC")

        # Get last 5 trades overall
        recent = broker.get_trades(last_n=5)

        # Get recent trades for cooldown logic
        last_trade = broker.get_last_trade("AAPL")
        if last_trade and last_trade.exit_reason == "stop_loss":
            # Implement cooldown after stop-out
            pass
    """
    result = self.trades

    # Filter by asset
    if asset is not None:
        result = [t for t in result if t.symbol == asset]

    # Apply last_n limit
    if last_n is not None and last_n > 0:
        result = result[-last_n:]

    return result

get_last_trade 

get_last_trade(asset=None)

Get the most recent completed trade.

Convenience method for strategies that need to check the last trade (e.g., for cooldown logic after stop-outs).

Parameters:

Name Type Description Default
asset str | None

Filter to only this asset. If None, returns last trade overall.

 None

Returns:

Type Description
Trade | None

Most recent Trade object, or None if no trades
Example

last = broker.get_last_trade("AAPL") if last and last.exit_reason == "stop_loss": # Was stopped out - implement cooldown cooldown_bars = 5

Source code in src/ml4t/backtest/broker.py 
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
def get_last_trade(self, asset: str | None = None) -> Trade | None:
    """Get the most recent completed trade.

    Convenience method for strategies that need to check the last trade
    (e.g., for cooldown logic after stop-outs).

    Args:
        asset: Filter to only this asset. If None, returns last trade overall.

    Returns:
        Most recent Trade object, or None if no trades

    Example:
        last = broker.get_last_trade("AAPL")
        if last and last.exit_reason == "stop_loss":
            # Was stopped out - implement cooldown
            cooldown_bars = 5
    """
    trades = self.get_trades(asset=asset, last_n=1)
    return trades[0] if trades else None

set_position_rules 

set_position_rules(rules, asset=None)

Set position rules globally or per-asset.

Parameters:

Name Type Description Default
rules PositionRule | None

PositionRule or RuleChain to apply. None explicitly disables rules for the selected scope.

 required
asset str | None

If provided, apply only to this asset; otherwise global

 None
Source code in src/ml4t/backtest/broker.py 
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
def set_position_rules(
    self,
    rules: PositionRule | None,
    asset: str | None = None,
) -> None:
    """Set position rules globally or per-asset.

    Args:
        rules: PositionRule or RuleChain to apply. ``None`` explicitly
            disables rules for the selected scope.
        asset: If provided, apply only to this asset; otherwise global
    """
    self._capture_lifecycle_mutation(risk_rules=True)
    if asset is not None:
        self._position_rules_by_asset[asset] = rules
    else:
        self._position_rules = rules

clear_position_rules 

clear_position_rules(asset=None)

Disable position rules globally or for one asset.

Source code in src/ml4t/backtest/broker.py 
1400
1401
1402
def clear_position_rules(self, asset: str | None = None) -> None:
    """Disable position rules globally or for one asset."""
    self.set_position_rules(None, asset=asset)

update_position_context 

update_position_context(asset, context)

Update context data for a position (used by signal-based rules).

Parameters:

Name Type Description Default
asset str

Asset symbol

 required
context dict

Dict of signal/indicator values (e.g., {'exit_signal': -0.5, 'atr': 2.5})

 required
Source code in src/ml4t/backtest/broker.py 
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
def update_position_context(self, asset: str, context: dict) -> None:
    """Update context data for a position (used by signal-based rules).

    Args:
        asset: Asset symbol
        context: Dict of signal/indicator values (e.g., {'exit_signal': -0.5, 'atr': 2.5})
    """
    self._capture_lifecycle_mutation(asset=asset)
    pos = self.positions.get(asset)
    if pos:
        pos.context.update(context)

get_contract_spec 

get_contract_spec(asset)

Return the configured contract specification, or None when absent.

Source code in src/ml4t/backtest/broker.py 
985
986
987
def get_contract_spec(self, asset: str) -> ContractSpec | None:
    """Return the configured contract specification, or None when absent."""
    return self._contract_specs.get(asset)

get_multiplier 

get_multiplier(asset)

Get contract multiplier for an asset (1.0 for equities).

Source code in src/ml4t/backtest/broker.py 
989
990
991
992
def get_multiplier(self, asset: str) -> float:
    """Get contract multiplier for an asset (1.0 for equities)."""
    spec = self._contract_specs.get(asset)
    return spec.multiplier if spec else 1.0

get_mark_price 

get_mark_price(asset, *, quantity=None, use_open=False)

Resolve the configured mark price for an asset.

Source code in src/ml4t/backtest/broker.py 
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
def get_mark_price(
    self,
    asset: str,
    *,
    quantity: float | None = None,
    use_open: bool = False,
) -> float | None:
    """Resolve the configured mark price for an asset."""
    mark_side = None
    if self.mark_price == ExecutionPrice.QUOTE_SIDE and quantity is not None:
        mark_side = OrderSide.SELL if quantity > 0 else OrderSide.BUY
    return self.get_price_for_source(
        self.mark_price,
        asset,
        side=mark_side,
        quantity=quantity,
        use_open=use_open,
    )

get_last_price 

get_last_price(asset)

Return the most recent positive reference price observed for an asset.

Source code in src/ml4t/backtest/broker.py 
1082
1083
1084
def get_last_price(self, asset: str) -> float | None:
    """Return the most recent positive reference price observed for an asset."""
    return self._last_prices.get(asset)

get_price_for_source 

get_price_for_source(
    source,
    asset,
    *,
    side=None,
    quantity=None,
    use_open=False,
)

Resolve a configured price source with sensible OHLCV fallbacks.

Source code in src/ml4t/backtest/broker.py 
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
def get_price_for_source(
    self,
    source: ExecutionPrice,
    asset: str,
    *,
    side: OrderSide | None = None,
    quantity: float | None = None,
    use_open: bool = False,
) -> float | None:
    """Resolve a configured price source with sensible OHLCV fallbacks."""
    if (
        use_open
        and self.execution_mode == ExecutionMode.NEXT_BAR
        and source
        not in {
            ExecutionPrice.BID,
            ExecutionPrice.ASK,
            ExecutionPrice.QUOTE_MID,
            ExecutionPrice.QUOTE_SIDE,
            # VWAP is already a whole-bar price, so the open short-circuit must not
            # override it. While VWAP was absent from this set the branch below was
            # unreachable under NEXT_BAR, and every VWAP fill silently returned the
            # open - the assumption a caller chooses VWAP specifically to avoid.
            ExecutionPrice.VWAP,
        }
    ):
        return self._current_opens.get(asset, self._current_prices.get(asset))

    if source == ExecutionPrice.PRICE:
        return self._current_prices.get(asset, self._current_closes.get(asset))
    if source == ExecutionPrice.CLOSE:
        return self._current_closes.get(asset, self._current_prices.get(asset))
    if source == ExecutionPrice.OPEN:
        return self._current_opens.get(asset, self._current_prices.get(asset))
    if source == ExecutionPrice.MID:
        high = self._current_highs.get(asset)
        low = self._current_lows.get(asset)
        if high is not None and low is not None:
            return (high + low) / 2.0
        return self._current_prices.get(asset, self._current_closes.get(asset))
    if source == ExecutionPrice.VWAP:
        # None, not a fallback and not a raise. A bar in which nothing traded has no
        # volume-weighted price, and that is an ordinary market state rather than an
        # error: callers already treat None as "this asset cannot be priced on this
        # bar" and skip it, which leaves the order unfilled and the prior position
        # standing - what happens to a real order resting in a bar with no prints.
        # Substituting the close would invent a price from a stale carried print.
        # A feed that declares no VWAP column at all is a different thing entirely,
        # and Engine rejects that configuration before the first bar.
        return self._current_vwaps.get(asset)
    if source == ExecutionPrice.BID:
        return self._current_bids.get(asset, self._current_prices.get(asset))
    if source == ExecutionPrice.ASK:
        return self._current_asks.get(asset, self._current_prices.get(asset))
    if source == ExecutionPrice.QUOTE_MID:
        return self.get_quote_mid(asset) or self._current_prices.get(asset)
    if source == ExecutionPrice.QUOTE_SIDE:
        if side is None and quantity is not None:
            side = OrderSide.BUY if quantity > 0 else OrderSide.SELL
        if side == OrderSide.BUY:
            return self._current_asks.get(
                asset,
                self._current_opens.get(asset) if use_open else self._current_prices.get(asset),
            )
        if side == OrderSide.SELL:
            return self._current_bids.get(
                asset,
                self._current_opens.get(asset) if use_open else self._current_prices.get(asset),
            )
        return self.get_quote_mid(asset) or self._current_prices.get(asset)
    return self._current_prices.get(asset, self._current_closes.get(asset))

get_quote_mid 

get_quote_mid(asset)

Return explicit quote midpoint or derive it from bid/ask.

Source code in src/ml4t/backtest/broker.py 
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
def get_quote_mid(self, asset: str) -> float | None:
    """Return explicit quote midpoint or derive it from bid/ask."""
    mid = self._current_mids.get(asset)
    if mid is not None:
        return mid
    bid = self._current_bids.get(asset)
    ask = self._current_asks.get(asset)
    if bid is not None and ask is not None:
        return (bid + ask) / 2.0
    return None

get_quote_context 

get_quote_context(asset, side=None)

Return quote context for fills and trade summaries.

Source code in src/ml4t/backtest/broker.py 
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
def get_quote_context(
    self, asset: str, side: OrderSide | None = None
) -> dict[str, float | None]:
    """Return quote context for fills and trade summaries."""
    bid = self._current_bids.get(asset)
    ask = self._current_asks.get(asset)
    quote_mid = self.get_quote_mid(asset)
    spread = ask - bid if bid is not None and ask is not None else None
    return {
        "reference_price": self._current_prices.get(asset),
        "quote_mid_price": quote_mid,
        "bid_price": bid,
        "ask_price": ask,
        "spread": spread,
        "bid_size": self._current_bid_sizes.get(asset),
        "ask_size": self._current_ask_sizes.get(asset),
        "available_size": self.get_available_size(asset, side),
    }

get_available_size 

get_available_size(asset, side=None)

Return side-aware quote size when available, otherwise bar volume.

Source code in src/ml4t/backtest/broker.py 
1105
1106
1107
1108
1109
1110
1111
def get_available_size(self, asset: str, side: OrderSide | None = None) -> float | None:
    """Return side-aware quote size when available, otherwise bar volume."""
    if side == OrderSide.BUY:
        return self._current_ask_sizes.get(asset, self._current_volumes.get(asset))
    if side == OrderSide.SELL:
        return self._current_bid_sizes.get(asset, self._current_volumes.get(asset))
    return self._current_volumes.get(asset)

configure_stats 

configure_stats(
    recent_window_size=None,
    track_session_stats=None,
    enabled=None,
    config=None,
)

Configure trading statistics tracking.

Can either pass individual parameters or a StatsConfig object. Individual parameters override config values if both are provided.

Parameters:

Name Type Description Default
recent_window_size int | None

Number of recent trades to track (default 50)

 None
track_session_stats bool | None

Whether to track per-session statistics

 None
enabled bool | None

Whether stats tracking is enabled

 None
config StatsConfig | None

StatsConfig object (alternative to individual params)

 None
Example

Using individual parameters

broker.configure_stats(recent_window_size=100)

Using StatsConfig

broker.configure_stats(config=StatsConfig( recent_window_size=100, track_session_stats=True, ))

Source code in src/ml4t/backtest/broker.py 
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
def configure_stats(
    self,
    recent_window_size: int | None = None,
    track_session_stats: bool | None = None,
    enabled: bool | None = None,
    config: StatsConfig | None = None,
) -> None:
    """Configure trading statistics tracking.

    Can either pass individual parameters or a StatsConfig object.
    Individual parameters override config values if both are provided.

    Args:
        recent_window_size: Number of recent trades to track (default 50)
        track_session_stats: Whether to track per-session statistics
        enabled: Whether stats tracking is enabled
        config: StatsConfig object (alternative to individual params)

    Example:
        # Using individual parameters
        broker.configure_stats(recent_window_size=100)

        # Using StatsConfig
        broker.configure_stats(config=StatsConfig(
            recent_window_size=100,
            track_session_stats=True,
        ))
    """
    self._capture_lifecycle_mutation(all_asset_stats=True)
    if config is not None:
        self._stats_config = config
    else:
        self._stats_config = StatsConfig()

    # Override with individual parameters if provided
    if recent_window_size is not None:
        self._stats_config.recent_window_size = recent_window_size
    if track_session_stats is not None:
        self._stats_config.track_session_stats = track_session_stats
    if enabled is not None:
        self._stats_config.enabled = enabled

    # Update existing stats deques to new window size
    new_size = self._stats_config.recent_window_size
    for stats in self._asset_stats.values():
        if stats.recent_pnls.maxlen != new_size:
            # Create new deque with updated maxlen, preserving recent data
            old_pnls = list(stats.recent_pnls)
            stats.recent_pnls = deque(old_pnls[-new_size:], maxlen=new_size)
            # Recalculate recent_wins from preserved data
            stats.recent_wins = sum(1 for pnl in stats.recent_pnls if pnl > 0)

get_asset_stats 

get_asset_stats(asset)

Get trading statistics for an asset.

Returns the AssetTradingStats object for the given asset, creating one if it doesn't exist. Stats are automatically updated when positions are closed or scaled down.

Parameters:

Name Type Description Default
asset str

Asset symbol (e.g., "BTC", "AAPL")

 required

Returns:

Type Description
AssetTradingStats

AssetTradingStats object with all-time and recent statistics
Example

stats = broker.get_asset_stats("BTC")

Check recent performance

if stats.recent_win_rate > 0.6: # Increase position size when winning size = base_size * 1.5 elif stats.recent_win_rate < 0.4: # Reduce size when losing size = base_size * 0.5

Check session performance (intraday)

if stats.session_trades > 3 and stats.session_win_rate < 0.25: # Stop trading this asset for today return

Source code in src/ml4t/backtest/broker.py 
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
def get_asset_stats(self, asset: str) -> AssetTradingStats:
    """Get trading statistics for an asset.

    Returns the AssetTradingStats object for the given asset, creating
    one if it doesn't exist. Stats are automatically updated when
    positions are closed or scaled down.

    Args:
        asset: Asset symbol (e.g., "BTC", "AAPL")

    Returns:
        AssetTradingStats object with all-time and recent statistics

    Example:
        stats = broker.get_asset_stats("BTC")

        # Check recent performance
        if stats.recent_win_rate > 0.6:
            # Increase position size when winning
            size = base_size * 1.5
        elif stats.recent_win_rate < 0.4:
            # Reduce size when losing
            size = base_size * 0.5

        # Check session performance (intraday)
        if stats.session_trades > 3 and stats.session_win_rate < 0.25:
            # Stop trading this asset for today
            return
    """
    if asset not in self._asset_stats:
        self._capture_lifecycle_mutation(asset=asset)
        self._asset_stats[asset] = AssetTradingStats(
            recent_pnls=deque(maxlen=self._stats_config.recent_window_size)
        )
    return self._asset_stats[asset]

set_session_config 

set_session_config(config)

Set session configuration for session-aware statistics.

When a session config is set, trading statistics are reset at session boundaries. This is useful for intraday strategies that want to track performance within each trading session.

Parameters:

Name Type Description Default
config SessionConfig | None

SessionConfig object from ml4t.backtest.sessions, or None to disable session-aware statistics resets.

 required
Example

from ml4t.backtest.sessions import SessionConfig

CME futures: sessions start 5pm CT previous day

session_config = SessionConfig( calendar="CME_Equity", timezone="America/Chicago", session_start_time="17:00", ) broker.set_session_config(session_config)

Source code in src/ml4t/backtest/broker.py 
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
def set_session_config(self, config: SessionConfig | None) -> None:
    """Set session configuration for session-aware statistics.

    When a session config is set, trading statistics are reset at
    session boundaries. This is useful for intraday strategies that
    want to track performance within each trading session.

    Args:
        config: SessionConfig object from ml4t.backtest.sessions, or None to
            disable session-aware statistics resets.

    Example:
        from ml4t.backtest.sessions import SessionConfig

        # CME futures: sessions start 5pm CT previous day
        session_config = SessionConfig(
            calendar="CME_Equity",
            timezone="America/Chicago",
            session_start_time="17:00",
        )
        broker.set_session_config(session_config)
    """
    self._capture_lifecycle_mutation()
    self._session_config = config
    self._session_boundary = (
        None
        if config is None
        else (
            ZoneInfo(config.timezone),
            config.get_session_timezone(),
            config.get_session_start_hour(),
            config.get_session_start_minute(),
        )
    )
    self._last_session_id = None

mark_account_positions 

mark_account_positions(use_open=False)

Synchronize account position marks using configured price semantics.

Source code in src/ml4t/backtest/broker.py 
1132
1133
1134
1135
1136
1137
1138
def mark_account_positions(self, use_open: bool = False) -> None:
    """Synchronize account position marks using configured price semantics."""
    self._capture_lifecycle_mutation(all_positions=True)
    for asset, position in self.account.positions.items():
        mark_price = self.get_mark_price(asset, quantity=position.quantity, use_open=use_open)
        if mark_price is not None:
            position.current_price = mark_price

evaluate_position_rules 

evaluate_position_rules()

Evaluate position rules for all open positions.

Called by Engine before processing orders. Returns list of exit orders. Handles defer_fill=True by storing pending exits for next bar.

Source code in src/ml4t/backtest/broker.py 
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
def evaluate_position_rules(self) -> list[Order]:
    """Evaluate position rules for all open positions.

    Called by Engine before processing orders. Returns list of exit orders.
    Handles defer_fill=True by storing pending exits for next bar.
    """
    self._capture_lifecycle_mutation(
        all_positions=True,
        all_pending_orders=True,
        risk_rules=True,
        all_asset_stats=True,
    )
    return self._risk_engine.evaluate_position_rules()

Domain Types

Order dataclass 

Order(
    asset,
    side,
    quantity,
    order_type=MARKET,
    limit_price=None,
    stop_price=None,
    trail_amount=None,
    parent_id=None,
    rebalance_id=None,
    order_id="",
    status=PENDING,
    created_at=None,
    filled_at=None,
    filled_price=None,
    filled_quantity=0.0,
    rejection_reason=None,
    requested_quantity=None,
    target_intent_id=None,
    child_intent_id=None,
    intent_idempotency_key=None,
    _rejection_code=None,
    _created_bar_index=0,
    _signal_price=None,
    _priority_notional=None,
    _risk_exit_reason=None,
    _exit_reason=None,
    _risk_fill_price=None,
)

rejection_code property 

rejection_code

Return a stable machine-readable category for the rejection reason.

reject 

reject(reason, code)

Move the order to a rejected state with a stable reason code.

Source code in src/ml4t/backtest/types.py 
208
209
210
211
212
def reject(self, reason: str, code: str) -> None:
    """Move the order to a rejected state with a stable reason code."""
    self.status = OrderStatus.REJECTED
    self.rejection_reason = reason
    self._rejection_code = code

Fill dataclass 

Fill(
    order_id,
    asset,
    side,
    quantity,
    price,
    timestamp,
    rebalance_id=None,
    commission=0.0,
    slippage=0.0,
    order_type="",
    limit_price=None,
    stop_price=None,
    price_source="",
    reference_price=None,
    quote_mid_price=None,
    bid_price=None,
    ask_price=None,
    spread=None,
    bid_size=None,
    ask_size=None,
    available_size=None,
    exit_reason="",
    exit_reason_detail=None,
    target_intent_id=None,
    child_intent_id=None,
    intent_idempotency_key=None,
)

Trade dataclass 

Trade(
    symbol,
    entry_time,
    exit_time,
    entry_price,
    exit_price,
    quantity,
    pnl,
    pnl_percent,
    bars_held,
    fees=0.0,
    exit_slippage=0.0,
    exit_reason="signal",
    exit_reason_detail=None,
    status="closed",
    mfe=0.0,
    mae=0.0,
    entry_slippage=0.0,
    multiplier=1.0,
    entry_quote_mid_price=None,
    entry_bid_price=None,
    entry_ask_price=None,
    entry_spread=None,
    entry_available_size=None,
    exit_quote_mid_price=None,
    exit_bid_price=None,
    exit_ask_price=None,
    exit_spread=None,
    exit_available_size=None,
    metadata=None,
)

Realized exit leg or open position mark.

This dataclass is part of the cross-library API specification, designed to produce identical Parquet output across Python, Numba, and Rust implementations.

Fully closed positions use status="closed". Incremental reductions use status="partial" so lifecycle analytics can exclude repeated position-level excursion and holding-period values. Open positions use status="open" and their exit fields represent end-of-backtest mark-to-market values.

Schema Alignment (v0.1.0a6): - symbol: Asset identifier (was 'asset' in earlier versions) - fees: Total transaction fees (was 'commission') - mfe/mae: Max favorable/adverse excursion (was 'max_favorable_excursion'/'max_adverse_excursion') - direction: Derived property from quantity sign

direction property 

direction

Return 'long' or 'short' based on quantity sign.

is_open property 

is_open

Return True if this is an open (mark-to-market) trade.

commission property 

commission

Backward-compat alias for validation scripts expecting commission.

gross_pnl property 

gross_pnl

Price-move P&L before fees: (exit - entry) * quantity * multiplier.

net_pnl property 

net_pnl

P&L after all costs. Alias for self.pnl.

gross_return property 

gross_return

Direction-aware gross return. Same as pnl_percent.

net_return property 

net_return

Direction-aware net return including fees.

total_slippage_cost property 

total_slippage_cost

Total slippage cost in dollars (entry + exit).

cost_drag property 

cost_drag

Total cost as fraction of notional: (fees + total slippage cost) / notional.

Position dataclass 

Position(
    asset,
    quantity,
    entry_price,
    entry_time,
    current_price=None,
    bars_held=0,
    high_water_mark=None,
    low_water_mark=None,
    max_favorable_excursion=0.0,
    max_adverse_excursion=0.0,
    initial_quantity=None,
    context=dict(),
    multiplier=1.0,
    entry_commission=0.0,
    entry_slippage=0.0,
)

Unified position tracking for strategy and accounting.

Supports both long and short positions with: - Weighted average cost basis tracking - Mark-to-market price tracking - Risk metrics (MFE/MAE, water marks) - Contract multipliers for futures

Attributes:

Name Type Description
asset str

Asset identifier (e.g., "AAPL", "ES")
quantity float

Position size (positive=long, negative=short)
entry_price float

Weighted average entry price (cost basis)
entry_time datetime

Timestamp when position was first opened
current_price float | None

Latest mark-to-market price (updated each bar)
bars_held int

Number of bars this position has been held

Examples:

Long position: Position("AAPL", 100, 150.0, datetime.now()) -> quantity=100, unrealized_pnl depends on current_price

Short position: Position("AAPL", -100, 150.0, datetime.now()) -> quantity=-100, profit if price drops

market_value property 

market_value

Current market value of the position.

For long positions: positive value (asset on balance sheet) For short positions: negative value (liability on balance sheet)

Returns:

Type Description
float

Market value = quantity × current_price

side property 

side

Return 'long' or 'short' based on quantity sign.

unrealized_pnl 

unrealized_pnl(current_price=None)

Calculate unrealized P&L including contract multiplier.

Parameters:

Name Type Description Default
current_price float | None

Price to calculate P&L at. If None, uses self.current_price.

 None

Returns:

Type Description
float

Unrealized P&L = (current_price - entry_price) × quantity × multiplier
Source code in src/ml4t/backtest/types.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
def unrealized_pnl(self, current_price: float | None = None) -> float:
    """Calculate unrealized P&L including contract multiplier.

    Args:
        current_price: Price to calculate P&L at. If None, uses self.current_price.

    Returns:
        Unrealized P&L = (current_price - entry_price) × quantity × multiplier
    """
    price = current_price if current_price is not None else self.current_price
    if price is None:
        price = self.entry_price
    return (price - self.entry_price) * self.quantity * self.multiplier

pnl_percent 

pnl_percent(current_price=None)

Calculate direction-aware percentage return on position.

For long positions: (price - entry) / entry For short positions: (entry - price) / entry

Parameters:

Name Type Description Default
current_price float | None

Price to calculate return at. If None, uses self.current_price.

 None
Source code in src/ml4t/backtest/types.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def pnl_percent(self, current_price: float | None = None) -> float:
    """Calculate direction-aware percentage return on position.

    For long positions: (price - entry) / entry
    For short positions: (entry - price) / entry

    Args:
        current_price: Price to calculate return at. If None, uses self.current_price.
    """
    price = current_price if current_price is not None else self.current_price
    if price is None:
        price = self.entry_price
    if self.entry_price == 0:
        return 0.0
    raw = (price - self.entry_price) / self.entry_price
    return raw if self.quantity >= 0 else -raw

notional_value 

notional_value(current_price=None)

Calculate notional value of position.

Parameters:

Name Type Description Default
current_price float | None

Price to calculate value at. If None, uses self.current_price.

 None
Source code in src/ml4t/backtest/types.py 
315
316
317
318
319
320
321
322
323
324
def notional_value(self, current_price: float | None = None) -> float:
    """Calculate notional value of position.

    Args:
        current_price: Price to calculate value at. If None, uses self.current_price.
    """
    price = current_price if current_price is not None else self.current_price
    if price is None:
        price = self.entry_price
    return abs(self.quantity) * price * self.multiplier

update_water_marks 

update_water_marks(
    current_price,
    bar_high=None,
    bar_low=None,
    use_high_for_hwm=False,
    use_low_for_lwm=False,
)

Update high/low water marks and excursion tracking.

Parameters:

Name Type Description Default
current_price float

Current bar's close price

 required
bar_high float | None

Bar's high price (used for HWM if use_high_for_hwm=True)

 None
bar_low float | None

Bar's low price (used for LWM if use_low_for_lwm=True)

 None
use_high_for_hwm bool

If True, use bar_high for HWM (VBT Pro OHLC mode). If False, use current_price (close) for HWM (default).

 False
use_low_for_lwm bool

If True, use bar_low for LWM (VBT Pro OHLC mode). If False, use current_price (close) for LWM (default).

 False
Source code in src/ml4t/backtest/types.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def update_water_marks(
    self,
    current_price: float,
    bar_high: float | None = None,
    bar_low: float | None = None,
    use_high_for_hwm: bool = False,
    use_low_for_lwm: bool = False,
) -> None:
    """Update high/low water marks and excursion tracking.

    Args:
        current_price: Current bar's close price
        bar_high: Bar's high price (used for HWM if use_high_for_hwm=True)
        bar_low: Bar's low price (used for LWM if use_low_for_lwm=True)
        use_high_for_hwm: If True, use bar_high for HWM (VBT Pro OHLC mode).
                          If False, use current_price (close) for HWM (default).
        use_low_for_lwm: If True, use bar_low for LWM (VBT Pro OHLC mode).
                         If False, use current_price (close) for LWM (default).
    """
    # Update current price
    self.current_price = current_price

    # Select HWM source based on configuration
    high_for_hwm = bar_high if use_high_for_hwm and bar_high is not None else current_price
    low_for_lwm = bar_low if use_low_for_lwm and bar_low is not None else current_price

    # Update water marks (guaranteed non-None after __post_init__)
    if self.high_water_mark is None or high_for_hwm > self.high_water_mark:
        self.high_water_mark = high_for_hwm
    if self.low_water_mark is None or low_for_lwm < self.low_water_mark:
        self.low_water_mark = low_for_lwm

    # Update MFE/MAE using bar extremes (more accurate than close only)
    # For longs: MFE from high, MAE from low
    # For shorts: MFE from low, MAE from high
    if self.quantity > 0:  # Long position
        mfe_return = self.pnl_percent(high_for_hwm)
        mae_return = self.pnl_percent(low_for_lwm)
    else:  # Short position
        mfe_return = self.pnl_percent(low_for_lwm)
        mae_return = self.pnl_percent(high_for_hwm)

    if mfe_return > self.max_favorable_excursion:
        self.max_favorable_excursion = mfe_return
    if mae_return < self.max_adverse_excursion:
        self.max_adverse_excursion = mae_return

__repr__ 

__repr__()

String representation for debugging.

Source code in src/ml4t/backtest/types.py 
378
379
380
381
382
383
384
385
386
387
388
def __repr__(self) -> str:
    """String representation for debugging."""
    direction = "LONG" if self.quantity > 0 else "SHORT"
    price = self.current_price if self.current_price is not None else self.entry_price
    pnl = self.unrealized_pnl()
    return (
        f"Position({direction} {abs(self.quantity):.2f} {self.asset} "
        f"@ ${self.entry_price:.2f}, "
        f"current ${price:.2f}, "
        f"PnL ${pnl:+.2f})"
    )

Enums

OrderType

Bases: Enum

OrderSide

Bases: Enum

ExecutionMode

Bases: str, Enum

Order execution timing mode.

StopFillMode

Bases: str, Enum

Stop/take-profit fill price mode.

Different frameworks handle stop order fills differently: - STOP_PRICE: Fill at exact stop/target price (standard model, default) Matches VectorBT Pro with OHLC and Backtrader behavior - CLOSE_PRICE: Fill at bar's close price when stop triggers Matches VectorBT Pro with close-only data - BAR_EXTREME: Fill at bar's low (stop-loss) or high (take-profit) Worst/best case model (conservative/optimistic) - NEXT_BAR_OPEN: Fill at next bar's open price when stop triggers Matches Zipline behavior (strategy-level stops)

CommissionType

Bases: str, Enum

Commission calculation method.

SlippageType

Bases: str, Enum

Slippage calculation method.

FillOrdering

Bases: str, Enum

Order processing sequence within a single bar.

Controls how pending orders are sequenced during fill processing:

EXIT_FIRST (default): All exits → mark-to-market → all entries (with gatekeeper validation). Capital-efficient: exits free cash before entries need it. Matches engines that explicitly process reductions before additions.

FIFO

Orders process in submission order with sequential cash updates. Each order's gatekeeper check sees cash from all prior fills. Matches Backtrader's submission-order processing.

SEQUENTIAL

Orders process in submission order (typically alphabetical by asset) without exit/entry separation. Cash updates after each individual fill. Unlike EXIT_FIRST, exits do not pre-free cash for later entries. Matches LEAN's per-order sequential buying-power model.

PRIORITY

Sort all orders as one sequence using entry_order_priority. This preserves combined reversal orders and matches portfolio simulators whose automatic call sequence does not separate exits from entries.

Results

BacktestResult dataclass 

BacktestResult(
    trades,
    equity_curve,
    fills,
    metrics,
    predictions=None,
    config=None,
    equity=None,
    trade_analyzer=None,
    portfolio_state=list(),
    rejected_orders=list(),
    funding_payments=list(),
    artifact_diagnostics=tuple(),
    _trades_df=None,
    _equity_df=None,
    _fills_df=None,
    _portfolio_state_df=None,
    _rejected_orders_df=None,
)

Structured backtest result with export capabilities.

This class wraps the raw output from Engine.run() and provides: - DataFrame conversion methods (trades, equity, daily P&L) - Parquet export/import for persistence - Integration with ml4t.diagnostic library - Backward-compatible dict export

Attributes:

Name Type Description
trades list[Trade]

List of completed Trade objects
equity_curve list[tuple[datetime, float]]

List of (timestamp, portfolio_value) tuples
fills list[Fill]

List of Fill objects (all order fills)
funding_payments list[FundingPayment]

Timestamped cash flows separate from trades and fills
rejected_orders list[Order]

Orders that reached the rejected terminal state. Orders cancelled under permissive insufficient-cash handling are not included.
predictions DataFrame | None

Raw prediction DataFrame passed into the backtest (optional)
metrics dict[str, Any]

Dictionary of computed performance metrics
config BacktestConfig | None

BacktestConfig used for the backtest (optional)
equity EquityCurve | None

EquityCurve analytics object
trade_analyzer TradeAnalyzer | None

TradeAnalyzer analytics object
artifact_diagnostics tuple[ArtifactDiagnostic, ...]

Structured omissions and recovery actions. Empty for artifacts loaded successfully in strict mode.

from_parquet classmethod 

from_parquet(path, *, recovery=False)

Load a validated result artifact.

Parameters:

Name Type Description Default
path str | Path

Directory containing files written by :meth:to_parquet.

 required
recovery bool

Permit manifest-free beta artifacts and omit unreadable components. Every omission is reported through artifact_diagnostics.

 False

Returns:

Type Description
BacktestResult

BacktestResult instance.

Raises:

Type Description
ArtifactError

If strict validation or component decoding fails.
Source code in src/ml4t/backtest/result.py 
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
@classmethod
def from_parquet(
    cls,
    path: str | Path,
    *,
    recovery: bool = False,
) -> BacktestResult:
    """Load a validated result artifact.

    Args:
        path: Directory containing files written by :meth:`to_parquet`.
        recovery: Permit manifest-free beta artifacts and omit unreadable components.
            Every omission is reported through ``artifact_diagnostics``.

    Returns:
        BacktestResult instance.

    Raises:
        ArtifactError: If strict validation or component decoding fails.
    """
    path = Path(path)
    artifact_path = path
    if not path.exists():
        raise ArtifactNotFoundError(f"Result artifact path does not exist: {path}")
    if not path.is_dir():
        raise ArtifactNotFoundError(f"Result artifact path is not a directory: {path}")

    diagnostics: list[ArtifactDiagnostic] = []
    entries = list(path.iterdir())
    if not entries and not recovery:
        raise ArtifactNotFoundError(f"Result artifact directory is empty: {path}")

    marker_path = path / _INCOMPLETE_MARKER
    if marker_path.exists():
        if not recovery:
            raise ArtifactIncompleteError(
                f"Result artifact contains {_INCOMPLETE_MARKER}; its write did not complete"
            )
        diagnostics.append(
            ArtifactDiagnostic(
                code="incomplete_write",
                component="manifest",
                message="Artifact write did not complete.",
            )
        )

    def discover_legacy_components() -> dict[str, str]:
        discovered = {
            name: filename
            for name, filename in _COMPONENT_FILES.items()
            if (artifact_path / filename).exists()
        }
        if "predictions" not in discovered and (artifact_path / "signals.parquet").exists():
            discovered["predictions"] = "signals.parquet"
        return discovered

    manifest_path = path / _MANIFEST_FILE
    components: dict[str, str] = {}
    manifest: dict[str, Any] | None = None
    if not manifest_path.exists():
        if not recovery:
            raise ArtifactManifestError(
                "Result artifact manifest is missing; pass recovery=True only for retained "
                "beta artifacts"
            )
        diagnostics.append(
            ArtifactDiagnostic(
                code="manifest_missing",
                component="manifest",
                message="Loaded a manifest-free beta artifact.",
            )
        )
        components = discover_legacy_components()
    else:
        try:
            with open(manifest_path) as file:
                manifest_data = json.load(file)
            if not isinstance(manifest_data, dict):
                raise TypeError("manifest root must be an object")
            manifest = manifest_data
        except Exception as exc:
            if not recovery:
                raise ArtifactManifestError(
                    f"Failed to read {_MANIFEST_FILE}: {type(exc).__name__}: {exc}"
                ) from exc
            diagnostics.append(
                ArtifactDiagnostic(
                    code="manifest_invalid",
                    component="manifest",
                    message=f"Ignored malformed manifest ({type(exc).__name__}).",
                )
            )
            components = discover_legacy_components()

    if manifest is not None:
        artifact_type = manifest.get("artifact_type")
        if artifact_type != _ARTIFACT_TYPE:
            message = f"Unsupported artifact type: {artifact_type!r}"
            if not recovery:
                raise ArtifactManifestError(message)
            diagnostics.append(ArtifactDiagnostic("manifest_invalid", "manifest", message))
            components = discover_legacy_components()
            manifest = None

    if manifest is not None:
        schema_version = manifest.get("schema_version")
        if schema_version != _ARTIFACT_SCHEMA_VERSION:
            raise UnsupportedArtifactVersionError(
                f"Unsupported result artifact schema version {schema_version!r}; "
                f"supported version is {_ARTIFACT_SCHEMA_VERSION}"
            )
        component_data = manifest.get("components")
        if not isinstance(component_data, dict) or not all(
            isinstance(name, str) and isinstance(filename, str)
            for name, filename in component_data.items()
        ):
            if not recovery:
                raise ArtifactManifestError("Manifest components must be a string mapping")
            diagnostics.append(
                ArtifactDiagnostic(
                    "manifest_invalid",
                    "manifest",
                    "Ignored invalid component mapping.",
                )
            )
            components = discover_legacy_components()
        else:
            unknown = sorted(set(component_data) - _COMPONENT_FILES.keys())
            noncanonical = sorted(
                name
                for name, filename in component_data.items()
                if name in _COMPONENT_FILES and filename != _COMPONENT_FILES[name]
            )
            if unknown or noncanonical:
                details = f"unknown={unknown}, noncanonical={noncanonical}"
                if not recovery:
                    raise ArtifactManifestError(f"Invalid manifest components: {details}")
                diagnostics.append(
                    ArtifactDiagnostic(
                        "manifest_invalid",
                        "manifest",
                        f"Ignored invalid manifest components: {details}.",
                    )
                )
                components = discover_legacy_components()
            else:
                components = dict(component_data)

    declared_incomplete = manifest is not None and manifest.get("complete") is not True
    if declared_incomplete and not recovery:
        raise ArtifactIncompleteError("Result artifact manifest marks the export incomplete")
    if declared_incomplete:
        diagnostics.append(
            ArtifactDiagnostic(
                code="manifest_incomplete",
                component="manifest",
                message="Manifest marks this as a selective or incomplete export.",
            )
        )

    missing_required_components = sorted(_REQUIRED_RESULT_COMPONENTS - components.keys())
    missing_files = sorted(
        name for name, filename in components.items() if not (path / filename).is_file()
    )
    if not recovery and (missing_required_components or missing_files):
        raise ArtifactIncompleteError(
            "Result artifact is incomplete: "
            f"missing components={missing_required_components}, missing files={missing_files}"
        )
    if recovery:
        missing_components = sorted((_COMPONENT_FILES.keys() - {"funding"}) - components.keys())
        diagnostics.extend(
            ArtifactDiagnostic(
                code="component_missing",
                component=name,
                message=f"Component {_COMPONENT_FILES[name]} is absent.",
            )
            for name in missing_components
        )
        for name in missing_files:
            diagnostics.append(
                ArtifactDiagnostic(
                    code="component_missing_file",
                    component=name,
                    message=f"Declared component {components[name]} is absent.",
                )
            )
            components.pop(name)

    component_read_ok: dict[str, bool] = {}

    def read_component(name: str, reader, default):
        filename = components.get(name)
        if filename is None:
            component_read_ok[name] = False
            return default
        try:
            value = reader(path / filename)
            component_read_ok[name] = True
            return value
        except Exception as exc:
            component_read_ok[name] = False
            if not recovery:
                raise ArtifactReadError(
                    f"Failed to read {filename}: {type(exc).__name__}: {exc}"
                ) from exc
            diagnostics.append(
                ArtifactDiagnostic(
                    code="component_invalid",
                    component=name,
                    message=f"Ignored unreadable {filename} ({type(exc).__name__}).",
                )
            )
            return default

    def read_trades(component_path: Path) -> list[Trade]:
        result: list[Trade] = []
        for row in pl.read_parquet(component_path).iter_rows(named=True):
            symbol = row.get("symbol") or row.get("asset", "")
            fees = row.get("fees")
            if fees is None:
                fees = row.get("commission", 0.0)
            result.append(
                Trade(
                    symbol=symbol,
                    entry_time=row["entry_time"],
                    exit_time=row["exit_time"],
                    entry_price=row["entry_price"],
                    exit_price=row["exit_price"],
                    quantity=row["quantity"],
                    pnl=row["pnl"],
                    pnl_percent=row["pnl_percent"],
                    bars_held=row["bars_held"],
                    fees=fees,
                    exit_slippage=row.get("exit_slippage", row.get("slippage", 0.0)),
                    exit_reason=row.get("exit_reason", "signal"),
                    exit_reason_detail=row.get("exit_reason_detail"),
                    status=row.get("status", "closed"),
                    mfe=row.get("mfe", 0.0),
                    mae=row.get("mae", 0.0),
                    entry_slippage=row.get("entry_slippage", 0.0),
                    multiplier=row.get("multiplier", 1.0),
                    entry_quote_mid_price=row.get("entry_quote_mid_price"),
                    entry_bid_price=row.get("entry_bid_price"),
                    entry_ask_price=row.get("entry_ask_price"),
                    entry_spread=row.get("entry_spread"),
                    entry_available_size=row.get("entry_available_size"),
                    exit_quote_mid_price=row.get("exit_quote_mid_price"),
                    exit_bid_price=row.get("exit_bid_price"),
                    exit_ask_price=row.get("exit_ask_price"),
                    exit_spread=row.get("exit_spread"),
                    exit_available_size=row.get("exit_available_size"),
                )
            )
        return result

    def read_fills(component_path: Path) -> list[Fill]:
        result: list[Fill] = []
        for row in pl.read_parquet(component_path).iter_rows(named=True):
            result.append(
                Fill(
                    order_id=row["order_id"],
                    rebalance_id=row.get("rebalance_id"),
                    asset=row["asset"],
                    side=OrderSide(row["side"]),
                    quantity=row["quantity"],
                    price=row["price"],
                    timestamp=row["timestamp"],
                    commission=row.get("commission", 0.0),
                    slippage=row.get("slippage", 0.0),
                    order_type=row.get("order_type", ""),
                    limit_price=row.get("limit_price"),
                    stop_price=row.get("stop_price"),
                    price_source=row.get("price_source", ""),
                    reference_price=row.get("reference_price"),
                    quote_mid_price=row.get("quote_mid_price"),
                    bid_price=row.get("bid_price"),
                    ask_price=row.get("ask_price"),
                    spread=row.get("spread"),
                    bid_size=row.get("bid_size"),
                    ask_size=row.get("ask_size"),
                    available_size=row.get("available_size"),
                    exit_reason=row.get("exit_reason", ""),
                    exit_reason_detail=row.get("exit_reason_detail"),
                )
            )
        return result

    def read_funding(component_path: Path) -> list[FundingPayment]:
        return [
            FundingPayment(
                timestamp=row["timestamp"],
                asset=row["asset"],
                quantity=row["quantity"],
                mark_price=row["mark_price"],
                multiplier=row["multiplier"],
                rate=row["rate"],
                amount_per_unit=row["amount_per_unit"],
                cash_delta=row["cash_delta"],
            )
            for row in pl.read_parquet(component_path).iter_rows(named=True)
        ]

    def read_rejected_orders(component_path: Path) -> list[Order]:
        result: list[Order] = []
        for row in pl.read_parquet(component_path).iter_rows(named=True):
            result.append(
                Order(
                    order_id=row["order_id"],
                    asset=row["symbol"],
                    created_at=row["timestamp"],
                    requested_quantity=row["requested_quantity"],
                    quantity=row.get("remaining_quantity", row["requested_quantity"]),
                    filled_quantity=row.get("filled_quantity", 0.0),
                    side=OrderSide(row["side"]),
                    order_type=OrderType(row["order_type"]),
                    limit_price=row.get("limit_price"),
                    stop_price=row.get("stop_price"),
                    trail_amount=row.get("trail_amount"),
                    parent_id=row.get("parent_id"),
                    rebalance_id=row.get("rebalance_id"),
                    status=OrderStatus(row["status"]),
                    rejection_reason=row.get("rejection_reason"),
                    _rejection_code=row.get("rejection_code"),
                )
            )
        return result

    def read_equity(component_path: Path) -> list[tuple[datetime, float]]:
        return [
            (row["timestamp"], row["equity"])
            for row in pl.read_parquet(component_path).iter_rows(named=True)
        ]

    def read_portfolio_state(
        component_path: Path,
    ) -> list[tuple[datetime, float, float, float, float, int]]:
        return [
            (
                row["timestamp"],
                row["equity"],
                row["cash"],
                row["gross_exposure"],
                row["net_exposure"],
                row["open_positions"],
            )
            for row in pl.read_parquet(component_path).iter_rows(named=True)
        ]

    def read_metrics(component_path: Path) -> dict[str, Any]:
        with open(component_path) as file:
            data = json.load(file)
        if not isinstance(data, dict):
            raise TypeError("metrics root must be an object")
        return _deserialize_metric_value(data)

    def read_config(component_path: Path):
        import yaml

        from .config import BacktestConfig

        with open(component_path) as file:
            data = yaml.safe_load(file)
        if not isinstance(data, dict):
            raise TypeError("config root must be a mapping")
        return BacktestConfig.from_dict(data)

    def read_spec_config(component_path: Path):
        import yaml

        from .config import BacktestConfig

        with open(component_path) as file:
            data = yaml.safe_load(file)
        if not isinstance(data, dict):
            raise TypeError("spec root must be a mapping")
        if data.get("version") != 1:
            raise ValueError(f"unsupported spec version {data.get('version')!r}")
        config_data = data.get("config")
        if not isinstance(config_data, dict):
            raise TypeError("spec config must be a mapping")
        return BacktestConfig.from_dict(config_data)

    trades = read_component("trades", read_trades, [])
    fills = read_component("fills", read_fills, [])
    funding_payments = read_component("funding", read_funding, [])
    rejected_orders = read_component("rejected_orders", read_rejected_orders, [])
    equity_curve = read_component("equity", read_equity, [])
    portfolio_state = read_component("portfolio_state", read_portfolio_state, [])
    metrics = read_component("metrics", read_metrics, {})
    if metrics.get("num_funding_events", 0) and not component_read_ok["funding"]:
        message = "funding.parquet is required for an artifact with funding events"
        if not recovery:
            raise ArtifactReadError(message)
        diagnostics.append(ArtifactDiagnostic("component_missing", "funding", message))
    if component_read_ok["funding"] and "num_funding_events" in metrics:
        reported_count = metrics["num_funding_events"]
        reported_total = metrics.get("total_funding")
        actual_total = sum(payment.cash_delta for payment in funding_payments)
        if (
            reported_count != len(funding_payments)
            or not isinstance(reported_total, (int, float))
            or not math.isclose(reported_total, actual_total, abs_tol=1e-9)
        ):
            message = "funding.parquet is inconsistent with funding metrics"
            if not recovery:
                raise ArtifactReadError(message)
            diagnostics.append(ArtifactDiagnostic("component_inconsistent", "funding", message))
    predictions = read_component("predictions", pl.read_parquet, None)
    daily_pnl = read_component("daily_pnl", pl.read_parquet, None)
    config = read_component("config", read_config, None)
    spec_config = read_component("spec", read_spec_config, None)
    if config is None:
        config = spec_config

    if daily_pnl is not None:
        if not component_read_ok["equity"]:
            diagnostics.append(
                ArtifactDiagnostic(
                    code="component_unverified",
                    component="daily_pnl",
                    message="daily_pnl.parquet could not be verified without equity.parquet",
                )
            )
        else:
            expected_daily_pnl = cls(
                trades=[],
                equity_curve=equity_curve,
                fills=[],
                metrics={},
            ).to_daily_pnl()
        if component_read_ok["equity"] and not daily_pnl.equals(expected_daily_pnl):
            message = "daily_pnl.parquet is inconsistent with equity.parquet"
            if not recovery:
                raise ArtifactReadError(message)
            diagnostics.append(
                ArtifactDiagnostic(
                    code="component_inconsistent",
                    component="daily_pnl",
                    message=message,
                )
            )

    return cls(
        trades=trades,
        equity_curve=equity_curve,
        fills=fills,
        predictions=predictions,
        portfolio_state=portfolio_state,
        rejected_orders=rejected_orders,
        funding_payments=funding_payments,
        metrics=metrics,
        config=config,
        artifact_diagnostics=tuple(diagnostics),
    )

to_trades_dataframe 

to_trades_dataframe()

Convert trades to Polars DataFrame.

Returns DataFrame with columns

symbol, entry_time, exit_time, entry_price, exit_price, quantity, direction, pnl, pnl_percent, bars_held, fees, exit_slippage, mfe, mae, entry_slippage, multiplier, gross_pnl, net_return, total_slippage_cost, cost_drag, exit_reason, exit_reason_detail, status

Cost decomposition columns

gross_pnl: Price-move P&L before fees net_return: Direction-aware net return including fees total_slippage_cost: Entry + exit slippage in dollars cost_drag: Total cost as fraction of notional

The status column indicates "closed" (flat-to-flat completion), "partial" (realized reduction), or "open" (mark-to-market at end of backtest).

Returns:

Type Description
DataFrame

Polars DataFrame with one row per trade
Source code in src/ml4t/backtest/result.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def to_trades_dataframe(self) -> pl.DataFrame:
    """Convert trades to Polars DataFrame.

    Returns DataFrame with columns:
        symbol, entry_time, exit_time, entry_price, exit_price,
        quantity, direction, pnl, pnl_percent, bars_held,
        fees, exit_slippage, mfe, mae, entry_slippage, multiplier,
        gross_pnl, net_return, total_slippage_cost, cost_drag,
        exit_reason, exit_reason_detail, status

    Cost decomposition columns:
        gross_pnl: Price-move P&L before fees
        net_return: Direction-aware net return including fees
        total_slippage_cost: Entry + exit slippage in dollars
        cost_drag: Total cost as fraction of notional

    The status column indicates "closed" (flat-to-flat completion), "partial"
    (realized reduction), or "open" (mark-to-market at end of backtest).

    Returns:
        Polars DataFrame with one row per trade
    """
    if self._trades_df is not None:
        return self._trades_df

    schema = self._trades_schema()
    schema["entry_time"] = self._timestamp_dtype(t.entry_time for t in self.trades)
    schema["exit_time"] = self._timestamp_dtype(t.exit_time for t in self.trades)
    if not self.trades:
        return pl.DataFrame(schema=schema)

    records = []
    for t in self.trades:
        records.append(
            {
                "symbol": t.symbol,
                "entry_time": t.entry_time,
                "exit_time": t.exit_time,
                "entry_price": t.entry_price,
                "exit_price": t.exit_price,
                "quantity": t.quantity,
                "direction": t.direction,
                "pnl": t.pnl,
                "pnl_percent": t.pnl_percent,
                "bars_held": t.bars_held,
                "fees": t.fees,
                "exit_slippage": t.exit_slippage,
                "mfe": t.mfe,
                "mae": t.mae,
                "entry_slippage": t.entry_slippage,
                "multiplier": t.multiplier,
                "entry_quote_mid_price": t.entry_quote_mid_price,
                "entry_bid_price": t.entry_bid_price,
                "entry_ask_price": t.entry_ask_price,
                "entry_spread": t.entry_spread,
                "entry_available_size": t.entry_available_size,
                "exit_quote_mid_price": t.exit_quote_mid_price,
                "exit_bid_price": t.exit_bid_price,
                "exit_ask_price": t.exit_ask_price,
                "exit_spread": t.exit_spread,
                "exit_available_size": t.exit_available_size,
                "gross_pnl": t.gross_pnl,
                "net_return": t.net_return,
                "total_slippage_cost": t.total_slippage_cost,
                "cost_drag": t.cost_drag,
                "exit_reason": t.exit_reason,
                "exit_reason_detail": t.exit_reason_detail,
                "status": t.status,
            }
        )

    self._trades_df = pl.DataFrame(records, schema=schema)
    return self._trades_df

to_fills_dataframe 

to_fills_dataframe()

Convert every execution fill to a stable Polars DataFrame.

The result includes order identity, quantity, execution costs, price source, nullable quote context, available size, and exit-reason fields. An empty result retains the same typed schema.

Source code in src/ml4t/backtest/result.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def to_fills_dataframe(self) -> pl.DataFrame:
    """Convert every execution fill to a stable Polars DataFrame.

    The result includes order identity, quantity, execution costs, price
    source, nullable quote context, available size, and exit-reason fields.
    An empty result retains the same typed schema.
    """
    if self._fills_df is not None:
        return self._fills_df

    schema = self._fills_schema()
    schema["timestamp"] = self._timestamp_dtype(fill.timestamp for fill in self.fills)
    if not self.fills:
        return pl.DataFrame(schema=schema)

    records = []
    for fill in self.fills:
        records.append(
            {
                "order_id": fill.order_id,
                "rebalance_id": fill.rebalance_id,
                "asset": fill.asset,
                "side": fill.side.value,
                "quantity": fill.quantity,
                "price": fill.price,
                "timestamp": fill.timestamp,
                "commission": fill.commission,
                "slippage": fill.slippage,
                "order_type": fill.order_type,
                "limit_price": fill.limit_price,
                "stop_price": fill.stop_price,
                "price_source": fill.price_source,
                "reference_price": fill.reference_price,
                "quote_mid_price": fill.quote_mid_price,
                "bid_price": fill.bid_price,
                "ask_price": fill.ask_price,
                "spread": fill.spread,
                "bid_size": fill.bid_size,
                "ask_size": fill.ask_size,
                "available_size": fill.available_size,
                "exit_reason": fill.exit_reason,
                "exit_reason_detail": fill.exit_reason_detail,
            }
        )

    self._fills_df = pl.DataFrame(records, schema=schema)
    return self._fills_df

to_funding_dataframe 

to_funding_dataframe()

Return timestamped funding cash flows separately from trades and fills.

Source code in src/ml4t/backtest/result.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def to_funding_dataframe(self) -> pl.DataFrame:
    """Return timestamped funding cash flows separately from trades and fills."""
    schema = {
        "timestamp": self._timestamp_dtype(p.timestamp for p in self.funding_payments),
        "asset": pl.String(),
        "quantity": pl.Float64(),
        "mark_price": pl.Float64(),
        "multiplier": pl.Float64(),
        "rate": pl.Float64(),
        "amount_per_unit": pl.Float64(),
        "cash_delta": pl.Float64(),
    }
    return pl.DataFrame(
        [
            {
                "timestamp": payment.timestamp,
                "asset": payment.asset,
                "quantity": payment.quantity,
                "mark_price": payment.mark_price,
                "multiplier": payment.multiplier,
                "rate": payment.rate,
                "amount_per_unit": payment.amount_per_unit,
                "cash_delta": payment.cash_delta,
            }
            for payment in self.funding_payments
        ],
        schema=schema,
    )

to_rejected_orders_dataframe 

to_rejected_orders_dataframe()

Convert rejected orders to a stable, machine-readable DataFrame.

Source code in src/ml4t/backtest/result.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def to_rejected_orders_dataframe(self) -> pl.DataFrame:
    """Convert rejected orders to a stable, machine-readable DataFrame."""
    if self._rejected_orders_df is not None:
        return self._rejected_orders_df
    schema = self._rejected_orders_schema()
    schema["timestamp"] = self._timestamp_dtype(
        order.created_at for order in self.rejected_orders
    )
    if not self.rejected_orders:
        return pl.DataFrame(schema=schema)

    records = [
        {
            "order_id": order.order_id,
            "symbol": order.asset,
            "timestamp": order.created_at,
            "requested_quantity": order.requested_quantity,
            "filled_quantity": order.filled_quantity,
            "remaining_quantity": order.quantity,
            "side": order.side.value,
            "order_type": order.order_type.value,
            "limit_price": order.limit_price,
            "stop_price": order.stop_price,
            "trail_amount": order.trail_amount,
            "parent_id": order.parent_id,
            "rebalance_id": order.rebalance_id,
            "status": order.status.value,
            "rejection_code": order.rejection_code,
            "rejection_reason": order.rejection_reason,
        }
        for order in self.rejected_orders
    ]
    self._rejected_orders_df = pl.DataFrame(
        records,
        schema=schema,
    )
    return self._rejected_orders_df

to_portfolio_state_dataframe 

to_portfolio_state_dataframe()

Convert portfolio state snapshots to Polars DataFrame.

Returns DataFrame with columns

timestamp, equity, cash, gross_exposure, net_exposure, open_positions

Returns:

Type Description
DataFrame

Polars DataFrame with one row per bar, sorted by timestamp
Source code in src/ml4t/backtest/result.py 
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
def to_portfolio_state_dataframe(self) -> pl.DataFrame:
    """Convert portfolio state snapshots to Polars DataFrame.

    Returns DataFrame with columns:
        timestamp, equity, cash, gross_exposure, net_exposure, open_positions

    Returns:
        Polars DataFrame with one row per bar, sorted by timestamp
    """
    if self._portfolio_state_df is not None:
        return self._portfolio_state_df

    schema = self._portfolio_state_schema()
    schema["timestamp"] = self._timestamp_dtype(row[0] for row in self.portfolio_state)
    if not self.portfolio_state:
        return pl.DataFrame(schema=schema)

    self._portfolio_state_df = pl.DataFrame(
        self.portfolio_state,
        schema=schema,
        orient="row",
    ).sort("timestamp")
    return self._portfolio_state_df

to_predictions_dataframe 

to_predictions_dataframe()

Return the raw prediction DataFrame used as backtest input.

Source code in src/ml4t/backtest/result.py 
428
429
430
431
432
def to_predictions_dataframe(self) -> pl.DataFrame:
    """Return the raw prediction DataFrame used as backtest input."""
    if self.predictions is None:
        return pl.DataFrame()
    return self.predictions

to_equity_dataframe 

to_equity_dataframe()

Convert equity curve to Polars DataFrame.

Returns DataFrame with columns

timestamp, equity, return, cumulative_return, drawdown, high_water_mark

Returns:

Type Description
DataFrame

Polars DataFrame with one row per bar, sorted by timestamp
Source code in src/ml4t/backtest/result.py 
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
def to_equity_dataframe(self) -> pl.DataFrame:
    """Convert equity curve to Polars DataFrame.

    Returns DataFrame with columns:
        timestamp, equity, return, cumulative_return,
        drawdown, high_water_mark

    Returns:
        Polars DataFrame with one row per bar, sorted by timestamp
    """
    if self._equity_df is not None:
        return self._equity_df

    if not self.equity_curve:
        return pl.DataFrame(schema=self._equity_schema())

    timestamps = [ts for ts, _ in self.equity_curve]
    values = [float(v) for _, v in self.equity_curve]

    # Build base DataFrame and sort by timestamp
    df = pl.DataFrame({"timestamp": timestamps, "equity": values}).sort("timestamp")

    # Vectorized computation using Polars
    df = df.with_columns(
        [
            # Returns: percent change, first bar has no return
            pl.col("equity").pct_change().fill_null(0.0).alias("return"),
            # Cumulative return from initial equity
            (pl.col("equity") / pl.first("equity") - 1.0).alias("cumulative_return"),
            # High water mark (running maximum)
            pl.col("equity").cum_max().alias("high_water_mark"),
        ]
    ).with_columns(
        # Drawdown: (equity / hwm) - 1, handle division by zero
        pl.when(pl.col("high_water_mark") > 0)
        .then(pl.col("equity") / pl.col("high_water_mark") - 1.0)
        .otherwise(0.0)
        .alias("drawdown")
    )

    # Reorder columns to match expected schema
    self._equity_df = df.select(
        ["timestamp", "equity", "return", "cumulative_return", "drawdown", "high_water_mark"]
    )

    return self._equity_df

to_daily_pnl 

to_daily_pnl(session_aligned=False)

Get daily P&L DataFrame.

Parameters:

Name Type Description Default
session_aligned bool

If True and session config is available, align P&L to trading sessions (e.g., CME 5pm-4pm CT). If False, use calendar day boundaries.

 False

Returns:

Type Description
DataFrame

DataFrame with columns: date, open_equity, close_equity, high_equity, low_equity, pnl, return_pct, cumulative_return, num_bars
Source code in src/ml4t/backtest/result.py 
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
def to_daily_pnl(self, session_aligned: bool = False) -> pl.DataFrame:
    """Get daily P&L DataFrame.

    Args:
        session_aligned: If True and session config is available,
            align P&L to trading sessions (e.g., CME 5pm-4pm CT).
            If False, use calendar day boundaries.

    Returns:
        DataFrame with columns:
            date, open_equity, close_equity, high_equity, low_equity,
            pnl, return_pct, cumulative_return, num_bars
    """
    if not self.equity_curve:
        return pl.DataFrame(
            schema={
                "date": pl.Date,
                "open_equity": pl.Float64,
                "close_equity": pl.Float64,
                "high_equity": pl.Float64,
                "low_equity": pl.Float64,
                "pnl": pl.Float64,
                "return_pct": pl.Float64,
                "cumulative_return": pl.Float64,
                "num_bars": pl.Int32,
            }
        )

    # Build equity DataFrame
    equity_df = self.to_equity_dataframe()

    if session_aligned and self.config and self.config.resolved_calendar:
        # Use session alignment
        from .sessions import SessionConfig, compute_session_pnl

        session_config = SessionConfig(
            calendar=self.config.resolved_calendar,
            timezone=self.config.resolved_timezone,
            session_start_time=self.config.resolved_session_start_time,
        )
        return compute_session_pnl(self.equity_curve, session_config)

    # Default: calendar day aggregation
    daily = (
        equity_df.with_columns(pl.col("timestamp").dt.date().alias("date"))
        .group_by("date")
        .agg(
            [
                pl.col("equity").first().alias("open_equity"),
                pl.col("equity").last().alias("close_equity"),
                pl.col("equity").max().alias("high_equity"),
                pl.col("equity").min().alias("low_equity"),
                pl.len().alias("num_bars"),
            ]
        )
        .sort("date")
    )

    # Compute daily P&L and returns
    daily = daily.with_columns(
        [
            (pl.col("close_equity") - pl.col("open_equity")).alias("pnl"),
        ]
    )

    # Return percent (handle first day)
    prev_close = daily.select(pl.col("close_equity").shift(1)).to_series()
    return_pct = (daily["close_equity"] - prev_close) / prev_close
    return_pct = return_pct.fill_null(0.0)

    # Cumulative return from first open
    initial = daily["open_equity"][0] if len(daily) > 0 else 1.0
    cum_return = (daily["close_equity"] / initial) - 1.0

    daily = daily.with_columns(
        [
            return_pct.alias("return_pct"),
            cum_return.alias("cumulative_return"),
        ]
    )

    return daily

to_daily_returns 

to_daily_returns(calendar=None, session_aligned=None)

Get daily returns as Polars Series for ml4t-diagnostic integration.

This method properly aggregates bar-level equity to daily returns, which is the correct input for computing risk metrics like Sharpe ratio. For intraday data, using bar-level returns would give incorrect results.

Parameters:

Name Type Description Default
calendar str | None

Trading calendar for context. If provided and known, enables session-aware aggregation. Common values: - "crypto": 365 days/year (24/7) - "NYSE", "NASDAQ": 252 days/year - "CME_Equity", etc: Uses pandas_market_calendars If None, uses config calendar or defaults to calendar day boundaries.

 None
session_aligned bool | None

If True, align to trading sessions (e.g., CME 5pm CT). If None, auto-detect from calendar (True for CME, False for crypto). If False, use calendar day boundaries.

 None

Returns:

Type Description
Series

Series of daily returns (percentage, e.g., 0.01 = 1%)
Example

result = engine.run() daily_returns = result.to_daily_returns(calendar="NYSE")

Use with ml4t-diagnostic

from ml4t.diagnostic.evaluation.metrics.risk_adjusted import sharpe_ratio sharpe = sharpe_ratio(daily_returns.to_numpy(), annualization_factor=252)

Source code in src/ml4t/backtest/result.py 
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
def to_daily_returns(
    self,
    calendar: str | None = None,
    session_aligned: bool | None = None,
) -> pl.Series:
    """Get daily returns as Polars Series for ml4t-diagnostic integration.

    This method properly aggregates bar-level equity to daily returns,
    which is the correct input for computing risk metrics like Sharpe ratio.
    For intraday data, using bar-level returns would give incorrect results.

    Args:
        calendar: Trading calendar for context. If provided and known,
            enables session-aware aggregation. Common values:
            - "crypto": 365 days/year (24/7)
            - "NYSE", "NASDAQ": 252 days/year
            - "CME_Equity", etc: Uses pandas_market_calendars
            If None, uses config calendar or defaults to calendar day boundaries.
        session_aligned: If True, align to trading sessions (e.g., CME 5pm CT).
            If None, auto-detect from calendar (True for CME, False for crypto).
            If False, use calendar day boundaries.

    Returns:
        Series of daily returns (percentage, e.g., 0.01 = 1%)

    Example:
        >>> result = engine.run()
        >>> daily_returns = result.to_daily_returns(calendar="NYSE")
        >>> # Use with ml4t-diagnostic
        >>> from ml4t.diagnostic.evaluation.metrics.risk_adjusted import sharpe_ratio
        >>> sharpe = sharpe_ratio(daily_returns.to_numpy(), annualization_factor=252)
    """
    # Determine session alignment
    if session_aligned is None:
        cal = calendar or (self.config.resolved_calendar if self.config else None)
        session_aligned = self._auto_session_aligned(cal)

    # Get daily P&L DataFrame
    daily_df = self.to_daily_pnl(session_aligned=session_aligned)

    if daily_df.is_empty():
        return pl.Series("daily_return", [], dtype=pl.Float64)

    # Return the return_pct column as a Series
    return daily_df["return_pct"].alias("daily_return")

to_returns_series 

to_returns_series()

Get period returns as Polars Series.

Note: This returns BAR-LEVEL returns, not daily returns. For risk metrics like Sharpe ratio, use to_daily_returns() instead.

Returns:

Type Description
Series

Series of period returns (one per bar)
Source code in src/ml4t/backtest/result.py 
634
635
636
637
638
639
640
641
642
643
644
def to_returns_series(self) -> pl.Series:
    """Get period returns as Polars Series.

    Note: This returns BAR-LEVEL returns, not daily returns.
    For risk metrics like Sharpe ratio, use to_daily_returns() instead.

    Returns:
        Series of period returns (one per bar)
    """
    equity_df = self.to_equity_dataframe()
    return equity_df["return"]

to_trade_records 

to_trade_records()

Convert trades to ml4t.diagnostic TradeRecord format.

Returns list of dictionaries matching the TradeRecord schema from ml4t.diagnostic.integration.

Returns:

Type Description
list[dict[str, Any]]

List of trade record dictionaries
Source code in src/ml4t/backtest/result.py 
646
647
648
649
650
651
652
653
654
655
656
657
def to_trade_records(self) -> list[dict[str, Any]]:
    """Convert trades to ml4t.diagnostic TradeRecord format.

    Returns list of dictionaries matching the TradeRecord schema
    from ml4t.diagnostic.integration.

    Returns:
        List of trade record dictionaries
    """
    from .analytics.bridge import to_trade_records

    return to_trade_records(self.trades)

to_dict 

to_dict()

Export as dictionary (backward compatible with Engine.run()).

Returns:

Type Description
dict[str, Any]

Dictionary with all metrics and raw data
Source code in src/ml4t/backtest/result.py 
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def to_dict(self) -> dict[str, Any]:
    """Export as dictionary (backward compatible with Engine.run()).

    Returns:
        Dictionary with all metrics and raw data
    """
    result = dict(self.metrics)
    result.update(
        {
            "trades": self.trades,
            "equity_curve": self.equity_curve,
            "fills": self.fills,
            "funding_payments": self.funding_payments,
            "portfolio_state": self.portfolio_state,
        }
    )
    if self.predictions is not None:
        result["predictions"] = self.predictions
    if self.equity is not None:
        result["equity"] = self.equity
    if self.trade_analyzer is not None:
        result["trade_analyzer"] = self.trade_analyzer
    return result

to_spec_dict 

to_spec_dict()

Export a resolved runtime spec for reproducibility.

Returns:

Type Description
dict[str, Any]

Dictionary containing the fully resolved config, library version,
dict[str, Any]

and realized run window. The nested config payload remains
dict[str, Any]

compatible with BacktestConfig.from_dict().
Source code in src/ml4t/backtest/result.py 
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
def to_spec_dict(self) -> dict[str, Any]:
    """Export a resolved runtime spec for reproducibility.

    Returns:
        Dictionary containing the fully resolved config, library version,
        and realized run window. The nested ``config`` payload remains
        compatible with ``BacktestConfig.from_dict()``.
    """
    config_dict = self.config.to_dict() if self.config is not None else {}
    start = self.equity_curve[0][0].isoformat() if self.equity_curve else None
    end = self.equity_curve[-1][0].isoformat() if self.equity_curve else None
    return {
        "version": 1,
        "library_version": __version__,
        "lifecycle_version": self.metrics.get("lifecycle_version"),
        "lifecycle_callback_counts": self.metrics.get("lifecycle_callback_counts", {}),
        "lifecycle_invocations": self.metrics.get("lifecycle_invocations", []),
        "execution_policy": self.metrics.get("execution_policy"),
        "target_intent_count": self.metrics.get("target_intent_count", 0),
        "child_order_intent_count": self.metrics.get("child_order_intent_count", 0),
        "intent_reconciliation_count": self.metrics.get("intent_reconciliation_count", 0),
        "target_rule_reconciliation_count": self.metrics.get(
            "target_rule_reconciliation_count", 0
        ),
        "target_intents": self.metrics.get("target_intents", []),
        "child_order_intents": self.metrics.get("child_order_intents", []),
        "intent_reconciliations": self.metrics.get("intent_reconciliations", []),
        "target_rule_reconciliations": self.metrics.get("target_rule_reconciliations", []),
        "config": config_dict,
        "window": {
            "start": start,
            "end": end,
        },
    }

to_parquet 

to_parquet(path, include=None, compression='zstd')

Export backtest result to Parquet files.

Creates directory structure

{path}/ trades.parquet fills.parquet funding.parquet rejected_orders.parquet predictions.parquet equity.parquet portfolio_state.parquet daily_pnl.parquet metrics.json config.yaml (if config available) spec.yaml (if config available) manifest.json

Parameters:

Name Type Description Default
path str | Path

Directory path to write files

 required
include list[str] | None

Components to include. Default: all. Options: ["trades", "fills", "funding", "rejected_orders", "predictions", "equity", "portfolio_state", "daily_pnl", "metrics", "config", "spec"]

 None
compression Literal['lz4', 'uncompressed', 'snappy', 'gzip', 'brotli', 'zstd']

Parquet compression codec (default: "zstd")

 'zstd'

Returns:

Type Description
dict[str, Path]

Dict mapping requested component names to file paths. The always-written
dict[str, Path]

manifest is returned under the additional "manifest" key; it is not
dict[str, Path]

a selectable component.

Raises:

Type Description
ArtifactWriteError

If a requested component is unavailable or cannot be written.
Source code in src/ml4t/backtest/result.py 
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
def to_parquet(
    self,
    path: str | Path,
    include: list[str] | None = None,
    compression: Literal["lz4", "uncompressed", "snappy", "gzip", "brotli", "zstd"] = "zstd",
) -> dict[str, Path]:
    """Export backtest result to Parquet files.

    Creates directory structure:
        {path}/
            trades.parquet
            fills.parquet
            funding.parquet
            rejected_orders.parquet
            predictions.parquet
            equity.parquet
            portfolio_state.parquet
            daily_pnl.parquet
            metrics.json
            config.yaml (if config available)
            spec.yaml (if config available)
            manifest.json

    Args:
        path: Directory path to write files
        include: Components to include. Default: all.
            Options: ["trades", "fills", "funding", "rejected_orders", "predictions", "equity",
                "portfolio_state", "daily_pnl", "metrics", "config", "spec"]
        compression: Parquet compression codec (default: "zstd")

    Returns:
        Dict mapping requested component names to file paths. The always-written
        manifest is returned under the additional ``"manifest"`` key; it is not
        a selectable component.

    Raises:
        ArtifactWriteError: If a requested component is unavailable or cannot be written.
    """
    explicitly_selected = include is not None
    requested = list(include) if include is not None else list(_COMPONENT_FILES)
    unknown = sorted(set(requested) - _COMPONENT_FILES.keys() - {"manifest"})
    if unknown:
        raise ArtifactWriteError(f"Unknown artifact components requested: {unknown}")
    requested = [name for name in requested if name != "manifest"]

    unavailable: dict[str, str] = {}
    if self.predictions is None:
        unavailable["predictions"] = "result has no predictions"
    if self.config is None:
        unavailable["config"] = "result has no config"
        unavailable["spec"] = "result has no config for a runtime spec"

    explicitly_unavailable = sorted(set(requested) & unavailable.keys())
    if explicitly_selected and explicitly_unavailable:
        details = ", ".join(f"{name}: {unavailable[name]}" for name in explicitly_unavailable)
        raise ArtifactWriteError(f"Requested artifact components are unavailable: {details}")

    selected = [name for name in requested if name not in unavailable]

    text_payloads: dict[str, str] = {}
    if "metrics" in selected:
        try:
            serializable_metrics = {
                key: _serialize_metric_value(value, path=f"metrics[{key!r}]")
                for key, value in self.metrics.items()
            }
            text_payloads["metrics"] = json.dumps(
                serializable_metrics,
                indent=2,
                allow_nan=False,
            )
        except ArtifactWriteError:
            raise
        except Exception as exc:
            raise ArtifactWriteError(f"Failed to serialize metrics: {exc}") from exc

    if "config" in selected or "spec" in selected:
        config = self.config
        if config is None:
            raise ArtifactWriteError("Config and spec components require a runtime config")
        try:
            import yaml
        except ImportError as exc:
            raise ArtifactWriteError("PyYAML is required to serialize config or spec") from exc
        if "config" in selected:
            try:
                text_payloads["config"] = yaml.safe_dump(
                    config.to_dict(),
                    default_flow_style=False,
                )
            except Exception as exc:
                raise ArtifactWriteError(
                    f"Failed to serialize config component: {exc}"
                ) from exc
        if "spec" in selected:
            try:
                text_payloads["spec"] = yaml.safe_dump(
                    self.to_spec_dict(),
                    default_flow_style=False,
                    sort_keys=False,
                )
            except Exception as exc:
                raise ArtifactWriteError(f"Failed to serialize spec component: {exc}") from exc

    path = Path(path)
    try:
        path.mkdir(parents=True, exist_ok=True)
    except Exception as exc:
        raise ArtifactWriteError(f"Failed to create artifact directory {path}: {exc}") from exc

    def write_component(name: str, writer) -> None:
        try:
            writer()
        except Exception as exc:
            raise ArtifactWriteError(f"Failed to write {name} component: {exc}") from exc

    marker_path = path / _INCOMPLETE_MARKER
    write_component(
        "incomplete marker",
        lambda: marker_path.write_text("Result artifact write did not complete.\n"),
    )
    manifest_path = path / _MANIFEST_FILE
    write_component("stale manifest removal", lambda: manifest_path.unlink(missing_ok=True))

    written: dict[str, Path] = {}

    if "trades" in selected:
        trades_path = path / "trades.parquet"
        write_component(
            "trades",
            lambda: self.to_trades_dataframe().write_parquet(
                trades_path,
                compression=compression,
            ),
        )
        written["trades"] = trades_path

    if "fills" in selected:
        fills_path = path / "fills.parquet"
        write_component(
            "fills",
            lambda: self.to_fills_dataframe().write_parquet(
                fills_path,
                compression=compression,
            ),
        )
        written["fills"] = fills_path

    if "funding" in selected:
        funding_path = path / "funding.parquet"
        write_component(
            "funding",
            lambda: self.to_funding_dataframe().write_parquet(
                funding_path,
                compression=compression,
            ),
        )
        written["funding"] = funding_path

    if "rejected_orders" in selected:
        rejected_orders_path = path / "rejected_orders.parquet"
        write_component(
            "rejected_orders",
            lambda: self.to_rejected_orders_dataframe().write_parquet(
                rejected_orders_path,
                compression=compression,
            ),
        )
        written["rejected_orders"] = rejected_orders_path

    if "predictions" in selected:
        predictions_path = path / "predictions.parquet"
        write_component(
            "predictions",
            lambda: self.to_predictions_dataframe().write_parquet(
                predictions_path,
                compression=compression,
            ),
        )
        written["predictions"] = predictions_path

    if "equity" in selected:
        equity_path = path / "equity.parquet"
        write_component(
            "equity",
            lambda: self.to_equity_dataframe().write_parquet(
                equity_path,
                compression=compression,
            ),
        )
        written["equity"] = equity_path

    if "portfolio_state" in selected:
        portfolio_state_path = path / "portfolio_state.parquet"
        write_component(
            "portfolio_state",
            lambda: self.to_portfolio_state_dataframe().write_parquet(
                portfolio_state_path,
                compression=compression,
            ),
        )
        written["portfolio_state"] = portfolio_state_path

    if "daily_pnl" in selected:
        daily_path = path / "daily_pnl.parquet"
        write_component(
            "daily_pnl",
            lambda: self.to_daily_pnl().write_parquet(
                daily_path,
                compression=compression,
            ),
        )
        written["daily_pnl"] = daily_path

    for name in ("metrics", "config", "spec"):
        if name not in selected:
            continue
        component_path = path / _COMPONENT_FILES[name]
        write_component(
            name,
            lambda component_path=component_path, payload=text_payloads[name]: (
                component_path.write_text(payload)
            ),
        )
        written[name] = component_path

    manifest = {
        "artifact_type": _ARTIFACT_TYPE,
        "schema_version": _ARTIFACT_SCHEMA_VERSION,
        "library_version": __version__,
        "complete": written.keys()
        >= (_REQUIRED_RESULT_COMPONENTS | ({"funding"} if self.funding_payments else set())),
        "components": {
            name: _COMPONENT_FILES[name] for name in _COMPONENT_FILES if name in written
        },
        "omitted_components": {
            name: reason for name, reason in unavailable.items() if name in requested
        },
    }
    manifest_payload = json.dumps(manifest, indent=2, allow_nan=False)
    write_component("manifest", lambda: manifest_path.write_text(manifest_payload))
    written["manifest"] = manifest_path
    write_component("incomplete marker removal", marker_path.unlink)

    return written

__getitem__ 

__getitem__(key)

Return a metric or raw result component, raising KeyError if absent.

Source code in src/ml4t/backtest/result.py 
719
720
721
def __getitem__(self, key: str) -> Any:
    """Return a metric or raw result component, raising KeyError if absent."""
    return self.to_dict()[key]

get 

get(key, default=None)

Return a metric or raw component, or default when the key is absent.

Source code in src/ml4t/backtest/result.py 
723
724
725
def get(self, key: str, default: Any = None) -> Any:
    """Return a metric or raw component, or default when the key is absent."""
    return self.to_dict().get(key, default)

keys 

keys()

Return keys from the backward-compatible dictionary representation.

Source code in src/ml4t/backtest/result.py 
727
728
729
def keys(self) -> KeysView[str]:
    """Return keys from the backward-compatible dictionary representation."""
    return self.to_dict().keys()

items 

items()

Return items from the backward-compatible dictionary representation.

Source code in src/ml4t/backtest/result.py 
731
732
733
def items(self) -> ItemsView[str, Any]:
    """Return items from the backward-compatible dictionary representation."""
    return self.to_dict().items()

Execution: Market Impact

LinearImpact dataclass 

LinearImpact(coefficient=0.1)

Bases: MarketImpactModel

Linear market impact model.

Impact = coefficient * (quantity / volume) * price

Simple model where impact scales linearly with participation rate. Appropriate for liquid markets with moderate order sizes.

Parameters:

Name Type Description Default
coefficient float

Impact scaling factor (default 0.1) Higher values = more impact per unit participation

 0.1

The impact this returns is entirely temporary: calculate sees one order and holds no reference to earlier slices of the same parent, so there is nothing for a permanent component to persist into. A permanent_fraction field was accepted and documented here until 0.1.7 and was never read; it is removed rather than defaulted so that asking for a permanent component raises instead of silently returning a fully temporary one.

Example

model = LinearImpact(coefficient=0.1)

10% participation at $100 price = $1.00 impact

calculate 

calculate(quantity, price, volume, is_buy)

Calculate linear impact.

Source code in src/ml4t/backtest/execution/impact.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def calculate(
    self,
    quantity: float,
    price: float,
    volume: float | None,
    is_buy: bool,
) -> float:
    """Calculate linear impact."""
    if volume is None or volume == 0:
        return 0.0

    participation = quantity / volume
    impact = self.coefficient * participation * price

    # Apply direction (buys push price up, sells push price down)
    return impact if is_buy else -impact

SquareRootImpact dataclass 

SquareRootImpact(
    coefficient=0.5, volatility=0.02, adv_factor=1.0
)

Bases: MarketImpactModel

Square root market impact model (Almgren-Chriss style).

Impact = coefficient * sigma * sqrt(quantity / ADV) * price

Based on academic market microstructure research. Impact scales with the square root of order size, which matches empirical observations.

Parameters:

Name Type Description Default
coefficient float

Scaling factor (default 0.5, typical range 0.1-1.0)

 0.5
volatility float

Daily volatility (sigma, default 0.02 = 2%)

 0.02
adv_factor float

Average daily volume as multiple of bar volume (default 1.0 for daily bars, 390 for minute bars)

 1.0
Example

model = SquareRootImpact(coefficient=0.5, volatility=0.02)

For order = 1% of ADV at 2% vol, $100 price:

Impact = 0.5 * 0.02 * sqrt(0.01) * 100 = $0.10

calculate 

calculate(quantity, price, volume, is_buy)

Calculate square root impact.

Source code in src/ml4t/backtest/execution/impact.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def calculate(
    self,
    quantity: float,
    price: float,
    volume: float | None,
    is_buy: bool,
) -> float:
    """Calculate square root impact."""
    if volume is None or volume == 0:
        return 0.0

    adv = volume * self.adv_factor
    participation = quantity / adv

    # Square root impact
    impact = self.coefficient * self.volatility * math.sqrt(participation) * price

    return impact if is_buy else -impact

PowerLawImpact dataclass 

PowerLawImpact(
    coefficient=0.1, exponent=0.5, min_impact=0.0
)

Bases: MarketImpactModel

Generalized power law impact model.

Impact = coefficient * (quantity / volume)^exponent * price

Flexible model that can represent various impact regimes. - exponent = 1.0: Linear (like LinearImpact) - exponent = 0.5: Square root (like SquareRootImpact) - exponent < 0.5: Concave (impact flattens for large orders) - exponent > 1.0: Convex (impact accelerates for large orders)

Parameters:

Name Type Description Default
coefficient float

Scaling factor (default 0.1)

 0.1
exponent float

Power law exponent (default 0.5)

 0.5
min_impact float

Minimum impact per trade (fixed cost, default 0)

 0.0
Example

model = PowerLawImpact(coefficient=0.1, exponent=0.6)

calculate 

calculate(quantity, price, volume, is_buy)

Calculate power law impact.

Source code in src/ml4t/backtest/execution/impact.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def calculate(
    self,
    quantity: float,
    price: float,
    volume: float | None,
    is_buy: bool,
) -> float:
    """Calculate power law impact."""
    if volume is None or volume == 0:
        return self.min_impact if is_buy else -self.min_impact

    participation = quantity / volume

    # Power law impact
    impact = self.coefficient * (participation**self.exponent) * price
    impact = max(impact, self.min_impact)

    return impact if is_buy else -impact

Risk: Position Rules

StopLoss dataclass 

StopLoss(pct)

Exit when stop price is breached during the bar.

Stop orders trigger when the bar's price range touches the stop level. Fill price depends on StopFillMode configuration: - STOP_PRICE: Fill at exact stop price (standard model, default) - BAR_EXTREME: Fill at bar's low (matches VectorBT Pro behavior)

For long positions: stop triggers if bar_low <= stop_price For short positions: stop triggers if bar_high >= stop_price

Parameters:

Name Type Description Default
pct float

Maximum loss as decimal (0.05 = 5% loss triggers exit)

 required
Example

rule = StopLoss(pct=0.05) # Exit at -5%

evaluate 

evaluate(state)

Exit if stop price was breached during the bar.

Source code in src/ml4t/backtest/risk/position/static.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def evaluate(self, state: PositionState) -> PositionAction:
    """Exit if stop price was breached during the bar."""
    # Get base price for stop level calculation (entry_price or signal_price)
    base_price = _get_stop_base_price(state, state.context)

    # Calculate stop price from base
    if state.is_long:
        stop_price = base_price * (1 - self.pct)
        # Check if stop was triggered during bar (low touched stop level)
        triggered = (
            state.bar_low is not None and state.bar_low <= stop_price
        ) or state.current_price <= stop_price
    else:  # short
        stop_price = base_price * (1 + self.pct)
        # Check if stop was triggered during bar (high touched stop level)
        triggered = (
            state.bar_high is not None and state.bar_high >= stop_price
        ) or state.current_price >= stop_price

    if triggered:
        # Determine fill price based on mode
        fill_mode = _get_stop_fill_mode(state.context)
        if fill_mode == StopFillMode.NEXT_BAR_OPEN:
            # Zipline model: defer exit to next bar, fill at open
            return PositionAction.exit_full(
                reason=f"stop_loss_{self.pct:.1%}",
                defer_fill=True,  # Broker will fill at next bar's open
            )
        elif fill_mode == StopFillMode.CLOSE_PRICE:
            # VectorBT Pro close-only model: always fill at close price
            fill_price = state.current_price
        elif fill_mode == StopFillMode.BAR_EXTREME:
            # Conservative model: fill at bar's extreme (worst case)
            if state.is_long:
                fill_price = state.bar_low if state.bar_low is not None else stop_price
            else:
                fill_price = state.bar_high if state.bar_high is not None else stop_price
        else:
            # Standard model (STOP_PRICE): fill at exact stop price if within bar range
            # If bar gaps through stop, fill at open (gap behavior)
            if state.is_long:
                # For long stops: check if bar opened below stop (gap down)
                # or if stop is within bar range
                if state.bar_open is not None and state.bar_open <= stop_price:
                    # Bar opened below stop - fill at open (Backtrader gap behavior)
                    fill_price = state.bar_open
                elif (
                    state.bar_low is not None
                    and state.bar_high is not None
                    and state.bar_low <= stop_price <= state.bar_high
                ):
                    # Stop within bar range - fill at exact stop
                    fill_price = stop_price
                else:
                    # Gap through (VBT behavior) - fill at close
                    fill_price = state.current_price
            else:
                # For short stops: check if bar opened above stop (gap up)
                if state.bar_open is not None and state.bar_open >= stop_price:
                    # Bar opened above stop - fill at open (gap behavior)
                    fill_price = state.bar_open
                elif (
                    state.bar_low is not None
                    and state.bar_high is not None
                    and state.bar_low <= stop_price <= state.bar_high
                ):
                    # Stop within bar range - fill at exact stop
                    fill_price = stop_price
                else:
                    # Gap through (VBT behavior) - fill at close
                    fill_price = state.current_price

        return PositionAction.exit_full(
            reason=f"stop_loss_{self.pct:.1%}",
            fill_price=fill_price,
        )
    return PositionAction.hold()

TakeProfit dataclass 

TakeProfit(pct)

Exit when target price is reached during the bar.

Take-profit orders trigger when the bar's price range touches the target. Fill price depends on StopFillMode configuration: - STOP_PRICE: Fill at exact target price (standard model, default) - BAR_EXTREME: Fill at bar's high (matches VectorBT Pro behavior)

For long positions: triggers if bar_high >= target_price For short positions: triggers if bar_low <= target_price

Parameters:

Name Type Description Default
pct float

Target profit as decimal (0.10 = 10% profit triggers exit)

 required
Example

rule = TakeProfit(pct=0.10) # Exit at +10%

evaluate 

evaluate(state)

Exit if target price was reached during the bar.

Source code in src/ml4t/backtest/risk/position/static.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def evaluate(self, state: PositionState) -> PositionAction:
    """Exit if target price was reached during the bar."""
    # Get base price for target level calculation (entry_price or signal_price)
    base_price = _get_stop_base_price(state, state.context)

    # Calculate target price from base
    if state.is_long:
        target_price = base_price * (1 + self.pct)
        # Check if target was reached during bar (high touched target)
        triggered = (
            state.bar_high is not None and state.bar_high >= target_price
        ) or state.current_price >= target_price
    else:  # short
        target_price = base_price * (1 - self.pct)
        # Check if target was reached during bar (low touched target)
        triggered = (
            state.bar_low is not None and state.bar_low <= target_price
        ) or state.current_price <= target_price

    if triggered:
        # Determine fill price based on mode
        fill_mode = _get_stop_fill_mode(state.context)
        if fill_mode == StopFillMode.NEXT_BAR_OPEN:
            # Zipline model: defer exit to next bar, fill at open
            return PositionAction.exit_full(
                reason=f"take_profit_{self.pct:.1%}",
                defer_fill=True,  # Broker will fill at next bar's open
            )
        elif fill_mode == StopFillMode.CLOSE_PRICE:
            # VectorBT Pro close-only model: always fill at close price
            fill_price = state.current_price
        elif fill_mode == StopFillMode.BAR_EXTREME:
            # Optimistic model: fill at bar's extreme (best case for profits)
            if state.is_long:
                fill_price = state.bar_high if state.bar_high is not None else target_price
            else:
                fill_price = state.bar_low if state.bar_low is not None else target_price
        else:
            # Standard model (STOP_PRICE): fill at exact target price if within bar range
            # If bar gaps through target, fill at open/close (gap behavior)
            if state.is_long:
                # For long targets: check if bar opened above target (price improvement)
                # or if target is within bar range
                if state.bar_open is not None and state.bar_open >= target_price:
                    # Bar opened above target - fill at open (Backtrader behavior)
                    fill_price = state.bar_open
                elif (
                    state.bar_low is not None
                    and state.bar_high is not None
                    and state.bar_low <= target_price <= state.bar_high
                ):
                    # Target within bar range - fill at exact target
                    fill_price = target_price
                else:
                    # Gap through - fill at close
                    fill_price = state.current_price
            else:
                # For short targets: check if bar opened below target (price improvement)
                if state.bar_open is not None and state.bar_open <= target_price:
                    # Bar opened below target - fill at open (price improvement)
                    fill_price = state.bar_open
                elif (
                    state.bar_low is not None
                    and state.bar_high is not None
                    and state.bar_low <= target_price <= state.bar_high
                ):
                    # Target within bar range - fill at exact target
                    fill_price = target_price
                else:
                    # Gap through - fill at close
                    fill_price = state.current_price

        return PositionAction.exit_full(
            reason=f"take_profit_{self.pct:.1%}",
            fill_price=fill_price,
        )
    return PositionAction.hold()

TimeExit dataclass 

TimeExit(max_bars)

Exit after holding for a specified number of bars.

Parameters:

Name Type Description Default
max_bars int

Maximum bars to hold position

 required
Example

rule = TimeExit(max_bars=20) # Exit after 20 bars

evaluate 

evaluate(state)

Exit if held too long.

Source code in src/ml4t/backtest/risk/position/static.py 
241
242
243
244
245
246
def evaluate(self, state: PositionState) -> PositionAction:
    """Exit if held too long."""
    if state.bars_held >= self.max_bars:
        # Time exits fill at current close price
        return PositionAction.exit_full(f"time_exit_{self.max_bars}bars")
    return PositionAction.hold()

TrailingStop dataclass 

TrailingStop(pct)

Exit when price retraces from high water mark.

For longs: Exit if price drops X% from highest price since entry For shorts: Exit if price rises X% from lowest price since entry

Fill price depends on StopFillMode configuration: - STOP_PRICE: Fill at exact trail level (default) - CLOSE_PRICE: Fill at bar's close price (VBT Pro behavior)

Parameters:

Name Type Description Default
pct float

Trail percentage as decimal (0.05 = 5% trail)

 required
Example

rule = TrailingStop(pct=0.05) # 5% trailing stop

evaluate 

evaluate(state)

Exit if price retraces beyond trail.

Uses bar_low/bar_high for intrabar trigger detection. Handles gap-through: if bar opens beyond stop level, fill at open.

Fill price depends on StopFillMode configuration: - STOP_PRICE: Fill at exact trail level (default) - CLOSE_PRICE: Fill at bar's close price - BAR_EXTREME: Fill at bar's low (long) or high (short)

Water mark timing depends on TrailStopTiming configuration: - LAGGED: Use water mark from PREVIOUS bar (default, 1-bar lag) - INTRABAR: Compute "live" water mark using current bar's extreme, then check. VBT Pro compatible: respects StopFillMode for fill price.

Gap-through handling: When bar opens beyond the stop level (gap down for longs, gap up for shorts), the fill is at the open price regardless of StopFillMode. This matches VBT Pro behavior.

Source code in src/ml4t/backtest/risk/position/dynamic.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def evaluate(self, state: PositionState) -> PositionAction:
    """Exit if price retraces beyond trail.

    Uses bar_low/bar_high for intrabar trigger detection.
    Handles gap-through: if bar opens beyond stop level, fill at open.

    Fill price depends on StopFillMode configuration:
    - STOP_PRICE: Fill at exact trail level (default)
    - CLOSE_PRICE: Fill at bar's close price
    - BAR_EXTREME: Fill at bar's low (long) or high (short)

    Water mark timing depends on TrailStopTiming configuration:
    - LAGGED: Use water mark from PREVIOUS bar (default, 1-bar lag)
    - INTRABAR: Compute "live" water mark using current bar's extreme, then check.
                VBT Pro compatible: respects StopFillMode for fill price.

    Gap-through handling: When bar opens beyond the stop level (gap down for
    longs, gap up for shorts), the fill is at the open price regardless of
    StopFillMode. This matches VBT Pro behavior.
    """

    fill_mode = _get_stop_fill_mode_for_trail(state.context)
    trail_timing = _get_trail_stop_timing(state.context)

    if state.is_long:
        return self._evaluate_long(state, fill_mode, trail_timing)
    else:
        return self._evaluate_short(state, fill_mode, trail_timing)

RuleChain dataclass 

RuleChain(rules)

Evaluate rules in order, first non-HOLD action wins.

This is the most common composition pattern - rules are checked in priority order and the first rule to trigger takes effect.

Parameters:

Name Type Description Default
rules list[PositionRule]

List of rules to evaluate in order

 required
Example

chain = RuleChain([ StopLoss(pct=0.05), # Highest priority ScaledExit([(0.10, 0.5)]), # Second priority TighteningTrailingStop([...]), # Third priority TimeExit(bars=20), # Lowest priority ])

evaluate 

evaluate(state)

Evaluate rules in order, return first non-HOLD action.

Source code in src/ml4t/backtest/risk/position/composite.py 
30
31
32
33
34
35
36
def evaluate(self, state: PositionState) -> PositionAction:
    """Evaluate rules in order, return first non-HOLD action."""
    for rule in self.rules:
        action = rule.evaluate(state)
        if action.action != ActionType.HOLD:
            return action
    return PositionAction.hold()

AllOf dataclass 

AllOf(rules)

All rules must return non-HOLD for the action to trigger.

Useful for requiring multiple conditions to be true before exiting. Returns the first rule's action details (pct, stop_price, etc.).

Parameters:

Name Type Description Default
rules list[PositionRule]

List of rules that must all agree

 required
Example

Exit only if both profitable AND held long enough

rule = AllOf([ TakeProfit(pct=0.0), # Must be profitable TimeExit(bars=5), # Must have held 5+ bars ])

evaluate 

evaluate(state)

Return action only if ALL rules return non-HOLD.

Source code in src/ml4t/backtest/risk/position/composite.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def evaluate(self, state: PositionState) -> PositionAction:
    """Return action only if ALL rules return non-HOLD."""
    actions = [rule.evaluate(state) for rule in self.rules]

    if all(a.action != ActionType.HOLD for a in actions):
        # All triggered - return first rule's action with combined reason
        reasons = [a.reason for a in actions if a.reason]
        first = actions[0]
        return PositionAction(
            action=first.action,
            pct=first.pct,
            stop_price=first.stop_price,
            reason=" AND ".join(reasons) if reasons else "",
        )

    return PositionAction.hold()

AnyOf dataclass 

AnyOf(rules)

First rule to return non-HOLD wins (alias for RuleChain).

Semantically equivalent to RuleChain but named for clarity when composing complex rule logic.

Parameters:

Name Type Description Default
rules list[PositionRule]

List of rules where any can trigger

 required
Example

Exit on stop-loss OR signal

rule = AnyOf([ StopLoss(pct=0.05), SignalExit(threshold=0.5), ])

evaluate 

evaluate(state)

Return first non-HOLD action (same as RuleChain).

Source code in src/ml4t/backtest/risk/position/composite.py 
 97
 98
 99
100
101
102
103
def evaluate(self, state: PositionState) -> PositionAction:
    """Return first non-HOLD action (same as RuleChain)."""
    for rule in self.rules:
        action = rule.evaluate(state)
        if action.action != ActionType.HOLD:
            return action
    return PositionAction.hold()

Risk: Portfolio Limits

MaxDrawdownLimit dataclass 

MaxDrawdownLimit(
    max_drawdown=0.2,
    action="liquidate",
    warn_threshold=None,
    reduction_pct=0.0,
)

Bases: PortfolioLimit

Liquidate or halt when drawdown exceeds threshold.

Parameters:

Name Type Description Default
max_drawdown float

Maximum allowed drawdown (0.0-1.0) Default 0.20 = 20% max drawdown

 0.2
action str

Action when breached ("warn", "reduce", "halt", "liquidate") Default "liquidate" - flatten positions and stop new trades

 'liquidate'
warn_threshold float | None

Optional earlier threshold for warnings

 None
reduction_pct float

Required fraction in (0, 1] when action is "reduce"

 0.0
Example

limit = MaxDrawdownLimit(max_drawdown=0.20, warn_threshold=0.15)

Warns at 15% drawdown, liquidates at 20%

MaxPositionsLimit dataclass 

MaxPositionsLimit(max_positions=10, action='halt')

Bases: PortfolioLimit

Limit maximum number of open positions.

Parameters:

Name Type Description Default
max_positions int

Maximum number of simultaneous positions

 10
action str

Action when breached ("warn", "halt")

 'halt'
Example

limit = MaxPositionsLimit(max_positions=10)

Prevents opening more than 10 positions

MaxExposureLimit dataclass 

MaxExposureLimit(max_exposure_pct=0.1, action='warn')

Bases: PortfolioLimit

Limit maximum exposure to a single asset.

Parameters:

Name Type Description Default
max_exposure_pct float

Maximum position size as % of equity (0.0-1.0) Default 0.10 = 10% max per asset

 0.1
action str

Action when breached

 'warn'
Example

limit = MaxExposureLimit(max_exposure_pct=0.10)

No single position can be > 10% of portfolio

DailyLossLimit dataclass 

DailyLossLimit(max_daily_loss_pct=0.02, action='liquidate')

Bases: PortfolioLimit

Liquidate or halt when daily loss exceeds threshold.

Parameters:

Name Type Description Default
max_daily_loss_pct float

Maximum daily loss as % of equity (0.0-1.0) Default 0.02 = 2% max daily loss

 0.02
action str

Action when breached ("warn", "halt", "liquidate")

 'liquidate'
Example

limit = DailyLossLimit(max_daily_loss_pct=0.02)

Halt if down more than 2% today

Strategy Templates

SignalFollowingStrategy

Bases: Strategy

Template for strategies that follow pre-computed signals.

Use this when you have ML predictions, technical indicators, or any pre-computed signal column in your DataFrame.

Class Attributes

signal_column: Name of the signal column in data (default: "signal") position_size: Fraction of equity per position (default: 0.10) allow_shorts: Whether to allow short positions (default: False)

Example

class MyMLStrategy(SignalFollowingStrategy): ... signal_column = "rf_prediction" ... position_size = 0.05 ... ... def should_enter_long(self, signal): ... return signal > 0.7 ... ... def should_exit(self, signal): ... return signal < 0.3

should_enter_long abstractmethod 

should_enter_long(signal)

Return True to open a long position.

Parameters:

Name Type Description Default
signal float

Current signal value for the asset

 required

Returns:

Type Description
bool

True if should enter long position
Source code in src/ml4t/backtest/strategies/templates.py 
62
63
64
65
66
67
68
69
70
71
@abstractmethod
def should_enter_long(self, signal: float) -> bool:
    """Return True to open a long position.

    Args:
        signal: Current signal value for the asset

    Returns:
        True if should enter long position
    """

should_exit abstractmethod 

should_exit(signal)

Return True to close current position.

Parameters:

Name Type Description Default
signal float

Current signal value for the asset

 required

Returns:

Type Description
bool

True if should exit position
Source code in src/ml4t/backtest/strategies/templates.py 
73
74
75
76
77
78
79
80
81
82
@abstractmethod
def should_exit(self, signal: float) -> bool:
    """Return True to close current position.

    Args:
        signal: Current signal value for the asset

    Returns:
        True if should exit position
    """

should_enter_short 

should_enter_short(signal)

Return True to open a short position.

Override this method for short strategies. Default returns False.

Parameters:

Name Type Description Default
signal float

Current signal value for the asset

 required

Returns:

Type Description
bool

True if should enter short position
Source code in src/ml4t/backtest/strategies/templates.py 
84
85
86
87
88
89
90
91
92
93
94
95
def should_enter_short(self, signal: float) -> bool:
    """Return True to open a short position.

    Override this method for short strategies. Default returns False.

    Args:
        signal: Current signal value for the asset

    Returns:
        True if should enter short position
    """
    return False

on_data 

on_data(timestamp, data, context, broker)

Process each bar and generate orders based on signals.

Source code in src/ml4t/backtest/strategies/templates.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def on_data(
    self,
    timestamp: datetime,
    data: dict[str, dict],
    context: dict[str, Any],
    broker: Broker,
) -> None:
    """Process each bar and generate orders based on signals."""
    for asset, bar in data.items():
        # Signals are nested under 'signals' dict in DataFeed output
        signals = bar.get("signals", {})
        signal = signals.get(self.signal_column, 0) if signals else 0
        if signal is None:
            signal = 0

        position = broker.get_position(asset)
        price = bar.get("close", 0)

        if position is None:
            # No position - check for entry
            fractional = _use_fractional(self.allow_fractional, broker)
            if self.should_enter_long(signal):
                equity = broker.get_account_value()
                raw_shares = (equity * self.position_size) / price if price > 0 else 0
                shares = raw_shares if fractional else int(raw_shares)
                if shares > 0:
                    broker.submit_order(asset, shares)
            elif self.allow_shorts and self.should_enter_short(signal):
                equity = broker.get_account_value()
                raw_shares = (equity * self.position_size) / price if price > 0 else 0
                shares = raw_shares if fractional else int(raw_shares)
                if shares > 0:
                    broker.submit_order(asset, -shares)
        else:
            # Have position - check for exit
            if self.should_exit(signal):
                broker.close_position(asset)

MomentumStrategy 

MomentumStrategy()

Bases: Strategy

Template for momentum/trend-following strategies.

Enters long when asset has positive momentum over lookback period, exits when momentum turns negative.

Class Attributes

lookback: Number of bars for momentum calculation (default: 20) entry_threshold: Minimum return to enter (default: 0.05 = 5%) exit_threshold: Return level to exit (default: -0.02 = -2%) position_size: Fraction of equity per position (default: 0.10)

Example

class MyMomentum(MomentumStrategy): ... lookback = 60 # 60-day momentum ... entry_threshold = 0.10 # Enter on 10% gain ... exit_threshold = 0.0 # Exit when momentum turns negative

Source code in src/ml4t/backtest/strategies/templates.py 
161
162
def __init__(self) -> None:
    self.price_history: dict[str, list[float]] = defaultdict(list)

calculate_momentum 

calculate_momentum(prices)

Calculate momentum as return over lookback period.

Parameters:

Name Type Description Default
prices list[float]

List of prices (most recent last)

 required

Returns:

Type Description
float

Return from first to last price
Source code in src/ml4t/backtest/strategies/templates.py 
164
165
166
167
168
169
170
171
172
173
174
175
def calculate_momentum(self, prices: list[float]) -> float:
    """Calculate momentum as return over lookback period.

    Args:
        prices: List of prices (most recent last)

    Returns:
        Return from first to last price
    """
    if len(prices) < 2 or prices[0] == 0:
        return 0.0
    return (prices[-1] / prices[0]) - 1

on_data 

on_data(timestamp, data, context, broker)

Process each bar and trade based on momentum.

Source code in src/ml4t/backtest/strategies/templates.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def on_data(
    self,
    timestamp: datetime,
    data: dict[str, dict],
    context: dict[str, Any],
    broker: Broker,
) -> None:
    """Process each bar and trade based on momentum."""
    for asset, bar in data.items():
        close = bar.get("close")
        if close is None or close <= 0:
            continue

        # Track price history
        self.price_history[asset].append(close)

        # Wait for enough history
        if len(self.price_history[asset]) < self.lookback:
            continue

        # Keep only lookback period
        self.price_history[asset] = self.price_history[asset][-self.lookback :]

        # Calculate momentum
        momentum = self.calculate_momentum(self.price_history[asset])
        position = broker.get_position(asset)

        if position is None and momentum > self.entry_threshold:
            # Enter long on strong momentum
            equity = broker.get_account_value()
            raw_shares = (equity * self.position_size) / close
            fractional = _use_fractional(self.allow_fractional, broker)
            shares = raw_shares if fractional else int(raw_shares)
            if shares > 0:
                broker.submit_order(asset, shares)
        elif position is not None and momentum < self.exit_threshold:
            # Exit on weak momentum
            broker.close_position(asset)

MeanReversionStrategy 

MeanReversionStrategy()

Bases: Strategy

Template for mean-reversion strategies.

Buys when price is below moving average by a threshold, sells when price reverts to the mean.

Class Attributes

lookback: Number of bars for mean calculation (default: 20) entry_zscore: Z-score threshold to enter (default: -2.0) exit_zscore: Z-score threshold to exit (default: 0.0) position_size: Fraction of equity per position (default: 0.10)

Example

class MyMeanReversion(MeanReversionStrategy): ... lookback = 30 ... entry_zscore = -2.5 # More extreme entry ... exit_zscore = 0.5 # Take profit above mean

Source code in src/ml4t/backtest/strategies/templates.py 
242
243
def __init__(self) -> None:
    self.price_history: dict[str, list[float]] = defaultdict(list)

calculate_zscore 

calculate_zscore(prices, current)

Calculate z-score of current price vs historical distribution.

Parameters:

Name Type Description Default
prices list[float]

Historical prices

 required
current float

Current price

 required

Returns:

Type Description
float | None

Z-score or None if insufficient data
Source code in src/ml4t/backtest/strategies/templates.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def calculate_zscore(self, prices: list[float], current: float) -> float | None:
    """Calculate z-score of current price vs historical distribution.

    Args:
        prices: Historical prices
        current: Current price

    Returns:
        Z-score or None if insufficient data
    """
    if len(prices) < 2:
        return None

    try:
        avg = mean(prices)
        std = stdev(prices)
        if std == 0:
            return None
        return (current - avg) / std
    except Exception:
        return None

on_data 

on_data(timestamp, data, context, broker)

Process each bar and trade based on mean reversion.

Source code in src/ml4t/backtest/strategies/templates.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def on_data(
    self,
    timestamp: datetime,
    data: dict[str, dict],
    context: dict[str, Any],
    broker: Broker,
) -> None:
    """Process each bar and trade based on mean reversion."""
    for asset, bar in data.items():
        close = bar.get("close")
        if close is None or close <= 0:
            continue

        # Track price history
        self.price_history[asset].append(close)

        # Wait for enough history
        if len(self.price_history[asset]) < self.lookback:
            continue

        # Keep only lookback period
        prices = self.price_history[asset][-self.lookback :]
        self.price_history[asset] = prices

        # Calculate z-score
        zscore = self.calculate_zscore(prices[:-1], close)
        if zscore is None:
            continue

        position = broker.get_position(asset)

        if position is None and zscore < self.entry_zscore:
            # Enter long on oversold condition
            equity = broker.get_account_value()
            raw_shares = (equity * self.position_size) / close
            fractional = _use_fractional(self.allow_fractional, broker)
            shares = raw_shares if fractional else int(raw_shares)
            if shares > 0:
                broker.submit_order(asset, shares)
        elif position is not None and zscore > self.exit_zscore:
            # Exit on mean reversion
            broker.close_position(asset)

LongShortStrategy 

LongShortStrategy()

Bases: Strategy

Template for long/short equity strategies.

Ranks assets by a signal and goes long top N, short bottom N.

Class Attributes

signal_column: Column to rank assets by (default: "signal") long_count: Number of assets to go long (default: 5) short_count: Number of assets to go short (default: 5) position_size: Fraction of equity per position (default: 0.05) rebalance_frequency: Bars between rebalancing (default: 20)

Example

class MyLongShort(LongShortStrategy): ... signal_column = "momentum_score" ... long_count = 10 ... short_count = 10 ... rebalance_frequency = 21 # Monthly

Source code in src/ml4t/backtest/strategies/templates.py 
339
340
341
342
def __init__(self) -> None:
    self.bar_count = 0
    self._schedule_config: BacktestConfig | None = None
    self._schedule_evaluator: _OnlineRebalanceEvaluator | None = None

on_prepare 

on_prepare(broker, config=None)

Retain causal calendar metadata for online schedule evaluation.

Source code in src/ml4t/backtest/strategies/templates.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
def on_prepare(
    self,
    broker: Any,
    config: BacktestConfig | None = None,
) -> None:
    """Retain causal calendar metadata for online schedule evaluation."""
    self._schedule_config = config
    self._schedule_evaluator = None
    if self.rebalance_schedule is not None and config is not None:
        self._schedule_evaluator = _OnlineRebalanceEvaluator(
            self.rebalance_schedule,
            calendar=config.resolved_calendar,
            timezone=config.resolved_timezone,
            session_start_time=config.resolved_session_start_time,
            data_frequency=config.resolved_data_frequency,
            timestamp_semantics=config.resolved_timestamp_semantics,
        )

rank_assets 

rank_assets(data)

Rank assets by signal and return long/short lists.

Parameters:

Name Type Description Default
data dict[str, dict]

Current bar data for all assets

 required

Returns:

Type Description
tuple[list[str], list[str]]

Tuple of (long_assets, short_assets)
Source code in src/ml4t/backtest/strategies/templates.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
def rank_assets(self, data: dict[str, dict]) -> tuple[list[str], list[str]]:
    """Rank assets by signal and return long/short lists.

    Args:
        data: Current bar data for all assets

    Returns:
        Tuple of (long_assets, short_assets)
    """
    # Collect signals (signals are nested under 'signals' dict)
    signals: list[tuple[str, float]] = []
    for asset, bar in data.items():
        bar_signals = bar.get("signals", {})
        signal = bar_signals.get(self.signal_column) if bar_signals else None
        if signal is not None:
            signals.append((asset, signal))

    if not signals:
        return [], []

    # Sort by signal (high to low)
    signals.sort(key=lambda x: x[1], reverse=True)

    # Top N for long, bottom N for short
    long_assets = [s[0] for s in signals[: self.long_count]]
    short_assets = [s[0] for s in signals[-self.short_count :]]

    # Don't short the same assets we're going long
    short_assets = [a for a in short_assets if a not in long_assets]

    return long_assets, short_assets

on_data 

on_data(timestamp, data, context, broker)

Rebalance portfolio periodically based on rankings.

Source code in src/ml4t/backtest/strategies/templates.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def on_data(
    self,
    timestamp: datetime,
    data: dict[str, dict],
    context: dict[str, Any],
    broker: Broker,
) -> None:
    """Rebalance portfolio periodically based on rankings."""
    self.bar_count += 1

    if self.rebalance_schedule is not None:
        if self._schedule_config is None:
            raise ValueError("rebalance_schedule is set but was not prepared before execution")
        if self._schedule_evaluator is None:
            raise RuntimeError("rebalance schedule evaluator was not initialized")
        if not self._schedule_evaluator.evaluate(timestamp):
            return
    elif self.bar_count % self.rebalance_frequency != 1:
        return

    # Get current rankings
    long_assets, short_assets = self.rank_assets(data)
    target_assets = set(long_assets + short_assets)

    # Close positions not in target
    for asset in list(broker.get_positions().keys()):
        if asset not in target_assets:
            broker.close_position(asset)

    # Open/adjust positions
    equity = broker.get_account_value()
    fractional = _use_fractional(self.allow_fractional, broker)

    for asset in long_assets:
        price = data.get(asset, {}).get("close", 0)
        if price <= 0:
            continue

        position = broker.get_position(asset)
        raw_shares = (equity * self.position_size) / price
        target_shares = raw_shares if fractional else int(raw_shares)

        if position is None and target_shares > 0:
            broker.submit_order(asset, target_shares)

    for asset in short_assets:
        price = data.get(asset, {}).get("close", 0)
        if price <= 0:
            continue

        position = broker.get_position(asset)
        raw_shares = (equity * self.position_size) / price
        target_shares = raw_shares if fractional else int(raw_shares)

        if position is None and target_shares > 0:
            broker.submit_order(asset, -target_shares)