Создание максимально стабильной автоматизированной торговой системы: от бектеста до реального бота

Привет, хабр!

Сегодня мы разберём полный цикл создания торговой системы на Python: от бэктеста стратегии до её запуска в реальном времени на бирже BingX. Статегия будет основа на индикаторах и математике, но они будут довольно неклассические и, думаю, что многим это будет интересно.

Я опишу логику стратегии, покажу код и объясню каждую часть шаг за шагом. Это не просто копипаст — это полноценный гайд, чтобы вы могли адаптировать систему под себя. Мы используем библиотеки вроде Pandas, NumPy, Matplotlib и API бирж (Binance для данных, BingX для торгов).

Предупреждение: Сейчас система находится в тесте около 2 недель. На данный момент профит составляет 5% к капиталу бота, но потеря капитала также возможна. Это не финансовый совет — тестируйте на демо-счёте. Я также постоянно подгоняю параметры, чтобы бот был актуален и периодически заменяю монетки в боте.

Все файлы этой торговой системы, а также pine script выложу на мой github.

Введение: Почему нужна торговая система?

Торговая система — это набор правил для входа/выхода из позиций, основанный на техническом анализе. Автоматизация позволяет:

• Исключить эмоции (страх, жадность).

• Тестировать стратегии на исторических данных (бэктест).

• Торговать 24/7.

• Управлять рисками (например, риск 1% на сделку).

Наша стратегия основана на индикаторах: Ichimoku Cloud, CCI, ADX, RSI, NATR, Bollinger Bands Width. Мы торгуем на паре SOLUSDT (или других) на 1-часовом таймфрейме. Вход по пересечению CCI, фильтры от других индикаторов. Выходы по TP/SL с безубытком (breakeven).

Система разделена на три файла:

1. main.py — бэктест и оптимизация.

2. bingx_client.py — клиент для API BingX (открытие/закрытие ордеров).

3. realtime.py — реалтайм бот, использующий данные с Binance и торговлю на BingX.

Почему Binance для данных? Он бесплатный и надёжный, не требует апи ключей. BingX — для торговли (самые низкие комиссии, Perpetual Futures).

Шаг 1: Логика стратегии

Ключевые индикаторы

• Ichimoku Cloud (Kumo): Облако для определения тренда. Входим в long выше облака, в short ниже.

• CCI (Commodity Channel Index): Пересечение +98 для long, -98 для short.

• ADX (Average Directional Index): Фильтр силы тренда (минимум для входа).

• RSI (Relative Strength Index): Фильтр перекупленности/перепроданности.

• NATR (Normalized Average True Range): Фильтр волатильности.

• BBW (Bollinger Bands Width): Фильтр сужения/расширения диапазона.

• MA (Moving Average): Фильтр направления тренда.

Правила входа

• Long: CCI пересекает +98 вверх + все фильтры пройдены + цена выше облака Ichimoku.

• Short: CCI пересекает -98 вниз + фильтры + цена ниже облака.

Выходы и управление

• TP: +X% от входа.

• SL: -Y% от входа.

• Безубыток: Когда цена проходит Z% в профит, SL перемещается в безубыток.

• Риск: 1% капитала на сделку (динамический расчёт).

• Флаги allow_long/short позволяют заблокировать какие-то направлению для сделок. В целом, мной эта функция почти не используется, но она всё равно имеет место быть.

Оптимизация

Перебор по параметрам: TP, SL, BE trigger, CCI length, ADX min.

Шаг 2: Бэктест (main.py)

Этот скрипт загружает данные с Binance, рассчитывает индикаторы, симулирует торговлю и оптимизирует параметры. Использует multiprocessing для ускорения.

Ключевые части кода

Сначала импорты и клиент Binance:

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from binance.client import Client
from multiprocessing import Pool, cpu_count
from itertools import product
import time

api_key = ''
api_secret = ''
client = Client(api_key, api_secret)

Функция загрузки данных (fetch_klines_paged): Загружает до 5000 свечей по 1000 за раз, чтобы обойти лимит API.

