Skip to main content

DeFi Liquidations API

Monitor lending protocol positions and detect liquidation opportunities in real-time. The DeFi API provides position health tracking, opportunity filtering, and WebSocket streaming for liquidation bots.

Key Features​

  • Position health monitoring - Track health factors across protocols
  • Opportunity detection - Find profitable liquidations
  • Protocol coverage - Aave, Compound, Morpho, Spark, and more
  • Profit estimation - Calculate expected profit including gas
  • WebSocket streaming - Real-time opportunity alerts

Supported Chains & Protocols​

Chains​

Ethereum, Optimism, Arbitrum, Polygon, Base, Gnosis

Protocols​

ProtocolChainsLiquidation Bonus
Aave V2ethereum, polygon5-15%
Aave V3ethereum, polygon, arbitrum, optimism, base4-10%
Compound V2ethereum5-8%
Compound V3ethereum, polygon, arbitrum5%
MorphoethereumVariable
Sparkethereum4-10%

Health Status​

Positions are classified by health factor (HF):

StatusHealth FactorRiskAction
HEALTHYHF > 1.5LowNo action needed
WARNING1.2 < HF <= 1.5MediumMonitor closely
AT_RISK1 < HF <= 1.2HighApproaching liquidation
LIQUIDATABLEHF <= 1CriticalCan be liquidated

REST Endpoints​

Get Position Health​

GET /v1/defi/{chain}/positions/{address}/health

Returns lending position health for an address across all protocols.

Request​

curl -H "X-API-Key: YOUR_KEY" \
  "https://api.axol.io/api/v1/defi/ethereum/positions/0x123.../health"

Response​

{
  "address": "0x123...",
  "chain": "ethereum",
  "positions": [
    {
      "protocol": "aave_v3",
      "health_factor": 1.35,
      "status": "WARNING",
      "total_collateral_usd": 150000,
      "total_debt_usd": 100000,
      "available_to_borrow_usd": 10000,
      "net_apy": 2.5,
      "collateral_assets": [
        {
          "token": "WETH",
          "amount": 50.0,
          "usd_value": 150000,
          "ltv": 0.8,
          "liquidation_threshold": 0.85
        }
      ],
      "debt_assets": [
        {
          "token": "USDC",
          "amount": 100000,
          "usd_value": 100000,
          "borrow_apy": 4.5
        }
      ]
    }
  ],
  "at_risk": true,
  "total_collateral_usd": 150000,
  "total_debt_usd": 100000
}

Query Parameters​

ParameterTypeDescription
protocolstringFilter to specific protocol (aave_v2, aave_v3, compound_v2, compound_v3)

Get Liquidation Opportunities​

GET /v1/defi/{chain}/liquidations/opportunities

Returns positions that can be profitably liquidated.

Request​

curl -H "X-API-Key: YOUR_KEY" \
  "https://api.axol.io/api/v1/defi/ethereum/liquidations/opportunities?min_profit_usd=100&protocol=aave_v3"

Response​

{
  "chain": "ethereum",
  "count": 5,
  "opportunities": [
    {
      "borrower": "0xabc...",
      "protocol": "aave_v3",
      "health_factor": 0.95,
      "total_debt_usd": 50000,
      "total_collateral_usd": 45000,
      "max_liquidatable_debt_usd": 25000,
      "collateral_to_receive": {
        "token": "WETH",
        "amount": 8.5,
        "usd_value": 25500
      },
      "debt_to_repay": {
        "token": "USDC",
        "amount": 25000,
        "usd_value": 25000
      },
      "liquidation_bonus_percent": 5,
      "gross_profit_usd": 500,
      "estimated_gas_usd": 50,
      "net_profit_usd": 450,
      "flash_loan_available": true,
      "timestamp": "2026-01-15T10:30:00Z"
    }
  ],
  "total_profit_usd": 2500
}

Query Parameters​

ParameterTypeDescription
limitintMax opportunities (1-500, default 100)
min_profit_usdfloatMinimum net profit
min_collateral_usdfloatMinimum collateral value
protocolstringFilter by protocol

Get Protocol Statistics​

GET /v1/defi/{chain}/protocols/stats

Returns TVL, utilization, and position counts.

Response​

{
  "chain": "ethereum",
  "protocols": [
    {
      "protocol": "aave_v3",
      "tvl_usd": 5000000000,
      "total_borrowed_usd": 2500000000,
      "utilization_rate": 0.5,
      "total_positions": 150000,
      "at_risk_positions": 1200,
      "liquidatable_positions": 45
    }
  ]
}

Get Supported Protocols​

GET /v1/defi/{chain}/protocols

Returns list of supported protocols for a chain.

{
  "chain": "ethereum",
  "protocols": ["aave_v2", "aave_v3", "compound_v2", "compound_v3", "morpho", "spark"]
}

WebSocket Streaming​

Connect​

wss://api.axol.io/api/v1/defi/{chain}/liquidations/stream

Subscribe​

