Liquidity Mining Strategies for AI Agents: Uniswap v3, IL & Auto-Compounding

from web3 import Web3
import json

w3 = Web3(Web3.HTTPProvider("https://base-mainnet.g.alchemy.com/v2/YOUR_KEY"))

# Uniswap v2 Pair ABI (simplified)
PAIR_ABI = json.loads('[{"constant":true,"inputs":[],"name":"getReserves","outputs":[{"name":"_reserve0","type":"uint112"},{"name":"_reserve1","type":"uint112"},{"name":"_blockTimestampLast","type":"uint32"}],"type":"function"},{"constant":true,"inputs":[],"name":"totalSupply","outputs":[{"name":"","type":"uint256"}],"type":"function"},{"constant":true,"inputs":[{"name":"","type":"address"}],"name":"balanceOf","outputs":[{"name":"","type":"uint256"}],"type":"function"}]')

def get_lp_position_value(pool_address: str, wallet_address: str) -> dict:
    """Get the current value of a v2 LP position."""
    pool = w3.eth.contract(address=Web3.to_checksum_address(pool_address), abi=PAIR_ABI)

    reserves     = pool.functions.getReserves().call()
    total_supply = pool.functions.totalSupply().call()
    lp_balance   = pool.functions.balanceOf(Web3.to_checksum_address(wallet_address)).call()

    if total_supply == 0:
        return {"share": 0, "token0": 0, "token1": 0}

    share    = lp_balance / total_supply
    token0   = reserves[0] * share / 1e18
    token1   = reserves[1] * share / 1e6   # USDC = 6 decimals

    return {
        "lp_balance":    lp_balance / 1e18,
        "share_pct":     share * 100,
        "token0_amount": token0,
        "token1_amount": token1,
        "current_price": reserves[1] / reserves[0] * 1e12  # token1/token0 in human units
    }

import math

def price_to_tick(price: float) -> int:
    """Convert a human-readable price to a Uniswap v3 tick."""
    return math.floor(math.log(price) / math.log(1.0001))

def tick_to_price(tick: int) -> float:
    """Convert a Uniswap v3 tick to human-readable price."""
    return 1.0001 ** tick

