bull put spread research script

spread_research.py

from alpaca.data.historical import StockHistoricalDataClient
from alpaca.data.requests import StockSnapshotRequest
from alpaca.data.enums import DataFeed
from alpaca.data.historical import OptionHistoricalDataClient
from alpaca.data.requests import OptionChainRequest, StockBarsRequest, OptionBarsRequest, OptionSnapshotRequest
from alpaca.data.enums import OptionsFeed
from alpaca.trading.enums import ContractType, AssetClass
from alpaca.trading.enums import QueryOrderStatus, OrderSide, OrderClass, TimeInForce, OrderStatus, OrderType, PositionIntent
from alpaca.trading.client import TradingClient
from alpaca.trading.requests import GetOrdersRequest, LimitOrderRequest, TakeProfitRequest, StopLimitOrderRequest, GetOptionContractsRequest, OptionLegRequest
from alpaca.data.timeframe import TimeFrame
import argparse
import os
import statistics
from datetime import datetime, timezone, timedelta
import time
import re
import pandas as pd
import numpy as np

API_KEY = "*********************"
SECRET_KEY = "************************"
PAPER = False

import re
from collections import defaultdict
from datetime import datetime, timedelta

# Existing imports from alpaca-py assumed above:
# OptionHistoricalDataClient, StockHistoricalDataClient, StockSnapshotRequest,
# OptionChainRequest, DataFeed, OptionsFeed, ContractType

TICKERS = ["XSP", "KO"]
NUMBER_OF_DAYS = 60
MINIUM_CREDIT = 0.05
MAX_STRIKE_WIDTH = 1
CUR_PRICE_FRAC = 0.98


def parse_occ_symbol(symbol: str):
    """Extracts strike price and expiration date from a standard OCC option symbol."""
    match = re.match(r"^([A-Z]+)(\d{6})([CP])(\d{8})$", symbol)
    if match:
        _, exp_str, _, strike_str = match.groups()
        strike = int(strike_str) / 1000.0
        return strike, exp_str
    return None, None


if __name__ == "__main__":
    client = OptionHistoricalDataClient(API_KEY, SECRET_KEY)
    data_client = StockHistoricalDataClient(API_KEY, SECRET_KEY)

    current_date = datetime.now()
    end_date = current_date + timedelta(days=NUMBER_OF_DAYS)

    for t in TICKERS:
        request_params = StockSnapshotRequest(symbol_or_symbols=t if t != "XSP" else "SPY", feed=DataFeed.SIP)
        snapshot = data_client.get_stock_snapshot(request_params)
        stock_data = snapshot[t if t != "XSP" else "SPY"]

        latest_ask_price = stock_data.latest_quote.ask_price
        latest_bid_price = stock_data.latest_quote.bid_price
        mid_price = round(((latest_ask_price + latest_bid_price) / 2) * CUR_PRICE_FRAC, 2)
        print(f"\n--- [{t}] Current Stock Mid Price: ${mid_price} ---")

        opt_params = OptionChainRequest(
            underlying_symbol=t,
            feed=OptionsFeed.OPRA,
            expiration_date_gte=current_date.date().isoformat(),
            expiration_date_lte=end_date.date().isoformat(),
            type=ContractType.PUT,
            strike_price_lte=mid_price,
        )
        resp = client.get_option_chain(opt_params)

        # Group valid option contracts by expiration
        exp_groups = defaultdict(list)
        for occ_symbol, snapshot_data in resp.items():
            if not snapshot_data.latest_quote:
                continue

            strike, exp_date = parse_occ_symbol(occ_symbol)
            if strike is not None:
                exp_groups[exp_date].append(
                    {
                        "symbol": occ_symbol,
                        "strike": strike,
                        "bid": snapshot_data.latest_quote.bid_price,
                        "ask": snapshot_data.latest_quote.ask_price,
                    }
                )

        # Evaluate option pairs per expiration date
        for exp_date, options in exp_groups.items():
            # Sort contracts ascending by strike price
            options.sort(key=lambda x: x["strike"])

            for i in range(len(options)):
                for j in range(i):
                    short_opt = options[i]  # Higher strike put (sold)
                    long_opt = options[j]  # Lower strike put (bought)

                    strike_width = round(short_opt["strike"] - long_opt["strike"], 2)

                    # Filter by allowed strike width (supports 0.5, 1.0, 2.0, etc.)
                    if 0.5 <= strike_width <= MAX_STRIKE_WIDTH:
                        # Natural Credit: Sell short at Bid, Buy long at Ask
                        natural_credit = round(short_opt["bid"] - long_opt["ask"], 2)

                        # Mid Credit: Spread difference between mid prices
                        short_mid = (short_opt["bid"] + short_opt["ask"]) / 2
                        long_mid = (long_opt["bid"] + long_opt["ask"]) / 2
                        mid_credit = round(short_mid - long_mid, 2)

                        meets_natural = natural_credit >= MINIUM_CREDIT
                        meets_mid = mid_credit >= MINIUM_CREDIT

                        # Print only if at least one pricing method meets the threshold
                        if meets_natural or meets_mid:
                            credit_details = []
                            if meets_natural:
                                credit_details.append(
                                    f"Natural Credit = ${natural_credit:.2f}"
                                )
                            if meets_mid:
                                credit_details.append(
                                    f"Mid Credit = ${mid_credit:.2f}"
                                )

                            pricing_str = " | ".join(credit_details)

                            print(f"\n[MATCH] {t} | Exp: {exp_date} | Width: ${strike_width:.2f}")
                            print(f"  Short OCC: {short_opt['symbol']} (Strike: ${short_opt['strike']:.2f}) Bid: ${short_opt['bid']}  Ask: ${short_opt['ask']}")
                            print(f"  Long OCC:  {long_opt['symbol']} (Strike: ${long_opt['strike']:.2f}) Bid: ${long_opt['bid']}  Ask: ${long_opt['ask']}")
                            print(f"  Satisfied: {pricing_str}")