def fetch_klines_paged(symbol, interval, limit=5000):
    klines = []
    end_time = None
    while len(klines) < limit:
        batch = client.get_klines(
            symbol=symbol,
            interval=interval,
            limit=min(1000, limit - len(klines)),
            endTime=end_time
        )
        if not batch:
            break
        klines = batch + klines
        end_time = batch[0][0] - 1

    df = pd.DataFrame(klines, columns=[
        "open_time", "open", "high", "low", "close", "volume",
        "close_time", "qav", "trades", "tbbav", "tbqav", "ignore"
    ])
    df = df[["open_time", "open", "high", "low", "close"]].astype(float)
    df["open_time"] = df["open_time"].astype(int)
    return df.reset_index(drop=True)

Основная функция бэктеста (run_backtest): Принимает параметры, загружает данные, рассчитывает индикаторы, симулирует торговлю.

• Параметры фиксированные + оптимизируемые.

• Расчёт индикаторов: Ichimoku, CCI, ADX, RSI, NATR, BBW.

• Сигналы входа: Логические условия с фильтрами.

• Симуляция: Проходим по свечам, проверяем выходы (SL/TP), входы только без позиции.

• Риск: 1% от текущего капитала на qty.

• Возврат: Прибыль, equity curve.

def run_backtest(params):
    # 1. Распаковка параметров оптимизации
    symbol, tp_pct, sl_pct, be_trig_pct, cci_length, adx_long_min = params
    
    # 2. Загрузка исторических данных
    df = fetch_klines_paged(symbol, Client.KLINE_INTERVAL_1HOUR, limit=5000)
    if df.empty or len(df) < 1000:
        return None
        
    # 3. Преобразование времени в читаемый формат
    df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
    
    # 4. Здесь задаются все фиксированные параметры стратегии (много констант)
    # conversionPeriods = 10, basePeriods = 26, cci_long_thr = 98.0 и т.д.
    
    # 5. Расчёт всех индикаторов (очень объёмный блок)
    # Ichimoku Cloud → kumoTop, kumoBottom, is_above_kumo, long_lines_pass...
    # CCI, ADX, RSI, NATR, BB Width, 200 EMA и вспомогательные серии (tr, plus_dm, etc.)
    
    # 6. Формирование колонок сигналов входа
    df['long_entry'] = (...)   # сложное логическое выражение с ~use_xxx
    df['short_entry'] = (...)  # аналогично для шорта
    
    # 7. Инициализация состояния симуляции
    capital = initial_capital = 10000.0
    position = 0.0
    entry_price = 0.0
    long_sl = long_tp = short_sl = short_tp = 0.0
    long_be_triggered = short_be_triggered = False
    allow_long = allow_short = True
    equity_curve = []
    dates = []
    
    # 8. Главный цикл симуляции по всем свечам (начиная с max_lookback)
    for i in range(max_lookback, len(df)):
        row = df.iloc[i]
        price = row['close']
        
        # Текущая оценка эквити (очень важно — учитываем нереализованную прибыль)
        current_equity = capital + position * price
        equity_curve.append(current_equity)
        dates.append(row['open_time'])
        
        # 9. Логика сброса флагов разрешения входа (по облаку Ишимоку)
        if row['hasKumo'] and (not wait_flag_reset_till_flat or position == 0):
            # условия для allow_long / allow_short
            
        # 10. Проверка условий выхода из позиции
        exit_price = None
        if position > 0:                    # Лонг
            # Breakeven logic
            if be_enabled and not long_be_triggered and price >= entry_price * (1 + be_trig_pct/100):
                long_be_triggered = True
                long_sl = max(long_sl, entry_price * (1 + be_offset_pct/100))
            
            if row['low'] <= long_sl:
                exit_price = long_sl
            elif row['high'] >= long_tp:
                exit_price = long_tp
                
        elif position < 0:                  # Шорт
            # аналогичная логика для шорта
            
        # 11. Если сработал выход — закрываем позицию с учётом комиссии
        if exit_price is not None:
            net_proceeds = abs(position) * exit_price * (1 - commission_rate)
            capital += net_proceeds
            position = 0.0
            long_be_triggered = short_be_triggered = False
            
        # 12. Проверка условий входа (только если сейчас нет позиции)
        current_risk_amount = capital * risk_per_trade   # 1% от текущего капитала
        
        if df['long_entry'].iloc[i] and position == 0 and allow_long:
            # Расчёт количества
            qty = current_risk_amount / price
            cost = qty * price
            commission = cost * commission_rate
            capital -= (cost + commission)
            position = qty
            
            # Установка уровней
            long_sl = price * (1 - sl_pct / 100)
            long_tp = price * (1 + tp_pct / 100)
            
            # Блокировка повторного входа в эту же сторону
            allow_long = False
            allow_short = True
            
        elif df['short_entry'].iloc[i] and position == 0 and allow_short:
            # Аналогично для шорта
            
    # 13. Финальный подсчёт (учитываем открытую позицию на последней свече)
    final_equity = capital + position * df.iloc[-1]['close']
    total_profit = final_equity - initial_capital
    
    # 14. Формирование результата
    return {
        'symbol': symbol,
        'tp_pct': tp_pct,
        # ... остальные параметры
        'total_profit': total_profit,
        'final_equity': final_equity,
        'equity_curve': pd.Series(equity_curve, index=dates)
    }

