DeFi Liquidation Alerts: Real-Time Risk Monitoring with SQL

DeFi liquidation alerts give lenders and borrowers early warning before a position's health factor crosses the liquidation threshold — often preventing losses that would otherwise happen within a single block. With RisingWave, a PostgreSQL-compatible streaming database, you can monitor collateral ratios across Aave, Compound, and other protocols in real time using SQL materialized views that update with every new on-chain event.

Why Liquidation Risk Matters in DeFi

In over-collateralized lending protocols, a borrower deposits collateral — say ETH — to borrow a stablecoin like USDC. The protocol tracks a health factor: a ratio that compares the value of collateral to the outstanding debt. When the health factor drops below 1.0, the position becomes eligible for liquidation.

Liquidations happen fast. A sharp ETH price drop can push thousands of positions below threshold in seconds. For borrowers, late awareness means losing collateral at a discount. For liquidators and risk systems, missing the window means lost revenue. For protocol risk teams, unmonitored concentration of undercollateralized positions can threaten protocol solvency.

The problem with traditional monitoring:

• Batch queries against an archive node run every few minutes — far too slow when a position can go from healthy to liquidated in under 60 seconds
• Point-in-time snapshots miss rapid price oscillations that briefly dip below threshold
• Manual dashboards require constant attention and can't trigger automated responses

How Streaming SQL Solves This

RisingWave ingests a continuous stream of protocol events — position updates, price oracle updates, borrow and repay events — and maintains continuously updated materialized views of every position's current health factor. Alert views filter for positions approaching the liquidation threshold, and sinks push those alerts to downstream systems without any polling.

Building It Step by Step

Step 1: Connect the Data Source

-- Position and price events from on-chain indexer via Kafka
CREATE SOURCE defi_events (
    event_id        VARCHAR,
    protocol        VARCHAR,          -- 'aave_v3', 'compound_v3'
    event_type      VARCHAR,          -- 'deposit', 'borrow', 'repay', 'price_update'
    wallet_address  VARCHAR,
    asset           VARCHAR,
    collateral_usd  NUMERIC,
    debt_usd        NUMERIC,
    health_factor   NUMERIC,
    ltv             NUMERIC,
    liquidation_threshold NUMERIC,
    price_usd       NUMERIC,
    block_number    BIGINT,
    event_time      TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'defi.position.events',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

Step 2: Build the Core Materialized View

-- Latest position state per wallet per protocol
CREATE MATERIALIZED VIEW position_health AS
SELECT DISTINCT ON (protocol, wallet_address)
    protocol,
    wallet_address,
    asset,
    collateral_usd,
    debt_usd,
    health_factor,
    ltv,
    liquidation_threshold,
    (liquidation_threshold - ltv)   AS ltv_buffer,
    event_time                       AS last_updated,
    block_number
FROM defi_events
WHERE event_type IN ('deposit', 'borrow', 'repay', 'price_update')
  AND health_factor IS NOT NULL
ORDER BY protocol, wallet_address, event_time DESC;

-- Risk tier distribution across protocols
CREATE MATERIALIZED VIEW risk_distribution AS
SELECT
    protocol,
    COUNT(*) FILTER (WHERE health_factor >= 2.0)          AS safe_count,
    COUNT(*) FILTER (WHERE health_factor BETWEEN 1.2 AND 2.0) AS watch_count,
    COUNT(*) FILTER (WHERE health_factor BETWEEN 1.0 AND 1.2) AS danger_count,
    COUNT(*) FILTER (WHERE health_factor < 1.0)           AS liquidatable_count,
    SUM(debt_usd) FILTER (WHERE health_factor < 1.2)      AS at_risk_debt_usd,
    MAX(event_time)                                        AS last_updated
FROM position_health
GROUP BY protocol;

Step 3: Add Alerts and Detection Logic

-- Positions approaching liquidation (health factor < 1.15)
CREATE MATERIALIZED VIEW liquidation_alerts AS
SELECT
    protocol,
    wallet_address,
    asset,
    collateral_usd,
    debt_usd,
    health_factor,
    liquidation_threshold,
    ltv,
    ltv_buffer,
    CASE
        WHEN health_factor < 1.0  THEN 'LIQUIDATABLE'
        WHEN health_factor < 1.05 THEN 'CRITICAL'
        WHEN health_factor < 1.1  THEN 'HIGH'
        ELSE                           'WATCH'
    END                             AS risk_level,
    last_updated
FROM position_health
WHERE health_factor < 1.15;

-- Sink critical alerts to Kafka
CREATE SINK liquidation_alert_sink AS
SELECT * FROM liquidation_alerts
WHERE risk_level IN ('LIQUIDATABLE', 'CRITICAL')
WITH (
    connector = 'kafka',
    topic = 'defi.alerts.liquidation',
    properties.bootstrap.server = 'kafka:9092'
) FORMAT PLAIN ENCODE JSON;

-- Track large positions with tight LTV buffers
CREATE MATERIALIZED VIEW whale_risk_positions AS
SELECT
    protocol,
    wallet_address,
    collateral_usd,
    debt_usd,
    health_factor,
    ltv_buffer,
    last_updated
FROM position_health
WHERE debt_usd > 100000
  AND ltv_buffer < 0.05
ORDER BY debt_usd DESC;

Step 4: Query for Real-Time Insights

-- All positions currently eligible for liquidation
SELECT
    protocol,
    wallet_address,
    asset,
    ROUND(collateral_usd, 2)        AS collateral_usd,
    ROUND(debt_usd, 2)              AS debt_usd,
    ROUND(health_factor, 4)         AS health_factor,
    last_updated
FROM liquidation_alerts
WHERE risk_level = 'LIQUIDATABLE'
ORDER BY debt_usd DESC
LIMIT 10;

-- Protocol-level risk summary
SELECT
    protocol,
    safe_count,
    watch_count,
    danger_count,
    liquidatable_count,
    ROUND(at_risk_debt_usd, 0)      AS at_risk_debt_usd
FROM risk_distribution
ORDER BY at_risk_debt_usd DESC;

Comparison: Batch vs Streaming

FAQ

Q: How do I keep collateral prices current without a separate price feed join?

The recommended approach is to include price oracle update events in the same Kafka topic (event_type = 'price_update'). Your on-chain indexer emits a position recalculation event every time the oracle updates. RisingWave then processes these events in order, and the position_health materialized view always reflects the latest health factor computed with the current oracle price.

Q: Can this work with multiple collateral assets per position?

Yes. Protocols like Aave V3 allow multi-asset collateral. Extend the source schema to include per-asset breakdowns and emit one event per asset update. The materialized view can then aggregate across assets using SUM(collateral_usd) grouped by wallet_address and protocol, giving an accurate combined health factor per position.

Q: What happens if there is a brief network partition and events arrive late?

RisingWave's watermark mechanism handles late-arriving events. For position state views using DISTINCT ON ordered by event_time, a late event with an earlier timestamp won't override a newer state. For windowed aggregations tracking protocol-level risk, configure an appropriate allowed lateness to absorb brief gaps without incorrect window closures.

Key Takeaways

• A single Kafka source captures deposit, borrow, repay, and price oracle events from multiple DeFi protocols
• The position_health materialized view maintains the latest health factor and LTV for every active position
• Risk tier views give protocol teams an instant overview of collateralization across their user base
• Positions crossing the 1.15 health factor threshold trigger alerts published to Kafka within milliseconds
• Whale position tracking identifies the largest at-risk accounts for prioritized liquidation or outreach

Ready to try this? Get started with RisingWave. Join our Slack community.