Building an Insider Trading Scanner

This guide walks you through building an insider trading scanner that detects significant insider purchases and sales. You’ll learn how to filter for meaningful signals and create alerts for potential trading opportunities.

What You’ll Build

By the end of this guide, you’ll have a scanner that:

Monitors insider transactions across multiple tickers
Filters for significant purchases (high conviction signals)
Detects cluster buying (multiple insiders buying)
Tracks C-suite executive activity
Generates alerts for notable transactions

Prerequisites

Python 3.7+
FinBrain API key
finbrain-python SDK installed

pip install finbrain-python

Step 1: Basic Insider Data Retrieval

First, let’s fetch insider transactions for a single ticker using as_dataframe=True:

from finbrain import FinBrainClient

fb = FinBrainClient(api_key="YOUR_API_KEY")

# Get insider transactions as a DataFrame
df = fb.insider_transactions.ticker("S&P 500", "AAPL", as_dataframe=True)

print(df.head())
#                 insiderTradings              relationship  transaction  shares    cost     USDValue
# date
# 2024-01-15         Tim Cook     Chief Executive Officer        Sale   50000  185.50      9275000
# 2024-01-10       Luca Maestri    Chief Financial Officer        Sale   25000  188.25      4706250

# Filter for purchases only
purchases = df[df["transaction"].str.contains("Buy", na=False)]
print(f"\nRecent purchases: {len(purchases)}")
print(purchases[["insiderTradings", "shares", "USDValue"]])

Step 2: Filter for Significant Purchases

Not all insider transactions are meaningful. Let’s filter for significant open-market purchases using DataFrame operations:

from finbrain import FinBrainClient
from datetime import datetime, timedelta

fb = FinBrainClient(api_key="YOUR_API_KEY")

def get_significant_purchases(market, ticker, min_value=100000, days=30):
    """Filter for significant purchases in the last N days"""
    df = fb.insider_transactions.ticker(market, ticker, as_dataframe=True)

    if df.empty:
        return df

    cutoff_date = datetime.now() - timedelta(days=days)

    # Filter for purchases with minimum value in recent days
    significant = df[
        (df["transaction"].str.contains("Buy", na=False)) &
        (df["USDValue"] >= min_value) &
        (df.index >= cutoff_date)
    ]

    return significant.sort_values("USDValue", ascending=False)

# Example usage
purchases = get_significant_purchases("S&P 500", "AAPL", min_value=500000)

print(f"Found {len(purchases)} significant purchases")
print(purchases[["insiderTradings", "relationship", "USDValue"]])

Step 3: C-Suite Tracking

Focus on the most important insiders - CEOs, CFOs, and board members using DataFrame filtering:

from finbrain import FinBrainClient

fb = FinBrainClient(api_key="YOUR_API_KEY")

C_SUITE_KEYWORDS = [
    "chief executive", "ceo",
    "chief financial", "cfo",
    "chief operating", "coo",
    "president",
    "chairman",
    "director"
]

def get_c_suite_purchases(market, ticker):
    """Filter for C-suite purchases only"""
    df = fb.insider_transactions.ticker(market, ticker, as_dataframe=True)

    if df.empty:
        return df

    # Filter for purchases
    purchases = df[df["transaction"].str.contains("Buy", na=False)]

    # Filter for C-suite relationships
    pattern = "|".join(C_SUITE_KEYWORDS)
    c_suite = purchases[purchases["relationship"].str.lower().str.contains(pattern, na=False)]

    return c_suite

# Example
c_suite = get_c_suite_purchases("S&P 500", "AAPL")

print("C-Suite Purchases:")
print(c_suite[["insiderTradings", "relationship", "shares", "USDValue"]])

Step 4: Cluster Buying Detection

Multiple insiders buying at the same time is a strong signal. Use DataFrame groupby for analysis:

from finbrain import FinBrainClient
from datetime import datetime, timedelta
import pandas as pd

fb = FinBrainClient(api_key="YOUR_API_KEY")

def detect_cluster_buying(market, ticker, window_days=14, min_buyers=2):
    """Detect cluster buying - multiple insiders buying within a time window"""
    df = fb.insider_transactions.ticker(market, ticker, as_dataframe=True)

    if df.empty:
        return []

    # Filter for purchases
    purchases = df[df["transaction"].str.contains("Buy", na=False)].copy()

    if purchases.empty:
        return []

    # Group purchases by rolling window
    purchases = purchases.sort_index()
    clusters = []

    for start_date in purchases.index.unique():
        window_end = start_date + timedelta(days=window_days)
        window = purchases[(purchases.index >= start_date) & (purchases.index <= window_end)]

        unique_buyers = window["insiderTradings"].unique()
        if len(unique_buyers) >= min_buyers:
            clusters.append({
                "start_date": start_date.strftime("%Y-%m-%d"),
                "buyers": list(unique_buyers),
                "num_buyers": len(unique_buyers),
                "total_value": window["USDValue"].sum()
            })

    return clusters

# Example
clusters = detect_cluster_buying("S&P 500", "AAPL")

for c in clusters:
    print(f"Cluster starting {c['start_date']}:")
    print(f"  {c['num_buyers']} buyers, ${c['total_value']:,} total")
    print(f"  Buyers: {', '.join(c['buyers'])}")

Step 5: Multi-Ticker Scanner

Now let’s scan multiple tickers using DataFrames:

from finbrain import FinBrainClient
from datetime import datetime, timedelta
import pandas as pd

fb = FinBrainClient(api_key="YOUR_API_KEY")

