Event-driven Python backtesting engine - Data - Part 3

# trading_engine/data.py

import pandas as pd

from abc import ABC, abstractmethod
from pathlib import Path

from events import MarketEvent
from utils import truncate_dfs


class DataHandler(ABC):
    """
    DataHandler is an abstract base class providing an interface for all subsequent inherited
    data handlers (live and historic).

    The goal of a derived data handler object is to output a generated set of bars (OHLCVI) for each
    symbol requested.

    This will replicate how a live strategy would function as current market data would be sent "down
    the pipe" (drip feed). An historic and live system will be treated identically in terms of data feed.
    """

    @abstractmethod
    def get_latest_bars(self, symbol, N=1):
        """
        Returns the last N bars from the latest_symbol list, or fewer if less bars are available.
        """
        raise NotImplementedError("Should implement get_latest_bars()")

    @abstractmethod
    def update_bars(self):
        """
        Pushes the latest bar to the latest symbol structure for all symbols in the symbol list
        """
        raise NotImplementedError("Should implement update_bars()")


class HistoricCsvDataHandler(DataHandler):
    """
    HistoricCsvDataHandler is designed to read CSV files for each requested symbol from disk and
    provide an interface to obtain the "latest" (in the context of historic csv data of course) bar
    in a manner identical to a live trading interface.
    """

    def __init__(self, events, csv_dir, symbol_list):
        """
        Initialises the historic data handler by requesting the location of the CSV files
        and a list of symbols.

        It will be assumed that all files are of the form 'symbol.csv', where symbol is a sting
        in the list.

        :param events: the Event queue -> Queue object
        :param csv_dir: directory path to the csv files -> Path or Path-like object
        :param symbol_list: a list of symbol strings -> str
        """

        self.events = events
        self.csv_dir = csv_dir
        self.symbol_list = symbol_list

        self.symbol_data = {}
        self.latest_symbol_data = {}
        self.continue_backtest = True

        self._open_convert_csv_files()

    def _open_convert_csv_files(self):
        """
        Opens the csv files from the data directory, converting them into pandas DataFrames
        within a symbol dictionary.
        """
      
        # Note: combining the index is particularly useful for mean-reverting strategies,
        # because it allows graceful handling of missing data
        comb_index = None
        for symbol in self.symbol_list:
            self.symbol_data[symbol] = pd.read_csv(
                Path.joinpath(self.csv_dir.joinpath(f"{symbol.lower()}.csv")),
                header=0,
                names=["datetime", "open", "high", "low", "close", "volume", "oi"]
            )

            # combine the index column to pad forward values
            if comb_index is None:
                comb_index = self.symbol_data[symbol].index
            else:
                comb_index.union(self.symbol_data[symbol].index)

            self.latest_symbol_data[symbol] = []  # initialise to empty list (for now)

        # reindex the data frames
        for symbol in self.symbol_list:
            self.symbol_data[symbol] = self.symbol_data[symbol].reindex(index=comb_index, method="pad").itertuples()

    def _get_new_bar(self, symbol):
        """
        Returns the latest bar from the data feed as a tuple of

        (symbol, datetime, open, high, low, close, volume)

        :param symbol: symbol to retrieve -> str
        """

        for bar in self.symbol_data[symbol]:
            yield tuple([
                symbol,
                bar.datetime,
                bar.open,
                bar.high,
                bar.low,
                bar.close,
                bar.volume
            ])

    def get_latest_bars(self, symbol, N=1):
        """
        Returns the last N bars from the latest_symbol list, or N - K if less are available.

        :param symbol: symbol to retrieve
        :param N: non-negative integer indicating number of bars to retrieve
        """

        try:
            bars_list = self.latest_symbol_data[symbol]
        except KeyError as e:
            print(f"Symbol {symbol} is not available in the historical data set.")
        else:
            return bars_list[-N:]

    def update_bars(self):
        """
        Pushes the latest bar to the latest_symbol data structure for all symbols in the list
        """

        for symbol in self.symbol_list:
            try:
                bar = self._get_new_bar(symbol).__next__()
            except StopIteration:
                self.continue_backtest = False
            else:
                if bar is not None:
                    self.latest_symbol_data[symbol].append(bar)

        self.events.put(MarketEvent())