Крипто-ассистент — самописный аналитический сервис под биржи (spot + USDT-perpetual, то есть и обычная торговля, и бессрочные фьючерсы).

Мой большой  проект, над которым очень долго работал, — 7 месяцев трудов. Статья будет очень большой, и это тоже моя первая статья на Хабре. Итак, вернёмся к сути.

Неторговый бот, который сам анализирует и не начинает мне ХХХХ — доход клепать, совсем нет, он мой советник. Он собирает рыночные данные, считывает индикаторы, прогоняет их через ML-модель (т.е. через обученный алгоритм, который пытается предсказать вероятность движения цены), проверяет кучей фильтров и присылает в Telegram карточку сигнала вида:

«BTCUSDT 15m LONG @ 64321, SL=…, TP=…, score=…»

Таким образом, приходит информация в Telegram:

А вот общая блок-схема, специально упрощённая, для трейдеров и непрофильных читателей. Чтобы стало понятно, о чём вообще речь, прежде чем начну погружать вас в технические детали:

Фичи — это индикаторы, на которых стоят столбы моего творения. Долго провозился над тем, чтобы для себя сформулировать, за что именно отвечает каждый из этих показателей. Вот итоговый набор:

• EMA 9/20/50/200, SMA 20 - скользящие средние с разным периодом. Показывают, куда смотрит цена «в среднем» на разных горизонтах.

• MACD (12, 26, 9), гистограмма — классический индикатор схождения/расхождения скользящих, помогает ловить смену импульса.

• Bollinger (20, 2), ширина канала-границы «нормального» движения цены и насколько они сейчас сжаты.

• ATR + ATR% к цене— средний истинный диапазон, по сути «насколько свечи большие». Дальше я использую его как универсальную меру волатильности.

• RSI, ADX, OBV -индекс относительной силы, сила тренда и накопленный объём с учётом направления.

• z-score по close и volume-отклонение цены и объёма от своего скользящего среднего в «сигмах». Помогает понять, насколько ситуация «нестандартная».

• trend_score ∈ [-1; 1] мой сводный показатель тренда, собирается из структуры EMA и наклона EMA50. Простой способ ответить на вопрос «насколько уверенно идёт движение». Я использовал метод «тройного барьера» из книги Advances in Financial Machine Learning (2018)* Автор: Marcos López de Prado, если кому-то нужна эта книжка, я поделюсь. Идея была такая: Каждой свече мы задаём вопрос: «слышь, мне открывать сделку сейчас или подождать, пока цена не рухнет, или войти в сделку и она достигнет тейк-профита». А здесь моделька-мудмазелька сразу же ответ даёт: становится меткой «y», на которой потом учится модель классификации (предсказывать направление будущего движения). Представляю вашему вниманию часть кода.

def trend_score_from_emas(df: pl.DataFrame) -> pl.DataFrame: # 1) Флаговая структура EMA: +1 / -1 / 0 df = df.with_columns([ pl.when((pl.col(“ema_20”) > pl.col(“ema_50”)) & (pl.col(“ema_50”) > pl.col(“ema_200”))).then(1.0) .when((pl.col(“ema_20”) < pl.col(“ema_50”)) & (pl.col(“ema_50”) < pl.col(“ema_200”))).then(-1.0) .otherwise(0.0).alias(“ema_stack”) ])

# 2) Наклон EMA50 за 3 бара. 
# Делю на саму EMA50, чтобы было относительно цены, а не в абсолюте.
# Защита от деления на ноль обязательна, вспоминаю как спот возвращал мне нули, я вообще не понимаю что происходит
slope_raw = (pl.col("ema_50") - pl.col("ema_50").shift(3))
denom = pl.when(pl.col("ema_50").shift(3).abs() < 1e-12)\
          .then(1e-12)\
          .otherwise(pl.col("ema_50").shift(3))
df = df.with_columns([(slope_raw / denom).alias("ema50_slope3")])

# 3) Зажимаем наклон в окно ±2% и нормируем к [-1; +1]
# 2% за 3 четверти часа если резкое движение
df = df.with_columns([
    (_clip(pl.col("ema50_slope3"), -0.02, 0.02) / 0.02).alias("ema50_slope3_n")
])

