True-streaming anomaly terminal: event-time MVs (RisingWave) + drift-adaptive detection, backtested

[puneethkotha]

Summary

Rearchitects StockStream from a Spark micro-batch ETL into a true-streaming,
backtested market anomaly terminal. The Spark foreachBatch + per-row
collect() consumer and the InfluxDB aggregation store are gone; RisingWave now
computes VWAP, moving averages, and rolling realized volatility as event-time
incremental materialized views and serves them over the Postgres wire, and a
drift-adaptive online detector (River) replaces the four static single-bar
thresholds. Everything runs here — the streaming path, the MV-correctness check,
and the backtest — and the numbers below are from real runs, pasted verbatim.

The nine-container stack (Kafka+Zookeeper, Spark master/worker/submit, InfluxDB,
Postgres, Grafana) collapses to four processes: Redpanda, RisingWave, the
detector, the UI/API.

Phase-by-phase

• Phase 0 — Honest baseline + recorded corpus. Deterministic generator
  (bench/generate_corpus.py), 11,700 minute bars / 6 symbols / 5 sessions, 17
  labeled events (earnings gaps, drawdowns, halts, shocks). Static baseline via
  the shipped AlertEngine, deterministic across two passes.
• Phase 1 — Redpanda + RisingWave; delete Spark/InfluxDB. New
  docker-compose.yaml; sql/00_sources.sql (watermarked raw_bars). Producer
  → Redpanda → RisingWave ingests all 11,700 bars; count(*) = 11700 exact.
• Phase 2 — Incremental feature MVs. sql/10_features.sql: returns_1m,
  features (per-minute VWAP/SMA/rvol/volume moments), vwap_5m_hop
  (EMIT ON WINDOW CLOSE). Verified equal to a numpy/pandas reference.
• Phase 3 — Drift-adaptive detector. detector/ (robust z/MAD +
  Half-Space Trees + ADWIN/Page-Hinkley); sql/20_anomalies.sql; live sidecar
  with exactly-once coverage across a crash/restart.
• Phase 4 — Backtest at matched recall. bench/backtest.py, range-based
  precision/recall; adaptive vs static on identical inputs.
• Phase 5 — Terminal UI + deploy + honest landing. Client-side replay on
  GitHub Pages (docs/demo/), FastAPI read API (api/), Caddy/VM deploy
  (deploy/), and corrected landing-page claims.

Evidence / metrics (real output)

Materialized-view correctness (bench/verify_mv.py, vs numpy/pandas,
np.isclose rtol 1e-6 / atol 1e-9):

features_pass: true
  sma_5m       max_abs_err 1.8e-12   rel 2.3e-15
  sma_20m      max_abs_err 1.8e-12   rel 1.6e-15
  vwap_5m      max_abs_err 3.2e-12   rel 4.5e-15
  rvol_5m      max_abs_err 7.7e-11   rel 1.1e-12
  vol_mean_20 / vol_std_20   exact (0.0)
hop VWAP (EMIT ON WINDOW CLOSE): max_rel_err 4.9e-16 over 11,741 windows
watermark: 42 of 11,820 windows correctly held back

Backtest, adaptive vs static (bench/backtest.py, range-based, 17 labeled
events / 6 symbols / 5 trading days):

                     recall  precision  false alerts/day    F1
static (5%/3%)         0.53       0.45          6.0         0.49
adaptive               0.76       0.88          0.6         0.82

At recall 0.76 (≥ the static baseline's 0.53) the adaptive detector fires 10x
fewer false alerts/day (0.6 vs 6.0) with near-double precision — it dominates
the static baseline on recall, precision, and false-alert rate. AUC-PR over the
primary robust-z sweep is 0.15 (pulled down by deliberately near-noise-floor
subtle events). The static rules also miss overnight gaps entirely (single-bar
open/close cannot see a cross-bar move).

Engine latency (bench/measure_latency.py, ingest → feature, 120 bars):
p50 17.7 ms, p95 27.1 ms (produce → raw_bars p50 8.1 ms). Scoped to engine
latency; the live yfinance producer's 3s poll of 15-min bars is reported
separately.

Exactly-once coverage (bench/restart_test.sh): deleted 4,680 tail rows to
simulate a crash; the detector resumed from its last (ts, stock) and refilled —
final 11,700 rows, 11,700 distinct keys (no duplicate), 0 gap.

Detector checks (bench/detector_checks.py): deterministic across two runs;
band-width vs rvol_5m Pearson r = 0.68 (> 0.6); all 4 regime-change events
(all 10 non-subtle events) trigger drift.

Needs your environment

• Live feed. The live path ships a paced corpus replay as the source
  (bench/producer_live.py). The real yfinance producer needs network access +
  open market hours; its bootstrap is now env-driven (KAFKA_BOOTSTRAP).
• Live VM. deploy/docker-compose.yml + deploy/Caddyfile need a domain and
  a small always-on VM (e.g. Hetzner CX22, ~$5/mo) for TLS + the live URL. The
  compose config is validated here; it is not stood up (no domain). The zero-infra
  GitHub Pages replay needs nothing.
• Pages. The replay auto-deploys via the existing pages.yml on merge to
  main; the demo lives at /demo/.

Run instructions

docker compose up -d redpanda risingwave
docker compose run --rm redpanda-init
docker compose build producer                         # image: ss_bench
for f in 00_sources 10_features 20_anomalies; do
  docker run --rm --network stockstream_ss -v "$PWD/sql:/sql" ss_bench \
    psql -h risingwave -p 4566 -U root -d dev -f /sql/$f.sql; done
docker run --rm --network stockstream_ss -v "$PWD/bench:/work" \
  -e KAFKA_BOOTSTRAP=redpanda:9092 ss_bench python /work/feed_kafka.py
# evidence
docker run --rm --network stockstream_ss -v "$PWD/bench:/work" -e RW_HOST=risingwave ss_bench python /work/verify_mv.py
docker run --rm -v "$PWD:/repo" -w /repo -e PYTHONPATH=/repo ss_bench python bench/backtest.py
NET=stockstream_ss bash bench/restart_test.sh
# replay demo (no backend)
python -m http.server 8000 --directory docs   # then open /demo/

Notes

• No Spark or InfluxDB remains in the running system; the migration is documented
  in CHANGELOG.md.
• The corpus is deterministic and synthetic (yfinance cannot serve intraday
  history for specific past events); it models realistic per-symbol volatility
  heterogeneity and labeled events. Every metric above is a real measurement over
  that corpus, reproducible with one command.