algo_engine.backtest

The backtesting framework provides data replay, simulated order matching, and performance metrics.

Replay Engines

class algo_engine.backtest.Replay(start_date: date = None, end_date: date = None, market_date: date = None, calendar: Sequence[date] = None, bod: MarketDateCallable = None, eod: MarketDateCallable = None)[source]
__init__(start_date: date = None, end_date: date = None, market_date: date = None, calendar: Sequence[date] = None, bod: MarketDateCallable = None, eod: MarketDateCallable = None) None[source]
add_bod(func: MarketDateCallable, priority: int = None) None[source]
add_eod(func: MarketDateCallable, priority: int = None)[source]
add_subscription(ticker: str, dtype: DataType | str)[source]
remove_subscription(ticker: str, dtype: DataType | str)[source]
class algo_engine.backtest.SimpleReplay(loader: MarketDataBulkLoader | MarketDataLoader = None, market_date: date = None, start_date: date = None, end_date: date = None, calendar: Sequence[date] = None, bod: MarketDateCallable = None, eod: MarketDateCallable = None)[source]

Bases: Replay

__init__(loader: MarketDataBulkLoader | MarketDataLoader = None, market_date: date = None, start_date: date = None, end_date: date = None, calendar: Sequence[date] = None, bod: MarketDateCallable = None, eod: MarketDateCallable = None)[source]
property progress: float
property tickers: list[str]
property dtypes: list[str]
property status: dict[date, str]
class algo_engine.backtest.ProgressReplay(loader: MarketDataBulkLoader | MarketDataLoader = None, market_date: date = None, start_date: date = None, end_date: date = None, calendar: Sequence[date] = None, bod: MarketDateCallable = None, eod: MarketDateCallable = None, **pbar_config)[source]

Bases: SimpleReplay

__init__(loader: MarketDataBulkLoader | MarketDataLoader = None, market_date: date = None, start_date: date = None, end_date: date = None, calendar: Sequence[date] = None, bod: MarketDateCallable = None, eod: MarketDateCallable = None, **pbar_config)[source]
class algo_engine.backtest.PyDataScope(*values)[source]
SCOPE_TRANSACTION = 1
SCOPE_ORDER = 2
SCOPE_TICK = 4
SCOPE_TICK_LITE = 8
SCOPE_ALL = 7
classmethod get_dtype(dtype: DataType | str) str | Literal['TickData', 'TickDataLite', 'OrderData', 'TransactionData'][source]
to_dtype() list[DataType][source]
to_int() list[int][source]
to_str() list[str][source]
from_str(dtype: Literal['TickData', 'TickDataLite', 'OrderData', 'TransactionData']) Self[source]

Simulated Matching

class algo_engine.backtest.SimMatch(ticker: str, event_engine=None, topic_set=None, seed: int = None, **kwargs)[source]
__init__(ticker: str, event_engine=None, topic_set=None, seed: int = None, **kwargs)[source]
static best_price(*price: float, side: TransactionSide | TransactionDirection) float[source]

Get best price for the given side.

static worst_price(*price: float, side: TransactionSide | TransactionDirection) float[source]

Get worst price for the given side.

register(topic_set: TopicSet = None, event_engine: EventEngine = None)[source]
unregister()[source]
launch_order(order: TradeInstruction, **kwargs)[source]
cancel_order(order: TradeInstruction = None, order_id: str = None, **kwargs)[source]
on_order(order, **kwargs)[source]
on_report(report, **kwargs)[source]
eod()[source]
clear()[source]
property market_time: datetime

Metrics

class algo_engine.backtest.metrics.TradeMetrics[source]
__init__()[source]
update(market_price: float)[source]
add_trades(side: int, price: float, timestamp: float, volume: float = None, trade_id: int | str = None)[source]
add_trades_batch(trade_logs: DataFrame)[source]
clear()[source]
property summary
property info
to_string() str[source]