{
  "action": "subscribe",
  "filter": {
    "min_profit_usd": 100,
    "protocols": ["aave_v3", "compound_v3"]
  }
}

Messages​

Connected​

{
  "type": "connected",
  "chain": "ethereum",
  "protocols": ["aave_v2", "aave_v3", "compound_v2", "compound_v3"]
}

Opportunity​

{
  "type": "opportunity",
  "data": {
    "borrower": "0xabc...",
    "protocol": "aave_v3",
    "health_factor": 0.98,
    "net_profit_usd": 250,
    "debt_to_repay": {"token": "USDC", "amount": 10000},
    "collateral_to_receive": {"token": "WETH", "amount": 3.5},
    "timestamp": "2026-01-15T10:30:00Z"
  }
}

Code Examples​

Python - Liquidation Bot​

import asyncio
import aiohttp
import websockets
import json
import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class LiquidationConfig:
    api_key: str
    chain: str = "ethereum"
    min_profit_usd: float = 100
    protocols: list[str] = None
    base_url: str = "https://api.axol.io/api/v1"

class LiquidationBot:
    def __init__(self, config: LiquidationConfig):
        self.config = config
        self.headers = {"X-API-Key": config.api_key}

    async def get_opportunities(self) -> list[dict]:
        """Fetch current liquidation opportunities."""
        params = {
            "min_profit_usd": self.config.min_profit_usd,
            "limit": 100
        }
        if self.config.protocols:
            params["protocol"] = self.config.protocols[0]

        async with aiohttp.ClientSession() as session:
            url = f"{self.config.base_url}/defi/{self.config.chain}/liquidations/opportunities"
            async with session.get(url, headers=self.headers, params=params) as resp:
                data = await resp.json()
                return data.get("opportunities", [])

    async def check_position(self, address: str) -> dict:
        """Check health of a specific position."""
        async with aiohttp.ClientSession() as session:
            url = f"{self.config.base_url}/defi/{self.config.chain}/positions/{address}/health"
            async with session.get(url, headers=self.headers) as resp:
                return await resp.json()

    async def should_liquidate(self, opportunity: dict) -> bool:
        """Determine if opportunity is still valid."""
        # Re-check position health
        position = await self.check_position(opportunity["borrower"])

        for p in position.get("positions", []):
            if p["protocol"] == opportunity["protocol"]:
                # Still liquidatable?
                return p["health_factor"] <= 1.0

        return False

    async def execute_liquidation(self, opportunity: dict) -> Optional[str]:
        """Execute the liquidation (implement your logic)."""
        print(f"Executing liquidation:")
        print(f"  Borrower: {opportunity['borrower']}")
        print(f"  Protocol: {opportunity['protocol']}")
        print(f"  Profit: ${opportunity['net_profit_usd']:.2f}")

        # Your liquidation execution logic here
        # This typically involves:
        # 1. Flash loan the debt token
        # 2. Call liquidation function on protocol
        # 3. Receive collateral
        # 4. Swap collateral to repay flash loan
        # 5. Keep profit

        return None  # Return tx hash

    async def stream_opportunities(self):
        """Stream liquidation opportunities via WebSocket."""
        uri = f"wss://api.axol.io/api/v1/defi/{self.config.chain}/liquidations/stream"

        while True:
            try:
                async with websockets.connect(
                    uri,
                    extra_headers=self.headers,
                    ping_interval=30
                ) as ws:
                    # Subscribe
                    await ws.send(json.dumps({
                        "action": "subscribe",
                        "filter": {
                            "min_profit_usd": self.config.min_profit_usd,
                            "protocols": self.config.protocols
                        }
                    }))

                    async for message in ws:
                        data = json.loads(message)

                        if data["type"] == "opportunity":
                            opp = data["data"]
                            print(f"\nNew opportunity: ${opp['net_profit_usd']:.2f}")

                            if await self.should_liquidate(opp):
                                await self.execute_liquidation(opp)

            except Exception as e:
                print(f"Connection error: {e}")
                await asyncio.sleep(5)

    async def run(self):
        """Run the liquidation bot."""
        print(f"Starting liquidation bot on {self.config.chain}")
        print(f"Min profit: ${self.config.min_profit_usd}")

        # Stream opportunities
        await self.stream_opportunities()

# Usage
config = LiquidationConfig(
    api_key=os.getenv("AXOL_API_KEY"),
    chain="ethereum",
    min_profit_usd=100,
    protocols=["aave_v3", "compound_v3"]
)

bot = LiquidationBot(config)
asyncio.run(bot.run())

TypeScript - Liquidation Monitor​

import WebSocket from 'ws';
import fetch from 'node-fetch';

interface LiquidationOpportunity {
  borrower: string;
  protocol: string;
  health_factor: number;
  net_profit_usd: number;
  debt_to_repay: { token: string; amount: number };
  collateral_to_receive: { token: string; amount: number };
}

