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,
)

Event-driven backtesting engine.

The Engine orchestrates the backtest by iterating through market data, managing the broker, and calling the strategy on each bar.

Execution Flow
  1. Initialize strategy (on_start)
  2. For each bar: a. Update broker with current prices b. Process pending exits (NEXT_BAR_OPEN mode) c. Evaluate position rules (stops, trails) d. Process pending orders e. Call strategy.on_data() f. Process new orders (SAME_BAR mode) g. Update water marks h. Record equity
  3. Close open positions
  4. Finalize strategy (on_end)

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 
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
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,
):
    from .config import BacktestConfig as ConfigCls

    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
    self.broker = Broker.from_config(
        self.config,
        contract_specs=contract_specs,
        market_impact_model=market_impact_model,
        execution_limits=execution_limits,
    )
    self.equity_curve: list[tuple[datetime, float]] = []
    self.portfolio_state: list[tuple[datetime, float, float, float, float, int]] = []

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

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.
Source code in src/ml4t/backtest/engine.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
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
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.
    """
    # Lazy calendar initialization (zero cost if unused)
    is_trading_day_fn = None
    if self.config and self.config.resolved_calendar:
        from .calendar import get_calendar, is_trading_day

        self._calendar = get_calendar(self.config.resolved_calendar)
        is_trading_day_fn = is_trading_day

    self.strategy.on_prepare(self.broker, self.feed.timestamps, self.config)
    self.strategy.on_start(self.broker)

    # Date-level cache for trading day checks (significant speedup for intraday data)
    trading_day_cache: dict[date, bool] = {}

    for timestamp, assets_data, context in self.feed:
        # Calendar session enforcement
        calendar_id = self.config.resolved_calendar if self.config else None
        if (
            self._calendar
            and calendar_id
            and self.config
            and self.config.enforce_sessions
            and is_trading_day_fn
        ):
            # For daily data, check trading day; for intraday, check market hours
            if self.config.resolved_data_frequency == DataFrequency.DAILY:
                if not is_trading_day_fn(calendar_id, timestamp.date()):
                    self._skipped_bars += 1
                    continue
            else:
                # Intraday: use cached trading day check (avoid expensive calendar.valid_days per bar)
                bar_date = timestamp.date()
                if bar_date not in trading_day_cache:
                    trading_day_cache[bar_date] = is_trading_day_fn(calendar_id, bar_date)
                if not trading_day_cache[bar_date]:
                    self._skipped_bars += 1
                    continue

        prices = getattr(assets_data, "_prices", None)
        opens = getattr(assets_data, "_opens", None)
        highs = getattr(assets_data, "_highs", None)
        lows = getattr(assets_data, "_lows", None)
        closes = getattr(assets_data, "_closes", None)
        volumes = getattr(assets_data, "_volumes", None)
        bids = getattr(assets_data, "_bids", None)
        asks = getattr(assets_data, "_asks", None)
        mids = getattr(assets_data, "_mids", None)
        bid_sizes = getattr(assets_data, "_bid_sizes", None)
        ask_sizes = getattr(assets_data, "_ask_sizes", None)
        signals = getattr(assets_data, "_signals", None)

        if (
            prices is None
            or opens is None
            or highs is None
            or lows is None
            or closes is None
            or volumes is None
            or bids is None
            or asks is None
            or mids is None
            or bid_sizes is None
            or ask_sizes is None
            or signals is None
        ):
            prices = {
                a: price
                for a, d in assets_data.items()
                if (price := d.get("price", d.get("close"))) is not None
            }
            opens = {}
            highs = {}
            lows = {}
            closes = {
                a: close
                for a, d in assets_data.items()
                if (close := d.get("close", d.get("price"))) is not None
            }
            for asset, data in assets_data.items():
                base_price = data.get("close")
                if base_price is None:
                    base_price = data.get("price")
                opens[asset] = data.get("open") if data.get("open") is not None else base_price
                highs[asset] = data.get("high") if data.get("high") is not None else base_price
                lows[asset] = data.get("low") if data.get("low") is not None else base_price
            volumes = {a: d.get("volume", 0) for a, d in assets_data.items()}
            bids = {a: d["bid"] for a, d in assets_data.items() if d.get("bid") is not None}
            asks = {a: d["ask"] for a, d in assets_data.items() if d.get("ask") is not None}
            mids = {a: d["mid"] for a, d in assets_data.items() if d.get("mid") is not None}
            bid_sizes = {
                a: d["bid_size"]
                for a, d in assets_data.items()
                if d.get("bid_size") is not None
            }
            ask_sizes = {
                a: d["ask_size"]
                for a, d in assets_data.items()
                if d.get("ask_size") is not None
            }
            signals = {a: d.get("signals", {}) for a, d in assets_data.items()}

        self.broker._update_time(
            timestamp,
            prices,
            opens,
            highs,
            lows,
            closes,
            volumes,
            bids,
            asks,
            mids,
            bid_sizes,
            ask_sizes,
            signals,
        )

        # Process pending exits from NEXT_BAR_OPEN mode (fills at open)
        # This must happen BEFORE evaluate_position_rules() to clear deferred exits
        self.broker._process_pending_exits()

        # Optional strategy phase for opening orders that must receive risk
        # protection during the current bar. Existing strategies inherit a no-op.
        self.strategy.on_before_risk(timestamp, assets_data, context, self.broker)

        # Evaluate position rules (stops, trails, etc.) - generates exit orders
        self.broker.evaluate_position_rules()

        if self.execution_mode == ExecutionMode.NEXT_BAR:
            # Next-bar mode: process pending orders at open price
            self.broker._process_orders(use_open=True)
            # Strategy generates new orders
            self.strategy.on_data(timestamp, assets_data, context, self.broker)
            # MOC orders are the one next-bar exception: they execute on the
            # current session close after strategy logic runs.
            self.broker._process_orders(
                order_types={OrderType.MOC},
                include_orders_this_bar=True,
            )
        else:
            # Same-bar mode: process before and after strategy
            self.broker._process_orders()
            self.strategy.on_data(timestamp, assets_data, context, self.broker)
            self.broker._process_orders()

        # Update water marks at END of bar, AFTER all orders processed
        # This ensures new positions get their HWM updated from entry bar's high
        # VBT Pro behavior: HWM updated at bar end, used in NEXT bar's trail evaluation
        self.broker._update_water_marks()

        self._record_portfolio_state(timestamp)

    self.strategy.on_end(self.broker)
    return self._generate_results()

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.
Source code in src/ml4t/backtest/engine.py 
259
260
261
262
263
264
265
266
267
268
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.
    """
    return self.run().to_dict()

