DeFi AI Agent
Strategies

import requests from typing import Dict, List BASE_URL = "https://wallet.purpleflea.com/v1" HEADERS = {"Authorization": "Bearer pf_sk_...", "Content-Type": "application/json"} PROTOCOLS = { "aave_v3_eth": "https://aave.com/api/v3/rates?asset=USDC&chain=ethereum", "compound_v3": "https://api.compound.finance/api/v2/ctoken?symbol=cUSDCv3", "morpho_blue": "https://api.morpho.org/markets?asset=USDC", "aave_v3_arb": "https://aave.com/api/v3/rates?asset=USDC&chain=arbitrum", } def fetch_apy(protocol: str, url: str) -> float: """Fetch current supply APY for USDC from a given protocol API.""" try: data = requests.get(url, timeout=5).json() # Response shapes vary — normalise to a float APY if "supplyApy" in data: return float(data["supplyApy"]) * 100 elif "supply_rate" in data: return float(data["supply_rate"]) / 1e18 * 365 * 100 return 0.0 except Exception: return 0.0 def get_best_protocol() -> tuple[str, float]: """Returns (protocol_name, apy_pct) for the highest current USDC yield.""" apys = {p: fetch_apy(p, url) for p, url in PROTOCOLS.items()} best = max(apys, key=apys.get) return best, apys[best] def should_rebalance( current_protocol: str, best_protocol: str, current_apy: float, best_apy: float, position_usd: float, estimated_gas_usd: float = 8.0, ) -> bool: """ Only rebalance if annual yield improvement exceeds switching cost. Assumes gas cost is recovered within 30 days of improved yield. """ if current_protocol == best_protocol: return False annual_gain = position_usd * (best_apy - current_apy) / 100 daily_gain = annual_gain / 365 breakeven_days = estimated_gas_usd / daily_gain if daily_gain > 0 else float("inf") return breakeven_days < 30 # rebalance only if payback < 30 days def rebalance_via_wallet(wallet_id: str, from_protocol: str, to_protocol: str, amount_usdc: float): """ Execute rebalance: withdraw from current protocol, deposit into best. Both operations sent via Purple Flea Wallet API as contract calls. Cross-chain rebalances use the wallet swap/bridge endpoint. """ # Step 1: withdraw from current protocol withdraw = requests.post(f"{BASE_URL}/wallet/contract-call", headers=HEADERS, json={ "wallet_id": wallet_id, "protocol": from_protocol, "action": "withdraw", "asset": "USDC", "amount": amount_usdc, }) print(f"Withdrew {amount_usdc} USDC from {from_protocol}: {withdraw.json()['tx_hash']}") # Step 2: deposit into best protocol (may include bridge if cross-chain) deposit = requests.post(f"{BASE_URL}/wallet/contract-call", headers=HEADERS, json={ "wallet_id": wallet_id, "protocol": to_protocol, "action": "deposit", "asset": "USDC", "amount": amount_usdc, }) print(f"Deposited {amount_usdc} USDC into {to_protocol}: {deposit.json()['tx_hash']}") def calculate_optimal_compound_frequency( position_usd: float, apy_pct: float, gas_cost_usd: float = 3.0, ) -> float: """ Returns optimal compound interval in hours. Compounds when: daily_accrued_reward > gas_cost * 3 (safety margin). """ daily_yield = position_usd * apy_pct / 100 / 365 hours_to_breakeven = (gas_cost_usd * 3) / (daily_yield / 24) return max(hours_to_breakeven, 1.0) # minimum 1 hour between compounds

