Crypto Derivatives for AI Agents: Futures, Perps, and Options Primer

import requests
import time
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging

log = logging.getLogger("basis_agent")

def fetch_market_data(api_key: str) -> dict:
    """Fetch spot and futures prices from Purple Flea Trading API."""
    resp = requests.get(
        "https://purpleflea.com/trading-api/market-data/btc",
        headers={"X-API-Key": api_key},
        params={"include_futures": True, "include_perps": True}
    )
    resp.raise_for_status()
    return resp.json()

def compute_basis_opportunities(data: dict) -> list:
    """
    Find futures contracts with annualized basis above threshold.
    Returns list of opportunities sorted by yield descending.
    """
    spot_price = data["spot"]["price"]
    opportunities = []

    for fut in data.get("futures", []):
        expiry = datetime.fromisoformat(fut["expiry"])
        days_to_expiry = (expiry - datetime.utcnow()).days

        if days_to_expiry <= 0:
            continue

        futures_price = fut["mark_price"]
        basis = futures_price - spot_price
        basis_pct = basis / spot_price * 100
        annualized_yield = basis_pct * (365 / days_to_expiry)

        opportunities.append({
            "symbol": fut["symbol"],
            "spot_price": spot_price,
            "futures_price": futures_price,
            "basis_usd": round(basis, 2),
            "basis_pct": round(basis_pct, 4),
            "days_to_expiry": days_to_expiry,
            "annualized_yield": round(annualized_yield, 4),
        })

    return sorted(opportunities, key=lambda x: x["annualized_yield"], reverse=True)


class BasisTradeAgent:
    def __init__(self, api_key: str, min_yield: float = 5.0, max_notional: float = 50_000):
        self.api_key = api_key
        self.min_yield = min_yield      # minimum annualized % to enter
        self.max_notional = max_notional
        self.active_trades = []

    def scan_and_trade(self) -> dict:
        data = fetch_market_data(self.api_key)
        opps = compute_basis_opportunities(data)

        if not opps:
            return {"status": "no_opportunities"}

        best = opps[0]
        log.info(f"Best basis: {best['symbol']} @ {best['annualized_yield']:.2f}% annualized")

        if best["annualized_yield"] < self.min_yield:
            return {"status": "yield_too_low", "best_yield": best["annualized_yield"]}

        # Position size: notional in USD
        notional = min(self.max_notional, self.get_available_capital())
        qty_btc = notional / best["spot_price"]

        # Execute: buy spot + sell futures simultaneously
        orders = [
            {"action": "buy", "market": "spot", "asset": "BTC", "qty": qty_btc},
            {"action": "sell", "market": "futures", "symbol": best["symbol"], "qty": qty_btc}
        ]

        resp = requests.post(
            "https://purpleflea.com/trading-api/batch-execute",
            json={"orders": orders},
            headers={"X-API-Key": self.api_key}
        )
        result = resp.json()

        self.active_trades.append({"trade": best, "qty": qty_btc, "notional": notional})
        log.info(f"Basis trade entered: {qty_btc:.4f} BTC, notional=${notional:,.0f}")
        return result

    def get_available_capital(self) -> float:
        resp = requests.get(
            "https://purpleflea.com/wallet-api/balance",
            headers={"X-API-Key": self.api_key}
        )
        return resp.json().get("usdc_balance", 0.0)

import requests
import asyncio
import logging
from datetime import datetime

log = logging.getLogger("funding_agent")

EIGHT_HOURS = 8 * 3600
ANNUAL_PERIODS = 365 * 3  # 3 funding periods per day


def get_funding_rates(api_key: str) -> dict:
    """Fetch current and predicted funding rates for all major perp markets."""
    resp = requests.get(
        "https://purpleflea.com/trading-api/perps/funding-rates",
        headers={"X-API-Key": api_key}
    )
    resp.raise_for_status()
    return resp.json()["funding_rates"]


def annualize_funding(rate_per_8h: float) -> float:
    return rate_per_8h * ANNUAL_PERIODS * 100


class FundingRateHarvester:
    def __init__(self, api_key: str,
                 min_annual_rate: float = 15.0,
                 max_notional: float = 30_000):
        self.api_key = api_key
        self.min_annual_rate = min_annual_rate
        self.max_notional = max_notional
        self.positions = {}

    def find_best_opportunity(self) -> dict:
        """Find the perp market with highest absolute funding rate."""
        rates = get_funding_rates(self.api_key)
        best = None
        best_annualized = 0.0

        for market, rate_data in rates.items():
            rate_8h = rate_data["current_rate"]
            annualized = annualize_funding(abs(rate_8h))

            if annualized > best_annualized:
                best_annualized = annualized
                best = {
                    "market": market,
                    "rate_8h": rate_8h,
                    "annualized_pct": annualized,
                    "direction": "short_perp" if rate_8h > 0 else "long_perp",
                    "next_funding": rate_data.get("next_funding_time")
                }

        return best

    def enter_harvest_position(self) -> dict:
        """
        Enter delta-neutral funding harvest:
        - If rate > 0: short perp + buy spot (longs pay us)
        - If rate < 0: long perp + short spot (shorts pay us)
        """
        opp = self.find_best_opportunity()

        if not opp or opp["annualized_pct"] < self.min_annual_rate:
            return {"status": "no_opportunity"}

        market = opp["market"]
        direction = opp["direction"]

        # Get spot price for position sizing
        spot = self._get_spot_price(market)
        qty = self.max_notional / spot

        if direction == "short_perp":
            orders = [
                {"action": "buy", "market": "spot", "asset": market, "qty": qty},
                {"action": "sell", "market": "perp", "symbol": f"{market}-PERP", "qty": qty}
            ]
        else:
            orders = [
                {"action": "sell", "market": "spot", "asset": market, "qty": qty},
                {"action": "buy", "market": "perp", "symbol": f"{market}-PERP", "qty": qty}
            ]

        resp = requests.post(
            "https://purpleflea.com/trading-api/batch-execute",
            json={"orders": orders},
            headers={"X-API-Key": self.api_key}
        )
        self.positions[market] = opp
        log.info(f"[FUNDING HARVEST] {market} | {direction} | {opp['annualized_pct']:.1f}% APY")
        return resp.json()

    def should_exit(self, market: str) -> bool:
        """Exit if funding rate drops below exit threshold."""
        rates = get_funding_rates(self.api_key)
        current_rate = rates.get(market, {}).get("current_rate", 0)
        current_annual = annualize_funding(abs(current_rate))
        return current_annual < (self.min_annual_rate * 0.5)  # exit at half entry threshold

    def _get_spot_price(self, asset: str) -> float:
        resp = requests.get(
            f"https://purpleflea.com/trading-api/market-data/{asset.lower()}/spot",
            headers={"X-API-Key": self.api_key}
        )
        return resp.json()["price"]


async def funding_harvest_loop(api_key: str):
    agent = FundingRateHarvester(api_key, min_annual_rate=15.0)
    while True:
        # Scan for new opportunities every 30 minutes
        result = agent.enter_harvest_position()
        log.info(f"Scan result: {result}")

        # Check existing positions before each funding window
        for market in list(agent.positions.keys()):
            if agent.should_exit(market):
                log.info(f"Exiting {market} funding position — rate dropped")
                # Exit logic here

        await asyncio.sleep(1800)  # 30 min

import numpy as np
from scipy.stats import norm
import requests
import asyncio

def black_scholes_delta(S: float, K: float, T: float,
                         r: float, sigma: float, option_type: str) -> float:
    """
    Black-Scholes delta for call or put.
    S: spot price, K: strike, T: time to expiry (years),
    r: risk-free rate, sigma: implied vol (decimal)
    """
    d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
    if option_type == "call":
        return norm.cdf(d1)
    elif option_type == "put":
        return norm.cdf(d1) - 1.0
    raise ValueError(f"Unknown option_type: {option_type}")


def black_scholes_gamma(S, K, T, r, sigma) -> float:
    """Gamma: rate of change of delta per unit spot move."""
    d1 = (np.log(S/K) + (r + 0.5*sigma**2)*T) / (sigma*np.sqrt(T))
    return norm.pdf(d1) / (S * sigma * np.sqrt(T))