Краткое резюме логики работы

1. Один вызов — одна комбинация параметров + один символ

2. Загружаем данные → считаем все индикаторы один раз

3. Проходим по каждой свече и имитируем реальное поведение трейдера: сначала проверяем, не пора ли выходить (SL/TP/BE), потом проверяем, можно ли войти (сигнал + нет позиции + разрешено флагом)

4. Риск-менеджмент реализуется через динамический расчёт размера позиции: всегда 1% от текущего капитала

5. Всё состояние (позиция, уровни SL/TP, флаги, капитал) сохраняется между свечами — это и есть главная идея симуляции

Именно благодаря такому пошаговому, свечному подходу бэктест максимально приближен к тому, как вела бы себя стратегия в реальной торговле.

Главный блок: Генерация комбинаций, multiprocessing, топ-5, график equity.

if __name__ == '__main__':
    symbols = ['ETHUSDT', 'DOGEUSDT', 'SOLUSDT']
    tp_pcts = [5.0, 6.0, 6.9, 8.0]
    # Другие списки
    configs = list(product(symbols, tp_pcts, ...))

    with Pool(cpu_count()) as pool:
        results = pool.map(run_backtest, configs)
  
    # Фильтр, сортировка, вывод топ-5
    results = [r for r in results if r is not None]
    print(f"Готово за {time.time() - start_time:.1f} сек.\n")

    top5 = sorted(results, key=lambda x: x['total_profit'], reverse=True)[:5]
    # Плот equity curve лучшей

Бэктест симулирует реальную торговлю, учитывая комиссии (0.035%). Комиссия 0.035% действует при реге по партнёрской ссылке, но даже без неё — 0.05% довольно маленькая. Оптимизация — полный перебор, можно улучшить алгоритмами (DEAP), но в данной ситуации перебора вполне достаточно. Результат: Топ конфигураций по прибыли, график кривой капитала.

Теперь давайте запустим бектест. Мы получим кривую капитала, список всех сделок, параметры. По итогу получаем лучшими параметрами:

<code>tp_pct = 5.0
sl_pct = 1.0
be_trig_pct = 2.0
cci_length = 25
adx_long_min = 16</code>

Кривая капитала впечатлает:

Ну в .csv файле мы видим, что распределение сделок по прибыльности адекватное, всё грамотно работает. Прикреплю equity curve и .csv файл на гитхаб — можете посмотреть и их.

Бектест позволил нам понять, что наша стратегия действительно работает. Далее давайте напишем реальную торговую систему для монетки SOLUSDT. Начнём с написания bingX SDK клиента для открытия сделок.

Шаг 3: Клиент для BingX (bingx_client.py)

BingX — биржа с Perpetual Futures. Клиент оборачивает API: подпись запросов, ордера. Этот клиент — стандартная SDK система, которую я написал уже довольно давно. Она упрощает работу с API, запросами, параметрами и сложными функциями. Не нужно каждый раз переписывать функции в коде.

Ключевые методы

Инициализация:

import time, hmac, hashlib, requests, json

class BingxClient:
    BASE_URL = "https://open-api-vst.bingx.com"

    def __init__(self, api_key: str, api_secret: str, symbol: str = None):
        self.api_key = api_key
        self.api_secret = api_secret
        self.symbol = self._to_bingx_symbol(symbol) if symbol else None
        self.time_offset = self.get_server_time_offset()

    def _to_bingx_symbol(self, symbol: str) -> str:
        return symbol.replace("USDT", "-USDT")

    def _sign(self, query: str) -> str:
        return hmac.new(self.api_secret.encode("utf-8"), query.encode("utf-8"), hashlib.sha256).hexdigest()

