DEX Trading with AI Agents:
Uniswap, Curve, and AMM Strategies

from dataclasses import dataclass
from typing import Tuple


@dataclass
class UniswapV2Pool:
    reserve_x: float    # token0 reserves
    reserve_y: float    # token1 reserves
    fee: float = 0.003  # default 0.3%

    @property
    def k(self) -> float:
        return self.reserve_x * self.reserve_y

    @property
    def spot_price_x_in_y(self) -> float:
        """Price of x denominated in y (e.g. ETH price in USDC)."""
        return self.reserve_y / self.reserve_x

    def get_amount_out(self, amount_in: float, x_to_y: bool) -> Tuple[float, float]:
        """
        Compute output amount and price impact for a given input.
        Returns (amount_out, price_impact_bps).
        """
        amount_in_eff = amount_in * (1 - self.fee)

        if x_to_y:
            reserve_in, reserve_out = self.reserve_x, self.reserve_y
            spot = self.spot_price_x_in_y
        else:
            reserve_in, reserve_out = self.reserve_y, self.reserve_x
            spot = 1 / self.spot_price_x_in_y

        amount_out = (
            reserve_out * amount_in_eff
            / (reserve_in + amount_in_eff)
        )
        exec_price = amount_out / amount_in
        price_impact_bps = abs(spot - exec_price) / spot * 10000
        return amount_out, price_impact_bps

    def max_input_for_impact(self, max_impact_bps: float, x_to_y: bool) -> float:
        """
        Binary search for max input that keeps price impact
        below max_impact_bps. Useful for slippage-limited agents.
        """
        lo, hi = 0.0, (self.reserve_x if x_to_y else self.reserve_y) * 0.5
        for _ in range(50):  # 50 iterations of bisection
            mid = (lo + hi) / 2
            _, impact = self.get_amount_out(mid, x_to_y)
            if impact < max_impact_bps:
                lo = mid
            else:
                hi = mid
        return lo


# Example usage
pool = UniswapV2Pool(
    reserve_x=1000.0,       # 1000 ETH
    reserve_y=3_000_000.0  # 3M USDC
)

out, impact = pool.get_amount_out(10.0, x_to_y=True)
print(f"10 ETH → {out:.2f} USDC (impact: {impact:.2f} bps)")
# → 10 ETH → 29611.86 USDC (impact: 129.47 bps)

max_in = pool.max_input_for_impact(50.0, x_to_y=True)
print(f"Max ETH for <50bps impact: {max_in:.4f}")
# → Max ETH for <50bps impact: 1.6693

import math
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple


@dataclass
class PoolEdge:
    pool_id: str
    token_in: str
    token_out: str
    pool: UniswapV2Pool


class DEXRouter:
    """
    Multi-hop DEX router using Bellman-Ford for optimal path.
    Supports Uniswap v2/v3 and Curve pools via a unified interface.
    """

    def __init__(self, pools: List[PoolEdge]):
        self.pools = pools
        # Build adjacency: token -> list of outgoing edges
        self.adj: Dict[str, List[PoolEdge]] = {}
        for edge in pools:
            self.adj.setdefault(edge.token_in, []).append(edge)

    def find_best_route(
        self,
        token_in: str,
        token_out: str,
        amount_in: float,
        max_hops: int = 3
    ) -> Tuple[Optional[List[str]], float]:
        """
        DFS with pruning to find max output route up to max_hops deep.
        Returns (route as list of pool_ids, expected output amount).
        """
        best_out = 0.0
        best_path: Optional[List[str]] = None

        def dfs(token: str, amount: float,
                path: List[str], visited: set):
            nonlocal best_out, best_path
            if token == token_out:
                if amount > best_out:
                    best_out, best_path = amount, path[:]
                return
            if len(path) >= max_hops:
                return
            for edge in self.adj.get(token, []):
                if edge.pool_id in visited:
                    continue
                is_x_to_y = edge.token_in == "x"  # simplified
                out_amt, _ = edge.pool.get_amount_out(amount, is_x_to_y)
                visited.add(edge.pool_id)
                path.append(f"{edge.pool_id}({edge.token_in}->{edge.token_out})")
                dfs(edge.token_out, out_amt, path, visited)
                path.pop()
                visited.discard(edge.pool_id)

        dfs(token_in, amount_in, [], set())
        return best_path, best_out

    def detect_arbitrage(self, token: str, amount: float) -> Optional[dict]:
        """Find circular arbitrage: token -> ... -> token with profit."""
        route, out = self.find_best_route(token, token, amount, max_hops=4)
        if route and out > amount:
            return {
                "route": route,
                "profit": out - amount,
                "profit_pct": (out - amount) / amount * 100
            }
        return None

import math
import numpy as np
from scipy.stats import norm


def il_v2(price_ratio: float) -> float:
    """Impermanent loss for Uniswap v2 given price_ratio = P_new/P_entry."""
    return 2 * math.sqrt(price_ratio) / (1 + price_ratio) - 1


