Skip to content
  • guides
  • congressional signals

Congressional Trade Signals

This guide shows you how to analyze trading data from both US House Representatives and Senators, and build systematic trading signals based on congressional activity.

FinBrain provides two separate endpoints for congressional trading data:

  • House Trades - US House Representatives (fb.house_trades)
  • Senate Trades - US Senators (fb.senate_trades)

Why Track Congressional Trades?

Section titled “Why Track Congressional Trades?”

Members of Congress have access to non-public information through their committee work, briefings, and legislative activities. While trading on material non-public information is illegal, their disclosed trades can still provide insights:

  • Committee knowledge - Members on finance committees may have early insights
  • Legislative impact - Understanding how laws affect specific companies
  • Historical accuracy - Some members have notable track records

Prerequisites

Section titled “Prerequisites”
Terminal window
pip install finbrain-python pandas

Step 1: Basic Congressional Data

Section titled “Step 1: Basic Congressional Data”

Let’s start by fetching congressional trading data using as_dataframe=True:

from finbrain import FinBrainClient


fb = FinBrainClient(api_key="YOUR_API_KEY")


# Get House trades as a DataFrame
house_df = fb.house_trades.ticker("S&P 500", "NVDA", as_dataframe=True)
print("=== House Trades ===")
print(house_df.head())
#                representative       type                   amount
# date
# 2024-01-08      Nancy Pelosi  Purchase  $1,000,001 - $5,000,000


# Get Senate trades as a DataFrame
senate_df = fb.senate_trades.ticker("S&P 500", "META", as_dataframe=True)
print("\n=== Senate Trades ===")
print(senate_df.head())
#                        senator       type              amount
# date
# 2025-11-13  Shelley Moore Capito  Purchase  $1,001 - $15,000


# Filter for purchases only
purchases = house_df[house_df["type"] == "Purchase"]
print(f"\nHouse Purchases: {len(purchases)}")

Step 2: Scan Multiple Tickers

Section titled “Step 2: Scan Multiple Tickers”

Scan your watchlist for congressional activity using DataFrames:

from finbrain import FinBrainClient


fb = FinBrainClient(api_key="YOUR_API_KEY")


def scan_congressional_activity(tickers, market="S&P 500"):
    """Scan multiple tickers for congressional trades"""
    results = []


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


            if not df.empty:
                purchases = df[df["type"] == "Purchase"]
                sales = df[df["type"] == "Sale"]


                results.append({
                    "ticker": ticker,
                    "total_trades": len(df),
                    "purchases": len(purchases),
                    "sales": len(sales),
                    "net_signal": "bullish" if len(purchases) > len(sales) else "bearish"
                })
        except Exception:
            continue


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


# Scan tech stocks
tickers = ["NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA"]
results = scan_congressional_activity(tickers)


print("\n=== Congressional Activity Scan ===")
for r in results:
    print(f"{r['ticker']}: {r['purchases']} buys, {r['sales']} sells ({r['net_signal']})")

Step 3: Track Specific Representatives

Section titled “Step 3: Track Specific Representatives”

Some representatives have notable trading track records. Use DataFrames for easy filtering:

from finbrain import FinBrainClient
import pandas as pd


fb = FinBrainClient(api_key="YOUR_API_KEY")


def track_representative(representative_name, tickers, market="S&P 500"):
    """Track all trades by a specific representative"""
    all_trades = []


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


            if not df.empty:
                # Filter for specific representative
                rep_trades = df[df["representative"].str.contains(representative_name, case=False, na=False)].copy()
                rep_trades["ticker"] = ticker
                all_trades.append(rep_trades)
        except Exception:
            continue


    if not all_trades:
        return pd.DataFrame()


    # Combine and sort by date
    combined = pd.concat(all_trades)
    return combined.sort_index(ascending=False)