# 4) Сводный балл: 70% структура EMA + 30% наклон. 
# Зажимаем в [-1; +1] на всякий случай могут вылезти кривые значения
df = df.with_columns([
    _clip(0.7 * pl.col("ema_stack") + 0.3 * pl.col("ema50_slope3_n"), -1.0, 1.0)
    .alias("trend_score")
])
return df

p.s простите, пожалуйста, модераторы, но у меня не получилось нормально вставить часть кода пришлось картинку прикрепить…

vol_regime: low/mid/high-режим волатильности по ATR. Типа, что с рынком? «Он спит?Умер? Бодр и свеж?»

squeeze play- флаг сжатия Боллинджера. Узкие BB часто предшествуют резкому движению.

ange_index- попытка отличить тренд от флэт (бокового движения).

Дополнительно для младшего таймфрейма 15m подмешиваю значение trend_score_4h (т.е тренд 4-часового таймфрейма). Делаю это через numpy.searchsorted- это просто быстрее и предсказуемее, чем join_asof из Polars. Так, модель видит контекст старшего ТФ, не теряя скорости.

Volume Profile - это профиль объёма вместо обычной свечной гистограммы строится распределение торгового объёма по уровням цены. Считаю его по 90-дневному окну, нахожу уровень с максимальным объёмом и диапазон, где прошло 70% торгов. Границы этого диапазона — вверх и вниз. Затем для каждой свечи записываю дистанции до этих уровней в единицах ATR -это потом используется и как фича, и как фильтр против входа «в стену».

Цель-мечта!

Избавиться от ручного мониторинга десятка пар на нескольких таймфреймах. Хотел получать свежую информацию из биржи и с возможностью успешно торговать — ручной труд не приносил мне нужного интереса, была усталость. В результате пришла в голову идея, почему бы не получать истории с биржи и не отправлять обработанные сигналы в Telegram? Тем самым видеть, в какие позиции сто́ит заходить, и играть на рынке более осознанно.

Что в итоге умеет система:

24/7 слушает рынок по WebSocket (это протокол постоянного соединения, через который биржа сама шлёт свежие свечи) и догружает историю по REST (одиночные HTTP-запросы для исторических данных).

Сама считывает фичи (EMA, RSI, MACD, ATR, BB, ADX, OBV, trend_score, режим волатильности). Выше немного описал, что это.

Оценивает вероятность движения вверх/вниз на горизонте H баров с помощью обученной модели - LightGBM (это градиентный бустинг, штука посерьёзнее) или логистическая регрессия + температурная калибровка (попроще, но быстрее).

Не лезет в каждое шевеление: есть фильтры (TrendAlign, SR-guard, Flip-guard, Volume Profile, кулдаун, Freeze на новостях/макро). Подробнее про каждый будет ниже.

Шлёт в Telegram только те карточки, где есть уровни SL/TP и пройдены все проверки. Считает ex-post по истории сигналов-winrate (доля прибыльных), RR (risk/reward, отношение прибыли к риску), ожидание. Чтобы я мог посмотреть, не врёт ли мой движок самому себе. А такое реально бывает, когда движок «в моменте» выглядит хорошо, а на истории сливает.

Отдельно есть Advisor- это наблюдатель/страж, который раз в N минут сканирует 4h/1d и ищет среднесрочные свинги (тренд 1d + swing 4h + удалённость от SR + брейкаут). Про него тоже расскажу подробно. И всё это из одной админки на Streamlit мне было лень каждый раз лазить в YAML-конфиг и вспоминать CLI-команды.

Стек Python 3.11, asyncio (асинхронный код, чтобы обрабатывать события с биржи без блокировок). Polars + PyArrow, для всего, что касается датафреймов и parquet-файлов.

aiohttp + websockets + tenacity, для общения с биржей (HTTP-запросы, WebSocket-подписки и автоматические повторы при сбоях).

scikit-learn + LightGBM-модели машинного обучения.

Pydantic v2 + YAML конфигурация. YAML - это просто текстовый формат с настройками, Pydantic проверяет, что все значения правильного типа.

Streamlit- для веб-админки (без неё пришлось бы всё делать через консоль, что для постоянного использования больно).

Prometheus client метрики, чтобы видеть состояние сервиса.

Хранилища как такового отсутствует, я специально не стал разворачивать базу данных, всё лежит в файлах формата parquet. Мне хватает.

