Funding Rate Arbitrage: How AI Agents Harvest Perpetual Funding

import asyncio
import aiohttp
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import logging

logger = logging.getLogger("FundingHarvester")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

PF_BASE = "https://api.purpleflea.com"
AGENT_KEY = "your-agent-api-key"

# Harvest thresholds
MIN_FUNDING_TO_ENTER = 0.025   # 0.025% per 8h minimum to enter
EXIT_FUNDING_THRESHOLD = 0.01   # Exit when funding drops below 0.01%
MIN_HOLD_PERIODS = 3            # Hold at least 3 funding periods (~24h)
MAX_POSITION_USD = 2000.0       # Max notional per symbol
MAX_SYMBOLS = 5                 # Max concurrent harvest positions
TAKER_FEE_PCT = 0.04            # 0.04% taker fee


@dataclass
class FundingOpportunity:
    symbol: str
    exchange: str
    funding_rate_8h: float    # positive = longs pay shorts
    mark_price: float
    spot_price: float
    premium_pct: float        # (mark - spot) / spot * 100
    annualized_yield: float   # funding_rate_8h * 1095 * 100

    @property
    def is_harvestable(self) -> bool:
        return abs(self.funding_rate_8h) >= MIN_FUNDING_TO_ENTER

    @property
    def harvest_direction(self) -> str:
        """Which side of the perp to hold to collect funding"""
        return "short" if self.funding_rate_8h > 0 else "long"

    def net_yield_after_fees(self, hold_periods: int = 10) -> float:
        """Estimated net yield after round-trip transaction costs"""
        gross = abs(self.funding_rate_8h) * hold_periods
        costs = (TAKER_FEE_PCT / 100) * 4  # 4 orders: open/close each leg
        return gross - costs


@dataclass
class HarvestPosition:
    symbol: str
    exchange: str
    perp_side: str            # 'long' or 'short' perp
    perp_size: float          # units
    spot_size: float          # units (hedge)
    entry_funding: float      # funding rate at entry
    entry_perp_price: float
    entry_spot_price: float
    entry_time: datetime = field(default_factory=datetime.utcnow)
    periods_held: int = 0
    total_funding_collected: float = 0.0


class FundingScanner:
    """Scans exchanges for funding rate opportunities"""

    def __init__(self, session: aiohttp.ClientSession, pf_headers: dict):
        self.session = session
        self.pf_headers = pf_headers

    async def scan_purple_flea(self, symbols: List[str]) -> List[FundingOpportunity]:
        """Fetch funding rates from Purple Flea trading API"""
        opportunities = []
        for symbol in symbols:
            try:
                async with self.session.get(
                    f"{PF_BASE}/trading/markets/{symbol}",
                    headers=self.pf_headers
                ) as r:
                    d = await r.json()
                funding = d.get("funding_rate", 0.0)
                mark = d.get("mark_price", 0.0)
                index = d.get("index_price", mark)
                premium = (mark - index) / index * 100 if index > 0 else 0.0
                opp = FundingOpportunity(
                    symbol=symbol,
                    exchange="purpleflea",
                    funding_rate_8h=funding,
                    mark_price=mark,
                    spot_price=index,
                    premium_pct=premium,
                    annualized_yield=abs(funding) * 1095 * 100
                )
                opportunities.append(opp)
            except Exception as e:
                logger.warning(f"Error scanning {symbol}: {e}")
        return opportunities

    async def scan_public_exchange(self, exchange: str, symbol: str) -> Optional[FundingOpportunity]:
        """Fetch funding rate from external exchange public API"""
        endpoints = {
            "binance": f"https://fapi.binance.com/fapi/v1/premiumIndex?symbol={symbol}",
            "bybit": f"https://api.bybit.com/v5/market/tickers?category=linear&symbol={symbol}",
        }
        url = endpoints.get(exchange)
        if not url:
            return None
        try:
            async with self.session.get(url) as r:
                data = await r.json()
            # Parse exchange-specific response format
            if exchange == "binance":
                funding = float(data.get("lastFundingRate", 0))
                mark = float(data.get("markPrice", 0))
                index = float(data.get("indexPrice", mark))
            elif exchange == "bybit":
                ticker = data.get("result", {}).get("list", [{}])[0]
                funding = float(ticker.get("fundingRate", 0))
                mark = float(ticker.get("markPrice", 0))
                index = float(ticker.get("indexPrice", mark))
            else:
                return None
            premium = (mark - index) / index * 100 if index > 0 else 0.0
            return FundingOpportunity(
                symbol=symbol, exchange=exchange,
                funding_rate_8h=funding,
                mark_price=mark, spot_price=index,
                premium_pct=premium,
                annualized_yield=abs(funding) * 1095 * 100
            )
        except Exception as e:
            logger.warning(f"Error scanning {exchange}/{symbol}: {e}")
            return None

    async def find_best_opportunities(self, symbols: List[str]) -> List[FundingOpportunity]:
        """Aggregate opportunities from all exchanges and rank by yield"""
        all_opps = []
        pf_opps = await self.scan_purple_flea(symbols)
        all_opps.extend(pf_opps)
        # Scan external exchanges for comparison
        binance_symbols = [s.replace("-PERP", "USDT") for s in symbols]
        external_tasks = []
        for pf_sym, bn_sym in zip(symbols, binance_symbols):
            external_tasks.append(self.scan_public_exchange("binance", bn_sym))
            external_tasks.append(self.scan_public_exchange("bybit", bn_sym))
        results = await asyncio.gather(*external_tasks, return_exceptions=True)
        for r in results:
            if isinstance(r, FundingOpportunity):
                all_opps.append(r)
        harvestable = [o for o in all_opps if o.is_harvestable]
        harvestable.sort(key=lambda o: abs(o.funding_rate_8h), reverse=True)
        return harvestable[:MAX_SYMBOLS]


