Skip to content

Instantly share code, notes, and snippets.

@tirinox
Last active August 22, 2021 08:27
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save tirinox/a077c2c53aa3231cfc5df504d5e2d25c to your computer and use it in GitHub Desktop.
Save tirinox/a077c2c53aa3231cfc5df504d5e2d25c to your computer and use it in GitHub Desktop.
This script sums all your ROWAN on balance and inside the Liquidity pools. USDC/USDT assets on balance are counted separately!
import asyncio
import sys
import aiohttp
def rowan_to_float(x, n=18):
return int(x) / (10 ** n)
def sep(n=100):
print('-' * n)
def is_stable(asset):
return asset in ('cusdc', 'cusdt')
def amount_to_float(x, asset):
table = {
'cusdc': 6,
'cusdt': 6,
}
return rowan_to_float(x, table.get(asset, 18))
class SifBalanceTracker:
def __init__(self, session: aiohttp.ClientSession, address: str):
self.session = session
self.address = address
self.total_rowan = 0
self.pooled_rowan = 0
self.usd_on_balance = 0
self.rowan_on_balance = 0
self.delegated_rowan = 0
async def get_json(self, url):
async with self.session.get(url) as resp:
return await resp.json()
async def get_delegation_total(self):
j = await self.get_json(
f'https://lcd-sifchain.cosmostation.io/cosmos/staking/v1beta1/delegations/{self.address}')
balance = 0.0
for d in j['delegation_responses']:
balance += rowan_to_float(d['balance']['amount'])
return balance
async def get_usd_per_rowan(self):
j = await self.get_json(
'https://api.coingecko.com/api/v3/coins/sifchain?tickers=true&market_data=false&community_data=false'
'&developer_data=false')
return float(j['tickers'][0]['converted_last']['usd'])
async def get_account_balances(self):
j = await self.get_json(f'https://api-int.sifchain.finance/bank/balances/{self.address}')
self.usd_on_balance = 0.0
self.rowan_on_balance = 0.0
for item in j['result']:
asset, amount = item['denom'], item['amount']
if asset == 'rowan':
self.rowan_on_balance = rowan_to_float(amount)
elif is_stable(asset):
self.usd_on_balance += amount_to_float(amount, asset)
async def get_lp_address(self):
j = await self.get_json(f'https://api-int.sifchain.finance/clp/getAssets?lpAddress={self.address}')
return [r['symbol'] for r in j['result']]
async def get_lp_info(self, pool):
url = f'https://api-int.sifchain.finance/clp/getLiquidityProvider?symbol={pool}&lpAddress={self.address}'
j = await self.get_json(url)
r = j['result']
native_asset_balance = rowan_to_float(r['native_asset_balance'])
external_asset_balance = amount_to_float(r['external_asset_balance'], pool)
units = int(r['liquidity_provider']['liquidity_provider_units'])
return native_asset_balance, external_asset_balance, units
async def get_lp_total_rowan(self, pool):
native_asset_balance, external_asset_balance, units = await self.get_lp_info(pool)
rowan_in_pool = native_asset_balance * 2
asset_in_pool = external_asset_balance * 2
print(f'In pool "{pool}" there are {rowan_in_pool:.1f} rowan OR {asset_in_pool:.3f} {pool}.')
print(f'If withdraw 50/50: {native_asset_balance:.1f} rowan AND {external_asset_balance:.3f} {pool}.')
return rowan_in_pool
async def process(self):
lp_pools = await self.get_lp_address()
print(f'Provides liquidity to pools {", ".join(lp_pools)}')
self.pooled_rowan = sum(
await asyncio.gather(*(self.get_lp_total_rowan(pool) for pool in lp_pools))
)
print(f'Pooled sum of rowan: {self.pooled_rowan:.1f} rowan')
self.delegated_rowan = await self.get_delegation_total()
print(f'Delegation is {self.delegated_rowan:.1f} rowan.')
await self.get_account_balances()
print(f'Account balance of {self.address}:')
print(f' USD = ${self.usd_on_balance:.2f}')
print(f' ROWAN = {self.rowan_on_balance:.1f} rowan')
self.total_rowan = self.rowan_on_balance + self.pooled_rowan + self.delegated_rowan
async def main(addresses):
async with aiohttp.ClientSession() as session:
total_rowan = 0
total_usd_on_balance = 0
sif = SifBalanceTracker(session, '')
usd_per_rowan = await sif.get_usd_per_rowan()
print(f'Rowan price: ${usd_per_rowan:.3f}.')
sep(100)
for address in addresses:
print(f'Address: {address}')
sif = SifBalanceTracker(session, address)
await sif.process()
sep(100)
total_rowan += sif.total_rowan
total_usd_on_balance += sif.usd_on_balance
total_rowan_usd = total_rowan * usd_per_rowan
grand_total = total_usd_on_balance + total_rowan_usd
print(f'Total Rowan (balance+pooled): {total_rowan:.1f} rowan')
print(f' or in USDT: ${total_rowan_usd:.2f} ')
print(f'Total Usd on balance: ${total_usd_on_balance:.2f}')
sep(50)
print(f'Grand total: ${grand_total:.2f}')
loop = asyncio.get_event_loop()
loop.run_until_complete(main(sys.argv[1:]))
@tirinox
Copy link
Author

tirinox commented Aug 22, 2021

Usage:
python3 sifchain_count_my_rowan.py sifADDR

If you are multi account guy:
python3 sifchain_count_my_rowan.py sifADDR1 sifADDR2 sifADDR3 ... in other words put there as many sif addresses as you want, separating them by a space.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment