Build a Working Unusual Options Activity Scanner in Python (Under 150 Lines)

If you want to build an unusual options activity scanner, you have two choices.

You can buy a packaged alert product from a vendor that picks the strategy for you and charges per ticker. Or you can call a transparent flow API yourself and write the scanner you actually want.

This article walks through the second route end to end: every line of Python, every parameter, every alert format. By the end you'll have a single-file scanner you can run locally, schedule with cron, or wire into Slack.

The underlying feed used here is the FlashAlpha Flow Signals API, which scores every block-sized print on a 0-100 scale with six documented components (premium, size-vs-OI, aggressor, sweep, opening bias, tenor). The full methodology is a separate article; the short version is that every score in the response carries a score_breakdown object you can audit. We'll filter, rank, render, and alert on top of that.

Prerequisites

pip install requests
export FLASHALPHA_API_KEY=your_key_here

You need Python 3.9+ and an Alpha-tier API key. The signals endpoints are gated to Alpha. requests is the only dependency for the scanner itself; if you want fancier terminal output, rich is a drop-in addition but the article doesn't use it.

Step 1: The signals call

One function, one HTTP call, one parsed JSON response.

import os
import requests

API_BASE = "https://lab.flashalpha.com"
API_KEY = os.environ["FLASHALPHA_API_KEY"]

