a alpaca based cli script to search bull put spreads

bull_put_search.py

import argparse
import json
import os
from datetime import datetime, timedelta
from alpaca.data.historical.option import OptionHistoricalDataClient
from alpaca.data.historical.stock import StockHistoricalDataClient
from alpaca.data.requests import (
    OptionChainRequest,
    StockLatestTradeRequest
)
from alpaca.data.enums import DataFeed

def get_alpaca_clients():
    """Initialize Alpaca Data Clients using environment variables."""
    api_key = os.getenv("ALPACA_API_KEY")
    secret_key = os.getenv("ALPACA_SECRET_KEY")
    
    if not api_key or not secret_key:
        raise ValueError("Missing ALPACA_API_KEY or ALPACA_SECRET_KEY environment variables.")
    
    option_client = OptionHistoricalDataClient(api_key, secret_key)
    stock_client = StockHistoricalDataClient(api_key, secret_key)
    return option_client, stock_client

def get_underlying_price(stock_client, symbol):
    """Fetch latest trade price or fallback to last close price."""
    try:
        req = StockLatestTradeRequest(symbol_or_symbols=symbol)
        trade = stock_client.get_stock_latest_trade(req)
        if symbol in trade and trade[symbol].price:
            return float(trade[symbol].price)
    except Exception:
        pass
    return None

