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

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

Скринер волатильности — это инструмент для поиска активов с аномальным движением цены или изменившимися характеристиками волатильности. Трейдеры используют его для поиска торговых возможностей: пробои, моментум-стратегии, стратегии 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  # для hourly данных
) -> 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 minimum
  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 (для интеграции с другими системами). Важный нюанс: alert должен снова активироваться после reset, иначе пользователь получит уведомление один раз и больше не увидит повторных спайков.

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 аудиторию на платформе.