Engines

BacktestEngine

BacktestEngine

BacktestEngine(strategy: Strategy, sesh: SeshSimulatorBase, _feed_type: Literal['tick', 'kline'] = 'tick')

Bases: _StoreBuilderMixin

Unified backtesting engine.

# Ticks
bt = BacktestEngine.by_ticks(
    strategy = MyStrategy(),
    data     = (TickData("BTCUSDT", ...),),
    sesh     = sesh,
)
bt.run()

# Multi-symbol klines
bt = BacktestEngine.by_klines(
    strategy = MyStrategy(),
    data     = (
        KlineData("BTCUSDT", ..., timeframe='5m'),
        KlineData("ETHUSDT", ..., timeframe='1m'),
    ),
    sesh     = sesh,
)
bt.run()

Result

engine.stats - performance metrics engine.strategy - the strategy with its final state engine.broker - the internal broker engine.sesh - the SeshSimulatorBase provided

Source code in src/tradetropy/backtest/engine.py

def __init__(
    self,
    strategy: Strategy,
    sesh: SeshSimulatorBase,
    _feed_type: Literal["tick", "kline"] = "tick",
):
    if not isinstance(sesh, SeshSimulatorBase):
        raise ConfigError(
            f"`sesh` must be a simulated session (SeshSimulatorBase), "
            f"received {type(sesh).__name__}. "
            "A live session cannot be used in a backtest. "
            "Use SeshMT5Sim, SeshBybitSim, SeshCCXTSim or another simulated session."
        )
    self.strategy = strategy
    self._sesh = sesh
    # Inject the feed_type into the session: builds the internal broker if the
    # session was created in automatic mode (feed_type=None) or validates that
    # the explicit feed_type matches the engine's.
    if hasattr(sesh, "_bind_feed_type"):
        sesh._bind_feed_type(_feed_type)
    self._broker = sesh._broker
    self._feed_type = _feed_type

    self._tick_stores: dict = {}
    self._ohlc_stores: dict = {}
    self._stats: "Stats | None" = None
    self._plot_config = {}

    # When True, run() finalizes metrics via the pandas-free
    # stats._fast.compute_stats_fast (optimize/pool worker hot path, so the
    # child process never imports pandas). Numeric parity with compute_stats
    # is guarded by test_stats_fast_parity.py. Default False keeps the full
    # pandas Stats object for normal run()/plot().
    self._fast_stats: bool = False

    # Controls the "Stats: insufficient sample" UserWarning emitted by
    # compute_stats() when the sample is too small to trust the
    # annualized/trade-distribution metrics. The gating itself (those
    # metrics zeroed to NaN, Stats["_low_sample"] = True) always applies
    # regardless of this flag; only the warning message is optional. run()
    # exposes this as stats_warn=. Default True keeps existing behavior.
    self._stats_warn: bool = True

    # Typed data - stored in by_ticks/by_klines, used in run()
    self._tick_inputs: tuple[TickData, ...] = ()
    self._kline_inputs: tuple[KlineData, ...] = ()

    # Timestamp alignment in multi-symbol tick mode
    self._align_by_ts: bool = False

    # Optional recorded L2 order book replayed alongside the ticks (tick
    # mode only). Built into LiveBookRings and drained in the tick loop so
    # DeepTrades / L2 metrics work in the backtest exactly as in replay.
    self._book_inputs: dict = {}
    self._sync_book: bool = False
    self._book_rings: dict = {}
    self._book_replayer = None
    self._book_sync_reports: dict = {}

stats property

stats: 'Stats | None'

Performance metrics calculated automatically at the end of run().

Returns None if there were no trades or no broker is configured.

Returns:

Name	Type	Description
Stats	'Stats | None'	Performance metrics or None

Example

bt = BacktestEngine.by_ticks(strategy, data=(ticks,), sesh=sesh).run() print(bt.stats) print(bt.stats['Sharpe Ratio']) trades = bt.stats.trades equity = bt.stats.equity_curve

out_of_money property

out_of_money: bool