def scan_insider_activity(tickers, market="S&P 500", min_value=100000, days=30):
    """Scan multiple tickers for significant insider activity"""
    results = []
    cutoff_date = datetime.now() - timedelta(days=days)

    for ticker in tickers:
        try:
            df = fb.insider_transactions.ticker(market, ticker, as_dataframe=True)

            if df.empty:
                continue

            # Filter for recent significant purchases
            purchases = df[
                (df["transaction"].str.contains("Buy", na=False)) &
                (df["USDValue"] >= min_value) &
                (df.index >= cutoff_date)
            ]

            if not purchases.empty:
                results.append({
                    "ticker": ticker,
                    "purchases": purchases,
                    "total_value": purchases["USDValue"].sum(),
                    "num_buyers": purchases["insiderTradings"].nunique()
                })

        except Exception as e:
            print(f"Error scanning {ticker}: {e}")
            continue

    # Sort by total value
    return sorted(results, key=lambda x: x["total_value"], reverse=True)

# Scan tech stocks
tech_tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "META", "TSLA", "NFLX"]
results = scan_insider_activity(tech_tickers)

print("\n=== Insider Buying Scanner Results ===\n")
for r in results:
    print(f"{r['ticker']}: ${r['total_value']:,} ({r['num_buyers']} buyers)")
    top_purchases = r["purchases"].nlargest(3, "USDValue")
    for _, p in top_purchases.iterrows():
        print(f"  - {p['insiderTradings']}: ${p['USDValue']:,}")

Step 6: Complete Scanner with Alerts

Here’s a complete scanner with all features using as_dataframe=True:

from finbrain import FinBrainClient
from datetime import datetime, timedelta
import pandas as pd

class InsiderScanner:
    def __init__(self, api_key):
        self.fb = FinBrainClient(api_key=api_key)
        self.c_suite_keywords = [
            "chief executive", "ceo", "chief financial", "cfo",
            "chief operating", "coo", "president", "chairman", "director"
        ]

    def is_c_suite(self, relationship):
        relationship_lower = str(relationship).lower()
        return any(kw in relationship_lower for kw in self.c_suite_keywords)

    def scan_ticker(self, market, ticker, days=30, min_value=100000):
        """Scan a single ticker for insider activity"""
        try:
            df = self.fb.insider_transactions.ticker(market, ticker, as_dataframe=True)
        except Exception:
            return None

        if df.empty:
            return None

        cutoff = datetime.now() - timedelta(days=days)

        # Filter recent transactions
        recent = df[df.index >= cutoff]

        if recent.empty:
            return None

        # Analyze purchases using DataFrame operations
        purchases = recent[recent["transaction"].str.contains("Buy", na=False)]

        if purchases.empty:
            return None

        significant = purchases[purchases["USDValue"] >= min_value]
        c_suite = purchases[purchases["relationship"].apply(self.is_c_suite)]

        # Cluster detection
        unique_buyers = purchases["insiderTradings"].nunique()
        is_cluster = unique_buyers >= 2

        return {
            "ticker": ticker,
            "total_purchases": len(purchases),
            "significant_purchases": len(significant),
            "c_suite_purchases": len(c_suite),
            "unique_buyers": unique_buyers,
            "is_cluster": is_cluster,
            "total_value": purchases["USDValue"].sum(),
            "top_transactions": significant.nlargest(5, "USDValue")
        }

    def scan_universe(self, tickers, market="S&P 500"):
        """Scan multiple tickers"""
        results = []

        for i, ticker in enumerate(tickers):
            if i % 10 == 0:
                print(f"Scanning... {i}/{len(tickers)}")

            result = self.scan_ticker(market, ticker)
            if result and result["total_purchases"] > 0:
                results.append(result)

        return sorted(results, key=lambda x: x["total_value"], reverse=True)

    def generate_alerts(self, results):
        """Generate alerts for notable activity"""
        alerts = []

        for r in results:
            # High value alert
            if r["total_value"] >= 1000000:
                alerts.append({
                    "type": "HIGH_VALUE",
                    "ticker": r["ticker"],
                    "message": f"${r['total_value']:,} in insider purchases"
                })

            # Cluster buying alert
            if r["is_cluster"]:
                alerts.append({
                    "type": "CLUSTER_BUYING",
                    "ticker": r["ticker"],
                    "message": f"{r['unique_buyers']} different insiders buying"
                })

            # C-suite alert
            if r["c_suite_purchases"] > 0:
                alerts.append({
                    "type": "C_SUITE",
                    "ticker": r["ticker"],
                    "message": f"{r['c_suite_purchases']} C-suite purchases"
                })

        return alerts


# Run the scanner
scanner = InsiderScanner(api_key="YOUR_API_KEY")

# Define your universe
universe = [
    "AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "META", "TSLA",
    "JPM", "BAC", "WFC", "GS", "MS",
    "JNJ", "PFE", "UNH", "MRK", "ABBV"
]

print("Starting Insider Trading Scanner...")
print("=" * 50)

results = scanner.scan_universe(universe)
alerts = scanner.generate_alerts(results)

print("\n=== ALERTS ===")
for alert in alerts:
    print(f"[{alert['type']}] {alert['ticker']}: {alert['message']}")

print("\n=== TOP INSIDER BUYING ===")
for r in results[:10]:
    print(f"\n{r['ticker']}: ${r['total_value']:,}")
    print(f"  Buyers: {r['unique_buyers']}, C-Suite: {r['c_suite_purchases']}")
    if r["is_cluster"]:
        print("  ** CLUSTER BUYING DETECTED **")

Next Steps

Add email or Slack notifications for alerts
Store historical data for backtesting
Combine with AI predictions for higher conviction signals
Create a web dashboard for visualization