def fetch_signals(
    symbol: str,
    window_minutes: int = 240,
    min_score: int = 0,
    intent: str | None = None,
    structure: str | None = None,
    limit: int = 50,
    expiry: str | None = None,
):
    """One call to /v1/flow/signals/{symbol}."""
    params = {
        "windowMinutes": window_minutes,
        "minScore": min_score,
        "limit": limit,
    }
    if intent: params["intent"] = intent
    if structure: params["structure"] = structure
    if expiry: params["expiry"] = expiry

    resp = requests.get(
        f"{API_BASE}/v1/flow/signals/{symbol}",
        headers={"X-Api-Key": API_KEY},
        params=params,
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

The response is a top-level metadata wrapper plus a signals array. Each signal carries the trade fields, classification, score, breakdown, and the chain context.

Step 2: A pretty terminal renderer

The point of a transparent feed is that you can read the per-component score, not just the composite. Render the breakdown alongside the signal.

from datetime import datetime

def render_signal(s: dict) -> str:
    ts = datetime.fromisoformat(s["ts"].replace("Z", "+00:00"))
    bd = s["score_breakdown"]
    bd_str = " ".join(f"{k[:3]}={v:>2}" for k, v in bd.items())
    tags = ",".join(s["tags"]) if s["tags"] else "-"
    return (
        f"{ts:%H:%M:%S}  "
        f"{s['expiry']} {s['strike']:>6}{s['right']}  "
        f"{s['side']:>4}  "
        f"size={s['size']:>5}  "
        f"${s['premium']:>10,.0f}  "
        f"score={s['score']:>3} ({s['conviction']:>7})  "
        f"{s['intent']:>7}/{s['open_close_bias']:>13}  "
        f"[{bd_str}]  "
        f"tags={tags}"
    )

def scan_and_print(symbol: str, **kwargs):
    data = fetch_signals(symbol, **kwargs)
    print(f"\n=== {symbol} - {data['count']} signals ===")
    print(
        f"chain: call_wall={data['chain']['call_wall']} "
        f"put_wall={data['chain']['put_wall']} "
        f"flip={data['chain']['gamma_flip']}"
    )
    for s in data["signals"]:
        print(render_signal(s))

if __name__ == "__main__":
    scan_and_print("SPY", min_score=70, structure="sweep", window_minutes=60)

One line per qualifying signal with every score component, the intent, the open/close bias, the chain context, and the tags. No interpretation layer, no opaque rank.

Step 3: A watchlist scanner

The summary endpoint returns a directional roll-up in one call per ticker. Concurrent fetches let you sweep a watchlist in under a second.

from concurrent.futures import ThreadPoolExecutor

WATCHLIST = ["SPY", "QQQ", "IWM", "NVDA", "TSLA", "AAPL", "META", "AMZN"]

def fetch_summary(symbol: str, window_minutes: int = 60):
    resp = requests.get(
        f"{API_BASE}/v1/flow/signals/{symbol}/summary",
        headers={"X-Api-Key": API_KEY},
        params={"windowMinutes": window_minutes},
        timeout=10,
    )
    resp.raise_for_status()
    return symbol, resp.json()

def watchlist_card(watchlist=WATCHLIST, window_minutes: int = 60):
    print(f"\n=== Watchlist Direction Card (window={window_minutes}m) ===")
    print(f"{'SYM':6} {'DIR':8} {'NET PREMIUM':>16} "
          f"{'OPEN':>14} {'CLOSE':>14} {'SIGNALS':>8}")

    with ThreadPoolExecutor(max_workers=8) as pool:
        for sym, d in pool.map(
            lambda s: fetch_summary(s, window_minutes), watchlist
        ):
            net = d["net_directional_premium"]
            direction = "BULLISH" if net > 0 else "BEARISH" if net < 0 else "NEUTRAL"
            print(
                f"{sym:6} {direction:8} "
                f"${net:>14,.0f}  "
                f"${d['opening_premium']:>12,.0f}  "
                f"${d['closing_premium']:>12,.0f}  "
                f"{d['signal_count']:>8}"
            )

Sample output on a busy session:

=== Watchlist Direction Card (window=60m) ===
SYM    DIR           NET PREMIUM           OPEN          CLOSE  SIGNALS
SPY    BULLISH    $   4,820,000   $  6,420,000   $  1,600,000        42
QQQ    BULLISH    $   1,140,000   $  2,810,000   $  1,670,000        28
NVDA   BEARISH    $  -2,250,000   $    410,000   $  2,660,000        17
TSLA   BULLISH    $   3,100,000   $  3,420,000   $    320,000        12
AAPL   NEUTRAL    $     -45,000   $     80,000   $    125,000         3
AMZN   NEUTRAL    $          0    $          0   $          0         0

SPY is bullish with mostly fresh positioning (open > close). NVDA looks bearish at the net but closing premium dominates opening, so it's mostly position unwinding, not fresh bearish conviction. The opening/closing split is doing real work here. Most UOA feeds publish only the net.

Step 4: Custom filters - whale sweeps only

Compose the filter parameters for the specific pattern you care about. Whale (premium ≥ $1M) sweeps with high conviction:

def whale_sweeps(symbol: str, window_minutes: int = 60):
    data = fetch_signals(
        symbol,
        window_minutes=window_minutes,
        structure="sweep",
        min_score=70,
        limit=50,
    )
    return [s for s in data["signals"] if "whale" in s["tags"]]

for sym in WATCHLIST:
    for s in whale_sweeps(sym):
        print(f"{sym}  {render_signal(s)}")

The whale tag fires automatically at $1,000,000+ premium.

Step 5: Golden signals only

The golden tag is a dual gate: top decile of the current result set and at or above 70 score. A quiet session doesn't promote weak signals.

def golden_signals(symbol: str, window_minutes: int = 240):
    data = fetch_signals(symbol, window_minutes=window_minutes)
    return [s for s in data["signals"] if "golden" in s["tags"]]

Step 6: Opening-only bullish scan

Most directional bets worth surfacing are opening flows, not closing flows.

def opening_bullish(symbol: str, window_minutes: int = 240, min_score: int = 60):
    data = fetch_signals(
        symbol,
        window_minutes=window_minutes,
        intent="bullish",
        min_score=min_score,
    )
    return [
        s for s in data["signals"]
        if s["open_close_bias"] == "opening_bias"
    ]

intent=bullish already drops every closing trade (closing flows collapse to Neutral). The client-side check is the explicit double-gate.

Step 7: Slack alerts

Wire any filter into a Slack incoming webhook. Send the score breakdown in the alert body so the human reading it can see why it fired without leaving Slack.

SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")

def slack_block(symbol: str, s: dict) -> dict:
    bd = s["score_breakdown"]
    color = (
        "#22c55e" if s["intent"] == "bullish"
        else "#ef4444" if s["intent"] == "bearish"
        else "#a1a1aa"
    )
    return {
        "attachments": [{
            "color": color,
            "title": (
                f"{symbol}  {s['expiry']} {s['strike']}{s['right']}  "
                f"{s['side'].upper()}  score={s['score']} ({s['conviction']})"
            ),
            "text": (
                f"size: {s['size']:,}  "
                f"premium: ${s['premium']:,.0f}  "
                f"intent: {s['intent']}\n"
                f"bias: {s['open_close_bias']}  "
                f"structure: {s['structure']}  "
                f"tags: {', '.join(s['tags']) or '-'}\n"
                f"breakdown: prem={bd['premium']} sz={bd['size_vs_oi']} "
                f"aggr={bd['aggressor']} swp={bd['sweep']} "
                f"open={bd['opening_bias']} ten={bd['tenor']}"
            ),
        }]
    }

def slack_post(symbol: str, s: dict):
    if not SLACK_WEBHOOK_URL: return
    requests.post(SLACK_WEBHOOK_URL, json=slack_block(symbol, s), timeout=5)

Step 8: De-duplication on a schedule

Polling every minute, the same signal appears in consecutive responses until it ages out of the window. Hash the natural key and remember.

import hashlib

_alerted: set[str] = set()

def signal_key(symbol: str, s: dict) -> str:
    raw = f"{symbol}|{s['ts']}|{s['expiry']}|{s['strike']}|{s['right']}|{s['size']}"
    return hashlib.sha1(raw.encode()).hexdigest()[:16]

def alert_once(symbol: str, s: dict):
    k = signal_key(symbol, s)
    if k in _alerted: return
    _alerted.add(k)
    slack_post(symbol, s)

Coalesced sweeps come through with their group's last-print timestamp and combined size, so the key is stable for the duration of the window. For long-running daemons, periodically prune entries older than the window.

Step 9: The single-file daemon

import time
from datetime import datetime, timezone

def loop_forever(
    watchlist=WATCHLIST,
    poll_seconds: int = 60,
    window_minutes: int = 30,
    min_score: int = 70,
):
    print(f"UOA daemon starting on {watchlist}, polling every {poll_seconds}s")
    while True:
        try:
            for sym in watchlist:
                for s in opening_bullish(sym, window_minutes, min_score):
                    if "golden" in s["tags"]:
                        alert_once(sym, s)
        except requests.HTTPError as e:
            now = datetime.now(timezone.utc).isoformat(timespec="seconds")
            print(f"{now}  HTTP {e.response.status_code} - retrying next tick")
        except Exception as e:
            now = datetime.now(timezone.utc).isoformat(timespec="seconds")
            print(f"{now}  {type(e).__name__}: {e} - retrying next tick")
        time.sleep(poll_seconds)

if __name__ == "__main__":
    loop_forever()

That's it. Under 150 lines, one dependency, alerts to Slack with full breakdown, deterministic de-dup, swappable filters.

What you avoid by building it yourself

Packaged UOA product:

• Black-box score with no per-component breakdown
• Alert thresholds the vendor picks, not you
• Per-ticker pricing that gets expensive fast
• Cannot inspect why an alert fired (or did not)
• Wire format is whatever the vendor provides

API + 150 lines of Python:

• Every component of every score in the response
• You pick filters, thresholds, watchlist
• One key, unlimited symbols, no per-ticker fees
• Every alert carries its audit trail
• Slack, Discord, Telegram, terminal, whatever you want

Five natural extensions

1. 0DTE-only filter. Pass expiry=<today's ET date>. Combine with structure=sweep for intraday momentum.
2. Cross-reference with chain context. Compare each signal's strike against chain.call_wall, chain.put_wall, chain.gamma_flip. A sweep through a wall is structurally different from one in no-man's-land.
3. Score-weighted leaderboard. Aggregate by ticker, sum premium weighted by score. Daily top-5 at 4:05 PM ET.
4. Cross with live GEX direction. Alert only when intent agrees with the live GEX regime shift.
5. Persist to a database. Drop alerts into Postgres or DuckDB keyed by signal_key. Build your own residual analysis: how often did "golden bullish opening sweep" precede a meaningful move? The breakdown makes per-component backtests trivial.

---

Full tutorial with worked examples and FAQ on the canonical: Build an Unusual Options Activity Scanner with Python. The methodology pillar that explains every field in the response: Flow Signals API methodology.