Ядро системы

Ниже будет блок-схема. Она разделена на три части. Большая схема целиком не вмещалась по размерам (пробовал сжимать, не получилось без потери читаемости). Поэтому идём по этапам.

Часть 1. Сбор и подготовка данных

Описание по схеме:

Источник данных биржи. Свечи 15-минутного таймфрейма приходят двумя путями: в реальном времени по WebSocket и в виде истории через REST API. Слой ингеста (ws.py+ history.py) принимает их, проверяет на подтверждённость. Всё ок записываем, не ок не трогаем.

Хранение.Все свечи складываются в «partitioned parquet» это просто способ разложить файлы по папкам так, чтобы поиск нужного куска данных был мгновенным. Разбивка по символу, таймфрейму и категории (spot/linear). Даёт быстрый доступ и компактный размер на диске. Иногда думаешь, надо всё-таки базу.

Ресемплер.На основе 15-минуток автоматически собираются старшие таймфреймы: 30m, 1h, 4h и 1d. Никаких отдельных закачек — всё считается из базового ТФ. Это гарантирует, что данные между таймфреймами согласованы. Другие варианты (закачивать каждый ТФ отдельно) я перестал даже рассматривать после первого расхождения по нескольким барам.

Контекст деривативов. Параллельно тянутся данные с биржи v5: ставка финансирования (funding rate комиссия, которую лонги платят шортам или, наоборот), открытый интерес (OI - сколько всего открытых позиций) и ликвидации (когда чьи-то позы принудительно закрывают). Всё это складывается в отдельный кэш cache/context.

Volume Profile. По 90-дневному окну считается профиль объёма — уровни. Вверх, вниз и точка наибольшего объёма. Эти уровни потом используются и в фичах, и в фильтрах движка.

Расчёт фич.На каждом таймфрейме считаются индикаторы (EMA, MACD, BB, ATR, RSI, ADX, OBV), режимы волатильности, флаги сжатия Боллинджера, trend_score. Сюда же подмешивается контекст деривативов и дистанции до VP-уровней. Результат — единый датасет фич в cache/features.

Вывод этого этапа: для каждой свечи, каждого символа есть строка с 30+ признаками, готовая к ML.

Кстати, ингест данных, для тех, кто не сталкивался с термином, — это просто процесс сбора, обработки и импорта данных для анализа и хранения. Можно думать о нём как о «приёмка товара на склад» проверили, сверили, что не битый, разложили по полкам, теперь можно работать.

Часть 2. Обучение модели и движок принятия решений

Левая часть — схемы-обучение модели. Схема была большая, я постарался всё разбить, чтобы описать в статье по каждому блоку отдельно. Очень надеюсь, что получилось.

Сборка ML-таблицы. Из кэша фич собирается обучающий датасет. Каждой свече присваивается метка по методу triple-barrier, для каждого бара выставляются верхний тей профит и нижний стоп лос. Барьеры в ATR, и через H баров в будущем смотрим, какой барьер был пробит первым. Это даёт честную бинарную классификацию: «выгоднее был лонг» /«выгоднее был шорт». Если ни тот, ни другой не пробит за H баров — этот бар выбрасываем как «нежелательный».

Обучение. В зависимости от настроек, которые указываются в интерфейсе админки: LightGBM (тяжёлая модель с ранней остановкой по метрике AUC, точнее, но медленнее) или LogisticRegression + StandardScaler (лёгкая, быстрая, без зависимостей; подойдёт, если LightGBM почему-то не ставится).

Калибровка температурой T. После обучения подбирается коэффициент T, усиливающий «стойкость» модели (в мире ML называют логитами), чтобы предсказанные вероятности соответствовали реальной частоте событий. Замеряются специальной мерой так, насколько модель адекватна в своих оценках до и после, для этого. Без этого шага модель часто переуверена и начинает «галлюцинировать» — выдавать 90% уверенности там, где по факту 60%. У меня как-то упало с 0.12 до 0.04, и это сразу стало заметно по качеству сигналов.

Модель и метаданные (список фич, барьеры, метрики, T) сохраняются в папку models, потом можно запустить для обучения модельки.

Правая часть схемы — движок в реальном времени.

На каждом закрытом баре движок берёт последние фичи символа из кэша, прогоняет через сохранённую модель, получает рост и падение.