import math from dataclasses import dataclass @dataclass class LPPosition: pool: str # e.g., "ETH/USDC 0.05%" tick_lower: int tick_upper: int liquidity: float fees_earned_usd: float position_value_usd: float entry_price: float def price_to_tick(price: float) -> int: """Convert a token price to the nearest Uniswap v3 tick.""" return int(math.log(price) / math.log(1.0001)) def calculate_il(entry_price: float, current_price: float) -> float: """ Returns impermanent loss as a percentage (0.0 to 1.0). IL formula for 50/50 constant-product pool. """ r = current_price / entry_price il = 2 * math.sqrt(r) / (1 + r) - 1 return abs(il) # always negative, return magnitude def should_rerange( position: LPPosition, current_price: float, gas_cost_usd: float = 15.0, out_of_range_hours: float = 0, ) -> tuple[bool, str]: """ Decide if a Uniswap v3 position should be reranged. Triggers: 1. Price is out of range (earning zero fees) for over 2 hours 2. Accrued fees cover rerange gas cost * 3 3. IL exceeds 5% (price moved significantly from entry) """ current_tick = price_to_tick(current_price) in_range = position.tick_lower <= current_tick <= position.tick_upper il = calculate_il(position.entry_price, current_price) if not in_range and out_of_range_hours >= 2: return True, "out_of_range_2h" if position.fees_earned_usd >= gas_cost_usd * 3: return True, "fees_cover_rerange" if il > 0.05: return True, "il_threshold" return False, "hold" def compute_new_range( current_price: float, volatility_24h_pct: float, range_width_multiplier: float = 2.0, ) -> tuple[float, float]: """ Set the new LP range as current_price ± (volatility * multiplier). Wider ranges earn fewer fees per unit of liquidity but rebalance less often. Start with 2x daily vol, tighten if volume is consistently high. """ band = current_price * (volatility_24h_pct / 100) * range_width_multiplier return current_price - band, current_price + band def rerange_position(wallet_id: str, position: LPPosition, new_lower: float, new_upper: float): """ Remove liquidity from old range, collect fees, re-add at new range. All steps executed via Purple Flea Wallet API contract calls. """ import requests base = "https://wallet.purpleflea.com/v1/wallet/contract-call" hdrs = {"Authorization": "Bearer pf_sk_..."} # 1. Remove liquidity + collect fees requests.post(base, headers=hdrs, json={ "wallet_id": wallet_id, "protocol": "uniswap_v3", "action": "remove_liquidity", "pool": position.pool, "tick_lower": position.tick_lower, "tick_upper": position.tick_upper, }) # 2. Re-add at new range requests.post(base, headers=hdrs, json={ "wallet_id": wallet_id, "protocol": "uniswap_v3", "action": "add_liquidity", "pool": position.pool, "tick_lower": price_to_tick(new_lower), "tick_upper": price_to_tick(new_upper), "amount_token0": "auto", # wallet splits optimally })