def il_v3(price_entry: float, price_now: float,
          range_low: float, range_high: float) -> float:
    """
    Impermanent loss for a Uniswap v3 concentrated position.
    Assumes equal USD value of both tokens at entry (50/50).
    """
    if price_now <= range_low:
        # All in token0 (base asset), price below range
        hodl_value = 0.5 * price_now / price_entry + 0.5
        pool_value = price_now / price_entry  # fully in base
    elif price_now >= range_high:
        # All in token1 (quote asset), price above range
        hodl_value = 0.5 * price_now / price_entry + 0.5
        pool_value = math.sqrt(range_high / range_low)  # fixed at range edge
    else:
        # In range: standard v2 IL but amplified by concentration
        r = price_now / price_entry
        hodl_value = 0.5 * r + 0.5
        pool_value = hodl_value + il_v2(r) * hodl_value
    return pool_value / hodl_value - 1


def expected_il_gbm(
    sigma_annual: float,
    days: int,
    simulations: int = 10000
) -> dict:
    """
    Monte Carlo simulation of expected IL over N days,
    assuming Geometric Brownian Motion price process.
    """
    sigma_daily = sigma_annual / math.sqrt(365)
    # Simulate log returns
    log_returns = np.random.normal(
        loc=-0.5 * sigma_daily**2,
        scale=sigma_daily,
        size=(simulations, days)
    )
    price_ratios = np.exp(log_returns.sum(axis=1))
    il_values = np.array([il_v2(r) for r in price_ratios])
    return {
        "mean_il_pct": il_values.mean() * 100,
        "p5_il_pct":  np.percentile(il_values, 5)  * 100,
        "p95_il_pct": np.percentile(il_values, 95) * 100,
    }


# Example: ETH/USDC pool over 90 days, sigma=80% annual
result = expected_il_gbm(
    sigma_annual=0.80,   # 80% annualized vol
    days=90
)
print(f"90-day expected IL: {result['mean_il_pct']:.2f}%")
print(f"5th percentile IL:  {result['p5_il_pct']:.2f}%")
# → 90-day expected IL: -3.82%
# → 5th percentile IL:  -14.31%  (worst 5% of scenarios)

import asyncio
import httpx
from typing import Optional

TRADING_API = "https://purpleflea.com/trading-api"
MIN_PROFIT_BPS = 20   # minimum profit after fees
SCAN_INTERVAL = 0.5   # seconds between scans


class DEXArbitrageAgent:
    """
    Monitors Purple Flea DEX + Uniswap prices and executes
    cross-venue arbitrage when profit exceeds threshold.
    Uses Purple Flea Trading API for CEX leg.
    """

    def __init__(self, api_key: str,
                 wallet_address: str,
                 capital_usd: float):
        self.api_key = api_key
        self.wallet = wallet_address
        self.capital = capital_usd
        self.client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {api_key}"}
        )
        self.total_pnl = 0.0
        self.trades_executed = 0

    async def get_cex_price(self, symbol: str) -> float:
        r = await self.client.get(
            f"{TRADING_API}/v1/price/{symbol}"
        )
        return float(r.json()["mid"])

    async def get_dex_quote(self, token_in: str,
                              token_out: str,
                              amount: float) -> float:
        """Get best DEX quote via Purple Flea DEX aggregator."""
        r = await self.client.get(
            f"{TRADING_API}/v1/dex/quote",
            params={"tokenIn": token_in, "tokenOut": token_out,
                    "amountIn": amount}
        )
        return float(r.json()["amountOut"])

    async def check_arb_opportunity(self, symbol: str) -> Optional[dict]:
        cex_price  = await self.get_cex_price(symbol)
        dex_out    = await self.get_dex_quote(
            "USDC", symbol.split("/")[0],
            self.capital
        )
        dex_price = self.capital / dex_out
        spread_bps = (cex_price - dex_price) / cex_price * 10000
        if abs(spread_bps) > MIN_PROFIT_BPS:
            return {
                "symbol": symbol,
                "direction": "buy_dex" if dex_price < cex_price else "sell_dex",
                "spread_bps": spread_bps,
                "cex_price": cex_price,
                "dex_price": dex_price,
            }
        return None

    async def run(self, symbols: list[str]):
        print(f"[ARB] Starting with ${self.capital:.0f} capital")
        while True:
            for sym in symbols:
                opp = await self.check_arb_opportunity(sym)
                if opp:
                    print(f"[ARB] Opportunity: {opp}")
                    # Execute both legs (implement atomically via flash loan)
                    await self._execute_arb(opp)
            await asyncio.sleep(SCAN_INTERVAL)

    async def _execute_arb(self, opp: dict):
        # Buy cheap leg, sell expensive leg via Purple Flea
        print(f"[ARB] Executing {opp['direction']} {opp['symbol']}"
              f" | spread: {opp['spread_bps']:.1f}bps")
        self.trades_executed += 1