Розробка бота на базі торговых індикаторів
Бот на основі технічних індикаторів — найпоширеніший клас торговельних алгоритмів. Стратегія кодифікує правила входу/виходу через 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 даних. Хороша стратегія показує позитивні результати й на OOS даних.