Привет, Хабр!
Сегодня мы разберём полный цикл создания торговой системы на 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 бектеста, перебор большего числа параметров.
Заключение
Мы создали полную систему: от идеи до деплоя. Логика основана на проверенных индикаторах, код — модульный. Это делает систему гибкой. Комбинация индикаторов и математики, сделанная мной — лишь один вариант. Возможно, можно подобрать даже лучшие комбинации. Но мы видим — эта система работает и работает грамотно.
Если вопросы есть вопросы — жду в комментариях. Удачных трейдов!
