Skip to content

Instantly share code, notes, and snippets.

@algolog
Created January 13, 2023 19:47
Show Gist options
  • Save algolog/8bd03e76ff33eafecbe70e2ec6bf89b4 to your computer and use it in GitHub Desktop.
Save algolog/8bd03e76ff33eafecbe70e2ec6bf89b4 to your computer and use it in GitHub Desktop.
Lists limit orders on Deflex DEX
import time
from datetime import datetime, timedelta
from dataclasses import dataclass
from base64 import b64decode
from pprint import pprint
from algosdk.v2client.algod import AlgodClient
from algosdk.v2client.indexer import IndexerClient
from algosdk.encoding import encode_address
from algofipy.state_utils import format_state
MAINNET_REGISTRY_APP_ID = 949209670
LIMIT_ORDER_MIN_FEE_BPS = 6
MAX_TIMEDELTA_DAYS = 10000
@dataclass
class LimitOrder:
user_address: str
beneficiary_address: str
limit_order_app_id: int
registry_app_id: int
platform_treasury_address: str
asset_in_id: int
asset_out_id: int
amount_in: int
amount_out: int
expiration_date: int
note: str = ''
fee_bps: int = LIMIT_ORDER_MIN_FEE_BPS
backend_address: str = None
escrow_address: str = None
class DeflexLimitOrdersClient:
def __init__(self, algod_client: AlgodClient, indexer_client: IndexerClient, registry_app_id: int, network: str):
"""Constructor method for generic deflex limit orders client
:param algod_client: a class:`AlgodClient` for interacting with the network
:type algod_client: class:`AlgodClient`
:param indexer_client: a class:`IndexerClient` for interacting with the network
:type indexer_client: class:`IndexerClient`
:param registry_app_id: application ID for orders registry on the network
:type registry_app_id: int
:param network: what network to connect to
:type network: str
"""
self.algod = algod_client
self.indexer = indexer_client
self.registry_app_id = registry_app_id
self.network = network
def fetch_registered_accounts(self):
accounts = []
next_page = ""
while next_page is not None:
resp = self.indexer.accounts(application_id=self.registry_app_id,
limit=500, next_page=next_page)
accounts.extend(resp.get("accounts", []))
next_page = resp.get("next-token", None)
return accounts
def limit_order_from_account(self, account):
app = [a for a in account["apps-local-state"] if a["id"] != self.registry_app_id][0]
app_id = app["id"]
fstate = format_state(app["key-value"])
expiration_date = fstate['expiration_date']
# cap expiration_date to prevent overflow of python datetime types
max_timestamp = int((datetime.utcnow() + timedelta(days=MAX_TIMEDELTA_DAYS)).timestamp())
if expiration_date > max_timestamp:
expiration_date = max_timestamp
limit_order = LimitOrder(
encode_address(b64decode(fstate['user_address'])),
encode_address(b64decode(fstate['beneficiary_address'])),
app_id,
fstate['registry_app_id'],
encode_address(b64decode(fstate['platform_treasury_address'])),
fstate['asset_in_id'],
fstate['asset_out_id'],
fstate['amount_in'],
fstate['amount_out'],
expiration_date,
fstate['note'],
fstate['fee_bps'],
encode_address(b64decode(fstate['backend_address'])),
account['address']
)
return limit_order
def fetch_all_orders(self):
return [self.limit_order_from_account(a) for a in self.fetch_registered_accounts()]
def fetch_all_open_orders(self):
limit_orders = self.fetch_all_orders()
now = time.time()
return [order for order in limit_orders if order.expiration_date >= now]
def fetch_all_expired_orders(self):
limit_orders = self.fetch_all_orders()
now = time.time()
return [order for order in limit_orders if order.expiration_date < now]
class DeflexLimitOrdersMainnetClient(DeflexLimitOrdersClient):
def __init__(self, algod_client=None, indexer_client=None):
if algod_client is None:
algod_client = AlgodClient("", "https://mainnet-api.algonode.cloud")
if indexer_client is None:
indexer_client = IndexerClient("", "https://mainnet-idx.algonode.cloud",
headers={"User-Agent": "algosdk"})
super().__init__(algod_client, indexer_client, MAINNET_REGISTRY_APP_ID, network="mainnet")
if __name__ == '__main__':
client = DeflexLimitOrdersMainnetClient()
orders = client.fetch_all_orders()
print("# asset_in asset_out amount_in price inv_price expires in")
now = time.time()
for o in orders:
dt = str(timedelta(seconds=(o.expiration_date - now))).rsplit('.', 1)[0]
price = o.amount_out / o.amount_in
print("{:10} {:10} {:19_} {:14.5f} {:14.5f} {:>25}".format(o.asset_in_id, o.asset_out_id,
o.amount_in, price, 1/price, dt))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment