Basis Trading for AI Agents: Cash-and-Carry Arbitrage

import statistics

def annualised_perp_basis(funding_rates_8h: list[float]) -> float:
    """
    Calculate annualised basis yield from a list of 8-hour funding rates.

    Args:
        funding_rates_8h: List of funding rates (as %, e.g. 0.0300 means 0.03%)
                          Positive = longs pay shorts (contango)

    Returns:
        Annualised yield in percent (e.g. 13.14 means 13.14% APY)
    """
    if not funding_rates_8h:
        return 0.0

    avg_rate = statistics.mean(funding_rates_8h)  # average 8h rate as %
    periods_per_year = 3 * 365                         # 3 periods/day × 365
    return avg_rate * periods_per_year

def annualised_dated_basis(spot: float, futures: float, dte: int) -> float:
    """
    Calculate annualised yield for a dated futures basis trade.

    Args:
        spot:    Spot price of the underlying asset (USD)
        futures: Futures contract price (USD)
        dte:     Days to expiry

    Returns:
        Annualised yield in percent
    """
    if dte <= 0 or spot <= 0:
        return 0.0
    raw_basis_pct = ((futures / spot) - 1.0) * 100
    return raw_basis_pct * (365 / dte)

# Example
historical_rates = [0.0251, 0.0312, 0.0288, 0.0345, 0.0299,
                    0.0261, 0.0318]  # last 7 days of 8h rates
print(f"Perp APY: {annualised_perp_basis(historical_rates):.2f}%")
# → Perp APY: 10.72%

print(f"Dated APY: {annualised_dated_basis(95000, 97850, 90):.2f}%")
# → Dated APY: 12.16%

"""
BasisTradingAgent for Purple Flea
- Scans all perpetual markets for funding rate opportunities
- Compares perpetual funding rate vs spot holding cost
- Enters delta-neutral long spot + short perp when basis exceeds MIN_APY
- Tracks cumulative funding income and exits when basis collapses
"""

import time
import statistics
from dataclasses import dataclass, field
from typing import Optional
import requests

# ── Config ──────────────────────────────────────────────────────────────────
BASE_URL    = "https://api.purpleflea.com"
API_KEY     = "YOUR_PURPLE_FLEA_API_KEY"
HEADERS     = {"Authorization": f"Bearer {API_KEY}"}

MIN_APY         = 8.0     # % annualised — minimum to enter a position
EXIT_APY        = 3.0     # % annualised — exit when rolling 3d avg drops below
MAX_POSITION_USD = 5000   # maximum USD size per position
MAX_LEVERAGE    = 3.0     # maximum leverage on the short perp leg
MARGIN_BUFFER   = 0.35    # keep 35% extra margin above initial requirement
HISTORY_PERIODS = 21      # funding periods (7 days) for rolling average
SCAN_INTERVAL   = 28800   # 8 hours in seconds (one funding period)

# ── Data Structures ─────────────────────────────────────────────────────────
@dataclass
class BasisOpportunity:
    market:          str
    spot_price:      float
    perp_price:      float
    current_funding: float   # 8h rate as %
    avg_funding_7d:  float   # 21-period rolling average as %
    annualised_apy:  float   # based on avg_funding_7d

@dataclass
class Position:
    market:          str
    spot_qty:        float
    perp_qty:        float
    entry_spot:      float
    entry_perp:      float
    entry_basis_pct: float
    total_funding_collected: float = 0.0
    funding_payments:        list  = field(default_factory=list)

# ── API Helpers ──────────────────────────────────────────────────────────────
def get_markets() -> list[dict]:
    """Fetch all active perpetual markets from Purple Flea trading API."""
    resp = requests.get(f"{BASE_URL}/trading/markets", headers=HEADERS, timeout=10)
    resp.raise_for_status()
    return [m for m in resp.json().get("markets", []) if m.get("type") == "perpetual"]

def get_spot_price(asset: str) -> float:
    """Get current spot price for an asset (USD)."""
    resp = requests.get(
        f"{BASE_URL}/trading/price/{asset}",
        headers=HEADERS, timeout=10
    )
    return resp.json()["spot_price"]

def get_funding_history(market: str, periods: int) -> list[float]:
    """
    Fetch the last N 8-hour funding rate snapshots for a market.
    Returns list of rates as percent (e.g. 0.0310 = 0.031%).
    """
    resp = requests.get(
        f"{BASE_URL}/trading/funding/{market}",
        params={"limit": periods},
        headers=HEADERS, timeout=10
    )
    return [r["rate"] for r in resp.json().get("history", [])]

