В обсуждении реализации обратного тестирования, основанного на событиях, ранее рассматривалась петля событий, иерархия классов событий и компонент обработки данных.
Объект "Стратегия" включает в себя все расчеты по рыночным данным, которые генерируют консультативные сигналы к объекту "Портфель". На данном этапе в разработке событийного бэкстестера нет концепции индикатора или фильтра, таких как те, которые встречаются в технической торговле.
Иерархия стратегии относительно проста, поскольку она состоит из абстрактного базового класса с одним чистым виртуальным методом для генерации объектов SignalEvent.
# strategy.py
import datetime
import numpy as np
import pandas as pd
import Queue
из abc import ABCMeta, абстрактный метод
от импорта события SignalEvent Абстрактный базовый класс Strategy просто определяет чистый виртуальный метод calculate_signals. В производных классах он используется для обработки генерации объектов SignalEvent на основе обновлений данных рынка:
# strategy.py
class Strategy(object):
"""
Strategy is an abstract base class providing an interface for
all subsequent (inherited) strategy handling objects.
The goal of a (derived) Strategy object is to generate Signal
objects for particular symbols based on the inputs of Bars
(OLHCVI) generated by a DataHandler object.
This is designed to work both with historic and live data as
the Strategy object is agnostic to the data source,
since it obtains the bar tuples from a queue object.
"""
__metaclass__ = ABCMeta
@abstractmethod
def calculate_signals(self):
"""
Provides the mechanisms to calculate the list of signals.
"""
raise NotImplementedError("Should implement calculate_signals()")
Определение "Стратегия ABC" простое. Наш первый пример подразделения объекта "Стратегия" использует стратегию покупки и удержания для создания класса BuyAndHoldStrategy. Это просто длится в определенной ценной бумаге в определенную дату и держит ее в портфеле. Таким образом, всегда генерируется только один сигнал на ценную бумагу.
Конструктор (Инит) требует обработки данных рынка баров и объекта очереди событий событий:
# strategy.py
class BuyAndHoldStrategy(Strategy):
"""
This is an extremely simple strategy that goes LONG all of the
symbols as soon as a bar is received. It will never exit a position.
It is primarily used as a testing mechanism for the Strategy class
as well as a benchmark upon which to compare other strategies.
"""
def __init__(self, bars, events):
"""
Initialises the buy and hold strategy.
Parameters:
bars - The DataHandler object that provides bar information
events - The Event Queue object.
"""
self.bars = bars
self.symbol_list = self.bars.symbol_list
self.events = events
# Once buy & hold signal is given, these are set to True
self.bought = self._calculate_initial_bought()
При инициировании BuyAndHoldStrategy у участника купленного словаря есть набор ключей для каждого символа, которые все настроены на False. Как только актив был
# strategy.py
def _calculate_initial_bought(self):
"""
Adds keys to the bought dictionary for all symbols
and sets them to False.
"""
bought = {}
for s in self.symbol_list:
bought[s] = False
return bought
Метод calculate_signals чисто виртуальный реализован конкретно в этом классе. Метод перемещается по всем символам в списке символов и извлекает последнюю строку из обработчика данных строк. Затем он проверяет, был ли этот символ куплен (т.е. находимся ли мы на рынке для этого символа или нет), и если нет, создает один объект SignalEvent. Это затем помещается в очередь событий, и купленный словарь правильно обновляется до True для этого конкретного ключа символа:
# strategy.py
def calculate_signals(self, event):
"""
For "Buy and Hold" we generate a single signal per symbol
and then no additional signals. This means we are
constantly long the market from the date of strategy
initialisation.
Parameters
event - A MarketEvent object.
"""
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
В следующей статье мы рассмотрим, как создать иерархию портфеля, которая отслеживает наши позиции с прибылью и убытком (PnL).