Build an Unusual Options Activity Scanner with Python (FlashAlpha Flow Signals API)

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, all hitting the FlashAlpha Flow Signals API.

The underlying feed scores every block-sized print on a 0-100 scale with six documented components. You filter, rank, render, and alert on top of that. By the end of this article you will have a single-file Python scanner you can run locally, schedule with cron, or wire into a Slack channel.

What You Need

• Python 3.9 or newer.
• An Alpha-tier FlashAlpha API key. The signals endpoints are gated to Alpha; view pricing or open the playground to confirm access.
• requests for HTTP. Optionally rich for pretty terminal output.

pip install requests rich
export FLASHALPHA_API_KEY=your_key_here

Step 1: The Signals Call

One function, one HTTP call, one parsed JSON response. The endpoint is GET /v1/flow/signals/{symbol}; the query parameters narrow the result without re-running the pipeline server-side.

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 shape: 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 so a human triaging the feed can see why each rank fired.

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 (window={data['window_minutes']}m) ===")
    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)

Run it and you get 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 (GET /v1/flow/signals/{symbol}/summary) returns the 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}"
            )

Output, on a busy session, looks like this:

=== 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
IWM    NEUTRAL    $          0    $    180,000   $    180,000         5
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
META   BULLISH    $     820,000   $    910,000   $     90,000         6
AMZN   NEUTRAL    $          0    $          0   $          0         0

Reading: SPY is bullish with mostly fresh positioning (open > close). NVDA is bearish but the closing premium dominates the opening premium - it is mostly position unwinding, not fresh bearish positioning. AMZN has no qualifying signals at all. This is a dashboard, not a black box.

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 when the trade's premium is at or above $1,000,000. Combine with structure=sweep for the cross-venue urgency signal and min_score=70 for the medium-conviction floor.

Step 5: Golden Signals Only

The golden tag fires on signals in the top decile of the current result set and at or above 70 score - a dual gate that prevents quiet sessions from promoting weak signals. Use it as your primary triage filter when you want quality without volume.

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"]]

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

Step 6: Opening-Only Bullish Scan

Most directional bets you want to surface are opening flows, not closing flows. The signals endpoint does not expose a ?bias= filter today, but the open_close_bias field is in every record.

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"
    ]

The intent=bullish server-side filter already drops every closing trade (closing flows collapse to Neutral intent - see the Opening vs Closing Bias article). The client-side open_close_bias == "opening_bias" check is the explicit double-gate when you want to be sure.

Step 7: Slack Alerts

Wire any of the filters above into a Slack incoming webhook. Send a formatted block per signal, including the score breakdown so the human reading the alert 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)

# Wiring example
for sym in WATCHLIST:
    for s in golden_signals(sym, window_minutes=15):
        slack_post(sym, s)

Replace golden_signals with whichever filter matches your alert taste. The breakdown is in every alert by design - it is the audit trail.

Step 8: De-duplication on a Schedule

If you poll every minute, the same signal will appear in consecutive responses until it ages out of the window. Hash the signal's natural key and remember what you have already alerted on.

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)

The natural key combines symbol, timestamp, contract, side, and size. 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 _alerted entries older than your window.

Step 9: A Single-File Daemon

Putting it all together: a small daemon that polls every minute, alerts on golden bullish openings, and never duplicates.

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 is the whole scanner. Less than 150 lines, no dependencies beyond requests, alerts to Slack with full breakdown, deterministic de-dup, swappable filters. Wire opening_bullish + "golden" in s["tags"] for your taste; change the watchlist and the poll cadence at the call site.

What You Avoid by Building It Yourself

Extending the Scanner

Five natural extensions you can drop in:

1. 0DTE-only filter

   Pass expiry=<today's ET date> to filter the chain to today's expiry. Combine with structure=sweep for intraday 0DTE momentum signals.

2. Cross-reference with chain context

   For each signal, compare its strike to the response's 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 signals across the watchlist by ticker, summing premium and weighting by score. Publish the daily top-5 to a Discord channel at 4:05 PM ET.

4. Cross with live GEX direction

   Combine with a call to /v1/flow/summary/{symbol}. Alert only on signals where the directional intent agrees with the live GEX regime shift (flow_direction=amplifying + bullish intent, for example).

5. Persist to a database

   Drop each alerted signal into Postgres or DuckDB keyed by signal_key. Build your own residual analysis: how often did "golden bullish opening sweep" signals precede a meaningful move? The score breakdown makes per-component backtests trivial.

Frequently Asked Questions

Related Reading

• Flow Signals API: Scoring Unusual Options Activity - the full methodology pillar covering all six scoring components and the response schema
• Options Sweep Detection: 500ms Coalescing - what the structure field actually means
• Opening vs Closing Bias on Options Flow - what the open_close_bias field means and why closing trades are Neutral
• Live Dealer Flow Monitor - the chain-level companion surface for combining intent with regime
• Effective Open Interest Methodology - the OI calibration story behind the opening-bias confidence weight
• Options Flow API: Recent Trades, Blocks, Leaderboard - the raw-trade surface for audit and custom logic