import requests WALLET_BASE = "https://wallet.purpleflea.com/v1" TRADING_BASE = "https://trading.purpleflea.com/v1" HEADERS = {"Authorization": "Bearer pf_sk_...", "Content-Type": "application/json"} MIN_FUNDING_RATE_APR = 15.0 # open hedge only above 15% APR CLOSE_FUNDING_RATE_APR = 5.0 # close hedge below 5% APR def get_funding_rate(symbol: str = "ETH-PERP") -> dict: """ Fetch current funding rate from Purple Flea Trading API. Returns: { "rate_8h": 0.0003, "rate_apr": 13.14, "next_funding_ts": 1234567890 } """ resp = requests.get( f"{TRADING_BASE}/markets/{symbol}/funding", headers=HEADERS, ) return resp.json() def open_delta_neutral_position( wallet_id: str, notional_usd: float, symbol: str = "ETH-PERP", ) -> dict: """ Open a delta-neutral funding harvest position: - Buy spot ETH via wallet swap (long leg) - Short equal notional ETH-PERP via Trading API (short leg) Net delta: ~zero. P&L comes from funding payments to the short. """ # Leg 1: buy spot ETH via wallet swap spot = requests.post(f"{WALLET_BASE}/wallet/swap", headers=HEADERS, json={ "wallet_id": wallet_id, "from_asset": "USDC", "to_asset": "ETH", "amount_usd": notional_usd, "slippage_bps": 30, }).json() # Leg 2: short perpetual via Trading API perp = requests.post(f"{TRADING_BASE}/orders", headers=HEADERS, json={ "symbol": symbol, "side": "sell", "type": "market", "notional_usd": notional_usd, "leverage": 1, # 1x leverage = fully collateralised, minimal liquidation risk "reduce_only": False, }).json() print(f"Delta-neutral position opened: spot={spot['eth_received']} ETH, perp={perp['order_id']}") return {"spot_tx": spot["tx_hash"], "perp_order": perp["order_id"]} def close_delta_neutral_position(wallet_id: str, perp_order_id: str, eth_amount: float): """Close both legs when funding rate compresses below threshold.""" # Close perp short requests.post(f"{TRADING_BASE}/orders", headers=HEADERS, json={ "symbol": "ETH-PERP", "side": "buy", "type": "market", "reduce_only": True, "close_order_id": perp_order_id, }) # Sell spot ETH back to USDC requests.post(f"{WALLET_BASE}/wallet/swap", headers=HEADERS, json={ "wallet_id": wallet_id, "from_asset": "ETH", "to_asset": "USDC", "amount_token": eth_amount, }) def funding_harvest_loop(wallet_id: str, notional_usd: float): """Main agent loop for funding rate harvesting.""" import time active_position = None while True: funding = get_funding_rate() apr = funding["rate_apr"] print(f"ETH-PERP funding APR: {apr:.2f}%") if active_position is None and apr >= MIN_FUNDING_RATE_APR: active_position = open_delta_neutral_position(wallet_id, notional_usd) elif active_position is not None and apr < CLOSE_FUNDING_RATE_APR: close_delta_neutral_position(wallet_id, active_position["perp_order"], notional_usd / 3000) active_position = None time.sleep(300) # check every 5 minutes

