1. The StableSwap Invariant
Curve's core innovation is the StableSwap invariant — a mathematical hybrid between a constant sum (x+y=k) and a constant product (xy=k) formula. The constant sum allows near-zero slippage near the peg; the constant product prevents price from going to zero when one asset depletes. The invariant is:
An^n * sum(x_i) + D = An^n * D + D^(n+1) / (n^n * prod(x_i))
Where A is the amplification coefficient, n is the number of tokens, x_i are token balances, and D is the invariant (total value when balanced). A higher A-parameter means more stable pricing near peg at the cost of worse pricing during depeg events.
from scipy.optimize import brentq
import numpy as np
def stableswap_D(balances: list, A: int) -> float:
"""
Compute the StableSwap invariant D using Newton's method.
balances: list of token amounts (normalized to same decimals)
A: amplification coefficient
"""
n = len(balances)
S = sum(balances)
if S == 0:
return 0.0
D = S
Ann = A * n**n
for _ in range(255):
D_P = D
for b in balances:
D_P = D_P * D / (n * b)
D_prev = D
D = (Ann * S + n * D_P) * D / ((Ann - 1) * D + (n + 1) * D_P)
if abs(D - D_prev) <= 1:
return D
return D
def stableswap_y(i: int, j: int, x: float, balances: list, A: int) -> float:
"""
Compute the output amount y[j] given input x[i].
i: index of input token
j: index of output token
"""
n = len(balances)
D = stableswap_D(balances, A)
Ann = A * n**n
# Replace balances[i] with new balance after receiving x
new_balances = balances.copy()
new_balances[i] += x
# Compute sum and product excluding j
S_ = sum(new_balances[k] for k in range(n) if k != j)
c = D
for k in range(n):
if k != j:
c = c * D / (n * new_balances[k])
c = c * D / (n * Ann)
b = S_ + D / Ann
# Solve y^2 + (b-D)y - c = 0 for y
y = D
for _ in range(255):
y_prev = y
y = (y**2 + c) / (2*y + b - D)
if abs(y - y_prev) <= 1:
break
return y
def get_dy(i: int, j: int, dx: float, balances: list, A: int, fee: float = 0.0004) -> float:
"""
Output amount for a swap, after fee.
fee: fraction (0.0004 = 0.04%)
"""
x = balances[i] + dx
y = stableswap_y(i, j, dx, balances, A)
dy = balances[j] - y - 1 # -1 for rounding
fee_amount = dy * fee
return dy - fee_amount
# Example: 3pool (DAI/USDC/USDT) with A=2000
balances = [30_000_000, 35_000_000, 25_000_000] # DAI, USDC, USDT
A = 2000
# Swap 1M USDC (index 1) for DAI (index 0)
output = get_dy(1, 0, 1_000_000, balances, A, fee=0.0001)
slippage = (1_000_000 - output) / 1_000_000
print(f"Swap 1M USDC → {output:,.0f} DAI")
print(f"Slippage: {slippage*100:.4f}%")
Amplification Parameter A: Typical values range from 100 (new pools) to 2000+ (established 3pool). A higher A compresses the StableSwap curve toward the constant sum line, reducing slippage but increasing depeg risk. Agents should monitor A-parameter governance proposals.
2. Tricrypto (USDT/WBTC/ETH) Mechanics
Curve's Tricrypto pools handle volatile assets (BTC, ETH) using the CryptoSwap invariant — a more complex geometric mean formula with a dynamic fee and an internal price oracle. Unlike stableswap, these pools use a "repegging" mechanism to adjust the invariant as prices change.
"""
Tricrypto Pool Interface
Uses the CryptoSwap invariant: K0 * prod(x_i) = D^n (approximately)
With fee structure that increases with recent volatility.
"""
TRICRYPTO2_ADDRESS = "0xD51a44d3FaE010294C616388b506AcdA1bfAAE46"
TRICRYPTO2_ABI = [
{"name": "get_dy", "inputs": [
{"name": "i", "type": "uint256"},
{"name": "j", "type": "uint256"},
{"name": "dx", "type": "uint256"}
], "outputs": [{"name": "", "type": "uint256"}], "type": "function"},
{"name": "price_oracle", "inputs": [{"name": "k", "type": "uint256"}],
"outputs": [{"name": "", "type": "uint256"}], "type": "function"},
{"name": "last_prices", "inputs": [{"name": "k", "type": "uint256"}],
"outputs": [{"name": "", "type": "uint256"}], "type": "function"},
{"name": "fee", "inputs": [], "outputs": [{"name": "", "type": "uint256"}], "type": "function"},
{"name": "A", "inputs": [], "outputs": [{"name": "", "type": "uint256"}], "type": "function"},
{"name": "gamma", "inputs": [], "outputs": [{"name": "", "type": "uint256"}], "type": "function"},
{"name": "balances", "inputs": [{"name": "i", "type": "uint256"}],
"outputs": [{"name": "", "type": "uint256"}], "type": "function"},
]
class TricryptoPool:
"""Interface to Curve Tricrypto2 pool."""
TOKENS = {0: ("USDT", 6), 1: ("WBTC", 8), 2: ("WETH", 18)}
def __init__(self, w3, address=TRICRYPTO2_ADDRESS):
self.contract = w3.eth.contract(address, abi=TRICRYPTO2_ABI)
def get_price_oracle(self, k: int) -> float:
"""Internal price oracle for asset k (in USDT terms)."""
raw = self.contract.functions.price_oracle(k).call()
return raw / 1e18
def get_balances(self) -> dict:
"""Current pool balances in token units."""
balances = {}
for i, (name, decimals) in self.TOKENS.items():
raw = self.contract.functions.balances(i).call()
balances[name] = raw / 10**decimals
return balances
def simulate_swap(self, from_token: int, to_token: int, amount: float) -> dict:
"""Simulate a swap and compute effective price and slippage."""
_, from_dec = self.TOKENS[from_token]
_, to_dec = self.TOKENS[to_token]
dx = int(amount * 10**from_dec)
dy = self.contract.functions.get_dy(from_token, to_token, dx).call()
out = dy / 10**to_dec
oracle_price = self.get_price_oracle(to_token if to_token != 0 else from_token)
expected_out = amount / oracle_price if from_token == 0 else amount * oracle_price
slippage = (expected_out - out) / expected_out
return {
"input": amount,
"output": out,
"slippage_pct": slippage * 100,
"effective_price": amount / out if to_token == 0 else out / amount,
}
def pool_imbalance(self) -> dict:
"""Compute pool imbalance relative to oracle prices."""
balances = self.get_balances()
oracle_eth = self.get_price_oracle(1) # WBTC price in USDT
oracle_wbtc = self.get_price_oracle(2) # WETH price in USDT
usdt_val = balances["USDT"]
wbtc_val = balances["WBTC"] * oracle_eth
eth_val = balances["WETH"] * oracle_wbtc
total_val = usdt_val + wbtc_val + eth_val
return {
"USDT_pct": usdt_val / total_val * 100,
"WBTC_pct": wbtc_val / total_val * 100,
"WETH_pct": eth_val / total_val * 100,
"total_tvl": total_val,
"imbalance_score": max(abs(usdt_val/total_val - 1/3),
abs(wbtc_val/total_val - 1/3),
abs(eth_val /total_val - 1/3))
}3. Gauge Voting with veToken Economics
Curve's tokenomics are built around vote-escrowed CRV (veCRV). Locking CRV gives voting power to direct CRV emissions to specific pool gauges. The amount of veCRV received depends on lock duration:
veCRV = CRV_locked * (lock_end - now) / 4_years
from datetime import datetime, timedelta
def veCRV_amount(crv_locked: float, lock_end: datetime) -> float:
"""Compute veCRV balance given a lock."""
now = datetime.utcnow()
four_years = timedelta(days=4*365)
time_remaining = lock_end - now
if time_remaining.total_seconds() <= 0:
return 0.0
return crv_locked * time_remaining.total_seconds() / four_years.total_seconds()
def voting_power_boost(veCRV_held: float, total_veCRV: float) -> float:
"""Share of total voting power."""
return veCRV_held / total_veCRV if total_veCRV > 0 else 0
def lp_boost_multiplier(veCRV_share: float, gauge_weight: float,
lp_share: float) -> float:
"""
Curve LP boost: up to 2.5x on CRV emissions.
min_boost = 1.0 (base)
max_boost = 2.5
Formula from Curve docs.
"""
working_balance = min(
0.4 * lp_share + 0.6 * veCRV_share,
lp_share
)
base_working = 0.4 * lp_share
if base_working == 0:
return 1.0
return working_balance / base_working
# Example
crv_locked = 100_000
lock_end = datetime.utcnow() + timedelta(days=4*365)
veCRV = veCRV_amount(crv_locked, lock_end)
total_veCRV = 500_000_000 # Total veCRV supply
our_share = voting_power_boost(veCRV, total_veCRV)
our_lp_share = 0.001 # 0.1% of pool
boost = lp_boost_multiplier(our_share, gauge_weight=0.02, lp_share=our_lp_share)
print(f"veCRV: {veCRV:,.0f}")
print(f"VP share: {our_share*100:.4f}%")
print(f"LP boost: {boost:.2f}x")4. Bribe Economics via Votium and Hidden Hand
Protocols that want to attract liquidity to their Curve pool need CRV emissions, which require gauge votes. Since not all veCRV holders vote optimally, a bribe market emerged: protocols pay veCRV holders to vote for their gauge. The two main bribe platforms are Votium (for vlCVX/veCRV) and Hidden Hand (for veCRV directly).
import aiohttp
import asyncio
VOTIUM_API = "https://api.votium.app/api/v1"
HIDDEN_HAND = "https://api.hiddenhand.finance/proposal/curve"
async def fetch_votium_bribes(session) -> list[dict]:
"""Fetch current round bribe data from Votium."""
async with session.get(f"{VOTIUM_API}/vlcvx/eligible") as r:
data = await r.json()
bribes = []
for gauge in data:
bribes.append({
"gauge": gauge["gauge"],
"protocol": gauge.get("protocol", "unknown"),
"bribe_usd": gauge.get("totalValue", 0),
"votes_cvx": gauge.get("totalVotes", 0),
})
return sorted(bribes, key=lambda x: x["bribe_usd"], reverse=True)
async def fetch_hidden_hand_bribes(session) -> list[dict]:
"""Fetch current veCRV bribe data from Hidden Hand."""
async with session.get(HIDDEN_HAND) as r:
data = await r.json()
bribes = []
for item in data.get("data", []):
bribes.append({
"gauge": item["title"],
"bribe_usd": item.get("totalValue", 0),
"bribe_token": item.get("bribeToken", "USDC"),
"max_votes": item.get("maxVotesAllowed", 0),
})
return sorted(bribes, key=lambda x: x["bribe_usd"], reverse=True)
def bribe_per_vote_roi(bribe_usd: float, votes: float,
crv_price: float, crv_per_vote: float) -> float:
"""
ROI of voting for a bribed gauge vs. the next-best unbiased allocation.
bribe_per_vote = bribe_usd / votes
opportunity_cost = crv_per_vote * crv_price (CRV you'd earn voting for your own pool)
"""
if votes == 0:
return float('inf')
bribe_per_vote = bribe_usd / votes
opportunity_cost = crv_per_vote * crv_price # Per unit of voting power
roi = bribe_per_vote / opportunity_cost - 1
return roi
class BribeScanner:
"""Scan bribe platforms and compute optimal voting allocations."""
def __init__(self, veCRV_balance: float, own_gauge: str):
self.veCRV = veCRV_balance
self.own_gauge = own_gauge
async def scan_and_rank(self) -> list[dict]:
"""Rank all bribe opportunities by ROI."""
async with aiohttp.ClientSession() as session:
vh_bribes, hh_bribes = await asyncio.gather(
fetch_votium_bribes(session),
fetch_hidden_hand_bribes(session),
return_exceptions=True
)
all_bribes = []
if isinstance(vh_bribes, list):
all_bribes.extend(vh_bribes)
if isinstance(hh_bribes, list):
all_bribes.extend(hh_bribes)
# Compute ROI for each
crv_price = 0.50 # Fetch from oracle in production
crv_per_vote = 2.0 # CRV per 1 veCRV per epoch (approximate)
ranked = []
for b in all_bribes:
roi = bribe_per_vote_roi(
b["bribe_usd"],
b.get("votes_cvx", b.get("max_votes", 1_000_000)),
crv_price,
crv_per_vote
)
b["roi"] = roi
ranked.append(b)
ranked.sort(key=lambda x: x["roi"], reverse=True)
return ranked
def compute_optimal_allocation(self, ranked_bribes: list, top_n: int = 5) -> dict:
"""
Allocate veCRV votes to maximize bribe income.
Simple greedy: allocate proportionally to bribe_usd weighted by ROI.
"""
# Always allocate some votes to own gauge for LP boost
own_gauge_allocation = 0.20 # 20% to own gauge
remaining = 1.0 - own_gauge_allocation
top = ranked_bribes[:top_n]
total_roi = sum(max(b["roi"], 0.01) for b in top)
allocation = {self.own_gauge: own_gauge_allocation}
for b in top:
weight = max(b["roi"], 0.01) / total_roi
allocation[b["gauge"]] = weight * remaining
return allocation5. Pool Imbalance Arbitrage
When a Curve stableswap pool becomes imbalanced (one asset exceeds its target share), the pool prices the overweighted asset at a discount relative to external markets. Agents that detect and exploit these imbalances provide a valuable service while earning risk-free profit.
import requests
def get_curve_pool_price(pool_contract, from_idx: int, to_idx: int,
amount: float, decimals_in: int, decimals_out: int) -> float:
"""Get effective exchange rate from Curve pool."""
dx = int(amount * 10**decimals_in)
dy = pool_contract.functions.get_dy(from_idx, to_idx, dx).call()
return (dy / 10**decimals_out) / amount
def find_arb_opportunity(
pool_prices: dict, # {'USDC→DAI': 0.9995, 'DAI→USDC': 1.0004}
market_prices: dict, # {'USDC': 1.00, 'DAI': 0.9998}
min_spread_bps: int = 3
) -> list[dict]:
"""
Find arbitrage opportunities between Curve pool and external market.
Returns list of profitable routes sorted by spread.
"""
opportunities = []
for route, curve_rate in pool_prices.items():
from_token, to_token = route.split("→")
# External conversion: buy to_token with from_token
market_rate = market_prices.get(to_token, 1.0) / market_prices.get(from_token, 1.0)
spread_bps = (curve_rate - market_rate) / market_rate * 10_000
if spread_bps > min_spread_bps:
opportunities.append({
"route": route,
"curve_rate": curve_rate,
"market_rate": market_rate,
"spread_bps": spread_bps,
"direction": "buy_from_curve" # Curve gives better rate
})
return sorted(opportunities, key=lambda x: x["spread_bps"], reverse=True)
class CurveArbBot:
"""Automated arbitrage bot for Curve stablecoin pool imbalances."""
MONITORED_POOLS = [
{"name": "3pool", "address": "0xbEbc44782C7dB0a1A60Cb6fe97d0b483032FF1C7"},
{"name": "FRAX/USDC", "address": "0xDcEF968d416a41Cdac0ED8702fAC8128A64241A2"},
{"name": "LUSD/3CRV", "address": "0xEd279fDD11cA84bEef15AF5D39BB4d4bEE23F0cA"},
]
def __init__(self, w3, wallet, chainlink_eth_usdc: str):
self.w3 = w3
self.wallet = wallet
self.oracle = chainlink_eth_usdc
def analyze_pool_imbalance(self, pool_address: str) -> dict:
"""
Check pool composition and compute imbalance metrics.
Returns opportunity score and recommended action.
"""
pool = self.w3.eth.contract(pool_address, abi=CURVE_POOL_ABI)
# Get balances (3pool: DAI=0, USDC=1, USDT=2)
b0 = pool.functions.balances(0).call() / 1e18 # DAI
b1 = pool.functions.balances(1).call() / 1e6 # USDC
b2 = pool.functions.balances(2).call() / 1e6 # USDT
total = b0 + b1 + b2
share0 = b0 / total
share1 = b1 / total
share2 = b2 / total
target = 1/3
# Imbalance: deviation from 33.3% for each token
imbalances = {
"DAI": share0 - target,
"USDC": share1 - target,
"USDT": share2 - target,
}
# Find most overweighted and underweighted
most_over = max(imbalances, key=imbalances.get)
most_under = min(imbalances, key=imbalances.get)
# Swap overweighted→underweighted to earn premium
over_idx = {"DAI": 0, "USDC": 1, "USDT": 2}[most_over]
under_idx = {"DAI": 0, "USDC": 1, "USDT": 2}[most_under]
test_amount = 100_000 # $100k test swap
dec_in = 18 if most_over == "DAI" else 6
dec_out = 18 if most_under == "DAI" else 6
rate = get_curve_pool_price(pool, over_idx, under_idx, test_amount, dec_in, dec_out)
# External rate should be ~1.0 for stablecoins
premium_bps = (rate - 1.0) * 10_000
return {
"pool": pool_address,
"most_over": most_over,
"most_under": most_under,
"imbalance": imbalances,
"curve_rate": rate,
"premium_bps": premium_bps,
"arb_viable": premium_bps > 3.0,
"optimal_size": self.estimate_optimal_arb_size(pool, over_idx, under_idx)
}
def estimate_optimal_arb_size(self, pool, i: int, j: int) -> float:
"""
Binary search for optimal trade size that maximizes profit.
Profit = (rate - 1) * size - gas_cost
"""
gas_cost = 15.0 # $15 for a swap on mainnet
def profit(size):
rate = get_curve_pool_price(pool, i, j, size, 6, 6)
return (rate - 1.0) * size - gas_cost
# Find size that maximizes profit (profit is concave)
from scipy.optimize import minimize_scalar
result = minimize_scalar(lambda x: -profit(x),
bounds=(1_000, 10_000_000),
method='bounded')
return result.x if profit(result.x) > 0 else 06. CRV Emission Optimization
CRV emissions follow a fixed schedule that decays over time. The total emissions per week are split between gauges in proportion to their gauge weights, which are set by veCRV voter decisions. Agents providing liquidity can optimize their CRV yield by:
- Choosing pools with high gauge weights relative to their TVL (high CRV yield per dollar)
- Locking CRV to accumulate veCRV for up to 2.5x boost
- Using Convex Finance to auto-compound CRV + CVX rewards
- Voting for their own pool's gauge to increase emissions
def crv_apr_estimate(
gauge_weight: float, # Fraction of total gauge weight (0 to 1)
pool_tvl: float, # Pool TVL in USD
crv_price: float, # CRV price in USD
weekly_crv_emissions: float = 2_000_000, # ~2M CRV/week total
boost: float = 1.0 # 1.0 to 2.5 based on veCRV holdings
) -> float:
"""Estimate annualized CRV yield for an LP position."""
weekly_pool_emissions = weekly_crv_emissions * gauge_weight
weekly_usd_emissions = weekly_pool_emissions * crv_price * boost
annual_usd_emissions = weekly_usd_emissions * 52
return annual_usd_emissions / pool_tvl
# Practical gauge APR scanner
CURVE_GAUGE_CONTROLLER = "0x2F50D538606Fa9EDD2B11E2446BEb18C9D5846bB"
async def scan_gauge_aprs(w3, crv_price: float) -> list[dict]:
"""Scan all gauges and rank by CRV APR."""
controller = w3.eth.contract(CURVE_GAUGE_CONTROLLER, abi=GAUGE_CONTROLLER_ABI)
# In production: iterate over all registered gauges
# Here we show the pattern for known gauges
KNOWN_GAUGES = [
{"name": "3pool", "gauge": "0xbFcF63294aD7105dEa65aA58F8AE5BE2D9d0952A", "tvl": 600_000_000},
{"name": "FRAX/USDC", "gauge": "0xCFc25170633581Bf896CB6CDeE170e3E3Aa59503", "tvl": 300_000_000},
{"name": "stETH/ETH", "gauge": "0x182B723a58739a9c974cFDB385ceaDb237453c28", "tvl": 1_200_000_000},
{"name": "USDT/WBTC/ETH","gauge": "0xDeFd8FdD20e0f34115C7018CCfb655796F6B2168","tvl": 150_000_000},
]
results = []
for g in KNOWN_GAUGES:
# Get gauge weight from controller
raw_weight = controller.functions.gauge_relative_weight(g["gauge"]).call()
weight = raw_weight / 1e18
apr = crv_apr_estimate(weight, g["tvl"], crv_price, boost=2.5)
results.append({
"name": g["name"],
"gauge_weight": weight * 100,
"tvl": g["tvl"],
"crv_apr_boosted": apr * 100,
"crv_apr_base": crv_apr_estimate(weight, g["tvl"], crv_price, boost=1.0) * 100,
})
return sorted(results, key=lambda x: x["crv_apr_boosted"], reverse=True)7. Convex Finance Integration
Convex Finance (CVX) aggregates veCRV to provide Curve LPs with boosted CRV rewards without requiring them to lock CRV individually. The trade-off: LPs share a portion of their boost with Convex stakers. Agents should evaluate whether to go native (own veCRV) vs. Convex based on capital size:
| Approach | Min Capital for Break-Even | Additional Yield | Complexity |
|---|---|---|---|
| Native veCRV (4yr lock) | $500K+ | Max 2.5x boost + voting power | High |
| Convex (cvxCRV) | Any | ~1.7x average boost + CVX | Low |
| Yearn Curve vaults | Any | Auto-compound + strategy | Very Low |
| Stake DAO | $10K+ | ~1.8x boost + SDT rewards | Medium |
CONVEX_BOOSTER = "0xF403C135812408BFbE8713b5A23a04b3D48AAE31"
BOOSTER_ABI = [
{"name": "poolLength", "inputs": [], "outputs": [{"type": "uint256"}], "type": "function"},
{"name": "poolInfo", "inputs": [{"name": "_pid", "type": "uint256"}],
"outputs": [
{"name": "lptoken", "type": "address"},
{"name": "token", "type": "address"},
{"name": "gauge", "type": "address"},
{"name": "crvRewards", "type": "address"},
{"name": "stash", "type": "address"},
{"name": "shutdown","type": "bool"},
], "type": "function"},
{"name": "deposit", "inputs": [
{"name": "_pid", "type": "uint256"},
{"name": "_amount", "type": "uint256"},
{"name": "_stake", "type": "bool"},
], "outputs": [], "type": "function"},
]
class ConvexStrategy:
"""Deposit Curve LP tokens into Convex for boosted CRV + CVX rewards."""
CVX_EXTRA_REWARDS_MULTIPLIER = 0.15 # Approximate CVX rewards on top of CRV
def __init__(self, w3, wallet_address: str, pool_pid: int):
self.w3 = w3
self.wallet = wallet_address
self.pid = pool_pid
self.booster = w3.eth.contract(CONVEX_BOOSTER, abi=BOOSTER_ABI)
def get_pool_info(self) -> dict:
"""Get pool metadata from Convex."""
info = self.booster.functions.poolInfo(self.pid).call()
return {
"lp_token": info[0],
"crv_rewards": info[3],
"is_active": not info[5]
}
def estimate_convex_apr(
self,
crv_apr_base: float,
cvx_price: float,
crv_price: float
) -> dict:
"""
Estimate total APR through Convex.
Convex keeps 16% of CRV, 1% to cvxCRV lockers.
LPs receive 17% boost (Convex's aggregate boost) + CVX incentives.
"""
convex_boost = 1.17 - 0.17 * 0.17 # ~1.17x net after Convex fee
crv_apr_cv = crv_apr_base * convex_boost
cvx_apr = crv_apr_cv * self.CVX_EXTRA_REWARDS_MULTIPLIER * (cvx_price / crv_price)
return {
"crv_apr": crv_apr_cv * 100,
"cvx_apr": cvx_apr * 100,
"total_apr": (crv_apr_cv + cvx_apr) * 100,
"convex_fee_pct": 17.0,
}
def deposit(self, lp_amount: int, stake: bool = True) -> dict:
"""Deposit LP tokens into Convex and optionally auto-stake."""
pool = self.get_pool_info()
if not pool["is_active"]:
return {"error": "pool_shutdown"}
# First approve booster to spend LP tokens
# Then call deposit(pid, amount, stake=True)
tx = self.booster.functions.deposit(self.pid, lp_amount, stake).build_transaction({
"from": self.wallet,
"gas": 300_000
})
return {"tx": tx, "pid": self.pid, "amount": lp_amount}8. Complete CurveAgent Implementation
The CurveAgent below combines pool monitoring, bribe scanning, gauge optimization, and auto-compounding into a single autonomous strategy:
"""
CurveAgent — Autonomous Curve Finance strategy manager.
Strategies: optimal pool selection, bribe voting, Convex deposit, arb scanning.
"""
import asyncio
import logging
from typing import Optional
log = logging.getLogger("CurveAgent")
class CurveAgentConfig:
capital_usdc: float = 100_000
enable_bribes: bool = True
enable_arb: bool = True
use_convex: bool = True
min_crv_apr: float = 5.0 # Skip pools with <5% CRV APR
arb_min_bps: int = 4 # Min 4 bps spread for arb
rebalance_interval: int = 7 # Days between pool rebalances
class CurveAgent:
def __init__(self, w3, wallet, config: CurveAgentConfig, own_gauge: str = ""):
self.w3 = w3
self.wallet = wallet
self.cfg = config
self.bribe_scanner = BribeScanner(0, own_gauge) # veCRV balance=0 initially
self.arb_bot = CurveArbBot(w3, wallet, "")
self.current_pool = None
self.stats = {
"arb_profits": 0.0,
"crv_harvested": 0.0,
"bribe_income": 0.0,
"pool_switches": 0,
}
async def select_optimal_pool(self, crv_price: float) -> dict:
"""Find the best pool to LP in based on CRV APR."""
gauges = await scan_gauge_aprs(self.w3, crv_price)
eligible = [g for g in gauges if g["crv_apr_boosted"] >= self.cfg.min_crv_apr]
if not eligible:
log.warning("No pools meet minimum CRV APR requirement")
return gauges[0] if gauges else {}
best = eligible[0]
log.info(f"Selected pool: {best['name']} | CRV APR: {best['crv_apr_boosted']:.1f}%")
return best
async def execute_bribe_vote(self, epoch_id: int):
"""Scan bribes and submit optimal vote allocation."""
ranked = await self.bribe_scanner.scan_and_rank()
allocation = self.bribe_scanner.compute_optimal_allocation(ranked, top_n=5)
log.info(f"Voting allocation: {allocation}")
# In production: call GaugeController.vote_for_gauge_weights for each gauge
total_bribe_income = sum(
b["bribe_usd"] * allocation.get(b["gauge"], 0)
for b in ranked
if b["gauge"] in allocation
)
self.stats["bribe_income"] += total_bribe_income
return {"allocation": allocation, "estimated_income": total_bribe_income}
async def run_arb_scanner(self):
"""Scan for pool imbalances and execute arbitrage."""
for pool_info in CurveArbBot.MONITORED_POOLS:
try:
analysis = self.arb_bot.analyze_pool_imbalance(pool_info["address"])
if analysis["arb_viable"] and analysis["optimal_size"] > 0:
size = analysis["optimal_size"]
profit_estimate = (analysis["curve_rate"] - 1.0) * size - 15.0
log.info(
f"Arb opportunity in {pool_info['name']}: "
f"{analysis['premium_bps']:.1f} bps, "
f"${profit_estimate:.2f} est. profit"
)
# Execute swap: buy overweighted asset from Curve, sell on CEX/Uniswap
# ... (transaction execution)
self.stats["arb_profits"] += profit_estimate
except Exception as e:
log.error(f"Arb scan error for {pool_info['name']}: {e}")
async def harvest_and_compound(self):
"""Claim CRV/CVX rewards and re-invest."""
# 1. Call BaseRewardPool.getReward() on Convex
# 2. Optionally lock CRV → veCRV if balance > threshold
# 3. Sell excess CVX and re-invest in LP
crv_harvested = 500.0 # Placeholder
self.stats["crv_harvested"] += crv_harvested
log.info(f"Harvested {crv_harvested:.2f} CRV")
async def run(self):
"""Main agent loop."""
log.info(f"CurveAgent starting | capital: ${self.cfg.capital_usdc:,.0f}")
crv_price = 0.50 # Fetch from oracle
best_pool = await self.select_optimal_pool(crv_price)
log.info(f"Initial deployment: {best_pool.get('name', 'unknown')}")
iteration = 0
while True:
try:
# Arb scanning: every 30 seconds
if self.cfg.enable_arb:
await self.run_arb_scanner()
# Harvest rewards: every 12 hours
if iteration % 1440 == 0:
await self.harvest_and_compound()
# Bribe voting: every 2 weeks (Curve epoch)
if iteration % 20160 == 0 and self.cfg.enable_bribes:
await self.execute_bribe_vote(epoch_id=iteration // 20160)
# Pool rebalance check: every 7 days
if iteration % (self.cfg.rebalance_interval * 1440) == 0:
new_pool = await self.select_optimal_pool(crv_price)
if new_pool.get("name") != (self.current_pool or {}).get("name"):
log.info(f"Switching pool: {self.current_pool} → {new_pool['name']}")
self.current_pool = new_pool
self.stats["pool_switches"] += 1
log.info(
f"Arb profits: ${self.stats['arb_profits']:.2f} | "
f"CRV harvested: {self.stats['crv_harvested']:.0f} | "
f"Bribe income: ${self.stats['bribe_income']:.2f}"
)
iteration += 1
except Exception as e:
log.error(f"CurveAgent error: {e}")
await asyncio.sleep(30) # 30-second loop
# Entrypoint
if __name__ == "__main__":
from web3 import Web3
w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"))
wallet = w3.eth.account.from_key("0x...")
agent = CurveAgent(
w3, wallet,
CurveAgentConfig(),
own_gauge="0xbFcF63294aD7105dEa65aA58F8AE5BE2D9d0952A"
)
asyncio.run(agent.run())Put Your Curve Profits to Work
Use Purple Flea to trade perpetual futures on 275+ markets — a natural complement to Curve LP positions for hedging directional risk.
Start on Purple Flea Perp Trading Docs9. Risk Considerations for Curve Strategies
Curve strategies carry specific risks that agents must model and monitor:
| Risk Type | Description | Mitigation |
|---|---|---|
| Depeg Risk | Stablecoin loses peg; LP stuck holding depegged asset | Diversify across multiple stablecoin pools; monitor oracle prices |
| Smart Contract Risk | Curve or Convex contract exploit | Use battle-tested pools; avoid new experimental pools |
| CRV Price Risk | CRV price drops, reducing yield | Hedge CRV with short position or convert to stablecoins regularly |
| Gauge Weight Risk | Pool loses gauge weight, emissions drop | Monitor biweekly gauge votes; re-allocate if weight drops >30% |
| Admin Key Risk | Pool A-parameter can be changed by admin (slowly) | Track governance proposals; exit if aggressive A-change proposed |
class CurveRiskMonitor:
"""Monitor Curve-specific risks and alert when thresholds are breached."""
def __init__(self, pool_addresses: list, stablecoin_oracles: dict):
self.pools = pool_addresses
self.oracles = stablecoin_oracles # {token: oracle_address}
self.alerts = []
def check_depeg(self, token: str, price: float, threshold: float = 0.005) -> bool:
"""Alert if stablecoin depegs by more than threshold."""
deviation = abs(price - 1.0)
if deviation > threshold:
self.alerts.append({
"type": "DEPEG",
"token": token,
"price": price,
"deviation_pct": deviation * 100
})
return True
return False
def check_pool_composition(self, pool_address: str,
balances: dict, threshold: float = 0.15) -> bool:
"""Alert if pool is more than threshold% imbalanced."""
n = len(balances)
target = 1.0 / n
for token, share in balances.items():
if abs(share - target) > threshold:
self.alerts.append({
"type": "POOL_IMBALANCE",
"pool": pool_address,
"token": token,
"share": share,
"deviation": share - target
})
return True
return False
def check_gauge_weight_drop(self, gauge: str, old_weight: float,
new_weight: float, threshold: float = 0.30) -> bool:
"""Alert if gauge weight drops significantly."""
drop = (old_weight - new_weight) / old_weight
if drop > threshold:
self.alerts.append({
"type": "GAUGE_WEIGHT_DROP",
"gauge": gauge,
"drop_pct": drop * 100
})
return True
return False
def get_alerts(self) -> list:
alerts = self.alerts.copy()
self.alerts.clear()
return alerts