Crypto Volatility Screener Development
A volatility screener is a tool for finding assets with anomalous price movement or changing volatility characteristics. Traders use it to find trading opportunities: breakouts, momentum strategies, mean-reversion strategies. Technically, it's a data collection pipeline, metric calculation, and real-time web interface.
Volatility Metrics
Realized Volatility
The most common metric is historical volatility based on 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 # for hourly data
) -> pd.Series:
"""
Calculates realized volatility as standard deviation of log-returns.
window: number of periods (candles)
"""
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 # in percent
ATR (Average True Range)
ATR is more robust to gaps and intraday range:
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 normalized to price — comparable across different assets"""
return atr(high, low, close, period) / close * 100
Bollinger Band Width
Bollinger Bands squeeze is a signal of upcoming breakout:
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
# Width normalized to midline
return (upper - lower) / rolling_mean * 100
Screener Architecture
Data pipeline
Binance/OKX WebSocket ──► Candle Aggregator ──► Redis (latest N candles)
│
VolatilityCalculator (every 60s)
│
PostgreSQL / TimescaleDB
│
REST API + WebSocket
│
Frontend (React table)
Market-wide Calculation
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 days
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),
# Volatility
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
volume_24h=float(df.volume.iloc[-24:].sum()),
volume_ratio=float(df.volume.iloc[-1] / df.volume.iloc[-24:].mean()),
)
# Anomalous volatility flag
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.5x above normal
results.append(row)
return sorted(results, key=lambda x: x.rv_24h, reverse=True)
Filters and Sorting
A screener is only useful with flexible filtering:
interface ScreenerFilters {
minVolume24h: number; // $1M minimum
minPriceChange: number; // |change_24h| > N%
volatilitySpike: boolean; // only anomalous
bbWidthMax: number; // Bollinger compression (breakout search)
bbWidthMin: number;
sector: string[]; // 'defi', 'layer1', 'meme'
exchange: 'binance' | 'bybit' | 'okx' | 'all';
}
// Sorting by: rv_24h, change_24h, volume_ratio, atr_percent
Alerts
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)
Alerts are delivered via Telegram bot, email, push notifications, and webhooks (for integration with other systems). Important nuance: the alert should re-activate after reset, otherwise the user will receive the notification once and not see subsequent spikes.
Frontend Table
Real-time table with WebSocket updates and color coding:
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>
);
};
A well-implemented volatility screener is a daily tool for professional traders. These are exactly the tools that create sticky audiences on a platform.