Trading Bot Based on Technical Indicators
A bot based on technical indicators is the most common class of trading algorithms. The strategy codifies entry/exit rules through RSI, MACD, Bollinger Bands, volume, and other indicators. The developer's main task is to correctly implement the logic, avoid lookahead bias, and ensure reliable execution.
Typical Indicator Strategies
RSI Oversold/Overbought Strategy
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:
# IMPORTANT: shift(1) — use only closed candles
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]
# Entry from oversold zone (crossover from below)
if previous_rsi < self.oversold and current_rsi >= self.oversold:
return 'BUY'
# Entry from overbought zone (crossover from above)
if previous_rsi > self.overbought and current_rsi <= self.overbought:
return 'SELL'
return 'HOLD'
MACD Strategy
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 and Signal line crossover
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]
# Additional filter: crossover above/below zero line
if crossover_up and macd_line.iloc[-1] < 0: # in negative zone — weaker
return 'BUY' if self.config.trade_below_zero else 'HOLD'
elif crossover_up:
return 'BUY'
elif crossover_down:
return 'SELL'
return 'HOLD'
Combined Strategy (multi-indicator confirmation)
One indicator gives many false signals. A combination of several is more accurate:
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 conditions: all three factors align
trend_up = ema_20.iloc[-1] > ema_50.iloc[-1]
rsi_ok = 40 < rsi.iloc[-1] < 65 # not overbought, but above neutral
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 conditions: trend broken
if ema_20.iloc[-1] < ema_50.iloc[-1] and rsi.iloc[-1] > 60:
return 'SELL'
return 'HOLD'
Data Processing and Architecture
Incremental Data Update
When working with live market, don't load full history at every candle:
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):
"""Load history on startup"""
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):
"""Add new candle, remove old"""
new_row = self.candle_to_row(new_candle)
if new_candle['time'] == self.df.index[-1]:
# Update current unclosed candle
self.df.iloc[-1] = new_row
else:
# Add new closed candle
self.df = pd.concat([self.df, pd.DataFrame([new_row])])
# Keep fixed history length
if len(self.df) > self.history_length:
self.df = self.df.iloc[-self.history_length:]
Stop Loss and 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
):
# Main order
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 order for simultaneous SL and 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)
)
Testing Indicator Strategy
Before running live — mandatory 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)): # start from 50 for indicator warmup
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% capital
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,
}
Key backtesting rules: execute on next candle open (not signal candle close), include fees in calculation, test on out-of-sample data. A good strategy shows positive results on OOS data too.