Source code for bitshares.asset

# -*- coding: utf-8 -*-
import json

from bitsharesbase import operations
from bitsharesbase.asset_permissions import (
    asset_permissions,
    force_flag,
    test_permissions,
    todict,
)
from .blockchainobject import BlockchainObject
from .exceptions import AssetDoesNotExistsException
from .instance import BlockchainInstance

from graphenecommon.asset import Asset as GrapheneAsset


[docs]@BlockchainInstance.inject class Asset(GrapheneAsset): """ Deals with Assets of the network. :param str Asset: Symbol name or object id of an asset :param bool lazy: Lazy loading :param bool full: Also obtain bitasset-data and dynamic asset data :param bitshares.bitshares.BitShares blockchain_instance: BitShares instance :returns: All data of an asset :rtype: dict .. note:: This class comes with its own caching function to reduce the load on the API server. Instances of this class can be refreshed with ``Asset.refresh()``. """
[docs] def define_classes(self): self.type_id = 3
def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # Permissions and flags self["permissions"] = todict(self["options"].get("issuer_permissions")) self["flags"] = todict(self["options"].get("flags")) try: self["description"] = json.loads(self["options"]["description"]) except Exception: self["description"] = self["options"]["description"] @property def market_fee_percent(self): return self["options"]["market_fee_percent"] / 100 / 100 @property def max_market_fee(self): from .amount import Amount return Amount( {"amount": self["options"]["max_market_fee"], "asset_id": self["id"]}, blockchain_instance=self.blockchain, ) @property def feeds(self): from .price import PriceFeed self.ensure_full() if not self.is_bitasset: return r = [] for feed in self["bitasset_data"]["feeds"]: r.append(PriceFeed(feed, blockchain_instance=self.blockchain)) return r @property def feed(self): from .price import PriceFeed assert self.is_bitasset self.ensure_full() return PriceFeed( self["bitasset_data"]["current_feed"], blockchain_instance=self.blockchain ) @property def calls(self): return self.get_call_orders(10)
[docs] def get_call_orders(self, limit=100): from .price import Price from .account import Account from .amount import Amount from .market import Market assert limit <= 100 assert self.is_bitasset self.ensure_full() r = [] bitasset = self["bitasset_data"] settlement_price = Price( bitasset["current_feed"]["settlement_price"], blockchain_instance=self.blockchain, ) ret = self.blockchain.rpc.get_call_orders(self["id"], limit) for call in ret[:limit]: collateral_amount = Amount( { "amount": call["collateral"], "asset_id": call["call_price"]["base"]["asset_id"], }, blockchain_instance=self.blockchain, ) debt_amount = Amount( { "amount": call["debt"], "asset_id": call["call_price"]["quote"]["asset_id"], }, blockchain_instance=self.blockchain, ) call_price = collateral_amount / ( debt_amount * (bitasset["current_feed"]["maintenance_collateral_ratio"] / 1000) ) latest = Market( "{}:{}".format( bitasset["options"]["short_backing_asset"], self["symbol"] ) ).ticker()["latest"] r.append( { "account": Account( call["borrower"], lazy=True, blockchain_instance=self.blockchain ), "collateral": collateral_amount, "debt": debt_amount, "call_price": call_price, "settlement_price": settlement_price, "ratio": ( float(collateral_amount) / float(debt_amount) * float(latest) ), } ) return r
@property def settlements(self): return self.get_settle_orders(10)
[docs] def get_settle_orders(self, limit=100): from .account import Account from .amount import Amount from .utils import formatTimeString assert limit <= 100 assert self.is_bitasset r = [] ret = self.blockchain.rpc.get_settle_orders(self["id"], limit) for settle in ret[:limit]: r.append( { "account": Account( settle["owner"], lazy=True, blockchain_instance=self.blockchain ), "amount": Amount( settle["balance"], blockchain_instance=self.blockchain ), "date": formatTimeString(settle["settlement_date"]), } ) return r
[docs] def halt(self): """Halt this asset from being moved or traded.""" from .account import Account nullaccount = Account( "null-account", # We set the null-account blockchain_instance=self.blockchain, ) flags = {"white_list": True, "transfer_restricted": True} options = self["options"] test_permissions(options["issuer_permissions"], flags) flags_int = force_flag(options["flags"], flags) options.update( { "flags": flags_int, "whitelist_authorities": [nullaccount["id"]], "blacklist_authorities": [], "whitelist_markets": [self["id"]], "blacklist_markets": [], } ) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def release( self, whitelist_authorities=None, blacklist_authorities=None, whitelist_markets=None, blacklist_markets=None, ): """ Release this asset and allow unrestricted transfer, trading, etc. :param list whitelist_authorities: List of accounts that serve as whitelist authorities :param list blacklist_authorities: List of accounts that serve as blacklist authorities :param list whitelist_markets: List of assets to allow trading with :param list blacklist_markets: List of assets to prevent trading with """ from .account import Account if whitelist_authorities is None: whitelist_authorities = [] if blacklist_authorities is None: blacklist_authorities = [] if whitelist_markets is None: whitelist_markets = [] if blacklist_markets is None: blacklist_markets = [] flags = {"white_list": False, "transfer_restricted": False} if whitelist_authorities or blacklist_authorities: flags["white_list"] = True options = self["options"] test_permissions(options["issuer_permissions"], flags) flags_int = force_flag(options["flags"], flags) options.update( { "flags": flags_int, "whitelist_authorities": [ Account(a, blockchain_instance=self.blockchain)["id"] for a in whitelist_authorities ], "blacklist_authorities": [ Account(a, blockchain_instance=self.blockchain)["id"] for a in blacklist_authorities ], "whitelist_markets": [Asset(a)["id"] for a in whitelist_markets], "blacklist_markets": [Asset(a)["id"] for a in blacklist_markets], } ) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def setoptions(self, flags, **kwargs): """ Enable a certain flag. Flags: * charge_market_fee * white_list * override_authority * transfer_restricted * disable_force_settle * global_settle * disable_confidential * witness_fed_asset * committee_fed_asset :param dict flag: dictionary of flags and boolean """ assert set(flags.keys()).issubset(asset_permissions.keys()), "unknown flag" options = self["options"] test_permissions(options["issuer_permissions"], flags) flags_int = force_flag(options["flags"], flags) options.update({"flags": flags_int}) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) if kwargs.get("return_op") is True: return op return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def enableflag(self, flag): """ Enable a certain flag. :param str flag: Flag name """ return self.setoptions({flag: True})
[docs] def disableflag(self, flag): """ Enable a certain flag. :param str flag: Flag name """ return self.setoptions({flag: False})
[docs] def seize(self, from_account, to_account, amount, **kwargs): """ Seize amount from an account and send to another. ... note:: This requires the ``override_authority`` to be set for this asset! :param bitshares.account.Account from_account: From this account :param bitshares.account.Account to_account: To this account :param bitshares.amount.Amount amount: Amount to seize """ options = self["options"] if not (options["flags"] & asset_permissions["override_authority"]): raise Exception("Insufficient Permissions/flags for seizure!") op = operations.Override_transfer( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "from": from_account["id"], "to": to_account["id"], "amount": amount.json(), "extensions": [], } ) if kwargs.get("return_op") is True: return op return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def add_authorities(self, type, authorities=None): """ Add authorities to an assets white/black list. :param str type: ``blacklist`` or ``whitelist`` :param list authorities: List of authorities (Accounts) """ assert type in ["blacklist", "whitelist"] assert isinstance(authorities, (list, set)) from .account import Account if authorities is None: authorities = [] flags = {"white_list": True} options = self["options"] test_permissions(options["issuer_permissions"], flags) flags_int = force_flag(options["flags"], flags) options.update({"flags": flags_int}) if type == "whitelist": options["whitelist_authorities"].extend( [ Account(a, blockchain_instance=self.blockchain)["id"] for a in authorities ] ) if type == "blacklist": options["blacklist_authorities"].extend( [ Account(a, blockchain_instance=self.blockchain)["id"] for a in authorities ] ) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def remove_authorities(self, type, authorities=None): """ Remove authorities from an assets white/black list. :param str type: ``blacklist`` or ``whitelist`` :param list authorities: List of authorities (Accounts) """ assert type in ["blacklist", "whitelist"] assert isinstance(authorities, (list, set)) from .account import Account if authorities is None: authorities = [] options = self["options"] if type == "whitelist": for a in authorities: options["whitelist_authorities"].remove( Account(a, blockchain_instance=self.blockchain)["id"] ) if type == "blacklist": for a in authorities: options["blacklist_authorities"].remove( Account(a, blockchain_instance=self.blockchain)["id"] ) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def add_markets(self, type, authorities=None, force_enable=True): """ Add markets to an assets white/black list. :param str type: ``blacklist`` or ``whitelist`` :param list markets: List of markets (assets) :param bool force_enable: Force enable ``white_list`` flag """ assert type in ["blacklist", "whitelist"] assert isinstance(authorities, (list, set)) if authorities is None: authorities = [] options = self["options"] if force_enable: test_permissions(options["issuer_permissions"], {"white_list": True}) flags_int = force_flag(options["flags"], {"white_list": True}) options.update({"flags": flags_int}) else: assert test_permissions( options["flags"], ["white_list"] ), "whitelist feature not enabled" if type == "whitelist": options["whitelist_markets"].extend( [ Asset(a, blockchain_instance=self.blockchain)["id"] for a in authorities ] ) if type == "blacklist": options["blacklist_markets"].extend( [ Asset(a, blockchain_instance=self.blockchain)["id"] for a in authorities ] ) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def remove_markets(self, type, authorities=None): """ Remove markets from an assets white/black list. :param str type: ``blacklist`` or ``whitelist`` :param list markets: List of markets (assets) """ assert type in ["blacklist", "whitelist"] assert isinstance(authorities, (list, set)) if authorities is None: authorities = [] options = self["options"] if type == "whitelist": for a in authorities: options["whitelist_markets"].remove( Asset(a, blockchain_instance=self.blockchain)["id"] ) if type == "blacklist": for a in authorities: options["blacklist_markets"].remove( Asset(a, blockchain_instance=self.blockchain)["id"] ) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def set_market_fee(self, percentage_fee, max_market_fee, **kwargs): """ Set trading percentage fee. :param float percentage_fee: Percentage of fee :param bitshares.amount.Amount max_market_fee: Max Fee """ assert percentage_fee <= 100 and percentage_fee > 0 flags = {"charge_market_fee": percentage_fee > 0} options = self["options"] test_permissions(options["issuer_permissions"], flags) flags_int = force_flag(options["flags"], flags) options.update( { "flags": flags_int, "market_fee_percent": percentage_fee * 100, "max_market_fee": int(max_market_fee), } ) op = operations.Asset_update( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_options": options, "extensions": [], } ) if kwargs.get("return_op") is True: return op return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def update_feed_producers(self, producers): """ Update bitasset feed producers. :param list producers: List of accounts that are allowed to produce a feed """ assert self.is_bitasset, "Asset needs to be a bitasset/market pegged asset" from .account import Account op = operations.Asset_update_feed_producers( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_feed_producers": [ Account(a, blockchain_instance=self.blockchain)["id"] for a in producers ], "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active")
[docs] def change_issuer(self, new_issuer, **kwargs): """ Change asset issuer (needs signing with owner key!) :param str new_issuer: account name """ from .account import Account new_issuer = Account(new_issuer, blockchain_instance=self.blockchain) op = operations.Asset_update_issuer( **{ "fee": {"amount": 0, "asset_id": "1.3.0"}, "issuer": self["issuer"], "asset_to_update": self["id"], "new_issuer": new_issuer["id"], "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "owner", **kwargs)
[docs] def issue(self, amount, to, memo=None, **kwargs): """ Issue new shares of an asset. :param float amount: Amount to issue :param str to: Recipient :param str memo: (optional) Memo message """ from .memo import Memo from .account import Account to = Account(to, blockchain_instance=self.blockchain) account = Account(self["issuer"], blockchain_instance=self.blockchain) memoObj = Memo( from_account=account, to_account=to, blockchain_instance=self.blockchain ) # append operation op = operations.Asset_issue( **{ "fee": { "amount": 0, "asset_id": "1.3.0", }, # Will be filled in automatically "issuer": account["id"], # the Issuer account "asset_to_issue": { "amount": int(float(amount) * 10 ** self["precision"]), "asset_id": self["id"], }, "issue_to_account": to["id"], "memo": memoObj.encrypt(memo), "extensions": [], } ) return self.blockchain.finalizeOp(op, self["issuer"], "active", **kwargs)