Skip to content

Fix/liquidation price #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tbssajal wants to merge 14 commits into
base: master
Choose a base branch
from
Open

Fix/liquidation price #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
3dee212
use specific python version
tbssajal Dec 29, 2025
c95859c
fix price calculation
tbssajal Dec 29, 2025
2f39bc8
fix issues
tbssajal Dec 29, 2025
b450262
fix bug
tbssajal Dec 29, 2025
d9a5cf8
use different price diff for short long to always yield bid
tbssajal Dec 29, 2025
ab16575
fix bug
tbssajal Dec 30, 2025
ca09bc7
fix precision issue
tbssajal Dec 30, 2025
2489292
use Decimal for MAE
tbssajal Dec 31, 2025
74d8c9f
convert properly
tbssajal Dec 31, 2025
89c9011
wait before mae check
tbssajal Dec 31, 2025
d62f0e0
fix bug
tbssajal Dec 31, 2025
f4be447
fix notifier bug
tbssajal Dec 31, 2025
d6216a1
fix bug
tbssajal Dec 31, 2025
51a7b7e
use logs properly
tbssajal Dec 31, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 42 additions & 21 deletions liquidation/__main__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,14 +52,19 @@ def main():
strategy = FullLiquidationPercentMAEStrategy(DMAE, reseller.validate_position)

logger.info("Found %d liquidatable accounts", len(accounts))
liquidated: list[LiquidatingAccountContext] = []
for account in accounts:
logger.info("%s - processing Liquidation", account.account_id)
account.populate()
logger.info(
"%s - number of positions %d", account.account_id, len(account.positions)
)

liquidation_started, dmae, dmmu = account.start_liquidation(strategy)
liquidation_started, request_failed, dmae, dmmu = account.start_liquidation(strategy)
if request_failed:
logger.info("%s - cannot initiate liquidation, skipping...", account.account_id)
continue

if liquidation_started:
logger.info("%s - requested liquidation", account.account_id)
else:
Expand Down Expand Up @@ -95,9 +100,28 @@ def main():
account.account_id,
format_int(mae_offered, account.collateral_decimals),
)
bids = FullLiquidationPercentMAEStrategy(
DMAE, reseller.validate_position
).construct_bids(account.positions)
skip, blocks_to_wait = account.wait_time_for(dmae, dmmu, mae_offered, account.auction_data.mmu_now)
if skip:
logger.info("%s - Delta MAE is too big", account.account_id)
continue
bid_possible, bids = strategy.construct_bids(
format_int(
account.auction_data.mae_at_initiation,
account.collateral_decimals
),
account.positions
)
if (not bid_possible) or (len(bids) == 0):
logger.info("%s - no valid bid constructed, skipping...", account.account_id)
continue
logger.info(
"%s - waiting for %s blocks before liquidation",
account.account_id,
blocks_to_wait,
)
## wait an extra block to ensure we are clear
wait_for_blocks(w3, blocks_to_wait + 1)

mae_check_failed, mae_over_mmu_exceeded = clearing.mae_check_on_bid(
w3.eth.default_account,
account.account_id,
Expand All @@ -112,40 +136,37 @@ def main():
mae_over_mmu_exceeded,
)
continue
blocks_to_wait = account.wait_time_for(dmae, dmmu)
logger.info(
"%s - waiting for %s before liquidation",
account.account_id,
blocks_to_wait,
)
## wait an extra block to ensure we are clear
wait_for_blocks(w3, blocks_to_wait + 1)
account.bid_liquidation(strategy)
if account.bid_liquidation(strategy):
liquidated.append(account)

if len(accounts) > 0:
content = f"Liquidation processed for {len(accounts)} margin accounts"
if len(liquidated) > 0:
content = f"Liquidation processed for {len(liquidated)} margin accounts"
notify_data = [
NotificationItem(
title=f"Account {account.account_id} liquidated",
values={
"Account ID": account.account_id,
"Collateral Asset": account.collateral_asset,
"MMU (before)": Decimal(account.auction_data.mae_at_initiation)
/ Decimal(10**account.collateral_decimals),
"MAE (before)": Decimal(account.auction_data.mae_at_initiation)
/ Decimal(10**account.collateral_decimals),
"MMU (before)": format_int(
account.auction_data.mmu_at_initiation,
account.collateral_decimals
),
"MAE (before)": format_int(
account.auction_data.mae_at_initiation,
account.collateral_decimals
),
"Positions": str(len(account.positions)),
},
)
for account in accounts
for account in liquidated
]
notifier.notify(
"Margin Accounts Liquidated",
content,
notify_data,
)

