from .helper import limits, unittest, mock, D import requests_mock import market @unittest.skipUnless("unit" in limits, "Unit skipped") class poloniexETest(unittest.TestCase): def setUp(self): super().setUp() self.wm = requests_mock.Mocker() self.wm.start() self.s = market.ccxt.poloniexE() def tearDown(self): self.wm.stop() super().tearDown() def test__init(self): with self.subTest("Nominal case"), \ mock.patch("market.ccxt.poloniexE.session") as session: session.request.return_value = "response" ccxt = market.ccxt.poloniexE() ccxt._market = mock.Mock ccxt._market.report = mock.Mock() ccxt._market.market_id = 3 ccxt._market.user_id = 3 ccxt.session.request("GET", "URL", data="data", headers={}) ccxt._market.report.log_http_request.assert_called_with('GET', 'URL', 'data', {'X-market-id': '3', 'X-user-id': '3'}, 'response') with self.subTest("Raising"),\ mock.patch("market.ccxt.poloniexE.session") as session: session.request.side_effect = market.ccxt.RequestException("Boo") ccxt = market.ccxt.poloniexE() ccxt._market = mock.Mock ccxt._market.report = mock.Mock() ccxt._market.market_id = 3 ccxt._market.user_id = 3 with self.assertRaises(market.ccxt.RequestException, msg="Boo") as cm: ccxt.session.request("GET", "URL", data="data", headers={}) ccxt._market.report.log_http_request.assert_called_with('GET', 'URL', 'data', {'X-market-id': '3', 'X-user-id': '3'}, cm.exception) def test_nanoseconds(self): with mock.patch.object(market.ccxt.time, "time") as time: time.return_value = 123456.7890123456 self.assertEqual(123456789012345, self.s.nanoseconds()) def test_nonce(self): with mock.patch.object(market.ccxt.time, "time") as time: time.return_value = 123456.7890123456 self.assertEqual(123456789012345, self.s.nonce()) def test_request(self): with mock.patch.object(market.ccxt.poloniex, "request") as request,\ mock.patch("market.ccxt.retry_call") as retry_call: with self.subTest(wrapped=True): with self.subTest(desc="public"): self.s.request("foo") retry_call.assert_called_with(request, delay=1, tries=10, fargs=["foo"], fkwargs={'api': 'public', 'method': 'GET', 'params': {}, 'headers': None, 'body': None}, exceptions=(market.ccxt.RequestTimeout, market.ccxt.InvalidNonce)) request.assert_not_called() with self.subTest(desc="private GET"): self.s.request("foo", api="private") retry_call.assert_called_with(request, delay=1, tries=10, fargs=["foo"], fkwargs={'api': 'private', 'method': 'GET', 'params': {}, 'headers': None, 'body': None}, exceptions=(market.ccxt.RequestTimeout, market.ccxt.InvalidNonce)) request.assert_not_called() with self.subTest(desc="private POST regexp"): self.s.request("returnFoo", api="private", method="POST") retry_call.assert_called_with(request, delay=1, tries=10, fargs=["returnFoo"], fkwargs={'api': 'private', 'method': 'POST', 'params': {}, 'headers': None, 'body': None}, exceptions=(market.ccxt.RequestTimeout, market.ccxt.InvalidNonce)) request.assert_not_called() with self.subTest(desc="private POST non-regexp"): self.s.request("getMarginPosition", api="private", method="POST") retry_call.assert_called_with(request, delay=1, tries=10, fargs=["getMarginPosition"], fkwargs={'api': 'private', 'method': 'POST', 'params': {}, 'headers': None, 'body': None}, exceptions=(market.ccxt.RequestTimeout, market.ccxt.InvalidNonce)) request.assert_not_called() retry_call.reset_mock() request.reset_mock() with self.subTest(wrapped=False): with self.subTest(desc="private POST non-matching regexp"): self.s.request("marginBuy", api="private", method="POST") request.assert_called_with("marginBuy", api="private", method="POST", params={}, headers=None, body=None) retry_call.assert_not_called() with self.subTest(desc="private POST non-matching non-regexp"): self.s.request("closeMarginPositionOther", api="private", method="POST") request.assert_called_with("closeMarginPositionOther", api="private", method="POST", params={}, headers=None, body=None) retry_call.assert_not_called() def test_order_precision(self): self.s.markets = { "FOO": { "precision": { "price": 5, "amount": 6, } }, "BAR": { "precision": { "price": 7, "amount": 8, } } } with mock.patch.object(self.s, "load_markets") as load_markets: self.assertEqual(5, self.s.order_precision("FOO")) load_markets.assert_called_once() def test_transfer_balance(self): with self.subTest(success=True),\ mock.patch.object(self.s, "privatePostTransferBalance") as t: t.return_value = { "success": 1 } result = self.s.transfer_balance("FOO", 12, "exchange", "margin") t.assert_called_once_with({ "currency": "FOO", "amount": 12, "fromAccount": "exchange", "toAccount": "margin", "confirmed": 1 }) self.assertTrue(result) with self.subTest(success=False),\ mock.patch.object(self.s, "privatePostTransferBalance") as t: t.return_value = { "success": 0 } self.assertFalse(self.s.transfer_balance("FOO", 12, "exchange", "margin")) def test_close_margin_position(self): with mock.patch.object(self.s, "privatePostCloseMarginPosition") as c: self.s.close_margin_position("FOO", "BAR") c.assert_called_with({"currencyPair": "BAR_FOO"}) def test_tradable_balances(self): with mock.patch.object(self.s, "privatePostReturnTradableBalances") as r: r.return_value = { "FOO": { "exchange": "12.1234", "margin": "0.0123" }, "BAR": { "exchange": "1", "margin": "0" }, } balances = self.s.tradable_balances() self.assertEqual(["FOO", "BAR"], list(balances.keys())) self.assertEqual(["exchange", "margin"], list(balances["FOO"].keys())) self.assertEqual(D("12.1234"), balances["FOO"]["exchange"]) self.assertEqual(["exchange", "margin"], list(balances["BAR"].keys())) def test_margin_summary(self): with mock.patch.object(self.s, "privatePostReturnMarginAccountSummary") as r: r.return_value = { "currentMargin": "1.49680968", "lendingFees": "0.0000001", "pl": "0.00008254", "totalBorrowedValue": "0.00673602", "totalValue": "0.01000000", "netValue": "0.01008254", } expected = { 'current_margin': D('1.49680968'), 'gains': D('0.00008254'), 'lending_fees': D('0.0000001'), 'total': D('0.01000000'), 'total_borrowed': D('0.00673602') } self.assertEqual(expected, self.s.margin_summary()) def test_parse_ticker(self): ticker = { "high24hr": "12", "low24hr": "10", "highestBid": "10.5", "lowestAsk": "11.5", "last": "11", "percentChange": "0.1", "quoteVolume": "10", "baseVolume": "20" } market = { "symbol": "BTC/ETC" } with mock.patch.object(self.s, "milliseconds") as ms: ms.return_value = 1520292715123 result = self.s.parse_ticker(ticker, market) expected = { "symbol": "BTC/ETC", "timestamp": 1520292715123, "datetime": "2018-03-05T23:31:55.123Z", "high": D("12"), "low": D("10"), "bid": D("10.5"), "ask": D("11.5"), "vwap": None, "open": None, "close": None, "first": None, "last": D("11"), "change": D("0.1"), "percentage": None, "average": None, "baseVolume": D("10"), "quoteVolume": D("20"), "info": ticker } self.assertEqual(expected, result) def test_fetch_margin_balance(self): with mock.patch.object(self.s, "privatePostGetMarginPosition") as get_margin_position: get_margin_position.return_value = { "BTC_DASH": { "amount": "-0.1", "basePrice": "0.06818560", "lendingFees": "0.00000001", "liquidationPrice": "0.15107132", "pl": "-0.00000371", "total": "0.00681856", "type": "short" }, "BTC_ETC": { "amount": "-0.6", "basePrice": "0.1", "lendingFees": "0.00000001", "liquidationPrice": "0.6", "pl": "0.00000371", "total": "0.06", "type": "short" }, "BTC_ETH": { "amount": "0", "basePrice": "0", "lendingFees": "0", "liquidationPrice": "-1", "pl": "0", "total": "0", "type": "none" } } balances = self.s.fetch_margin_balance() self.assertEqual(2, len(balances)) expected = { "DASH": { "amount": D("-0.1"), "borrowedPrice": D("0.06818560"), "lendingFees": D("1E-8"), "pl": D("-0.00000371"), "liquidationPrice": D("0.15107132"), "type": "short", "total": D("0.00681856"), "baseCurrency": "BTC" }, "ETC": { "amount": D("-0.6"), "borrowedPrice": D("0.1"), "lendingFees": D("1E-8"), "pl": D("0.00000371"), "liquidationPrice": D("0.6"), "type": "short", "total": D("0.06"), "baseCurrency": "BTC" } } self.assertEqual(expected, balances) def test_sum(self): self.assertEqual(D("1.1"), self.s.sum(D("1"), D("0.1"))) def test_fetch_balance(self): with mock.patch.object(self.s, "load_markets") as load_markets,\ mock.patch.object(self.s, "privatePostReturnCompleteBalances") as balances,\ mock.patch.object(self.s, "common_currency_code") as ccc: ccc.side_effect = ["ETH", "BTC", "DASH"] balances.return_value = { "ETH": { "available": "10", "onOrders": "1", }, "BTC": { "available": "1", "onOrders": "0", }, "DASH": { "available": "0", "onOrders": "3" } } expected = { "info": { "ETH": {"available": "10", "onOrders": "1"}, "BTC": {"available": "1", "onOrders": "0"}, "DASH": {"available": "0", "onOrders": "3"} }, "ETH": {"free": D("10"), "used": D("1"), "total": D("11")}, "BTC": {"free": D("1"), "used": D("0"), "total": D("1")}, "DASH": {"free": D("0"), "used": D("3"), "total": D("3")}, "free": {"ETH": D("10"), "BTC": D("1"), "DASH": D("0")}, "used": {"ETH": D("1"), "BTC": D("0"), "DASH": D("3")}, "total": {"ETH": D("11"), "BTC": D("1"), "DASH": D("3")} } result = self.s.fetch_balance() load_markets.assert_called_once() self.assertEqual(expected, result) def test_fetch_balance_per_type(self): with mock.patch.object(self.s, "privatePostReturnAvailableAccountBalances") as balances: balances.return_value = { "exchange": { "BLK": "159.83673869", "BTC": "0.00005959", "USDT": "0.00002625", "XMR": "0.18719303" }, "margin": { "BTC": "0.03019227" } } expected = { "info": { "exchange": { "BLK": "159.83673869", "BTC": "0.00005959", "USDT": "0.00002625", "XMR": "0.18719303" }, "margin": { "BTC": "0.03019227" } }, "exchange": { "BLK": D("159.83673869"), "BTC": D("0.00005959"), "USDT": D("0.00002625"), "XMR": D("0.18719303") }, "margin": {"BTC": D("0.03019227")}, "BLK": {"exchange": D("159.83673869")}, "BTC": {"exchange": D("0.00005959"), "margin": D("0.03019227")}, "USDT": {"exchange": D("0.00002625")}, "XMR": {"exchange": D("0.18719303")} } result = self.s.fetch_balance_per_type() self.assertEqual(expected, result) def test_fetch_all_balances(self): import json with mock.patch.object(self.s, "load_markets") as load_markets,\ mock.patch.object(self.s, "privatePostGetMarginPosition") as margin_balance,\ mock.patch.object(self.s, "privatePostReturnCompleteBalances") as balance,\ mock.patch.object(self.s, "privatePostReturnAvailableAccountBalances") as balance_per_type: with open("test_samples/poloniexETest.test_fetch_all_balances.1.json") as f: balance.return_value = json.load(f) with open("test_samples/poloniexETest.test_fetch_all_balances.2.json") as f: margin_balance.return_value = json.load(f) with open("test_samples/poloniexETest.test_fetch_all_balances.3.json") as f: balance_per_type.return_value = json.load(f) result = self.s.fetch_all_balances() expected_doge = { "total": D("-12779.79821852"), "exchange_used": D("0E-8"), "exchange_total": D("0E-8"), "exchange_free": D("0E-8"), "margin_available": 0, "margin_in_position": 0, "margin_borrowed": D("12779.79821852"), "margin_total": D("-12779.79821852"), "margin_pending_gain": 0, "margin_lending_fees": D("-9E-8"), "margin_pending_base_gain": D("0.00024059"), "margin_position_type": "short", "margin_liquidation_price": D("0.00000246"), "margin_borrowed_base_price": D("0.00599149"), "margin_borrowed_base_currency": "BTC" } expected_btc = {"total": D("0.05432165"), "exchange_used": D("0E-8"), "exchange_total": D("0.00005959"), "exchange_free": D("0.00005959"), "margin_available": D("0.03019227"), "margin_in_position": D("0.02406979"), "margin_borrowed": 0, "margin_total": D("0.05426206"), "margin_pending_gain": D("0.00093955"), "margin_lending_fees": 0, "margin_pending_base_gain": 0, "margin_position_type": None, "margin_liquidation_price": 0, "margin_borrowed_base_price": 0, "margin_borrowed_base_currency": None } expected_xmr = {"total": D("0.18719303"), "exchange_used": D("0E-8"), "exchange_total": D("0.18719303"), "exchange_free": D("0.18719303"), "margin_available": 0, "margin_in_position": 0, "margin_borrowed": 0, "margin_total": 0, "margin_pending_gain": 0, "margin_lending_fees": 0, "margin_pending_base_gain": 0, "margin_position_type": None, "margin_liquidation_price": 0, "margin_borrowed_base_price": 0, "margin_borrowed_base_currency": None } self.assertEqual(expected_xmr, result["XMR"]) self.assertEqual(expected_doge, result["DOGE"]) self.assertEqual(expected_btc, result["BTC"]) def test_create_order(self): with self.subTest(type="market"),\ self.assertRaises(market.ExchangeError): self.s.create_order("FOO", "market", "buy", "10") with self.subTest(type="limit", account="margin"),\ mock.patch.object(self.s, "load_markets") as load_markets,\ mock.patch.object(self.s, "privatePostMarginBuy") as margin_buy,\ mock.patch.object(self.s, "privatePostMarginSell") as margin_sell,\ mock.patch.object(self.s, "market") as market_mock,\ mock.patch.object(self.s, "price_to_precision") as ptp,\ mock.patch.object(self.s, "amount_to_precision") as atp: margin_buy.return_value = { "orderNumber": 123 } margin_sell.return_value = { "orderNumber": 456 } market_mock.return_value = { "id": "BTC_ETC", "symbol": "BTC_ETC" } ptp.return_value = D("0.1") atp.return_value = D("12") order = self.s.create_order("BTC_ETC", "limit", "buy", "12", account="margin", price="0.1") self.assertEqual(123, order["id"]) margin_buy.assert_called_once_with({"currencyPair": "BTC_ETC", "rate": D("0.1"), "amount": D("12")}) margin_sell.assert_not_called() margin_buy.reset_mock() margin_sell.reset_mock() order = self.s.create_order("BTC_ETC", "limit", "sell", "12", account="margin", lending_rate="0.01", price="0.1") self.assertEqual(456, order["id"]) margin_sell.assert_called_once_with({"currencyPair": "BTC_ETC", "rate": D("0.1"), "amount": D("12"), "lendingRate": "0.01"}) margin_buy.assert_not_called() with self.subTest(type="limit", account="exchange"),\ mock.patch.object(self.s, "load_markets") as load_markets,\ mock.patch.object(self.s, "privatePostBuy") as exchange_buy,\ mock.patch.object(self.s, "privatePostSell") as exchange_sell,\ mock.patch.object(self.s, "market") as market_mock,\ mock.patch.object(self.s, "price_to_precision") as ptp,\ mock.patch.object(self.s, "amount_to_precision") as atp: exchange_buy.return_value = { "orderNumber": 123 } exchange_sell.return_value = { "orderNumber": 456 } market_mock.return_value = { "id": "BTC_ETC", "symbol": "BTC_ETC" } ptp.return_value = D("0.1") atp.return_value = D("12") order = self.s.create_order("BTC_ETC", "limit", "buy", "12", account="exchange", price="0.1") self.assertEqual(123, order["id"]) exchange_buy.assert_called_once_with({"currencyPair": "BTC_ETC", "rate": D("0.1"), "amount": D("12")}) exchange_sell.assert_not_called() exchange_buy.reset_mock() exchange_sell.reset_mock() order = self.s.create_order("BTC_ETC", "limit", "sell", "12", account="exchange", lending_rate="0.01", price="0.1") self.assertEqual(456, order["id"]) exchange_sell.assert_called_once_with({"currencyPair": "BTC_ETC", "rate": D("0.1"), "amount": D("12")}) exchange_buy.assert_not_called() with self.subTest(account="unknown"), self.assertRaises(NotImplementedError),\ mock.patch.object(self.s, "load_markets") as load_markets: self.s.create_order("symbol", "type", "side", "amount", account="unknown") def test_common_currency_code(self): self.assertEqual("FOO", self.s.common_currency_code("FOO")) def test_currency_id(self): self.assertEqual("FOO", self.s.currency_id("FOO")) def test_amount_to_precision(self): self.s.markets = { "FOO": { "precision": { "price": 5, "amount": 6, } }, "BAR": { "precision": { "price": 7, "amount": 8, } } } self.assertEqual("0.0001", self.s.amount_to_precision("FOO", D("0.0001"))) self.assertEqual("0.0000001", self.s.amount_to_precision("BAR", D("0.0000001"))) self.assertEqual("0.000001", self.s.amount_to_precision("FOO", D("0.000001"))) def test_price_to_precision(self): self.s.markets = { "FOO": { "precision": { "price": 5, "amount": 6, } }, "BAR": { "precision": { "price": 7, "amount": 8, } } } self.assertEqual("0.0001", self.s.price_to_precision("FOO", D("0.0001"))) self.assertEqual("0.0000001", self.s.price_to_precision("BAR", D("0.0000001"))) self.assertEqual("0", self.s.price_to_precision("FOO", D("0.000001"))) def test_is_dust_trade(self): self.assertTrue(self.s.is_dust_trade(D("0.0000009"), D("1000"))) self.assertTrue(self.s.is_dust_trade(D("0.000001"), D("10"))) self.assertFalse(self.s.is_dust_trade(D("0.000001"), D("100")))