Разработка бота на базе торговых индикаторов
Бот на основе технических индикаторов — наиболее распространённый класс торговых алгоритмов. Стратегия кодифицирует правила входа/выхода через RSI, MACD, Bollinger Bands, объём и другие индикаторы. Главная задача разработчика — правильно реализовать логику, избежать lookahead bias и обеспечить надёжное исполнение.
Типовые индикаторные стратегии
RSI Oversold/Overbought стратегия
import pandas_ta as ta
import pandas as pd
class RSIStrategy:
def __init__(self, rsi_period=14, oversold=30, overbought=70):
self.rsi_period = rsi_period
self.oversold = oversold
self.overbought = overbought
def generate_signal(self, df: pd.DataFrame) -> str:
# ВАЖНО: shift(1) — используем только закрытые свечи
rsi = ta.rsi(df['close'], length=self.rsi_period).shift(1)
prev_rsi = rsi.shift(1)
current_rsi = rsi.iloc[-1]
previous_rsi = prev_rsi.iloc[-1]
# Вход из oversold зоны (пересечение снизу вверх)
if previous_rsi < self.oversold and current_rsi >= self.oversold:
return 'BUY'
# Вход из overbought зоны (пересечение сверху вниз)
if previous_rsi > self.overbought and current_rsi <= self.overbought:
return 'SELL'
return 'HOLD'
MACD стратегия
class MACDStrategy:
def __init__(self, fast=12, slow=26, signal=9):
self.fast = fast
self.slow = slow
self.signal = signal
def generate_signal(self, df: pd.DataFrame) -> str:
macd_data = ta.macd(df['close'], fast=self.fast, slow=self.slow, signal=self.signal)
macd_line = macd_data[f'MACD_{self.fast}_{self.slow}_{self.signal}'].shift(1)
signal_line = macd_data[f'MACDs_{self.fast}_{self.slow}_{self.signal}'].shift(1)
# Пересечение MACD и Signal линии
crossover_up = macd_line.iloc[-2] < signal_line.iloc[-2] and macd_line.iloc[-1] >= signal_line.iloc[-1]
crossover_down = macd_line.iloc[-2] > signal_line.iloc[-2] and macd_line.iloc[-1] <= signal_line.iloc[-1]
# Дополнительный фильтр: пересечение выше/ниже нулевой линии
if crossover_up and macd_line.iloc[-1] < 0: # в отрицательной зоне — слабее
return 'BUY' if self.config.trade_below_zero else 'HOLD'
elif crossover_up:
return 'BUY'
elif crossover_down:
return 'SELL'
return 'HOLD'
Комбинированная стратегия (multi-indicator confirmation)
Один индикатор даёт много ложных сигналов. Комбинация нескольких — точнее:
class MultiIndicatorStrategy:
def generate_signal(self, df: pd.DataFrame) -> str:
rsi = ta.rsi(df['close'], 14).shift(1)
ema_20 = ta.ema(df['close'], 20).shift(1)
ema_50 = ta.ema(df['close'], 50).shift(1)
volume_ma = df['volume'].rolling(20).mean().shift(1)
current_close = df['close'].iloc[-1]
# Условия BUY: все три фактора совпадают
trend_up = ema_20.iloc[-1] > ema_50.iloc[-1]
rsi_ok = 40 < rsi.iloc[-1] < 65 # не перекуплен, но выше нейтрали
volume_confirm = df['volume'].iloc[-1] > volume_ma.iloc[-1] * 1.3
price_above_ema = current_close > ema_20.iloc[-1]
if trend_up and rsi_ok and volume_confirm and price_above_ema:
return 'BUY'
# Условия SELL: тренд сломан
if ema_20.iloc[-1] < ema_50.iloc[-1] and rsi.iloc[-1] > 60:
return 'SELL'
return 'HOLD'
Обработка данных и архитектура
Incremental data update
При работе с live рынком не нужно загружать всю историю при каждой свече:
class IncrementalCandleManager:
def __init__(self, symbol: str, interval: str, history_length: int = 200):
self.symbol = symbol
self.interval = interval
self.history_length = history_length
self.df: pd.DataFrame = None
async def initialize(self):
"""Загружаем историю при старте"""
candles = await self.exchange.get_klines(
self.symbol, self.interval, limit=self.history_length
)
self.df = self.to_dataframe(candles)
def update(self, new_candle: dict):
"""Добавляем новую свечу, удаляем старую"""
new_row = self.candle_to_row(new_candle)
if new_candle['time'] == self.df.index[-1]:
# Обновляем текущую незакрытую свечу
self.df.iloc[-1] = new_row
else:
# Добавляем новую закрытую свечу
self.df = pd.concat([self.df, pd.DataFrame([new_row])])
# Держим фиксированную длину истории
if len(self.df) > self.history_length:
self.df = self.df.iloc[-self.history_length:]
Stop Loss и Take Profit
class PositionManager:
async def open_with_sl_tp(
self,
symbol: str,
side: str,
amount: float,
entry_price: float,
sl_percent: float,
tp_percent: float
):
# Основной ордер
order = await self.exchange.place_order(symbol, side, amount)
if side == 'buy':
sl_price = entry_price * (1 - sl_percent / 100)
tp_price = entry_price * (1 + tp_percent / 100)
else:
sl_price = entry_price * (1 + sl_percent / 100)
tp_price = entry_price * (1 - tp_percent / 100)
# OCO ордер для одновременной установки SL и TP
await self.exchange.place_oco_order(
symbol=symbol,
side='sell' if side == 'buy' else 'buy',
quantity=amount,
price=tp_price, # limit (take profit)
stop_price=sl_price, # stop trigger
stop_limit_price=sl_price * (0.995 if side == 'buy' else 1.005)
)
Тестирование индикаторной стратегии
Перед запуском live — обязательный backtesting:
def backtest_strategy(strategy, candles_df: pd.DataFrame, initial_capital: float = 10000):
capital = initial_capital
position = None
trades = []
for i in range(50, len(candles_df)): # начинаем с 50 для прогрева индикаторов
window = candles_df.iloc[:i]
signal = strategy.generate_signal(window)
current_price = candles_df['close'].iloc[i]
if signal == 'BUY' and position is None:
size = capital * 0.95 / current_price # 95% капитала
position = {'side': 'buy', 'price': current_price, 'size': size}
capital -= size * current_price
elif signal == 'SELL' and position and position['side'] == 'buy':
pnl = (current_price - position['price']) * position['size']
capital += position['size'] * current_price
trades.append({'pnl': pnl, 'return_pct': pnl / (position['price'] * position['size']) * 100})
position = None
return {
'final_capital': capital,
'roi': (capital / initial_capital - 1) * 100,
'num_trades': len(trades),
'win_rate': sum(1 for t in trades if t['pnl'] > 0) / len(trades) * 100 if trades else 0,
}
Ключевые правила backtesting: исполнять по open следующей свечи (не по close сигнальной), включать комиссии в расчёт, тестировать на out-of-sample данных. Хорошая стратегия показывает positive results и на OOS данных.