logger.info("Liquidation bids submitted for %d margin accounts", len(accounts))
logger.info("Liquidation bids submitted for %d margin accounts", len(liquidated))
# Here we check all margin accounts for positions, just in case some accounts were not liquidated
# in previous runs
reseller.populate()
Expand Down
105 changes: 89 additions & 16 deletions liquidation/bid.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from abc import ABC, abstractmethod
from decimal import Decimal
import math
from typing import Callable

import afp.bindings
Expand All @@ -18,14 +19,16 @@ class BidStrategy(ABC):
"""

@abstractmethod
def construct_bids(self, positions: list[Position]) -> list[afp.bindings.BidData]:
def construct_bids(self, mae_initial: Decimal, positions: list[Position]) -> (bool, list[afp.bindings.BidData]):
"""
Construct bids for the liquidating account based on its positions.

Args:
mae_initial (Decimal): Initial MAE of the account without the collateral token precision
positions (list[Position]): List of positions to construct bids for.

Returns:
bool: True if bid construction is possible, False otherwise
list[afp.bindings.BidData]: List of bid data objects.
"""
pass
Expand All @@ -46,23 +49,25 @@ def __init__(self, position_validator: Callable[[HexBytes], bool]):
"""
self.position_validator = position_validator

def construct_bids(self, positions: list[Position]) -> list[afp.bindings.BidData]:
def construct_bids(self, mae_initial: Decimal, positions: list[Position]) -> (bool, list[afp.bindings.BidData]):
"""
Construct bids for every position held by the liquidating account at the mark price.

Args:
mae_initial (Decimal): Initial MAE of the account without the collateral token precision
positions (list[Position]): List of positions to construct bids for.

Returns:
bool: True if bid construction is possible, False otherwise
list[afp.bindings.BidData]: List of bid data objects.
"""
bids = []
for position in positions:
if not self.position_validator(position.position_id):
continue
mark_price = position.mark_price
quantity = -position.quantity
side = afp.bindings.Side.BID if quantity < 0 else afp.bindings.Side.ASK
quantity = position.quantity
side = afp.bindings.Side.BID if quantity > 0 else afp.bindings.Side.ASK
bids.append(
afp.bindings.BidData(
product_id=position.position_id,
Expand All @@ -71,7 +76,7 @@ def construct_bids(self, positions: list[Position]) -> list[afp.bindings.BidData
side=side,
)
)
return bids
return True, bids


class FullLiquidationPercentMAEStrategy(BidStrategy):
Expand All @@ -82,6 +87,8 @@ class FullLiquidationPercentMAEStrategy(BidStrategy):
reducing the MAE of the liquidating account by percent_mae.
"""

percent_mae: Decimal

def __init__(
self, percent_mae: Decimal, position_validator: Callable[[HexBytes], bool]
):
Expand All @@ -93,40 +100,106 @@ def __init__(
self.percent_mae = percent_mae
self.position_validator = position_validator

def construct_bids(self, positions: list[Position]) -> list[afp.bindings.BidData]:
def construct_bids(self, mae_initial: Decimal, positions: list[Position]) -> (bool, list[afp.bindings.BidData]):
"""
Construct bids for full liquidation based on a percentage of the mark price.

Args:
mae_initial (Decimal): Initial MAE of the account without the collateral token precision
positions (list[Position]): List of positions to construct bids for.

Returns:
bool: True if bid construction is possible, False otherwise
list[afp.bindings.BidData]: List of bid data objects.
"""
bids = []
skip_bids = []
sum_notional_long = Decimal(0)
sum_notional_short = Decimal(0)
for position in positions:
if not self.position_validator(position.position_id):
skip_bids.append(True)
continue
if position.mark_price == 0:
# just a safety check for further calculation
raise RuntimeError("Got zero mark price for position ", position.position_id)

if position.quantity > 0:
sum_notional_long += position.notional_at_mark()
else:
sum_notional_short += position.notional_at_mark()

skip_bids.append(False)

# we calculate a constant difference percentage between mark price and bid price for all positions
# so that we take `mae_initial * percent_mae` amount of MAE from the liquidating account
#
# the equation we solve here is the following:
# `percent_dmark_long * sum_notional_long + percent_dmark_short * sum_notional_short >= dmae`
# we solve this assuming `percent_dmark_long = percent_dmark_short = percent_dmark`
sum_notional = sum_notional_long + sum_notional_short
dmae = mae_initial * self.percent_mae
percent_dmark = dmae / sum_notional

if percent_dmark > Decimal(1):
# for long positions, bid price cannot differ more than 100% from mark price
# we can use different percentage for long and short positions to compensate this
# we will take 100% for long, which means the bid price will be zero
for index, position in enumerate(positions):
if skip_bids[index]:
continue

if position.quantity < 0:
continue

bids.append(
afp.bindings.BidData(
product_id=position.position_id,
quantity=abs(position.quantity),
price=0,
side=afp.bindings.Side.BID,
)
)
skip_bids[index] = True

# for short, we will calculate the price difference percentage from here
# as we assumed `percent_dmark_long = 1`, we get `percent_dmark_short * sum_notional_short >= dmae - sum_notional_long`
# so we update `sum_notional` and `dmae` accordingly
sum_notional = sum_notional_short
dmae -= sum_notional_long
percent_dmark = dmae / sum_notional

for index, position in enumerate(positions):
if skip_bids[index]:
continue
mark_price = Decimal(position.mark_price) / Decimal(10**position.tick_size)
tick_size = position.tick_size
quantity = position.quantity
mark_price = position.mark_price
side = afp.bindings.Side.BID if quantity > 0 else afp.bindings.Side.ASK
bid_price = (
mark_price
* (1 - self.percent_mae / (abs(quantity) * position.point_value))
if quantity > 0
else mark_price
* (1 + self.percent_mae / (abs(quantity) * position.point_value))
math.floor(
Decimal(mark_price) * (1 - percent_dmark)
) if quantity > 0
else math.ceil(
Decimal(mark_price) * (1 + percent_dmark)
)
)
bids.append(
afp.bindings.BidData(
product_id=position.position_id,
quantity=abs(quantity),
price=parse_decimal(bid_price, tick_size),
price=bid_price,
side=side,
)
)
return bids

# update `dmae` and `sum_notional` and calculate `percent_dmark` again
dmae -= position.dmae(bid_price)
sum_notional -= position.notional_at_mark()
if dmae < Decimal(0):
percent_dmark = Decimal(0)
elif sum_notional > Decimal(0):
percent_dmark = dmae / sum_notional
return True, bids


class OrderedPercentMAEStrategy(BidStrategy):
Expand All @@ -135,6 +208,6 @@ class OrderedPercentMAEStrategy(BidStrategy):
def __init__(self, percent_mae: Decimal):
self.percent_mae = percent_mae

def construct_bids(self, positions: list[Position]) -> list[afp.bindings.BidData]:
def construct_bids(self, mae_initial: Decimal, positions: list[Position]) -> (bool, list[afp.bindings.BidData]):
# ToDo: bid the largest positions
raise NotImplementedError("OrderedPercentMAEStrategy not implemented yet.")
12 changes: 12 additions & 0 deletions liquidation/model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,3 +35,15 @@ def __init__(
self.tick_size = tick_size
self.point_value = point_value

def notional_at_mark(self) -> Decimal:
return Decimal(self.mark_price * abs(self.quantity)) * self.point_value / Decimal(10**self.tick_size)

def notional_at_price(self, price: int) -> Decimal:
return Decimal(price * abs(self.quantity)) * self.point_value / Decimal(10**self.tick_size)

# Returns the amount of MAE (without the collateral token precision) decreased with this `bid_price`
def dmae(self, bid_price: int) -> Decimal:
return (
self.notional_at_mark() - self.notional_at_price(bid_price) if self.quantity > 0
else self.notional_at_price(bid_price) - self.notional_at_mark()
)
Loading