Arbitrage Strategies for AI Agents: Finding and Exploiting Price Inefficiencies

# Core arbitrage profit model
from dataclasses import dataclass
from typing import Optional
from decimal import Decimal

@dataclass
class ArbOpportunity:
    buy_venue: str
    sell_venue: str
    asset: str
    buy_price: Decimal
    sell_price: Decimal
    available_size: Decimal
    buy_fee_bps: Decimal
    sell_fee_bps: Decimal
    transfer_cost: Decimal = Decimal("0")
    transfer_time_ms: int = 0

def calculate_arb_pnl(opp: ArbOpportunity, size: Decimal) -> dict:
    """
    Calculate net PnL for an arbitrage opportunity.
    All fees in basis points (1 bps = 0.01%).
    """
    # Gross spread
    gross_spread = opp.sell_price - opp.buy_price
    gross_spread_bps = (gross_spread / opp.buy_price) * 10000

    # Total fee drag
    total_fee_bps = opp.buy_fee_bps + opp.sell_fee_bps
    fee_cost = (total_fee_bps / 10000) * opp.buy_price * size

    # Net profit
    gross_profit = gross_spread * size
    net_profit = gross_profit - fee_cost - opp.transfer_cost
    net_spread_bps = gross_spread_bps - total_fee_bps

    return {
        "gross_spread_bps": float(gross_spread_bps),
        "total_fee_bps": float(total_fee_bps),
        "net_spread_bps": float(net_spread_bps),
        "gross_profit": float(gross_profit),
        "net_profit": float(net_profit),
        "is_profitable": net_profit > 0,
        "return_pct": float((net_profit / (opp.buy_price * size)) * 100)
    }

# Example: 15 bps gross spread, 5 bps buy fee, 5 bps sell fee
opp = ArbOpportunity(
    buy_venue="purpleflea_trading",
    sell_venue="external_venue_b",
    asset="USDC/USDT",
    buy_price=Decimal("1.0000"),
    sell_price=Decimal("1.0015"),
    available_size=Decimal("10000"),
    buy_fee_bps=Decimal("2"),   # Purple Flea maker fee
    sell_fee_bps=Decimal("3")    # taker fee at venue B
)

result = calculate_arb_pnl(opp, Decimal("10000"))
print(f"Net spread: {result['net_spread_bps']:.1f} bps")
print(f"Net profit on $10k: ${result['net_profit']:.2f}")
print(f"Profitable: {result['is_profitable']}")

# Temporal arbitrage: detect primary-venue price moves, fade stale secondary prices
import asyncio
import time
from collections import deque
from decimal import Decimal

class TemporalArbAgent:
    def __init__(self, min_move_bps=5, max_position_usd=10000):
        self.primary_prices = deque(maxlen=1000)
        self.secondary_prices = {}
        self.min_move_bps = min_move_bps
        self.max_pos = Decimal(str(max_position_usd))
        self.active_positions = {}
        self.trades_today = 0
        self.pnl_today = Decimal("0")

    async def on_primary_tick(self, asset: str, price: Decimal, ts_ms: int):
        # Record primary price tick with timestamp
        self.primary_prices.append((asset, price, ts_ms))

        # Check for significant move vs. lookback window
        if len(self.primary_prices) < 2:
            return

        prev_price, prev_ts = self._get_prev_price(asset)
        if prev_price is None:
            return

        move_bps = abs((price - prev_price) / prev_price) * 10000

        if move_bps >= self.min_move_bps:
            # Significant move detected — check secondary venues for stale prices
            await self._scan_secondary_venues(asset, price, move_bps)

    async def _scan_secondary_venues(self, asset, primary_price, move_bps):
        # Fetch prices from all secondary venues concurrently
        tasks = [
            self._fetch_secondary_price(venue, asset)
            for venue in self.secondary_venues
        ]
        secondary_quotes = await asyncio.gather(*tasks)

        for venue, quote in zip(self.secondary_venues, secondary_quotes):
            if quote is None:
                continue

            # If secondary price is stale (hasn't updated), arb may exist
            stale_spread_bps = (abs(primary_price - quote["mid"]) / quote["mid"]) * 10000

            if stale_spread_bps > 10:  # 10 bps threshold after fees
                await self._execute_temporal_arb(
                    asset=asset,
                    venue=venue,
                    direction="buy" if primary_price > quote["mid"] else "sell",
                    size=min(self.max_pos, quote["available_size"])
                )

    def _get_prev_price(self, asset: str):
        # Return most recent prior tick for this asset
        for a, p, t in reversed(list(self.primary_prices)[:-1]):
            if a == asset:
                return p, t
        return None, None

# Statistical arbitrage: pairs trading with cointegration
import numpy as np
from statsmodels.tsa.stattools import coint, adfuller
from statsmodels.regression.linear_model import OLS
from statsmodels.tools import add_constant
from typing import Tuple, Optional
import requests

