Этот пост посвящен реализации стратегии grid trading’a на Python.
Что такое grid-трейдинг?
Стратегия Grid Trading размещает набор ордеров на покупку и продажу через равные промежутки времени выше и ниже текущей цены. Когда цена колеблется вверх или вниз, исполняются long и short ордера, которые будут приносить прибыль, поскольку цена продолжает двигаться вверх и вниз по сетке.
Существует два варианта стратегии:
- По диапазону рынка
- По рыночному тренду
Логика торговли
В этой записи будет показано, как реализовать стратегию grid торговли для ранжирования рынка. Логика торговли выглядит следующим образом:
Определение параметров для grid стратегии:
- Размер сетки: 1%
- Количество сеток 5
- Take profit: 1% выше и ниже сетки
- Stop loss: 1% выше и ниже сетки
- Период сброса сетки: 7 дней
Мы будем использовать индикатор Chop, чтобы определить является ли текущий рынок трендовым или колеблющимся. Если текущий рынок находится в диапазоне, то мы открываем 5 лимитных ордеров на покупку и 5 лимитных ордеров на продажу выше и ниже текущей рыночной цены. Также устанавливаем соответствующие уровни тейк-профита и стоп-лосса на 1% выше и ниже лимитной цены.
Шаг 1: Расчет данных для индикатора Chop
Прежде всего, мы определяем параметры сетки при инициализации.
from AlgoAPI import AlgoAPIUtil, AlgoAPI_Backtest
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import pandas_ta as ta
class AlgoEvent:
def __init__(self):
self.timer = datetime(1970,1,1)
self.grid_setup_time = datetime(1970,1,1)
self.grid_size_pct = 0.01
self.grid_num = 5
self.tp_pct = 0.01
self.sl_pct = 0.01
self.numObs = 30
self.is_runningGrid = False
self.trade_size = 0.1
В этой заметке будет использоваться функция API getHistoricalBar для сбора исторических данных свечей. Затем применяется библиотека pandas_ta.chop для расчета серий данных chop.
def compute_CHOP(self, data):
high = [data[t]['h'] for t in data]
low = [data[t]['l'] for t in data]
close = [data[t]['c'] for t in data]
df = pd.DataFrame.from_dict({"high":high,"low":low,"close":close})
chop_series = ta.chop(high=df['high'], low=df['low'], close=df['close'], length=14, atr_length=1)
return chop_series.to_numpy()
def on_marketdatafeed(self, md, ab):
# вызов стратегий каждые 24 часа
if md.timestamp >= self.timer+timedelta(hours=24):
# update timer
self.timer = md.timestamp
# получить последние OHLC свечи
res = self.evt.getHistoricalBar(
contract = {"instrument":md.instrument},
numOfBar = self.numObs,
interval = "D"
)
if len(res)<self.numObs:
return
chop_series = self.compute_CHOP(res)
self.evt.consoleLog(chop_series)Шаг 2: Определение диапазона рынка и настройка сетки
Индикатор CHOP располагается в диапазоне от 0 до 100. Это означает, что рынок находится в диапазоне, если значение закрывается до 100. И рынок является трендовым, если значение закрывается до 0. Таким образом, в логике кода представленного ниже (строки 21-23) мы считаем, что рынок находится в зоне диапазона, когда предыдущее значение CHOP ниже 50, а текущее значение выше 50.
Настройка сетки лимитных ордеров на покупку описана в строках 32-45; сетка лимитных ордеров на продажу в строке 47-60.
def on_marketdatafeed(self, md, ab):
# вызов стратегий каждые 24 часа
if md.timestamp >= self.timer+timedelta(hours=24):
# update timer
self.timer = md.timestamp
# получение последних OHLC свечей
res = self.evt.getHistoricalBar(
contract = {"instrument":md.instrument},
numOfBar = self.numObs,
interval = "D"
)
if len(res)<self.numObs:
return
chop_series = self.compute_CHOP(res)
self.evt.consoleLog(chop_series)
# проверка нахождения рынка в тренде или в диапазоне
is_ranging = True
if chop_series[-1]>50 and chop_series[-2]<=50:
is_ranging = False
# установка нового grid для is_ranging рынка
if not self.is_runningGrid and is_ranging:
self.is_runningGrid = True
self.grid_setup_time = md.timestamp
for i in range(1,self.grid_num+1):
# лимитный ордер на покупку
price = md.bidPrice*(1-i*self.grid_size_pct)
#ask = md.askPrice*(1-i*self.grid_size_pct)
order = AlgoAPIUtil.OrderObject(
instrument = md.instrument,
openclose = 'open',
buysell = 1, # 1=buy, -1=sell
ordertype = 1, # 1=limit order
price = price,
volume = self.trade_size,
takeProfitLevel = price*(1+self.tp_pct),
stopLossLevel = price*(1-self.sl_pct)
)
self.evt.sendOrder(order)
# лимитный ордер на продажу
#bid = md.bidPrice*(1+i*self.grid_size_pct)
price = md.askPrice*(1+i*self.grid_size_pct)
order = AlgoAPIUtil.OrderObject(
instrument = md.instrument,
openclose = 'open',
buysell = -1, # 1=buy, -1=sell
ordertype = 1, # 1=лимитный ордер
price = price,
volume = self.trade_size,
takeProfitLevel = price*(1-self.tp_pct),
stopLossLevel = price*(1+self.sl_pct)
)
self.evt.sendOrder(order)Шаг 3: Сброс сетки
В логике нашей стратегий мы будем сбрасывать сетку каждые 7 дней. В функций reset_grid() отменяются все невыполненные ордера.
def reset_grid(self):
# отмена всех невыполненных заявок
positions, osOrder, pendOrder = self.evt.getSystemOrders()
for tradeID in pendOrder:
order = AlgoAPIUtil.OrderObject(
tradeID=tradeID,
openclose='cancel'
)
self.evt.sendOrder(order)
def on_marketdatafeed(self, md, ab):
# вызов стратегий каждые 24 часа
if md.timestamp >= self.timer+timedelta(hours=24):
# update timer
self.timer = md.timestamp
# ....
# сброс сетки
if md.timestamp>self.grid_setup_time+timedelta(days=7):
self.grid_setup_time = md.timestamp
self.reset_grid()
self.is_runningGrid = FalseПолный исходный код
Подытожив вышеперечисленное, рабочий скрипт будет представлен ниже.
from AlgoAPI import AlgoAPIUtil, AlgoAPI_Backtest
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import pandas_ta as ta
class AlgoEvent:
def __init__(self):
self.timer = datetime(1970,1,1)
self.grid_setup_time = datetime(1970,1,1)
self.grid_size_pct = 0.01
self.grid_num = 5
self.tp_pct = 0.01
self.sl_pct = 0.01
self.numObs = 30
self.is_runningGrid = False
self.trade_size = 0.1
def start(self, mEvt):
self.evt = AlgoAPI_Backtest.AlgoEvtHandler(self, mEvt)
self.evt.start()
def compute_CHOP(self, data):
high = [data[t]['h'] for t in data]
low = [data[t]['l'] for t in data]
close = [data[t]['c'] for t in data]
df = pd.DataFrame.from_dict({"high":high,"low":low,"close":close})
chop_series = ta.chop(high=df['high'], low=df['low'], close=df['close'], length=14, atr_length=1)
return chop_series.to_numpy()
def reset_grid(self):
# отменить все неисполненные ордера
positions, osOrder, pendOrder = self.evt.getSystemOrders()
for tradeID in pendOrder:
order = AlgoAPIUtil.OrderObject(
tradeID=tradeID,
openclose='cancel'
)
self.evt.sendOrder(order)
def on_marketdatafeed(self, md, ab):
# вызов стратегий каждые 24 часа
if md.timestamp >= self.timer+timedelta(hours=24):
# update timer
self.timer = md.timestamp
# получение последних OHLC свечей
res = self.evt.getHistoricalBar(
contract = {"instrument":md.instrument},
numOfBar = self.numObs,
interval = "D"
)
if len(res)<self.numObs:
return
chop_series = self.compute_CHOP(res)
self.evt.consoleLog(chop_series)
# проверка нахождения рынка в тренде или в диапазоне
is_ranging = True
if chop_series[-1]>50 and chop_series[-2]<=50:
is_ranging = False
# установка нового grid'a для is_ranging рынка
if not self.is_runningGrid and is_ranging:
self.is_runningGrid = True
self.grid_setup_time = md.timestamp
for i in range(1,self.grid_num+1):
# лимитный ордер на покупку
price = md.bidPrice*(1-i*self.grid_size_pct)
#ask = md.askPrice*(1-i*self.grid_size_pct)
order = AlgoAPIUtil.OrderObject(
instrument = md.instrument,
openclose = 'open',
buysell = 1, # 1=buy, -1=sell
ordertype = 1, # 1=limit order
price = price,
volume = self.trade_size,
takeProfitLevel = price*(1+self.tp_pct),
stopLossLevel = price*(1-self.sl_pct)
)
self.evt.sendOrder(order)
# лимитный ордер на продажу
#bid = md.bidPrice*(1+i*self.grid_size_pct)
price = md.askPrice*(1+i*self.grid_size_pct)
order = AlgoAPIUtil.OrderObject(
instrument = md.instrument,
openclose = 'open',
buysell = -1, # 1=buy, -1=sell
ordertype = 1, # 1=limit order
price = price,
volume = self.trade_size,
takeProfitLevel = price*(1-self.tp_pct),
stopLossLevel = price*(1+self.sl_pct)
)
self.evt.sendOrder(order)
# сброс сетки
if md.timestamp>self.grid_setup_time+timedelta(days=7):
self.grid_setup_time = md.timestamp
self.reset_grid()
self.is_runningGrid = FalseРезультат работы стратегий
Давайте попробуем протестировать скрипт. Настройки теста:
- Инструмент: BTCUSD
- Период: 2021.01 – 2021.12
- Стартовый депозит: $100,000
- Интервал: 1-час
- Разрешить Short: True
