Source code for pybroker.strategy

"""Contains implementation for backtesting trading strategies."""

"""Copyright (C) 2023 Edward West. All rights reserved.

This code is licensed under Apache 2.0 with Commons Clause license
(see LICENSE for details).
"""

import dataclasses
import numpy as np
import pandas as pd
from pybroker.cache import CacheDateFields
from pybroker.common import (
    BarData,
    DataCol,
    Day,
    IndicatorSymbol,
    ModelSymbol,
    PriceType,
    get_unique_sorted_dates,
    quantize,
    to_datetime,
    to_decimal,
    to_seconds,
    verify_data_source_columns,
    verify_date_range,
)
from pybroker.config import StrategyConfig
from pybroker.context import (
    ExecContext,
    ExecResult,
    PosSizeContext,
    set_exec_ctx_data,
    set_pos_size_ctx_data,
)
from pybroker.data import AlpacaCrypto, DataSource
from pybroker.eval import BootstrapResult, EvalMetrics, EvaluateMixin
from pybroker.indicator import Indicator, IndicatorsMixin
from pybroker.model import ModelSource, ModelsMixin, TrainedModel
from pybroker.portfolio import (
    Order,
    Portfolio,
    PortfolioBar,
    PositionBar,
    StopRecord,
    Trade,
)
from pybroker.scope import (
    ColumnScope,
    IndicatorScope,
    ModelInputScope,
    PendingOrderScope,
    PredictionScope,
    PriceScope,
    StaticScope,
    get_signals,
)
from pybroker.slippage import SlippageModel
from collections import defaultdict, deque
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from numpy.typing import NDArray
from typing import (
    Callable,
    Iterable,
    Iterator,
    Literal,
    Mapping,
    MutableMapping,
    NamedTuple,
    Optional,
    Union,
)


def _between(
    df: pd.DataFrame, start_date: datetime, end_date: datetime
) -> pd.DataFrame:
    if df.empty:
        return df
    return df[
        (df[DataCol.DATE.value].dt.tz_localize(None) >= start_date)
        & (df[DataCol.DATE.value].dt.tz_localize(None) <= end_date)
    ]


def _sort_by_score(result: ExecResult) -> float:
    return 0.0 if result.score is None else result.score