def get_account_balance() -> float:
    """Return available USDC balance in the trading account."""
    resp = requests.get(f"{BASE_URL}/trading/account", headers=HEADERS, timeout=10)
    return resp.json().get("usdc_balance", 0.0)

def place_order(market: str, side: str, size: float, order_type: str = "limit",
               price: Optional[float] = None) -> dict:
    """Submit an order to Purple Flea trading API."""
    payload = {"market": market, "side": side, "size": size, "type": order_type}
    if price is not None:
        payload["price"] = price
    resp = requests.post(
        f"{BASE_URL}/trading/order",
        headers=HEADERS, json=payload, timeout=15
    )
    resp.raise_for_status()
    return resp.json()

def get_spot_wallet(asset: str) -> dict:
    """Get spot wallet balance for an asset."""
    resp = requests.get(f"{BASE_URL}/wallets/{asset.lower()}", headers=HEADERS, timeout=10)
    return resp.json()

# ── Core Logic ───────────────────────────────────────────────────────────────
def scan_basis_opportunities() -> list[BasisOpportunity]:
    """Scan all perpetual markets and return ranked list of basis opportunities."""
    markets = get_markets()
    opportunities = []

    for mkt in markets:
        try:
            asset       = mkt["base_asset"]         # e.g. "BTC"
            perp_name   = mkt["name"]                # e.g. "BTC-PERP"
            perp_price  = float(mkt["mark_price"])
            spot_price  = get_spot_price(asset)

            funding_hist = get_funding_history(perp_name, HISTORY_PERIODS)
            if len(funding_hist) < 3:
                continue  # insufficient history

            current_rate  = funding_hist[0]             # most recent
            avg_rate_7d   = statistics.mean(funding_hist) # 7-day rolling mean
            annualised    = avg_rate_7d * 3 * 365

            if annualised > MIN_APY:
                opportunities.append(BasisOpportunity(
                    market=perp_name,
                    spot_price=spot_price,
                    perp_price=perp_price,
                    current_funding=current_rate,
                    avg_funding_7d=avg_rate_7d,
                    annualised_apy=annualised
                ))
        except (KeyError, ValueError, requests.RequestException) as e:
            print(f"  Warning: skipping {mkt.get('name', '?')} — {e}")
            continue

    # Sort by annualised APY descending
    return sorted(opportunities, key=lambda x: x.annualised_apy, reverse=True)

def calculate_position_size(opp: BasisOpportunity, balance: float) -> float:
    """
    Calculate USD size for the basis position.
    Uses 50% of available balance, capped at MAX_POSITION_USD.
    """
    target = min(balance * 0.50, MAX_POSITION_USD)
    return round(target, 2)

def enter_basis_position(opp: BasisOpportunity, size_usd: float) -> Optional[Position]:
    """
    Execute the two-legged cash-and-carry entry:
    1. Buy spot (asset purchase via wallet API)
    2. Short the perpetual (futures short via trading API)
    Both legs sized to same USD notional.
    """
    asset  = opp.market.split("-")[0]   # "BTC" from "BTC-PERP"
    qty    = round(size_usd / opp.spot_price, 6)

    print(f"  → Entering basis trade on {opp.market}")
    print(f"    APY: {opp.annualised_apy:.2f}% | Size: ${size_usd:.2f} | Qty: {qty}")

    # Leg 1: buy spot at market (or limit slightly above bid)
    spot_order = place_order(
        market=f"{asset}-SPOT", side="buy",
        size=qty, order_type="market"
    )
    actual_spot_price = float(spot_order.get("avg_fill_price", opp.spot_price))

    # Leg 2: short the perpetual at market
    perp_order = place_order(
        market=opp.market, side="sell",
        size=qty, order_type="market"
    )
    actual_perp_price = float(perp_order.get("avg_fill_price", opp.perp_price))

    entry_basis_pct = ((actual_perp_price / actual_spot_price) - 1.0) * 100
    print(f"    Entry basis: {entry_basis_pct:.4f}%")

    return Position(
        market=opp.market,
        spot_qty=qty,
        perp_qty=qty,
        entry_spot=actual_spot_price,
        entry_perp=actual_perp_price,
        entry_basis_pct=entry_basis_pct
    )

def record_funding_payment(position: Position) -> float:
    """
    Read the most recent funding payment received on the short perp leg.
    Returns funding payment in USD for this period.
    """
    resp = requests.get(
        f"{BASE_URL}/trading/funding-payments",
        params={"market": position.market, "limit": 1},
        headers=HEADERS, timeout=10
    )
    payments = resp.json().get("payments", [])
    if payments:
        usd_amount = float(payments[0].get("amount_usd", 0))
        position.total_funding_collected += usd_amount
        position.funding_payments.append(usd_amount)
        print(f"  Funding received: ${usd_amount:.4f} | Total: ${position.total_funding_collected:.4f}")
        return usd_amount
    return 0.0

