http://habrahabr.ru/company/itinvest/blog/264141/



В предыдущей статье мы поговорили о том, что такое событийно-ориентированная система бэктестинга и разобрали иерархию классов, которую необходимо для нее разработать. Сегодня речь пойдет о том, как подобные системы используют рыночные данные как в контексте исторического тестирования, так и для «живой» работы на бирже.

Работа с рыночными данными

Одной из задач при создании событийно ориентированной торговой системы является минимизация необходимости писать разный код для одних и тех же задач в контексте тестирования на исторических данных и для реальной торговли. В идеале, следует использовать единую методологию генерации сигналов и управления портфолио для каждого из этих случаев. Чтобы этого добиться, объект Strategy, который генерирует торговые сигналы (Signals), и объект Portfolio, который на их основе генерирует ордера (Orders), должны использовать один интерфейс доступа к рыночным данным как в контексте исторического тестирования, так и работы в реальном времени.

Именно эта необходимость привела к появлению концепции иерархии классов, основанной на объекте DataHandler, который предоставляет подклассам интерфейс для передачи рыночных данных остальным компонентам системы. В такой конфигурации обработчик любого подкласса можно просто «выбросить», и это никак не скажется на работе компонентов, отвечающих за стратегию и обработку портфолио.

Среди таких подклассов могут быть HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler и так далее. Здесь мы рассмотрим только создание обработчика CSV с историческими данными, который будет загружать соответствующий CSV-файл финансовых данных внутри дня в формате баров (значения цены Low, High, Close, а также объем торгов Volume и открытый интерес OpenInterest). На основе этих данных при каждом «ударе сердца» системы (heartbeat) можно уже проводить углубленный анализ компонентами Strategy и Portfolio, что позволит избежать различных искажений.

На первом шаге нужно импортировать требуемые библиотеки, в частности pandas и abstract base class. Поскольку DataHandler генерирует события MarketEvents, нужно также импортировать и event.py:

# data.py

import datetime
import os, os.path
import pandas as pd

from abc import ABCMeta, abstractmethod

from event import MarketEvent

DataHandler — это абстрактный базовый класс (АБК), что означает невозможность создания экземпляра напрямую. Это можно сделать только с помощью подклассов. Обоснование этого заключается в том, что АБК, предоставляет интерфейс для подлежащих подклассов DataHandler, который они должны использовать, что позволяет добиться совместимости с другими классами, с которыми может осуществляться взаимодействие.

Чтобы Python «понял», что имеет дело с абстрактным базовым классом, мы будем использовать свойство _metaclass_. Также с помощью декоратора @abstractmethod указывается, что метод будет переопределен в подклассах (в точности аналогично полностью виртуальному методу в C++).

Два интересующих нас метода — это get_latest_bars и update_bars. Первый из них возвращает последние N баров из текущей временной метки «удара сердца» системы, что полезно для осуществления вычислений для классов Strategy. Последний метод предоставляет механизм анализа для наложения информацию бара на новую структуру данных, что полностью позволяет избавиться от прогнозных искажений. Если произойдет попытка созданий экземпляра класса, возникнет исключение:

# data.py

class DataHandler(object):
    """
    DataHandler — абстрактный базовый класс, предоставляющий интерфейс для всех наследованных обработчиков (для живой торговли и работы с историческими данными)

Цель (выделенного) объекта DataHandler заключается в выводе сгенерированного набора баров (OLHCVI) для каждого запрощенного финансового инструмента.

Это нужно для получения понимания о том, как будет функционировать стратегия, при использовании реальных торговых данных. Таким образом реальная и историческая система во всем наборе инструментов бэктестинга рассматриваются одинаково.
    """

    __metaclass__ = ABCMeta

    @abstractmethod
    def get_latest_bars(self, symbol, N=1):
        """
        Возвращает последние N баров из списка    latest_symbol или меньше, если столько баров еще недоступно.
        """
        raise NotImplementedError("Should implement get_latest_bars()")

    @abstractmethod
    def update_bars(self):
        """
        Накладывает последний бар на последнюю структуру инструмента для всех инструментов в списке.
        """
        raise NotImplementedError("Should implement update_bars()")

После описания класса DataHandler следующим шагом является создание обработчика для исторических CSV-файлов. HistoricCSVDataHandler будет брать множество CSV-файлов (по одному для каждого финансового инструмента) и конвертировать их в словарь фреймов DataFrames для pandas.

Обработчику нужно несколько параметров — очередь событий (Event Queue), в которую публиковать рыночную информацию MarketEvent, абсолютный путь к CSV-файлам и список инструментов. Вот так выглядит инициализация класса:

# data.py

