1
0

Partisia blockchain

This commit is contained in:
Jon Michael Aanes 2024-06-02 18:33:25 +02:00
parent f74945d6bf
commit 949244656a
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
4 changed files with 160 additions and 5 deletions

View File

@ -1,3 +1,3 @@
__all__ = ['defi_kraken', 'defi_kucoin']
__all__ = ['defi_kraken', 'defi_kucoin', 'defi_partisia_blockchain']
from . import defi_kraken, defi_kucoin
from . import defi_kraken, defi_kucoin, defi_partisia_blockchain

View File

@ -12,6 +12,8 @@ logger = logging.getLogger(__name__)
class KrakenDepoFetcher:
def __init__(self, kraken_key: str, kraken_secret: str):
assert kraken_key is not None, 'Missing kraken_key'
assert kraken_secret is not None, 'Missing kraken_secret'
self.client = krakenex.API(
kraken_key,
kraken_secret,
@ -24,6 +26,7 @@ class KrakenDepoFetcher:
assets: dict[fin_defs.Asset, Decimal] = {}
for account, balance_str in result['result'].items():
account = account.removesuffix('.HOLD')
asset = fin_defs.WELL_KNOWN_SYMBOLS[account]
balance = Decimal(balance_str)
if balance != 0:
@ -32,6 +35,6 @@ class KrakenDepoFetcher:
return DepoSingle(
name='Kraken',
assets=assets,
_assets=assets,
updated_time=now,
)

View File

@ -12,6 +12,9 @@ logger = logging.getLogger(__name__)
class KucoinDepoFetcher:
def __init__(self, kucoin_key: str, kucoin_secret: str, kucoin_pass: str):
assert kucoin_key is not None, 'Missing kucoin_key'
assert kucoin_secret is not None, 'Missing kucoin_secret'
assert kucoin_pass is not None, 'Missing kucoin_pass'
self.client = kucoin.client.Client(
kucoin_key,
kucoin_secret,
@ -24,14 +27,14 @@ class KucoinDepoFetcher:
for account_data in self.client.get_accounts():
asset = fin_defs.WELL_KNOWN_SYMBOLS[account_data['currency']]
balance = Decimal(account_data['balance'])
assets_for_account_type = assets_by_account_type.get(
assets_for_account_type = assets_by_account_type.setdefault(
account_data['type'],
{},
)
assets_for_account_type[asset] = (
assets_for_account_type.get(asset, Decimal(0)) + balance
)
del account_data, asset, balance
del account_data, asset, balance, assets_for_account_type
return DepoGroup(
'Kucoin',
now,

View File

@ -0,0 +1,149 @@
import dataclasses
import datetime
import email.utils
import json
import logging
from collections.abc import Mapping
from decimal import Decimal
import fin_defs
import requests
from frozendict import frozendict
from . import DepoGroup, DepoSingle
logger = logging.getLogger(__name__)
# mainnet: https://reader.partisiablockchain.com
# testnet: https://node1.testnet.partisiablockchain.com
HOSTNAME = 'reader.partisiablockchain.com'
URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local'
URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global'
URL_CONTRACT_STATE = 'https://{hostname}/{shard}blockchain/contracts/{address}?requireContractState=false'
MPC_DECIMALS = 10000
def shard_id_for_address(address: str) -> str:
if address.endswith('a'):
return 'shards/Shard0/' # TODO
elif address.endswith('2'):
return 'shards/Shard1/' # TODO
else:
return 'shards/Shard2/' # TODO
@dataclasses.dataclass(frozen=True)
class Balances:
update_time: datetime.datetime
mpc: Decimal
byoc: Mapping[str, Decimal]
@dataclasses.dataclass(frozen=True)
class PbcClient:
session: requests.Session
def get_json(
self,
url: str,
data: Mapping[str, str] = frozendict(),
method='POST',
) -> tuple[dict, datetime.datetime]:
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json',
}
response = self.session.request(
method,
url,
headers=headers,
data=json.dumps(data),
)
response.raise_for_status()
date_text = response.headers.get('last-modified') or response.headers.get(
'date',
)
date = email.utils.parsedate_to_datetime(date_text)
json_data = response.json()
if json_data is None:
msg = 'No result data for ' + url
raise Exception(msg)
return (json_data, date)
def determine_coins(self) -> list[dict[str, str]]:
data: dict = {'path': []}
url = URL_ACCOUNT_PLUGIN_GLOBAL.format(
hostname=HOSTNAME,
shard='',
)
json_data, date = self.get_json(url, data=data)
return json_data['coins']['coins']
def get_account_balances(self, address: str) -> Balances:
coins = self.determine_coins()
url = URL_ACCOUNT_PLUGIN.format(
hostname=HOSTNAME,
shard=shard_id_for_address(address),
)
data: dict = {
'path': [
{'type': 'field', 'name': 'accounts'},
{'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address},
],
}
account_data, date = self.get_json(url, data=data)
byoc: dict[str, Decimal] = {}
mpc = Decimal(account_data['mpcTokens']) / MPC_DECIMALS
for coin_idx, amount_data in enumerate(account_data['accountCoins']):
coin_data = coins[coin_idx]
byoc_balance = Decimal(amount_data['balance'])
denominator = Decimal(coin_data['conversionRate']['denominator'])
native_balance = byoc_balance / denominator
byoc[coin_data['symbol']] = native_balance
del coin_idx, coin_data
return Balances(date, mpc, byoc)
def get_contract_state(self, address: str) -> tuple[dict, datetime.datetime]:
url = URL_CONTRACT_STATE.format(
hostname=HOSTNAME,
shard=shard_id_for_address(address),
address=address,
)
data: dict = {'path': []}
return self.get_json(url, data=data)
class PartisiaBlockchainAccountDepoFetcher:
def __init__(self, session, blockchain_address: str):
self.client = PbcClient(session)
self.blockchain_address = blockchain_address
def get_depo(self) -> DepoGroup:
balances = self.client.get_account_balances(self.blockchain_address)
depo_mpc = DepoSingle('MPC', balances.update_time, {fin_defs.MPC: balances.mpc})
depo_byoc = DepoSingle(
'BYOC',
balances.update_time,
{fin_defs.WELL_KNOWN_SYMBOLS[k]: v for k, v in balances.byoc.items()},
)
return DepoGroup(
'Partisia Blockchain',
balances.update_time,
[depo_mpc, depo_byoc],
)