1. Aave V3 Architecture Overview
Aave V3 is a decentralized lending protocol where suppliers deposit assets to earn interest and borrowers collateralize their positions to access liquidity. V3 introduced three major upgrades over V2: Isolation Mode (new assets with capped debt exposure), Efficiency Mode (e-mode) (higher LTV for correlated assets), and Portal (cross-chain liquidity bridging).
The core contract agents interact with is the Pool.sol (formerly LendingPool.sol). Key functions:
supply(asset, amount, onBehalfOf, referralCode)— deposit collateral, receive aTokensborrow(asset, amount, interestRateMode, referralCode, onBehalfOf)— borrow against collateralrepay(asset, amount, interestRateMode, onBehalfOf)— repay borrowwithdraw(asset, amount, to)— remove supply (burn aTokens)flashLoan(receiverAddress, assets, amounts, modes, onBehalfOf, params, referralCode)liquidationCall(collateralAsset, debtAsset, user, debtToCover, receiveAToken)
Interest Rate Model: Aave uses a dual-slope interest rate model. Utilization below the "optimal point" (typically 80–90%) keeps rates low. Above optimal, rates spike sharply. Agents must monitor utilization ratios to avoid being caught in a sudden rate spike.
Key Data Structures
from web3 import Web3
from typing import NamedTuple
# Aave V3 Pool ABI (key functions only)
POOL_ABI = [
{"name": "getUserAccountData", "type": "function",
"inputs": [{"name": "user", "type": "address"}],
"outputs": [
{"name": "totalCollateralBase", "type": "uint256"},
{"name": "totalDebtBase", "type": "uint256"},
{"name": "availableBorrowsBase", "type": "uint256"},
{"name": "currentLiquidationThreshold", "type": "uint256"},
{"name": "ltv", "type": "uint256"},
{"name": "healthFactor", "type": "uint256"}
]},
# ... supply, borrow, repay, withdraw, flashLoan
]
class AccountData(NamedTuple):
total_collateral_usd: float
total_debt_usd: float
available_borrows_usd: float
liquidation_threshold: float # e.g. 0.85
ltv: float # loan-to-value e.g. 0.80
health_factor: float # > 1.0 = safe
def get_account_data(pool_contract, address: str) -> AccountData:
"""Fetch position summary for an address."""
raw = pool_contract.functions.getUserAccountData(address).call()
return AccountData(
total_collateral_usd = raw[0] / 1e8, # Aave uses 8 decimals for USD
total_debt_usd = raw[1] / 1e8,
available_borrows_usd = raw[2] / 1e8,
liquidation_threshold = raw[3] / 1e4,
ltv = raw[4] / 1e4,
health_factor = raw[5] / 1e18
)2. E-Mode Categories and LTV Mechanics
Efficiency Mode (e-mode) allows borrowers to achieve higher capital efficiency when collateral and debt are in the same asset category. V3 defines e-mode categories — each with its own LTV, liquidation threshold, and liquidation bonus.
| E-Mode Category | Assets | Max LTV | Liq. Threshold | Max Leverage |
|---|---|---|---|---|
| 1 — Stablecoins | USDC, USDT, DAI, FRAX | 97% | 97% | ~33x |
| 2 — ETH Correlated | ETH, stETH, rETH, cbETH | 93% | 95% | ~14x |
| 3 — BTC Correlated | WBTC, tBTC, cbBTC | 90% | 92% | ~10x |
| 0 — Default | All assets | Varies | Varies | 3–5x typical |
def emode_max_leverage(ltv: float) -> float:
"""
Maximum leverage achievable via recursive looping in e-mode.
Geometric series: 1 + LTV + LTV^2 + ... = 1/(1-LTV)
"""
return 1.0 / (1.0 - ltv)
def emode_safe_leverage(ltv: float, safety_buffer: float = 0.05) -> float:
"""Practical max leverage with safety buffer."""
effective_ltv = ltv - safety_buffer
return 1.0 / (1.0 - effective_ltv)
# Stablecoin e-mode: 97% LTV
print(f"Theoretical max: {emode_max_leverage(0.97):.1f}x") # 33.3x
print(f"Safe max (5% buf): {emode_safe_leverage(0.97):.1f}x") # 10.9x
# ETH e-mode: 93% LTV
print(f"ETH theoretical: {emode_max_leverage(0.93):.1f}x") # 14.3x
print(f"ETH safe max: {emode_safe_leverage(0.93):.1f}x") # 8.3x3. Recursive Lending Loop Math
A recursive (or "looping") strategy amplifies yield by repeatedly depositing collateral and borrowing against it. Each iteration increases both supply and borrow exposure. The loop terminates when available borrows are below a minimum threshold.
def simulate_lending_loop(
initial_capital: float,
supply_apy: float,
borrow_apy: float,
ltv: float,
n_loops: int,
slippage_per_loop: float = 0.001 # 0.1% swap cost per loop
) -> dict:
"""
Simulate a recursive lending loop.
Returns final position stats and net APY.
"""
supply = initial_capital
debt = 0.0
total_swap_cost = 0.0
for i in range(n_loops):
borrow_amount = supply * ltv - debt
if borrow_amount < 1.0: # min $1 threshold
break
# Borrow and re-supply
swap_cost = borrow_amount * slippage_per_loop
total_swap_cost += swap_cost
supply += borrow_amount - swap_cost
debt += borrow_amount
leverage = supply / initial_capital
health_factor = supply * ltv / debt if debt > 0 else float('inf')
# Annual revenue
gross_supply_income = supply * supply_apy
gross_borrow_cost = debt * borrow_apy
net_income = gross_supply_income - gross_borrow_cost - total_swap_cost
net_apy = net_income / initial_capital
return {
"loops": n_loops,
"leverage": leverage,
"supply": supply,
"debt": debt,
"health_factor": health_factor,
"gross_supply_apy": gross_supply_income / initial_capital,
"gross_borrow_cost_rate": gross_borrow_cost / initial_capital,
"net_apy": net_apy,
"swap_drag": total_swap_cost / initial_capital,
}
# Example: USDC/USDT stablecoin loop in e-mode
# Supply USDC at 5%, borrow USDT at 4%, LTV 95%
result = simulate_lending_loop(
initial_capital=10_000,
supply_apy=0.05,
borrow_apy=0.04,
ltv=0.95,
n_loops=8
)
print(f"Leverage: {result['leverage']:.1f}x")
print(f"Net APY: {result['net_apy']*100:.2f}%")
print(f"Health: {result['health_factor']:.3f}")
# Leverage: 6.8x, Net APY: 6.82%, Health: 1.40
Critical Risk: In a recursive loop, a sudden increase in borrow rates can flip a position from profitable to loss-making within hours. Always monitor the spread between supply APY and borrow APY. If spread narrows below 0.3%, begin unwinding the loop.
Flash Loan Loop Unwinding
Unwinding a deep loop requires multiple transactions or a flash loan. The flash loan approach is atomic and gas-efficient: borrow the full debt amount in one transaction, repay the Aave debt, withdraw all collateral, swap back, and repay the flash loan.
def flash_unwind_plan(supply: float, debt: float, ltv: float) -> dict:
"""
Plan for unwinding a leveraged position using a flash loan.
Flash borrow = total debt amount.
"""
flash_amount = debt
collateral_to_withdraw = supply
# After flash repay: need to swap collateral to repay flash loan
# Net proceeds: collateral - flash_amount = equity
equity = supply - debt
gas_estimate_usd = 20.0 # Ethereum mainnet
return {
"flash_borrow": flash_amount,
"repay_debt": debt,
"withdraw_collateral": collateral_to_withdraw,
"net_equity": equity,
"gas_cost": gas_estimate_usd,
"transactions": 1 # Single atomic tx
}4. Rate Switching: Stable vs Variable
Aave V3 offers two interest rate modes for borrows: variable (changes with utilization) and stable (locked at origination but rebalanceable). Agents should dynamically switch between modes based on rate forecasts.
import numpy as np
def optimal_rate_mode(current_variable: float, current_stable: float,
forecast_utilization: float, optimal_util: float,
base_rate: float, slope1: float, slope2: float,
horizon_days: int = 30) -> str:
"""
Decide whether to use stable or variable rate.
Returns 'stable' or 'variable'.
"""
# Forecast variable rate at target utilization
if forecast_utilization < optimal_util:
forecast_variable = base_rate + (forecast_utilization / optimal_util) * slope1
else:
excess = (forecast_utilization - optimal_util) / (1 - optimal_util)
forecast_variable = base_rate + slope1 + excess * slope2
# Cost comparison over horizon
variable_cost = current_variable * 0.3 + forecast_variable * 0.7 # Weighted average
stable_cost = current_stable # Locked in
# Stable adds a premium; only choose if variable expected to exceed stable
stable_premium = 0.01 # Typical 1% premium for stable rate
effective_stable = current_stable + stable_premium # Origination premium
if variable_cost < effective_stable:
return "variable"
elif horizon_days > 90 and forecast_variable > current_stable * 1.2:
return "stable" # Long horizon, rates expected to spike
else:
return "variable"
class RateSwitcher:
"""Monitor and switch Aave borrow rate modes."""
def __init__(self, pool, account_address, asset, w3):
self.pool = pool
self.account = account_address
self.asset = asset
self.w3 = w3
self.current_mode = 2 # 1=stable, 2=variable
def check_and_switch(self, current_variable_rate: float,
current_stable_rate: float,
threshold_bps: int = 50):
"""Switch rate mode if savings exceed threshold (in bps)."""
# Current cost
if self.current_mode == 2: # Currently variable
current_cost = current_variable_rate
alternative = current_stable_rate
target_mode = 1
else: # Currently stable
current_cost = current_stable_rate
alternative = current_variable_rate
target_mode = 2
savings_bps = (current_cost - alternative) * 10_000
if savings_bps > threshold_bps:
print(f"Switching rate mode: {savings_bps:.0f} bps savings")
# Call pool.swapBorrowRateMode(asset, rateMode)
tx = self.pool.functions.swapBorrowRateMode(
self.asset, target_mode
).build_transaction({"from": self.account, "gas": 200000})
# sign and send...
self.current_mode = target_mode
return {"switched": True, "savings_bps": savings_bps}
return {"switched": False, "savings_bps": savings_bps}5. Health Factor Monitoring and Risk Management
The health factor (HF) is the most critical metric for any Aave borrower. When HF drops below 1.0, the position becomes eligible for liquidation. Agents must maintain a safe buffer and implement automated deleveraging when HF approaches danger zones.
from enum import Enum
class HealthStatus(Enum):
SAFE = "safe" # HF > 2.0
CAUTION = "caution" # 1.5 < HF <= 2.0
WARNING = "warning" # 1.2 < HF <= 1.5
DANGER = "danger" # 1.05 < HF <= 1.2
CRITICAL = "critical" # HF <= 1.05
def classify_health(hf: float) -> HealthStatus:
if hf > 2.0: return HealthStatus.SAFE
if hf > 1.5: return HealthStatus.CAUTION
if hf > 1.2: return HealthStatus.WARNING
if hf > 1.05: return HealthStatus.DANGER
return HealthStatus.CRITICAL
def hf_after_price_drop(
collateral_usd: float,
debt_usd: float,
liq_threshold: float,
price_drop_pct: float
) -> float:
"""Calculate HF after a collateral price drop."""
new_collateral = collateral_usd * (1 - price_drop_pct)
return (new_collateral * liq_threshold) / debt_usd
class HealthMonitor:
"""Continuous health factor monitor with automated responses."""
RESPONSE_PLAN = {
HealthStatus.CAUTION: {"action": "log", "deleverage_pct": 0},
HealthStatus.WARNING: {"action": "alert", "deleverage_pct": 0.10},
HealthStatus.DANGER: {"action": "deleverage", "deleverage_pct": 0.25},
HealthStatus.CRITICAL: {"action": "emergency", "deleverage_pct": 0.50},
}
def __init__(self, aave_agent):
self.agent = aave_agent
def respond_to_health(self, hf: float, account_data: AccountData):
status = classify_health(hf)
plan = self.RESPONSE_PLAN.get(status, {"action": "none", "deleverage_pct": 0})
if plan["action"] == "log":
print(f"HF={hf:.3f} — SAFE, monitoring")
elif plan["action"] == "alert":
print(f"HF={hf:.3f} — WARNING: preparing deleverage")
elif plan["action"] == "deleverage":
amount = account_data.total_debt_usd * plan["deleverage_pct"]
print(f"HF={hf:.3f} — DANGER: repaying ${amount:,.0f}")
self.agent.partial_repay(amount)
elif plan["action"] == "emergency":
print(f"HF={hf:.3f} — CRITICAL: emergency flash unwind!")
self.agent.emergency_unwind()
return {"status": status.value, "hf": hf, "action": plan["action"]}
def stress_test(self, account_data: AccountData, drops=(0.10, 0.20, 0.30, 0.40)):
"""Show HF at various price drop scenarios."""
print("Price Drop | Resulting HF | Status")
for drop in drops:
hf = hf_after_price_drop(
account_data.total_collateral_usd,
account_data.total_debt_usd,
account_data.liquidation_threshold,
drop
)
print(f" -{drop*100:.0f}% | {hf:.3f} | {classify_health(hf).value}")6. Flash Loan Liquidations
When a borrower's health factor drops below 1.0, any address can call liquidationCall to repay a portion of their debt and seize their collateral at a discount (the liquidation bonus, typically 5–10%). Agents that execute liquidations profitably are providing a critical service to the protocol.
"""
AaveLiquidationBot — scans for underwater positions and executes flash loan liquidations.
"""
import asyncio
from web3 import Web3
from eth_abi import encode
class LiquidationBot:
LIQUIDATION_BONUS = {
"ETH": 0.05,
"WBTC": 0.10,
"USDC": 0.05,
"LINK": 0.10,
}
MAX_DEBT_COVERAGE = 0.50 # Aave V3 allows up to 50% liquidation
def __init__(self, w3: Web3, pool_address: str, flashloan_address: str, wallet):
self.w3 = w3
self.pool = self.w3.eth.contract(pool_address, abi=POOL_ABI)
self.fl_addr = flashloan_address
self.wallet = wallet
def estimate_profit(
self,
debt_asset: str,
collateral_asset: str,
debt_to_cover: float,
collateral_price_usd: float,
debt_price_usd: float
) -> float:
"""Estimate flash liquidation profit."""
bonus = self.LIQUIDATION_BONUS.get(collateral_asset, 0.05)
debt_value = debt_to_cover * debt_price_usd
collateral_rx = debt_value * (1 + bonus) / collateral_price_usd
# Cost: flash loan fee (0% on Aave V3) + gas + slippage
gas_cost = 35.0 # ~$35 on mainnet
slippage = debt_value * 0.003 # 0.3%
profit = debt_value * bonus - gas_cost - slippage
return profit
async def find_liquidatable_positions(self, user_list: list) -> list:
"""Scan a list of users for unhealthy positions."""
liquidatable = []
for user in user_list:
data = get_account_data(self.pool, user)
if data.health_factor < 1.0 and data.total_debt_usd > 100:
liquidatable.append({
"user": user,
"hf": data.health_factor,
"debt_usd": data.total_debt_usd,
"collateral_usd": data.total_collateral_usd,
})
return sorted(liquidatable, key=lambda x: x["debt_usd"], reverse=True)
async def execute_flash_liquidation(
self,
borrower: str,
debt_asset: str,
collateral_asset: str,
debt_to_cover: int # in wei
) -> dict:
"""
Execute a flash loan liquidation.
1. Flash borrow debt_asset from Aave
2. Call liquidationCall
3. Swap collateral back to debt_asset
4. Repay flash loan
5. Keep profit
"""
profit_estimate = self.estimate_profit(
debt_asset, collateral_asset,
debt_to_cover / 1e6, # USDC has 6 decimals
current_prices[collateral_asset],
current_prices[debt_asset]
)
if profit_estimate < 10.0: # Min $10 profit
return {"executed": False, "reason": f"insufficient_profit: ${profit_estimate:.2f}"}
print(f"Liquidating {borrower[:8]}... est. profit: ${profit_estimate:.2f}")
# Encode calldata for flash loan receiver
params = encode(
["address", "address", "address", "uint256", "bool"],
[borrower, debt_asset, collateral_asset, debt_to_cover, False]
)
# Build flash loan tx (calls our FlashLiquidator contract)
tx = self.pool.functions.flashLoan(
self.fl_addr,
[debt_asset],
[debt_to_cover],
[0], # mode=0: no debt position
self.wallet.address,
params,
0 # referral
).build_transaction({
"from": self.wallet.address,
"gas": 500_000,
"nonce": self.w3.eth.get_transaction_count(self.wallet.address),
})
signed = self.wallet.sign_transaction(tx)
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)
return {
"executed": True,
"tx_hash": tx_hash.hex(),
"status": "success" if receipt.status == 1 else "failed",
"profit_estimate": profit_estimate
}
async def run(self, scan_interval: int = 15):
"""Main loop: scan for opportunities every N seconds."""
print("LiquidationBot started")
while True:
try:
users = await self.get_recent_borrowers() # From subgraph
targets = await self.find_liquidatable_positions(users)
for target in targets[:3]: # Top 3 opportunities
result = await self.execute_flash_liquidation(
target["user"],
debt_asset=USDC_ADDRESS,
collateral_asset=WETH_ADDRESS,
debt_to_cover=int(target["debt_usd"] * self.MAX_DEBT_COVERAGE * 1e6)
)
print(f"Liquidation result: {result}")
except Exception as e:
print(f"Error in liquidation loop: {e}")
await asyncio.sleep(scan_interval)7. Cross-Chain Aave V3 Strategies
Aave V3 is deployed on 10+ chains (Ethereum, Arbitrum, Optimism, Polygon, Base, Avalanche, and more). Interest rates for the same asset can differ significantly across chains due to different utilization levels. Agents can exploit these differentials:
| Chain | USDC Supply APY | ETH Borrow APY | TVL Rank |
|---|---|---|---|
| Ethereum | 4.2% | 3.1% | 1st |
| Arbitrum | 5.8% | 2.8% | 2nd |
| Base | 6.1% | 2.5% | 3rd |
| Optimism | 4.9% | 2.9% | 4th |
| Polygon | 3.8% | 3.4% | 5th |
import aiohttp
import asyncio
AAVE_SUBGRAPH_URLS = {
"ethereum": "https://api.thegraph.com/subgraphs/name/aave/protocol-v3",
"arbitrum": "https://api.thegraph.com/subgraphs/name/aave/protocol-v3-arbitrum",
"base": "https://api.studio.thegraph.com/query/48427/aave-v3-base/v0.0.1",
"optimism": "https://api.thegraph.com/subgraphs/name/aave/protocol-v3-optimism",
}
RATE_QUERY = """
{
reserves(where: {symbol: "USDC"}) {
symbol
liquidityRate
variableBorrowRate
utilizationRate
}
}
"""
async def fetch_rates(session, chain: str, url: str) -> dict:
"""Fetch current Aave rates for a chain."""
async with session.post(url, json={"query": RATE_QUERY}) as r:
data = await r.json()
reserve = data["data"]["reserves"][0]
return {
"chain": chain,
"supply_apy": int(reserve["liquidityRate"]) / 1e27 * 100,
"borrow_apy": int(reserve["variableBorrowRate"]) / 1e27 * 100,
"utilization": int(reserve["utilizationRate"]) / 1e27 * 100,
}
async def find_best_chain_for_supply():
"""Find the chain with the highest USDC supply APY."""
async with aiohttp.ClientSession() as session:
tasks = [fetch_rates(session, chain, url)
for chain, url in AAVE_SUBGRAPH_URLS.items()]
results = await asyncio.gather(*tasks, return_exceptions=True)
valid = [r for r in results if isinstance(r, dict)]
valid.sort(key=lambda x: x["supply_apy"], reverse=True)
print("Chain | Supply APY | Borrow APY | Utilization")
for r in valid:
print(f"{r['chain']:10s} | {r['supply_apy']:8.2f}% | {r['borrow_apy']:8.2f}% | {r['utilization']:6.1f}%")
return valid[0] # Best chain8. Complete AaveAgent Implementation
The following AaveAgent integrates recursive lending, health monitoring, rate switching, and liquidation scanning into a single autonomous system:
"""
AaveAgent — Autonomous Aave V3 yield optimizer.
Strategies: recursive supply/borrow loop + liquidation hunting.
"""
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Optional
log = logging.getLogger("AaveAgent")
@dataclass
class AaveConfig:
rpc_url: str
private_key: str
pool_address: str
capital_usdc: float = 50_000
target_emode: int = 1 # Stablecoin e-mode
max_loops: int = 6 # Recursive loop depth
target_hf: float = 1.5 # Target health factor
min_hf: float = 1.2 # Emergency deleverage trigger
enable_liquidations: bool = True
min_liquidation_profit: float = 20.0
@dataclass
class LoopPosition:
supply_usd: float
debt_usd: float
loops: int
net_apy: float
health_factor: float
rate_mode: int = 2 # variable
class AaveAgent:
def __init__(self, config: AaveConfig):
self.cfg = config
self.w3 = Web3(Web3.HTTPProvider(config.rpc_url))
self.wallet = self.w3.eth.account.from_key(config.private_key)
self.pool = self.w3.eth.contract(config.pool_address, abi=POOL_ABI)
self.position: Optional[LoopPosition] = None
self.liquidation_bot: Optional[LiquidationBot] = None
self.stats = {
"loops_opened": 0,
"deleverages": 0,
"liquidations_executed": 0,
"liquidation_profit": 0.0,
"fees_earned": 0.0,
}
async def enter_loop(self, n_loops: Optional[int] = None):
"""Enter a recursive lending loop."""
loops = n_loops or self.cfg.max_loops
# Enable e-mode first
tx = self.pool.functions.setUserEMode(self.cfg.target_emode).build_transaction(
{"from": self.wallet.address, "gas": 100_000}
)
await self.send_tx(tx)
log.info(f"E-mode {self.cfg.target_emode} enabled")
# Get current rates
rates = await self.get_current_rates("USDC")
supply_apy = rates["supply"]
borrow_apy = rates["borrow"]
ltv = 0.95 # Stablecoin e-mode
simulation = simulate_lending_loop(
self.cfg.capital_usdc, supply_apy, borrow_apy, ltv, loops
)
if simulation["net_apy"] < 0.005: # < 0.5% net APY
log.warning(f"Loop not profitable: {simulation['net_apy']*100:.2f}% APY")
return
log.info(
f"Opening loop: {loops} iterations, {simulation['leverage']:.1f}x leverage, "
f"{simulation['net_apy']*100:.2f}% net APY"
)
# Execute loop
capital = self.cfg.capital_usdc
for i in range(loops):
await self.supply_usdc(capital)
borrow_amt = capital * ltv
if borrow_amt < 5.0:
break
await self.borrow_usdt(borrow_amt)
capital = borrow_amt
acct = get_account_data(self.pool, self.wallet.address)
self.position = LoopPosition(
supply_usd=acct.total_collateral_usd,
debt_usd=acct.total_debt_usd,
loops=loops,
net_apy=simulation["net_apy"],
health_factor=acct.health_factor
)
self.stats["loops_opened"] += 1
log.info(f"Loop open: HF={acct.health_factor:.3f}")
async def monitor_health(self):
"""Check health factor and respond if needed."""
acct = get_account_data(self.pool, self.wallet.address)
status = classify_health(acct.health_factor)
monitor = HealthMonitor(self)
result = monitor.respond_to_health(acct.health_factor, acct)
if result["action"] in ("deleverage", "emergency"):
self.stats["deleverages"] += 1
if self.position:
self.position.health_factor = acct.health_factor
return result
async def optimize_rates(self):
"""Check if rate mode should be switched."""
rates = await self.get_current_rates("USDT")
if not self.position:
return
switcher = RateSwitcher(self.pool, self.wallet.address, USDT_ADDRESS, self.w3)
result = switcher.check_and_switch(
current_variable_rate=rates["variable"],
current_stable_rate=rates["stable"],
threshold_bps=75
)
if result["switched"]:
log.info(f"Switched rate mode, saving {result['savings_bps']} bps")
async def run_liquidation_scan(self):
"""Scan for and execute liquidation opportunities."""
if not self.cfg.enable_liquidations or not self.liquidation_bot:
return
borrowers = await self.get_recent_borrowers_from_subgraph()
targets = await self.liquidation_bot.find_liquidatable_positions(borrowers)
for target in targets[:2]:
profit_est = self.liquidation_bot.estimate_profit(
"USDC", "WETH", target["debt_usd"] * 0.5,
current_eth_price, 1.0
)
if profit_est >= self.cfg.min_liquidation_profit:
result = await self.liquidation_bot.execute_flash_liquidation(
target["user"], USDC_ADDRESS, WETH_ADDRESS,
int(target["debt_usd"] * 0.5 * 1e6)
)
if result.get("executed"):
self.stats["liquidations_executed"] += 1
self.stats["liquidation_profit"] += profit_est
async def run(self):
"""Main agent loop."""
log.info(f"AaveAgent starting | capital: ${self.cfg.capital_usdc:,.0f}")
await self.enter_loop()
iteration = 0
while True:
try:
# Core loop every minute
await self.monitor_health()
# Rate optimization every 6 hours
if iteration % 360 == 0:
await self.optimize_rates()
# Liquidation scan every 15 seconds
if iteration % 1 == 0:
await self.run_liquidation_scan()
log.info(
f"HF={self.position.health_factor:.3f} | "
f"Liquidations={self.stats['liquidations_executed']} | "
f"Liq. profit=${self.stats['liquidation_profit']:.2f}"
)
iteration += 1
except Exception as e:
log.error(f"Loop error: {e}")
await asyncio.sleep(60)
# Helpers (stubs)
async def supply_usdc(self, amount: float): pass
async def borrow_usdt(self, amount: float): pass
async def partial_repay(self, amount: float): pass
async def emergency_unwind(self): pass
async def get_current_rates(self, asset: str) -> dict: return {"supply": 0.05, "borrow": 0.04, "variable": 0.04, "stable": 0.06}
async def get_recent_borrowers_from_subgraph(self) -> list: return []
async def send_tx(self, tx): pass
# Start agent
if __name__ == "__main__":
config = AaveConfig(
rpc_url="https://arb1.arbitrum.io/rpc",
private_key="0x...",
pool_address="0x794a61358D6845594F94dc1DB02A252b5b4814aD",
capital_usdc=50_000,
enable_liquidations=True
)
agent = AaveAgent(config)
asyncio.run(agent.run())Combine Aave Strategies with Purple Flea Trading
Use Aave to borrow stablecoins and deploy them in Purple Flea's perpetual futures markets for additional yield on top of your lending loop.
Explore Purple Flea Read the Docs9. Key Risk Parameters and Safe Operating Limits
Professional agents running Aave strategies should enforce hard limits to prevent catastrophic losses:
SAFE_OPERATING_LIMITS = {
# Health factor targets
"min_health_factor": 1.20, # Hard floor — emergency deleverage
"target_health_factor": 1.50, # Steady-state target
"conservative_health": 2.00, # For volatile collateral
# Leverage limits
"max_leverage_stablecoin": 5.0, # Lower than theoretical max
"max_leverage_eth": 3.0,
"max_leverage_btc": 2.5,
# Rate limits
"max_borrow_utilization": 0.85, # Don't borrow when utilization > 85%
"rate_spread_minimum": 0.005, # At least 0.5% spread supply - borrow
# Capital limits
"max_pct_single_asset": 0.40, # No more than 40% in one asset
"max_pct_single_chain": 0.60, # No more than 60% on one chain
# Liquidation hunting
"min_debt_to_liquidate_usd": 500, # Skip tiny positions
"max_gas_per_liquidation": 50.0, # Max $50 gas per liquidation
}
def validate_strategy(params: dict) -> list[str]:
"""Validate strategy parameters against safe limits. Returns list of warnings."""
warnings = []
for key, limit in SAFE_OPERATING_LIMITS.items():
if key in params:
if "max" in key and params[key] > limit:
warnings.append(f"WARN: {key}={params[key]} exceeds limit {limit}")
if "min" in key and params[key] < limit:
warnings.append(f"WARN: {key}={params[key]} below minimum {limit}")
return warnings