FX and Stablecoin Arbitrage for AI Agents

def compute_breakeven_spread(
    bridge_fee_bps: float,
    source_gas_usd: float,
    dest_gas_usd: float,
    dex_fee_bps: float,
    amount_usd: float,
    price_risk_bps: float = 5,  # spread movement risk during bridge time
) -> float:
    """
    Returns the minimum observed spread (in bps) required for
    a cross-chain stablecoin arb to be profitable after all costs.
    """
    # Convert fixed costs to basis points
    gas_bps = (source_gas_usd + dest_gas_usd) / amount_usd * 10000

    # Total cost in bps
    total_cost_bps = bridge_fee_bps + gas_bps + dex_fee_bps + price_risk_bps

    return total_cost_bps


# Example: USDC arb Ethereum → Arbitrum via Across
breakeven = compute_breakeven_spread(
    bridge_fee_bps=6,       # Across fee
    source_gas_usd=4.50,   # Ethereum send gas
    dest_gas_usd=0.05,     # Arbitrum receive gas
    dex_fee_bps=1,         # Curve 3pool fee
    amount_usd=50000,      # trade size
    price_risk_bps=4,      # 2-min bridge, stablecoin vol
)
print(f"Breakeven spread: {breakeven:.1f} bps")
# Output: Breakeven spread: 12.0 bps
# Minimum observable spread to trade: ~12 bps

# Scale sensitivity: how much does size matter?
for size in [10_000, 50_000, 200_000, 1_000_000]:
    be = compute_breakeven_spread(6, 4.50, 0.05, 1, size, 4)
    print(f"  ${size:>10,} → breakeven {be:.1f} bps")

# Output:
#   $    10,000 → breakeven 57.5 bps
#   $    50,000 → breakeven 12.0 bps
#   $   200,000 → breakeven  8.3 bps
#   $ 1,000,000 → breakeven  7.2 bps

import asyncio
import httpx
import logging
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
from enum import Enum
import time

logger = logging.getLogger("FXArbitrageAgent")

# Purple Flea infrastructure endpoints
PF_TRADING_URL = "https://trading.purpleflea.com/api/v1"
PF_WALLET_URL = "https://wallet.purpleflea.com/api/v1"

# Stablecoin price feeds (Chainlink aggregators via RPC or price API)
PRICE_FEEDS = {
    "USDC": "https://api.coingecko.com/api/v3/simple/price?ids=usd-coin&vs_currencies=usd",
    "USDT": "https://api.coingecko.com/api/v3/simple/price?ids=tether&vs_currencies=usd",
    "DAI":  "https://api.coingecko.com/api/v3/simple/price?ids=dai&vs_currencies=usd",
    "FRAX": "https://api.coingecko.com/api/v3/simple/price?ids=frax&vs_currencies=usd",
}

COINGECKO_ID_MAP = {
    "USDC": "usd-coin",
    "USDT": "tether",
    "DAI":  "dai",
    "FRAX": "frax",
}


class ArbType(Enum):
    SPATIAL = "spatial"          # same chain, different venues
    TRIANGULAR = "triangular"    # A→B→C→A
    CROSS_CHAIN = "cross_chain"  # same coin, different chains
    DEPEG = "depeg"              # stablecoin trading at discount


@dataclass
class StablecoinPrice:
    symbol: str
    price: float
    venue: str
    chain: str
    timestamp: float = field(default_factory=time.time)

    @property
    def depeg_bps(self) -> float:
        """Deviation from $1.00 in basis points."""
        return (1.0 - self.price) * 10000


@dataclass
class ArbOpportunity:
    arb_type: ArbType
    buy_venue: str
    sell_venue: str
    buy_price: float
    sell_price: float
    spread_bps: float
    input_token: str
    output_token: str
    estimated_profit_bps: float
    max_size_usd: float
    confidence: float  # 0-1

    @property
    def is_viable(self) -> bool:
        return self.estimated_profit_bps > 5 and self.confidence > 0.7