Контекст подмешивается тут же: данные деривативов и состояние Freeze (новости/макро) влияют на решение. Для новостей я использую API CryptoPanic - он отдаёт ленту по криптовалютам, и если попадает что-то критичное (взлом, делистинг, расследование), движок ставит «заморозку» на N минут и в это время никаких новых входов не делает.

Динамические пороги. Базовые лонг/шорты/gap умножаются на множители режима-волатильности, типа рынка (trend/flat) и торговой сессии (ASIA/EU/US). Это позволяет одной и той же модели адаптироваться к разным условиям. На спокойном азиатском рынке пороги мягче, но бывают не всегда, а вот открытие Нью-Йорка — ускоренно всё просто.

Цепочка фильтров. Она выглядит примерно так, Freeze — Cooldown — TrendAlign — SR-guard — VP-filter — Flip-guard — проверка score. Если хоть один фильтр сказал «нет» — сигнал превращается в ожидание с указанием причины. Жди тоже логируется, и в админке потом смотрю, какой фильтр чаще всего режет сигналы. Иногда вижу, что половину входов выбрасывает Flip-guard - и думаю, может, я его слишком жёстко настроил.

**Вывод **этого этапа:если все проверки пройдены — формируется карточка лонг/шорт с уровнями СЛ/ТП, рассчитанными по ATR, и отправляется в Telegram. Все события (и сигналы, и отказы) пишутся в журнал data/signals сразу в двух форматах:— parquet (для быстрой аналитики) и CSV (для быстрого открытия в Excel глазами).

Часть 3. Анализ результатов и Advisor

Эта схема про то, как система проверяет сама себя и работает на средних горизонтах. Описание, что имею в виду. Левая ветка — ex-post симулятор.

Ex-post мы берём уже выданные сигналы и симулируем, как было бы «если б стали миллионерами» или, наоборот, «хорошо, что не зашёл»

Источник. Берётся журнал сигналов из data/signals и исторические 15-минутные свечи из кеша.

Симуляция исполнения. Для каждого сигнала лонг/шорт симулирует реальный вход — по открытию следующего бара после сигнала (чтобы не подсматривать в будущее) — это Ex-post.

Метрики на выходе:winrate (% прибыльных сделок), средний RR (риск/ревард), ожидаемость в RR, медианное время удержания, разбивка по символам и действиям. Результаты сохраняются в data.

Зачем это вообще нужно? Без ex-post легко повестись на разводняк, что движок работает хорошо. С ex-post видишь правду, где модель ошибается, галлюцинация есть или её нет? Какие фильтры режут прибыль, а какие, наоборот, спасают от потерь. Это, наверное, самая отрезвляющая часть всего проекта. Это о чём выше написал.

Правая ветка — это Advisor(страж). Я о нём вышел, писал.

Параллельный сервис. Advisor работает независимо от основного движка(добавил службу, чтобы они не мешали друг друга) и сканирует фичи на старших таймфреймах (4h и 1d) с заданным интервалом (по умолчанию раз в 5 минут).

Логика поиска свингов. Необходимо найти сочетание тренда на днёвки, если достаточный свинг на 4ч, тогда необходимо создать свинг-карточку.

Вывод. Карточки уходят в Telegram (отдельным потоком от основных сигналов — я разделил, чтобы случайно не спутать) и в журнал data/advisor. Это нужно, чтобы основной движок ловил локальные входы на 15m, а Advisor-крупные свинги на днёвке. У них разные ритмы и разные цели, поэтому они разнесены по разным процессам.

Streamlit-админка связывает всё воедино. Из неё можно запустить и остановить движок и Advisor, посмотреть журналы, прогнать ex-post за нужный период, переобучить модель и применить новые настройки.

