ВАЖНО: данная статья создана исключительно в образовательных целях. Торговля на бирже связана с большим риском и может привести к значительным убыткам.

Первая часть статьи доступна по ссылке.

Введение (и напоминание)

В предыдущей части статьи мы разобрали, как написать простейшего торгового бота на python с использованием Binance API. Стратегия, которую мы использовали, основывалась на статичном коридоре цены на паре ETHUSDT. Такая стратегия крайне редко применима на практике и в нашем случае служила только в качестве примера, однако, она проиллюстрировала важные методы при торговле с реальными стратегиями.

Кратко напомним основные шаги стратегии из предыдущей части статьи:

1. Запросить последнюю цену актива на бирже (без разделения на BID или ASK в простейшем случае);

2. Если мы вышли за верхнюю границу коридора: закрыть лонг и открыть шорт;

3. Если мы вышли за нижнюю границу коридора: закрыть шорт и открыть лонг.

Теперь, используя алгоритм выше, мы можем напомнить общую структуру реализации данной стратегии на языке python:

# imports and functions are omitted here as they stay exactly the same 

# flags to track the exchanges to avoid duplicated entries
in_short = False
in_long = False

# main loop
while True:
    # get latest price for the symbol
    latest_price = client.futures_symbol_ticker(symbol=symbol)['price']
    latest_price = float(latest_price)
    
    # check the condition (these conditions could be substitued by any other)
    if "<condintion>":
        # if we are in long position, close it
        if in_long:
            long_close(symbol=symbol, quantity=1)
            in_long = False
        # if we are not in short position, open it
        if not in_short:
            short_open(symbol=symbol, quantity=1)
            in_short = True
    elif "<other_contition>":
        # if we are in long position, close it
        if not in_long:
            long_open(symbol=symbol, quantity=1)
            in_long = True
        # if we are in short position, close it
        if in_short:
            short_close(symbol=symbol, quantity=1)
            in_short = False

    # 1 second sleep
    time.sleep(1)

То есть, если заменить условия <condition> и <other_condition> на некоторые реальные условия, то код выше может служить основой для полноценной торговой стратегии.

Базовые (статистические) стратегии

Переходя к настоящим торговым стратегиям на одном активе (торговля несколькими активами заслуживает отдельного серьезного разговора) хочется выделить три основных направления в статистических стратегиях (статистическими стратегиями будем называть те стратегии, которые основываются только на данных о предыдущей цене и объемах торговли и не учитывают "внешних" факторов влияющих на рынок):

1. Пересечение скользящих средних (moving average crossover): стратегия рассматривает две скользящие средние с разными периодами и входит в лонг, когда линия с меньшим периодом пересекает линию с бОльшим периодом снизу вверх и входит в шорт, наоборот, при пересечении линии с меньшим периодом линии с бОльшим периодом сверху вниз.

2. Динамический коридор цены (channel breakout): стратегия рассматривает одну скользящую среднюю а также коридор цены, верхняя граница которого соответствует максимуму цены за предыдущие n свеч, а нижняя -- минимуму за предыдущие n свеч.

3. Линии Боллинджера (Bollinger bands): стратегия также строит коридор цены, но по другим правилам.

   Верхняя граница коридора = скользящая средняя + 2 стандартных отклонения цены;

   Нижняя граница коридора = скользящая средняя - 2 стандартных отклонения цены^.

   Правила для входа и выхода стратегии определяются следующим образом:

   При пересечении цены верхней линии Боллинджера (снизу вверх) открывается шорт. Шорт закрывается при пересечении ценой скользящей средней (сверху вниз).

   При пересечении цены нижней линии Боллинджера (сверху вниз) открывается лонг. Лонг закрывается при пересечении ценой скользящей средней (снизу вверх).

Линии Боллинджера

В данной статье мы подробно разберем линии Боллинджера (реализация других стратегий, после освоения линий Боллинджера, не должна составить труда).

Начнем с определения:

Линии (полосы) Боллинджера (Bollinger bands) — инструмент технического анализа финансовых рынков, отражающий текущие отклонения цены акции, товара или валюты.

Основное предположение стратегии состоит в том, что при значительном отклонении цены от скользящей средней, она вернется назад. Преимущество линий Боллинджера -- простота их вычисления. Представим, что мы имеем данные о ценах закрытия 5 минутных свечей некоторого актива (см таблицу ниже), тогда мы можем вычислить линии Боллинджера используя следующие формулы:

В таблице представлены цены закрытия а также значения скользящих средних и отклонений, а также значения линий Боллиндджера:

На рисунке 1 ниже представлены линии Боллинджера (синий и фиолетовый цвета), а также скользящая средняя (оранжевый цвет). Помимо линий Боллинджера представлены точки входа и выхода из сделок (согласно стратегии). Они отмечены вертикальными линиями. Вертикальная пунктирная линия соответствует моменту входа в сделку, а вертикальная сплошная линия -- моменту выхода из сделки. Число рядом с подписью внизу сделки обозначает ее доход (убыток) в процентах.

Имплементация на Python

Теперь мы можем использовать теорию выше для имплементации стратегии на языке python. Существенное отличие текущей стратегии от стратегии статичного коридора цены в первой части статьи состоит в том, что теперь мы хотим получать цену актива только лишь в момент закрытия свечи, то есть, каждые 5 минут (константу 5 минут можно легко изменять).

Вычисление линий Боллинджера

Код ниже показывает, как можно вычислять линии Боллинджера для интервалов в 5 минут с периодом 3 свечи (15 минут):

import numpy as np
import time
from datetime import datetime
from binance.client import Client

# list to keep all closing prices
prices = []

# lists to keep all stats needed
moving_average_values = []
bollinger_band_high_values = []
bollinger_band_low_values = []

