Привет, Хабр!
Сегодня мы разберём полный цикл создания торговой системы на 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).
Система разделена на три файла:
main.py — бэктест и оптимизация.
bingx_client.py — клиент для API BingX (открытие/закрытие ордеров).
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) }
Краткое резюме логики работы
Один вызов — одна комбинация параметров + один символ
Загружаем данные → считаем все индикаторы один раз
Проходим по каждой свече и имитируем реальное поведение трейдера: сначала проверяем, не пора ли выходить (SL/TP/BE), потом проверяем, можно ли войти (сигнал + нет позиции + разрешено флагом)
Риск-менеджмент реализуется через динамический расчёт размера позиции: всегда 1% от текущего капитала
Всё состояние (позиция, уровни 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), но в данной ситуации перебора вполне достаточно. Результат: Топ конфигураций по прибыли, график кривой капитала.
Теперь давайте запустим бектест. Мы получим кривую капитала, список всех сделок, параметры. По итогу получаем лучшими параметрами:
tp_pct = 5.0 sl_pct = 1.0 be_trig_pct = 2.0 cci_length = 25 adx_long_min = 16
Кривая капитала впечатлает:
Ну в .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: Запуск и тестирование
Установите библиотеки: pip install pandas numpy matplotlib python-binance requests.
Вставьте ключи.
Бэктест: python main.py — получите топ и график.
Реалтайм: python realtime.py — бот запустится, логи в консоли.
Тестируйте на малом капитале или демо.
Улучшения, которые можно внести в проект в будущем:
Мониторинг: Telegram-бот для алертов.
Оптимизация Walk-forward бектеста, перебор большего числа параметров.
Заключение
Мы создали полную систему: от идеи до деплоя. Логика основана на проверенных индикаторах, код — модульный. Это делает систему гибкой. Комбинация индикаторов и математики, сделанная мной — лишь один вариант. Возможно, можно подобрать даже лучшие комбинации. Но мы видим — эта система работает и работает грамотно.
Если вопросы есть вопросы — жду в комментариях. Удачных трейдов!