True if the backtest ended because the account was wiped (equity <= 0) or a margin stop-out fired, liquidating open positions and stopping the run cleanly (mirrors backtesting.py's out-of-money behavior).

stopped_early property

stopped_early: bool

True if the backtest terminated early (out-of-money or stop-out).

feed_type property

feed_type: FeedType

Engine feed type: 'tick' or 'kline'.

tick_store property

tick_store: 'TickDataStore | None'

First available TickDataStore (compatibility).

tick_stores property

tick_stores: dict

Full access: {symbol: TickDataStore}.

by_ticks classmethod

by_ticks(strategy: Strategy, data: Tuple[TickData, ...], sesh: 'SeshSimulatorBase | None' = None, align_by_ts: bool = False, book: 'BookData | tuple | list | None' = None, sync_book: bool = False) -> 'BacktestEngine'

Prepare engine for tick data.

Data must be a tuple of TickData, one per symbol. The engine does NOT run here - call .run() to start the loop.

Parameters:

Name	Type	Description	Default
strategy	Strategy	Trading strategy instance	required
data	tuple	Tuple of TickData objects	required
sesh	SeshSimulatorBase	Optional session for testing	None
align_by_ts	bool	Align symbols by timestamp instead of position	False
book	BookData | tuple | list | None	Optional recorded L2 order book replayed alongside the ticks. Pass a single BookData or a tuple/list of them; the symbol is read from each BookData.symbol (no {symbol: BookData} mapping). Enables DeepTrades and the L2 book metrics in a backtest (book_as_of is populated causally as the tick cursor advances, exactly as in replay). A sync preflight warns if the book is desynchronized from the trades.	None
sync_book	bool	When True and the preflight finds a recoverable clock offset between the book and the ticks, the book timestamps are shifted onto the trade clock. Default False (only warn on desync). An unrecoverable desync is never shifted.	False

Returns:

Name	Type	Description
BacktestEngine	'BacktestEngine'	Prepared engine instance

Note

align_by_ts controls how multi-symbol tick data is synchronized:

• False (default): Symbols aligned by matrix row index (fast path). All symbols must have identical number of rows. Each on_data() call processes the same row index across all symbols, regardless of their actual timestamps. Risk: introduces look-ahead bias if symbols have different tick frequencies or timestamps.

• True: Symbols aligned by real timestamp (merge path). Builds a unified timeline from all unique timestamps across symbols, sorted chronologically. Each symbol forward-fills to its most recent tick with ts <= current_ts. on_data() fires only when at least one symbol produces a new tick. Eliminates look-ahead bias for realistic multi-feed scenarios (e.g., BTC + ADA from independent sources). Slower than fast path due to np.searchsorted() lookups.

With a single symbol, this parameter has no effect (dispatcher always uses fast path).

Example

ticks = TickData('BTCUSDT', tick_matrix, tick_step=0.01) bt = BacktestEngine.by_ticks(MyStrategy(), data=(ticks,), sesh=sesh) bt.run()

Source code in src/tradetropy/backtest/engine.py

@classmethod
def by_ticks(
    cls,
    strategy: Strategy,
    data: Tuple[TickData, ...],
    sesh: "SeshSimulatorBase | None" = None,
    align_by_ts: bool = False,
    book: "BookData | tuple | list | None" = None,
    sync_book: bool = False,
) -> "BacktestEngine":
    '''
    Prepare engine for tick data.

    Data must be a tuple of TickData, one per symbol.
    The engine does NOT run here - call .run() to start the loop.

    Args:
        strategy (Strategy): Trading strategy instance
        data (tuple): Tuple of TickData objects
        sesh (SeshSimulatorBase): Optional session for testing
        align_by_ts (bool): Align symbols by timestamp instead of position
        book (BookData | tuple | list | None): Optional recorded L2 order
            book replayed alongside the ticks. Pass a single BookData or a
            tuple/list of them; the symbol is read from each
            ``BookData.symbol`` (no ``{symbol: BookData}`` mapping). Enables
            DeepTrades and the L2 book metrics in a backtest (book_as_of is
            populated causally as the tick cursor advances, exactly as in
            replay). A sync preflight warns if the book is desynchronized
            from the trades.
        sync_book (bool): When True and the preflight finds a recoverable
            clock offset between the book and the ticks, the book timestamps
            are shifted onto the trade clock. Default False (only warn on
            desync). An unrecoverable desync is never shifted.

    Returns:
        BacktestEngine: Prepared engine instance

    Note:
        align_by_ts controls how multi-symbol tick data is synchronized:

        - False (default): Symbols aligned by matrix row index (fast path).
          All symbols must have identical number of rows. Each on_data() call
          processes the same row index across all symbols, regardless of their
          actual timestamps. Risk: introduces look-ahead bias if symbols have
          different tick frequencies or timestamps.

        - True: Symbols aligned by real timestamp (merge path). Builds a
          unified timeline from all unique timestamps across symbols, sorted
          chronologically. Each symbol forward-fills to its most recent tick
          with ts <= current_ts. on_data() fires only when at least one symbol
          produces a new tick. Eliminates look-ahead bias for realistic
          multi-feed scenarios (e.g., BTC + ADA from independent sources).
          Slower than fast path due to np.searchsorted() lookups.

        With a single symbol, this parameter has no effect (dispatcher always
        uses fast path).

    Example:
        ticks = TickData('BTCUSDT', tick_matrix, tick_step=0.01)
        bt = BacktestEngine.by_ticks(MyStrategy(), data=(ticks,), sesh=sesh)
        bt.run()
    '''
    inputs, feed_type = _normalize_data(data)
    if feed_type != "tick":
        raise ConfigError(
            "by_ticks() received KlineData. Use by_klines() for candles."
        )

    if sesh is None:
        sesh = SeshSimulatorBase()

    engine = cls(strategy, sesh=sesh, _feed_type="tick")
    engine._tick_inputs = inputs
    engine._align_by_ts = align_by_ts
    engine._book_inputs = _normalize_book(book)
    engine._sync_book = bool(sync_book)

    for inp in inputs:
        sesh.add_symbol(inp.config)

    return engine

by_klines classmethod

by_klines(strategy: Strategy, data: Tuple[KlineData, ...], sesh: 'SeshSimulatorBase | None' = None, book: 'BookData | tuple | list | None' = None) -> 'BacktestEngine'

Prepare engine for OHLC candle data (klines).

Data must be a tuple of KlineData, one per symbol. Each KlineData must have interval_ms configured. subscribe_ticks() is not supported in this mode. The engine does NOT run here - call .run() to start the loop.

Parameters:

Name	Type	Description	Default
strategy	Strategy	Trading strategy instance	required
data	tuple	Tuple of KlineData objects	required
sesh	SeshSimulatorBase	Optional session for testing	None
book	BookData | tuple | list | None	Not supported in kline mode - the L2 order book needs per-trade timestamps for the causal book_as_of lookup. Passing a book raises ConfigError; use by_ticks(book=...).	None

Returns:

Name	Type	Description
BacktestEngine	'BacktestEngine'	Prepared engine instance

Note

If sesh is omitted, a SeshSimulatorBase() is built by default (balance 10,000, no commission). The internal broker's feed_type is inferred automatically from the engine mode.

Example

klines_btc = KlineData('BTCUSDT', btc_matrix, timeframe='5m') klines_eth = KlineData('ETHUSDT', eth_matrix, timeframe='1m') bt = BacktestEngine.by_klines( MyStrategy(), data=(klines_btc, klines_eth), sesh=sesh, ) bt.run()

Source code in src/tradetropy/backtest/engine.py

@classmethod
def by_klines(
    cls,
    strategy: Strategy,
    data: Tuple[KlineData, ...],
    sesh: "SeshSimulatorBase | None" = None,
    book: "BookData | tuple | list | None" = None,
) -> "BacktestEngine":
    '''
    Prepare engine for OHLC candle data (klines).

    Data must be a tuple of KlineData, one per symbol.
    Each KlineData must have interval_ms configured.
    subscribe_ticks() is not supported in this mode.
    The engine does NOT run here - call .run() to start the loop.

    Args:
        strategy (Strategy): Trading strategy instance
        data (tuple): Tuple of KlineData objects
        sesh (SeshSimulatorBase): Optional session for testing
        book (BookData | tuple | list | None): Not supported in kline mode -
            the L2 order book needs per-trade timestamps for the causal
            book_as_of lookup. Passing a book raises ConfigError; use
            by_ticks(book=...).

    Returns:
        BacktestEngine: Prepared engine instance

    Note:
        If sesh is omitted, a SeshSimulatorBase() is built by default
        (balance 10,000, no commission). The internal broker's feed_type
        is inferred automatically from the engine mode.

    Example:
        klines_btc = KlineData('BTCUSDT', btc_matrix, timeframe='5m')
        klines_eth = KlineData('ETHUSDT', eth_matrix, timeframe='1m')
        bt = BacktestEngine.by_klines(
            MyStrategy(),
            data=(klines_btc, klines_eth),
            sesh=sesh,
        )
        bt.run()
    '''
    inputs, feed_type = _normalize_data(data)
    if feed_type != "kline":
        raise ConfigError(
            "by_klines() received TickData. Use by_ticks() for ticks."
        )
    if book is not None:
        raise ConfigError(
            "by_klines() does not support book=. The L2 order book needs "
            "per-trade timestamps for the causal book_as_of lookup; use "
            "BacktestEngine.by_ticks(..., book=...) instead."
        )

    if sesh is None:
        sesh = SeshSimulatorBase()

    engine = cls(strategy, sesh=sesh, _feed_type="kline")
    engine._kline_inputs = inputs

    for inp in inputs:
        sesh.add_symbol(inp.config)

    return engine

run

run(params: dict = None, verbose: bool = False, save_log: 'bool | None' = None, stats_warn: bool = True) -> 'BacktestEngine'

Execute the complete backtest.

Call after by_ticks() or by_klines(). Returns self for optional chaining.

Parameters:

Name	Type	Description	Default
params	dict	Optional dict to override parameters before running	None
verbose	bool	Enable strategy logger output to console	False
save_log	bool	Write log to file. None uses engine default	None
stats_warn	bool	Emit the "Stats: insufficient sample" UserWarning when the sample (duration/closed trades) is too small to trust the annualized/trade-distribution metrics. The gating that zeroes those metrics to NaN and sets stats["_low_sample"] always applies; set stats_warn=False to silence just the warning for quick exploratory runs on small datasets.	True

Returns:

Name	Type	Description
BacktestEngine	'BacktestEngine'	Self for method chaining

Note

save_log=None uses _DEFAULT_SAVE_LOG (False in backtest). Requires verbose=True and log_file defined in strategy.

Example

result = BacktestEngine.by_ticks( strategy, data=(ticks,), sesh=sesh, ).run()

Source code in src/tradetropy/backtest/engine.py

def run(self, params: dict = None, verbose: bool = False,
        save_log: "bool | None" = None,
        stats_warn: bool = True) -> "BacktestEngine":
    '''
    Execute the complete backtest.

    Call after by_ticks() or by_klines(). Returns self for optional chaining.

    Args:
        params (dict): Optional dict to override parameters before running
        verbose (bool): Enable strategy logger output to console
        save_log (bool): Write log to file. None uses engine default
        stats_warn (bool): Emit the "Stats: insufficient sample" UserWarning
                          when the sample (duration/closed trades) is too
                          small to trust the annualized/trade-distribution
                          metrics. The gating that zeroes those metrics to
                          NaN and sets stats["_low_sample"] always applies;
                          set stats_warn=False to silence just the warning
                          for quick exploratory runs on small datasets.

    Returns:
        BacktestEngine: Self for method chaining

    Note:
        save_log=None uses _DEFAULT_SAVE_LOG (False in backtest).
        Requires verbose=True and log_file defined in strategy.

    Example:
        result = BacktestEngine.by_ticks(
            strategy, data=(ticks,), sesh=sesh,
        ).run()
    '''
    if save_log is None:
        save_log = self._DEFAULT_SAVE_LOG

    self._stats_warn = stats_warn

    if params is not None:
        self.strategy.update_params(params)

    if self._tick_inputs:
        data_dict = _inputs_to_dict(self._tick_inputs)
        self._run_ticks(data_dict, verbose=verbose, save_log=save_log)
    elif self._kline_inputs:
        data_dict = _inputs_to_dict(self._kline_inputs)
        self._run_klines(data_dict, verbose=verbose, save_log=save_log)
    elif self._feed_type == "tick":
        raise TradingError(
            "No data. Use BacktestEngine.by_ticks("
            "strategy, data=(TickData(...),), sesh=sesh) "
            "before calling run()."
        )
    else:
        raise TradingError(
            "No data. Use BacktestEngine.by_klines("
            "strategy, data=(KlineData(...),), sesh=sesh) "
            "before calling run()."
        )

    self._finalize_open_trades()
    self._stats = self._build_stats()
    return self

rerun

rerun(params: dict = None) -> 'BacktestEngine'

Create new engine with same data, run with new parameters.

Creates a new engine with the same data and a cloned session, runs the backtest with the provided parameters and returns the new instance.

Parameters:

Name	Type	Description	Default
params	dict	New parameters for the backtest	None

Returns:

Name	Type	Description
BacktestEngine	'BacktestEngine'	New engine with results

Example

bt_optimal = bt_original.rerun(params=result.best_params) bt_optimal.plot()

Source code in src/tradetropy/backtest/engine.py

def rerun(self, params: dict = None) -> "BacktestEngine":
    '''
    Create new engine with same data, run with new parameters.

    Creates a new engine with the same data and a cloned session, runs
    the backtest with the provided parameters and returns the new instance.

    Args:
        params (dict): New parameters for the backtest

    Returns:
        BacktestEngine: New engine with results

    Example:
        bt_optimal = bt_original.rerun(params=result.best_params)
        bt_optimal.plot()
    '''
    new_engine = self._clone_engine()
    if params is not None:
        new_engine.strategy.update_params(params)
    new_engine.run()
    return new_engine

plot

plot(theme: str = 'light', width: int = 1200, plot_trades: bool = True, plot_volume: bool = True, ohlc_style: str = 'candle', plot_drawdown: bool = False, plot_footprint: bool = True, plot_stats: bool = False, plot_pl: bool = False, pl_height: int = 100, ohlc_height: int = 410, equity_height: int = 110, drawdown_height: int = 80, indicator_height: int = 100, footprint_zoom_range: int = 40, labels_zoom_range: int = 100, max_candles: int = 10000, equity_mode: str = 'return', equity_unit: str = 'percent', output: str = 'show', filename: str = 'backtest.html', title: str = 'Backtest', resample_timeframe: str | None = None, align_trades_to_candle: bool = True, max_trailing_dd: float | None = None, output_backend: str = 'webgl', show_price_tag: bool = False, header_text: str | None = None, footer_text: str | None = None, superimpose: bool | str = False, derender_on_pan: bool = True) -> None

Generate interactive backtest chart.

Requires: pip install bokeh

Parameters:

Name	Type	Description	Default
theme	str	'light' or 'dark'	'light'
width	int	Total width in pixels	1200
plot_trades	bool	Show trade entry/exit lines	True
plot_volume	bool	Show volume bars	True
ohlc_style	str	Price panel style - 'candle' (Japanese candlesticks, default) or 'bar' (OHLC bars: High-Low vertical with open/close ticks)	'candle'
plot_drawdown	bool	Show drawdown panel	False
plot_footprint	bool	Show footprint (requires FpProxy)	True
plot_stats	bool	Show statistics bar	False
plot_pl	bool	Show P&L panel per trade	False
pl_height	int	P&L panel height in pixels	100
ohlc_height	int	Main OHLC panel height	410
equity_height	int	Equity/return panel height	110
drawdown_height	int	Drawdown panel height	80
indicator_height	int	Default height for indicator panels	100
footprint_zoom_range	int	Max candles to display footprint	40
labels_zoom_range	int	Max candles in view to display text labels	100
max_candles	int	Candle render limit (performance)	10000
equity_mode	str	'none' | 'balance' | 'return'	'return'
equity_unit	str	'currency' | 'percent'	'percent'
output	str	'notebook' | 'file' | 'show'	'show'
filename	str	HTML filename (only if output='file')	'backtest.html'
title	str	Browser tab / window title. Default 'Backtest'.	'Backtest'
resample_timeframe	str	Resample candles to given timeframe (e.g. '1m', '5m', '1h', '1d').	None
align_trades_to_candle	bool	Align trade timestamps to candle open	True
max_trailing_dd	float	Maximum trailing drawdown (e.g. 0.2 = 20%)	None
output_backend	str	'canvas' | 'webgl' | 'svg'. Default 'webgl' for smooth pan/zoom/crosshair interaction; use 'canvas' for deterministic headless PNG export and 'svg' for vector export.	'webgl'
show_price_tag	bool	DEPRECATED, removed. The floating price box has been removed from the chart entirely; this kwarg is still accepted for backwards compatibility and silently ignored, emitting a DeprecationWarning. Will be removed in a future version.	False
header_text	str	Optional manual note rendered above the whole chart (e.g. a title). Raw HTML, not escaped -- do not pass untrusted/external text. None (default) renders nothing.	None
footer_text	str	Optional manual note rendered below the whole chart (e.g. a disclaimer). Same HTML caveat as header_text. None (default) renders nothing.	None
superimpose	bool | str	Overlay translucent, wick-less higher-timeframe candle bodies behind the main OHLC. True auto-selects a higher TF from the chart's own interval; a string (e.g. '4h') sets it explicitly. False (default) disables it. See PlotConfig.superimpose for the full auto-selection rules.	False
derender_on_pan	bool	Temporarily hide heavy overlay glyphs (scatter markers, text labels) while dragging/wheel-zooming, restoring them once the interaction settles. Default True; set to False to disable the whole mechanism.	True

Example

bt.plot() bt.plot(theme='dark', plot_trades=False) bt.plot(theme='dark', width=1400, equity_mode='balance')

Source code in src/tradetropy/backtest/engine.py

def plot(
    self,
    theme: str = "light",
    width: int = 1200,
    plot_trades: bool = True,
    plot_volume: bool = True,
    ohlc_style: str = "candle",
    plot_drawdown: bool = False,
    plot_footprint: bool = True,
    plot_stats: bool = False,
    plot_pl: bool = False,
    pl_height: int = 100,
    ohlc_height: int = 410,
    equity_height: int = 110,
    drawdown_height: int = 80,
    indicator_height: int = 100,
    footprint_zoom_range: int = 40,
    labels_zoom_range: int = 100,
    max_candles: int = 10_000,
    equity_mode: str = "return",
    equity_unit: str = "percent",
    output: str = "show",
    filename: str = "backtest.html",
    title: str = "Backtest",
    resample_timeframe: str | None = None,
    align_trades_to_candle: bool = True,
    max_trailing_dd: float | None = None,
    output_backend: str = "webgl",
    show_price_tag: bool = False,
    header_text: str | None = None,
    footer_text: str | None = None,
    superimpose: bool | str = False,
    derender_on_pan: bool = True,
) -> None:
    '''
    Generate interactive backtest chart.

    Requires: pip install bokeh

    Args:
        theme (str): 'light' or 'dark'
        width (int): Total width in pixels
        plot_trades (bool): Show trade entry/exit lines
        plot_volume (bool): Show volume bars
        ohlc_style (str): Price panel style - 'candle' (Japanese
            candlesticks, default) or 'bar' (OHLC bars: High-Low vertical
            with open/close ticks)
        plot_drawdown (bool): Show drawdown panel
        plot_footprint (bool): Show footprint (requires FpProxy)
        plot_stats (bool): Show statistics bar
        plot_pl (bool): Show P&L panel per trade
        pl_height (int): P&L panel height in pixels
        ohlc_height (int): Main OHLC panel height
        equity_height (int): Equity/return panel height
        drawdown_height (int): Drawdown panel height
        indicator_height (int): Default height for indicator panels
        footprint_zoom_range (int): Max candles to display footprint
        labels_zoom_range (int): Max candles in view to display text labels
        max_candles (int): Candle render limit (performance)
        equity_mode (str): 'none' | 'balance' | 'return'
        equity_unit (str): 'currency' | 'percent'
        output (str): 'notebook' | 'file' | 'show'
        filename (str): HTML filename (only if output='file')
        title (str): Browser tab / window title. Default 'Backtest'.
        resample_timeframe (str): Resample candles to given timeframe
            (e.g. '1m', '5m', '1h', '1d').
        align_trades_to_candle (bool): Align trade timestamps to candle open
        max_trailing_dd (float): Maximum trailing drawdown (e.g. 0.2 = 20%)
        output_backend (str): 'canvas' | 'webgl' | 'svg'. Default 'webgl'
            for smooth pan/zoom/crosshair interaction; use 'canvas' for
            deterministic headless PNG export and 'svg' for vector export.
        show_price_tag (bool): DEPRECATED, removed. The floating price
            box has been removed from the chart entirely; this kwarg is
            still accepted for backwards compatibility and silently
            ignored, emitting a DeprecationWarning. Will be removed in a
            future version.
        header_text (str): Optional manual note rendered above the whole
            chart (e.g. a title). Raw HTML, not escaped -- do not pass
            untrusted/external text. None (default) renders nothing.
        footer_text (str): Optional manual note rendered below the whole
            chart (e.g. a disclaimer). Same HTML caveat as header_text.
            None (default) renders nothing.
        superimpose (bool | str): Overlay translucent, wick-less
            higher-timeframe candle bodies behind the main OHLC. True
            auto-selects a higher TF from the chart's own interval; a
            string (e.g. '4h') sets it explicitly. False (default)
            disables it. See PlotConfig.superimpose for the full
            auto-selection rules.
        derender_on_pan (bool): Temporarily hide heavy overlay glyphs
            (scatter markers, text labels) while dragging/wheel-zooming,
            restoring them once the interaction settles. Default True;
            set to False to disable the whole mechanism.

    Example:
        bt.plot()
        bt.plot(theme='dark', plot_trades=False)
        bt.plot(theme='dark', width=1400, equity_mode='balance')
    '''
    kwargs = dict(
        theme=theme,
        width=width,
        plot_trades=plot_trades,
        plot_volume=plot_volume,
        ohlc_style=ohlc_style,
        plot_drawdown=plot_drawdown,
        plot_footprint=plot_footprint,
        plot_stats=plot_stats,
        plot_pl=plot_pl,
        pl_height=pl_height,
        ohlc_height=ohlc_height,
        equity_height=equity_height,
        drawdown_height=drawdown_height,
        indicator_height=indicator_height,
        footprint_zoom_range=footprint_zoom_range,
        labels_zoom_range=labels_zoom_range,
        max_candles=max_candles,
        equity_mode=equity_mode,
        equity_unit=equity_unit,
        output=output,
        filename=filename,
        title=title,
        resample_timeframe=resample_timeframe,
        align_trades_to_candle=align_trades_to_candle,
        max_trailing_dd=max_trailing_dd,
        output_backend=output_backend,
        show_price_tag=show_price_tag,
        header_text=header_text,
        footer_text=footer_text,
        superimpose=superimpose,
        derender_on_pan=derender_on_pan,
    )
    self._plot_config = kwargs
    from tradetropy.plotting import plot as _plot, PlotConfig
    _plot(self, PlotConfig(**kwargs))

optimize

optimize(maximize=None, minimize=None, constraints=None, method='grid', iterations=100, workers=None, progress=True, **param_lists)

Optimize strategy parameters.

Each parameter is passed as an explicit list of values to explore:

bt.optimize(
    maximize = 'Sharpe Ratio',
    fast_ma = [10, 20, 30, 50],
    threshold = [0.1, 0.5, 1.0],
    mode = ['trend', 'mean_revert'],
)

Parameters:

Name	Type	Description	Default
maximize	str	Name of the metric to maximize	None
minimize	str	Name of the metric to minimize	None
constraints		Function (params: dict) -> bool for filtering	None
method	str	'grid' or 'random'	'grid'
iterations	int	Number of random samples (only with method='random')	100
workers	int	Parallel processes (None -> cpu_count)	None
progress	bool	Show a single aggregate progress bar counting completed backtests (default True). One bar for the whole optimization, not one per backtest.	True
**param_lists		Parameter name -> list of values to explore	{}

Returns:

Name	Type	Description
OptimizationResult		With best_params, best_fitness, best_stats, top(n), and to_dataframe() method

Source code in src/tradetropy/backtest/engine.py

def optimize(self, maximize=None, minimize=None, constraints=None,
             method="grid", iterations=100, workers=None, progress=True,
             **param_lists):
    '''
    Optimize strategy parameters.

    Each parameter is passed as an explicit list of values to explore:

        bt.optimize(
            maximize = 'Sharpe Ratio',
            fast_ma = [10, 20, 30, 50],
            threshold = [0.1, 0.5, 1.0],
            mode = ['trend', 'mean_revert'],
        )

    Args:
        maximize (str): Name of the metric to maximize
        minimize (str): Name of the metric to minimize
        constraints: Function (params: dict) -> bool for filtering
        method (str): 'grid' or 'random'
        iterations (int): Number of random samples (only with method='random')
        workers (int): Parallel processes (None -> cpu_count)
        progress (bool): Show a single aggregate progress bar counting
            completed backtests (default True). One bar for the whole
            optimization, not one per backtest.
        **param_lists: Parameter name -> list of values to explore

    Returns:
        OptimizationResult: With best_params, best_fitness, best_stats, top(n),
            and to_dataframe() method
    '''
    from tradetropy.optimize import (
        ParameterSpace,
        FitnessMetric,
        OptimizationResult,
        GridSearchOptimizer,
        RandomSearchOptimizer,
    )
    from tradetropy.optimize.task import _create_evaluation_function
    from tradetropy.backtest.pool_adapter import PoolEvaluator

    if maximize is None and minimize is None:
        raise ConfigError('Specify maximize or minimize.')
    if maximize is not None and minimize is not None:
        raise ConfigError('Specify only one: maximize or minimize.')
    metric = maximize or minimize
    is_max = maximize is not None

    space = ParameterSpace(**param_lists)
    if constraints is not None:
        space.add_constraint(constraints)

    fitness = FitnessMetric(metric=metric, maximize=is_max)
    evaluate_fn = _create_evaluation_function(_run_backtest_candidate, fitness)

    # Ship the input matrices to workers via shared_memory (allocated once
    # in the parent) instead of pickling a full copy into each worker. Only
    # tiny descriptors + metadata travel in the pickle stream. Mirrors
    # PoolBacktestEngine; workers only read the arrays.
    from tradetropy.backtest._shm_bundle import build_shm_bundle, release_shm
    data_bundle, shm_refs = build_shm_bundle(
        strategy_cls=type(self.strategy),
        sesh=self._sesh,
        tick_inputs=self._tick_inputs,
        kline_inputs=self._kline_inputs,
        align_by_ts=self._align_by_ts,
    )
    evaluator = PoolEvaluator(evaluate_fn, data_bundle, workers=workers,
                              progress=progress, desc="Optimize")

    if method == 'grid':
        optimizer = GridSearchOptimizer(space, fitness)
    elif method == 'random':
        optimizer = RandomSearchOptimizer(space, fitness, iterations=iterations)
    else:
        raise ConfigError(f'Unknown optimization method: {method!r}. Use grid or random.')

    try:
        optimizer.run(evaluator)
    finally:
        release_shm(shm_refs)
    return OptimizationResult(optimizer.results, maximize=is_max)

montecarlo

montecarlo(n_sims=1000, methods=None, metrics=None, confidence=(0.95, 0.99), seed=None, workers=None)

Run a Monte Carlo robustness test on the finished backtest.

Generates many randomized variants of the result to estimate the distribution of performance metrics, confidence intervals, the probability of loss, the risk of ruin and a composite robustness score.

Methods operate at one of three levels (they cannot be mixed across levels in a single call):

- Trade level (fast, no re-run): 'shuffle_order', 'resample_trades',
  'skip_trades', 'randomize_slippage', 'random_start_index'.
- Data level (re-runs the engine): 'randomize_prices',
  'random_start_bar'.
- Parameter level (re-runs the engine): 'randomize_parameters'
  (pass an instance with a search space).

Parameters:

Name	Type	Description	Default
n_sims	int	Number of simulations to run.	1000
methods	Sequence	Method identifiers (str) or MCMethod instances. Defaults to ['resample_trades'].	None
metrics	Sequence[str]	Stats keys to track. Defaults to a standard set (Return, Max Drawdown, Sharpe, Profit Factor, Win Rate).	None
confidence	Sequence[float]	Confidence levels in (0, 1).	(0.95, 0.99)
seed	int	Base seed for reproducibility.	None
workers	int	Parallel processes for the re-run path (None -> cpu_count). Ignored on the trade-level path.	None

Returns:

Name	Type	Description
MonteCarloResult		With summary(), to_dataframe(), percentile(), confidence_interval(), probability_of_loss, risk_of_ruin() and robustness_score.

Example

bt.run() mc = bt.montecarlo(n_sims=1000, methods=['shuffle_order'], seed=42) print(mc.summary()) print(mc.robustness_score)

Source code in src/tradetropy/backtest/engine.py

def montecarlo(self, n_sims=1000, methods=None, metrics=None,
               confidence=(0.95, 0.99), seed=None, workers=None):
    '''
    Run a Monte Carlo robustness test on the finished backtest.

    Generates many randomized variants of the result to estimate the
    distribution of performance metrics, confidence intervals, the
    probability of loss, the risk of ruin and a composite robustness score.

    Methods operate at one of three levels (they cannot be mixed across
    levels in a single call):

        - Trade level (fast, no re-run): 'shuffle_order', 'resample_trades',
          'skip_trades', 'randomize_slippage', 'random_start_index'.
        - Data level (re-runs the engine): 'randomize_prices',
          'random_start_bar'.
        - Parameter level (re-runs the engine): 'randomize_parameters'
          (pass an instance with a search space).

    Args:
        n_sims (int): Number of simulations to run.
        methods (Sequence): Method identifiers (str) or MCMethod instances.
            Defaults to ['resample_trades'].
        metrics (Sequence[str]): Stats keys to track. Defaults to a standard
            set (Return, Max Drawdown, Sharpe, Profit Factor, Win Rate).
        confidence (Sequence[float]): Confidence levels in (0, 1).
        seed (int): Base seed for reproducibility.
        workers (int): Parallel processes for the re-run path (None ->
            cpu_count). Ignored on the trade-level path.

    Returns:
        MonteCarloResult: With summary(), to_dataframe(), percentile(),
            confidence_interval(), probability_of_loss, risk_of_ruin() and
            robustness_score.

    Example:
        bt.run()
        mc = bt.montecarlo(n_sims=1000, methods=['shuffle_order'], seed=42)
        print(mc.summary())
        print(mc.robustness_score)
    '''
    from tradetropy.robustness import MonteCarlo, MonteCarloConfig

    if self._stats is None:
        raise TradingError(
            'Run the backtest before montecarlo(): no stats available.'
        )

    kwargs = dict(
        n_sims=n_sims,
        methods=methods if methods is not None else ['resample_trades'],
        confidence=confidence,
        seed=seed,
        workers=workers,
    )
    if metrics is not None:
        kwargs['metrics'] = metrics

    config = MonteCarloConfig(**kwargs)
    return MonteCarlo(self, config).run()

PoolBacktestEngine

PoolBacktestEngine

Runs multiple strategies in parallel with shared memory.

USAGE

With ticks [N x >=7]:

results = PoolBacktestEngine.by_ticks( strategies = [S1, S2, S3], data = (TickData("BTCUSDT", tick_matrix, tick_size=0.01),), workers = 8, )

With klines:

results = PoolBacktestEngine.by_klines( strategies = [S1, S2, S3], data = (KlineData("BTCUSDT", kline_matrix, timeframe=60_000),), workers = 8, )

The data type (TickData/KlineData) determines the mode -- it is never inferred from shape. Symbol config travels with the data.

Advantages

• SMA(20) on BTCUSDT shared across 400 strategies -> computed only once.
• Footprints pre-computed in bulk -> zero recalculation in workers.
• Works on Linux (fork), macOS and Windows (spawn).

by_ticks classmethod

by_ticks(strategies: list, data, workers: int | None = None, book: 'BookData | tuple | list | None' = None, sync_book: bool = False) -> list

Run strategies with tick data.

data : TickData or tuple/list of TickData (one per symbol). subscribe_ticks() and subscribe_footprint() are valid in this mode.

BookData or tuple/list of BookData with the recorded L2 book

to replay alongside ticks (enables DeepTrades / L2 metrics in the pool, same as in backtest). The symbol is taken from each BookData.symbol (no {symbol: BookData} map). A sync preflight warns if the book is out of sync with the trades.

sync_book : if True and the preflight finds a recoverable clock offset, shifts the book to the trades' clock. Default False (warn only).

Source code in src/tradetropy/backtest/pool.py

@classmethod
def by_ticks(
    cls,
    strategies: list,
    data,
    workers: int | None = None,
    book: "BookData | tuple | list | None" = None,
    sync_book: bool = False,
) -> list:
    """
    Run strategies with tick data.

    data : TickData or tuple/list of TickData (one per symbol).
    subscribe_ticks() and subscribe_footprint() are valid in this mode.

    book : BookData or tuple/list of BookData with the recorded L2 book
        to replay alongside ticks (enables DeepTrades / L2 metrics in the
        pool, same as in backtest). The symbol is taken from each
        ``BookData.symbol`` (no ``{symbol: BookData}`` map). A sync
        preflight warns if the book is out of sync with the trades.
    sync_book : if True and the preflight finds a recoverable clock offset,
        shifts the book to the trades' clock. Default False (warn only).
    """
    inputs, feed_type = _normalize_data(data)
    if feed_type != "tick":
        raise ConfigError(
            "PoolBacktestEngine.by_ticks() received KlineData. "
            "Use by_klines() for bars."
        )
    data_dict = {inp.symbol: inp.data for inp in inputs}
    return cls._run(
        strategies, data_dict, workers, feed_type="tick",
        book=_normalize_book(book), sync_book=sync_book,
    )

by_klines classmethod

by_klines(strategies: list, data, workers: int | None = None, book: 'BookData | tuple | list | None' = None) -> list

Run strategies with OHLC bar data.

data : KlineData or tuple/list of KlineData (each with interval_ms). subscribe_ticks() and subscribe_footprint() are not supported.

Source code in src/tradetropy/backtest/pool.py

@classmethod
def by_klines(
    cls,
    strategies: list,
    data,
    workers: int | None = None,
    book: "BookData | tuple | list | None" = None,
) -> list:
    """
    Run strategies with OHLC bar data.

    data : KlineData or tuple/list of KlineData (each with interval_ms).
    subscribe_ticks() and subscribe_footprint() are not supported.
    """
    inputs, feed_type = _normalize_data(data)
    if feed_type != "kline":
        raise ConfigError(
            "PoolBacktestEngine.by_klines() received TickData. "
            "Use by_ticks() for ticks."
        )
    if book is not None:
        raise ConfigError(
            "PoolBacktestEngine.by_klines() does not support book=. The L2 "
            "book needs per-trade timestamps for causal book_as_of; use "
            "by_ticks(..., book=...)."
        )
    data_dict = {inp.symbol: inp.data for inp in inputs}
    return cls._run(strategies, data_dict, workers, feed_type="kline")

run classmethod

run(strategies: list, data: 'SymbolInput | tuple[SymbolInput, ...] | list[SymbolInput]', workers: int | None = None, save_log: 'bool | None' = None, book: 'BookData | tuple | list | None' = None, sync_book: bool = False) -> list

Unified entry point for PoolBacktestEngine.

.. note:: save_log is accepted for API consistency with other engines, but is a no-op in pool/optimize: each worker uses NullLogger (zero I/O), so nothing is ever written to a file.

Accepts two forms:

Typed API (v2.2+, preferred): data is TickData, KlineData, or a list of them. The mode is inferred from the type -- there is no mode parameter. Symbol config travels with the data; no need to add it manually.

Example::

    PoolBacktestEngine.run(
        strategies = [S1, S2],
        data       = TickData("BTCUSDT", tick_matrix,
                              tick_step=0.25, tick_value=1.25),
        workers    = 4,
    )

    PoolBacktestEngine.run(
        strategies = [S1, S2],
        data       = [
            TickData("BTCUSDT", btc_ticks),
            TickData("ETHUSDT", eth_ticks),
        ],
    )

Legacy API (dict, compatible with v2.0/v2.1): data is a dict {symbol: ndarray}. The mode is inferred from shape[1] as in earlier versions. Kept for backwards compatibility.

Example::

    PoolBacktestEngine.run(
        strategies = [S1, S2],
        data       = {"BTCUSDT": tick_matrix},   # [N x >=7]
        workers    = 4,
    )

Source code in src/tradetropy/backtest/pool.py

@classmethod
def run(
    cls,
    strategies: list,
    data: "SymbolInput | tuple[SymbolInput, ...] | list[SymbolInput]",
    workers: int | None = None,
    save_log: "bool | None" = None,
    book: "BookData | tuple | list | None" = None,
    sync_book: bool = False,
) -> list:
    """
    Unified entry point for PoolBacktestEngine.

    .. note::
        ``save_log`` is accepted for API consistency with other
        engines, but is a **no-op** in pool/optimize: each worker uses
        ``NullLogger`` (zero I/O), so nothing is ever written to a file.

    Accepts two forms:

    **Typed API (v2.2+, preferred):**
        data is ``TickData``, ``KlineData``, or a list of them.
        The mode is inferred from the type -- there is no ``mode`` parameter.
        Symbol config travels with the data; no need to add it manually.

        Example::

            PoolBacktestEngine.run(
                strategies = [S1, S2],
                data       = TickData("BTCUSDT", tick_matrix,
                                      tick_step=0.25, tick_value=1.25),
                workers    = 4,
            )

            PoolBacktestEngine.run(
                strategies = [S1, S2],
                data       = [
                    TickData("BTCUSDT", btc_ticks),
                    TickData("ETHUSDT", eth_ticks),
                ],
            )

    **Legacy API (dict, compatible with v2.0/v2.1):**
        data is a dict ``{symbol: ndarray}``. The mode is inferred from
        ``shape[1]`` as in earlier versions. Kept for backwards
        compatibility.

        Example::

            PoolBacktestEngine.run(
                strategies = [S1, S2],
                data       = {"BTCUSDT": tick_matrix},   # [N x >=7]
                workers    = 4,
            )
    """
    if not strategies:
        return []
    inputs, feed_type = _normalize_data(data)
    data_dict = {inp.symbol: inp.data for inp in inputs}
    return cls._run(
        strategies, data_dict, workers, feed_type=feed_type,
        book=_normalize_book(book), sync_book=sync_book,
    )

LiveEngine

LiveEngine

LiveEngine(strategy: Strategy, sesh=None, _feed_type: Literal['tick', 'kline'] = 'tick', _poll_interval: float = 1.0, chart_ohlc_interval_ms: int = 60000, require_warmup: bool = True)

Engine for live trading.

Handles automatic warmup (historical data loading), indicator initialization, and real-time feed loop for tick or candle-based strategies.

Preferred constructors: - LiveEngine.by_ticks(strategy, sesh=sesh) - warmup in ticks - LiveEngine.by_klines(strategy, sesh=sesh) - warmup in candles

Automatic Warmup: - by_ticks: Loads strategy.warmup ticks (with hybrid historical candle support) - by_klines: Loads strategy.warmup candles

Typical Usage

engine = LiveEngine.by_ticks(MyStrategy(), sesh=sesh) engine.run()

With chart (automatic wiring):

engine.run(live_chart=chart)

With chart (manual, fine control):

engine.prepare() engine.attach_chart(chart) chart.start() engine.run()

Source code in src/tradetropy/live/engine.py

def __init__(
    self,
    strategy: Strategy,
    sesh=None,
    _feed_type: Literal["tick", "kline"] = "tick",
    _poll_interval: float = 1.0,
    chart_ohlc_interval_ms: int = 60_000,
    require_warmup: bool = True,
):
    self.strategy = strategy
    self._sesh = sesh
    self._feed_type: FeedType = _feed_type
    # When False, a broker returning zero historical ticks/klines is
    # tolerated (warning instead of DataError): the engine starts with
    # empty rings and warms up from the live feed itself. Used by Recorder,
    # where "no history yet" (new listing, quiet market) is expected and
    # recording should still start. Strategies keep the strict default
    # (True) so they never run silently blind on missing data.
    self._require_warmup: bool = bool(require_warmup)

    # Interval (ms) of the chart-only OHLC proxy auto-injected for tick
    # strategies that declare no subscribe_ohlc(). Default 1 minute.
    self._chart_ohlc_interval_ms: int = int(chart_ohlc_interval_ms)
    # OHLC proxies created by the engine purely so the chart has a candle
    # backbone (NOT declared by the strategy). They are excluded from the
    # REST top-up so a headless tick strategy triggers no network fetch.
    self._auto_ohlc_proxies: list[OhlcProxy] = []
    # If the session is simulated and was created in auto mode (feed_type=None),
    # builds its internal broker according to the engine mode. For real live
    # sessions (without _bind_feed_type) this is a no-op.
    if sesh is not None and hasattr(sesh, "_bind_feed_type"):
        sesh._bind_feed_type(_feed_type)
    self._poll_interval: float = _poll_interval

    self._tick_rings: dict[str, LiveRingBuffer] = {}
    self._ohlc_rings: dict[str, list[LiveOhlcRing]] = {}
    self._fp_rings: dict[str, list[LiveFpRing]] = {}
    self._book_rings: dict[str, list] = {}
    self._mbo_rings: dict[str, list] = {}
    self._ind_state: list[dict] = []

    self._loop_thread: Optional[threading.Thread] = None
    self._stop_event: threading.Event = threading.Event()
    self._preparado: bool = False

    # Rebuild gating (used by ReplayEngine to rewind a replay in place).
    # When _suppress_on_data is True, _process_tick/_process_bar update the
    # rings/indicators/broker but skip strategy.on_data() (warmup re-feed).
    # When _suppress_chart is True, the chart is not notified per tick during
    # a bulk re-feed; a single resync is issued at the end instead.
    self._suppress_on_data: bool = False
    self._suppress_chart: bool = False

    self._historical_candles: int = 0

    self._chart: "LiveChart | None" = None

    from tradetropy.session.base import SeshSimulatorBase
    self._is_simulated: bool = isinstance(sesh, SeshSimulatorBase)

sesh property

sesh

Get the session (broker) connected to the engine.

Returns:

Name	Type	Description
Session		The broker instance.

feed_type property

feed_type: FeedType

Get the engine feed type.

Returns:

Name	Type	Description
FeedType	FeedType	Either 'tick' or 'kline'.

by_ticks classmethod

by_ticks(strategy: Strategy, sesh=None, poll_interval: float = 1.0, chart_ohlc_interval_ms: int = 60000, require_warmup: bool = True) -> 'LiveEngine'

Create LiveEngine with tick-based warmup.

Parameters:

Name	Type	Description	Default
strategy	Strategy	Strategy instance (must have warmup attribute).	required
sesh		Optional session (broker) instance.	None
poll_interval	float	Polling interval in seconds.	1.0
chart_ohlc_interval_ms	int	Interval (ms) of the chart-only OHLC proxy auto-injected when the strategy subscribes ticks but no OHLC. Default 60_000 (1 minute). Used only for the chart; never fed to on_data().	60000
require_warmup	bool	If False, a broker returning zero historical ticks or klines is tolerated (warning instead of DataError) and the engine starts with empty rings. Default True (strict).	True

Returns:

Name	Type	Description
LiveEngine	'LiveEngine'	Engine configured for tick-based operation.

Source code in src/tradetropy/live/engine.py

@classmethod
def by_ticks(
    cls,
    strategy: Strategy,
    sesh=None,
    poll_interval: float = 1.0,
    chart_ohlc_interval_ms: int = 60_000,
    require_warmup: bool = True,
) -> "LiveEngine":
    """
    Create LiveEngine with tick-based warmup.

    Args:
        strategy: Strategy instance (must have warmup attribute).
        sesh: Optional session (broker) instance.
        poll_interval: Polling interval in seconds.
        chart_ohlc_interval_ms: Interval (ms) of the chart-only OHLC proxy
            auto-injected when the strategy subscribes ticks but no OHLC.
            Default 60_000 (1 minute). Used only for the chart; never fed
            to on_data().
        require_warmup: If False, a broker returning zero historical ticks
            or klines is tolerated (warning instead of DataError) and the
            engine starts with empty rings. Default True (strict).

    Returns:
        LiveEngine: Engine configured for tick-based operation.
    """
    return cls(
        strategy,
        sesh=sesh,
        _feed_type="tick",
        _poll_interval=poll_interval,
        chart_ohlc_interval_ms=chart_ohlc_interval_ms,
        require_warmup=require_warmup,
    )

by_klines classmethod

by_klines(strategy: Strategy, sesh=None, poll_interval: float = 1.0, require_warmup: bool = True) -> 'LiveEngine'

Create LiveEngine with candle-based warmup.

Parameters:

Name	Type	Description	Default
strategy	Strategy	Strategy instance (must have warmup attribute).	required
sesh		Optional session (broker) instance.	None
poll_interval	float	Polling interval in seconds.	1.0
require_warmup	bool	If False, a broker returning zero historical klines is tolerated (warning instead of DataError) and the engine starts with empty rings. Default True (strict).	True

Returns:

Name	Type	Description
LiveEngine	'LiveEngine'	Engine configured for candle-based operation.

Source code in src/tradetropy/live/engine.py

@classmethod
def by_klines(
    cls,
    strategy: Strategy,
    sesh=None,
    poll_interval: float = 1.0,
    require_warmup: bool = True,
) -> "LiveEngine":
    """
    Create LiveEngine with candle-based warmup.

    Args:
        strategy: Strategy instance (must have warmup attribute).
        sesh: Optional session (broker) instance.
        poll_interval: Polling interval in seconds.
        require_warmup: If False, a broker returning zero historical klines
            is tolerated (warning instead of DataError) and the engine
            starts with empty rings. Default True (strict).

    Returns:
        LiveEngine: Engine configured for candle-based operation.
    """
    return cls(
        strategy, sesh=sesh, _feed_type="kline",
        _poll_interval=poll_interval, require_warmup=require_warmup,
    )

attach_chart

attach_chart(chart) -> None

Connect a LiveChart to the engine for real-time visualization.

Parameters:

Name	Type	Description	Default
chart		LiveChart instance to attach.	required

Raises:

Type	Description
TradingError	If chart is not a LiveChart instance.

Source code in src/tradetropy/live/engine.py

def attach_chart(self, chart) -> None:
    """
    Connect a LiveChart to the engine for real-time visualization.

    Args:
        chart: LiveChart instance to attach.

    Raises:
        TradingError: If chart is not a LiveChart instance.
    """
    from tradetropy.plotting.live.chart import LiveChart as _LiveChart

    if not isinstance(chart, _LiveChart):
        raise TradingError(
            "attach_chart() requires a LiveChart instance. "
            f"Received: {type(chart).__name__}"
        )

    self._chart          = chart
    chart._engine        = self
    chart._strategy      = self.strategy
    _inner_broker = getattr(self.sesh, "_broker", None)
    chart._broker = _inner_broker if _inner_broker is not None else self.sesh

prepare

prepare(historico: 'dict | None' = None) -> None

Prepares the engine: fetches historical data, builds rings, warms indicators and calls strategy.init().

Parameters:

Name	Type	Description	Default
historico	'dict | None'	optional dict with preloaded data (for tests). Format: {symbol: tick_array} and/or {(symbol, interval_ms): kline_array}.	None

Source code in src/tradetropy/live/engine.py

def prepare(self, historico: "dict | None" = None) -> None:
    """
    Prepares the engine: fetches historical data, builds rings, warms
    indicators and calls strategy.init().

    Args:
        historico: optional dict with preloaded data (for tests).
                   Format: {symbol: tick_array} and/or
                           {(symbol, interval_ms): kline_array}.
    """
    if self._feed_type == "kline" and self.strategy._tick_proxies:
        syms = [tp.symbol for tp in self.strategy._tick_proxies]
        raise TradingError(
            f"LiveEngine.by_klines() does not support subscribe_ticks(). "
            f"Problematic symbols: {syms}."
        )

    self.strategy._sesh = self.sesh
    self.strategy._feed_type = self._feed_type
    self.strategy._set_run_mode("live")
    self.strategy._save_log = getattr(self, "_save_log", self._DEFAULT_SAVE_LOG)

    self.strategy.init()

    self._maybe_inject_chart_ohlc()

    if historico is not None:
        hist = historico
    elif self.sesh is not None:
        hist = _fetch_auto_history(self)
    else:
        hist = {}

    _build_rings(self, hist)
    _warm(self, hist)

    if self._feed_type == "tick" and historico is None and self.sesh is not None:
        self._topup_ohlc_rings()

    _build_fp_rings(self, hist)
    _build_pattern_stores_live(self, hist)
    _build_book_rings(self, hist)
    _build_mbo_rings(self, hist)
    self._historical_candles = _count_closed_candles_rings(self)

    # Print warmup only for pure LiveEngine. ReplayEngine already prints its
    # own real warmup in _calc_warmup; the one here would be a reconstructed
    # value (misleading) because it does not know ticks_per_candle.
    if type(self).__name__ == "LiveEngine":
        from tradetropy.models._warmup_policy import resolve_warmup, log_warmup
        _wu = resolve_warmup(self.strategy, feed_type=self._feed_type)
        log_warmup(self.strategy, self._feed_type, _wu, "LiveEngine")

    self._preparado = True

run

run(params: dict = None, verbose: bool = False, blocking: bool = True, live_chart=None, save_log: 'bool | None' = None)

Start the feed loop. Returns self for chaining.

Parameters:

Name	Type	Description	Default
params	dict	Optional dict to override strategy parameters before starting (equivalent to BacktestEngine.run).	None
verbose	bool	Enable diagnostic prints.	False
blocking	bool	True -> blocks current thread. False -> launches daemon thread.	True
live_chart		Optional LiveChart. If passed, engine connects and starts it automatically (equivalent to: prepare() -> attach_chart() -> start() -> run()). Done in correct order, so it's sufficient to call: engine.run(live_chart=chart)	None
save_log	'bool | None'	Write strategy log to file (strategy.log_file). None -> uses engine default (_DEFAULT_SAVE_LOG = True in live, False in replay). Requires log_file to be defined.	None

Returns:

Name	Type	Description
self		Engine instance for chaining.

Raises:

Type	Description
TradingError	If both positional and live_chart= LiveChart passed.

Source code in src/tradetropy/live/engine.py

def run(
    self,
    params: dict = None,
    verbose: bool = False,
    blocking: bool = True,
    live_chart=None,
    save_log: "bool | None" = None,
):
    """
    Start the feed loop. Returns self for chaining.

    Args:
        params: Optional dict to override strategy parameters before starting
                (equivalent to BacktestEngine.run).
        verbose: Enable diagnostic prints.
        blocking: True -> blocks current thread. False -> launches daemon thread.
        live_chart: Optional LiveChart. If passed, engine connects and starts
                    it automatically (equivalent to: prepare() -> attach_chart()
                    -> start() -> run()). Done in correct order, so it's
                    sufficient to call: engine.run(live_chart=chart)
        save_log: Write strategy log to file (strategy.log_file). None -> uses
                  engine default (_DEFAULT_SAVE_LOG = True in live, False in
                  replay). Requires log_file to be defined.

    Returns:
        self: Engine instance for chaining.

    Raises:
        TradingError: If both positional and live_chart= LiveChart passed.
    """
    from tradetropy.plotting.live.chart import LiveChart as _LiveChart
    if isinstance(params, _LiveChart):
        if live_chart is not None:
            raise TradingError('run(): passed a positional LiveChart and also live_chart=. Use only one.')
        live_chart, params = params, None
    if params is not None:
        self.strategy.update_params(params)
    self._verbose = verbose

    if save_log is None:
        save_log = self._DEFAULT_SAVE_LOG
    # Store so that prepare() can pick it up; _set_save_log invalidates the
    # lazy logger if prepare() already ran (init() may have created it).
    self._save_log = save_log
    self.strategy._set_save_log(save_log)

    if not self._preparado:
        self.prepare()

    # Automatic chart wiring: attach + start in the correct order,
    # right after prepare() (which creates the proxies the chart validates)
    # and before the loop (the Bokeh server must be ready before data).
    if live_chart is not None:
        self.attach_chart(live_chart)
        live_chart.start()

    # Hook for subclasses that need to start something between chart.start()
    # and the loop (e.g. ReplayEngine starts its ReplayController here).
    # No-op by default in LiveEngine.
    self._before_loop()

    self._stop_event.clear()

    if blocking:
        self._run_guarded()
    else:
        self._loop_thread = threading.Thread(
            target=self._run_guarded,
            daemon=True,
            name="LiveEngine-loop",
        )
        self._loop_thread.start()

    return self

stop

stop()

Stop the run() loop cleanly.

In live mode, flushes any pending tick/OHLC records to disk before stopping the loop thread.

Source code in src/tradetropy/live/engine.py

def stop(self):
    """
    Stop the run() loop cleanly.

    In live mode, flushes any pending tick/OHLC records to disk before
    stopping the loop thread.
    """
    if self._stop_event.is_set():
        return
    self._stop_event.set()
    if not self._is_simulated:
        from tradetropy.io.io import _consolidate_npz_record
        for tp in self.strategy._tick_proxies:
            if tp._record_config is not None:
                self._flush_tick_proxy(tp)
                _consolidate_npz_record(tp._record_config.path)
        for op in self.strategy._ohlc_proxies:
            if op._record_config is not None:
                self._flush_ohlc_proxy(op)
                _consolidate_npz_record(op._record_config.path)
        for bp in self.strategy._book_proxies:
            if bp._record_config is not None:
                self._flush_book_proxy(bp)
                _consolidate_npz_record(bp._record_config.path)
        for mp in self.strategy._mbo_proxies:
            if mp._record_config is not None:
                self._flush_mbo_proxy(mp)
                _consolidate_npz_record(mp._record_config.path)
    if self._loop_thread is not None and self._loop_thread.is_alive():
        self._loop_thread.join(timeout=10)

on_tick

on_tick(symbol: str, tick: ndarray)

Process a tick in tick mode (manual feed).

Parameters:

Name	Type	Description	Default
symbol	str	Trading symbol.	required
tick	ndarray	Tick array [ts_ms, bid, ask, volume, flags, volume_real, price].	required

Source code in src/tradetropy/live/engine.py

def on_tick(self, symbol: str, tick: np.ndarray):
    """
    Process a tick in tick mode (manual feed).

    Args:
        symbol: Trading symbol.
        tick: Tick array [ts_ms, bid, ask, volume, flags, volume_real, price].
    """
    _process_tick(self, symbol, tick)

on_kline

on_kline(symbol: str, kline: ndarray)

Process a candle (kline) in kline mode (manual feed).

Parameters:

Name	Type	Description	Default
symbol	str	Trading symbol.	required
kline	ndarray	Kline array [ts_ms, open, high, low, close, volume, turnover].	required

Source code in src/tradetropy/live/engine.py

def on_kline(self, symbol: str, kline: np.ndarray):
    """
    Process a candle (kline) in kline mode (manual feed).

    Args:
        symbol: Trading symbol.
        kline: Kline array [ts_ms, open, high, low, close, volume, turnover].
    """
    _process_bar(self, symbol, kline)

ReplayEngine

ReplayEngine

ReplayEngine(strategy: Strategy, data, sesh: '_ReplaySesh | None' = None, history: 'dict | None' = None, warmup_ticks: 'int | None' = None, warmup_pct: float = 0.2, _feed_type: str = 'tick', _poll_interval: float = 0.0, speed: float = 1.0, base_rate: float = 1.0, book: 'BookData | tuple | list | None' = None, sync_book: bool = False, chart_ohlc_interval_ms: int = 60000)

Bases: BaseEngine

Automated, forward-only replay engine over historical data.

Specialization of BaseEngine for auditing a programmatic strategy: it keeps backward navigation disabled (_ALLOW_BACKWARD = False) so the Step Backward (« ) control is hidden, matching PaperEngine. Use Restart to rewind to the beginning and replay from scratch. Replay does not persist the log by default (_DEFAULT_SAVE_LOG = False), same as backtest.

Quick mode (data + broker kwargs -> automatic ReplaySesh): engine = ReplayEngine.by_ticks(strategy, data=data, commission=1.5)

Full control mode (pre-configured ReplaySesh): sesh = ReplaySesh(feed_type='tick', commission=1.5) engine = ReplayEngine.by_ticks(strategy, data=data, sesh=sesh)

Source code in src/tradetropy/playback/base.py

def __init__(
    self,
    strategy: Strategy,
    data,
    sesh: "_ReplaySesh | None" = None,
    history: "dict | None" = None,
    warmup_ticks: "int | None" = None,
    warmup_pct: float = 0.20,
    _feed_type: str = "tick",
    _poll_interval: float = 0.0,
    speed: float = 1.0,
    base_rate: float = 1.0,
    book: "BookData | tuple | list | None" = None,
    sync_book: bool = False,
    chart_ohlc_interval_ms: int = 60_000,
):
    if data is None:
        raise ConfigError(f"{type(self).__name__} requires `data`.")

    # Lazy imports break the import cycle (replay package imports BaseEngine).
    from tradetropy.replay.replay_sesh import ReplaySesh as _ReplaySesh
    from tradetropy.replay.controller import PlaybackController as _PlaybackController

    # Normalize typed data (TickData/KlineData) -> tuple + feed_type.
    inputs, data_feed_type = _normalize_data(data)
    if data_feed_type != _feed_type:
        method = "by_ticks" if _feed_type == "tick" else "by_klines"
        expected = "TickData" if _feed_type == "tick" else "KlineData"
        raise ConfigError(
            f"{type(self).__name__}.{method}() expected {expected}, "
            f"received data of type {data_feed_type!r}."
        )

    # Internal dict {symbol: ndarray} consumed by _calc_warmup/_bind_data.
    data_dict = {inp.symbol: inp.data for inp in inputs}

    # Normalize the recorded book: single BookData or tuple/list of them,
    # keyed by each BookData.symbol (symmetric with `data`).
    book = _normalize_book(book)

    # Resolve sesh
    if sesh is None:
        sesh = _ReplaySesh()
    elif not isinstance(sesh, _ReplaySesh):
        raise ConfigError(
            f"`sesh` must be a ReplaySesh, received {type(sesh).__name__}. "
            "Use ReplaySesh(commission=...) to create one."
        )

    # Register SymbolConfig for each input (same as BacktestEngine).
    # add_symbol queues if the broker is not yet built; it is flushed when
    # the feed_type is bound in super().__init__.
    for inp in inputs:
        sesh.add_symbol(inp.config)

    # Calculate warmup_ticks if not passed explicitly
    if warmup_ticks is None:
        warmup_ticks = self._calc_warmup(
            strategy, data_dict, _feed_type, warmup_pct, sesh,
        )

    # Bind data to the sesh
    sesh._bind_data(data_dict, history=history, warmup_ticks=warmup_ticks)

    super().__init__(
        strategy       = strategy,
        sesh           = sesh,
        _feed_type     = _feed_type,
        _poll_interval = _poll_interval,
        chart_ohlc_interval_ms = chart_ohlc_interval_ms,
    )

    self._ctrl = _PlaybackController(
        sesh, speed=speed, base_rate=base_rate,
        allow_backward=self._ALLOW_BACKWARD,
    )
    # Let the controller drive in-place rewinds (restart / step back).
    self._ctrl._engine = self

    # Serializes the feed loop against an in-place rewind so rings are never
    # mutated by both at the same time.
    self._rebuild_lock = threading.RLock()
    # Per-symbol cursor tracking for the tick loop's drain. Stored on the
    # instance (not as loop locals) so a rewind can reset them coherently.
    self._replay_ultimo_idx: dict[str, int] = {}
    self._replay_ultimo_ts:  dict[str, int] = {}

    # Thread-safe command queue: UI callbacks (Bokeh IOLoop thread) enqueue
    # callables that mutate broker/ring state; they are drained and executed
    # ON THE ENGINE THREAD at the top of each feed iteration, preserving the
    # single-writer model (only the engine thread mutates engine state).
    self._cmd_queue: "deque" = deque()
    self._cmd_lock = threading.Lock()

    # Recorded L2 order book to replay alongside the ticks (optional). Each
    # recorded row is a full top-K image; the shared BookReplayer applies it
    # as a snapshot to the book ring when the tick cursor reaches its
    # timestamp, so book_as_of / DeepTrades behave identically to the
    # live/streaming replay path and to the tick backtest.
    from tradetropy.data._book_replay import BookReplayer, resolve_book_sync
    self._book_replayer = BookReplayer(book or {})

    # Sync preflight: measure how the recorded book aligns with the ticks,
    # warn on desync, and (only when recoverable and sync_book=True) shift
    # the book clock onto the trade clock. Tick mode only - klines have no
    # per-trade timestamp for book_as_of.
    if book and _feed_type == "tick":
        from tradetropy.core.constants import _TICK_COL as _TC
        trades = {
            sym: (arr[:, _TC["ts"]], arr[:, _TC["price"]])
            for sym, arr in data_dict.items()
        }
        reports = resolve_book_sync(
            book, trades, sync_book=sync_book,
            engine_label=type(self).__name__,
        )
        for sym, (_rep, off) in reports.items():
            if off:
                self._book_replayer.set_offset(sym, off)
        self._book_sync_reports = {s: r for s, (r, _o) in reports.items()}
    else:
        self._book_sync_reports = {}

PaperEngine

PaperEngine

PaperEngine(strategy: Strategy, data, sesh: '_ReplaySesh | None' = None, history: 'dict | None' = None, warmup_ticks: 'int | None' = None, warmup_pct: float = 0.2, _feed_type: str = 'tick', _poll_interval: float = 0.0, speed: float = 1.0, base_rate: float = 1.0, book: 'BookData | tuple | list | None' = None, sync_book: bool = False, chart_ohlc_interval_ms: int = 60000)

Bases: BaseEngine

Manual, forward-only paper-trading engine driven by the UI Order Ticket.

Specialization of BaseEngine that forbids backward navigation (_ALLOW_BACKWARD = False) so the user cannot peek ahead or rewind single trades; the only reset is Restart (wipes the ledger, restores the baseline balance and seeks to index 0). It synthesizes its strategy from a PaperConfig, so no programmatic strategy is supplied.

Source code in src/tradetropy/playback/base.py

def __init__(
    self,
    strategy: Strategy,
    data,
    sesh: "_ReplaySesh | None" = None,
    history: "dict | None" = None,
    warmup_ticks: "int | None" = None,
    warmup_pct: float = 0.20,
    _feed_type: str = "tick",
    _poll_interval: float = 0.0,
    speed: float = 1.0,
    base_rate: float = 1.0,
    book: "BookData | tuple | list | None" = None,
    sync_book: bool = False,
    chart_ohlc_interval_ms: int = 60_000,
):
    if data is None:
        raise ConfigError(f"{type(self).__name__} requires `data`.")

    # Lazy imports break the import cycle (replay package imports BaseEngine).
    from tradetropy.replay.replay_sesh import ReplaySesh as _ReplaySesh
    from tradetropy.replay.controller import PlaybackController as _PlaybackController

    # Normalize typed data (TickData/KlineData) -> tuple + feed_type.
    inputs, data_feed_type = _normalize_data(data)
    if data_feed_type != _feed_type:
        method = "by_ticks" if _feed_type == "tick" else "by_klines"
        expected = "TickData" if _feed_type == "tick" else "KlineData"
        raise ConfigError(
            f"{type(self).__name__}.{method}() expected {expected}, "
            f"received data of type {data_feed_type!r}."
        )

    # Internal dict {symbol: ndarray} consumed by _calc_warmup/_bind_data.
    data_dict = {inp.symbol: inp.data for inp in inputs}

    # Normalize the recorded book: single BookData or tuple/list of them,
    # keyed by each BookData.symbol (symmetric with `data`).
    book = _normalize_book(book)

    # Resolve sesh
    if sesh is None:
        sesh = _ReplaySesh()
    elif not isinstance(sesh, _ReplaySesh):
        raise ConfigError(
            f"`sesh` must be a ReplaySesh, received {type(sesh).__name__}. "
            "Use ReplaySesh(commission=...) to create one."
        )

    # Register SymbolConfig for each input (same as BacktestEngine).
    # add_symbol queues if the broker is not yet built; it is flushed when
    # the feed_type is bound in super().__init__.
    for inp in inputs:
        sesh.add_symbol(inp.config)

    # Calculate warmup_ticks if not passed explicitly
    if warmup_ticks is None:
        warmup_ticks = self._calc_warmup(
            strategy, data_dict, _feed_type, warmup_pct, sesh,
        )

    # Bind data to the sesh
    sesh._bind_data(data_dict, history=history, warmup_ticks=warmup_ticks)

    super().__init__(
        strategy       = strategy,
        sesh           = sesh,
        _feed_type     = _feed_type,
        _poll_interval = _poll_interval,
        chart_ohlc_interval_ms = chart_ohlc_interval_ms,
    )

    self._ctrl = _PlaybackController(
        sesh, speed=speed, base_rate=base_rate,
        allow_backward=self._ALLOW_BACKWARD,
    )
    # Let the controller drive in-place rewinds (restart / step back).
    self._ctrl._engine = self

    # Serializes the feed loop against an in-place rewind so rings are never
    # mutated by both at the same time.
    self._rebuild_lock = threading.RLock()
    # Per-symbol cursor tracking for the tick loop's drain. Stored on the
    # instance (not as loop locals) so a rewind can reset them coherently.
    self._replay_ultimo_idx: dict[str, int] = {}
    self._replay_ultimo_ts:  dict[str, int] = {}

    # Thread-safe command queue: UI callbacks (Bokeh IOLoop thread) enqueue
    # callables that mutate broker/ring state; they are drained and executed
    # ON THE ENGINE THREAD at the top of each feed iteration, preserving the
    # single-writer model (only the engine thread mutates engine state).
    self._cmd_queue: "deque" = deque()
    self._cmd_lock = threading.Lock()

    # Recorded L2 order book to replay alongside the ticks (optional). Each
    # recorded row is a full top-K image; the shared BookReplayer applies it
    # as a snapshot to the book ring when the tick cursor reaches its
    # timestamp, so book_as_of / DeepTrades behave identically to the
    # live/streaming replay path and to the tick backtest.
    from tradetropy.data._book_replay import BookReplayer, resolve_book_sync
    self._book_replayer = BookReplayer(book or {})

    # Sync preflight: measure how the recorded book aligns with the ticks,
    # warn on desync, and (only when recoverable and sync_book=True) shift
    # the book clock onto the trade clock. Tick mode only - klines have no
    # per-trade timestamp for book_as_of.
    if book and _feed_type == "tick":
        from tradetropy.core.constants import _TICK_COL as _TC
        trades = {
            sym: (arr[:, _TC["ts"]], arr[:, _TC["price"]])
            for sym, arr in data_dict.items()
        }
        reports = resolve_book_sync(
            book, trades, sync_book=sync_book,
            engine_label=type(self).__name__,
        )
        for sym, (_rep, off) in reports.items():
            if off:
                self._book_replayer.set_offset(sym, off)
        self._book_sync_reports = {s: r for s, (r, _o) in reports.items()}
    else:
        self._book_sync_reports = {}

by_ticks classmethod

by_ticks(config: PaperConfig, data, sesh: 'ReplaySesh | None' = None, history: 'dict | None' = None, warmup_ticks: 'int | None' = None, warmup_pct: float = 0.2, speed: float = 1.0, base_rate: float = 1.0, poll_interval: float = 0.0, book: 'BookData | tuple | list | None' = None, chart_ohlc_interval_ms: int = 60000) -> 'PaperEngine'

Build a forward-only paper-trading engine over historical ticks.

Parameters:

Name	Type	Description	Default
config	PaperConfig	Declarative session configuration.	required
data		Tuple of TickData (one per symbol).	required
sesh	ReplaySesh | None	Optional pre-configured session; if omitted a ReplaySesh is built from config.initial_balance / commission.	None
	others	Same as ReplayEngine.by_ticks.	required

Returns:

Name	Type	Description
PaperEngine	'PaperEngine'	Configured forward-only paper-trading engine.

Source code in src/tradetropy/paper/engine.py

@classmethod
def by_ticks(
    cls,
    config: PaperConfig,
    data,
    sesh: "ReplaySesh | None" = None,
    history: "dict | None" = None,
    warmup_ticks: "int | None" = None,
    warmup_pct: float = 0.20,
    speed: float = 1.0,
    base_rate: float = 1.0,
    poll_interval: float = 0.0,
    book: "BookData | tuple | list | None" = None,
    chart_ohlc_interval_ms: int = 60_000,
) -> "PaperEngine":
    """
    Build a forward-only paper-trading engine over historical ticks.

    Args:
        config (PaperConfig): Declarative session configuration.
        data: Tuple of TickData (one per symbol).
        sesh (ReplaySesh | None): Optional pre-configured session; if omitted
            a ReplaySesh is built from config.initial_balance / commission.
        (others): Same as ReplayEngine.by_ticks.

    Returns:
        PaperEngine: Configured forward-only paper-trading engine.
    """
    strategy = build_manual_strategy_class(config)()
    sesh = cls._build_sesh(config, sesh)
    return super().by_ticks(
        strategy,
        data,
        sesh=sesh,
        history=history,
        warmup_ticks=warmup_ticks,
        warmup_pct=warmup_pct,
        speed=speed,
        base_rate=base_rate,
        poll_interval=poll_interval,
        book=book,
        chart_ohlc_interval_ms=chart_ohlc_interval_ms,
    )

by_klines classmethod

by_klines(config: PaperConfig, data, sesh: 'ReplaySesh | None' = None, history: 'dict | None' = None, warmup_ticks: 'int | None' = None, warmup_pct: float = 0.2, speed: float = 1.0, base_rate: float = 1.0, poll_interval: float = 0.0) -> 'PaperEngine'

Build a forward-only paper-trading engine over historical klines.

Parameters:

Name	Type	Description	Default
config	PaperConfig	Declarative session configuration.	required
data		Tuple of KlineData (one per symbol-interval pair).	required
sesh	ReplaySesh | None	Optional pre-configured session.	None
	others	Same as ReplayEngine.by_klines.	required

Returns:

Name	Type	Description
PaperEngine	'PaperEngine'	Configured forward-only paper-trading engine.

Source code in src/tradetropy/paper/engine.py

@classmethod
def by_klines(
    cls,
    config: PaperConfig,
    data,
    sesh: "ReplaySesh | None" = None,
    history: "dict | None" = None,
    warmup_ticks: "int | None" = None,
    warmup_pct: float = 0.20,
    speed: float = 1.0,
    base_rate: float = 1.0,
    poll_interval: float = 0.0,
) -> "PaperEngine":
    """
    Build a forward-only paper-trading engine over historical klines.

    Args:
        config (PaperConfig): Declarative session configuration.
        data: Tuple of KlineData (one per symbol-interval pair).
        sesh (ReplaySesh | None): Optional pre-configured session.
        (others): Same as ReplayEngine.by_klines.

    Returns:
        PaperEngine: Configured forward-only paper-trading engine.
    """
    strategy = build_manual_strategy_class(config)()
    sesh = cls._build_sesh(config, sesh)
    return super().by_klines(
        strategy,
        data,
        sesh=sesh,
        history=history,
        warmup_ticks=warmup_ticks,
        warmup_pct=warmup_pct,
        speed=speed,
        base_rate=base_rate,
        poll_interval=poll_interval,
    )