class LiquidationMonitor {
  private apiKey: string;
  private chain: string;
  private baseUrl: string;
  private minProfit: number;

  constructor(
    apiKey: string,
    chain = 'ethereum',
    minProfit = 100,
    baseUrl = 'https://api.axol.io/api/v1'
  ) {
    this.apiKey = apiKey;
    this.chain = chain;
    this.baseUrl = baseUrl;
    this.minProfit = minProfit;
  }

  async getOpportunities(): Promise<LiquidationOpportunity[]> {
    const url = `${this.baseUrl}/defi/${this.chain}/liquidations/opportunities?min_profit_usd=${this.minProfit}`;
    const response = await fetch(url, {
      headers: { 'X-API-Key': this.apiKey }
    });
    const data = await response.json() as any;
    return data.opportunities || [];
  }

  streamOpportunities(callback: (opp: LiquidationOpportunity) => void): void {
    const connect = () => {
      const ws = new WebSocket(
        `wss://api.axol.io/api/v1/defi/${this.chain}/liquidations/stream`,
        { headers: { 'X-API-Key': this.apiKey } }
      );

      ws.on('open', () => {
        console.log('Connected to liquidation stream');
        ws.send(JSON.stringify({
          action: 'subscribe',
          filter: { min_profit_usd: this.minProfit }
        }));
      });

      ws.on('message', (data) => {
        const msg = JSON.parse(data.toString());
        if (msg.type === 'opportunity') {
          callback(msg.data);
        }
      });

      ws.on('close', () => {
        console.log('Connection closed, reconnecting...');
        setTimeout(connect, 5000);
      });

      ws.on('error', (error) => {
        console.error('WebSocket error:', error);
      });
    };

    connect();
  }
}

// Usage
const monitor = new LiquidationMonitor(process.env.AXOL_API_KEY!);

monitor.streamOpportunities((opp) => {
  console.log(`Opportunity: $${opp.net_profit_usd.toFixed(2)} profit`);
  console.log(`  Borrower: ${opp.borrower}`);
  console.log(`  Protocol: ${opp.protocol}`);
  console.log(`  Health: ${opp.health_factor}`);
});

Python - Position Health Monitor​

import asyncio
import aiohttp
from typing import Callable

class PositionMonitor:
    def __init__(
        self,
        api_key: str,
        address: str,
        chain: str = "ethereum",
        alert_threshold: float = 1.3
    ):
        self.api_key = api_key
        self.address = address
        self.chain = chain
        self.alert_threshold = alert_threshold
        self.base_url = "https://api.axol.io/api/v1"

    async def check_health(self) -> dict:
        """Check position health."""
        headers = {"X-API-Key": self.api_key}
        url = f"{self.base_url}/defi/{self.chain}/positions/{self.address}/health"

        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers) as resp:
                return await resp.json()

    async def monitor(
        self,
        callback: Callable[[dict], None],
        interval: int = 60
    ):
        """Continuously monitor position health."""
        while True:
            try:
                health = await self.check_health()

                for position in health.get("positions", []):
                    hf = position["health_factor"]

                    if hf < self.alert_threshold:
                        callback({
                            "type": "warning" if hf > 1.0 else "critical",
                            "health_factor": hf,
                            "protocol": position["protocol"],
                            "collateral_usd": position["total_collateral_usd"],
                            "debt_usd": position["total_debt_usd"]
                        })

            except Exception as e:
                print(f"Error checking health: {e}")

            await asyncio.sleep(interval)

# Usage
def on_alert(alert: dict):
    print(f"ALERT [{alert['type'].upper()}]: HF={alert['health_factor']:.2f}")

monitor = PositionMonitor(
    api_key=os.getenv("AXOL_API_KEY"),
    address="0x123...",
    alert_threshold=1.3
)

asyncio.run(monitor.monitor(on_alert, interval=30))

Best Practices​

1. Verify Before Executing​

Always re-check position health before liquidating:

opportunity = await get_opportunity()
current_health = await check_position(opportunity["borrower"])

if current_health["health_factor"] > 1.0:
    print("Position recovered, skipping")
    return

2. Consider Gas Costs​

Factor in current gas prices:

gas = await oracle.get_gas("ethereum")
gas_cost = gas["fast"]["max_fee_gwei"] * LIQUIDATION_GAS_ESTIMATE * eth_price / 1e9

if opportunity["gross_profit_usd"] - gas_cost < MIN_PROFIT:
    print("Not profitable after gas")
    return

3. Use Flash Loans​

For larger liquidations, use flash loans to avoid capital requirements:

if opportunity["flash_loan_available"]:
    # Execute with flash loan
    await execute_with_flash_loan(opportunity)
else:
    # Need to have capital on hand
    await execute_with_capital(opportunity)

4. Handle Competition​

Other bots are watching the same opportunities:

# Use higher gas to beat competition
gas_price = gas["instant"]["max_fee_gwei"] * 1.1

# Or use Flashbots to avoid gas wars
await submit_via_flashbots(tx)

See Also​