class HistoricCSVDataHandler(DataHandler):
    """
    HistoricCSVDataHandler создан для чтения CSV-файло с диска и создания интерфейса для получения «последнего» бара, как при реальной торговле.

    """

    def __init__(self, events, csv_dir, symbol_list):
        """
        Инициализирует обработчик исторических данных запросом местоположения CSV-файлов и списка инструментов.

Предполагается, что все файлы имеют форму  'symbol.csv', где symbol — это строка списка.


        Параметры:
        events - очередь событий.
        csv_dir - Абсолютный путь к директории с CSV-файлами.
        symbol_list - Список строк инструментов.
        """
        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()

Он будет пытаться открыть файлы в формате “SYMBOL.csv”, в которым SYMBOL — это тикер инструмента. Использованный здесь формат совпадает с предлагаемым поставщиком данных DTN IQFeed, но его легко можно модифицировать для работы с другими форматами. Открытие файлов обрабатывается методом _open_convert_csv_files.

Одно из преимуществ использования пакета pandas для хранения данных внутри HistoricCSVDataHandler заключается в том, что индексы всех отслеживаемых инструментов можно слить воедино. Это позволяет интерполировать даже отсутствующие данные, что полезно для побарового сравнения инструментов (бывает нужно в стратегиях mean reversion). При комбинировании индексов для инструментов используются методы union и reindex:

# data.py

    def _open_convert_csv_files(self):
        """
       Открывает CSV-файлы из директории, конвертирует их в pandas DataFrames внутри словаря инструментов.

Для данного обработчика предположим, что данные берутся из фида DTN IQFeed, и работа идет с этим форматом.
        """
        comb_index = None
        for s in self.symbol_list:
            # Загрузка CSV-файла без заголовочной информации, индексированный по дате

            self.symbol_data[s] = pd.io.parsers.read_csv(
                                      os.path.join(self.csv_dir, '%s.csv' % s),
                                      header=0, index_col=0, 
                                      names=['datetime','open','low','high','close','volume','oi']
                                  )

            # Комбинируется индекс для «подкладывания» значений
            if comb_index is None:
                comb_index = self.symbol_data[s].index
            else:
                comb_index.union(self.symbol_data[s].index)

            # Set the latest symbol_data to None
            self.latest_symbol_data[s] = []

        # Reindex the dataframes
        for s in self.symbol_list:
            self.symbol_data[s] = self.symbol_data[s].reindex(index=comb_index, method='pad').iterrows()

Метод _get_new_bar создает генератор для создания форматированной версии данных в барах. Это означается, что последующие вызовы метода результируются в новом баре (и так до того момента, пока не будет достигнут конец строки данных по инструментам):

# data.py

    def _get_new_bar(self, symbol):
        """
        Возвращает последний бар из дата-фида в формате: 
        (sybmbol, datetime, open, low, high, close, volume).
        """
        for b in self.symbol_data[symbol]:
            yield tuple([symbol, datetime.datetime.strptime(b[0], '%Y-%m-%d %H:%M:%S'), 
                        b[1][0], b[1][1], b[1][2], b[1][3], b[1][4]])

Первый абстрактный метод из DataHаndler, который нужно реализовать — это get_latest_bars. Он просто выводит список последних N баров из структуры latest_symbol_data. Установка N = 1 позволяет получать текущий бар:

# data.py

    def get_latest_bars(self, symbol, N=1):
        """
        Возвращает N последних баров из списка latest_symbol, или N-k, если доступно меньше.

        """
        try:
            bars_list = self.latest_symbol_data[symbol]
        except KeyError:
            print "That symbol is not available in the historical data set."
        else:
            return bars_list[-N:]  

Последний метод — update_bars, это второй абстрактный метод из DataHandler. Он генерирует события (MarketEvent), которые попадают в очередь, как последние бары добавляются в latest_symbol_data:

# data.py

    def update_bars(self):
        """
        Отправляет последний бар в структуру данных инструментов для всех инструментов в списке.
"""
        for s in self.symbol_list:
            try:
                bar = self._get_new_bar(s).next()
            except StopIteration:
                self.continue_backtest = False
            else:
                if bar is not None:
                    self.latest_symbol_data[s].append(bar)
        self.events.put(MarketEvent())

Таким образом, у нас есть DataHandler — выделенный объект, который используется остальными компонентами системы для отслеживания рыночных данных. Для работы объектам Stragety, Portfolio и ExecutionHandler требуется текущая рыночная информация, поэтому имеет смысл работать с ней централизованно, чтобы избежать возможного дублировани хранения.

От информации до торгового сигнала: стратегия

Объект Strategy инкапсулирует все вычисления, связанные с обработкой рыночных данных, для создания рекомендательных сигналов объекту Portfolio. На этой стадии разработки событийно ориентированного бэктестера нет понятий индикаторов или фильтров, которые используются в техническом анализе. Для их реализации можно создать отдельную структуру данных, но это уже выходит за рамки данной статьи.

