How to Build an AI Trading Bot Using Congressional Insider Data

pip install requests alpaca-trade-api python-dotenv schedule

GOVGREED_API_KEY=your_govgreed_key_here
ALPACA_API_KEY=your_alpaca_key_here
ALPACA_SECRET_KEY=your_alpaca_secret_here
ALPACA_BASE_URL=https://paper-api.alpaca.markets

import requests
import os
from dotenv import load_dotenv
from datetime import datetime, timedelta

load_dotenv()

GOVGREED_BASE = "https://api.govgreed.com/v1"
HEADERS = {"Authorization": f"Bearer {os.getenv('GOVGREED_API_KEY')}"}


def get_triple_signals(min_score=50, limit=20):
    """
    Fetch triple signals: politician on committee + traded stock + took contributions.
    These are the highest-conviction insider signals.
    """
    resp = requests.get(
        f"{GOVGREED_BASE}/signals/triple",
        headers=HEADERS,
        params={
            "min_score": min_score,
            "limit": limit,
            "order": "opportunity_score.desc"
        }
    )
    resp.raise_for_status()
    return resp.json().get("data", [])


def get_upcoming_markups(days_ahead=14):
    """
    Fetch committee markups scheduled in the next N days.
    Markups = bill is moving. This is the buy window.
    """
    resp = requests.get(
        f"{GOVGREED_BASE}/signals/markup-calendar",
        headers=HEADERS,
        params={
            "days_ahead": days_ahead,
            "has_signals": True
        }
    )
    resp.raise_for_status()
    return resp.json().get("data", [])


def get_exec_timing_signals(min_score=2.0):
    """
    Corporate exec buys within 90 days before a congressional vote.
    Use as confirmation — exec agrees with the congressional signal.
    """
    resp = requests.get(
        f"{GOVGREED_BASE}/signals/exec-timing",
        headers=HEADERS,
        params={
            "category": "ahead_of_vote",
            "transaction_type": "purchase",
            "min_timing_score": min_score
        }
    )
    resp.raise_for_status()
    return resp.json().get("data", [])


def build_signal_universe():
    """
    Merge all signal types into a ranked list of actionable tickers.
    Returns: [{ticker, score, sources, days_to_markup}]
    """
    triple = get_triple_signals(min_score=50)
    markups = get_upcoming_markups(days_ahead=14)
    exec_signals = get_exec_timing_signals()

    # Build ticker → score map
    ticker_scores = {}

    for signal in triple:
        ticker = signal["ticker"]
        if ticker not in ticker_scores:
            ticker_scores[ticker] = {"score": 0, "sources": [], "days_to_markup": None}
        ticker_scores[ticker]["score"] += signal.get("opportunity_score", 0) * 1.2
        ticker_scores[ticker]["sources"].append("triple")

    for markup in markups:
        for ticker in markup.get("affected_tickers", []):
            if ticker not in ticker_scores:
                ticker_scores[ticker] = {"score": 0, "sources": [], "days_to_markup": None}
            ticker_scores[ticker]["score"] += 30  # markup bonus
            ticker_scores[ticker]["sources"].append("markup")
            ticker_scores[ticker]["days_to_markup"] = markup.get("days_away")

    for exec_sig in exec_signals:
        ticker = exec_sig["ticker"]
        if ticker not in ticker_scores:
            ticker_scores[ticker] = {"score": 0, "sources": [], "days_to_markup": None}
        ticker_scores[ticker]["score"] += exec_sig.get("timing_score", 0) * 0.8
        ticker_scores[ticker]["sources"].append("exec")

    # Sort by composite score, return top candidates
    ranked = sorted(
        [{"ticker": k, **v} for k, v in ticker_scores.items()],
        key=lambda x: x["score"], reverse=True
    )
    return [r for r in ranked if r["score"] > 40]  # filter weak signals

import alpaca_trade_api as tradeapi
import os
from dotenv import load_dotenv

load_dotenv()

alpaca = tradeapi.REST(
    os.getenv("ALPACA_API_KEY"),
    os.getenv("ALPACA_SECRET_KEY"),
    base_url=os.getenv("ALPACA_BASE_URL")
)


def get_portfolio_value():
    account = alpaca.get_account()
    return float(account.portfolio_value)


def get_current_positions():
    return {p.symbol: p for p in alpaca.list_positions()}


def size_position(score, portfolio_value, max_position_pct=0.05):
    """
    Kelly-inspired sizing: higher score = larger position.
    Never exceed 5% of portfolio on a single political signal.
    """
    # Score 40-500+ → position 1%-5% of portfolio
    raw_pct = (score - 40) / (500 - 40) * max_position_pct
    pct = min(max(0.01, raw_pct), max_position_pct)
    return portfolio_value * pct