Ядро код как это выглядит Чтобы модель училась, ей нужны правильные ответы на вопросы: «Вот свеча/бар. Что надо тебе делать — Лонг или шорт?» Попробую ответить, что делает этот алгоритм. Берём каждую свечу из истории (здесь у нас большая будет история, 180 дней) и смотрим: — Цена растёт? Она на нужной величине — значит, надо лонговать — Цена упала на нужную величину — значит, надо было продавать Ни то, ни другое за отведённое время — ожидание, — ситуация неясна. Ждём точки входа. Вот и всё. Прохожусь так по всей истории, получаю миллион примеров с готовыми ответами. Скармливаю всю историю модели. Модель учится и потом на живых данных пытается угадать ответ сама. Угадала — молодец, получает развитие, если несколько раз промахнулась, переучивается. Почему не сделать проще — наблюдать выросла цена или нет? Потому что это обман. Цена могла сначала упасть и выбить стоп-лосс, а потом скакануть вверх, как они любят. По простой проверке получается «угадал». По факту — уже давно вышел с убытком. Способ модели на честность: смотрю, что происходило по истории, а не только в конечной точке.

def triple_barrier_label(df: pl.DataFrame, horizon: int, up_k: float, dn_k: float) -> pl.DataFrame: df = df.sort(["symbol", "ts"])

if "atr" not in df.columns:
    print("[triple_barrier] atr нет в df, барьеры будут нулевые")

parts = []
for g in _partition_by(df, "symbol"):
    n = len(g)
    if n < 2:
        continue

    close = g["close"].to_numpy()
    high = g["high"].to_numpy()
    low = g["low"].to_numpy()

    if "atr" in g.columns:
        atr = g["atr"].fill_null(strategy="forward").fill_null(0.0).to_numpy()
    else:
        atr = np.zeros(n)

    y = np.zeros(n, dtype=np.int8)
    tp_px = np.full(n, np.nan)
    sl_px = np.full(n, np.nan)

    for i in range(n - 1):
        a = atr[i]
        c = close[i]
        if a <= 0 or np.isnan(c):
            continue

        up = c + up_k * a
        dn = c - dn_k * a
        tp_px[i] = up
        sl_px[i] = dn

        # сколько баров реально доступно вперёд
        end = i + 1 + horizon
        if end > n:
            end = n

        up_j = -1
        dn_j = -1
        for j in range(i + 1, end):
            if up_j < 0 and high[j] >= up:
                up_j = j
            if dn_j < 0 and low[j] <= dn:
                dn_j = j
            if up_j >= 0 and dn_j >= 0:
                break

        # метод FIFO
        if up_j < 0 and dn_j < 0:
            continue  # y[i] уже 0
        if dn_j < 0:
            y[i] = 1
        elif up_j < 0:
            y[i] = -1
        else:
            y[i] = 1 if up_j < dn_j else -1

    parts.append(
        g.select(["ts", "symbol", "tf"]).with_columns(
            pl.Series("y", y),
            pl.Series("tp_price", tp_px),
            pl.Series("sl_price", sl_px),
        )
    )

if not parts:
    # тут пустой вход
    return df

labels = pl.concat(parts)
return df.join(labels, on=["symbol", "ts", "tf"], how="left")

Запомнил на всю жизнь: нельзя смешивать разные символы в один проход, если вдруг что-то смешалось, то метрики получатся ложные. Калибровка температурой уникальнейшая штука. Без неё вообще никак.

def sigmoid(x: np.ndarray) -> np.ndarray: x = np.asarray(x, dtype=np.float64) out = np.empty_like(x) pos = x >= 0 neg = ~pos out[pos] = 1.0 / (1.0 + np.exp(-x[pos])) ex = np.exp(x[neg]) out[neg] = ex / (1.0 + ex) return out def temperature_scale(logits: np.ndarray, T: float) -> np.ndarray: # делим логиты на T — при T>1 вероятности «сжимаются» к 0.5 # при T<1 наоборот, становятся более крайними return sigmoid(logits / T) def grid_search_temperature(logits_valid, y_valid, Tmin=0.6, Tmax=2.5, step=0.05): # Тупой перебор по сетке. На валидации, чтобы не подсматривать в трейн. best_T, best_ll = 1.0, float(“inf”) for T in np.arange(Tmin, Tmax + 1e-9, step): p = temperature_scale(logits_valid, T) ll = log_loss(y_valid, p, labels=[0, 1]) if ll < best_ll: best_ll, best_T = ll, T return float(best_T)

Интерфейс программы простой

Надеюсь, вы дочитали до конца! Цели на будущее: Модернизировать проект буду, на очереди биржи Binance и BINGX, кстати, вот что заметил, по некоторым монетам, цена у бирж отличается. Поэтому нужно доработать механизм, чтобы отображались разные цены на сделки.