Write Covered Calls Automatically with AI

import requests
import time
from dataclasses import dataclass
from typing import Optional

BASE    = "https://purpleflea.com/api/v1"
API_KEY = "pf_live_your_key_here"
HDR     = {"X-API-Key": API_KEY}

@dataclass
class CallPosition:
    position_id:    str
    asset:          str
    strike:         float
    expiry:         str
    premium_usd:    float
    quantity:       float
    current_delta:  float = 0.30

class CoveredCallAgent:
    """
    Writes covered calls on BTC/ETH holdings.
    Automatically rolls when delta exceeds threshold.
    """

    def __init__(
        self,
        asset: str           = "ETH",
        target_delta: float  = 0.30,
        roll_delta: float    = 0.70,  # roll if delta exceeds this
        quantity: float      = 1.0,   # units of asset to cover
    ):
        self.asset        = asset
        self.target_delta = target_delta
        self.roll_delta   = roll_delta
        self.quantity     = quantity
        self.positions    = []
        self.total_premium= 0.0

    def get_best_call(self) -> Optional[dict]:
        """Find the best call to write at our target delta."""
        r = requests.get(f"{BASE}/options/calls", headers=HDR, params={
            "asset":       self.asset,
            "delta_min":   self.target_delta - 0.05,
            "delta_max":   self.target_delta + 0.05,
            "expiry_type": "weekly"
        })
        calls = r.json()["calls"]
        if not calls:
            return None
        # Pick highest premium-to-delta ratio
        return max(calls, key=lambda c: c["premium_usd"] / c["delta"])

    def write_call(self, call: dict) -> CallPosition:
        """Execute the covered call write."""
        r = requests.post(f"{BASE}/options/write", headers=HDR, json={
            "asset":    self.asset,
            "strike":   call["strike"],
            "expiry":   call["expiry"],
            "quantity": self.quantity
        })
        data = r.json()
        pos  = CallPosition(
            position_id  = data["position_id"],
            asset        = self.asset,
            strike       = call["strike"],
            expiry       = call["expiry"],
            premium_usd  = data["premium_usd"],
            quantity     = self.quantity
        )
        self.total_premium += pos.premium_usd
        print(f"Wrote {asset} call | strike=${pos.strike:,.0f} | exp={pos.expiry} | premium=${pos.premium_usd:.2f}")
        return pos

    def monitor_and_roll(self, pos: CallPosition):
        """Check current delta and roll if needed."""
        r = requests.get(f"{BASE}/portfolio/coverage", headers=HDR,
                        params={"position_id": pos.position_id})
        live_delta = r.json()["current_delta"]
        print(f"Position {pos.position_id}: current delta={live_delta:.2f}")

        if live_delta >= self.roll_delta:
            print(f"Delta {live_delta:.2f} >= threshold {self.roll_delta} → rolling up")
            roll_r = requests.post(f"{BASE}/options/roll", headers=HDR, json={
                "position_id":  pos.position_id,
                "roll_type":    "roll_up",
                "new_delta":    self.target_delta
            })
            roll_data = roll_r.json()
            print(f"Rolled → new strike=${roll_data['new_strike']:,.0f} | net_credit=${roll_data['net_credit']:.2f}")
            self.total_premium += roll_data["net_credit"]

    def run(self, check_interval_sec: int = 3600):
        """Main loop: write calls, monitor, roll as needed."""
        print(f"CoveredCallAgent starting | asset={self.asset} | delta={self.target_delta}")
        while True:
            if not self.positions:
                call = self.get_best_call()
                if call:
                    pos = self.write_call(call)
                    self.positions.append(pos)
            else:
                for pos in self.positions:
                    self.monitor_and_roll(pos)
            print(f"Total premium collected: ${self.total_premium:.2f}")
            time.sleep(check_interval_sec)

if __name__ == "__main__":
    agent = CoveredCallAgent(asset="ETH", target_delta=0.30, quantity=2.0)
    agent.run()