# flags to track the exchanges to avoid duplicated entries
in_short = False
in_long = False

# period of Bollinger bands
period = 3

last_time = ""

# main loop
while True:
    current_time = datetime.now()
    # if current candlestick has closed and current time is not equal to the
    # latest record (to avoid duplicated price records)
    if current_time.minute % 5 == 0 and current_time.strftime('%Y-%m-%d %H:%M') != last_time:
        last_time = current_time.strftime('%Y-%m-%d %H:%M')
        # get latest price for the symbol
        latest_price = client.futures_symbol_ticker(symbol=symbol)['price']
        latest_price = float(latest_price)
        prices.append(latest_price)
    
        # print latest price with current time
        print(current_time, latest_price)

        # calculate moving average and deviation
        ma = np.mean(prices[-period:])
        moving_average_values.append(ma)

        std = np.std(prices[-period:], ddof=1)

        # calculate Bollinger bands
        bb_high = ma + 2 * std
        bb_low = ma - 2 * std

        bollinger_band_high_values.append(bb_high)
        bollinger_band_low_values.append(bb_low)

Правила входа (выхода) в сделки

Когда мы имеем значения линий Боллинджера и скользящей средней, мы можем перейти к имплементации самой стратегии. Для начала разберем условия входа в сделку:

# check if current closing price is higher than the upper band
# but previous closing price is lower than the upper band
# this indicates that the crossing occured on current candlestick
if prices[-2] < bollinger_band_high_values[-2] and prices[-1] > bollinger_band_high_values[-1]:
    # if we are not in short position, open it
    if not in_short:
        short_open(symbol=symbol, quantity=1)
        in_short = True

# check if current closing price is lower than the lower band
# but previous closing price is higher than the lower band
# this indicates that the crossing occured on current candlestick
if prices[-1] < bollinger_band_low_values[-1] and prices[-2] > bollinger_band_low_values[-2]:
    # if we are not in long position, open it
    if not in_long:
        long_open(symbol=symbol, quantity=1)
        in_long = True

Теперь разберем условия выхода из сделки:

# check if current closing price is lower than the moving average
# but previous closing price is higher than the moving average
# this indicates that the crossing occured on current candlestick
if prices[-2] > moving_average_values[-2] and prices[-1] < moving_average_values[-1]:
    if in_short:
        short_close(symbol=symbol, quantity=1)
        in_short = False

# vice versa
if prices[-2] < moving_average_values[-2] and prices[-1] > moving_average_values[-1]:
    if in_long:
        long_close(symbol=symbol, quantity=1)
        in_long = False

Наконец, соберем все кусочки вместе и получим полноценную стратегию!

import numpy as np
import time
from datetime import datetime
from binance.client import Client

# exchange operations:
def short_open(symbol, quantity):
    client.futures_create_order(symbol=symbol, 
                                side='SELL', 
                                type='MARKET', 
                                quantity=quantity)
    
def short_close(symbol, quantity):
    client.futures_create_order(symbol=symbol, 
                                side='BUY', 
                                type='MARKET', 
                                quantity=quantity)
    
def long_open(symbol, quantity):
    client.futures_create_order(symbol=symbol, 
                                side='BUY', 
                                type='MARKET', 
                                quantity=quantity)

def long_close(symbol, quantity):
    client.futures_create_order(symbol=symbol, 
                                side='SELL', 
                                type='MARKET', 
                                quantity=quantity)

# initialize Client with your API keys
api_key = "<your_api_key>"
api_secret = "<your_private_api_key>"
client = Client(api_key,  api_secret)

# traded symbol
symbol = "ETHUSDT"

# list to keep all closing prices
prices = []

# lists to keep all stats needed
moving_average_values = []
bollinger_band_high_values = []
bollinger_band_low_values = []

# flags to track the exchanges to avoid duplicated entries
in_short = False
in_long = False

# period of Bollinger bands
period = 3

last_time = ""

# main loop
while True:
    current_time = datetime.now()
    # if current candlestick has closed and current time is not equal to the
    # latest record (to avoid duplicated price records)
    if current_time.minute % 5 == 0 and current_time.strftime('%Y-%m-%d %H:%M') != last_time:
        last_time = current_time.strftime('%Y-%m-%d %H:%M')
        # get latest price for the symbol
        latest_price = client.futures_symbol_ticker(symbol=symbol)['price']
        latest_price = float(latest_price)
        prices.append(latest_price)
    
        # print latest price with current time
        print(current_time, latest_price)

        # calculate moving average and deviation
        ma = np.mean(prices[-period:])
        moving_average_values.append(ma)

        std = np.std(prices[-period:], ddof=1)

        # calculate Bollinger bands
        bb_high = ma + 2 * std
        bb_low = ma - 2 * std

        bollinger_band_high_values.append(bb_high)
        bollinger_band_low_values.append(bb_low)

        # descion of the strategy
        # short exchanges
        if prices[-2] < bollinger_band_high_values[-2] and prices[-1] > bollinger_band_high_values[-1]:
            if not in_short:
                short_open(symbol=symbol, quantity=1)
                in_short = True

        if prices[-2] > moving_average_values[-2] and prices[-1] < moving_average_values[-1]:
            if in_short:
                short_close(symbol=symbol, quantity=1)
                in_short = False

        # long conditions
        if prices[-1] < bollinger_band_low_values[-1] and prices[-2] > bollinger_band_low_values[-2]:
            if not in_long:
                long_open(symbol=symbol, quantity=1)
                in_long = True
        
        if prices[-2] < moving_average_values[-2] and prices[-1] > moving_average_values[-1]:
            if in_long:
                long_close(symbol=symbol, quantity=1)
                in_long = False

    time.sleep(1)

Спасибо за прочтение!