Liquidation Helper
Find liquidatable positions and execute profitable liquidations on Scallop.
Overview
Liquidation occurs when an obligation's risk level reaches >= 1.0. Liquidators can repay debt and receive collateral at a discount (liquidation bonus).
Liquidation Mechanics
When Can Liquidation Occur?
risk_level = total_debt_value / (total_collateral_value * collateral_factor)
If risk_level >= 1.0 → Position is liquidatable
Liquidation Bonus
Liquidators receive collateral at a discount — e.g. repay $100 of USDC debt → receive ~$105 worth of SUI collateral. The bonus is a per-asset protocol parameter (5% SUI, 3% USDC, 5% wETH/wBTC) and can be changed by governance, so read it from protocol state rather than hardcoding it.
Max Liquidation Amount
- Close Factor: ~50% of debt can be liquidated per transaction
- Multiple transactions may be needed for full liquidation
Each transaction re-evaluates the position against the prices posted in it, so a position measured as liquidatable off-chain can be healthy by execution time and the call will fail. Full mechanics: liquidation-flow.md.
Find Liquidatable Positions
Check Specific Obligation
query = client.create_query()
# Check a known obligation for liquidation eligibility
ob = query.get_obligation_with_prices(obligation_id)
if ob.is_liquidatable:
print(f"Obligation {ob.obligation_id} is liquidatable!")
print(f"Risk Level: {ob.risk_level}")
print(f"Total Debt: ${ob.total_debt_usd:,.2f}")
print(f"Total Collateral: ${ob.total_collateral_usd:,.2f}")
Monitor Known Obligations
Monitor a list of obligation IDs for liquidation opportunities:
def monitor_for_liquidations(client, obligation_ids, min_profit=10):
"""Monitor known obligations approaching liquidation."""
query = client.create_query()
while True:
for ob_id in obligation_ids:
ob = query.get_obligation_with_prices(ob_id)
if ob.is_liquidatable:
profit = calculate_profit(ob)
if profit >= min_profit:
print(f"💰 Liquidation opportunity!")
print(f" Obligation: {ob.obligation_id}")
print(f" Est. Profit: ${profit:,.2f}")
time.sleep(1) # Check frequently
Note: The Python SDK does not have a bulk query for all liquidatable obligations. You must track obligation IDs externally (e.g., from on-chain events or indexer).
Profitability Calculation
def calculate_liquidation_profit(obligation, repay_amount, debt_coin, collateral_coin):
"""Calculate expected profit from liquidation."""
# Get prices
debt_price = query.get_coin_price(debt_coin)
collateral_price = query.get_coin_price(collateral_coin)
# Get liquidation bonus for collateral
bonus = get_liquidation_bonus(collateral_coin) # e.g., 1.05 for 5%
# Calculate collateral received
repay_value_usd = repay_amount * debt_price
collateral_received_usd = repay_value_usd * bonus
# Profit = collateral value - repay cost - gas
estimated_gas = 0.01 # ~0.01 SUI
gas_cost_usd = estimated_gas * query.get_coin_price("sui")
profit = collateral_received_usd - repay_value_usd - gas_cost_usd
return {
'repay_amount': repay_amount,
'repay_value_usd': repay_value_usd,
'collateral_received_usd': collateral_received_usd,
'gas_cost_usd': gas_cost_usd,
'net_profit_usd': profit,
'roi_percent': (profit / repay_value_usd) * 100
}
This is a gross estimate. It omits exit slippage (usually the largest cost), gas burnt on lost races, flash loan fees, and price drift between decision and execution — so do not gate execution on
net_profit_usd > 0. See profitability.md.
Execute Liquidation
Using Quick Method
tx = builder.create_tx_block()
# Liquidate: repay USDC debt, receive SUI collateral
# Returns tuple[TxResult, TxResult]: (remaining_debt, collateral)
remaining_debt_idx, collateral_idx = tx.liquidate_quick(
obligation_id=target_obligation_id,
repay_amount=100_000_000, # 100 USDC
debt_coin_name="usdc",
collateral_coin_name="sui",
sender=wallet_address,
)
tx.transfer_objects([remaining_debt_idx, collateral_idx], wallet_address)
result = builder.sign_and_send_tx_block(tx)
print(f"Liquidation TX: {result.digest}")
Using Flash Loan for Capital Efficiency
tx = builder.create_tx_block()
# 1. Borrow USDC via flash loan (no collateral needed!)
usdc_coin, flash_receipt = tx.borrow_flash_loan(
amount=100_000_000,
coin_name="usdc"
)
# 2. Liquidate using borrowed USDC
# Returns tuple[TxResult, TxResult]: (remaining_debt, collateral)
remaining_debt, collateral = tx.liquidate(
obligation_id=target_obligation_id,
repay_coin_idx=usdc_coin,
debt_coin_name="usdc",
collateral_coin_name="sui",
)
# 3. Swap some collateral back to USDC (via DEX)
# ... DEX swap logic ...
# 4. Repay flash loan
tx.repay_flash_loan(usdc_for_repay, flash_receipt, "usdc")
# 5. Keep profit (remaining collateral + remaining debt coin)
tx.transfer_objects([remaining_collateral], wallet_address)
result = builder.sign_and_send_tx_block(tx)
Liquidation Bot Template
import time
from sui_scallop_sdk import ScallopClient
class LiquidationBot:
def __init__(self, secret_key, network="mainnet"):
self.client = ScallopClient(secret_key=secret_key, network=network)
self.query = self.client.create_query()
self.builder = self.client.create_builder()
self.min_profit_usd = 5.0
def run(self):
print("🤖 Liquidation bot started...")
while True:
try:
opportunities = self.find_opportunities()
for opp in opportunities:
if opp['profit'] >= self.min_profit_usd:
self.execute_liquidation(opp)
except Exception as e:
print(f"Error: {e}")
time.sleep(1)
def find_opportunities(self):
"""Find profitable liquidation opportunities."""
opportunities = []
# Check tracked obligations (must be populated externally)
for ob_id in self.tracked_obligations:
ob = self.query.get_obligation_with_prices(ob_id)
if not ob.is_liquidatable:
continue
for debt in ob.debts:
for collateral in ob.collaterals:
profit_info = calculate_liquidation_profit(
ob,
debt.amount * 0.5, # Max 50%
debt.coin_name,
collateral.coin_name
)
if profit_info['net_profit_usd'] > 0:
opportunities.append({
'obligation_id': ob.obligation_id,
'debt_coin': debt.coin_name,
'collateral_coin': collateral.coin_name,
'repay_amount': profit_info['repay_amount'],
'profit': profit_info['net_profit_usd']
})
# Sort by profit descending
opportunities.sort(key=lambda x: x['profit'], reverse=True)
return opportunities
def execute_liquidation(self, opportunity):
"""Execute a liquidation."""
print(f"Executing liquidation: {opportunity}")
tx = self.builder.create_tx_block()
remaining_debt, collateral = tx.liquidate_quick(
obligation_id=opportunity['obligation_id'],
repay_amount=opportunity['repay_amount'],
debt_coin_name=opportunity['debt_coin'],
collateral_coin_name=opportunity['collateral_coin'],
sender=self.client.wallet_address,
)
tx.transfer_objects([remaining_debt, collateral], self.client.wallet_address)
result = self.builder.sign_and_send_tx_block(tx)
print(f"✅ Liquidation successful: {result.digest}")
print(f" Est. Profit: ${opportunity['profit']:,.2f}")
# Run bot
if __name__ == "__main__":
bot = LiquidationBot(secret_key="...")
bot.run()
Safety Considerations
- Check Position First: Verify position is actually liquidatable
- Price Freshness: Ensure oracle prices are current
- Gas Estimation: Account for gas costs
- Competition: Other liquidators may front-run
- Capital Requirements: Have enough to repay debt
Error Handling
| Error | Cause | Solution |
|---|---|---|
NotLiquidatable |
Position is healthy | Risk level must be >= 1.0 |
LiquidationAmountTooHigh |
Exceeds close factor | Reduce repay amount |
InsufficientBalance |
Not enough to repay | Ensure sufficient balance |
References
- Liquidation Flow - Step-by-step process
- Profitability - Profit calculations
- Error Codes - Error reference