def is_tradeable(ticker):
    """Check if ticker is tradeable on Alpaca"""
    try:
        asset = alpaca.get_asset(ticker)
        return asset.tradable and asset.status == "active"
    except:
        return False


def execute_signal(ticker, score, reason=""):
    """
    Execute a buy order for a given signal.
    Uses market order — signal is more important than price precision.
    """
    if not is_tradeable(ticker):
        print(f"⚠️  {ticker} not tradeable, skipping")
        return None

    positions = get_current_positions()
    if ticker in positions:
        print(f"⏭️  Already holding {ticker}, skipping")
        return None

    portfolio_value = get_portfolio_value()
    position_value = size_position(score, portfolio_value)

    # Get current price for share calculation
    latest = alpaca.get_latest_trade(ticker)
    shares = int(position_value / latest.price)

    if shares < 1:
        print(f"⚠️  Position too small for {ticker} ({shares} shares)")
        return None

    order = alpaca.submit_order(
        symbol=ticker,
        qty=shares,
        side="buy",
        type="market",
        time_in_force="day"
    )

    print(f"✅  Bought {shares}x {ticker} @ ~${latest.price:.2f}")
    print(f"   Signal score: {score:.1f} | {reason}")
    return order

import schedule
import time
from govgreed_signals import build_signal_universe
from executor import execute_signal, get_current_positions

# Configuration
MAX_POSITIONS = 8        # never hold more than 8 political signals at once
MIN_SIGNAL_SCORE = 60   # only act on strong signals
REQUIRE_MARKUP = False   # set True to only trade markup-linked signals


def run_bot():
    print(f"\n{'='*50}")
    print(f"🏛️  GovGreed Bot — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
    print(f"{'='*50}")

    # Check position limits
    current_positions = get_current_positions()
    available_slots = MAX_POSITIONS - len(current_positions)

    if available_slots <= 0:
        print("📊  Portfolio full. No new positions today.")
        return

    # Fetch and rank all signals
    signals = build_signal_universe()
    print(f"📡  Found {len(signals)} signals above threshold")

    new_trades = 0
    for signal in signals:
        if new_trades >= available_slots:
            break

        ticker = signal["ticker"]
        score = signal["score"]
        sources = signal["sources"]
        days_to_markup = signal.get("days_to_markup")

        # Quality filters
        if score < MIN_SIGNAL_SCORE:
            continue
        if REQUIRE_MARKUP and "markup" not in sources:
            continue
        if ticker in current_positions:
            continue

        # Build reason string for logging
        reason_parts = []
        if "triple" in sources: reason_parts.append("TRIPLE SIGNAL")
        if "markup" in sources: reason_parts.append(f"MARKUP IN {days_to_markup}d")
        if "exec" in sources: reason_parts.append("EXEC BUY")
        reason = " + ".join(reason_parts)

        order = execute_signal(ticker, score, reason)
        if order:
            new_trades += 1

    print(f"\n✅  Session complete. {new_trades} new positions.")


# Schedule: run daily at 9:45am ET (after open volatility settles)
schedule.every().day.at("09:45").do(run_bot)

print("🤖  GovGreed Trading Bot started. Waiting for market open...")
while True:
    schedule.run_pending()
    time.sleep(60)

import requests
import pandas as pd
from datetime import timedelta

# 1. Pull historical triple signals via GovGreed API
historical = requests.get(
    "https://api.govgreed.com/v1/trades/historical",
    headers=HEADERS,
    params={"has_triple_signal": True, "date_from": "2018-01-01", "date_to": "2024-01-01"}
).json()["data"]

# 2. For each signal, fetch forward price returns from FMP
results = []
for signal in historical:
    ticker = signal["ticker"]
    signal_date = signal["trade_date"]

    price_history = requests.get(
        f"https://financialmodelingprep.com/stable/historical-price-eod/full?symbol={ticker}&apikey=YOUR_FMP_KEY"
    ).json()["historical"]

    df = pd.DataFrame(price_history).set_index("date").sort_index()

    try:
        entry_price = df.loc[signal_date]["close"]
        ret_30d = (df.iloc[df.index.get_loc(signal_date, method="nearest") + 30]["close"] - entry_price) / entry_price
        results.append({"ticker": ticker, "signal_date": signal_date, "ret_30d": ret_30d})
    except: pass

# 3. Analyze
df_results = pd.DataFrame(results)
print(f"Avg 30d return: {df_results['ret_30d'].mean():.1%}")
print(f"Win rate: {(df_results['ret_30d'] > 0).mean():.1%}")
print(f"Sharpe (annualized): {df_results['ret_30d'].mean() / df_results['ret_30d'].std() * (252/30)**0.5:.2f}")