В этой статье разберём полный цикл разработки торговой системы: от формализации идеи до запуска реального бота на фьючерсной бирже. Проект состоит из двух частей: скрипта бэктеста (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 биржи.
Управление риском и ордерами.
Главное — дисциплина разработки. Алготрейдинг — это не про индикаторы, а про корректную архитектуру и контроль состояния.
Всегда рад ответить на любые ваши вопросы!
Комментарии (0)
Войдите или зарегистрируйтесь, чтобы оставить комментарий
Загрузка комментариев…