Розробка скринера за волатильністю

Розроблення скринера з волатильності криптовалют

Скринер волатильності — це інструмент для пошуку активів з аномальним рухом ціни або змінилися характеристиками волатильності. Трейдери використовують його для пошуку торговельних можливостей: пробої, моментум-стратегії, стратегії mean-reversion. Технічно це pipeline збору даних, розрахунку метрик та real-time веб-інтерфейсу.

Метрики волатильності

Реалізована волатильність (Realized Volatility)

Найпоширеніша метрика — історична волатильність на основі log-returns:

import numpy as np
import pandas as pd
from typing import Optional

def realized_volatility(
    closes: pd.Series,
    window: int = 24,
    annualize: bool = True,
    periods_per_year: int = 8760  # для годинних даних
) -> pd.Series:
    """
    Розраховує реалізовану волатильність як стандартне відхилення log-returns.
    window: кількість періодів (свічок)
    """
    log_returns = np.log(closes / closes.shift(1))
    rv = log_returns.rolling(window).std()

    if annualize:
        rv = rv * np.sqrt(periods_per_year)

    return rv * 100  # у відсотках

ATR (Average True Range)

ATR більш стійкий до гепів та внутрішньоденного діапазону:

def atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series:
    prev_close = close.shift(1)
    true_range = pd.concat([
        high - low,
        (high - prev_close).abs(),
        (low - prev_close).abs()
    ], axis=1).max(axis=1)

    return true_range.ewm(span=period, adjust=False).mean()

def atr_percent(high, low, close, period=14) -> pd.Series:
    """ATR нормалізований до ціни — зіставний між різними активами"""
    return atr(high, low, close, period) / close * 100

Ширина смуг Болінджера (Bollinger Band Width)

Стиснення Bollinger Bands — сигнал предстоячого пробою:

def bollinger_bandwidth(close: pd.Series, window: int = 20, num_std: float = 2.0) -> pd.Series:
    rolling_mean = close.rolling(window).mean()
    rolling_std = close.rolling(window).std()

    upper = rolling_mean + num_std * rolling_std
    lower = rolling_mean - num_std * rolling_std

    # Ширина смуг нормалізована до середної лінії
    return (upper - lower) / rolling_mean * 100

Архітектура скринера

Data pipeline

Binance/OKX WebSocket ──► Candle Aggregator ──► Redis (latest N candles)
                                                        │
                                          VolatilityCalculator (кожні 60s)
                                                        │
                                             PostgreSQL / TimescaleDB
                                                        │
                                              REST API + WebSocket
                                                        │
                                              Frontend (React table)

Розрахунок по всьому ринку

class VolatilityScreener:
    def __init__(self, symbols: list[str]):
        self.symbols = symbols

    async def calculate_screener_data(self) -> list[ScreenerRow]:
        results = []

        for symbol in self.symbols:
            candles = await self.get_candles(symbol, interval='1h', limit=168)  # 7 днів
            df = pd.DataFrame(candles, columns=['time', 'open', 'high', 'low', 'close', 'volume'])

            if len(df) < 24:
                continue

            row = ScreenerRow(
                symbol=symbol,
                price=float(df.close.iloc[-1]),
                change_1h=float(df.close.pct_change(1).iloc[-1] * 100),
                change_24h=float(df.close.pct_change(24).iloc[-1] * 100),
                change_7d=float(df.close.pct_change(168).iloc[-1] * 100),

                # Волатильність
                rv_1h=float(realized_volatility(df.close, 1, annualize=False).iloc[-1]),
                rv_24h=float(realized_volatility(df.close, 24, annualize=False).iloc[-1]),
                rv_7d=float(realized_volatility(df.close, 168, annualize=False).iloc[-1]),
                atr_percent=float(atr_percent(df.high, df.low, df.close).iloc[-1]),
                bb_width=float(bollinger_bandwidth(df.close).iloc[-1]),

                # Обсяг
                volume_24h=float(df.volume.iloc[-24:].sum()),
                volume_ratio=float(df.volume.iloc[-1] / df.volume.iloc[-24:].mean()),
            )

            # Позначка аномальної волатильності
            rv_mean = realized_volatility(df.close, 24, annualize=False).mean()
            rv_current = row.rv_1h
            row.volatility_spike = rv_current > rv_mean * 2.5  # у 2.5 рази вище норми

            results.append(row)

        return sorted(results, key=lambda x: x.rv_24h, reverse=True)

Фільтри та сортування

Скринер корисний тільки з гнучкою фільтрацією:

interface ScreenerFilters {
  minVolume24h: number;       // мінімум $1M
  minPriceChange: number;     // |change_24h| > N%
  volatilitySpike: boolean;   // тільки аномальні
  bbWidthMax: number;         // Bollinger compression (пошук пробоїв)
  bbWidthMin: number;
  sector: string[];           // 'defi', 'layer1', 'meme'
  exchange: 'binance' | 'bybit' | 'okx' | 'all';
}

// Сортування по: rv_24h, change_24h, volume_ratio, atr_percent

Сповіщення

class VolatilityAlertsEngine:
    async def check_alerts(self, symbol: str, current_data: ScreenerRow):
        user_alerts = await self.db.get_active_alerts(symbol)

        for alert in user_alerts:
            triggered = False

            if alert.type == 'rv_threshold':
                triggered = current_data.rv_24h > alert.threshold

            elif alert.type == 'volume_spike':
                triggered = current_data.volume_ratio > alert.multiplier

            elif alert.type == 'bb_squeeze':
                triggered = current_data.bb_width < alert.threshold

            elif alert.type == 'price_change':
                triggered = abs(current_data.change_1h) > alert.threshold

            if triggered and not alert.is_triggered:
                await self.send_alert(alert, current_data)
                await self.db.mark_alert_triggered(alert.id)

Сповіщення доставляються через Telegram bot, email, push-уведомлення та webhook (для інтеграції з іншими системами). Важливий нюанс: сповіщення має повторно активуватися після скиду, інакше користувач отримає уведомлення один раз і не побачить повторних спайків.

Frontend таблиця

Real-time таблиця з WebSocket оновленнями та кольоровим кодуванням:

const VolatilityTable: React.FC = () => {
  const [data, setData] = useState<ScreenerRow[]>([]);

  useEffect(() => {
    const ws = new WebSocket('wss://api.yourplatform.com/screener/volatility');
    ws.onmessage = (e) => {
      const update = JSON.parse(e.data);
      setData(prev => prev.map(row =>
        row.symbol === update.symbol ? { ...row, ...update } : row
      ));
    };
    return () => ws.close();
  }, []);

  return (
    <table>
      <tbody>
        {data.map(row => (
          <tr key={row.symbol} className={row.volatilitySpike ? 'bg-yellow-50' : ''}>
            <td>{row.symbol}</td>
            <td className={row.change24h > 0 ? 'text-green-600' : 'text-red-600'}>
              {row.change24h.toFixed(2)}%
            </td>
            <td>{row.rv24h.toFixed(1)}%</td>
            <td>{row.atrPercent.toFixed(2)}%</td>
            <td className={row.bbWidth < 3 ? 'font-bold text-orange-500' : ''}>
              {row.bbWidth.toFixed(1)}%
            </td>
          </tr>
        ))}
      </tbody>
    </table>
  );
};

Хорошо реалізований скринер волатильності — це щоденний інструмент професійних трейдерів. Саме такі інструменти формують sticky аудиторію на платформі.