Подпись и запрос:

def parseParam(self, paramsMap: dict) -> str:
        # Сортировка и timestamp
        # ...

    def send_request(self, method: str, path: str, urlpa: str, payload: dict):
        sign = self._sign(urlpa)
        url = f"{self.APIURL}{path}?{urlpa}&signature={sign}"
        headers = {'X-BX-APIKEY': self.api_key}
        response = requests.request(method, url, headers=headers, data=payload)
        return response.json()

    def _request(self, method: str, path: str, params=None):
        # Подготовка и отправка
        # ...

get_mark_price: Марк-цена для избежания ликвидации.

place_market_order: Рыночный ордер с SL/TP:

def place_market_order(self, side: str, qty: float, symbol: str = None, stop: float = None, tp: float = None):
        side_param = "BUY" if side == "long" else "SELL"
        s = symbol or self.symbol
        pos_side = 'LONG' if side =='long' else 'SHORT'
        pos_side = 'BOTH' if pos_side_BOTH == True else pos_side

        params = {
            "symbol": s,
            "side": side_param,
            "positionSide": pos_side,
            "type": "MARKET",
            # ...
        }
        if stop is not None:
            params["stopLoss"] = json.dumps({"type": "STOP_MARKET", "stopPrice": stop, ...})
        if tp is not None:
            params["takeProfit"] = json.dumps({...})
        return self._request("POST", "/openApi/swap/v2/trade/order", params)

• set_multiple_sl/tp: Множественные стопы/тейки (для частичного закрытия).

Объяснение: API BingX требует HMAC-подписи. Метод place_market_order открывает позицию с прикреплёнными SL/TP. Для фьючей positionSide=«BOTH» (для хеджа, для однопозиционного режима — LONG/SHORT).

Шаг 4: Реалтайм бот (realtime.py)

Использует WebSocket Binance для свечей, BingX для ордеров. Обновляет DF, рассчитывает индикаторы, входит/выходит.

Ключевые части

Импорты и настройки:

import pandas as pd
import numpy as np
import time
import logging
from binance.client import Client as BinanceClient
from binance import ThreadedWebsocketManager
from binance.enums import *
from bingx_client import BingxClient

# Логи, ключи, параметры
# ...
binance_client = BinanceClient('', '')
bingx_client = BingxClient(BINGX_API_KEY, BINGX_API_SECRET, symbol=SYMBOL)

Добавим также функцию получения баланса:

# === ПОЛУЧЕНИЕ ТЕКУЩЕГО БАЛАНСА USDT НА BINGX ===
def get_usdt_balance():
    try:
        # BingX не имеет прямого метода баланса в твоём клиенте — добавим простой запрос
        path = "/openApi/swap/v2/user/balance"
        resp = bingx_client._request("GET", path)
        if resp and resp.get('code') == 0:
            for asset in resp['data']:
                if asset['asset'] == 'USDT':
                    return float(asset['availableBalance'])
        return 10.0  # fallback
    except Exception as e:
        logger.error(f"Ошибка получения баланса: {e}")
        return 10.0

Обработка свечи (process_candle): Когда свеча закрыта, добавляем в DF, рассчитываем индикаторы, проверяем сигналы:

def process_candle(msg):
    global df, position, entry_price, long_sl, long_tp, short_sl, short_tp
    global long_be_triggered, short_be_triggered, allow_long, allow_short, capital

    global df
    df = pd.concat([df, pd.DataFrame([new_row])], ignore_index=True)
    df = df.tail(700)  # Хватит для всех индикаторов

    conversionLine = (donchian_high(df['high'], conversionPeriods) + donchian_low(df['low'], conversionPeriods))/2
    baseLine = (donchian_high(df['high'], basePeriods) + donchian_low(df['low'], basePeriods)) / 2
    leadLine1 = (conversionLine + baseLine) / 2
    leadLine2 = (donchian_high(df['high'], laggingSpan2Periods) + donchian_low(df['low'], laggingSpan2Periods)) / 2

    spanA = leadLine1.shift(displacement - 1)
    spanB = leadLine2.shift(displacement - 1)
    df['kumoTop'] = np.maximum(spanA, spanB)
    df['kumoBottom'] = np.minimum(spanA, spanB)
    df['hasKumo'] = df['kumoTop'].notna() & df['kumoBottom'].notna()
    df['is_above_kumo'] = df['hasKumo'] & (df['close'] > df['kumoTop'])
    df['is_below_kumo'] = df['hasKumo'] & (df['close'] < df['kumoBottom'])
    #ДАЛЕЕ РАСЧИТЫВАЕМ ВСЮ ОСТАЛЬНУЮ ИНДИКАЦИЮ И МАТЕМАТИКУ

    #УСЛОВИЯ ВХОДА:
    last = len(df) - 1
    long_signal = (
      (df['cci_val'].iloc[last-1] < cci_long_thr) and (df['cci_val'].iloc[last] > cci_long_thr) and
      (~use_ichi_cloud or df['is_above_kumo'].iloc[last]) and
      (~use_ichi_lines or df['long_lines_pass'].iloc[last]) and
      (~use_ma_dir or df['long_ma_pass'].iloc[last]) and
      (~use_adx_filter or (df['adx'].iloc[last] >= adx_long_min and df['adx'].iloc[last] <= adx_long_max)) and
      (~use_rsi_filter or (df['rsi'].iloc[last] >= rsi_long_min and df['rsi'].iloc[last] <= rsi_long_max)) and
      (~use_natr_filter or (df['natr'].iloc[last] >= natr_long_min and df['natr'].iloc[last] <= natr_long_max)) and
      (~use_bbw_filter or df['bb_w'].iloc[last] >= bbw_min_trend)
          )

    short_signal = (
      (df['cci_val'].iloc[last-1] > cci_short_thr) and (df['cci_val'].iloc[last] < cci_short_thr) and
      (~use_ichi_cloud or df['is_below_kumo'].iloc[last]) and
      (~use_ichi_lines or df['short_lines_pass'].iloc[last]) and
      (~use_ma_dir or df['short_ma_pass'].iloc[last]) and
      (~use_adx_filter or (df['adx'].iloc[last] >= adx_short_min and df['adx'].iloc[last] <= adx_short_max)) and
      (~use_rsi_filter or (df['rsi'].iloc[last] >= rsi_short_min and df['rsi'].iloc[last] <= rsi_short_max)) and
    (~use_natr_filter or (df['natr'].iloc[last] >= natr_short_min and df['natr'].iloc[last] <= natr_short_max)) and
      (~use_bbw_filter or df['bb_w'].iloc[last] >= bbw_min_trend)
        )

    price = df['close'].iloc[-1]

Если наши long_signa либо short_signal исполняются (==True), то тогда уже открываем позицию. Делаем мы это следующим образом:

qty = risk_amount / price
logger.info(f"ОТКРЫВАЕМ LONG: {qty:.6f} {SYMBOL} по {price}")
resp = bingx_client.place_market_order("long", qty, stop=round(price * (1 - sl_pct / 100), 1), tp=round(price * (1 + tp_pct / 100), 1))

Ну и последнее — создаем функцию main(), где подключаемся к вебсокету бинанса для постоянного получения данных:

# === ЗАПУСК БОТА ===
def main():
    logger.info("Запуск бота Third Eye на BingX + Binance candles")

    # Загрузка истории
    global df
    klines = binance_client.get_klines(symbol=SYMBOL, interval=INTERVAL, limit=700)
    df = pd.DataFrame(klines, columns=['open_time', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'qav', 'trades', 'tbbav', 'tbqav', 'ignore'])
    df = df[['open_time', 'open', 'high', 'low', 'close']].astype(float)
    df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')

    # WebSocket Binance
    twm = ThreadedWebsocketManager()
    twm.start()
    twm.start_kline_socket(callback=process_candle, symbol=SYMBOL, interval=INTERVAL)

    logger.info("Бот запущен. Ожидание сигналов...")
    twm.join()


if __name__ == '__main__':
    main()

Шаг 5: Запуск и тестирование

Улучшения, которые можно внести в проект в будущем:

• Мониторинг: Telegram-бот для алертов.

• Оптимизация Walk-forward бектеста, перебор большего числа параметров.

Заключение

Мы создали полную систему: от идеи до деплоя. Логика основана на проверенных индикаторах, код — модульный. Это делает систему гибкой. Комбинация индикаторов и математики, сделанная мной — лишь один вариант. Возможно, можно подобрать даже лучшие комбинации. Но мы видим — эта система работает и работает грамотно.

Если вопросы есть вопросы — жду в комментариях. Удачных трейдов!