Source code for qf_lib.backtesting.trading_session.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
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#     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 abc import ABCMeta
from typing import Sequence

from qf_lib.backtesting.broker.broker import Broker
from qf_lib.backtesting.contract.contract_to_ticker_conversion.base import ContractTickerMapper
from qf_lib.backtesting.events.event_manager import EventManager
from qf_lib.backtesting.events.notifiers import Notifiers
from qf_lib.backtesting.monitoring.abstract_monitor import AbstractMonitor
from qf_lib.backtesting.order.order_factory import OrderFactory
from qf_lib.backtesting.orders_filter.orders_filter import OrdersFilter
from qf_lib.backtesting.position_sizer.position_sizer import PositionSizer
from qf_lib.common.enums.frequency import Frequency
from qf_lib.common.utils.dateutils.timer import Timer
from qf_lib.common.utils.logging.qf_parent_logger import qf_logger
from qf_lib.data_providers.data_provider import DataProvider
from qf_lib.settings import Settings


[docs]class TradingSession(metaclass=ABCMeta): """ Base class for all Trading Sessions. It configures all the elements of the trading environment. """ def __init__(self): self.trading_session_name = None # type: str self.settings = None # type: Settings self.data_provider = None # type: DataProvider self.timer = None # type: Timer self.monitor = None # type: AbstractMonitor self.broker = None # type: Broker self.contract_ticker_mapper = None # type: ContractTickerMapper self.order_factory = None # type: OrderFactory self.position_sizer = None # type: PositionSizer self.orders_filters = [] # type: Sequence[OrdersFilter] self.event_manager = None # type: EventManager self.notifiers = None # type: Notifiers self.frequency = None # type: Frequency self.logger = qf_logger.getChild(self.__class__.__name__)
[docs] def start_trading(self) -> None: """ Carries out an while loop that processes incoming events. The loop continues until the EndTradingEvent occurs (e.g. no more data for the backtest). """ self.logger.info("Trading Session - start trading...") while self.event_manager.continue_trading: self.event_manager.dispatch_next_event() self.logger.info("Trading Session - trading finished...") self.monitor.end_of_trading_update()
@staticmethod def _create_event_manager(timer: Timer, notifiers: Notifiers): event_manager = EventManager(timer) event_manager.register_notifiers([ notifiers.all_event_notifier, notifiers.empty_queue_event_notifier, notifiers.end_trading_event_notifier, notifiers.scheduler ]) return event_manager