from_config classmethod 

from_config(
    feed,
    strategy,
    config,
    *,
    contract_specs=None,
    market_impact_model=None,
    execution_limits=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

Returns:

Type Description
Engine

Configured Engine instance
Source code in src/ml4t/backtest/engine.py 
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
@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,
) -> 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.)

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

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,
)

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

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 
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
587
588
589
590
591
592
593
594
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,
) -> 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.)

    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,
    ).run()

Strategy

Bases: ABC

Base strategy class.

on_before_risk 

on_before_risk(timestamp, data, context, broker)

Called before position rules are evaluated for the current bar.

Source code in src/ml4t/backtest/strategy.py 
12
13
14
15
16
17
18
19
20
def on_before_risk(
    self,
    timestamp: datetime,
    data: dict[str, dict],
    context: dict[str, Any],
    broker: Any,
) -> None:
    """Called before position rules are evaluated for the current bar."""
    return None

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 
22
23
24
25
26
27
28
29
30
31
@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_start 

on_start(broker)

Called before backtest starts.

Source code in src/ml4t/backtest/strategy.py 
33
34
35
def on_start(self, broker: Any) -> None:  # noqa: B027
    """Called before backtest starts."""
    pass

on_prepare 

on_prepare(broker, timestamps, config=None)

Called before on_start with access to the full feed timestamp universe.

Source code in src/ml4t/backtest/strategy.py 
37
38
39
40
41
42
43
44
def on_prepare(
    self,
    broker: Any,
    timestamps: Sequence[datetime],
    config: Any | None = None,
) -> None:
    """Called before on_start with access to the full feed timestamp universe."""
    return None

on_end 

on_end(broker)

Called after backtest ends.

Source code in src/ml4t/backtest/strategy.py 
46
47
48
def on_end(self, broker: Any) -> None:  # noqa: B027
    """Called after backtest ends."""
    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,
    bid_col=None,
    ask_col=None,
    mid_col=None,
    bid_size_col=None,
    ask_size_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.

