Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# blockchain dict init
amount = Amount({
"amount": 1 * 10 ** precision,
"asset_id": asset["id"]
}, steem_instance=stm)
self.dotest(amount, 1, symbol)
# API dict init
amount = Amount({
"amount": 1.3 * 10 ** precision,
"asset": asset["id"]
}, steem_instance=stm)
self.dotest(amount, 1.3, symbol)
# Asset as symbol
amount = Amount(1.3, Asset("SBD"), steem_instance=stm)
self.dotest(amount, 1.3, symbol)
# Asset as symbol
amount = Amount(1.3, symbol, steem_instance=stm)
self.dotest(amount, 1.3, symbol)
# keyword inits
amount = Amount(amount=1.3, asset=Asset("SBD", steem_instance=stm), steem_instance=stm)
self.dotest(amount, 1.3, symbol)
amount = Amount(amount=1.3001, asset=Asset("SBD", steem_instance=stm), steem_instance=stm)
self.dotest(amount, 1.3001, symbol)
amount = Amount(amount=1.3001, asset=Asset("SBD", steem_instance=stm), fixed_point_arithmetic=True, steem_instance=stm)
self.dotest(amount, 1.3, symbol)
def test_init(self):
stm = self.bts
# String init
asset = Asset("SBD", steem_instance=stm)
symbol = asset["symbol"]
precision = asset["precision"]
amount = Amount("1 {}".format(symbol), steem_instance=stm)
self.dotest(amount, 1, symbol)
# Amount init
amount = Amount(amount, steem_instance=stm)
self.dotest(amount, 1, symbol)
# blockchain dict init
amount = Amount({
"amount": 1 * 10 ** precision,
"asset_id": asset["id"]
}, steem_instance=stm)
self.dotest(amount, 1, symbol)
# API dict init
amount = Amount({
"amount": 1.3 * 10 ** precision,
"asset": asset["id"]
}, steem_instance=stm)
self.dotest(amount, 1.3, symbol)
def test_Transfer_broadcast(self):
nodelist = NodeList()
stm = Steem(node=self.nodes,
keys=[self.active_key],
nobroadcast=True,
expiration=120,
num_retries=10)
tx = TransactionBuilder(use_condenser_api=True, expiration=10, steem_instance=stm)
tx.appendOps(Transfer(**{"from": "beem",
"to": "beem1",
"amount": Amount("1 STEEM", steem_instance=stm),
"memo": ""}))
tx.appendSigner("beem", "active")
tx.sign()
tx.broadcast()
def test_order(self):
order = Order(Amount("2 SBD"), Amount("1 STEEM"))
self.assertTrue(repr(order) is not None)
def test_amount(self, node_param):
if node_param == "instance":
stm = Steem(node="https://abc.d", autoconnect=False, num_retries=1)
set_shared_steem_instance(self.bts)
o = Amount("1 SBD")
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Amount("1 SBD", steem_instance=stm)
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = Amount("1 SBD", steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Amount("1 SBD")
'qustodian': 0.123 STEEM, 'jpphotography': 0.002 STEEM, 'thinkingmind': 0.001 STEEM,
'oups': 0.006 STEEM, 'mattockfs': 0.001 STEEM, 'holger80': 0.003 STEEM, 'michaelizer': 0.004 STEEM,
'flugschwein': 0.010 STEEM, 'ulisessabeque': 0.000 STEEM, 'hakancelik': 0.002 STEEM, 'sbi2': 0.008 STEEM,
'zcool': 0.000 STEEM, 'steemhq': 0.002 STEEM, 'rowdiya': 0.000 STEEM, 'qurator-tier-1-2': 0.012 STEEM
}
}
"""
median_price = Price(self.steem.get_current_median_history(), steem_instance=self.steem)
pending_rewards = False
total_vote_weight = self["total_vote_weight"]
if not self["allow_curation_rewards"]:
max_rewards = Amount(0, self.steem.steem_symbol, steem_instance=self.steem)
unclaimed_rewards = max_rewards.copy()
elif not self.is_pending():
max_rewards = Amount(self["curator_payout_value"], steem_instance=self.steem)
unclaimed_rewards = Amount(self["total_payout_value"], steem_instance=self.steem) * 0.25 - max_rewards
total_vote_weight = 0
for vote in self["active_votes"]:
total_vote_weight += int(vote["weight"])
else:
if pending_payout_value is None:
pending_payout_value = Amount(self["pending_payout_value"], steem_instance=self.steem)
elif isinstance(pending_payout_value, (float, integer_types)):
pending_payout_value = Amount(pending_payout_value, self.steem.sbd_symbol, steem_instance=self.steem)
elif isinstance(pending_payout_value, str):
pending_payout_value = Amount(pending_payout_value, steem_instance=self.steem)
if pending_payout_SBD:
max_rewards = (pending_payout_value * 0.25)
else:
max_rewards = median_price.as_base(self.steem.steem_symbol) * (pending_payout_value * 0.25)
unclaimed_rewards = max_rewards.copy()
def sbd_to_rshares(self, sbd, not_broadcasted_vote=False, use_stored_data=True):
""" Obtain the r-shares from SBD
:param sbd: SBD
:type sbd: str, int, amount.Amount
:param bool not_broadcasted_vote: not_broadcasted or already broadcasted vote (True = not_broadcasted vote).
Only impactful for very high amounts of SBD. Slight modification to the value calculation, as the not_broadcasted
vote rshares decreases the reward pool.
"""
if isinstance(sbd, Amount):
sbd = Amount(sbd, blockchain_instance=self)
elif isinstance(sbd, string_types):
sbd = Amount(sbd, blockchain_instance=self)
else:
sbd = Amount(sbd, self.sbd_symbol, blockchain_instance=self)
if sbd['symbol'] != self.sbd_symbol:
raise AssertionError('Should input SBD, not any other asset!')
# If the vote was already broadcasted we can assume the blockchain values to be true
if not not_broadcasted_vote:
return int(float(sbd) / self.get_sbd_per_rshares(use_stored_data=use_stored_data))
# If the vote wasn't broadcasted (yet), we have to calculate the rshares while considering
# the change our vote is causing to the recent_claims. This is more important for really
# big votes which have a significant impact on the recent_claims.
reward_fund = self.get_reward_funds(use_stored_data=use_stored_data)
median_price = self.get_median_price(use_stored_data=use_stored_data)
recent_claims = int(reward_fund["recent_claims"])
reward_balance = Amount(reward_fund["reward_balance"], blockchain_instance=self)
reward_pool_sbd = median_price * reward_balance
if sbd > reward_pool_sbd:
'leprechaun': 0.006 STEEM, 'timcliff': 0.186 STEEM,
'st3llar': 0.000 STEEM, 'crokkon': 0.015 STEEM, 'feedyourminnows': 0.003 STEEM,
'isnochys': 0.003 STEEM, 'loshcat': 0.001 STEEM, 'greenorange': 0.000 STEEM,
'qustodian': 0.123 STEEM, 'jpphotography': 0.002 STEEM, 'thinkingmind': 0.001 STEEM,
'oups': 0.006 STEEM, 'mattockfs': 0.001 STEEM, 'holger80': 0.003 STEEM, 'michaelizer': 0.004 STEEM,
'flugschwein': 0.010 STEEM, 'ulisessabeque': 0.000 STEEM, 'hakancelik': 0.002 STEEM, 'sbi2': 0.008 STEEM,
'zcool': 0.000 STEEM, 'steemhq': 0.002 STEEM, 'rowdiya': 0.000 STEEM, 'qurator-tier-1-2': 0.012 STEEM
}
}
"""
median_price = Price(self.steem.get_current_median_history(), steem_instance=self.steem)
pending_rewards = False
total_vote_weight = self["total_vote_weight"]
if not self["allow_curation_rewards"]:
max_rewards = Amount(0, self.steem.steem_symbol, steem_instance=self.steem)
unclaimed_rewards = max_rewards.copy()
elif not self.is_pending():
max_rewards = Amount(self["curator_payout_value"], steem_instance=self.steem)
unclaimed_rewards = Amount(self["total_payout_value"], steem_instance=self.steem) * 0.25 - max_rewards
total_vote_weight = 0
for vote in self["active_votes"]:
total_vote_weight += int(vote["weight"])
else:
if pending_payout_value is None:
pending_payout_value = Amount(self["pending_payout_value"], steem_instance=self.steem)
elif isinstance(pending_payout_value, (float, integer_types)):
pending_payout_value = Amount(pending_payout_value, self.steem.sbd_symbol, steem_instance=self.steem)
elif isinstance(pending_payout_value, str):
pending_payout_value = Amount(pending_payout_value, steem_instance=self.steem)
if pending_payout_SBD:
max_rewards = (pending_payout_value * 0.25)
def copy(self):
""" Copy the instance and make sure not to use a reference
"""
return Amount(
amount=self["amount"],
asset=self["asset"].copy(),
new_appbase_format=self.new_appbase_format,
fixed_point_arithmetic=self.fixed_point_arithmetic,
steem_instance=self.steem)
def get_sbd_per_rshares(self, not_broadcasted_vote_rshares=0, use_stored_data=True):
""" Returns the current rshares to SBD ratio
"""
reward_fund = self.get_reward_funds(use_stored_data=use_stored_data)
reward_balance = float(Amount(reward_fund["reward_balance"], blockchain_instance=self))
recent_claims = float(reward_fund["recent_claims"]) + not_broadcasted_vote_rshares
fund_per_share = reward_balance / (recent_claims)
median_price = self.get_median_price(use_stored_data=use_stored_data)
if median_price is None:
return 0
SBD_price = float(median_price * (Amount(1, self.steem_symbol, blockchain_instance=self)))
return fund_per_share * SBD_price