[docs] class Execution(NamedTuple): r"""Represents an execution of a :class:`.Strategy`. Holds a reference to a :class:`Callable` that implements trading logic. Attributes: id: Unique ID. symbols: Ticker symbols used for execution of ``fn``. fn: Implements trading logic. model_names: Names of :class:`pybroker.model.ModelSource`\ s used for execution of ``fn``. indicator_names: Names of :class:`pybroker.indicator.Indicator`\ s used for execution of ``fn``. """ id: int symbols: frozenset[str] fn: Optional[Callable[[ExecContext], None]] model_names: frozenset[str] indicator_names: frozenset[str]
[docs] class BacktestMixin: """Mixin implementing backtesting functionality."""
[docs] def backtest_executions( self, config: StrategyConfig, executions: set[Execution], before_exec_fn: Optional[Callable[[Mapping[str, ExecContext]], None]], after_exec_fn: Optional[Callable[[Mapping[str, ExecContext]], None]], sessions: Mapping[str, MutableMapping], models: Mapping[ModelSymbol, TrainedModel], indicator_data: Mapping[IndicatorSymbol, pd.Series], test_data: pd.DataFrame, portfolio: Portfolio, pos_size_handler: Optional[Callable[[PosSizeContext], None]], exit_dates: Mapping[str, np.datetime64], train_only: bool = False, slippage_model: Optional[SlippageModel] = None, enable_fractional_shares: bool = False, round_fill_price: bool = True, warmup: Optional[int] = None, ) -> dict[str, pd.DataFrame]: r"""Backtests a ``set`` of :class:`.Execution`\ s that implement trading logic. Args: config: :class:`pybroker.config.StrategyConfig`. executions: :class:`.Execution`\ s to run. sessions: :class:`Mapping` of symbols to :class:`Mapping` of custom data that persists for every bar during the :class:`.Execution`. models: :class:`Mapping` of :class:`pybroker.common.ModelSymbol` pairs to :class:`pybroker.common.TrainedModel`\ s. indicator_data: :class:`Mapping` of :class:`pybroker.common.IndicatorSymbol` pairs to :class:`pandas.Series` of :class:`pybroker.indicator.Indicator` values. test_data: :class:`pandas.DataFrame` of test data. portfolio: :class:`pybroker.portfolio.Portfolio`. pos_size_handler: :class:`Callable` that sets position sizes when placing orders for buy and sell signals. exit_dates: :class:`Mapping` of symbols to exit dates. train_only: Whether the backtest is run with trading rules or only trains models. enable_fractional_shares: Whether to enable trading fractional shares. round_fill_price: Whether to round fill prices to the nearest cent. warmup: Number of bars that need to pass before running the executions. Returns: Dictionary of :class:`pandas.DataFrame`\ s containing bar data, indicator data, and model predictions for each symbol when :attr:`pybroker.config.StrategyConfig.return_signals` is ``True``. """ test_dates = get_unique_sorted_dates(test_data[DataCol.DATE.value]) test_syms = sorted(test_data[DataCol.SYMBOL.value].unique()) test_data = ( test_data.reset_index(drop=True) .set_index([DataCol.SYMBOL.value, DataCol.DATE.value]) .sort_index() ) col_scope = ColumnScope(test_data) ind_scope = IndicatorScope(indicator_data, test_dates) input_scope = ModelInputScope(col_scope, ind_scope, models) pred_scope = PredictionScope(models, input_scope) if train_only: if config.return_signals: return get_signals(test_syms, col_scope, ind_scope, pred_scope) return {} sym_end_index: dict[str, int] = defaultdict(int) price_scope = PriceScope(col_scope, sym_end_index, round_fill_price) pending_order_scope = PendingOrderScope() exec_ctxs: dict[str, ExecContext] = {} exec_fns: dict[str, Callable[[ExecContext], None]] = {} for sym in test_syms: for exec in executions: if sym not in exec.symbols: continue exec_ctxs[sym] = ExecContext( symbol=sym, config=config, portfolio=portfolio, col_scope=col_scope, ind_scope=ind_scope, input_scope=input_scope, pred_scope=pred_scope, pending_order_scope=pending_order_scope, models=models, sym_end_index=sym_end_index, session=sessions[sym], ) if exec.fn is not None: exec_fns[sym] = exec.fn sym_exec_dates = { sym: frozenset(test_data.loc[pd.IndexSlice[sym, :]].index.values) for sym in exec_ctxs.keys() } cover_sched: dict[np.datetime64, list[ExecResult]] = defaultdict(list) buy_sched: dict[np.datetime64, list[ExecResult]] = defaultdict(list) sell_sched: dict[np.datetime64, list[ExecResult]] = defaultdict(list) if pos_size_handler is not None: pos_ctx = PosSizeContext( config=config, portfolio=portfolio, col_scope=col_scope, ind_scope=ind_scope, input_scope=input_scope, pred_scope=pred_scope, pending_order_scope=pending_order_scope, models=models, sessions=sessions, sym_end_index=sym_end_index, ) logger = StaticScope.instance().logger logger.backtest_executions_start(test_dates) cover_results: deque[ExecResult] = deque() buy_results: deque[ExecResult] = deque() sell_results: deque[ExecResult] = deque() exit_ctxs: deque[ExecContext] = deque() active_ctxs: dict[str, ExecContext] = {} for i, date in enumerate(test_dates): active_ctxs.clear() for sym, ctx in exec_ctxs.items(): if date not in sym_exec_dates[sym]: continue sym_end_index[sym] += 1 if warmup and sym_end_index[sym] <= warmup: continue active_ctxs[sym] = ctx set_exec_ctx_data(ctx, date) if ( exit_dates and sym in exit_dates and date == exit_dates[sym] ): exit_ctxs.append(ctx) is_cover_sched = date in cover_sched is_buy_sched = date in buy_sched is_sell_sched = date in sell_sched if ( config.max_long_positions is not None or pos_size_handler is not None ): if is_cover_sched: cover_sched[date].sort(key=_sort_by_score, reverse=True) elif is_buy_sched: buy_sched[date].sort(key=_sort_by_score, reverse=True) if is_sell_sched and ( config.max_short_positions is not None or pos_size_handler is not None ): sell_sched[date].sort(key=_sort_by_score, reverse=True) if pos_size_handler is not None and ( is_cover_sched or is_buy_sched or is_sell_sched ): pos_size_buy_results = None if is_cover_sched: pos_size_buy_results = cover_sched[date] elif is_buy_sched: pos_size_buy_results = buy_sched[date] self._set_pos_sizes( pos_size_handler=pos_size_handler, pos_ctx=pos_ctx, buy_results=pos_size_buy_results, sell_results=sell_sched[date] if is_sell_sched else None, ) portfolio.check_stops(date, price_scope) if is_cover_sched: self._place_buy_orders( date=date, price_scope=price_scope, pending_order_scope=pending_order_scope, buy_sched=cover_sched, portfolio=portfolio, enable_fractional_shares=enable_fractional_shares, ) if is_sell_sched: self._place_sell_orders( date=date, price_scope=price_scope, pending_order_scope=pending_order_scope, sell_sched=sell_sched, portfolio=portfolio, enable_fractional_shares=enable_fractional_shares, ) if is_buy_sched: self._place_buy_orders( date=date, price_scope=price_scope, pending_order_scope=pending_order_scope, buy_sched=buy_sched, portfolio=portfolio, enable_fractional_shares=enable_fractional_shares, ) portfolio.capture_bar(date, test_data) if before_exec_fn is not None and active_ctxs: before_exec_fn(active_ctxs) for sym, ctx in active_ctxs.items(): if sym in exec_fns: exec_fns[sym](ctx) if after_exec_fn is not None and active_ctxs: after_exec_fn(active_ctxs) for ctx in active_ctxs.values(): if slippage_model and (ctx.buy_shares or ctx.sell_shares): self._apply_slippage(slippage_model, ctx) result = ctx.to_result() if result is None: continue if result.buy_shares is not None: if result.cover: cover_results.append(result) else: buy_results.append(result) if result.sell_shares is not None: sell_results.append(result) while cover_results: self._schedule_order( result=cover_results.popleft(), created=date, sym_end_index=sym_end_index, delay=config.buy_delay, sched=cover_sched, col_scope=col_scope, pending_order_scope=pending_order_scope, ) while buy_results: self._schedule_order( result=buy_results.popleft(), created=date, sym_end_index=sym_end_index, delay=config.buy_delay, sched=buy_sched, col_scope=col_scope, pending_order_scope=pending_order_scope, ) while sell_results: self._schedule_order( result=sell_results.popleft(), created=date, sym_end_index=sym_end_index, delay=config.sell_delay, sched=sell_sched, col_scope=col_scope, pending_order_scope=pending_order_scope, ) while exit_ctxs: self._exit_position( portfolio=portfolio, date=date, ctx=exit_ctxs.popleft(), exit_cover_fill_price=config.exit_cover_fill_price, exit_sell_fill_price=config.exit_sell_fill_price, price_scope=price_scope, ) portfolio.incr_bars() if i % 10 == 0 or i == len(test_dates) - 1: logger.backtest_executions_loading(i + 1) return ( get_signals(test_syms, col_scope, ind_scope, pred_scope) if config.return_signals else {} )
def _apply_slippage( self, slippage_model: SlippageModel, ctx: ExecContext, ): buy_shares = to_decimal(ctx.buy_shares) if ctx.buy_shares else None sell_shares = to_decimal(ctx.sell_shares) if ctx.sell_shares else None slippage_model.apply_slippage( ctx, buy_shares=buy_shares, sell_shares=sell_shares ) def _exit_position( self, portfolio: Portfolio, date: np.datetime64, ctx: ExecContext, exit_cover_fill_price: Union[ PriceType, Callable[[str, BarData], Union[int, float, Decimal]] ], exit_sell_fill_price: Union[ PriceType, Callable[[str, BarData], Union[int, float, Decimal]] ], price_scope: PriceScope, ): buy_fill_price = price_scope.fetch(ctx.symbol, exit_cover_fill_price) sell_fill_price = price_scope.fetch(ctx.symbol, exit_sell_fill_price) portfolio.exit_position( date, ctx.symbol, buy_fill_price=buy_fill_price, sell_fill_price=sell_fill_price, ) def _set_pos_sizes( self, pos_size_handler: Callable[[PosSizeContext], None], pos_ctx: PosSizeContext, buy_results: Optional[list[ExecResult]], sell_results: Optional[list[ExecResult]], ): set_pos_size_ctx_data( ctx=pos_ctx, buy_results=buy_results, sell_results=sell_results ) pos_size_handler(pos_ctx) for id, shares in pos_ctx._signal_shares.items(): if id < 0: raise ValueError(f"Invalid ExecSignal id: {id}") if buy_results is not None and sell_results is not None: if id >= (len(buy_results) + len(sell_results)): raise ValueError(f"Invalid ExecSignal id: {id}") if id < len(buy_results): buy_results[id].buy_shares = to_decimal(shares) else: sell_results[ id - len(buy_results) ].sell_shares = to_decimal(shares) elif buy_results is not None: if id >= len(buy_results): raise ValueError(f"Invalid ExecSignal id: {id}") buy_results[id].buy_shares = to_decimal(shares) elif sell_results is not None: if id >= len(sell_results): raise ValueError(f"Invalid ExecSignal id: {id}") sell_results[id].sell_shares = to_decimal(shares) else: raise ValueError( "buy_results and sell_results cannot both be None." ) def _schedule_order( self, result: ExecResult, created: np.datetime64, sym_end_index: Mapping[str, int], delay: int, sched: Mapping[np.datetime64, list[ExecResult]], col_scope: ColumnScope, pending_order_scope: PendingOrderScope, ): date_loc = sym_end_index[result.symbol] - 1 dates = col_scope.fetch(result.symbol, DataCol.DATE.value) if dates is None: raise ValueError("Dates not found.") logger = StaticScope.instance().logger if date_loc + delay < len(dates): date = dates[date_loc + delay] order_type: Literal["buy", "sell"] if result.buy_shares is not None: order_type = "buy" shares = result.buy_shares limit_price = result.buy_limit_price fill_price = result.buy_fill_price elif result.sell_shares is not None: order_type = "sell" shares = result.sell_shares limit_price = result.sell_limit_price fill_price = result.sell_fill_price else: raise ValueError("buy_shares or sell_shares needs to be set.") result.pending_order_id = pending_order_scope.add( type=order_type, symbol=result.symbol, created=created, exec_date=date, shares=shares, limit_price=limit_price, fill_price=fill_price, ) sched[date].append(result) logger.debug_schedule_order(date, result) else: logger.debug_unscheduled_order(result) def _place_buy_orders( self, date: np.datetime64, price_scope: PriceScope, pending_order_scope: PendingOrderScope, buy_sched: dict[np.datetime64, list[ExecResult]], portfolio: Portfolio, enable_fractional_shares: bool, ): buy_results = buy_sched[date] for result in buy_results: if result.buy_shares is None: continue if ( result.pending_order_id is None or not pending_order_scope.contains(result.pending_order_id) ): continue pending_order_scope.remove(result.pending_order_id) buy_shares = self._get_shares( result.buy_shares, enable_fractional_shares ) fill_price = price_scope.fetch( result.symbol, result.buy_fill_price ) order = portfolio.buy( date=date, symbol=result.symbol, shares=buy_shares, fill_price=fill_price, limit_price=result.buy_limit_price, stops=result.long_stops, ) logger = StaticScope.instance().logger if order is None: logger.debug_unfilled_buy_order( date=date, symbol=result.symbol, shares=buy_shares, fill_price=fill_price, limit_price=result.buy_limit_price, ) else: logger.debug_filled_buy_order( date=date, symbol=result.symbol, shares=buy_shares, fill_price=fill_price, limit_price=result.buy_limit_price, ) del buy_sched[date] def _place_sell_orders( self, date: np.datetime64, price_scope: PriceScope, pending_order_scope: PendingOrderScope, sell_sched: dict[np.datetime64, list[ExecResult]], portfolio: Portfolio, enable_fractional_shares: bool, ): sell_results = sell_sched[date] for result in sell_results: if result.sell_shares is None: continue if ( result.pending_order_id is None or not pending_order_scope.contains(result.pending_order_id) ): continue pending_order_scope.remove(result.pending_order_id) sell_shares = self._get_shares( result.sell_shares, enable_fractional_shares ) fill_price = price_scope.fetch( result.symbol, result.sell_fill_price ) order = portfolio.sell( date=date, symbol=result.symbol, shares=sell_shares, fill_price=fill_price, limit_price=result.sell_limit_price, stops=result.short_stops, ) logger = StaticScope.instance().logger if order is None: logger.debug_unfilled_sell_order( date=date, symbol=result.symbol, shares=sell_shares, fill_price=fill_price, limit_price=result.sell_limit_price, ) else: logger.debug_filled_sell_order( date=date, symbol=result.symbol, shares=sell_shares, fill_price=fill_price, limit_price=result.sell_limit_price, ) del sell_sched[date] def _get_shares( self, shares: Union[int, float, Decimal], enable_fractional_shares: bool, ) -> Decimal: if enable_fractional_shares: return to_decimal(shares) else: return to_decimal(int(shares))
[docs] class WalkforwardWindow(NamedTuple): """Contains ``train_data`` and ``test_data`` of a time window used for `Walkforward Analysis <https://www.pybroker.com/en/latest/notebooks/6.%20Training%20a%20Model.html#Walkforward-Analysis>`_. Attributes: train_data: Train data. test_data: Test data. """ train_data: NDArray[np.int_] test_data: NDArray[np.int_]
[docs] class WalkforwardMixin: """Mixin implementing logic for `Walkforward Analysis <https://www.pybroker.com/en/latest/notebooks/6.%20Training%20a%20Model.html#Walkforward-Analysis>`_. """
[docs] def walkforward_split( self, df: pd.DataFrame, windows: int, lookahead: int, train_size: float = 0.9, shuffle: bool = False, ) -> Iterator[WalkforwardWindow]: r"""Splits a :class:`pandas.DataFrame` containing data for multiple ticker symbols into an :class:`Iterator` of train/test time windows for `Walkforward Analysis <https://www.pybroker.com/en/latest/notebooks/6.%20Training%20a%20Model.html#Walkforward-Analysis>`_. Args: df: :class:`pandas.DataFrame` of data to split into train/test windows for Walkforward Analysis. windows: Number of walkforward time windows. lookahead: Number of bars in the future of the target prediction. For example, predicting returns for the next bar would have a ``lookahead`` of ``1``. This quantity is needed to prevent training data from leaking into the test boundary. train_size: Amount of data in ``df`` to use for training, where the max ``train_size`` is ``1``. For example, a ``train_size`` of ``0.9`` would result in 90% of data in ``df`` being used for training and the remaining 10% of data being used for testing. shuffle: Whether to randomly shuffle the data used for training. Defaults to ``False``. Returns: :class:`Iterator` of :class:`.WalkforwardWindow`\ s containing train and test data. """ if windows <= 0: raise ValueError("windows needs to be > 0.") if lookahead <= 0: raise ValueError("lookahead needs to be > 0.") if train_size < 0: raise ValueError("train_size cannot be negative.") if df.empty: raise ValueError("DataFrame is empty.") date_col = DataCol.DATE.value dates = df[[date_col]] window_dates = get_unique_sorted_dates(df[date_col]) error_msg = f""" Invalid params for {len(window_dates)} dates: windows: {windows} lookahead: {lookahead} train_size: {train_size} """ if train_size == 0 or train_size == 1: window_length = int(len(window_dates) / windows) offset = len(window_dates) - window_length * windows for i in range(windows): start = offset + i * window_length end = start + window_length if train_size == 0: test_idx = dates[ (dates[date_col] >= window_dates[start]) & (dates[date_col] <= window_dates[end - 1]) ] test_idx = test_idx.index.to_numpy() yield WalkforwardWindow(np.array(tuple()), test_idx) else: train_idx = dates[ (dates[date_col] >= window_dates[start]) & (dates[date_col] <= window_dates[end - 1]) ] train_idx = train_idx.index.to_numpy() if shuffle: np.random.shuffle(train_idx) yield WalkforwardWindow(train_idx, np.array(tuple())) elif windows == 1: res = len(window_dates) - 1 - lookahead if res <= 0: raise ValueError(error_msg) train_length = int(res * train_size) test_length = int(res * (1 - train_size)) train_start = 0 train_end = train_length test_start = train_end + lookahead if test_start >= len(window_dates): raise ValueError(error_msg) test_end = test_start + test_length train_idx = dates[ (dates[date_col] >= window_dates[train_start]) & (dates[date_col] <= window_dates[train_end]) ] test_idx = dates[ (dates[date_col] >= window_dates[test_start]) & (dates[date_col] <= window_dates[test_end]) ] train_idx = train_idx.index.to_numpy() test_idx = test_idx.index.to_numpy() if shuffle: np.random.shuffle(train_idx) yield WalkforwardWindow(train_idx, test_idx) else: res = len(window_dates) - (lookahead - 1) * windows window_length = res / windows # type: ignore[assignment] train_length = int(window_length * train_size) test_length = int(window_length * (1 - train_size)) if train_length < 0 or test_length < 0: raise ValueError(error_msg) while True: rem = (res - (train_length + test_length * windows)) / windows train_incr = int(rem * train_size) test_incr = int(rem * (1 - train_size)) if train_incr == 0 or test_incr == 0: break train_length += train_incr test_length += test_incr if train_length == 0 and test_length == 0: raise ValueError(error_msg) window_idx = [] for i in range(windows): test_end = i * test_length test_start = test_end + test_length train_end = test_start + lookahead - 1 train_start = train_end + train_length window_idx.append( (train_start, train_end, test_start, test_end) ) window_idx.reverse() window_dates = window_dates[::-1] for train_start, train_end, test_start, test_end in window_idx: train_idx = dates[ (dates[date_col] > window_dates[train_start]) & (dates[date_col] <= window_dates[train_end]) ] test_idx = dates[ (dates[date_col] > window_dates[test_start]) & (dates[date_col] <= window_dates[test_end]) ] train_idx = train_idx.index.to_numpy() test_idx = test_idx.index.to_numpy() if shuffle: np.random.shuffle(train_idx) yield WalkforwardWindow(train_idx, test_idx)
[docs] @dataclass(frozen=True) class TestResult: r"""Contains the results of backtesting a :class:`.Strategy`. Attributes: start_date: Starting date of backtest. end_date: Ending date of backtest. portfolio: :class:`pandas.DataFrame` of :class:`pybroker.portfolio.Portfolio` balances for every bar. positions: :class:`pandas.DataFrame` of :class:`pybroker.portfolio.Position` balances for every bar. orders: :class:`pandas.DataFrame` of all orders that were placed. trades: :class:`pandas.DataFrame` of all trades that were made. metrics: Evaluation metrics. metrics_df: :class:`pandas.DataFrame` of evaluation metrics. bootstrap: Randomized bootstrap evaluation metrics. signals: Dictionary of :class:`pandas.DataFrame`\ s containing bar data, indicator data, and model predictions for each symbol when :attr:`pybroker.config.StrategyConfig.return_signals` is ``True``. stops: :class:`pandas.DataFrame` containing stop data per-bar when :attr:`pybroker.config.StrategyConfig.return_stops` is ``True``. """ start_date: datetime end_date: datetime portfolio: pd.DataFrame positions: pd.DataFrame orders: pd.DataFrame trades: pd.DataFrame metrics: EvalMetrics metrics_df: pd.DataFrame bootstrap: Optional[BootstrapResult] signals: Optional[dict[str, pd.DataFrame]] stops: Optional[pd.DataFrame]
[docs] class Strategy( BacktestMixin, EvaluateMixin, IndicatorsMixin, ModelsMixin, WalkforwardMixin, ): """Class representing a trading strategy to backtest. Args: data_source: :class:`pybroker.data.DataSource` or :class:`pandas.DataFrame` of backtesting data. start_date: Starting date of the data to fetch from ``data_source`` (inclusive). end_date: Ending date of the data to fetch from ``data_source`` (inclusive). config: ``Optional`` :class:`pybroker.config.StrategyConfig`. """ _execution_id: int = 0 def __init__( self, data_source: Union[DataSource, pd.DataFrame], start_date: Union[str, datetime], end_date: Union[str, datetime], config: Optional[StrategyConfig] = None, ): self._verify_data_source(data_source) self._data_source = data_source self._start_date = to_datetime(start_date) self._end_date = to_datetime(end_date) verify_date_range(self._start_date, self._end_date) if config is not None: self._verify_config(config) self._config = config else: self._config = StrategyConfig() self._executions: set[Execution] = set() self._before_exec_fn: Optional[ Callable[[Mapping[str, ExecContext]], None] ] = None self._after_exec_fn: Optional[ Callable[[Mapping[str, ExecContext]], None] ] = None self._pos_size_handler: Optional[Callable[[PosSizeContext], None]] = ( None ) self._slippage_model: Optional[SlippageModel] = None self._scope = StaticScope.instance() self._logger = self._scope.logger def _verify_config(self, config: StrategyConfig): if config.initial_cash <= 0: raise ValueError("initial_cash must be greater than 0.") if ( config.max_long_positions is not None and config.max_long_positions <= 0 ): raise ValueError("max_long_positions must be greater than 0.") if ( config.max_short_positions is not None and config.max_short_positions <= 0 ): raise ValueError("max_short_positions must be greater than 0.") if config.buy_delay <= 0: raise ValueError("buy_delay must be greater than 0.") if config.sell_delay <= 0: raise ValueError("sell_delay must be greater than 0.") if config.bootstrap_samples <= 0: raise ValueError("bootstrap_samples must be greater than 0.") if config.bootstrap_sample_size <= 0: raise ValueError("bootstrap_sample_size must be greater than 0.") def _verify_data_source( self, data_source: Union[DataSource, pd.DataFrame] ): if isinstance(data_source, pd.DataFrame): verify_data_source_columns(data_source) elif not isinstance(data_source, DataSource): raise TypeError(f"Invalid data_source type: {type(data_source)}")
[docs] def set_slippage_model(self, slippage_model: Optional[SlippageModel]): """Sets :class:`pybroker.slippage.SlippageModel`.""" self._slippage_model = slippage_model
[docs] def add_execution( self, fn: Optional[Callable[[ExecContext], None]], symbols: Union[str, Iterable[str]], models: Optional[Union[ModelSource, Iterable[ModelSource]]] = None, indicators: Optional[Union[Indicator, Iterable[Indicator]]] = None, ): r"""Adds an execution to backtest. Args: fn: :class:`Callable` invoked on every bar of data during the backtest and passed an :class:`pybroker.context.ExecContext` for each ticker symbol in ``symbols``. symbols: Ticker symbols used to run ``fn``, where ``fn`` is called separately for each symbol. models: :class:`Iterable` of :class:`pybroker.model.ModelSource`\ s to train/load for backtesting. indicators: :class:`Iterable` of :class:`pybroker.indicator.Indicator`\ s to compute for backtesting. """ symbols = ( frozenset((symbols,)) if isinstance(symbols, str) else frozenset(symbols) ) if not symbols: raise ValueError("symbols cannot be empty.") for sym in symbols: for exec in self._executions: if sym in exec.symbols: raise ValueError( f"{sym} was already added to an execution." ) if models is not None: for model in ( (models,) if isinstance(models, ModelSource) else models ): if not self._scope.has_model_source(model.name): raise ValueError( f"ModelSource {model.name!r} was not registered." ) if model is not self._scope.get_model_source(model.name): raise ValueError( f"ModelSource {model.name!r} does not match " "registered ModelSource." ) model_names = ( ( frozenset((models.name,)) if isinstance(models, ModelSource) else frozenset(model.name for model in models) ) if models is not None else frozenset() ) if indicators is not None: for ind in ( (indicators,) if isinstance(indicators, Indicator) else indicators ): if not self._scope.has_indicator(ind.name): raise ValueError( f"Indicator {ind.name!r} was not registered." ) if ind is not self._scope.get_indicator(ind.name): raise ValueError( f"Indicator {ind.name!r} does not match registered " "Indicator." ) ind_names = ( ( frozenset((indicators.name,)) if isinstance(indicators, Indicator) else frozenset(ind.name for ind in indicators) ) if indicators is not None else frozenset() ) self._execution_id += 1 self._executions.add( Execution( id=self._execution_id, symbols=symbols, fn=fn, model_names=model_names, indicator_names=ind_names, ) )
[docs] def set_before_exec( self, fn: Optional[Callable[[Mapping[str, ExecContext]], None]] ): r""":class:`Callable[[Mapping[str, ExecContext]]` that runs before all execution functions. Args: fn: :class:`Callable` that takes a :class:`Mapping` of all ticker symbols to :class:`ExecContext`\ s. """ self._before_exec_fn = fn
[docs] def set_after_exec( self, fn: Optional[Callable[[Mapping[str, ExecContext]], None]] ): r""":class:`Callable[[Mapping[str, ExecContext]]` that runs after all execution functions. Args: fn: :class:`Callable` that takes a :class:`Mapping` of all ticker symbols to :class:`ExecContext`\ s. """ self._after_exec_fn = fn
[docs] def clear_executions(self): """Clears executions that were added with :meth:`.add_execution`.""" self._executions.clear()
[docs] def set_pos_size_handler( self, fn: Optional[Callable[[PosSizeContext], None]] ): r"""Sets a :class:`Callable` that determines position sizes to use for buy and sell signals. Args: fn: :class:`Callable` invoked before placing orders for buy and sell signals, and is passed a :class:`pybroker.context.PosSizeContext`. """ self._pos_size_handler = fn
[docs] def backtest( self, start_date: Optional[Union[str, datetime]] = None, end_date: Optional[Union[str, datetime]] = None, timeframe: str = "", between_time: Optional[tuple[str, str]] = None, days: Optional[Union[str, Day, Iterable[Union[str, Day]]]] = None, lookahead: int = 1, train_size: int = 0, shuffle: bool = False, calc_bootstrap: bool = False, disable_parallel: bool = False, warmup: Optional[int] = None, portfolio: Optional[Portfolio] = None, ) -> TestResult: """Backtests the trading strategy by running executions that were added with :meth:`.add_execution`. Args: start_date: Starting date of the backtest (inclusive). Must be within ``start_date`` and ``end_date`` range that was passed to :meth:`.__init__`. end_date: Ending date of the backtest (inclusive). Must be within ``start_date`` and ``end_date`` range that was passed to :meth:`.__init__`. timeframe: Formatted string that specifies the timeframe resolution of the backtesting data. The timeframe string supports the following units: - ``"s"``/``"sec"``: seconds - ``"m"``/``"min"``: minutes - ``"h"``/``"hour"``: hours - ``"d"``/``"day"``: days - ``"w"``/``"week"``: weeks An example timeframe string is ``1h 30m``. between_time: ``tuple[str, str]`` of times of day e.g. ('9:30', '16:00') used to filter the backtesting data (inclusive). days: Days (e.g. ``"mon"``, ``"tues"`` etc.) used to filter the backtesting data. lookahead: Number of bars in the future of the target prediction. For example, predicting returns for the next bar would have a ``lookahead`` of ``1``. This quantity is needed to prevent training data from leaking into the test boundary. train_size: Amount of :class:`pybroker.data.DataSource` data to use for training, where the max ``train_size`` is ``1``. For example, a ``train_size`` of ``0.9`` would result in 90% of data being used for training and the remaining 10% of data being used for testing. shuffle: Whether to randomly shuffle the data used for training. Defaults to ``False``. Disabled when model caching is enabled via :meth:`pybroker.cache.enable_model_cache`. calc_bootstrap: Whether to compute randomized bootstrap evaluation metrics. Defaults to ``False``. disable_parallel: If ``True``, :class:`pybroker.indicator.Indicator` data is computed serially. If ``False``, :class:`pybroker.indicator.Indicator` data is computed in parallel using multiple processes. Defaults to ``False``. warmup: Number of bars that need to pass before running the executions. portfolio: Custom :class:`pybroker.portfolio.Portfolio` to use for backtests. Returns: :class:`.TestResult` containing portfolio balances, order history, and evaluation metrics. """ return self.walkforward( windows=1, lookahead=lookahead, start_date=start_date, end_date=end_date, timeframe=timeframe, between_time=between_time, days=days, train_size=train_size, shuffle=shuffle, calc_bootstrap=calc_bootstrap, disable_parallel=disable_parallel, warmup=warmup, portfolio=portfolio, )
[docs] def walkforward( self, windows: int, lookahead: int = 1, start_date: Optional[Union[str, datetime]] = None, end_date: Optional[Union[str, datetime]] = None, timeframe: str = "", between_time: Optional[tuple[str, str]] = None, days: Optional[Union[str, Day, Iterable[Union[str, Day]]]] = None, train_size: float = 0.5, shuffle: bool = False, calc_bootstrap: bool = False, disable_parallel: bool = False, warmup: Optional[int] = None, portfolio: Optional[Portfolio] = None, ) -> TestResult: """Backtests the trading strategy using `Walkforward Analysis <https://www.pybroker.com/en/latest/notebooks/6.%20Training%20a%20Model.html#Walkforward-Analysis>`_. Backtesting data supplied by the :class:`pybroker.data.DataSource` is divided into ``windows`` number of equal sized time windows, with each window split into train and test data as specified by ``train_size``. The backtest "walks forward" in time through each window, running executions that were added with :meth:`.add_execution`. Args: windows: Number of walkforward time windows. start_date: Starting date of the Walkforward Analysis (inclusive). Must be within ``start_date`` and ``end_date`` range that was passed to :meth:`.__init__`. end_date: Ending date of the Walkforward Analysis (inclusive). Must be within ``start_date`` and ``end_date`` range that was passed to :meth:`.__init__`. timeframe: Formatted string that specifies the timeframe resolution of the backtesting data. The timeframe string supports the following units: - ``"s"``/``"sec"``: seconds - ``"m"``/``"min"``: minutes - ``"h"``/``"hour"``: hours - ``"d"``/``"day"``: days - ``"w"``/``"week"``: weeks An example timeframe string is ``1h 30m``. between_time: ``tuple[str, str]`` of times of day e.g. ('9:30', '16:00') used to filter the backtesting data (inclusive). days: Days (e.g. ``"mon"``, ``"tues"`` etc.) used to filter the backtesting data. lookahead: Number of bars in the future of the target prediction. For example, predicting returns for the next bar would have a ``lookahead`` of ``1``. This quantity is needed to prevent training data from leaking into the test boundary. train_size: Amount of :class:`pybroker.data.DataSource` data to use for training, where the max ``train_size`` is ``1``. For example, a ``train_size`` of ``0.9`` would result in 90% of data being used for training and the remaining 10% of data being used for testing. shuffle: Whether to randomly shuffle the data used for training. Defaults to ``False``. Disabled when model caching is enabled via :meth:`pybroker.cache.enable_model_cache`. calc_bootstrap: Whether to compute randomized bootstrap evaluation metrics. Defaults to ``False``. disable_parallel: If ``True``, :class:`pybroker.indicator.Indicator` data is computed serially. If ``False``, :class:`pybroker.indicator.Indicator` data is computed in parallel using multiple processes. Defaults to ``False``. warmup: Number of bars that need to pass before running the executions. portfolio: Custom :class:`pybroker.portfolio.Portfolio` to use for backtests. Returns: :class:`.TestResult` containing portfolio balances, order history, and evaluation metrics. """ if warmup is not None and warmup < 1: raise ValueError("warmup must be > 0.") scope = StaticScope.instance() try: scope.freeze_data_cols() if not self._executions: raise ValueError("No executions were added.") start_dt = ( self._start_date if start_date is None else to_datetime(start_date) ) if start_dt < self._start_date or start_dt > self._end_date: raise ValueError( f"start_date must be between {self._start_date} and " f"{self._end_date}." ) end_dt = ( self._end_date if end_date is None else to_datetime(end_date) ) if end_dt < self._start_date or end_dt > self._end_date: raise ValueError( f"end_date must be between {self._start_date} and " f"{self._end_date}." ) if start_dt is not None and end_dt is not None: verify_date_range(start_dt, end_dt) self._logger.walkforward_start(start_dt, end_dt) df = self._fetch_data(timeframe) day_ids = self._to_day_ids(days) df = self._filter_dates( df=df, start_date=start_dt, end_date=end_dt, between_time=between_time, days=day_ids, ) tf_seconds = to_seconds(timeframe) indicator_data = self._fetch_indicators( df=df, cache_date_fields=CacheDateFields( start_date=start_dt, end_date=end_dt, tf_seconds=tf_seconds, between_time=between_time, days=day_ids, ), disable_parallel=disable_parallel, ) train_only = ( self._before_exec_fn is None and self._after_exec_fn is None and all(map(lambda e: e.fn is None, self._executions)) ) if portfolio is None: portfolio = Portfolio( self._config.initial_cash, self._config.fee_mode, self._config.fee_amount, self._config.subtract_fees, self._fractional_shares_enabled(), self._config.max_long_positions, self._config.max_short_positions, self._config.return_stops, ) signals = self._run_walkforward( portfolio=portfolio, df=df, indicator_data=indicator_data, tf_seconds=tf_seconds, between_time=between_time, days=day_ids, windows=windows, lookahead=lookahead, train_size=train_size, shuffle=shuffle, train_only=train_only, warmup=warmup, ) if train_only: self._logger.walkforward_completed() return self._to_test_result( start_dt, end_dt, portfolio, calc_bootstrap, train_only, signals if self._config.return_signals else None, ) finally: scope.unfreeze_data_cols()
def _to_day_ids( self, days: Optional[Union[str, Day, Iterable[Union[str, Day]]]] ) -> Optional[tuple[int]]: if days is None: return None days = ( (days,) if isinstance(days, str) or isinstance(days, Day) else days ) return tuple( sorted( (day.value if isinstance(day, Day) else Day[day.upper()].value) # type: ignore[union-attr] for day in set(days) # type: ignore[arg-type] ) ) # type: ignore[return-value] def _fractional_shares_enabled(self): return self._config.enable_fractional_shares or isinstance( self._data_source, AlpacaCrypto ) def _run_walkforward( self, portfolio: Portfolio, df: pd.DataFrame, indicator_data: dict[IndicatorSymbol, pd.Series], tf_seconds: int, between_time: Optional[tuple[str, str]], days: Optional[tuple[int]], windows: int, lookahead: int, train_size: float, shuffle: bool, train_only: bool, warmup: Optional[int], ) -> dict[str, pd.DataFrame]: sessions: dict[str, dict] = defaultdict(dict) exit_dates: dict[str, np.datetime64] = {} if self._config.exit_on_last_bar: for exec in self._executions: for sym in exec.symbols: sym_dates = df[df[DataCol.SYMBOL.value] == sym][ DataCol.DATE.value ].values if len(sym_dates): sym_dates.sort() exit_dates[sym] = sym_dates[-1] signals: dict[str, pd.DataFrame] = {} for train_idx, test_idx in self.walkforward_split( df=df, windows=windows, lookahead=lookahead, train_size=train_size, shuffle=shuffle, ): models: dict[ModelSymbol, TrainedModel] = {} train_data = df.loc[train_idx] test_data = df.loc[test_idx] if not train_data.empty: model_syms = { ModelSymbol(model_name, sym) for sym in train_data[DataCol.SYMBOL.value].unique() for execution in self._executions for model_name in execution.model_names if sym in execution.symbols } train_dates = get_unique_sorted_dates( train_data[DataCol.DATE.value] ) models = self.train_models( model_syms=model_syms, train_data=train_data, test_data=test_data, indicator_data=indicator_data, cache_date_fields=CacheDateFields( start_date=to_datetime(train_dates[0]), end_date=to_datetime(train_dates[-1]), tf_seconds=tf_seconds, between_time=between_time, days=days, ), ) if test_data.empty: return signals split_signals = self.backtest_executions( config=self._config, executions=self._executions, before_exec_fn=self._before_exec_fn, after_exec_fn=self._after_exec_fn, sessions=sessions, models=models, indicator_data=indicator_data, test_data=test_data, portfolio=portfolio, pos_size_handler=self._pos_size_handler, exit_dates=exit_dates, train_only=train_only, slippage_model=self._slippage_model, enable_fractional_shares=self._fractional_shares_enabled(), round_fill_price=self._config.round_fill_price, warmup=warmup, ) for sym, signals_df in split_signals.items(): if sym in signals: signals[sym] = pd.concat([signals[sym], signals_df]) else: signals[sym] = signals_df return signals def _filter_dates( self, df: pd.DataFrame, start_date: datetime, end_date: datetime, between_time: Optional[tuple[str, str]], days: Optional[tuple[int]], ) -> pd.DataFrame: if start_date != self._start_date or end_date != self._end_date: df = _between(df, start_date, end_date).reset_index(drop=True) if df[DataCol.DATE.value].dt.tz is not None: # Fixes bug on Windows. # https://stackoverflow.com/questions/51827582/message-exception-ignored-when-dealing-pandas-datetime-type df[DataCol.DATE.value] = df[DataCol.DATE.value].dt.tz_convert(None) is_time_range = between_time is not None or days is not None if is_time_range: df = df.reset_index(drop=True).set_index(DataCol.DATE.value) if days is not None: self._logger.info_walkforward_on_days(days) df = df[df.index.weekday.isin(frozenset(days))] if between_time is not None: if len(between_time) != 2: raise ValueError( "between_time must be a tuple[str, str] of start time and" f" end time, received {between_time!r}." ) self._logger.info_walkforward_between_time(between_time) df = df.between_time(*between_time) if is_time_range: df = df.reset_index() return df def _fetch_indicators( self, df: pd.DataFrame, cache_date_fields: CacheDateFields, disable_parallel: bool, ) -> dict[IndicatorSymbol, pd.Series]: indicator_syms = set() for execution in self._executions: for sym in execution.symbols: for model_name in execution.model_names: ind_names = self._scope.get_indicator_names(model_name) for ind_name in ind_names: indicator_syms.add(IndicatorSymbol(ind_name, sym)) for ind_name in execution.indicator_names: indicator_syms.add(IndicatorSymbol(ind_name, sym)) return self.compute_indicators( df=df, indicator_syms=indicator_syms, cache_date_fields=cache_date_fields, disable_parallel=disable_parallel, ) def _fetch_data(self, timeframe: str) -> pd.DataFrame: unique_syms = { sym for execution in self._executions for sym in execution.symbols } if isinstance(self._data_source, DataSource): df = self._data_source.query( unique_syms, self._start_date, self._end_date, timeframe ) else: df = _between(self._data_source, self._start_date, self._end_date) df = df[df[DataCol.SYMBOL.value].isin(unique_syms)] if df.empty: raise ValueError("DataSource is empty.") return df.reset_index(drop=True) def _to_test_result( self, start_date: datetime, end_date: datetime, portfolio: Portfolio, calc_bootstrap: bool, train_only: bool, signals: Optional[dict[str, pd.DataFrame]], ) -> TestResult: if train_only: return TestResult( start_date=start_date, end_date=end_date, portfolio=pd.DataFrame(), positions=pd.DataFrame(), orders=pd.DataFrame(), trades=pd.DataFrame(), metrics=EvalMetrics(), metrics_df=pd.DataFrame(), bootstrap=None, signals=signals, stops=None, ) pos_df = pd.DataFrame.from_records( portfolio.position_bars, columns=PositionBar._fields ) for col in ( "close", "equity", "market_value", "margin", "unrealized_pnl", ): pos_df[col] = quantize(pos_df, col, self._config.round_test_result) pos_df.set_index(["symbol", "date"], inplace=True) portfolio_df = pd.DataFrame.from_records( portfolio.bars, columns=PortfolioBar._fields, index="date" ) for col in ( "cash", "equity", "margin", "market_value", "pnl", "unrealized_pnl", "fees", ): portfolio_df[col] = quantize( portfolio_df, col, self._config.round_test_result ) orders_df = pd.DataFrame.from_records( portfolio.orders, columns=Order._fields, index="id" ) for col in ("limit_price", "fill_price", "fees"): orders_df[col] = quantize( orders_df, col, self._config.round_test_result ) trades_df = pd.DataFrame.from_records( portfolio.trades, columns=Trade._fields, index="id" ) trades_df["bars"] = trades_df["bars"].astype(int) for col in ( "entry", "exit", "pnl", "return_pct", "agg_pnl", "pnl_per_bar", "mae", "mfe", ): trades_df[col] = quantize( trades_df, col, self._config.round_test_result ) shares_type = float if self._fractional_shares_enabled() else int pos_df["long_shares"] = pos_df["long_shares"].astype(shares_type) pos_df["short_shares"] = pos_df["short_shares"].astype(shares_type) orders_df["shares"] = orders_df["shares"].astype(shares_type) trades_df["shares"] = trades_df["shares"].astype(shares_type) eval_result = self.evaluate( portfolio_df=portfolio_df, trades_df=trades_df, calc_bootstrap=calc_bootstrap, bootstrap_sample_size=self._config.bootstrap_sample_size, bootstrap_samples=self._config.bootstrap_samples, bars_per_year=self._config.bars_per_year, ) metrics = [ (k, v) for k, v in dataclasses.asdict(eval_result.metrics).items() if v is not None ] metrics_df = pd.DataFrame(metrics, columns=["name", "value"]) stops_df = None if self._config.return_stops: stops_df = pd.DataFrame.from_records( portfolio._stop_records, columns=StopRecord._fields ) self._logger.walkforward_completed() return TestResult( start_date=start_date, end_date=end_date, portfolio=portfolio_df, positions=pos_df, orders=orders_df, trades=trades_df, metrics=eval_result.metrics, metrics_df=metrics_df, bootstrap=eval_result.bootstrap, signals=signals, stops=stops_df, )