class FundingHarvester:
    """Delta-neutral funding rate harvester"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
        self.positions: Dict[str, HarvestPosition] = {}
        self._session: Optional[aiohttp.ClientSession] = None
        self.scanner: Optional[FundingScanner] = None
        self._funding_settle_times = [0, 8, 16]  # UTC hours of settlement

    async def __aenter__(self):
        self._session = aiohttp.ClientSession()
        self.scanner = FundingScanner(self._session, self.headers)
        return self

    async def __aexit__(self, *args):
        if self._session:
            await self._session.close()

    async def open_harvest_position(self, opp: FundingOpportunity) -> HarvestPosition:
        """Open delta-neutral position: perp + spot hedge"""
        size_usd = min(MAX_POSITION_USD, 2000)
        perp_size = size_usd / opp.mark_price
        spot_size = perp_size  # 1:1 hedge ratio
        perp_side = opp.harvest_direction  # short if funding > 0
        logger.info(f"Opening harvest: {opp.symbol} {perp_side} perp, funding={opp.funding_rate_8h*100:.3f}% 8h")
        # Open perpetual leg
        perp_payload = {"symbol": opp.symbol, "side": perp_side, "size": perp_size, "type": "market"}
        async with self._session.post(f"{PF_BASE}/trading/orders", json=perp_payload, headers=self.headers) as r:
            perp_order = await r.json()
        # Open spot hedge via Wallet API (hold or buy spot)
        # For positive funding (short perp): hold spot as natural hedge
        # For negative funding (long perp): borrow + short spot (advanced)
        pos = HarvestPosition(
            symbol=opp.symbol, exchange=opp.exchange,
            perp_side=perp_side, perp_size=perp_size, spot_size=spot_size,
            entry_funding=opp.funding_rate_8h,
            entry_perp_price=opp.mark_price, entry_spot_price=opp.spot_price
        )
        self.positions[opp.symbol] = pos
        logger.info(f"Harvest opened: {opp.symbol} | Expected yield: {opp.net_yield_after_fees():.3f}%")
        return pos

    async def close_harvest_position(self, symbol: str, reason: str):
        pos = self.positions.get(symbol)
        if not pos:
            return
        close_side = "long" if pos.perp_side == "short" else "short"
        payload = {"symbol": symbol, "side": close_side, "size": pos.perp_size, "type": "market", "reduce_only": True}
        async with self._session.post(f"{PF_BASE}/trading/orders", json=payload, headers=self.headers) as r:
            await r.json()
        logger.info(f"Closed harvest: {symbol} | Reason: {reason} | Periods: {pos.periods_held} | Collected: ${pos.total_funding_collected:.2f}")
        del self.positions[symbol]

    async def on_funding_settled(self, current_rates: Dict[str, float]):
        """Called after each 8h funding settlement"""
        for symbol, pos in list(self.positions.items()):
            current_rate = current_rates.get(symbol, 0.0)
            payout = abs(current_rate) * MAX_POSITION_USD
            pos.periods_held += 1
            pos.total_funding_collected += payout
            logger.info(f"Funding settled: {symbol} | Rate={current_rate*100:.3f}% | Payout=${payout:.2f} | Total=${pos.total_funding_collected:.2f}")
            # Check exit conditions
            funding_flipped = (pos.entry_funding > 0) != (current_rate > 0)
            rate_too_low = abs(current_rate) < EXIT_FUNDING_THRESHOLD
            held_long_enough = pos.periods_held >= MIN_HOLD_PERIODS
            if held_long_enough and (rate_too_low or funding_flipped):
                reason = "funding_reversed" if funding_flipped else "rate_too_low"
                await self.close_harvest_position(symbol, reason)

    async def run(self, watch_symbols: List[str]):
        """Main harvest loop"""
        logger.info(f"FundingHarvester started. Watching: {watch_symbols}")
        while True:
            try:
                # Scan for opportunities
                opportunities = await self.scanner.find_best_opportunities(watch_symbols)
                logger.info(f"Found {len(opportunities)} harvestable opportunities")
                for opp in opportunities:
                    logger.info(f"  {opp.exchange}/{opp.symbol}: {opp.funding_rate_8h*100:.3f}% 8h = {opp.annualized_yield:.1f}% APY")
                # Open new positions (up to MAX_SYMBOLS limit)
                for opp in opportunities:
                    if opp.symbol not in self.positions and len(self.positions) < MAX_SYMBOLS:
                        await self.open_harvest_position(opp)
                # Fetch current rates for settled positions
                current_rates = {}
                for sym in list(self.positions.keys()):
                    async with self._session.get(f"{PF_BASE}/trading/markets/{sym}", headers=self.headers) as r:
                        d = await r.json()
                    current_rates[sym] = d.get("funding_rate", 0.0)
                await self.on_funding_settled(current_rates)
                # Log portfolio state
                total_collected = sum(p.total_funding_collected for p in self.positions.values())
                logger.info(f"Portfolio: {len(self.positions)} positions | Total collected: ${total_collected:.2f}")
            except Exception as e:
                logger.error(f"Harvester loop error: {e}")
            # Wait until next funding period (every 8 hours)
            await asyncio.sleep(28800)  # 8 * 60 * 60 = 28800 seconds


async def main():
    watch_symbols = [
        "BTC-PERP", "ETH-PERP", "SOL-PERP",
        "AVAX-PERP", "BNB-PERP", "ARB-PERP", "OP-PERP"
    ]
    async with FundingHarvester(AGENT_KEY) as harvester:
        await harvester.run(watch_symbols)

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from datetime import datetime, timezone

class FundingMonitor:
    """Lightweight monitoring layer for funding harvest positions"""

    def __init__(self, harvester):
        self.harvester = harvester
        self.basis_alerts = []
        self.settlement_log = []

    async def check_basis(self, symbol: str, perp_price: float, spot_price: float) -> bool:
        """Returns True if basis is within safe range"""
        basis_pct = abs(perp_price - spot_price) / spot_price * 100
        if basis_pct > 2.0:
            msg = f"BASIS ALERT: {symbol} perp/spot divergence = {basis_pct:.2f}%"
            self.basis_alerts.append({"time": datetime.now(timezone.utc).isoformat(), "msg": msg})
            return False
        return True

    async def reconcile_settlement(self, symbol: str, expected_payout: float, actual_payout: float):
        discrepancy = abs(expected_payout - actual_payout)
        discrepancy_pct = discrepancy / max(0.01, expected_payout) * 100
        entry = {
            "symbol": symbol,
            "time": datetime.now(timezone.utc).isoformat(),
            "expected": expected_payout,
            "actual": actual_payout,
            "discrepancy_pct": discrepancy_pct
        }
        self.settlement_log.append(entry)
        if discrepancy_pct > 10:
            import logging
            logging.warning(f"Settlement discrepancy {symbol}: expected=${expected_payout:.4f} actual=${actual_payout:.4f} ({discrepancy_pct:.1f}%)")

    def report(self) -> dict:
        total_collected = sum(p.total_funding_collected for p in self.harvester.positions.values())
        return {
            "active_positions": len(self.harvester.positions),
            "total_collected_usd": total_collected,
            "basis_alerts_24h": len([a for a in self.basis_alerts[-100:]]),
            "settlement_records": len(self.settlement_log)
        }