import asyncio import aiohttp from dataclasses import dataclass from typing import Optional @dataclass class ArbOpportunity: asset: str buy_chain: str sell_chain: str buy_price: float sell_price: float spread_pct: float estimated_profit_usd: float bridge_cost_usd: float bridge_time_minutes: float DEX_PRICE_APIS = { "ethereum": "https://api.1inch.dev/price/v1.1/1/{token}", "arbitrum": "https://api.1inch.dev/price/v1.1/42161/{token}", "optimism": "https://api.1inch.dev/price/v1.1/10/{token}", "base": "https://api.1inch.dev/price/v1.1/8453/{token}", } BRIDGE_COSTS_USD = { ("ethereum", "arbitrum"): 3.50, ("ethereum", "optimism"): 3.50, ("ethereum", "base"): 2.00, ("arbitrum", "optimism"): 0.50, # L2-to-L2 is cheaper } async def get_token_price_on_chain( session: aiohttp.ClientSession, chain: str, token_address: str, ) -> float: url = DEX_PRICE_APIS[chain].format(token=token_address) async with session.get(url) as r: data = await r.json() return float(data.get(token_address, {}).get("usd", 0)) async def find_arb_opportunities( session: aiohttp.ClientSession, token_address: str, asset_symbol: str, trade_size_usd: float, min_profit_usd: float = 20.0, ) -> list[ArbOpportunity]: """Scan all chain pairs for a profitable arbitrage on the given token.""" chains = list(DEX_PRICE_APIS.keys()) prices = await asyncio.gather( *[get_token_price_on_chain(session, c, token_address) for c in chains], return_exceptions=True ) price_map = {c: p for c, p in zip(chains, prices) if isinstance(p, float) and p > 0} opps = [] for buy_chain, buy_price in price_map.items(): for sell_chain, sell_price in price_map.items(): if buy_chain == sell_chain: continue bridge_cost = BRIDGE_COSTS_USD.get((buy_chain, sell_chain), 5.0) tokens = trade_size_usd / buy_price gross_profit = tokens * (sell_price - buy_price) net_profit = gross_profit - bridge_cost - 3.0 # $3 gas estimate spread_pct = (sell_price - buy_price) / buy_price * 100 if net_profit >= min_profit_usd: opps.append(ArbOpportunity( asset=asset_symbol, buy_chain=buy_chain, sell_chain=sell_chain, buy_price=buy_price, sell_price=sell_price, spread_pct=spread_pct, estimated_profit_usd=net_profit, bridge_cost_usd=bridge_cost, bridge_time_minutes=2.0 if "arbitrum" in (buy_chain, sell_chain) else 15.0, )) return sorted(opps, key=lambda o: o.estimated_profit_usd, reverse=True) def execute_arb(wallet_id: str, opp: ArbOpportunity, amount_usd: float): """ Execute cross-chain arb via Purple Flea wallet cross-chain swap endpoint. The wallet API selects the optimal bridge, handles token approvals, and tracks the bridge transaction until confirmed on the destination chain. """ import requests resp = requests.post( "https://wallet.purpleflea.com/v1/wallet/cross-chain-swap", headers={"Authorization": "Bearer pf_sk_..."}, json={ "wallet_id": wallet_id, "asset": opp.asset, "from_chain": opp.buy_chain, "to_chain": opp.sell_chain, "amount_usd": amount_usd, "slippage_bps": 50, "bridge_strategy": "fastest", # "cheapest" | "fastest" | "safest" } ) print(f"Arb submitted: {opp.asset} {opp.buy_chain}→{opp.sell_chain} est_profit=${opp.estimated_profit_usd:.2f}") return resp.json()

import requests from datetime import datetime STABLE_PROTOCOLS = { "aave_v3": {"chain": "ethereum", "max_allocation_pct": 40, "audit_score": 9}, "compound_v3": {"chain": "ethereum", "max_allocation_pct": 30, "audit_score": 9}, "morpho": {"chain": "ethereum", "max_allocation_pct": 20, "audit_score": 8}, "sky_savings": {"chain": "ethereum", "max_allocation_pct": 20, "audit_score": 8}, "aave_arb": {"chain": "arbitrum", "max_allocation_pct": 25, "audit_score": 9}, } @dataclass # from dataclasses import dataclass class StableAllocation: protocol: str amount_usdc: float apy_pct: float chain: str deposited_at: str def get_risk_adjusted_apy(raw_apy: float, audit_score: int) -> float: """ Discount APY by protocol risk score (1-10, 10 = most trusted). A 6% APY from an audit_score=6 protocol is treated as ~4.8% adjusted. """ risk_multiplier = audit_score / 10 return raw_apy * risk_multiplier def build_optimal_allocation( total_usdc: float, current_apys: dict, # {protocol: apy_pct} ) -> list[StableAllocation]: """ Allocate USDC across protocols in risk-adjusted APY order, respecting per-protocol concentration limits. """ scored = [] for protocol, meta in STABLE_PROTOCOLS.items(): raw_apy = current_apys.get(protocol, 0) adj_apy = get_risk_adjusted_apy(raw_apy, meta["audit_score"]) max_alloc = total_usdc * meta["max_allocation_pct"] / 100 scored.append((protocol, adj_apy, max_alloc, meta["chain"])) scored.sort(key=lambda x: x[1], reverse=True) allocations = [] remaining = total_usdc for protocol, apy, max_alloc, chain in scored: if remaining <= 0 or apy <= 0: break alloc = min(remaining, max_alloc) allocations.append(StableAllocation( protocol=protocol, amount_usdc=alloc, apy_pct=apy, chain=chain, deposited_at=datetime.utcnow().isoformat(), )) remaining -= alloc return allocations def deploy_allocations(wallet_id: str, allocations: list): """Deploy USDC to each protocol via Purple Flea Wallet API.""" for alloc in allocations: resp = requests.post( "https://wallet.purpleflea.com/v1/wallet/contract-call", headers={"Authorization": "Bearer pf_sk_..."}, json={ "wallet_id": wallet_id, "protocol": alloc.protocol, "action": "deposit", "asset": "USDC", "amount": alloc.amount_usdc, "chain": alloc.chain, } ) print(f"Deployed {alloc.amount_usdc:.0f} USDC to {alloc.protocol} ({alloc.apy_pct:.2f}% adj APY)")