Бэк-тестирование - ключевой компонент алгоритмической торговли
Машинное обучение с алгоритмом классификации и алгоритмической торговой стратегией
Алгоритм классификации и обратное тестирование с данными природного газа
Https://sarit-maitra.medium.com/membership
Бэк-тестирование в форме тестирования производительности - одна из самых сложных задач, входящих в систему исследования алгоритмической торговли. Существует несколько факторов, таких как задержки программного обеспечения, задержки в сети, проскальзывания, комиссии и т. Д., Которые используются для построения алгоритмической торговой структуры. Основная идея заключается в том, что, учитывая исторические данные, какова будет производительность торговой стратегии.
Здесь мы рассмотрим, как мы можем использовать алгоритм машинного обучения для прогнозирования будущего направления и определения стратегии торговли. Кроме того, как только стратегия определена, мы проведем тестирование производительности в форме предварительной проверки, чтобы определить силу определенной стратегии.
Мы использовали ежедневные данные о природном газе, полученные от Управления энергетической информации (EIA) Федеральной статистической системы США.
Функция для извлечения данных:
def NaturalGasData():
NaturalGas = quandl.get("CHRIS/CME_NG1", authtoken = "token")
NaturalGas = NaturalGas.loc['2010-01-01':,]
NaturalGas = NaturalGas [['Open', 'High', 'Low', 'Last', 'Settle', 'Volume']].copy()
NaturalGas.rename(columns =
{
'Last': 'Close',
'Settle': 'Adj Close'
}, inplace=True)
return NaturalGas
NaturalGas = NaturalGasData()
print(NaturalGas)
Подготовка данных:
Ниже мы создали несколько функций. Наша простая торговая стратегия - покупать по низкой цене и продавать по высокой. Для этого мы рассчитали разницу в цене закрытия между двумя последовательными днями. Сигнал генерируется, если значение отрицательное, означает, что текущая цена ниже, чем в предыдущий день, и система генерирует сигнал на покупку. Если это значение положительное, это означает сигнал на продажу. Следовательно, у нас 0, когда нам нужно купить, и 1, когда нам нужно продать.
Функция для определения функций:
def ClassificationCondition():
df = NaturalGas.copy()
df['closeReturn'] = df['Close'].pct_change()
df['highReturn'] = df['High'].pct_change()
df['lowReturn'] = df['Low'].pct_change()
df['volReturn'] = df['Volume'].pct_change()
df['dailyChange'] = (df['Close'] - df['Open']) / df['Open']
df['priceDirection'] = (df['Close'].shift(-1) - df['Close'])
df = df.replace([np.inf, -np.inf], np.nan)
df['signal'] = 0.0
df['signal'] = np.where((df.loc[:,'priceDirection'] > 0), 1.0, 0.0)
df.drop(columns =['Open', 'High', 'Low', 'Close', 'Adj Close', 'Volume', 'priceDirection' ], axis=1, inplace=True)
df.dropna(inplace=True)
return df
df = ClassificationCondition()
print(df)
Переменные X, y:
def features():
X = df.drop( columns = ['signal'], axis=1)
y = df.signal
return X,y
X,y = features()
# store machine learning data frame
MLDataFrame = NaturalGas.tail(len(X))[['Open', 'High', 'Low','Close', 'Volume']]
print(MLDataFrame) # sanity check
sns.countplot(x = 'signal', data=pd.DataFrame(y), hue='signal') plt.show()
Мы видим, что наши классы (0,1) почти равны.
Тренировочные и тестовые наборы:
Мы использовали перекрестную проверку, чтобы избежать переобучения. Мы имеем дело с данными временных рядов. Здесь мы ввели промежутки между обучающим и тестовым набором, чтобы смягчить временную зависимость. Здесь функция разделения разбивает данные на Kfold.
gkcv = GapKFold(n_splits=10, gap_before=2, gap_after=1)
for trainIndex, testIndex in gkcv.split(X):
# print("TRAIN:", trainIndex, "TEST:", testIndex)
xTrain, xTest = X.values[trainIndex], X.values[testIndex];
yTrain, yTest = y.values[trainIndex], y.values[testIndex];
# print('Observations: %d' % (len(xTrain) + len(xTest)))
print('Training Observations: %d' % (len(xTrain)))
print('Testing Observations: %d' % (len(xTest)))
Прогнозные модели с использованием машинного обучения:
Мы поэкспериментировали с несколькими алгоритмами классификации и обнаружили, что логистическая регрессия наиболее эффективна при данных условиях.
RF = RandomForestClassifier(bootstrap=True, ccp_alpha=0.0, class_weight="balanced", max_leaf_nodes=None, min_impurity_decrease=0.0, min_impurity_split=None, min_samples_leaf=1, min_samples_split=2, min_weight_fraction_leaf=0.0, n_estimators=1000, random_state=0, verbose=0, warm_start=False)
LR = LogisticRegression(penalty='l2',random_state = 0)
ensembleClassifiers = VotingClassifier(estimators=[
('RandomForest', RF), ('LogReg', LR)],voting='hard').fit(X, y)
ensembleScore = cross_val_score(ensembleClassifiers, X, y, cv=gkcv).mean()
print(ensembleScore)
LR = LogisticRegression(penalty='l2',random_state = 0).fit(X,y)
print(cross_val_score(LR, X, y, cv=gkcv).mean())
Наконец, прогноз был выполнен, как показано ниже. У нас есть -1, когда нам нужно купить, и 1, когда нам нужно продать.
ONNX время выполнения
initialType = [('floatInput', FloatTensorType([None, X.shape[1]]))]
onx = convert_sklearn(LR, initial_types=initialType)
with open("LRModel.onnx", "wb") as f:
f.write(onx.SerializeToString())
session = rt.InferenceSession("LRModel.onnx")
inputName = sess.get_inputs()[0].name
labelName = sess.get_outputs()[0].name
predictions = DataFrame(session.run([labelName], {inputName: X.values.astype(np.float32)})[0])
predictions.rename({0: 'PredictedSignal'}, axis = 'columns', inplace=True)
predictions.index = MLDataFrame.index
# print(predictions)
MLDataFrame = pd.concat([MLDataFrame, predictions], 1)
# print(MLDataFrame)
Можно заметить, что я использовал sklearn-onnx для преобразования модели в формат ONNX, который затем можно использовать для вычисления прогнозов с выбранным нами сервером. Scikit-learn и его зависимости (Python, numpy scipy) накладывают большие накладные расходы на память и хранилище. Хотя обычно это не проблема во время обучения модели, это может быть препятствием при развертывании функции прогнозирования одной обученной модели на целевой платформе. Подробнее об этом можно прочитать здесь.
MLDataFrame['PredictedSignal']= MLDataFrame['PredictedSignal'].diff(periods=1) print(MLDataFrame)
Однако не стоит постоянно покупать или продавать, когда рынок продолжает двигаться вниз или вверх. Поэтому, чтобы ограничить покупку / продажу, применил diff () к столбцу PredictedSignal. Мы можем быстро взглянуть на количество сигналов, сгенерированных моим алгоритмом классификации.
График сигналов на покупку / продажу (Торговая стратегия):
ts = MLDataFrame['2021-01-01':]
print(ts.PredictedSignal.value_counts())
# Buy/Sell signals plot
buys = ts.loc[ts["PredictedSignal"] == 1];
sells = ts.loc[ts["PredictedSignal"] == -1];
# Plot
fig = plt.figure(figsize=(20, 5));
plt.plot(ts.index, ts['Close'], lw=2., label='Price');
plt.plot(buys.index, ts.loc[buys.index]['Close'], '^', markersize=10, color='red', lw=2., label='Buy');
# down arrow when we sell one share
plt.plot(sells.index, ts.loc[sells.index]['Close'], 'v', markersize = 10, color='green', lw=2., label='Sell');
plt.ylabel('Price (USD)'); plt.xlabel('Date');
plt.title('Buy and Sell signals since Jan 2021'); plt.legend(loc='best');
plt.grid(True)
plt.show()
- стрелка вверх, когда мы покупаем один
- стрелка вниз, когда мы продаем один
Теперь, когда мы разработали нашу стратегию, давайте углубимся в тестирование на истории, чтобы получить статистику, показывающую, насколько эффективна наша принятая стратегия.
Класс для определения столбцов OHLCV:
from __future__ import (absolute_import, division, print_function,
unicode_literals)
...
prices = MLDataFrame['2021-01-01':].copy()
OHLCV = ['Open', 'High', 'Low', 'Close', 'Volume']
...
# class to define the columns we will provide
class NaturalGasData(PandasData):
"""Define pandas DataFrame structure"""
cols = OHLCV + ['PredictedSignal']
lines = tuple(cols)
# define parameters
params = {c: -1 for c in cols}
params.update({'datetime': None})
params = tuple(params.items())
Класс для определения стратегии тестирования:
Мы автоматизировали стоп-лосс и стоп-лосс на основе процента от исходной цены покупки: стоп-лосс - это мера ограничения риска, и есть много способов установить стоп-лосс на основе торговой стратегии. Вот упрощенный вариант:
- если текущая цена на 10% ниже первоначальной цены покупки, продайте акции
- если текущая цена на 20% выше первоначальной цены покупки, продайте акции
Выше приведены для иллюстрации, а не для продуманного стратегического результата. В следующем коде показан класс TradingStrategy.
class NGStrategy(bt.Strategy):
params = (('percents', 0.9),
('stopLoss', 0.10),
('stopWin', 0.20)
)
def __init__(self):
# import ipdb; ipdb.set_trace()
for handler in logging.root.handlers[:]:
logging.root.removeHandler(handler)
logging.basicConfig(format='%(message)s', level = logging.CRITICAL, handlers=[logging.FileHandler("LOG.log"), logging.StreamHandler()])
self.startCash = self.broker.getvalue()
date = self.data.datetime.date()
close = prices.Close[0]
print('{}: Close: ${}, Position Size: {}'.format(date, close, self.position.size))
self.data_PredictedSignal = self.datas[0].PredictedSignal
self.data_Open = self.datas[0].Open
self.data_Close = self.datas[0].Close
# keep track of pending orders/buy price/buy commission
self.order = None
self.price = None
self.stop_price = None
self.comm = None
self.trade = None
self.buystop_order = None
self.sellstop_order = None
self.qty = 1
def log(self, txt, doprint=True):
dt = self.datas[0].datetime.date(0)
print('{0},{1}'.format(dt.isoformat(), txt))
# control structure
def notify_order(self, order):
date = self.data.datetime.date()
# If order is submitted/accepted, do nothing
if order.status in [order.Submitted, order.Accepted]:
return
# If order is buy/sell executed, report price executed
if order.status == order.Completed:
if order.isbuy():
self.log('BUY@ Price: {0:8.2f}, Cost: {1:8.2f}, Comm: {2:8.2f}'.format(order.executed.price, order.executed.value, order.executed.comm))
else:
self.log('SELL@ Price: {0:8.2f}, Cost: {1:8.2f}, Comm{2:8.2f}'.format(order.executed.price, order.executed.value,
order.executed.comm))
self.bar_executed = len(self) # when was trade executed
# If order is canceled/margin/rejected, report order canceled
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
"""When system receives a buy or sell signal, we can instruct it to create an order. However, that order won’t be executed until the next bar is called, at whatever price that may be"""
if order.status in [order.Completed]:
if order.isbuy():
if(self.params.stopLoss):
self.stopOrder = self.sell(price = order.executed.price,exectype = bt.Order.StopTrail, trailpercent = self.params.stopLoss)
if(self.params.stopWin):
self.stopOrder = self.sell(price=(order.executed.price*(1+self.params.stopWin)),exectype=bt.Order.Limit,oco=self.stopOrder)
# set no pending order
self.order = None
def notify_trade(self, trade):
date = self.data.datetime.date()
if not trade.isclosed:
return
self.log(f'OPERATION RESULT --- Gross: {trade.pnl:.2f}, Net: {trade.pnlcomm:.2f}')
def next_open(self):
if self.order: # check if order is pending, if so, then break out
return
# since there is no order pending, are we in the market?
if not self.position:
if self.data_PredictedSignal == 1:
size = int(self.broker.getcash() / self.datas[0].Open)
self.buy(size=size)
else:
if self.data_PredictedSignal == -1:
# sell order
self.log(f'SELL CREATED --- Size: {self.position.size}')
self.sell(size = self.position.size)
class Test(unittest.TestCase):
def setUp(self):
self.trading_strategy = NGStrategy()
Тестирование торговой стратегии имеет решающее значение. Нам нужно проверить, будет ли торговая стратегия выставлять правильные ордера.
Функция печати результатов технического анализа:
def printTradeAnalysis(analyzer):
# Get the results we are interested in
totalOpen = analyzer.total.open
totalClosed = analyzer.total.closed
totalWon = analyzer.won.total
totalLost = analyzer.lost.total
winStreak = analyzer.streak.won.longest
loseStreak = analyzer.streak.lost.longest
pnlNet = round(analyzer.pnl.net.total,2)
strikeRate = (totalWon / totalClosed) * 100
# Designate the rows\
a = ['Total Open', 'Total Closed', 'Total Won', 'Total Lost']
b = ['Strike Rate','Win Streak', 'Losing Streak', 'PnL Net']
c = [totalOpen, totalClosed,totalWon,totalLost]
d = [strikeRate, winStreak, loseStreak, pnlNet]
# Check which set of headers is the longest.
if len(a) > len(b):
header_length = len(a)
else:
header_length = len(b)
# Print the rows
print_list = [a,b,c,d]
row_format ="{:<20}" * (header_length + 1)
print("Trade Analysis Results:")
for row in print_list:
print(row_format.format('',*row))
def printSQN(analyzer):
sqn = round(analyzer.sqn,2)
print('SQN: {}'.format(sqn))
Создать экземпляр NaturalGasData:
data = NaturalGasData(dataname=prices)
def runStrategy():
# Variable for our starting cash
startCash = 5000.0
NGtrade = bt.Cerebro(stdstats = False, cheat_on_open=True, maxcpus=1)
NGtrade.addstrategy(NGStrategy)
NGtrade.addanalyzer(bt.analyzers.PyFolio, _name='pyfolio')
# Add the analyzers we are interested in
NGtrade.addanalyzer(bt.analyzers.TradeAnalyzer, _name="ta")
NGtrade.addanalyzer(bt.analyzers.SQN, _name="sqn")
NGtrade.addobserver(bt.observers.Value)
NGtrade.addanalyzer(bt.analyzers.SharpeRatio, riskfreerate=0.0)
NGtrade.addanalyzer(bt.analyzers.Returns)
NGtrade.addanalyzer(bt.analyzers.DrawDown)
NGtrade.adddata(data)
# Set desired cash start
NGtrade.broker.setcash(startCash)
NGtrade.broker.setcommission(commission=0.001)
startPortfolioValue = NGtrade.broker.getvalue()
print('Starting Portfolio Value:', startPortfolioValue)
print()
strategies = NGtrade.run(runonce=False)
firstStrategy = strategies[0]
# print the analyzers
print()
printTradeAnalysis(firstStrategy.analyzers.ta.get_analysis())
printSQN(firstStrategy.analyzers.sqn.get_analysis())
# Get final portfolio Value
endPortfolioValue = NGtrade.broker.getvalue()
print()
print(f'Final Portfolio Value: {endPortfolioValue:.2f}')
pnl = endPortfolioValue - startPortfolioValue
print()
print(f'PnL: {pnl:.2f}')
NGtrade.plot(style='candlestick', barup='green', bardown='red',
subtxtsize=8)[0][0].savefig('samplefigure.png', dpi=300)
#files.download('samplefigure.png')
if __name__ == '__main__':
runStrategy()
На этом мы подошли к концу нашего процесса оценки. Это был довольно упрощенный подход, чтобы проиллюстрировать, как можно использовать прогнозную модель для формулирования стратегии торговли акциями, а также провести тестирование производительности для проверки стратегии.
Ключевые выводы:
Алгоритмическая торговля - это чрезвычайно конкурентный и прибыльный бизнес. Тем не менее, торговля также сопряжена с рисками, и она постоянно развивается. В приведенном выше случае есть два основных раздела: 1) прогнозное моделирование, 2) тестирование производительности.
Соответствующие функции имеют решающее значение для оценки точности прогнозной модели. Здесь мы использовали некоторые внутренние особенности для иллюстрации; однако есть внешние функции из альтернативных источников данных, например температура, хранение, другие цены на акции и т. д. могут иметь корреляцию и причинно-следственную связь с ценами на природный газ, которые можно изучить. Перед развертыванием модели решающее значение имеет тщательное тестирование производительности. Многие общие проблемы с реальными торговыми стратегиями, такие как поведение или работа не в соответствии с ожиданиями, довольно часты и распространены в таких случаях использования. Таким образом, постоянные исследования и настройка производительности не менее важны в этом развивающемся характере аналитического торгового бизнеса.
Со мной можно связаться здесь.
Заявление об ограничении ответственности: Описанные здесь программы являются экспериментальными и не должны использоваться в коммерческих целях. Вы используете это на свой страх и риск.
