fin-depo/fin_depo/defi_partisia_blockchain.py

"""See `PartisiaBlockchainAccountDepoFetcher` for documentation."""

import dataclasses
import datetime
import email.utils
import json
import logging
from collections.abc import Mapping
from decimal import Decimal
from typing import Any

import fin_defs
import requests
from frozendict import frozendict

from .data import DepoFetcher, 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:
    # Very rough implementation
    if address is None:
        msg = 'Address must not be None'
        raise TypeError(msg)
    if address.endswith('a'):
        return 'shards/Shard0/'
    if address.endswith('2'):
        return 'shards/Shard1/'
    return 'shards/Shard2/'


@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, Any] = frozendict(),
        method='POST',
    ) -> tuple[Any, 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, Any]]:
        data: dict = {'path': []}

        url: str = 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)


BYOC_ASSETS = {
    'ETH': fin_defs.WELL_KNOWN_SYMBOLS['ETH'],
    'POLYGON_USDC': fin_defs.WELL_KNOWN_SYMBOLS['USDC'],
    'BNB': fin_defs.WELL_KNOWN_SYMBOLS['BNB'],
    'MATIC': fin_defs.WELL_KNOWN_SYMBOLS['MATIC'],
}


class PartisiaBlockchainAccountDepoFetcher(DepoFetcher):
    """Depository fetcher for individual [Partisia
    Blockchain](https://partisiablockchain.com/) accounts, including [MPC](https://partisiablockchain.gitlab.io/documentation/pbc-fundamentals/mpc-token-model-and-account-elements.html) and
    [Bring-Your-Own-Coin](https://partisiablockchain.gitlab.io/documentation/pbc-fundamentals/byoc/introduction-to-byoc.html).

    Requirements for use:
    - Account on Partisia Blockchain.

    Depository structure: A nested depo containing both a `Native` and a `BYOC`
    depo.
    """

    def __init__(self, session: requests.Session, blockchain_address: str):
        self.assert_param('session', requests.Session, session)
        self.assert_param('blockchain_address', str, blockchain_address)

        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(
            'Native',
            balances.update_time,
            {fin_defs.MPC: balances.mpc},
        )
        depo_byoc = DepoSingle(
            'Bring Your Own Coin',
            balances.update_time,
            {BYOC_ASSETS[k]: v for k, v in balances.byoc.items()},
        )

        return DepoGroup(
            'Partisia Blockchain',
            balances.update_time,
            [depo_mpc, depo_byoc],
        )