Memory Efficiency
  • 1M bars: ~100 MB (was ~1 GB with pre-converted dicts)
  • 10M bars: ~1 GB (vs ~10+ GB with dicts)
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 
 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
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
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,
    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,
):
    if feed_spec is not None and contract is not None:
        raise ValueError("Pass either feed_spec or contract, not both")

    self.prices = (
        prices_df
        if prices_df is not None
        else (pl.scan_parquet(prices_path).collect() if prices_path else None)
    )
    self.signals = (
        signals_df
        if signals_df is not None
        else (pl.scan_parquet(signals_path).collect() if signals_path else None)
    )
    self.context = (
        context_df
        if context_df is not None
        else (pl.scan_parquet(context_path).collect() 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)
    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,
        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
    self._entity_col = self.feed_spec.entity_col
    self._price_col = self.feed_spec.price_col
    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._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:
        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._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_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,
    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,
    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(),
)

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 QuantConnect LEAN's default behavior - "realistic": Conservative settings for realistic simulation - "ibkr_us_stocks_fixed": Interactive Brokers US stocks fixed pricing

Source code in src/ml4t/backtest/config.py 
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
@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 QuantConnect LEAN's default behavior
    - "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 
1149
1150
1151
1152
1153
1154
1155
@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 
 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
 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
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
@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",
            "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",
            },
            "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"},
            "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",
                "bid_col",
                "ask_col",
                "mid_col",
                "bid_size_col",
                "ask_size_col",
                "calendar",
                "timezone",
                "data_frequency",
                "bar_type",
                "timestamp_semantics",
                "session_start_time",
            },
        }
        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", {})
    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")),
        # 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")),
        # 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,
        metadata=dict(metadata),
    )

to_yaml 

to_yaml(path)

Save config to YAML file.

Source code in src/ml4t/backtest/config.py 
1143
1144
1145
1146
1147
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 
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
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,
        },
        "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,
        },
        "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),
        "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 
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
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
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] = []

    # 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_spread < 0:
        issues.append(f"slippage_spread ({self.slippage_spread}) must be >= 0")

    if any(spread < 0 for spread in self.slippage_spread_by_asset.values()):
        issues.append("slippage_spread_by_asset values must all be >= 0")

    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. "
            "Common values: 0 (instant), 1 (T+1), 2 (T+2 US equities)."
        )

    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 
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
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 
451
452
453
454
455
456
457
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 
460
461
462
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,
    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 
 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
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
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,
    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._bar_index: int = 0

    # 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

    # 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.positions: dict[str, Position] = {}
    self.orders: list[Order] = []
    self.pending_orders: list[Order] = []
    self.fills: list[Fill] = []
    self.trades: list[Trade] = []
    self._order_counter = 0
    self._current_time: datetime | None = None
    self._current_prices: dict[str, float] = {}  # FeedSpec.price_col values
    self._current_opens: dict[str, float] = {}  # open prices for next-bar execution
    self._current_highs: dict[str, float] = {}  # high prices for limit/stop checks
    self._current_lows: dict[str, float] = {}  # low prices for limit/stop checks
    self._current_closes: dict[str, float] = {}
    self._current_volumes: dict[str, float] = {}
    self._current_bids: dict[str, float] = {}
    self._current_asks: dict[str, float] = {}
    self._current_mids: dict[str, float] = {}
    self._current_bid_sizes: dict[str, float] = {}
    self._current_ask_sizes: dict[str, float] = {}
    self._current_signals: dict[str, dict[str, float]] = {}
    self._last_prices: dict[str, float] = {}
    self._asset_bars_seen: dict[str, int] = {}
    self._rebalance_counter = 0
    self._orders_this_bar: list[Order] = []  # Orders placed this bar (for next-bar mode)
    self._orders_this_bar_ids: set[str] = set()

    # Risk management
    self._position_rules: Any = None  # Global position rules
    self._position_rules_by_asset: dict[str, Any] = {}  # Per-asset rules
    self._pending_exits: dict[str, dict] = {}  # asset -> {reason, pct} for NEXT_BAR_OPEN mode

    # Execution model (volume limits and market impact)
    self.execution_limits = execution_limits  # ExecutionLimits instance
    self.market_impact_model = market_impact_model  # MarketImpactModel instance
    self._partial_orders: dict[str, float] = {}  # order_id -> remaining quantity
    self._filled_this_bar: set[str] = set()  # order_ids that had fills this bar

    # VBT Pro compatibility: prevent same-bar re-entry after stop exit
    self._stop_exits_this_bar: set[str] = set()  # assets that had stop exits this bar

    # VBT Pro compatibility: track positions created this bar
    # New positions should NOT have HWM updated from entry bar's high
    # VBT Pro uses CLOSE for initial HWM on entry bar, then updates from HIGH next bar
    self._positions_created_this_bar: set[str] = set()

    # 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)

    # 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._last_session_id: int | None = None  # Track current session for boundary detection

    # Extracted orchestration components (Phase B1 alpha-reset)
    self._order_book = OrderBook(self)
    self._risk_engine = RiskEngine(self)
    self._fill_engine = FillEngine(self)
    self._execution_engine = ExecutionEngine(self)
    self._portfolio_ledger = PortfolioLedger(self)

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 
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
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)
    """
    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 
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
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

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 
 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
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

cancel_order 

cancel_order(order_id)
Source code in src/ml4t/backtest/broker.py 
994
995
def cancel_order(self, order_id: str) -> bool:
    return self._order_book.cancel_order(order_id)

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 
716
717
718
719
720
721
722
723
724
725
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 
727
728
729
730
731
732
733
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 
735
736
737
738
739
740
741
def get_cash(self) -> float:
    """Get current cash balance.

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