# Track Nancy Pelosi's trades
tickers = ["NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "CRM", "RBLX", "DIS", "TSLA"]
pelosi_trades = track_representative("Pelosi", tickers)


print("\n=== Nancy Pelosi Recent Trades ===")
print(pelosi_trades[["ticker", "type", "amount"]].head(10))

Step 4: Large Trade Alerts

Section titled “Step 4: Large Trade Alerts”

Focus on significant trades using DataFrame filtering:

from finbrain import FinBrainClient
import pandas as pd


fb = FinBrainClient(api_key="YOUR_API_KEY")


LARGE_AMOUNTS = [
    "$500,001 - $1,000,000",
    "$1,000,001 - $5,000,000",
    "Over $5,000,000"
]


def get_large_trades(tickers, market="S&P 500"):
    """Find large congressional trades"""
    all_trades = []


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


            if not df.empty:
                # Filter for large amounts
                large = df[df["amount"].isin(LARGE_AMOUNTS)].copy()
                large["ticker"] = ticker
                all_trades.append(large)
        except Exception:
            continue


    if not all_trades:
        return pd.DataFrame()


    combined = pd.concat(all_trades)


    # Add estimated value column for sorting
    def estimate_value(amount):
        if "Over $5,000,000" in str(amount):
            return 7500000
        elif "$1,000,001 - $5,000,000" in str(amount):
            return 3000000
        elif "$500,001 - $1,000,000" in str(amount):
            return 750000
        return 0


    combined["estimated_value"] = combined["amount"].apply(estimate_value)
    return combined.sort_values("estimated_value", ascending=False)


# Find large trades
tickers = ["NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META"]
large = get_large_trades(tickers)


print("\n=== Large Congressional Trades ===")
print(large[["ticker", "representative", "type", "amount"]].head(10))

Step 5: Bipartisan Consensus

Section titled “Step 5: Bipartisan Consensus”

Find stocks both parties are buying using DataFrame aggregation:

from finbrain import FinBrainClient


fb = FinBrainClient(api_key="YOUR_API_KEY")


def find_bipartisan_buys(tickers, market="S&P 500"):
    """Find stocks where multiple representatives are buying"""
    results = []


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


            if df.empty:
                continue


            purchases = df[df["type"] == "Purchase"]
            unique_buyers = purchases["representative"].nunique()


            if unique_buyers >= 2:
                results.append({
                    "ticker": ticker,
                    "unique_buyers": unique_buyers,
                    "total_purchases": len(purchases)
                })
        except Exception:
            continue


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


# Find stocks with multiple buyers
tickers = ["NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META"]
multi_buyers = find_bipartisan_buys(tickers)


print("\n=== Multiple Representatives Buying ===")
for b in multi_buyers:
    print(f"{b['ticker']}: {b['unique_buyers']} unique buyers, {b['total_purchases']} purchases")

Step 6: Trade Analysis by Representative

Section titled “Step 6: Trade Analysis by Representative”

Analyze trading patterns by representative using DataFrame operations:

from finbrain import FinBrainClient
import pandas as pd


fb = FinBrainClient(api_key="YOUR_API_KEY")


def analyze_representative_activity(tickers, market="S&P 500"):
    """Analyze trading activity by representative"""
    all_trades = []


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


            if not df.empty:
                df = df.copy()
                df["ticker"] = ticker
                all_trades.append(df)
        except Exception:
            continue


    if not all_trades:
        return pd.DataFrame()


    combined = pd.concat(all_trades)


    # Group by representative and analyze
    analysis = combined.groupby("representative").agg(
        total_trades=("type", "count"),
        purchases=("type", lambda x: (x == "Purchase").sum()),
        sales=("type", lambda x: (x == "Sale").sum()),
        tickers=("ticker", lambda x: list(x.unique()))
    ).sort_values("total_trades", ascending=False)


    return analysis


# Analyze activity
tickers = ["NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META"]
activity = analyze_representative_activity(tickers)


print("\n=== Representative Activity Analysis ===")
print(activity[["total_trades", "purchases", "sales"]].head(10))

Step 7: Complete Signal Generator

Section titled “Step 7: Complete Signal Generator”

Here’s a complete congressional signal generator using as_dataframe=True:

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


class CongressionalSignals:
    def __init__(self, api_key):
        self.fb = FinBrainClient(api_key=api_key)
        self.large_amounts = [
            "$500,001 - $1,000,000",
            "$1,000,001 - $5,000,000",
            "Over $5,000,000"
        ]


    def generate_signal(self, market, ticker):
        """Generate trading signal from congressional activity"""
        try:
            df = self.fb.house_trades.ticker(market, ticker, as_dataframe=True)
        except Exception:
            return {"signal": "neutral", "reason": "no_data"}


        if df.empty:
            return {"signal": "neutral", "reason": "no_trades"}


        # Recent trades (last 60 days)
        cutoff = datetime.now() - timedelta(days=60)
        recent = df[df.index > cutoff]


        if recent.empty:
            return {"signal": "neutral", "reason": "no_recent_trades"}


        # Analyze trades using DataFrame operations
        purchases = recent[recent["type"] == "Purchase"]
        sales = recent[recent["type"] == "Sale"]


        large_purchases = purchases[purchases["amount"].isin(self.large_amounts)]
        large_sales = sales[sales["amount"].isin(self.large_amounts)]


        # Multiple buyers check
        unique_buyers = purchases["representative"].nunique()
        has_multiple_buyers = unique_buyers >= 2


        # Generate signal
        score = 0
        reasons = []


        # Purchase/sale ratio
        if len(purchases) > len(sales) * 2:
            score += 2
            reasons.append(f"{len(purchases)} purchases vs {len(sales)} sales")
        elif len(sales) > len(purchases) * 2:
            score -= 2
            reasons.append(f"More sales than purchases")


        # Large trades
        if len(large_purchases) > 0:
            score += 2
            reasons.append(f"{len(large_purchases)} large purchases")
        if len(large_sales) > 0:
            score -= 1
            reasons.append(f"{len(large_sales)} large sales")


        # Multiple buyers
        if has_multiple_buyers:
            score += 1
            reasons.append(f"{unique_buyers} unique buyers")


        # Determine signal
        if score >= 3:
            signal = "strong_buy"
        elif score >= 1:
            signal = "buy"
        elif score <= -3:
            signal = "strong_sell"
        elif score <= -1:
            signal = "sell"
        else:
            signal = "neutral"


        return {
            "ticker": ticker,
            "signal": signal,
            "score": score,
            "reasons": reasons,
            "purchases": len(purchases),
            "sales": len(sales),
            "unique_buyers": unique_buyers
        }


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


        for ticker in tickers:
            result = self.generate_signal(market, ticker)
            if result["signal"] != "neutral":
                signals.append(result)


        # Sort by score
        return sorted(signals, key=lambda x: x["score"], reverse=True)




# Use the signal generator
signals = CongressionalSignals(api_key="YOUR_API_KEY")


universe = [
    "NVDA", "AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA",
    "JPM", "BAC", "GS", "V", "MA",
    "JNJ", "PFE", "UNH", "ABBV"
]


results = signals.scan_universe(universe)


print("\n=== Congressional Signals ===")
for r in results:
    print(f"\n{r['ticker']}: {r['signal']} (score: {r['score']})")
    print(f"  Purchases: {r['purchases']}, Sales: {r['sales']}")
    print(f"  Unique Buyers: {r['unique_buyers']}")
    for reason in r["reasons"]:
        print(f"  - {reason}")

Important Considerations

Section titled “Important Considerations”
  1. Disclosure delays - Trades are reported up to 45 days after execution
  2. Not investment advice - This data is for research purposes only
  3. Past performance - Historical results don’t guarantee future returns
  4. Legal boundaries - Only use publicly available information
Section titled “Related Resources”