@dataclass
class AgentConfig:
    api_key: str
    wallet_id: str
    max_position_usd: float = 25_000
    min_profit_bps: float = 5      # 0.05% minimum net profit
    depeg_threshold_bps: float = 50 # alert on 0.5% depeg
    scan_interval_sec: float = 15
    max_daily_trades: int = 500
    chains: List[str] = field(default_factory=lambda: ["ethereum", "arbitrum", "optimism"])
    stablecoins: List[str] = field(default_factory=lambda: ["USDC", "USDT", "DAI", "FRAX"])


class FXArbitrageAgent:
    """
    Autonomous stablecoin FX arbitrage agent for AI-native financial infrastructure.

    Monitors USDC/USDT/DAI/FRAX prices across multiple venues and chains,
    detects spatial, triangular, and cross-chain arbitrage opportunities,
    and executes via Purple Flea's Trading API with automated risk controls.

    Register and fund via https://faucet.purpleflea.com (free $1 USDC to start).
    """

    def __init__(self, config: AgentConfig):
        self.config = config
        self._client: Optional[httpx.AsyncClient] = None
        self._price_cache: Dict[str, StablecoinPrice] = {}
        self._trade_count_today: int = 0
        self._pnl_today_usd: float = 0.0
        self._last_reset: datetime = datetime.utcnow()
        self._running: bool = False

    async def _client_get(self) -> httpx.AsyncClient:
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=8.0)
        return self._client

    async def _fetch_prices(self) -> Dict[str, float]:
        """Fetch current stablecoin prices from CoinGecko."""
        client = await self._client_get()
        ids = ",".join(COINGECKO_ID_MAP.values())
        url = f"https://api.coingecko.com/api/v3/simple/price?ids={ids}&vs_currencies=usd"
        try:
            resp = await client.get(url)
            resp.raise_for_status()
            data = resp.json()
            prices = {}
            for symbol, cg_id in COINGECKO_ID_MAP.items():
                if cg_id in data:
                    prices[symbol] = data[cg_id]["usd"]
            return prices
        except Exception as e:
            logger.warning(f"Price fetch failed: {e}")
            return {}

    def _compute_fee_bps(self, path_hops: int) -> float:
        """Estimate all-in fee for a trade path."""
        curve_fee_bps = 4  # 0.04% Curve 3pool fee
        gas_bps = 8         # ~8 bps gas at typical sizes
        return (curve_fee_bps + gas_bps) * path_hops

    def _detect_spatial_arb(self, prices: Dict[str, float]) -> List[ArbOpportunity]:
        """Find pairs trading at spreads exceeding cost of swap."""
        opportunities = []
        symbols = list(prices.keys())
        for i, s1 in enumerate(symbols):
            for s2 in symbols[i+1:]:
                p1, p2 = prices[s1], prices[s2]
                spread_bps = abs(p1 - p2) * 10000
                fee_bps = self._compute_fee_bps(1)
                net_profit_bps = spread_bps - fee_bps

                if net_profit_bps >= self.config.min_profit_bps:
                    buy_sym = s1 if p1 < p2 else s2
                    sell_sym = s2 if p1 < p2 else s1
                    buy_price = min(p1, p2)
                    sell_price = max(p1, p2)

                    # Confidence based on spread magnitude and data freshness
                    confidence = min(1.0, net_profit_bps / 20)

                    opportunities.append(ArbOpportunity(
                        arb_type=ArbType.SPATIAL,
                        buy_venue="Curve",
                        sell_venue="Curve",
                        buy_price=buy_price,
                        sell_price=sell_price,
                        spread_bps=spread_bps,
                        input_token=buy_sym,
                        output_token=sell_sym,
                        estimated_profit_bps=net_profit_bps,
                        max_size_usd=min(self.config.max_position_usd, 100_000),
                        confidence=confidence,
                    ))
        return opportunities

    def _detect_depegs(self, prices: Dict[str, float]) -> List[ArbOpportunity]:
        """Detect stablecoins trading significantly below $1."""
        opportunities = []
        for symbol, price in prices.items():
            depeg_bps = (1.0 - price) * 10000
            if depeg_bps > self.config.depeg_threshold_bps:
                logger.warning(f"DEPEG ALERT: {symbol} = ${price:.6f} ({depeg_bps:.0f} bps off peg)")
                # Arb: buy cheap stablecoin, redeem or sell once peg recovers
                net_profit_bps = depeg_bps - self._compute_fee_bps(1) - 20  # extra safety margin
                if net_profit_bps > 0:
                    opportunities.append(ArbOpportunity(
                        arb_type=ArbType.DEPEG,
                        buy_venue="Curve",
                        sell_venue="Redemption",
                        buy_price=price,
                        sell_price=1.0,
                        spread_bps=depeg_bps,
                        input_token="USDC",
                        output_token=symbol,
                        estimated_profit_bps=net_profit_bps,
                        max_size_usd=min(self.config.max_position_usd * 0.2, 5_000),
                        confidence=0.6 if depeg_bps > 200 else 0.8,
                    ))
        return opportunities

    async def _execute_arb(self, opp: ArbOpportunity) -> bool:
        """Submit arb trade via Purple Flea Trading API."""
        client = await self._client_get()
        size = min(opp.max_size_usd, self.config.max_position_usd)

        logger.info(
            f"Executing {opp.arb_type.value} arb: "
            f"{opp.input_token}→{opp.output_token} at {opp.buy_venue}, "
            f"est. profit={opp.estimated_profit_bps:.1f}bps, size=${size:,.0f}"
        )

        try:
            resp = await client.post(
                f"{PF_TRADING_URL}/swap",
                headers={"X-API-Key": self.config.api_key},
                json={
                    "from_token": opp.input_token,
                    "to_token": opp.output_token,
                    "amount_usd": size,
                    "wallet_id": self.config.wallet_id,
                    "slippage_bps": int(opp.spread_bps * 0.5),
                    "mev_protection": "flashbots",
                    "label": f"fx_arb_{opp.arb_type.value}",
                },
            )
            resp.raise_for_status()
            result = resp.json()
            logger.info(f"Trade executed: tx={result.get('tx_hash', 'pending')}")
            self._trade_count_today += 1
            actual_profit = result.get("realized_pnl_usd", 0)
            self._pnl_today_usd += actual_profit
            return True
        except Exception as e:
            logger.error(f"Trade execution failed: {e}")
            return False

    def _reset_daily_stats_if_needed(self) -> None:
        now = datetime.utcnow()
        if (now - self._last_reset).days >= 1:
            logger.info(f"Daily stats: trades={self._trade_count_today}, PnL=${self._pnl_today_usd:.2f}")
            self._trade_count_today = 0
            self._pnl_today_usd = 0
            self._last_reset = now

    async def _scan_cycle(self) -> None:
        """One full scan-detect-execute cycle."""
        self._reset_daily_stats_if_needed()

        if self._trade_count_today >= self.config.max_daily_trades:
            logger.warning("Daily trade limit reached, pausing.")
            return

        prices = await self._fetch_prices()
        if not prices:
            return

        # Detect all opportunity types
        opps = self._detect_spatial_arb(prices)
        opps += self._detect_depegs(prices)

        # Filter viable opportunities, sort by expected profit
        viable = [o for o in opps if o.is_viable]
        viable.sort(key=lambda o: o.estimated_profit_bps, reverse=True)

        if viable:
            logger.info(f"Found {len(viable)} viable arb opportunities")
            # Execute top opportunity only (conservative: one trade per cycle)
            await self._execute_arb(viable[0])
        else:
            logger.debug(f"No viable arb. Prices: {prices}")

    async def run(self) -> None:
        """Main agent loop. Runs indefinitely until stopped."""
        logger.info("FXArbitrageAgent starting. Powered by purpleflea.com")
        self._running = True
        while self._running:
            try:
                await self._scan_cycle()
            except Exception as e:
                logger.error(f"Scan cycle error: {e}")
            await asyncio.sleep(self.config.scan_interval_sec)

    async def stop(self) -> None:
        logger.info(f"Agent stopped. Daily PnL: ${self._pnl_today_usd:.2f}")
        self._running = False
        if self._client:
            await self._client.aclose()


# Entry point
async def main():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )
    config = AgentConfig(
        api_key="YOUR_PURPLE_FLEA_API_KEY",
        wallet_id="YOUR_WALLET_ID",
        max_position_usd=10_000,
        min_profit_bps=8,
        scan_interval_sec=15,
    )
    agent = FXArbitrageAgent(config)
    try:
        await agent.run()
    except KeyboardInterrupt:
        await agent.stop()

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