def find_bull_put_spreads(args):
    option_client, stock_client = get_alpaca_clients()
    results = []
    today = datetime.now().date()

    for symbol in args.tick:
        current_price = get_underlying_price(stock_client, symbol)
        
        if not current_price:
            if args.debug:
                print(f"[DEBUG] Could not fetch price for {symbol}")
            continue

        # Calculate DTE date range
        min_exp_date = today + timedelta(days=args.min_dte)
        max_exp_date = today + timedelta(days=args.max_dte)

        # Request option chain from Alpaca Premium OPRA feed
        req = OptionChainRequest(
            underlying_symbol=symbol,
            type="put",
            expiration_date_gte=min_exp_date,
            expiration_date_lte=max_exp_date
        )

        try:
            chain = option_client.get_option_chain(req)
        except Exception as e:
            if args.debug:
                print(f"[DEBUG] Failed to fetch option chain for {symbol}: {e}")
            continue

        if not chain:
            continue

        # Group chain snapshots by expiration date
        exp_groups = {}
        for occ_symbol, snapshot in chain.items():
            # Alpaca OCC symbol parsing or snapshot contract inspection
            # Standard OCC symbol: SYMBOLYYMMDD[C/P]STRIKE
            exp_date_str = snapshot.details.expiration_date if hasattr(snapshot, 'details') and snapshot.details else None
            
            # Fallback OCC parsing if details field is sparse
            if not exp_date_str:
                # Extract expiration from OCC string (e.g. AAPL261016P00215000 -> 261016)
                raw_code = occ_symbol.replace(symbol, "")
                exp_date_str = f"20{raw_code[:2]}-{raw_code[2:4]}-{raw_code[4:6]}"

            if exp_date_str not in exp_groups:
                exp_groups[exp_date_str] = []
            exp_groups[exp_date_str].append((occ_symbol, snapshot))

        # Process each expiration group
        for exp_str, options_list in exp_groups.items():
            exp_date = datetime.strptime(exp_str, "%Y-%m-%d").date()
            dte = (exp_date - today).days

            # Extract strike and price metadata with overnight fallback
            parsed_puts = []
            for occ_symbol, snap in options_list:
                latest_quote = snap.latest_quote
                latest_trade = snap.latest_trade
                greeks = snap.greeks

                # 1. Bid/Ask extraction
                bid = float(latest_quote.bid_price) if latest_quote and latest_quote.bid_price else 0.0
                ask = float(latest_quote.ask_price) if latest_quote and latest_quote.ask_price else 0.0
                
                # 2. Last known market data fallback (if quotes clear to 0 overnight)
                if bid == 0.0 and ask == 0.0:
                    if latest_trade and latest_trade.price:
                        bid = ask = float(latest_trade.price)

                # 3. Native Alpaca Greeks (Theta)
                theta = float(greeks.theta) if greeks and greeks.theta is not None else 0.0
                iv = float(snap.implied_volatility) if snap and snap.implied_volatility is not None else 0.0

                # Derive strike from contract details or OCC string
                if hasattr(snap, 'details') and snap.details and snap.details.strike_price:
                    strike = float(snap.details.strike_price)
                else:
                    # Parse strike from OCC symbol (last 8 digits divided by 1000)
                    strike = float(occ_symbol[-8:]) / 1000.0

                parsed_puts.append({
                    "occ_symbol": occ_symbol,
                    "strike": strike,
                    "bid": bid,
                    "ask": ask,
                    "mid": (bid + ask) / 2.0 if (bid + ask) > 0 else 0.0,
                    "theta": theta,
                    "iv": iv
                })

            max_short_strike = current_price * args.max_short_frac
            min_short_strike = current_price * args.min_short_frac

            short_candidates = [
                p for p in parsed_puts 
                if min_short_strike <= p['strike'] <= max_short_strike and p['mid'] > 0
            ]

            for short_opt in short_candidates:
                short_strike = short_opt['strike']
                
                # Find valid Long Puts
                long_candidates = [
                    p for p in parsed_puts 
                    if (short_strike - args.max_width) <= p['strike'] < short_strike
                ]

                for long_opt in long_candidates:
                    long_strike = long_opt['strike']
                    strike_width = short_strike - long_strike
                    
                    net_mid_credit = short_opt['mid'] - long_opt['mid']
                    net_nat_credit = short_opt['bid'] - long_opt['ask']

                    if net_mid_credit < args.min_credit:
                        continue

                    # Native Alpaca daily Theta per contract ($ = theta * 100)
                    # Short theta is positive income (+), long theta is negative decay (-)
                    net_theta_usd = (short_opt['theta'] - long_opt['theta']) * 100.0

                    if args.min_theta is not None and net_theta_usd < args.min_theta:
                        continue

                    spread_data = {
                        "underlying": symbol,
                        "underlying_price": round(current_price, 2),
                        "expiration": exp_str,
                        "dte": dte,
                        "metrics": {
                            "strike_width": round(strike_width, 2),
                            "max_credit_mid": round(net_mid_credit, 2),
                            "credit_natural": round(net_nat_credit, 2),
                            "max_loss": round(strike_width - net_mid_credit, 2),
                            "return_on_risk_pct": round((net_mid_credit / (strike_width - net_mid_credit)) * 100, 2) if (strike_width - net_mid_credit) > 0 else 0,
                            "net_theta_daily_usd": round(net_theta_usd, 2)
                        },
                        "legs": {
                            "short_put": {
                                "occ_symbol": short_opt['occ_symbol'],
                                "strike": short_strike,
                                "bid": round(short_opt['bid'], 2),
                                "ask": round(short_opt['ask'], 2),
                                "mid": round(short_opt['mid'], 2),
                                "iv": round(short_opt['iv'], 4)
                            },
                            "long_put": {
                                "occ_symbol": long_opt['occ_symbol'],
                                "strike": long_strike,
                                "bid": round(long_opt['bid'], 2),
                                "ask": round(long_opt['ask'], 2),
                                "mid": round(long_opt['mid'], 2),
                                "iv": round(long_opt['iv'], 4)
                            }
                        }
                    }
                    results.append(spread_data)

    return results

def main():
    parser = argparse.ArgumentParser(description="Find Bull Put Spreads using Alpaca OPRA Premium Data.")
    
    parser.add_argument("--tick", nargs="+", required=True, help="List of ticker symbols")
    parser.add_argument("--max-short-frac", type=float, default=0.98, help="Max short strike / market price ratio")
    parser.add_argument("--min-short-frac", type=float, default=0.80, help="Min short strike / market price ratio")
    parser.add_argument("--max-width", type=float, default=5.0, help="Max strike width in dollars")
    parser.add_argument("--min-credit", type=float, default=0.05, help="Min net credit collected (default: 0.01)")
    parser.add_argument("--min-dte", type=int, default=14, help="Minimum days to expiration")
    parser.add_argument("--max-dte", type=int, default=180, help="Maximum days to expiration")
    parser.add_argument("--min-theta", type=float, default=None, help="Minimum daily net theta yield in USD")
    parser.add_argument("--debug", action="store_true", help="Print debug logs")

    args = parser.parse_args()
    
    output = find_bull_put_spreads(args)
    print(json.dumps(output, indent=2))

if __name__ == "__main__":
    main()