def should_exit(position: Position) -> bool:
    """
    Exit the position if the 3-period rolling funding average drops below EXIT_APY.
    """
    if len(position.funding_payments) < 3:
        return False  # not enough data yet
    recent  = position.funding_payments[-3:]
    avg_8h  = statistics.mean(recent) / (position.entry_spot * position.spot_qty / 100)
    apy_est = avg_8h * 3 * 365 * 100
    if apy_est < EXIT_APY:
        print(f"  Exit signal: rolling APY {apy_est:.2f}% < {EXIT_APY}% threshold")
        return True
    return False

def exit_basis_position(position: Position):
    """Close both legs of the basis position."""
    asset = position.market.split("-")[0]
    print(f"  → Exiting basis trade on {position.market}")

    # Close spot leg (sell)
    place_order(market=f"{asset}-SPOT", side="sell",
               size=position.spot_qty, order_type="market")

    # Close perp leg (buy back the short)
    place_order(market=position.market, side="buy",
               size=position.perp_qty, order_type="market")

    print(f"  Total funding collected: ${position.total_funding_collected:.4f}")
    print(f"  Funding payments: {position.funding_payments}")

# ── Main Agent Loop ─────────────────────────────────────────────────────────
def run_basis_agent():
    print("=== BasisTradingAgent starting ===")
    active_positions: dict[str, Position] = {}

    while True:
        print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] Scanning basis opportunities...")

        # 1. Record funding for existing positions
        for mkt, pos in list(active_positions.items()):
            record_funding_payment(pos)
            if should_exit(pos):
                exit_basis_position(pos)
                del active_positions[mkt]

        # 2. Find new opportunities
        opportunities = scan_basis_opportunities()
        print(f"  Found {len(opportunities)} markets above {MIN_APY}% APY threshold")
        for o in opportunities[:5]:
            print(f"    {o.market:15s} | APY: {o.annualised_apy:6.2f}% | Funding: {o.current_funding:.4f}%")

        # 3. Enter the best available opportunity not already held
        for opp in opportunities:
            if opp.market not in active_positions:
                balance = get_account_balance()
                if balance < 50:
                    print("  Insufficient balance to enter new position")
                    break
                size  = calculate_position_size(opp, balance)
                pos   = enter_basis_position(opp, size)
                if pos:
                    active_positions[opp.market] = pos
                break  # one new position per cycle

        print(f"  Active positions: {len(active_positions)}")
        print(f"  Sleeping {SCAN_INTERVAL//3600}h until next funding period...")
        time.sleep(SCAN_INTERVAL)

if __name__ == "__main__":
    run_basis_agent()

from dataclasses import dataclass
from typing import List

@dataclass
class Allocation:
    market: str
    weight: float    # fraction of total capital to allocate
    target_usd: float

def allocate_basis_portfolio(
    opportunities: List[BasisOpportunity],
    total_capital: float,
    max_positions: int = 4,
    min_apy: float = 8.0
) -> List[Allocation]:
    """
    Allocate capital across top basis opportunities using
    yield-weighted allocation with a minimum APY floor.

    Higher-APY positions receive proportionally more capital
    but no single market exceeds 40% of total.
    """
    eligible = [o for o in opportunities[:max_positions] if o.annualised_apy >= min_apy]
    if not eligible:
        return []

    total_yield = sum(o.annualised_apy for o in eligible)
    allocations  = []

    for opp in eligible:
        raw_weight   = opp.annualised_apy / total_yield
        capped_weight = min(raw_weight, 0.40)   # cap any single position at 40%
        target_usd   = round(total_capital * capped_weight, 2)
        allocations.append(Allocation(
            market=opp.market,
            weight=capped_weight,
            target_usd=target_usd
        ))

    # Normalize weights to sum to 1.0 after capping
    total_weight = sum(a.weight for a in allocations)
    for a in allocations:
        a.weight     = a.weight / total_weight
        a.target_usd = round(total_capital * a.weight, 2)

    return allocations

# Example output for $10,000 capital across 4 markets:
# LINK-PERP : 28% → $2,800  (APY 22.4%)
# SOL-PERP  : 26% → $2,600  (APY 18.1%)
# ETH-PERP  : 24% → $2,400  (APY 15.9%)
# BTC-PERP  : 22% → $2,200  (APY 11.3%)