Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_market(self):
bts = self.bts
m1 = Market(u'STEEM', u'SBD', steem_instance=bts)
self.assertEqual(m1.get_string(), u'SBD:STEEM')
m2 = Market(steem_instance=bts)
self.assertEqual(m2.get_string(), u'SBD:STEEM')
m3 = Market(u'STEEM:SBD', steem_instance=bts)
self.assertEqual(m3.get_string(), u'STEEM:SBD')
self.assertTrue(m1 == m2)
base = Asset("SBD", steem_instance=bts)
quote = Asset("STEEM", steem_instance=bts)
m = Market(base, quote, steem_instance=bts)
self.assertEqual(m.get_string(), u'STEEM:SBD')
def test_volume(self):
bts = self.bts
m = Market(u'STEEM:SBD', steem_instance=bts)
volume = m.volume24h()
self.assertEqual(volume['STEEM']["symbol"], u'STEEM')
self.assertEqual(volume['SBD']["symbol"], u'SBD')
def test_buy(self):
bts = self.bts
m = Market(u'STEEM:SBD', steem_instance=bts)
bts.txbuffer.clear()
tx = m.buy(5, 0.1, account="test")
self.assertEqual(
(tx["operations"][0][0]),
"limit_order_create"
)
op = tx["operations"][0][1]
self.assertIn("test", op["owner"])
self.assertEqual(str(Amount('0.100 STEEM', steem_instance=bts)), op["min_to_receive"])
self.assertEqual(str(Amount('0.500 SBD', steem_instance=bts)), op["amount_to_sell"])
p = Price(5, u"SBD:STEEM", steem_instance=bts)
tx = m.buy(p, 0.1, account="test")
op = tx["operations"][0][1]
self.assertEqual(str(Amount('0.100 STEEM', steem_instance=bts)), op["min_to_receive"])
self.assertEqual(str(Amount('0.500 SBD', steem_instance=bts)), op["amount_to_sell"])
def test_market(self, node_param):
if node_param == "instance":
set_shared_steem_instance(self.bts)
o = Market()
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Market(steem_instance=Steem(node="https://abc.d", autoconnect=False, num_retries=1))
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = Market(steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Market()
def test_market(self, node_param):
if node_param == "instance":
set_shared_steem_instance(self.bts)
o = Market()
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Market(steem_instance=Steem(node="https://abc.d", autoconnect=False, num_retries=1))
else:
set_shared_steem_instance(Steem(node="https://abc.d", autoconnect=False, num_retries=1))
stm = self.bts
o = Market(steem_instance=stm)
self.assertIn(o.steem.rpc.url, self.urls)
with self.assertRaises(
RPCConnection
):
Market()
def test_orderbook(self):
bts = self.bts
m = Market(u'STEEM:SBD', steem_instance=bts)
orderbook = m.orderbook(limit=10)
self.assertEqual(len(orderbook['asks_date']), 10)
self.assertEqual(len(orderbook['asks']), 10)
self.assertEqual(len(orderbook['bids_date']), 10)
self.assertEqual(len(orderbook['bids']), 10)
"""
self.steem = steem_instance or shared_steem_instance()
if quote is None and isinstance(base, str):
quote_symbol, base_symbol = assets_from_string(base)
quote = Asset(quote_symbol, steem_instance=self.steem)
base = Asset(base_symbol, steem_instance=self.steem)
super(Market, self).__init__({"base": base, "quote": quote})
elif base and quote:
quote = Asset(quote, steem_instance=self.steem)
base = Asset(base, steem_instance=self.steem)
super(Market, self).__init__({"base": base, "quote": quote})
elif base is None and quote is None:
quote = Asset("SBD", steem_instance=self.steem)
base = Asset("STEEM", steem_instance=self.steem)
super(Market, self).__init__({"base": base, "quote": quote})
else:
raise ValueError("Unknown Market config")
if not accounts:
accounts = [stm.config["default_account"]]
if not comment and not curation and not post:
post = True
permlink = True
if days < 0:
days = 1
utc = pytz.timezone('UTC')
now = utc.localize(datetime.utcnow())
limit_time = now - timedelta(days=days)
for account in accounts:
sum_reward = [0, 0, 0, 0, 0]
account = Account(account, steem_instance=stm)
median_price = Price(stm.get_current_median_history(), steem_instance=stm)
m = Market(steem_instance=stm)
latest = m.ticker()["latest"]
if author and permlink:
t = PrettyTable(["Author", "Permlink", "Payout", "SBD", "SP + STEEM", "Liquid USD", "Invested USD"])
elif author and title:
t = PrettyTable(["Author", "Title", "Payout", "SBD", "SP + STEEM", "Liquid USD", "Invested USD"])
elif author:
t = PrettyTable(["Author", "Payout", "SBD", "SP + STEEM", "Liquid USD", "Invested USD"])
elif not author and permlink:
t = PrettyTable(["Permlink", "Payout", "SBD", "SP + STEEM", "Liquid USD", "Invested USD"])
elif not author and title:
t = PrettyTable(["Title", "Payout", "SBD", "SP + STEEM", "Liquid USD", "Invested USD"])
else:
t = PrettyTable(["Received", "SBD", "SP + STEEM", "Liquid USD", "Invested USD"])
t.align = "l"
rows = []
start_op = account.estimate_virtual_op_num(limit_time)
def orderbook(chart, limit, show_date, width, height, ascii):
"""Obtain orderbook of the internal market"""
stm = shared_steem_instance()
if stm.rpc is not None:
stm.rpc.rpcconnect()
market = Market(steem_instance=stm)
orderbook = market.orderbook(limit=limit, raw_data=False)
if not show_date:
header = ["Asks Sum SBD", "Sell Orders", "Bids Sum SBD", "Buy Orders"]
else:
header = ["Asks date", "Sell Orders", "Bids date", "Buy Orders"]
t = PrettyTable(header, hrules=0)
t.align = "r"
asks = []
bids = []
asks_date = []
bids_date = []
sumsum_asks = []
sum_asks = 0
sumsum_bids = []
sum_bids = 0
n = 0
accounts = [stm.config["default_account"]]
if not comment and not curation and not post:
post = True
permlink = True
if days < 0:
days = 1
if days > 7:
days = 7
utc = pytz.timezone('UTC')
limit_time = utc.localize(datetime.utcnow()) - timedelta(days=days)
for account in accounts:
sum_reward = [0, 0, 0, 0]
account = Account(account, steem_instance=stm)
median_price = Price(stm.get_current_median_history(), steem_instance=stm)
m = Market(steem_instance=stm)
latest = m.ticker()["latest"]
if author and permlink:
t = PrettyTable(["Author", "Permlink", "Cashout", "SBD", "SP", "Liquid USD", "Invested USD"])
elif author and title:
t = PrettyTable(["Author", "Title", "Cashout", "SBD", "SP", "Liquid USD", "Invested USD"])
elif author:
t = PrettyTable(["Author", "Cashout", "SBD", "SP", "Liquid USD", "Invested USD"])
elif not author and permlink:
t = PrettyTable(["Permlink", "Cashout", "SBD", "SP", "Liquid USD", "Invested USD"])
elif not author and title:
t = PrettyTable(["Title", "Cashout", "SBD", "SP", "Liquid USD", "Invested USD"])
else:
t = PrettyTable(["Cashout", "SBD", "SP", "Liquid USD", "Invested USD"])
t.align = "l"
rows = []
c_list = {}