class DeltaHedgeAgent:
    """
    Maintains delta-neutral book by auto-hedging with perp contracts.
    Suitable for: market-making, vol trading, funding harvest overlay.
    """

    def __init__(self, api_key: str, hedge_threshold: float = 0.05):
        self.api_key = api_key
        self.hedge_threshold = hedge_threshold  # rebalance if |delta| > threshold
        self.option_book = []  # list of option positions
        self.perp_hedge = 0.0  # current perp position (negative = short)

    def add_option(self, strike: float, expiry_days: float,
                   option_type: str, qty: float, position: str = "long"):
        sign = 1 if position == "long" else -1
        self.option_book.append({
            "strike": strike, "T": expiry_days/365,
            "type": option_type, "qty": qty * sign
        })

    def compute_portfolio_delta(self, spot: float, iv: float) -> float:
        """Sum of deltas across all option positions + perp hedge."""
        total_delta = 0.0
        for opt in self.option_book:
            d = black_scholes_delta(spot, opt["strike"], opt["T"],
                                    r=0.05, sigma=iv/100, option_type=opt["type"])
            total_delta += d * opt["qty"]
        total_delta += self.perp_hedge  # perp contributes delta 1:1
        return total_delta

    def rebalance(self, spot: float, iv: float) -> dict:
        """Compute required hedge adjustment and execute."""
        current_delta = self.compute_portfolio_delta(spot, iv)

        if abs(current_delta) < self.hedge_threshold:
            return {"status": "balanced", "delta": round(current_delta, 4)}

        # To neutralize delta: trade -delta units of the perp
        adjustment = -current_delta
        new_perp_position = self.perp_hedge + adjustment

        action = "buy" if adjustment > 0 else "sell"
        qty = abs(adjustment)

        resp = requests.post(
            "https://purpleflea.com/trading-api/execute",
            json={
                "action": action,
                "market": "perp",
                "symbol": "BTC-PERP",
                "qty": round(qty, 6),
                "reason": "delta_hedge"
            },
            headers={"X-API-Key": self.api_key}
        )

        self.perp_hedge = new_perp_position
        return {
            "status": "rebalanced",
            "prev_delta": round(current_delta, 4),
            "adjustment": round(adjustment, 4),
            "new_perp": round(new_perp_position, 4),
            "order": resp.json()
        }

import asyncio
import logging
from dataclasses import dataclass
from enum import Enum

log = logging.getLogger("derivatives_agent")

class Strategy(Enum):
    BASIS = "basis"
    FUNDING = "funding"
    DELTA_NEUTRAL_VOL = "delta_neutral_vol"

@dataclass
class Opportunity:
    strategy: Strategy
    expected_annual_return: float
    risk_score: float   # 1=low, 10=high
    details: dict

    @property
    def risk_adjusted_return(self) -> float:
        return self.expected_annual_return / self.risk_score


class UnifiedDerivativesAgent:
    def __init__(self, api_key: str, total_capital: float):
        self.api_key = api_key
        self.total_capital = total_capital
        self.basis_agent = BasisTradeAgent(api_key, max_notional=total_capital * 0.4)
        self.funding_agent = FundingRateHarvester(api_key, max_notional=total_capital * 0.4)
        self.delta_agent = DeltaHedgeAgent(api_key)

    async def evaluate_opportunities(self) -> list:
        """Score all available opportunities and return sorted list."""
        opportunities = []

        # Check basis trade
        try:
            data = fetch_market_data(self.api_key)
            basis_opps = compute_basis_opportunities(data)
            if basis_opps:
                best_basis = basis_opps[0]
                opportunities.append(Opportunity(
                    strategy=Strategy.BASIS,
                    expected_annual_return=best_basis["annualized_yield"],
                    risk_score=2,  # low risk: market neutral
                    details=best_basis
                ))
        except Exception as e:
            log.error(f"Basis scan error: {e}")

        # Check funding rate
        try:
            funding_opp = self.funding_agent.find_best_opportunity()
            if funding_opp:
                opportunities.append(Opportunity(
                    strategy=Strategy.FUNDING,
                    expected_annual_return=funding_opp["annualized_pct"],
                    risk_score=3,
                    details=funding_opp
                ))
        except Exception as e:
            log.error(f"Funding scan error: {e}")

        return sorted(opportunities, key=lambda x: x.risk_adjusted_return, reverse=True)

    async def run(self):
        log.info(f"Unified Derivatives Agent started | Capital: ${self.total_capital:,.0f}")
        while True:
            opps = await self.evaluate_opportunities()
            for opp in opps[:2]:  # top 2 strategies
                log.info(f"Strategy: {opp.strategy.value} | RAR: {opp.risk_adjusted_return:.2f}")
            await asyncio.sleep(1800)