PriceFlare

Exchange-agnostic real-time price crash and pump detector over WebSocket.

Any trading bot needs to react immediately to sudden price movements. PriceFlare handles the detection — you supply the WebSocket URL, a price parser, and your callbacks.

WebSocket stream → price parser → rolling window → CRASH / PUMP → your callback

Install

pip install priceflare

Quickstart

from priceflare import Sentinel
from priceflare import parsers

def on_alert(alert):
    print(f"[{alert['type']}] {alert['change_pct']:+.2f}% in {alert['window_sec']//60}min")
    print(f"  {alert['ref_price']} → {alert['cur_price']}")

def on_crash(price):
    print(f"Crash at {price} — checking stop-loss...")

def on_pump(price):
    print(f"Pump at {price} — checking opportunity...")

sentinel = Sentinel(
    ws_url           = "wss://stream.binance.com:9443/ws/btcusdt@trade",
    price_parser     = parsers.binance,
    crash_threshold  = 3.0,   # % drop to trigger CRASH
    pump_threshold   = 3.0,   # % rise to trigger PUMP
    window_seconds   = 300,   # rolling window (5 min)
    cooldown_seconds = 600,   # min delay between alerts (10 min)
    on_alert         = on_alert,
    on_crash         = on_crash,   # optional — CRASH only
    on_pump          = on_pump,    # optional — PUMP only
)

thread = sentinel.start()

Built-in parsers

Custom parser

A parser is any callable (raw: str) -> float | None. Return None to silently skip non-price messages.

import json

def my_parser(raw: str) -> float | None:
    data = json.loads(raw)
    if data.get("type") != "trade":
        return None
    return float(data["price"])

sentinel = Sentinel(
    ws_url       = "wss://my-exchange.com/ws",
    price_parser = my_parser,
    on_alert     = on_alert,
)

Exchanges requiring a subscription message (Kraken, Coinbase)

Some exchanges expect a JSON subscription message after the WebSocket connection is established. Use the subscribe_message parameter:

import json

sentinel = Sentinel(
    ws_url            = "wss://ws.kraken.com/v2",
    price_parser      = parsers.kraken,
    subscribe_message = json.dumps({
        "method": "subscribe",
        "params": {"channel": "ticker", "symbol": ["BTC/USD"]},
    }),
    on_alert = on_alert,
)

# Coinbase Advanced Trade
sentinel = Sentinel(
    ws_url            = "wss://advanced-trade-ws.coinbase.com",
    price_parser      = parsers.coinbase,
    subscribe_message = json.dumps({
        "type": "subscribe",
        "product_ids": ["BTC-USD"],
        "channel": "ticker",
    }),
    on_alert = on_alert,
)

Alert format

{
    "type":       "CRASH",        # or "PUMP"
    "change_pct": -4.21,          # signed percentage
    "ref_price":  67000.0,        # price at start of window
    "cur_price":  64178.0,        # current price
    "window_sec": 300,
    "timestamp":  "2026-03-31T08:00:00+00:00",
}

Backtesting / offline use

Feed prices directly without a WebSocket connection using feed(). Leave ws_url empty — start() is never called.

sentinel = Sentinel(
    ws_url          = "",         # unused in offline mode
    price_parser    = parsers.binance,
    crash_threshold = 3.0,
    cooldown_seconds = 0,
    on_alert        = on_alert,
)

for price, ts in zip(historical_prices, historical_timestamps):
    alert = sentinel.feed(price, timestamp=ts)   # timestamp = Unix float
    if alert:
        print(alert)

API reference

Sentinel(ws_url, price_parser, **kwargs)

Methods

Tests

pip install -e ".[dev]"
pytest                   # 61 tests
pytest --cov=priceflare  # with coverage

License

MIT — free to use, modify, and redistribute.