Source code for qf_lib.backtesting.trading_session.backtest_trading_session

#     Copyright 2016-present CERN – European Organization for Nuclear Research
#     Licensed under the Apache License, Version 2.0 (the "License");
#     you may not use this file except in compliance with the License.
#     You may obtain a copy of the License at
#     Unless required by applicable law or agreed to in writing, software
#     distributed under the License is distributed on an "AS IS" BASIS,
#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#     See the License for the specific language governing permissions and
#     limitations under the License.
from typing import Union, Sequence

from import BacktestBroker
from qf_lib.backtesting.contract.contract_to_ticker_conversion.base import ContractTickerMapper
from qf_lib.backtesting.data_handler.data_handler import DataHandler
from import EventManager
from import Notifiers
from qf_lib.backtesting.monitoring.backtest_monitor import BacktestMonitor
from qf_lib.backtesting.monitoring.backtest_result import BacktestResult
from qf_lib.backtesting.order.order_factory import OrderFactory
from qf_lib.backtesting.orders_filter.orders_filter import OrdersFilter
from qf_lib.backtesting.portfolio.portfolio import Portfolio
from qf_lib.backtesting.position_sizer.position_sizer import PositionSizer
from qf_lib.backtesting.trading_session.trading_session import TradingSession
from qf_lib.common.enums.frequency import Frequency
from qf_lib.common.enums.price_field import PriceField
from qf_lib.common.tickers.tickers import Ticker
from qf_lib.common.utils.dateutils.relative_delta import RelativeDelta
from qf_lib.common.utils.dateutils.timer import SettableTimer
from qf_lib.common.utils.logging.qf_parent_logger import qf_logger
from qf_lib.common.utils.miscellaneous.to_list_conversion import convert_to_list
from qf_lib.containers.helpers import compute_container_hash

[docs]class BacktestTradingSession(TradingSession): """ Encapsulates the settings and components for carrying out a backtest session. Pulls for data every day. """ def __init__(self, contract_ticker_mapper: ContractTickerMapper, start_date, end_date, position_sizer: PositionSizer, orders_filters: Sequence[OrdersFilter], data_handler: DataHandler, timer: SettableTimer, notifiers: Notifiers, portfolio: Portfolio, events_manager: EventManager, monitor: BacktestMonitor, broker: BacktestBroker, order_factory: OrderFactory, frequency: Frequency, backtest_result: BacktestResult): """ Set up the backtest variables according to what has been passed in. The data_provider parameter of the BacktestTradingSession points to a Data Handler object. """ super().__init__() self.logger = qf_logger.getChild(self.__class__.__name__) self.contract_ticker_mapper = contract_ticker_mapper self.start_date = start_date self.end_date = end_date self.notifiers = notifiers self.event_manager = events_manager self.data_handler = data_handler self.data_provider = data_handler # type: DataHandler self.portfolio = portfolio self.position_sizer = position_sizer self.orders_filters = orders_filters self.monitor = monitor self.timer = timer self.order_factory = order_factory = broker self.frequency = frequency self.backtest_result = backtest_result self._hash_of_data_bundle = None def use_data_preloading(self, tickers: Union[Ticker, Sequence[Ticker]], time_delta: RelativeDelta = None): if time_delta is None: time_delta = RelativeDelta(years=1) data_start = self.start_date - time_delta # The tickers and price fields are sorted in order to always return the same hash of the data bundle for # the same set of tickers and fields tickers, _ = convert_to_list(tickers, Ticker) self.data_handler.use_data_bundle(sorted(tickers), sorted(PriceField.ohlcv()), data_start, self.end_date, self.frequency) self._hash_of_data_bundle = compute_container_hash(self.data_handler.data_provider.data_bundle)"Preloaded data hash value {}".format(self._hash_of_data_bundle))
[docs] def get_preloaded_data_checksum(self) -> str: """ Returns the checksum value computed as a hexadecimal digest on the preloaded data bundle. Returns ------- str checksum of the preloaded data bundle """ if self._hash_of_data_bundle is not None: return self._hash_of_data_bundle else: raise ValueError("Not able to compute checksum of data bundle. The data has not been preloaded yet.")
[docs] def verify_preloaded_data(self, expected_checksum: str): """ Verifies if the checksum computed on the preloaded data bundle is equal to the expected value. In case of differences ValueError is raised. Parameters ----------- expected_checksum: str The expected checksum of the data bundle. """ if self._hash_of_data_bundle is None: raise ValueError("Not able to compute checksum of data bundle. The data has not been preloaded yet.") elif self._hash_of_data_bundle != expected_checksum: raise ValueError("Data preloading was not successful. The expected checksum does not match the actual " "value.")