В этой статье разберём полный цикл разработки торговой системы: от формализации идеи до запуска реального бота на фьючерсной бирже. Проект состоит из двух частей: скрипта бэктеста (Back.py), realtime-бота (Realtime.py).
Цель статьи — показать не только торговую идею, но и инженерную реализацию: архитектуру, контроль состояния, обработку данных, синхронизацию, и различие между backtest-движком и real-time исполнением.
Оба файла загружены на GitHub. Советую работать с файлом на протяжении статьи, так как не все функции описаны целиком, многие базовые моменты + зеркальное логика опущены.
1. Торговая гипотеза и концепция стратегии
В основе системы лежит классическая идея продолжения тренда после коррекции. Гипотеза формулируется следующим образом: если рынок формирует направленный импульс, то откат к уровню 0.5 по Фибоначчи от последнего импульса является вероятной зоной возобновления движения.
Речь идёт не о механическом построении сетки Фибоначчи на произвольных экстремумах, а о системной работе со структурой рынка. Структура определяется последовательностью экстремумов: Higher High и Higher Low для восходящего тренда, Lower High и Lower Low для нисходящего. Таким образом, тренд — это не индикаторное значение, а последовательность ценовых событий.
Импульс в рамках этой логики — это движение от подтверждённого swing low к swing high (или наоборот), зафиксированное алгоритмически. После формирования импульса вычисляется уровень 0.5:
Архитектура Back.py
Backtest реализован как пошаговая симуляция рынка. Его цель - показать реальный результат системы. Учтём комиссии, грамотно будем входить (на открытии следующей свечи) и отдавать приоритетность именно стоп-лоссу.
Загрузка исторических данных
Обычно данные подтягиваются через Binance API:
def fetch_klines(symbol='BTCUSDT', interval='5m', limit=1000): client = Client(api_key, api_secret) klines = client.get_klines(symbol=symbol, interval=interval, limit=limit) df = pd.DataFrame(klines, columns=[ 'open_time','open','high','low','close','volume', 'close_time','qav','trades','tbbav','tbqav','ignore' ]) df['open'] = df['open'].astype(float) df['high'] = df['high'].astype(float) df['low'] = df['low'].astype(float) df['close'] = df['close'].astype(float) return df
Здесь важно привести данные к float и исключить неиспользуемые поля. Для backtest достаточно OHLC.
Определение рыночной структуры
Структура вычисляется через анализ локальных экстремумов.
def detect_structure(df): highs = [] lows = [] for i in range(2, len(df) - 2): if df['high'][i] > df['high'][i-1] and df['high'][i] > df['high'][i+1]: highs.append((i, df['high'][i])) if df['low'][i] < df['low'][i-1] and df['low'][i] < df['low'][i+1]: lows.append((i, df['low'][i])) return highs, lows
Мы ищем swing-точки через простую фрактальную модель. В реальной версии для примера можно использовать ATR-фильтрацию или adaptive threshold, но базовая модель достаточна. Далее мы увидим её отработку.
Поиск импульса и расчет Fibonacci
После определения swing-точек необходимо выделить последний импульс.
def get_last_impulse(highs, lows): if len(highs) < 1 or len(lows) < 1: return None last_high_index, last_high = highs[-1] last_low_index, last_low = lows[-1] if last_high_index > last_low_index: return last_low, last_high else: return last_high, last_low
Теперь расчет уровня 0.5:
def calculate_fib_50(low, high): return low + (high - low) * 0.5
Получаем данные о последних свечах, а далее работаем с определением импульса, используя понятие свингов/фракталов/минимумов (кому как удобнее).
Логика входа в сделку
Backtest выполняется пошагово:
def run_backtest(df, commission_rate=0.00025): # 0.04% round-trip (Binance futures taker ~0.02% per side) tracker = MarketStructureTracker() trades = [] current_trade = None equity_curve = [] # список (timestamp, cumulative_pnl) current_equity = 0.0 position_size = 1.0 # 1 BTC (можно менять; PnL в USDT) for idx, row in df.iterrows(): i = df.index.get_loc(idx) # numeric position high = row['high'] low = row['low'] close = row['close'] tracker.add_candle(i, high, low)
Далее мы по очереди реализуем логику выходов, входов, трейлинга. Используем класс marketstructuretracker() для определения структуры - там мы ранее прописывали все функции.
В классических бектест системах я довольно редко вижу работу с трейлинг-стопом. Так как это основа моей стратегии, то функция трейлинга будет выглядеть следующим образом:
if current_trade: if current_trade['type'] == 'long': current_trade['max_price'] = max(current_trade['max_price'], high) if not current_trade['trailing_active'] and close > current_trade['impulse_high']: current_trade['trailing_active'] = True current_stop = current_trade['stop'] if current_trade['trailing_active']: trail_stop = current_trade['max_price'] - current_trade['trailing_distance'] current_stop = max(current_stop, trail_stop) if low <= current_stop: exit_price = current_stop gross_profit = (exit_price - current_trade['entry']) * position_size commission = commission_rate * (current_trade['entry'] + exit_price) * position_size net_profit = gross_profit - commission current_equity += net_profit trades.append({ 'entry_idx': current_trade['entry_idx'], 'entry_time': df.index[current_trade['entry_idx']], 'entry_price': current_trade['entry'], 'exit_idx': i, 'exit_time': idx, 'exit_price': exit_price, 'gross_profit': round(gross_profit, 2), 'commission': round(commission, 2), 'profit': round(net_profit, 2), 'type': 'long' }) equity_curve.append((idx, current_equity)) current_trade = None
Мы проверяем направления движения, ищем максимальную цену и сравниваем её с ценой активации трейлинга, чтобы активировать его при необходимость.
А уже далее выставляем трейлинг стоп и переставляем его в зависимости от дистанции, после чего сравниваем лой с этим стопом.
Часть входа в сделку же будет выглядеть следующим образом:
if not current_trade and tracker.current_impulse: impulse = tracker.current_impulse if impulse['high_price'] is None or impulse['low_price'] is None: continue # Только после формирования импульса (i > индексы экстремумов) if i <= impulse.get('high_index', -1) or i <= impulse.get('low_index', -1): continue if impulse['type'] == 'bull': fib_05 = impulse['high_price'] - 0.5 * (impulse['high_price'] - impulse['low_price']) if low <= fib_05 and close > fib_05: entry_price = close sl = impulse['low_price'] trailing_distance = 0.5 * (impulse['high_price'] - entry_price) current_trade = { 'type': 'long', 'entry': entry_price, 'entry_idx': i, 'stop': sl, 'trailing_distance': trailing_distance, 'max_price': high, 'impulse_high': impulse['high_price'], 'trailing_active': False, 'direction': 1 }
Определяем импульс, после чего при его формировании отмеряем fib_05 (уровень фибоначчи, половина импульса) и ждём прихода к этому уровню.
Когда подходим к этому уровню открываем нашу позицию.
Визуализация
Также для более грамотного понимания работы системы, я сделаю расчёт основных коэффициентов, а также выведу кривую капитала. С её помощью мы увидим стабильности стратегии и её выход.
За ряд бектестов я выявил наилучшую работу на активе ETHUSDT. Тестировать будем на отрезке в 50к часов, чтобы исключить любой скепсис. Использовать я буду библиотеку matplotlib, выглядит функция отрисовки таким образом:
def plot_equity_curve(equity_curve, title="Equity Curve (Cumulative Net PnL after Fees)"): if not equity_curve: print("Нет сделок — equity curve пустая") return dates = [t for t, e in equity_curve] equities = [e for t, e in equity_curve] plt.figure(figsize=(14, 7)) plt.plot(dates, equities, color='blue', linewidth=2, label='Cumulative Net PnL') plt.fill_between(dates, equities, color='blue', alpha=0.1) plt.title(title) plt.xlabel('Время') plt.ylabel('Накопленный Net PnL (USDT)') plt.grid(True, alpha=0.3) plt.legend() plt.tight_layout() plt.show() df = fetch_klines_paged("ETHUSDT", "1h", 50000) trades, stats, equity_curve = run_backtest(df)
Теперь давайте запустим скрипт и посмотрим на получившийся график:
Как вы можете видеть, получаем отличные результат. За приблизительно шесть лет стратегия сделала +25000$ с базовым активом 1 ETH на позицию. Эти результаты здесь показаны без учёта реинвестирования, что не позволяет графику 'улетать' очень высоко. Это просто объективный и качественный результат, который действительно можно достигнуть при базовом сценарии.
Также я вывел ряд интересных данных, выглядит эта статистика следующим образом:
============================================================ Символ: ETHUSDT Таймфрейм: 1h Всего баров: 50000 Всего трейдов: 4389 Общий PnL: +27582.57 USDT Winrate: 58.19% Avg Win: +35.24 Avg Loss: -34.02 Profit Factor: 1.44 Max Win: +641.79 Max Loss: -337.28
Архитектура Realtime.py
Realtime-движок принципиально отличается. Здесь есть:
API клиент
синхронизация по времени
контроль открытых позиций
защита от повторного входа
управление ордерами
Для начала, введём class BingxClient. С ним мы будем работать. Для работы в реалтайме буду использовать биржу bingX, ввиду небольших комиссий и неплохого API.
Получение данных в реальном времени
def get_klines(self, interval="1h", limit=1000): path = "/openApi/swap/v2/quote/klines" params = { "symbol": self.symbol, "interval": interval, "limit": str(limit), 'timestamp': int(time.time() * 1000) } data = self._public_request(path, params) if data.get('code') == 0: return data.get('data', []) logging.error(f"Klines error: {data.get('msg')}") return []
Как и в первом случае, получим OHLC данные для работы в реальном времени. Функция здесь уже немного отличается, но в любом случае на выходе будет то же самое.
Проверка открытой позиции
def has_open_position(client, symbol): positions = client.get_positions(symbol=symbol) for p in positions: if float(p['positionAmt']) != 0: return True return False
В реальном коде обязательно нужно проверять направление, размер и маржу. Делается это для безопасности. Функция get_positions() возвращает все открытые позиции. С помощью перебора находим позицию именно по нашему символу.
Открытие сделки
def place_market_order(self, side: str, qty: float, stop: float = None): side_param = "BUY" if side == "long" else "SELL" positionSide = 'LONG' if side == "long" else 'SHORT' params = { "symbol": self.symbol, "side": side_param, "positionSide": positionSide, "type": "MARKET", "quantity": str(qty), "recvWindow": "5000", } if stop is not None: params["stopLoss"] = json.dumps({ "type": "STOP_MARKET", "stopPrice": str(stop), "workingType": "MARK_PRICE" }) return self.send_request("POST", "/openApi/swap/v2/trade/order", params) .... resp = client.place_market_order("short", QTY, stop=sl)
Функции открытия, постановки ордеров и прочее есть в моём клиенте для биржи, всегда прикрепляю его на github.
Логика тут простая - функции просто обращаются к API, чтобы открыть позицию в заданном направлении на заданный объём, после чего выставляют стоп-лосс и тейк профит
Главные циклы бота
Что же касается трейлинга, то здесь я решил не завязываться на API, а мониторить его через код механически. Безусловно, у этого способа есть и ряд минусов, но если брать то, что у нас есть стоп-лосс, то мы здесь можем себе это позволит.
Такой подход позволит избежать закрытия позиций на свипах. Реализация через код следующая:
def monitor_trailing(): global current_trade while True: if current_trade: mark = client.get_mark_price() if mark is None: time.sleep(TRAIL_POLL_SEC) continue if current_trade['type'] == 'long': current_trade['max_price'] = max(current_trade['max_price'], mark) if not current_trade['trailing_active'] and mark > current_trade['impulse_high']: current_trade['trailing_active'] = True logging.info("Trailing активирован (long)") if current_trade['trailing_active']: trail_stop = current_trade['max_price'] - current_trade['trailing_distance'] current_trade['stop'] = max(current_trade['stop'], trail_stop) if mark <= current_trade['stop']: client.close_position() logging.info(f"Trailing hit LONG → закрытие @ {mark:.2f}") current_trade = None
Основной цикл работы выглядит так:
if __name__ == "__main__": logging.info("Запуск реал-тайм скрипта BingX (ETH-USDT 1h)") load_initial_history() # Запускаем мониторинг trailing в отдельном потоке threading.Thread(target=monitor_trailing, daemon=True).start() while True: try: check_new_candle() time.sleep(CANDLE_POLL_SEC) except KeyboardInterrupt: logging.info("Остановка по Ctrl+C") if current_trade: client.close_position() break except Exception as e: logging.error(f"Ошибка в главном цикле: {e}") time.sleep(30)
5. Что важно с инженерной точки зрения
Стратегия должна быть функцией от данных, а не от состояния глобальных переменных.
Backtest и Realtime должны использовать одинаковую торговую логику.
Нельзя допускать look-ahead bias в бектесте.
Синхронизация времени обязательна.
Ордер-менеджмент — отдельный модуль.
Всё это учтено в бектесте и реалтайм боте. Есть полная синхронизация логики, всё грамотно работает. Сейчас бот тестируется мной в реалтайм и показывает положительный результат, который сейчас действительно соответствует бектесту.
6. Итог
Проект представляет собой полноценный цикл разработки алгоритмической стратегии:
Формулировка гипотезы (трендовый трейдингчерез 0.5 Fibonacci).
Реализация backtest движка.
Проверка статистики.
Перенос логики в realtime-движок.
Интеграция с API биржи.
Управление риском и ордерами.
Главное — дисциплина разработки. Алготрейдинг — это не про индикаторы, а про корректную архитектуру и контроль состояния.
Всегда рад ответить на любые ваши вопросы!
