. Вдохновившись недавней статьей на Veles Finance "«Bomberman»: стратегия для крипторынка с индикаторами BOP, Mean Reversion и Дончиана", я решил воплотить эту идею в жизнь. Не просто в теории, а в коде: создал полноценного алгобота на Python, который автоматизирует торговлю, тестирует параметры и визуализирует результаты.
Этот бот прозрачен: использует открытые данные с binance, классические индикаторы технического анализа и строгий walk-forward бэктест, чтобы избежать look-ahead bias (смещения в будущее).
В статье разберем логику стратегии, архитектуру бота, ключевые функции и реальные результаты на исторических данных BTC/USDT. Если вы программист с интересом к финансам или трейдер, жаждущий автоматизации, — добро пожаловать. Мы пройдемся по коду, формулам и рискам.
Почему Bomberman
Стратегия "Bomberman", описанная в оригинальной статье, черпает вдохновение из аркадного хита 1980-х. Вкратце, основная идея самой стратегии в комбинации трех индикаторов на разных таймфреймах:
BOP (Balance of Power) на 30-минутном (M30) — измеряет баланс сил между быками и медведями.
Mean Reversion Channel на 15-минутном (M15) — канал возврата к среднему для зон перепроданности/перекупленности.
Donchian Channel на 5-минутном (M5) — прорывы для подтверждения импульса.
Бот реализует это с учетом шортов (коротких позиций), леверидж�� и строгого риск-менеджмента (1% на сделку). В отличие от ручной торговли, бот вычисляет индикаторы только на исторических данных до текущего бара, имитируя реальную торговлю без подглядывания в будущее. Это не просто бэктест — это симуляция, готовая к деплою на реальном API. Достаточно будет просто добавить подгрузку актуальных данных в бота и открытие сделок с помощью клиента для биржи (ранее писали собственный клиент для bingX).
Логика стратегии: Вход, выход и "бомбы"
Основа бота — функция backtest_one, которая симулирует торговлю. Давайте разберем правила стратегии. Все расчеты на основе OHLCV-данных (Open, High, Low, Close, Volume).
1. Индикаторы
Бот вычисляет индикаторы динамически, чтобы избежать предвзятости. Вот формулы:
BOP (Balance of Power):
Mean Reversion Channel (на M15):
Donchian Channel (на M5):
2. Условия входа: "Подрыв барьера"
Толерантность tol (0.5–1.5%) добавляет гибкости для приближения к уровням.
Лонг (LONG):
BOP > 0 (быки доминируют на M30).
Цена ≤ Lower × (1 + tol) (перепроданность на M15).
Цена > предыдущий Donchian_high (прорыв на M5).
Шорт (SHORT):
BOP < 0 (медведи доминируют).
Цена ≥ Upper × (1 - tol) (перекупленность).
Цена < предыдущий Donchian_low (прорыв вниз).
Размер позиции расчитываем исходя из указанного риска и капитала. Это в первую очередь нужно при реализации реальной торговле, для бектеста пока что достаточно получить прибыль в %.
3. Условия выхода: "Взрыв и отступление"
Выход минимизирует убытки и фиксирует прибыль:
Для лонга:
Тейк-профит: цена ≥ Upper (возврат к сопротивлению).
Стоп-лосс: цена ≤ entry × 0.99 (1% убыток).
Для шорта:
Тейк-профит: цена ≤ Lower.
Стоп-лосс: цена ≥ entry × 1.01 (1% убыток).
Эта логика обеспечивает баланс: mean reversion ловит отскоки, Donchian — трендовые прорывы, BOP — фильтр тренда. В "both" режиме бот торгует и лонг, и шорт.
Архитектура бота: Функции и модули
Бот — модульный скрипт на ~300 строк Python. Зависимости: ccxt (для данных), pandas/numpy (анализ), matplotlib (визуализация). Нет ML или сложных фреймворков — чистый TA-Lib стиль, но самописный для контроля.
Для начала сделаем импорты:
import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import warnings
from datetime import datetime
import time1. Загрузка данных: fetch_binance_ohlcv
Публичный API binance(без ключей). Фетчит OHLCV за N баров назад:
from binance.client import Client
client = Client()
def fetch_binance_ohlcv(symbol='ETHUSDT', interval='15m', total_bars=50000, client=client):
limit = 1000
data = []
current_end = None
while len(data) < total_bars:
bars_to_fetch = min(limit, total_bars - len(data))
try:
klines = client.futures_klines(
symbol=symbol.upper(),
interval=interval,
limit=bars_to_fetch,
endTime=current_end
)
except Exception as e:
print("Ошибка Binance API:", e)
break
if not klines:
break
data = klines + data # prepend
current_end = klines[0][0] - 1
time.sleep(0.1)
if not data:
return pd.DataFrame()
df = pd.DataFrame(data, columns=[
'timestamp','open','high','low','close','volume',
'close_time','quote_asset_volume','number_of_trades',
'taker_buy_base','taker_buy_quote','ignore'
])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df[['open','high','low','close','volume']] = df[['open','high','low','close','volume']].astype(float)
df = df.drop_duplicates('timestamp').sort_values('timestamp').reset_index(drop=True)
df = df.rename(columns={'timestamp': 'time'})
return df[["time", "open", "high", "low", "close", "volume"]]Синхронизирует таймфреймы: ~2000 баров M30 ≈ 4000 M15 ≈ 12000 M5 (~1 год данных). Обработка дубликатов и NaN.
2. Вычисление индикаторов: add_bop, add_mean_reversion, add_donchian
Простые rolling-операции на pandas. Для каждого бара (данные свечи) режем DF до current_time и пересчитываем только прошлое. Также необходимо синхронизовать таймфреймы, так как нам нужно чтобы все индексы свечей совпадали и не было ненужных нам смещений.
def add_bop(df: pd.DataFrame, smooth: int = 1) -> pd.DataFrame:
df = df.copy()
df['bop'] = (df['close'] - df['open']) / (df['high'] - df['low']).replace(0, np.nan)
if smooth > 1:
df['bop'] = df['bop'].rolling(smooth).mean()
return df
def add_mean_reversion(df: pd.DataFrame, period: int = 20, mult: float = 2.0) -> pd.DataFrame:
df = df.copy()
df['sma'] = df['close'].rolling(period).mean()
df['std'] = df['close'].rolling(period).std()
df['upper'] = df['sma'] + mult * df['std']
df['lower'] = df['sma'] - mult * df['std']
return df
def add_donchian(df: pd.DataFrame, period: int = 20) -> pd.DataFrame:
df = df.copy()
df['donchian_high'] = df['high'].rolling(period).max()
df['donchian_low'] = df['low'].rolling(period).min()
return df
# ------------------------------------------------------------------
# 3. Синхронизация (walk-forward)
# ------------------------------------------------------------------
def sync_timeframes(df30, df15, df5):
df = df30.copy()
df = df.join(df15[['upper', 'lower']], how='left')
df = df.join(df5[['donchian_high', 'donchian_low']], how='left')
cols = ['upper', 'lower', 'donchian_high', 'donchian_low']
df[cols] = df[cols].ffill() # Только прошлые данные
return dfТри таймфрейма по этим индикатором (M30 → M15 → M5) создают иерархическую фильтрацию:
M30: "Кто сильнее?"
M15: "Где перекупленность?"
M5: "Есть ли прорыв?"
Такой мультитаймфреймовый подход - основа стратегии.
3. Бэктест: backtest_one
Walk-forward симуляция: цикл по барам M30, слайс данных + индикаторы + сигналы. Возвращает словарь с результатом и основыми данными.
def backtest_one(df: pd.DataFrame,
tol: float,
direction: str = "both",
initial_capital: float = 10_000,
leverage: int = 5,
risk_per_trade: float = 0.01) -> dict:
capital = initial_capital
position = 0 # 1 = long, -1 = short, 0 = flat
entry_price = 0.0
size = 0.0
equity = [capital]
trades = []
for i in range(50, len(df)):
row = df.iloc[i]
prev = df.iloc[i-1]
price = row['close']
# --- ВХОД ---
if position == 0:
# LONG
if direction in ["long", "both"]:
long_cond = (
row['bop'] > 0 and
price <= row['lower'] * (1 + tol) and
price > prev['donchian_high']
)
if long_cond:
size = (capital * risk_per_trade * leverage) / price
entry_price = price
position = 1
trades.append(f"LONG at {price:.5f}")
# SHORT
if direction in ["short", "both"]:
short_cond = (
row['bop'] < 0 and
price >= row['upper'] * (1 - tol) and
price < prev['donchian_low']
)
if short_cond:
size = (capital * risk_per_trade * leverage) / price
entry_price = price
position = -1
trades.append(f"SHORT at {price:.5f}")
# --- ВЫХОД ---
if position == 1: # LONG
if price >= row['upper']:
pnl = size * (price - entry_price) * leverage - size * 0.0005
capital += pnl
position = 0
trades.append(f"EXIT LONG at {price:.5f} | PnL: {pnl:+.2f}")
elif price <= entry_price * 0.99:
pnl = size * (price - entry_price) * leverage - size * 0.0005
capital += pnl
position = 0
trades.append(f"STOP LONG at {price:.5f} | PnL: {pnl:+.2f}")
if position == -1: # SHORT
if price <= row['lower']:
pnl = (size * (entry_price - price) * leverage) - size * 0.0005
capital += pnl
position = 0
trades.append(f"EXIT SHORT at {price:.5f} | PnL: {pnl:+.2f}")
elif price >= entry_price * 1.01:
pnl = size * (entry_price - price) * leverage - size * 0.0005
capital += pnl
position = 0
trades.append(f"STOP SHORT at {price:.5f} | PnL: {pnl:+.2f}")
equity.append(capital)
total_ret = (capital - initial_capital) / initial_capital
return {
'final_capital': capital,
'total_return': total_ret,
'equity': equity,
'trades': trades
}Обратите внимание на расчёт прибыли:
pnl = size (price - entry_price) leverage - size * 0.005
Комиссия учтена и она равняется size * 0.005 (для bingx со скидкой на комиссию taker 0.0025=0.25%). Для binance, bybit - 0.01, что будет уже не так приятно, но в рамках не самого большого количества сделок всё еще терпимо.
4. Оптимизация: grid_search
Грид по 5+ параметрам (576 комбинаций в полном режиме, но для теста — подмножество). Скрипт выполняется довольно быстро ввиду не самого большого количества свечей. Если хотите доработать его и добавить новые фильтры и перебор большего числа параметров - несложно можно добавить multiprocessing (параллелизм).
def grid_search(df30, df15, df5):
param_grid = {
'mr_period': [15, 20],
'mr_mult' : [2.0, 2.5],
'dc_period': [15, 20],
'bop_smooth': [1],
'tol' : [0.015],
'direction': ['both']
}
results = []
total = np.prod([len(v) for v in param_grid.values()])
print(f"Грид-поиск: {total} комбинаций...")
cnt = 0
for mr_p in param_grid['mr_period']:
for mr_m in param_grid['mr_mult']:
for dc_p in param_grid['dc_period']:
for bop_s in param_grid['bop_smooth']:
for tol in param_grid['tol']:
for dir_ in param_grid['direction']:
cnt += 1
print(f"[{cnt}/{total}] MR={mr_p}*{mr_m}, DC={dc_p}, BOP_s={bop_s}, tol={tol*100:.1f}%, {dir_}")
df_mr = add_mean_reversion(df15.copy(), mr_p, mr_m)
df_dc = add_donchian(df5.copy(), dc_p)
df_bop = add_bop(df30.copy(), bop_s)
df_test = sync_timeframes(df_bop, df_mr, df_dc)
res = backtest_one(df_test, tol, direction=dir_)
results.append({
'mr_period': mr_p,
'mr_mult': mr_m,
'dc_period': dc_p,
'bop_smooth': bop_s,
'tolerance': tol,
'direction': dir_,
'final_capital': res['final_capital'],
'total_return': res['total_return'],
})
return pd.DataFrame(results)По сути, мы здесь просто для каждой комбинации прогоняем бектест и собираем результаты - далее будем их сравнивать и выбирать лучшее.
5. Визуализация: plot_best
Построение графика для лучшей комбинации:
Верхний: Цена + каналы + сигналы (стрелки лонг/шорт).
Нижний: Equity-кривой. Matplotlib генерирует интерактивный plot.
def plot_best(df30, df15, df5, best_row):
df_mr = add_mean_reversion(df15.copy(),
int(best_row['mr_period']),
best_row['mr_mult'])
df_dc = add_donchian(df5.copy(),
int(best_row['dc_period']))
df_bop = add_bop(df30.copy(),
int(best_row['bop_smooth']))
df = sync_timeframes(df_bop, df_mr, df_dc)
# Пересчёт с лучшими параметрами
res = backtest_one(df,
best_row['tolerance'],
direction=best_row['direction'])
# Сигналы
df['signal'] = 0
for i, trade in enumerate(res['trades']):
if 'LONG' in trade or 'SHORT' in trade:
timestamp = df.index[50 + i // 2]
df.loc[timestamp, 'signal'] = 1 if 'LONG' in trade else -1
plt.figure(figsize=(16, 10))
ax1 = plt.subplot(3, 1, 1)
ax1.plot(df.index, df['close'], label='BTC/USDT', color='steelblue')
ax1.plot(df.index, df['lower'], '--', color='green', label='Lower')
ax1.plot(df.index, df['upper'], '--', color='red', label='Upper')
ax1.scatter(df.index[df['signal'] == 1], df['close'][df['signal'] == 1],
marker='^', color='lime', s=100, label='LONG')
ax1.scatter(df.index[df['signal'] == -1], df['close'][df['signal'] == -1],
marker='v', color='red', s=100, label='SHORT')
ax1.set_title(f'Bomberman | {best_row["direction"].upper()} | '
f'Годовых: {best_row["annualized"]*100:.1f}%')
ax1.legend()
ax3 = plt.subplot(3, 1, 3)
ax3.plot(res['equity'], color='gold')
ax3.set_title(f'Капитал: ${res["final_capital"]:,.0f}')
plt.tight_layout()
plt.show()Запустим нашу стратегию
Запустим подгрузку свечей по всем тф и используем наши функции:
if __name__ == "__main__":
print("Загрузка данных...")
df_30m = fetch_binance_ohlcv('BTCUSDT', '30m', 5000)
df_15m = fetch_binance_ohlcv('BTCUSDT', '15m', 10000)
df_5m = fetch_binance_ohlcv('BTCUSDT', '5m', 30000)
print(f" 30m: {len(df_30m)} | 15m: {len(df_15m)} | 5m: {len(df_5m)}")
results_df = grid_search(df_30m, df_15m, df_5m)
top5 = results_df.sort_values('annualized', ascending=False).head(5)
print("\n" + "="*80)
print("ТОП-5 ПАРАМЕТРОВ (с LONG/SHORT/BOTH)")
print("="*80)
print(top5[['direction', 'mr_period', 'mr_mult', 'dc_period',
'bop_smooth', 'tolerance', 'annualized', 'final_capital']])
best = top5.iloc[0]
print(f"\nВизуализация лучшего: {best['direction']} | "
f"Годовых: {best['annualized']*100:.1f}%")
plot_best(df_30m, df_15m, df_5m, best)Результаты
На исторических данных BTC/USDT бот показал солидные цифры. В полном грид-серче (576 runs) лучшая комбинация дала 38% прибыли за период в 30000 5m свечей, что равно приблизительно 104 дням. Это отличный результат за такой небольшой период.
Конечно, можно добавлять дополнительные фильтры, делать больше расчётов. Но по графику доходности мы видим, что она стабильна - так что работоспособность стратегии подтвердилась!