equity 

equity()

Calculate current marked account equity.

Source code in src/ml4t/backtest/broker.py 
743
744
745
def equity(self) -> float:
    """Calculate current marked account equity."""
    return self._portfolio_ledger.get_account_value()

get_account_value 

get_account_value()

Alias for equity().

Source code in src/ml4t/backtest/broker.py 
747
748
749
def get_account_value(self) -> float:
    """Alias for equity()."""
    return self.equity()

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 
751
752
753
754
755
756
757
758
759
760
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)

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 
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
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
    """
    if asset is not None:
        self._position_rules_by_asset[asset] = rules
    else:
        self._position_rules = rules

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 
1494
1495
1496
1497
1498
1499
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
1556
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
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

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,
    _created_bar_index=0,
    _signal_price=None,
    _risk_exit_reason=None,
    _exit_reason=None,
    _risk_fill_price=None,
)

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,
)

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",
    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,
)

Round-trip trade (closed or open).

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

For open trades (status="open"), exit_time and exit_price represent mark-to-market values at the end of the backtest period.

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 
244
245
246
247
248
249
250
251
252
253
254
255
256
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 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
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 
275
276
277
278
279
280
281
282
283
284
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 
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
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
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 
338
339
340
341
342
343
344
345
346
347
348
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 VectorBT call_seq='auto' behavior.

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.

Results

BacktestResult dataclass 

BacktestResult(
    trades,
    equity_curve,
    fills,
    metrics,
    predictions=None,
    config=None,
    equity=None,
    trade_analyzer=None,
    portfolio_state=list(),
    _trades_df=None,
    _equity_df=None,
    _fills_df=None,
    _portfolio_state_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)
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

to_fills_dataframe 

to_fills_dataframe()

Convert fills to Polars DataFrame.

Source code in src/ml4t/backtest/result.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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def to_fills_dataframe(self) -> pl.DataFrame:
    """Convert fills to Polars DataFrame."""
    if self._fills_df is not None:
        return self._fills_df

    if not self.fills:
        return pl.DataFrame(schema=self._fills_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,
            }
        )

    self._fills_df = pl.DataFrame(records, schema=self._fills_schema())
    return self._fills_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 
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
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

    if not self.portfolio_state:
        return pl.DataFrame(schema=self._portfolio_state_schema())

    self._portfolio_state_df = (
        pl.DataFrame(
            self.portfolio_state,
            schema=[
                "timestamp",
                "equity",
                "cash",
                "gross_exposure",
                "net_exposure",
                "open_positions",
            ],
            orient="row",
        )
        .sort("timestamp")
        .cast(self._portfolio_state_schema())
    )
    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 
208
209
210
211
212
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_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, 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" (actually exited) 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 
 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 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, 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" (actually exited) 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

    if not self.trades:
        return pl.DataFrame(schema=self._trades_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,
                "status": t.status,
            }
        )

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

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 
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
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_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 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
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,
            "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_parquet 

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

Export backtest result to Parquet files.

Creates directory structure

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

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", "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 component names to file paths
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
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
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
            predictions.parquet
            equity.parquet
            portfolio_state.parquet
            daily_pnl.parquet
            metrics.json
            config.yaml (if config available)
            spec.yaml (if config available)

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

    Returns:
        Dict mapping component names to file paths
    """
    path = Path(path)
    path.mkdir(parents=True, exist_ok=True)

    if include is None:
        include = [
            "trades",
            "fills",
            "predictions",
            "equity",
            "portfolio_state",
            "daily_pnl",
            "metrics",
            "config",
            "spec",
        ]

    written: dict[str, Path] = {}

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

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

    if "predictions" in include and self.predictions is not None:
        predictions_path = path / "predictions.parquet"
        self.to_predictions_dataframe().write_parquet(predictions_path, compression=compression)
        written["predictions"] = predictions_path

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

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

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

    if "metrics" in include:
        metrics_path = path / "metrics.json"
        # Filter to JSON-serializable metrics
        serializable = {}
        for k, v in self.metrics.items():
            if isinstance(v, int | float | str | bool | type(None)):
                serializable[k] = v
            elif isinstance(v, datetime):
                serializable[k] = v.isoformat()
            else:
                # Handle numpy scalars (np.float64, np.int64, etc.)
                try:
                    import numpy as np

                    if isinstance(v, np.generic):
                        serializable[k] = v.item()
                except (ImportError, AttributeError):
                    pass  # Skip if numpy not available or not a numpy type
        with open(metrics_path, "w") as f:
            json.dump(serializable, f, indent=2)
        written["metrics"] = metrics_path

    if "config" in include and self.config is not None:
        config_path = path / "config.yaml"
        try:
            import yaml

            with open(config_path, "w") as f:
                yaml.dump(self.config.to_dict(), f, default_flow_style=False)
            written["config"] = config_path
        except (ImportError, AttributeError):
            pass  # Skip if yaml not available or config has no to_dict

    if "spec" in include and self.config is not None:
        spec_path = path / "spec.yaml"
        try:
            import yaml

            with open(spec_path, "w") as f:
                yaml.dump(self.to_spec_dict(), f, default_flow_style=False, sort_keys=False)
            written["spec"] = spec_path
        except (ImportError, AttributeError):
            pass

    return written