class PairsTradingAgent:
    """
    Statistical arbitrage agent using cointegration-based pairs trading.
    Fetches price history from Purple Flea Trading API.
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://purpleflea.com/trading-api"
        self.pairs = {}  # active pairs and their parameters
        self.positions = {}
        self.entry_z_threshold = 2.0
        self.exit_z_threshold = 0.5
        self.stop_loss_z = 4.0

    def fetch_price_series(self, asset: str, bars: int = 500) -> np.ndarray:
        resp = requests.get(
            f"{self.base_url}/ohlcv",
            params={"asset": asset, "bars": bars, "interval": "1h"},
            headers={"X-API-Key": self.api_key}
        )
        return np.array([c["close"] for c in resp.json()["candles"]])

    def test_cointegration(
        self,
        prices_a: np.ndarray,
        prices_b: np.ndarray,
        significance: float = 0.05
    ) -> Tuple[bool, float, float]:
        """
        Test for cointegration using Engle-Granger.
        Returns: (is_cointegrated, p_value, hedge_ratio)
        """
        # Engle-Granger cointegration test
        score, pvalue, _ = coint(prices_a, prices_b)
        is_cointegrated = pvalue < significance

        # Estimate hedge ratio via OLS regression
        X = add_constant(prices_b)
        model = OLS(prices_a, X).fit()
        hedge_ratio = model.params[1]

        return is_cointegrated, pvalue, hedge_ratio

    def calculate_spread_zscore(
        self,
        prices_a: np.ndarray,
        prices_b: np.ndarray,
        hedge_ratio: float,
        lookback: int = 60
    ) -> float:
        """
        Calculate rolling z-score of the spread.
        Positive z-score: A is overpriced vs B (sell A, buy B).
        """
        spread = prices_a - (hedge_ratio * prices_b)
        mean = spread[-lookback:].mean()
        std = spread[-lookback:].std()
        if std == 0:
            return 0.0
        return float((spread[-1] - mean) / std)

    def get_signal(self, pair_id: str) -> Optional[str]:
        if pair_id not in self.pairs:
            return None

        p = self.pairs[pair_id]
        prices_a = self.fetch_price_series(p["asset_a"])
        prices_b = self.fetch_price_series(p["asset_b"])

        z = self.calculate_spread_zscore(prices_a, prices_b, p["hedge_ratio"])
        in_pos = pair_id in self.positions

        # Entry signals
        if not in_pos:
            if z > self.entry_z_threshold:
                return "short_spread"   # A expensive: sell A, buy B
            if z < -self.entry_z_threshold:
                return "long_spread"    # A cheap: buy A, sell B
        # Exit and stop-loss signals
        else:
            if abs(z) < self.exit_z_threshold:
                return "close"
            if abs(z) > self.stop_loss_z:
                return "stop_loss"

        return None

#!/usr/bin/env python3
# Multi-strategy arbitrage scanner — Purple Flea Trading API
import asyncio
import aiohttp
import logging
from datetime import datetime
from decimal import Decimal
from typing import List, Dict, Optional
from dataclasses import dataclass, field

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("arb_scanner")

@dataclass
class ScannerConfig:
    api_key: str
    purpleflea_url: str = "https://purpleflea.com/trading-api"
    assets: List[str] = field(default_factory=lambda: [
        "BTC/USDC", "ETH/USDC", "SOL/USDC"
    ])
    spatial_min_bps: float = 8.0
    temporal_min_move_bps: float = 5.0
    stat_z_threshold: float = 2.0
    max_position_usd: float = 5000.0
    daily_loss_limit_usd: float = 500.0
    scan_interval_ms: int = 100

class MultiStrategyArbScanner:
    def __init__(self, config: ScannerConfig):
        self.cfg = config
        self.session: Optional[aiohttp.ClientSession] = None
        self.daily_pnl = Decimal("0")
        self.trade_log: List[Dict] = []
        self.running = False
        self.positions: Dict[str, Dict] = {}
        self.price_cache: Dict[str, Dict] = {}

    async def start(self):
        self.session = aiohttp.ClientSession(
            headers={"X-API-Key": self.cfg.api_key},
            timeout=aiohttp.ClientTimeout(total=5)
        )
        self.running = True
        log.info("Multi-strategy arb scanner starting...")

        try:
            await asyncio.gather(
                self._spatial_scan_loop(),
                self._temporal_scan_loop(),
                self._stat_arb_loop(),
                self._position_manager_loop(),
                self._risk_monitor_loop()
            )
        finally:
            await self.session.close()

    async def _spatial_scan_loop(self):
        while self.running:
            if self._at_loss_limit():
                await asyncio.sleep(1)
                continue

            tasks = [self._check_spatial_arb(a) for a in self.cfg.assets]
            opportunities = await asyncio.gather(*tasks, return_exceptions=True)

            for opp in opportunities:
                if isinstance(opp, dict) and opp.get("profitable"):
                    await self._execute_spatial_arb(opp)

            await asyncio.sleep(self.cfg.scan_interval_ms / 1000)

    async def _risk_monitor_loop(self):
        while self.running:
            if self._at_loss_limit():
                log.warning(f"Daily loss limit hit: ${self.daily_pnl}. Halting all trading.")
                await self._close_all_positions()
                self.running = False
                return
            await asyncio.sleep(5)

    def _at_loss_limit(self) -> bool:
        return float(self.daily_pnl) < -self.cfg.daily_loss_limit_usd

    async def _close_all_positions(self):
        log.info(f"Closing {len(self.positions)} open positions...")
        for pos_id, pos in list(self.positions.items()):
            await self._close_position(pos_id, reason="emergency_close")

# Run the scanner
if __name__ == "__main__":
    config = ScannerConfig(api_key="YOUR_PURPLEFLEA_API_KEY")
    scanner = MultiStrategyArbScanner(config)
    asyncio.run(scanner.start())