Как создать маркет-сканер волатильности криптовалют с кластеризацией аномалий

🔍 Что это и зачем нужно?

В криптовалютной торговле время решает всё. Если вы первым заметите резкое движение цены или объёма — это может стать началом пампа, инсайдерской закупки или просто короткого арбитражного окна. Но ручной мониторинг десятков монет невозможен. Именно поэтому я разработал скрипт-сканер волатильности, который:

• Подключается к Binance через API

• Анализирует данные по десяткам криптовалют

• Находит резкие изменения цены и объёма

• Применяет кластеризацию (DBSCAN) для выявления аномалий

• Показывает только те монеты, которые реально выбиваются из общего движения

⚙️ Установка и настройка

Перед запуском скрипта убедитесь, что у вас установлен Python 3. Установите необходимые библиотеки:

pip install ccxt pandas scikit-learn numpy

Что это за библиотеки

• ccxt — библиотека для подключения к криптовалютным биржам

• pandas — для обработки табличных данных

• scikit-learn — машинное обучение, включая кластеризацию

• numpy — численные операции и массивы

🧠 Как работает скрипт

Разберём код пошагово, чтобы понять его логику.

Импорт библиотек

import ccxt
import pandas as pd
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
import numpy as np

Здесь мы подключаем всё необходимое — от API-библиотеки ccxt до модулей для машинного обучения.

Загрузка рынков с Binance

exchange = ccxt.binance()
symbols = [s['symbol'] for s in exchange.load_markets().values() if '/USDT' in s['symbol'] and s['active']]

📌 Мы выбираем все активные торговые пары с USDT, т.к. это основная ликвидная валюта на Binance.

Сбор данных по монетам

data = []

for symbol in symbols[:50]:  # Ограничиваем до 50 монет для скорости
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, '5m')[-3:]  # последние 15 минут
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        pct_change = (df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0]
        vol_change = df['volume'].iloc[-1] - df['volume'].iloc[0]
        data.append([symbol, pct_change, vol_change])
    except:
        continue

🟢 Здесь мы:

• Получаем OHLCV-данные (Open, High, Low, Close, Volume) с интервалом в 5 минут

• Считаем:

  • Изменение цены (pct_change) — процент роста или падения

  • Изменение объёма (vol_change) — насколько вырос/упал объём за последние 15 минут

Формируем DataFrame

df = pd.DataFrame(data, columns=['symbol', 'pct_change', 'vol_change'])

📊 Получаем таблицу с монетами и двумя метриками — готово к машинному обучению.

Кластеризация с DBSCAN

X = StandardScaler().fit_transform(df[['pct_change', 'vol_change']])
clustering = DBSCAN(eps=0.8, min_samples=3).fit(X)
df['cluster'] = clustering.labels_

🧠 Здесь начинается магия:

• Стандартизация (StandardScaler) нужна, чтобы pct_change и vol_change были в одной шкале

• DBSCAN (Density-Based Spatial Clustering) определяет группы монет, которые похожи по динамике

• Те монеты, которые не попали ни в один кластер, получают cluster = -1 — это аномалии

Выводим только аномальные монеты

anomalies = df[df['cluster'] == -1]
print("📊 Обнаружены аномалии:\n", anomalies)

🔥 Это и есть те монеты, которые ведут себя не как остальные. Например:

• Резко выросли, в то время как другие падают

• Подняли объём при стабильной цене

• Повторяют паттерны пампа

📈 Зачем это нужно?

Такой инструмент — это не просто индикатор, а целый интеллект-модуль. Он может быть полезен для:

• 🚀 Поиска пампов на ранней стадии

• 🕵️ Выявления "инсайдерских" закупок

• ⚠️ Поиска арбитражных и нелогичных движений

• 🤖 Автоматических сигналов для Telegram-бота или скальпинг-бота

✅ Вывод

Маркет-сканер на Python с использованием кластеризации — это настоящий криптовалютный радар, который поможет вам:

• Быть на шаг впереди толпы

• Работать с аномалиями, а не с шумом

• Создавать собственные сигналы и стратегии

Если вы хотите получить больше таких инструментов, подписывайтесь на itsfera.pro — здесь я делюсь реальными Python-решениями для трейдинга, крипты и автоматизации.

import ccxt
import pandas as pd
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler
import numpy as np

exchange = ccxt.binance()
symbols = [s['symbol'] for s in exchange.load_markets().values() if '/USDT' in s['symbol'] and s['active']]

data = []

for symbol in symbols[:50]:  # Ограничиваем до 50 монет
    try:
        ohlcv = exchange.fetch_ohlcv(symbol, '5m')[-3:]  # последние 15 минут
        df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        pct_change = (df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0]
        vol_change = df['volume'].iloc[-1] - df['volume'].iloc[0]
        data.append([symbol, pct_change, vol_change])
    except:
        continue

df = pd.DataFrame(data, columns=['symbol', 'pct_change', 'vol_change'])

# Нормализация и кластеризация
X = StandardScaler().fit_transform(df[['pct_change', 'vol_change']])
clustering = DBSCAN(eps=0.8, min_samples=3).fit(X)
df['cluster'] = clustering.labels_

# Отображаем аномалии
anomalies = df[df['cluster'] == -1]
print("📊 Обнаружены аномалии:\n", anomalies)