Execution: Market Impact

LinearImpact dataclass 

LinearImpact(coefficient=0.1, permanent_fraction=0.5)

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
permanent_fraction float

Fraction of impact that is permanent (0-1) Remainder is temporary and reverts

 0.5
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 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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,
)

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
Example

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

Warns at 15% drawdown, halts 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", "reduce", "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 
60
61
62
63
64
65
66
67
68
69
@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 
71
72
73
74
75
76
77
78
79
80
@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 
82
83
84
85
86
87
88
89
90
91
92
93
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 
 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
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 
159
160
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 
162
163
164
165
166
167
168
169
170
171
172
173
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 
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
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 
240
241
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 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
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 
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
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 
337
338
339
def __init__(self) -> None:
    self.bar_count = 0
    self._resolved_schedule: frozenset[datetime] | None = None

on_prepare 

on_prepare(broker, timestamps, config=None)

Resolve optional schedule-based rebalance gating before the run starts.

Source code in src/ml4t/backtest/strategies/templates.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def on_prepare(
    self,
    broker: Any,
    timestamps: Sequence[datetime],
    config: BacktestConfig | None = None,
) -> None:
    """Resolve optional schedule-based rebalance gating before the run starts."""
    if self.rebalance_schedule is None:
        self._resolved_schedule = None
        return
    calendar = config.resolved_calendar if config is not None else None
    timezone = config.resolved_timezone if config is not None else "UTC"
    feed_spec = config.resolved_feed_spec if config is not None else None
    resolved = resolve_rebalance_timestamps(
        timestamps,
        self.rebalance_schedule,
        feed_spec=feed_spec,
        calendar=calendar,
        timezone=timezone,
    )
    self._resolved_schedule = frozenset(resolved.to_list())

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 
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
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 
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
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._resolved_schedule is None:
            raise ValueError("rebalance_schedule is set but was not prepared before execution")
        if timestamp not in self._resolved_schedule:
            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)