priceflare 0.2.0

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

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

Parser	Exchange	WebSocket URL
parsers.binance	Binance Spot/Futures	wss://stream.binance.com:9443/ws/<symbol>@trade
parsers.kraken	Kraken	wss://ws.kraken.com/v2
parsers.coinbase	Coinbase Advanced Trade	wss://advanced-trade-ws.coinbase.com

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)

Parameter	Type	Default	Description
ws_url	str	required	WebSocket URL. Empty string allowed for offline/feed-only use.
price_parser	callable	required	(str) -> float | None
crash_threshold	float	3.0	% drop to trigger CRASH
pump_threshold	float	3.0	% rise to trigger PUMP
window_seconds	int	300	Rolling window in seconds
cooldown_seconds	int	600	Min seconds between alerts
max_reconnects	int	50	Max reconnection attempts (must be > 0)
subscribe_message	str | None	None	JSON string sent after connection (Kraken, Coinbase)
on_alert	callable	None	Called on CRASH or PUMP with alert dict
on_crash	callable	None	Called on CRASH with current price
on_pump	callable	None	Called on PUMP with current price

Methods

Method	Description
start()	Start in a background daemon thread. Returns the thread. Raises SentinelError if ws_url is empty.
stop()	Signal the sentinel to stop.
feed(price, timestamp=None)	Inject a price directly. timestamp: Unix float for historical data. Returns alert dict or None.
status()	Returns current state dict.

---

Tests

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

---

License

MIT — free to use, modify, and redistribute.