funding rate harvest: neutral crypto carry when momentum breaks

import asyncio
import ccxt.pro as ccxtpro
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timezone
import logging

logger = logging.getLogger(__name__)


@dataclass
class FundingRate:
    exchange: str
    symbol: str
    rate: float            # 8-hour rate as decimal
    next_funding_time: int # Unix ms
    timestamp: int

    @property
    def annualized(self) -> float:
        """3 funding periods/day × 365"""
        return self.rate * 3 * 365

    @property
    def hours_to_next(self) -> float:
        now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
        return max((self.next_funding_time - now_ms) / (1000 * 3600), 0.0)


@dataclass
class HarvestSignal:
    symbol: str
    best_exchange: str
    annualized_rate: float
    hedge_cost_annual: float   # estimated round-trip taker fee × annual frequency
    net_yield: float
    viable: bool
    size_usd: float
    close_trigger: bool        # True if rate fell below minimum, close position


class FundingRateMonitor:
    """
    Multi-exchange funding rate monitor for delta-neutral crypto carry.

    Strategy:
      - Long spot BTC (or via low-fee CEX/OTC)
      - Short equal-sized BTC perpetual on exchange with highest rate
      - Net position: delta-neutral, collects 8h funding payments
      - Exit: rate drops below threshold OR approaches funding payment time
        while rate is negative

    Risk considerations:
      - Margin calls on short perp leg during extreme BTC moves
      - Counterparty/exchange risk (Binance, Kraken)
      - Slippage on entry/exit
      - Spot/perp basis risk on close
    """

    MIN_YIELD_THRESHOLD = 0.08    # 8% annualized minimum
    CLOSE_THRESHOLD = 0.03        # close if rate drops below 3% annualized
    MAX_POSITION_PCT = 0.12       # 12% of crypto allocation per symbol
    TAKER_FEE_BPS = 4             # ~4bps per side (aggressive but realistic)
    ANNUAL_TURNS = 12             # assume ~12 entries/exits per year
    CHECK_INTERVAL_SEC = 300      # check rates every 5 minutes

    def __init__(self, crypto_allocation_usd: float):
        self.allocation = crypto_allocation_usd
        self.exchanges: Dict[str, ccxtpro.Exchange] = {}
        self.active_positions: Dict[str, HarvestSignal] = {}
        self._rate_history: List[FundingRate] = []

    async def initialize(self):
        """Connect to exchanges"""
        try:
            self.exchanges['binance'] = ccxtpro.binance({
                'apiKey': self._cred('BINANCE_API_KEY'),
                'secret': self._cred('BINANCE_API_SECRET'),
                'options': {'defaultType': 'future'},
                'enableRateLimit': True
            })
            self.exchanges['kraken'] = ccxtpro.kraken({
                'apiKey': self._cred('KRAKEN_API_KEY'),
                'secret': self._cred('KRAKEN_API_SECRET'),
                'enableRateLimit': True
            })
            await asyncio.gather(
                self.exchanges['binance'].load_markets(),
                self.exchanges['kraken'].load_markets()
            )
            logger.info("FundingRateMonitor: exchanges initialized")
        except Exception as e:
            logger.critical(f"Exchange init failed: {e}")
            raise

    async def fetch_rate(self,
                          exchange: str,
                          symbol: str = "BTC/USDT:USDT") -> Optional[FundingRate]:
        """Fetch single funding rate snapshot"""
        try:
            exch = self.exchanges[exchange]
            info = await exch.fetch_funding_rate(symbol)
            return FundingRate(
                exchange=exchange,
                symbol=symbol,
                rate=float(info.get('fundingRate') or 0),
                next_funding_time=int(info.get('fundingTimestamp') or 0),
                timestamp=int(datetime.now(timezone.utc).timestamp() * 1000)
            )
        except Exception as e:
            logger.warning(f"fetch_rate failed [{exchange} {symbol}]: {e}")
            return None

    async def evaluate(self, symbol: str = "BTC") -> Optional[HarvestSignal]:
        """Evaluate harvest opportunity across all exchanges"""
        perp_sym = f"{symbol}/USDT:USDT"
        rates = await asyncio.gather(*[
            self.fetch_rate(ex, perp_sym)
            for ex in self.exchanges
        ])

        valid = [r for r in rates if r is not None]
        if not valid:
            return None

        self._rate_history.extend(valid)

        best = max(valid, key=lambda r: r.rate)

        # Hedge cost: TAKER_FEE_BPS × 2 sides × ANNUAL_TURNS / 10000
        hedge_cost = (self.TAKER_FEE_BPS / 10000) * 2 * self.ANNUAL_TURNS
        net_yield = best.annualized - hedge_cost

        # Dynamic size: scales with yield quality, capped at max pct
        raw_size = self.allocation * (net_yield / self.MIN_YIELD_THRESHOLD) * 0.04
        size = min(raw_size, self.allocation * self.MAX_POSITION_PCT)
        size = max(size, 0.0)

        # Close trigger: position open AND rate below close threshold
        is_open = symbol in self.active_positions
        close_now = is_open and best.annualized < self.CLOSE_THRESHOLD

        return HarvestSignal(
            symbol=symbol,
            best_exchange=best.exchange,
            annualized_rate=best.annualized,
            hedge_cost_annual=hedge_cost,
            net_yield=net_yield,
            viable=net_yield >= self.MIN_YIELD_THRESHOLD,
            size_usd=size,
            close_trigger=close_now
        )

    async def run(self):
        """Main monitoring loop"""
        logger.info("Starting funding harvest monitor")
        while True:
            try:
                sig = await self.evaluate("BTC")
                if sig:
                    self._log_signal(sig)
                    if sig.close_trigger:
                        await self._emit_close(sig)
                    elif sig.viable and "BTC" not in self.active_positions:
                        await self._emit_open(sig)
                await asyncio.sleep(self.CHECK_INTERVAL_SEC)
            except Exception as e:
                logger.error(f"Monitor loop error: {e}")
                await asyncio.sleep(60)

    async def _emit_open(self, sig: HarvestSignal):
        """Emit open signal to execution layer via Redis"""
        payload = {
            'strategy': 'funding_harvest',
            'action': 'open',
            'symbol': sig.symbol,
            'exchange': sig.best_exchange,
            'size_usd': sig.size_usd,
            'annualized_yield': sig.net_yield
        }
        # would push to Redis queue here
        logger.info(f"OPEN signal: {payload}")
        self.active_positions[sig.symbol] = sig

    async def _emit_close(self, sig: HarvestSignal):
        """Emit close signal — rate below threshold"""
        logger.warning(
            f"CLOSE trigger: {sig.symbol} rate={sig.annualized_rate*100:.1f}% "
            f"below threshold={self.CLOSE_THRESHOLD*100:.1f}%"
        )
        # push close to Redis
        self.active_positions.pop(sig.symbol, None)

    def _log_signal(self, sig: HarvestSignal):
        logger.info(
            f"BTC funding: {sig.best_exchange}={sig.annualized_rate*100:.1f}% ann | "
            f"net={sig.net_yield*100:.1f}% | viable={sig.viable} | "
            f"size=${sig.size_usd:,.0f}"
        )

    @staticmethod
    def _cred(key: str) -> str:
        import os
        val = os.environ.get(key)
        if not val:
            raise ValueError(f"Missing env var: {key}")
        return val