AI Agent Options Strategies: Selling Premium and Managing Greeks

import numpy as np
from scipy.stats import norm
from dataclasses import dataclass
from typing import Literal

@dataclass
class OptionGreeks:
    delta: float
    gamma: float
    theta: float    # per calendar day
    vega: float     # per 1% IV move
    rho: float
    price: float
    iv: float
    intrinsic: float
    extrinsic: float

class BlackScholes:
    """
    Black-Scholes option pricing and Greeks.
    All IV inputs are annualized (e.g., 0.80 = 80% annualized IV).
    """

    def __init__(self, risk_free_rate: float = 0.05):
        self.r = risk_free_rate

    def _d1_d2(self, S, K, T, sigma):
        # Avoid log(0) when deep ITM/OTM at expiry
        if T <= 0 or sigma <= 0:
            return 0.0, 0.0
        d1 = (np.log(S / K) + (self.r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
        d2 = d1 - sigma * np.sqrt(T)
        return d1, d2

    def price_and_greeks(
        self,
        option_type: Literal['call', 'put'],
        S: float,       # Current spot price
        K: float,       # Strike price
        T: float,       # Time to expiry in years (e.g., 30/365)
        sigma: float,   # Annualized implied volatility (e.g., 0.80)
    ) -> OptionGreeks:

        d1, d2 = self._d1_d2(S, K, T, sigma)
        r = self.r
        N = norm.cdf
        n = norm.pdf

        if option_type == 'call':
            price = S * N(d1) - K * np.exp(-r * T) * N(d2)
            delta = N(d1)
            intrinsic = max(S - K, 0)
        else:
            price = K * np.exp(-r * T) * N(-d2) - S * N(-d1)
            delta = N(d1) - 1
            intrinsic = max(K - S, 0)

        # Gamma is the same for calls and puts
        gamma = n(d1) / (S * sigma * np.sqrt(T)) if T > 0 else 0

        # Theta: rate of change per calendar day
        if option_type == 'call':
            theta_annual = (
                -(S * n(d1) * sigma) / (2 * np.sqrt(T))
                - r * K * np.exp(-r * T) * N(d2)
            ) if T > 0 else 0
        else:
            theta_annual = (
                -(S * n(d1) * sigma) / (2 * np.sqrt(T))
                + r * K * np.exp(-r * T) * N(-d2)
            ) if T > 0 else 0
        theta = theta_annual / 365

        # Vega: price change per 1% move in IV
        vega = S * n(d1) * np.sqrt(T) / 100 if T > 0 else 0

        # Rho: price change per 1% move in risk-free rate
        if option_type == 'call':
            rho = K * T * np.exp(-r * T) * N(d2) / 100
        else:
            rho = -K * T * np.exp(-r * T) * N(-d2) / 100

        return OptionGreeks(
            delta=round(delta, 4),
            gamma=round(gamma, 6),
            theta=round(theta, 4),
            vega=round(vega, 4),
            rho=round(rho, 4),
            price=round(price, 4),
            iv=sigma,
            intrinsic=round(intrinsic, 4),
            extrinsic=round(price - intrinsic, 4),
        )

# Example: BTC 30-delta put, 30 DTE, IV=80%
bs = BlackScholes(risk_free_rate=0.05)
greeks = bs.price_and_greeks(
    option_type='put',
    S=85000,    # BTC spot
    K=78000,    # Strike (~8% OTM)
    T=30/365,   # 30 days to expiry
    sigma=0.80  # 80% annualized IV
)
print(f"Price: ${greeks.price:.2f}")
print(f"Delta: {greeks.delta:.4f} | Gamma: {greeks.gamma:.6f}")
print(f"Theta: ${greeks.theta:.2f}/day | Vega: ${greeks.vega:.2f}/1%")
print(f"Extrinsic (collectible premium): ${greeks.extrinsic:.2f}")

import asyncio
import aiohttp
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, List
from black_scholes import BlackScholes, OptionGreeks

PURPLEFLEA_API = "https://purpleflea.com/trading-api"
API_KEY = "your-api-key-here"

@dataclass
class MarketConditions:
    symbol: str
    spot: float
    current_iv: float
    iv_rank: float      # 0-100, where 100 = highest IV in past year
    iv_percentile: float
    realized_vol_30d: float
    trend: str           # 'bullish', 'bearish', 'neutral'
    atr_pct: float        # Average True Range as % of spot

@dataclass
class OptionsPosition:
    symbol: str
    strategy: str
    legs: List[dict]
    net_credit: float
    max_profit: float
    max_loss: float
    prob_profit: float
    expiry: str
    greeks: dict = field(default_factory=dict)

class OptionsAgent:
    """
    Systematic options premium-selling agent.
    Uses Purple Flea Trading API for market data and order execution.
    Supports: covered calls, CSPs, iron condors, short strangles.
    """

    def __init__(self, api_key: str, portfolio_value: float):
        self.api_key = api_key
        self.portfolio_value = portfolio_value
        self.bs = BlackScholes()
        self.max_position_pct = 0.05   # 5% max per position
        self.target_dte = 30           # Target 30 DTE for entry
        self.profit_target_pct = 0.50  # Close at 50% max profit
        self.stop_loss_mult = 2.0      # Close at 2x credit received
        self.open_positions: List[OptionsPosition] = []

    async def get_market_conditions(self, symbol: str) -> MarketConditions:
        async with aiohttp.ClientSession() as session:
            headers = {"X-API-Key": self.api_key}
            url = f"{PURPLEFLEA_API}/options/market-data/{symbol}"
            async with session.get(url, headers=headers) as resp:
                data = await resp.json()

        return MarketConditions(
            symbol=symbol,
            spot=data['spot'],
            current_iv=data['implied_volatility'],
            iv_rank=data['iv_rank'],
            iv_percentile=data['iv_percentile'],
            realized_vol_30d=data['realized_vol_30d'],
            trend=data['trend'],
            atr_pct=data['atr_pct'],
        )

    def select_strategy(self, conditions: MarketConditions) -> str:
        """
        Strategy selection matrix based on IV rank and trend.
        Returns strategy name for position construction.
        """
        iv_rank = conditions.iv_rank
        trend = conditions.trend

        # Very high IV: iron condors and short strangles capture max premium
        if iv_rank >= 60:
            if trend == 'neutral':
                return 'iron_condor'
            else:
                return 'strangle'

        # Medium-high IV: directional plays with defined risk
        elif iv_rank >= 30:
            if trend == 'bullish':
                return 'cash_secured_put'
            elif trend == 'bearish':
                return 'covered_call'
            else:
                return 'iron_condor'

        # Low IV: avoid selling premium, wait for better conditions
        else:
            return 'no_trade'  # IV too low to sell

    def build_iron_condor(
        self, conditions: MarketConditions
    ) -> Optional[OptionsPosition]:
        """
        Constructs iron condor at standard parameters:
        - Short strikes at 0.20 delta (~16% OTM)
        - Long strikes 5% further OTM for defined risk
        - 30 DTE entry
        """
        S = conditions.spot
        sigma = conditions.current_iv
        T = self.target_dte / 365

        # Find strikes at target deltas by iteration
        short_put_strike = self._find_strike_by_delta(
            S, sigma, T, 'put', target_delta=-0.20
        )
        short_call_strike = self._find_strike_by_delta(
            S, sigma, T, 'call', target_delta=0.20
        )
        long_put_strike = short_put_strike * 0.95   # 5% below short put
        long_call_strike = short_call_strike * 1.05  # 5% above short call

        # Price each leg
        sp_g = self.bs.price_and_greeks('put', S, short_put_strike, T, sigma)
        lp_g = self.bs.price_and_greeks('put', S, long_put_strike, T, sigma)
        sc_g = self.bs.price_and_greeks('call', S, short_call_strike, T, sigma)
        lc_g = self.bs.price_and_greeks('call', S, long_call_strike, T, sigma)

        net_credit = (sp_g.price - lp_g.price) + (sc_g.price - lc_g.price)
        put_spread_width = short_put_strike - long_put_strike
        call_spread_width = long_call_strike - short_call_strike
        max_loss = max(put_spread_width, call_spread_width) - net_credit

        # Probability of profit: approximately N(d2) for short strikes
        prob_profit = 1 - (sp_g.delta + sc_g.delta)  # rough estimate

        expiry = (datetime.now() + timedelta(days=self.target_dte)).strftime("%Y-%m-%d")

        return OptionsPosition(
            symbol=conditions.symbol,
            strategy='iron_condor',
            legs=[
                {'type': 'put', 'action': 'sell', 'strike': short_put_strike},
                {'type': 'put', 'action': 'buy', 'strike': long_put_strike},
                {'type': 'call', 'action': 'sell', 'strike': short_call_strike},
                {'type': 'call', 'action': 'buy', 'strike': long_call_strike},
            ],
            net_credit=round(net_credit, 4),
            max_profit=round(net_credit, 4),
            max_loss=round(max_loss, 4),
            prob_profit=round(prob_profit, 3),
            expiry=expiry,
            greeks={
                'net_delta': sp_g.delta + sc_g.delta - lp_g.delta - lc_g.delta,
                'net_theta': sp_g.theta + sc_g.theta - lp_g.theta - lc_g.theta,
                'net_vega': sp_g.vega + sc_g.vega - lp_g.vega - lc_g.vega,
            }
        )

    def _find_strike_by_delta(self, S, sigma, T, opt_type, target_delta):
        # Binary search for strike matching target delta
        low, high = S * 0.50, S * 1.50
        for _ in range(50):  # 50 iterations is precise enough
            mid = (low + high) / 2
            g = self.bs.price_and_greeks(opt_type, S, mid, T, sigma)
            if abs(g.delta - target_delta) < 0.001:
                break
            if opt_type == 'call':
                if g.delta > target_delta: low = mid
                else: high = mid
            else:
                if g.delta < target_delta: low = mid
                else: high = mid
        return round(mid / 100) * 100  # Round to nearest $100 strike


    async def run_once(self, symbols: List[str]):
        for symbol in symbols:
            conditions = await self.get_market_conditions(symbol)
            strategy = self.select_strategy(conditions)
            print(f"[{symbol}] IV Rank: {conditions.iv_rank:.0f}% | Strategy: {strategy}")

            if strategy == 'iron_condor':
                position = self.build_iron_condor(conditions)
                if position:
                    print(f"  Credit: ${position.net_credit:.2f} | PoP: {position.prob_profit:.1%}")
                    print(f"  Theta/day: ${position.greeks['net_theta']:.3f}")

# Run the agent
async def main():
    agent = OptionsAgent(api_key=API_KEY, portfolio_value=100000)
    await agent.run_once(['BTC-USD', 'ETH-USD', 'SOL-USD'])

asyncio.run(main())