Иерархия стратегии относительно проста — она состоит из абстрактного базового класса с единственным виртуальным методом для создания объектов SignalEvents. Для создания иерархии стратегии необходимо импортировать NumPy, pandas, объект Queue, инструмент abstract base tools и SignalEvent:

# strategy.py

import datetime
import numpy as np
import pandas as pd
import Queue

from abc import ABCMeta, abstractmethod

from event import SignalEvent

Абстрактный базовый класс Strategy определяет виртуальный метод calculate_signals. Он используется для обработки создания объектов SignalEvent на основе обновлений рыночных данных:

# strategy.py

class Strategy(object):
    """
    Strategy — абстрактный базовый класс, предоставляющий интерфейс для подлежащих (наследованных) объектов для обработки стратегии.


    Цель выделенного объекта Strategy заключается в генерировании сигнальных объектов для конкретных инструментов на основе входящих баров (OLHCVI), сгенерированных объектом DataHandler.

    Эту конфигурацию можно использовать как для работы с историческими данными, так и для работы на реальном рынке — объект Strategy не зависит от от источника данных, он получает бары из очереди.

    """

    __metaclass__ = ABCMeta

    @abstractmethod
    def calculate_signals(self):
        """
        Предоставляет механизмы для вычисления списка сигналов.

        """
        raise NotImplementedError("Should implement calculate_signals()")

Определение абстрактного базового класса Strategy довольно проста. Первый пример использования подклассов в объекте Strategy заключается в использовании стратегий buy и hold и создании соответствующего класса BuyAndHoldStrategy. Он будет покупать конкретную акцию в определенный день и удерживает позицию. Таким образом на одну акцию генерируется только один сигнал.

Конструктор (__init__) требует наличия обработчика рыночных данных bars и объекта очереди событий events:

# strategy.py

class BuyAndHoldStrategy(Strategy):
    """
    Крайне простая стратегия, которая входит в длинную позициию при полуении бара и никогда из нее не выходит.

    Используется в качестве механизма тестирования класса Strategy и бенчмарка для сравнения разных стратегий.
    """

    def __init__(self, bars, events):
        """
        Инициализирует стратегию buy and hold.

        Параметры:
        bars - Объект DataHandler, который предоставляет информацию о барах
        events - Объект очереди событий.
        """
        self.bars = bars
        self.symbol_list = self.bars.symbol_list
        self.events = events

        # Когда получен сигнал на покупку и удержание акции, устанавливается в True 
        self.bought = self._calculate_initial_bought()

При инициализации стратегии BuyAndHoldStrategy в словаре bought содержится набор ключей для каждого инструмента, которые установлены в False. Когда определенный инструмент покупается (открывается длинная позиция), то ключ переводится в положение True. Это позволяет объекту Strategy понимать, открыта ли позиция:

# strategy.py

    def _calculate_initial_bought(self):
        """
        Добавляются ключи в словарь bought и устанавливаются в False.
        """
        bought = {}
        for s in self.symbol_list:
            bought[s] = False
        return bought

Виртуальный метод calculate_signals имплементирован именно в этом классе. Метод проходит по всем инструментам в списке и получает последний бар из обработчика bars. Затем он проверяет, был ли инструмент «куплен» (находимся ли мы в рынке по нему, или нет), а затем создается сигнальный объект SignalEvent. Затем он помещается в очередь событий, а словарь bought обновляется соответствующей информацией (True для купленного инструмента):

# strategy.py

    def calculate_signals(self, event):
        """
       Для "Buy and Hold" генерируем один сигнал на инструмент. Это значит, что мы только открываем длинные позиции с момента инициализации стратегии.

        Параметры:
        event - Объект MarketEvent. 
        """
        if event.type == 'MARKET':
            for s in self.symbol_list:
                bars = self.bars.get_latest_bars(s, N=1)
                if bars is not None and bars != []:
                    if self.bought[s] == False:
                        # (Symbol, Datetime, Type = LONG, SHORT or EXIT)
                        signal = SignalEvent(bars[0][0], bars[0][1], 'LONG')
                        self.events.put(signal)
                        self.bought[s] = True

Это очень простая стратегия, но ее достаточно для того, чтобы продемонстрировать природу иерархии событийно ориентированной стратегии. В следующей статьей мы рассмотрим более сложные стратегии, например, парную торговлю. Также в следующей статье речь пойдет о создании иерархии Portfolio, которая будет отслеживать прибыль и убыток по позициям (profit and loss, PnL).

Продолжение следует…

P. S. Ранее в нашем блоге на Хабре мы уже рассматривали различные этапы разработки торговых систем. Есть и онлайн-курсы по данной тематике.