def align_tick(tick: int, tick_spacing: int) -> int:
    """Round a tick to the nearest valid multiple of tick_spacing."""
    return (tick // tick_spacing) * tick_spacing

def get_position_amounts(
    liquidity: float,
    current_sqrt_price: float,
    lower_sqrt_price: float,
    upper_sqrt_price: float
) -> tuple[float, float]:
    """
    Calculate token amounts for a given liquidity at current price.
    Uses the standard Uniswap v3 formulas.
    """
    if current_sqrt_price <= lower_sqrt_price:
        # Position is fully in token0 (price below range)
        amount0 = liquidity * (upper_sqrt_price - lower_sqrt_price) / (lower_sqrt_price * upper_sqrt_price)
        return amount0, 0.0
    elif current_sqrt_price >= upper_sqrt_price:
        # Position is fully in token1 (price above range)
        amount1 = liquidity * (upper_sqrt_price - lower_sqrt_price)
        return 0.0, amount1
    else:
        # Price is in range — both tokens
        amount0 = liquidity * (upper_sqrt_price - current_sqrt_price) / (current_sqrt_price * upper_sqrt_price)
        amount1 = liquidity * (current_sqrt_price - lower_sqrt_price)
        return amount0, amount1

# Example: ETH/USDC pool, price = $3,000, range $2,500–$3,500
current_price = 3000
lower_price   = 2500
upper_price   = 3500
liquidity     = 1_000_000  # in Uniswap v3 liquidity units

sq = math.sqrt
eth, usdc = get_position_amounts(
    liquidity,
    sq(current_price), sq(lower_price), sq(upper_price)
)
print(f"ETH in position: {eth:.4f}")
print(f"USDC in position: {usdc:.2f}")

import math
import numpy as np
import matplotlib
# matplotlib.use('Agg')  # uncomment if running headless

def impermanent_loss_v2(price_ratio: float) -> float:
    """
    Calculate impermanent loss for a v2 pool.
    price_ratio = current_price / entry_price
    Returns IL as a negative decimal (e.g., -0.057 for -5.7%)
    """
    k = price_ratio
    return (2 * math.sqrt(k) / (1 + k)) - 1

def impermanent_loss_v3(
    entry_price: float,
    current_price: float,
    lower_price: float,
    upper_price: float
) -> float:
    """
    Calculate impermanent loss for a v3 concentrated liquidity position.
    More complex: IL depends on whether price is in range.
    """
    if current_price <= lower_price or current_price >= upper_price:
        # Out of range — position is 100% one token, IL is crystallized
        if current_price <= lower_price:
            # Fully in token0 — missed downside of token0 relative to HODL
            il = (current_price / entry_price) - 1
        else:
            # Fully in token1 — missed upside of token0 relative to HODL
            il = 1 - (entry_price / current_price)
        return max(il, -1.0)

    # In-range: use v2 formula on the effective sub-range
    sq = math.sqrt
    la = sq(lower_price)
    lb = sq(upper_price)
    lc = sq(current_price)
    le = sq(entry_price)

    # Virtual liquidity positions
    value_lp   = lc - la + lb / lc - lb / le + le / lc - la / le
    value_hold = entry_price * lc / le + 1  # simplified
    return (value_lp / value_hold) - 1 if value_hold != 0 else 0.0

def il_table(price_changes: list[float]) -> None:
    """Print an IL table for a range of price changes."""
    print(f"{'Price Change':>14} {'Price Ratio':>12} {'IL (v2)':>10}")
    print("-" * 40)
    for pct in price_changes:
        ratio = 1 + pct / 100
        il    = impermanent_loss_v2(ratio) * 100
        print(f"{pct:>13.0f}% {ratio:>12.2f} {il:>9.2f}%")

il_table([-75, -50, -25, -10, 0, 10, 25, 50, 100, 200, 400])

import math
import numpy as np
from scipy import stats
import requests

def get_price_history(pool: str, days: int = 30) -> list[float]:
    """Fetch historical prices from Uniswap v3 subgraph."""
    query = """
    {
      poolDayDatas(
        first: %d
        orderBy: date
        orderDirection: desc
        where: { pool: "%s" }
      ) {
        date
        token0Price
      }
    }
    """ % (days, pool.lower())
    resp = requests.post(
        "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3",
        json={"query": query}
    )
    data = resp.json()["data"]["poolDayDatas"]
    return [float(d["token0Price"]) for d in reversed(data)]

def compute_volatility(prices: list[float], annualize: bool = True) -> float:
    """Daily log-return standard deviation, optionally annualized."""
    log_returns = np.diff(np.log(prices))
    sigma_daily = np.std(log_returns)
    return sigma_daily * math.sqrt(365) if annualize else sigma_daily

def optimal_range(
    current_price: float,
    annual_volatility: float,
    holding_period_days: int = 7,
    confidence: float = 0.95
) -> tuple[float, float]:
    """
    Select the tightest range that will stay in-range with
    the given confidence level over the holding period.
    Uses GBM (Geometric Brownian Motion) price model.
    """
    sigma_period = annual_volatility * math.sqrt(holding_period_days / 365)
    z = stats.norm.ppf((1 + confidence) / 2)  # two-tailed

    lower = current_price * math.exp(-z * sigma_period)
    upper = current_price * math.exp(+z * sigma_period)
    return lower, upper

def range_fee_multiplier(
    current_price: float,
    lower: float,
    upper: float
) -> float:
    """
    Capital efficiency multiplier vs v2 for a given range.
    Higher = more fees per dollar of liquidity.
    """
    sq = math.sqrt
    sq_c = sq(current_price)
    sq_a = sq(lower)
    sq_b = sq(upper)
    # Uniswap v3 capital efficiency formula
    return sq_c / (sq_c - sq_a) if sq_c < sq_b else sq_b / (sq_b - sq_a)

# Example
prices = get_price_history("0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640", days=30)  # ETH/USDC 0.05%
vol    = compute_volatility(prices)
price  = prices[-1]
lower, upper = optimal_range(price, vol, holding_period_days=7, confidence=0.95)
multiplier   = range_fee_multiplier(price, lower, upper)

print(f"Current price: ${price:,.2f}")
print(f"Annual vol:    {vol*100:.1f}%")
print(f"7-day 95% range: ${lower:,.2f} — ${upper:,.2f}")
print(f"Capital efficiency: {multiplier:.1f}x vs v2")

import time
import requests
import os

class V3AutoCompounder:
    """
    Monitors a Uniswap v3 position and compounds fees when profitable.
    """
    GAS_COST_USD_ESTIMATE = 2.50  # approx compound transaction cost on Base

    def __init__(self, position_id: int, pf_api_key: str):
        self.position_id = position_id
        self.pf_api_key  = pf_api_key
        self.base_url    = "https://wallet.purpleflea.com/api"

    def get_uncollected_fees(self) -> dict:
        """Query current uncollected fee amounts via Purple Flea wallet API."""
        resp = requests.get(
            f"{self.base_url}/lp/positions/{self.position_id}/fees",
            headers={"X-API-Key": self.pf_api_key}
        )
        resp.raise_for_status()
        return resp.json()  # {"fee0_usd": ..., "fee1_usd": ..., "total_usd": ...}

    def should_compound(self, fees: dict) -> bool:
        """Return True if it's profitable to compound now."""
        total_fees = fees.get("total_usd", 0)
        # Only compound if fees are at least 2x the gas cost
        return total_fees >= 2 * self.GAS_COST_USD_ESTIMATE

    def compound(self) -> dict:
        """Execute a compound: collect fees and reinvest into position."""
        resp = requests.post(
            f"{self.base_url}/lp/positions/{self.position_id}/compound",
            headers={"X-API-Key": self.pf_api_key},
            json={"slippage_bps": 50}  # 0.5% max slippage
        )
        resp.raise_for_status()
        return resp.json()

    def run(self, check_interval_seconds: int = 3600):
        """Main loop: check every hour, compound when profitable."""
        print(f"Auto-compounder started for position #{self.position_id}")
        while True:
            fees = self.get_uncollected_fees()
            print(f"Uncollected fees: ${fees['total_usd']:.4f}")

            if self.should_compound(fees):
                print(f"Compounding ${fees['total_usd']:.4f} in fees...")
                result = self.compound()
                print(f"Compounded: {result}")
            else:
                print(f"Not yet profitable (need ${2*self.GAS_COST_USD_ESTIMATE:.2f})")

            time.sleep(check_interval_seconds)

import math
import requests
import time
import os
from dataclasses import dataclass

@dataclass
class Position:
    token_id:    int
    lower_price: float
    upper_price: float
    liquidity:   float
    entry_price: float

class GammaStyleManager:
    """
    Active v3 position manager:
    - Rebalances when price moves within REBALANCE_THRESHOLD of range boundary
    - Uses trend-adjusted ranges (skews toward trend direction)
    - Tracks IL and exits if it exceeds MAX_IL_PCT
    """
    REBALANCE_THRESHOLD = 0.05   # rebalance when within 5% of range boundary
    MAX_IL_PCT          = 0.08   # exit position if IL exceeds 8%
    BASE_RANGE_PCT      = 0.20   # default ±20% range around current price

    def __init__(self, pf_api_key: str):
        self.pf_api_key = pf_api_key
        self.position: Position = None

    def get_current_price(self, pool: str) -> float:
        resp = requests.get(
            f"https://wallet.purpleflea.com/api/lp/pools/{pool}/price",
            headers={"X-API-Key": self.pf_api_key}
        )
        resp.raise_for_status()
        return resp.json()["price"]

    def get_trend(self, pool: str, lookback_hours: int = 24) -> float:
        """Returns trend coefficient: +1=strong uptrend, -1=strong downtrend."""
        resp = requests.get(
            f"https://wallet.purpleflea.com/api/lp/pools/{pool}/ohlc?hours={lookback_hours}",
            headers={"X-API-Key": self.pf_api_key}
        )
        if not resp.ok:
            return 0.0
        data  = resp.json()
        opens = [c["open"] for c in data]
        closes= [c["close"] for c in data]
        if not opens or opens[0] == 0:
            return 0.0
        return (closes[-1] - opens[0]) / opens[0]  # total return as trend proxy

    def compute_range(self, current_price: float, trend: float) -> tuple[float, float]:
        """Compute an asymmetric range biased toward the trend direction."""
        base = self.BASE_RANGE_PCT
        # Skew: up to 10% extra range in trend direction
        skew = trend * 0.10
        upper_pct = base + max(skew, 0)
        lower_pct = base + max(-skew, 0)
        return current_price * (1 - lower_pct), current_price * (1 + upper_pct)

    def should_rebalance(self, price: float) -> bool:
        if not self.position:
            return False
        range_width = self.position.upper_price - self.position.lower_price
        near_lower = (price - self.position.lower_price) / range_width < self.REBALANCE_THRESHOLD
        near_upper = (self.position.upper_price - price) / range_width < self.REBALANCE_THRESHOLD
        out_of_range = price < self.position.lower_price or price > self.position.upper_price
        return near_lower or near_upper or out_of_range

    def rebalance(self, pool: str):
        """Withdraw current position and deploy a new one at current price."""
        current_price = self.get_current_price(pool)
        trend         = self.get_trend(pool)
        lower, upper  = self.compute_range(current_price, trend)

        # Withdraw existing position
        if self.position:
            requests.delete(
                f"https://wallet.purpleflea.com/api/lp/positions/{self.position.token_id}",
                headers={"X-API-Key": self.pf_api_key}
            )

        # Open new position
        resp = requests.post(
            f"https://wallet.purpleflea.com/api/lp/positions",
            headers={"X-API-Key": self.pf_api_key},
            json={"pool": pool, "lower_price": lower, "upper_price": upper, "amount_usd": 1000}
        )
        if resp.ok:
            data = resp.json()
            self.position = Position(
                token_id=data["token_id"], lower_price=lower,
                upper_price=upper, liquidity=data["liquidity"],
                entry_price=current_price
            )
            print(f"Rebalanced: ${lower:.2f} — ${upper:.2f} (trend={trend:.3f})")

import requests
import os

PF_API_KEY = os.environ["PF_API_KEY"]
WALLET_API = "https://wallet.purpleflea.com/api"
HEADERS    = {"X-API-Key": PF_API_KEY, "Content-Type": "application/json"}

# ── Open a new LP position ────────────────────────────────────────
def open_lp_position(pool: str, lower: float, upper: float, amount_usd: float) -> dict:
    resp = requests.post(f"{WALLET_API}/lp/positions", headers=HEADERS, json={
        "pool":        pool,
        "lower_price": lower,
        "upper_price": upper,
        "amount_usd":  amount_usd,
        "fee_tier":    500  # 0.05%
    })
    resp.raise_for_status()
    return resp.json()

# ── Get position details ──────────────────────────────────────────
def get_position(token_id: int) -> dict:
    resp = requests.get(f"{WALLET_API}/lp/positions/{token_id}", headers=HEADERS)
    resp.raise_for_status()
    return resp.json()

# ── Collect fees ──────────────────────────────────────────────────
def collect_fees(token_id: int) -> dict:
    resp = requests.post(
        f"{WALLET_API}/lp/positions/{token_id}/collect", headers=HEADERS
    )
    resp.raise_for_status()
    return resp.json()

# ── Close position ────────────────────────────────────────────────
def close_position(token_id: int) -> dict:
    resp = requests.delete(
        f"{WALLET_API}/lp/positions/{token_id}", headers=HEADERS
    )
    resp.raise_for_status()
    return resp.json()

# ── Full example ──────────────────────────────────────────────────
ETH_USDC_POOL = "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640"

# Open position
pos = open_lp_position(ETH_USDC_POOL, lower=2800.0, upper=3200.0, amount_usd=500.0)
print(f"Position opened: token_id={pos['token_id']}")

# After some time — check fees
details = get_position(pos["token_id"])
print(f"Uncollected fees: ${details['uncollected_fees_usd']:.4f}")

# Compound if worthwhile
if details["uncollected_fees_usd"] > 5.0:
    result = collect_fees(pos["token_id"])
    print(f"Fees collected: {result}")

def project_compounded_yield(
    principal:     float,
    annual_apr:    float,
    months:        int,
    il_monthly:    float = 0.005,   # 0.5% IL per month (typical for ETH/USDC)
    gas_monthly:   float = 15.0,    # gas costs per month in USD
    compounds_per_day: int = 1
) -> list[dict]:
    """Project compounded LP returns over time."""
    balance   = principal
    compounds = compounds_per_day * 30  # per month
    results   = []

    for month in range(1, months + 1):
        # Fees earned this month (compounded)
        monthly_rate = annual_apr / 12
        fees = balance * ((1 + monthly_rate / compounds) ** compounds - 1)

        # Subtract IL and gas
        il_loss   = balance * il_monthly
        net_gain  = fees - il_loss - gas_monthly
        balance  += net_gain

        results.append({
            "month":    month,
            "balance":  round(balance, 2),
            "fees":     round(fees, 2),
            "il_loss":  round(il_loss, 2),
            "net_gain": round(net_gain, 2),
            "total_roi": round((balance - principal) / principal * 100, 2)
        })
    return results

# ETH/USDC 0.05% pool, 25% APR, $10k principal
results = project_compounded_yield(10_000, 0.25, 12)
print(f"{'Month':>6} {'Balance':>10} {'Fees':>8} {'IL Loss':>8} {'ROI':>8}")
for r in results:
    print(f"{r['month']:>6} ${r['balance']:>9,.2f} ${r['fees']:>7,.2f} ${r['il_loss']:>7,.2f} {r['total_roi']:>7.1f}%")