import requests
import requests_mock
from io import StringIO
-import portfolio, market, main
+import portfolio, market, main, store
limits = ["acceptance", "unit"]
for test_type in limits:
self.m.debug = False
self.patchers = [
- mock.patch.multiple(portfolio.Portfolio, last_date=None, data=None, liquidities={}),
+ mock.patch.multiple(market.Portfolio,
+ last_date=None,
+ data=None,
+ liquidities={},
+ report=mock.Mock()),
mock.patch.multiple(portfolio.Computation,
computations=portfolio.Computation.computations),
]
@unittest.skipUnless("unit" in limits, "Unit skipped")
class PortfolioTest(WebMockTestCase):
- def fill_data(self):
- if self.json_response is not None:
- portfolio.Portfolio.data = self.json_response
-
def setUp(self):
super(PortfolioTest, self).setUp()
with open("test_samples/test_portfolio.json") as example:
self.json_response = example.read()
- self.wm.get(portfolio.Portfolio.URL, text=self.json_response)
+ self.wm.get(market.Portfolio.URL, text=self.json_response)
- def test_get_cryptoportfolio(self):
- self.wm.get(portfolio.Portfolio.URL, [
+ @mock.patch.object(market.Portfolio, "parse_cryptoportfolio")
+ def test_get_cryptoportfolio(self, parse_cryptoportfolio):
+ self.wm.get(market.Portfolio.URL, [
{"text":'{ "foo": "bar" }', "status_code": 200},
{"text": "System Error", "status_code": 500},
{"exc": requests.exceptions.ConnectTimeout},
])
- portfolio.Portfolio.get_cryptoportfolio(self.m)
- self.assertIn("foo", portfolio.Portfolio.data)
- self.assertEqual("bar", portfolio.Portfolio.data["foo"])
+ market.Portfolio.get_cryptoportfolio()
+ self.assertIn("foo", market.Portfolio.data)
+ self.assertEqual("bar", market.Portfolio.data["foo"])
self.assertTrue(self.wm.called)
self.assertEqual(1, self.wm.call_count)
- self.m.report.log_error.assert_not_called()
- self.m.report.log_http_request.assert_called_once()
- self.m.report.log_http_request.reset_mock()
-
- portfolio.Portfolio.get_cryptoportfolio(self.m)
- self.assertIsNone(portfolio.Portfolio.data)
+ market.Portfolio.report.log_error.assert_not_called()
+ market.Portfolio.report.log_http_request.assert_called_once()
+ parse_cryptoportfolio.assert_called_once_with()
+ market.Portfolio.report.log_http_request.reset_mock()
+ parse_cryptoportfolio.reset_mock()
+ market.Portfolio.data = None
+
+ market.Portfolio.get_cryptoportfolio()
+ self.assertIsNone(market.Portfolio.data)
self.assertEqual(2, self.wm.call_count)
- self.m.report.log_error.assert_not_called()
- self.m.report.log_http_request.assert_called_once()
- self.m.report.log_http_request.reset_mock()
-
+ parse_cryptoportfolio.assert_not_called()
+ market.Portfolio.report.log_error.assert_not_called()
+ market.Portfolio.report.log_http_request.assert_called_once()
+ market.Portfolio.report.log_http_request.reset_mock()
+ parse_cryptoportfolio.reset_mock()
+
+ market.Portfolio.data = "Foo"
+ market.Portfolio.get_cryptoportfolio()
+ self.assertEqual(2, self.wm.call_count)
+ parse_cryptoportfolio.assert_not_called()
- portfolio.Portfolio.data = "Foo"
- portfolio.Portfolio.get_cryptoportfolio(self.m)
- self.assertEqual("Foo", portfolio.Portfolio.data)
+ market.Portfolio.get_cryptoportfolio(refetch=True)
+ self.assertEqual("Foo", market.Portfolio.data)
self.assertEqual(3, self.wm.call_count)
- self.m.report.log_error.assert_called_once_with("get_cryptoportfolio",
+ market.Portfolio.report.log_error.assert_called_once_with("get_cryptoportfolio",
exception=mock.ANY)
- self.m.report.log_http_request.assert_not_called()
+ market.Portfolio.report.log_http_request.assert_not_called()
def test_parse_cryptoportfolio(self):
- portfolio.Portfolio.parse_cryptoportfolio(self.m)
+ market.Portfolio.data = store.json.loads(self.json_response, parse_int=D,
+ parse_float=D)
+ market.Portfolio.parse_cryptoportfolio()
self.assertListEqual(
["medium", "high"],
- list(portfolio.Portfolio.liquidities.keys()))
+ list(market.Portfolio.liquidities.keys()))
- liquidities = portfolio.Portfolio.liquidities
+ liquidities = market.Portfolio.liquidities
self.assertEqual(10, len(liquidities["medium"].keys()))
self.assertEqual(10, len(liquidities["high"].keys()))
'XCP': (D("0.1"), "long"),
}
self.assertDictEqual(expected, liquidities["medium"][date])
- self.assertEqual(portfolio.datetime(2018, 1, 15), portfolio.Portfolio.last_date)
-
- self.m.report.log_http_request.assert_called_once_with("GET",
- portfolio.Portfolio.URL, None, mock.ANY, mock.ANY)
- self.m.report.log_http_request.reset_mock()
-
- # It doesn't refetch the data when available
- portfolio.Portfolio.parse_cryptoportfolio(self.m)
- self.m.report.log_http_request.assert_not_called()
-
- self.assertEqual(1, self.wm.call_count)
-
- portfolio.Portfolio.parse_cryptoportfolio(self.m, refetch=True)
- self.assertEqual(2, self.wm.call_count)
- self.m.report.log_http_request.assert_called_once()
-
- def test_repartition(self):
- expected_medium = {
- 'BTC': (D("1.1102e-16"), "long"),
- 'USDT': (D("0.1"), "long"),
- 'ETC': (D("0.1"), "long"),
- 'FCT': (D("0.1"), "long"),
- 'OMG': (D("0.1"), "long"),
- 'STEEM': (D("0.1"), "long"),
- 'STRAT': (D("0.1"), "long"),
- 'XEM': (D("0.1"), "long"),
- 'XMR': (D("0.1"), "long"),
- 'XVC': (D("0.1"), "long"),
- 'ZRX': (D("0.1"), "long"),
- }
- expected_high = {
- 'USDT': (D("0.1226"), "long"),
- 'BTC': (D("0.1429"), "long"),
- 'ETC': (D("0.1127"), "long"),
- 'ETH': (D("0.1569"), "long"),
- 'FCT': (D("0.3341"), "long"),
- 'GAS': (D("0.1308"), "long"),
+ self.assertEqual(portfolio.datetime(2018, 1, 15), market.Portfolio.last_date)
+
+ @mock.patch.object(market.Portfolio, "get_cryptoportfolio")
+ def test_repartition(self, get_cryptoportfolio):
+ market.Portfolio.liquidities = {
+ "medium": {
+ "2018-03-01": "medium_2018-03-01",
+ "2018-03-08": "medium_2018-03-08",
+ },
+ "high": {
+ "2018-03-01": "high_2018-03-01",
+ "2018-03-08": "high_2018-03-08",
+ }
}
+ market.Portfolio.last_date = "2018-03-08"
- self.assertEqual(expected_medium, portfolio.Portfolio.repartition(self.m))
- self.assertEqual(expected_medium, portfolio.Portfolio.repartition(self.m, liquidity="medium"))
- self.assertEqual(expected_high, portfolio.Portfolio.repartition(self.m, liquidity="high"))
-
- self.assertEqual(1, self.wm.call_count)
-
- portfolio.Portfolio.repartition(self.m)
- self.assertEqual(1, self.wm.call_count)
-
- portfolio.Portfolio.repartition(self.m, refetch=True)
- self.assertEqual(2, self.wm.call_count)
- self.m.report.log_http_request.assert_called()
- self.assertEqual(2, self.m.report.log_http_request.call_count)
+ self.assertEqual("medium_2018-03-08", market.Portfolio.repartition())
+ get_cryptoportfolio.assert_called_once_with()
+ self.assertEqual("medium_2018-03-08", market.Portfolio.repartition(liquidity="medium"))
+ self.assertEqual("high_2018-03-08", market.Portfolio.repartition(liquidity="high"))
- @mock.patch.object(portfolio.time, "sleep")
- @mock.patch.object(portfolio.Portfolio, "repartition")
- def test_wait_for_recent(self, repartition, sleep):
+ @mock.patch.object(market.time, "sleep")
+ @mock.patch.object(market.Portfolio, "get_cryptoportfolio")
+ def test_wait_for_recent(self, get_cryptoportfolio, sleep):
self.call_count = 0
- def _repartition(market, refetch):
- self.assertEqual(self.m, market)
- self.assertTrue(refetch)
+ def _get(refetch=False):
+ if self.call_count != 0:
+ self.assertTrue(refetch)
+ else:
+ self.assertFalse(refetch)
self.call_count += 1
- portfolio.Portfolio.last_date = portfolio.datetime.now()\
- - portfolio.timedelta(10)\
- + portfolio.timedelta(self.call_count)
- repartition.side_effect = _repartition
+ market.Portfolio.last_date = store.datetime.now()\
+ - store.timedelta(10)\
+ + store.timedelta(self.call_count)
+ get_cryptoportfolio.side_effect = _get
- portfolio.Portfolio.wait_for_recent(self.m)
+ market.Portfolio.wait_for_recent()
sleep.assert_called_with(30)
self.assertEqual(6, sleep.call_count)
- self.assertEqual(7, repartition.call_count)
- self.m.report.print_log.assert_called_with("Attempt to fetch up-to-date cryptoportfolio")
+ self.assertEqual(7, get_cryptoportfolio.call_count)
+ market.Portfolio.report.print_log.assert_called_with("Attempt to fetch up-to-date cryptoportfolio")
sleep.reset_mock()
- repartition.reset_mock()
- portfolio.Portfolio.last_date = None
+ get_cryptoportfolio.reset_mock()
+ market.Portfolio.last_date = None
self.call_count = 0
- portfolio.Portfolio.wait_for_recent(self.m, delta=15)
+ market.Portfolio.wait_for_recent(delta=15)
sleep.assert_not_called()
- self.assertEqual(1, repartition.call_count)
+ self.assertEqual(1, get_cryptoportfolio.call_count)
sleep.reset_mock()
- repartition.reset_mock()
- portfolio.Portfolio.last_date = None
+ get_cryptoportfolio.reset_mock()
+ market.Portfolio.last_date = None
self.call_count = 0
- portfolio.Portfolio.wait_for_recent(self.m, delta=1)
+ market.Portfolio.wait_for_recent(delta=1)
sleep.assert_called_with(30)
self.assertEqual(9, sleep.call_count)
- self.assertEqual(10, repartition.call_count)
+ self.assertEqual(10, get_cryptoportfolio.call_count)
@unittest.skipUnless("unit" in limits, "Unit skipped")
class AmountTest(WebMockTestCase):
self.assertEqual("Foo", m.fetch_fees())
self.ccxt.fetch_fees.assert_not_called()
- @mock.patch.object(portfolio.Portfolio, "repartition")
+ @mock.patch.object(market.Portfolio, "repartition")
@mock.patch.object(market.Market, "get_ticker")
@mock.patch.object(market.TradeStore, "compute_trades")
def test_prepare_trades(self, compute_trades, get_ticker, repartition):
m.report.log_balances.assert_called_once_with(tag="tag")
- @mock.patch.object(portfolio.time, "sleep")
+ @mock.patch.object(market.time, "sleep")
@mock.patch.object(market.TradeStore, "all_orders")
def test_follow_orders(self, all_orders, time_mock):
for debug, sleep in [
self.assertListEqual(["USDT", "XVG", "XMR", "ETC"], list(balance_store.currencies()))
self.m.report.log_balances.assert_called_with(tag="foo")
- @mock.patch.object(portfolio.Portfolio, "repartition")
+ @mock.patch.object(market.Portfolio, "repartition")
def test_dispatch_assets(self, repartition):
self.m.ccxt.fetch_all_balances.return_value = self.fetch_balance
repartition.return_value = repartition_hash
amounts = balance_store.dispatch_assets(portfolio.Amount("BTC", "11.1"))
- repartition.assert_called_with(self.m, liquidity="medium")
+ repartition.assert_called_with(liquidity="medium")
self.assertIn("XEM", balance_store.currencies())
self.assertEqual(D("2.6"), amounts["BTC"].value)
self.assertEqual(D("7.5"), amounts["XEM"].value)
processor.run_action("wait_for_recent", "bar", "baz")
- method_mock.assert_called_with(self.m, foo="bar")
+ method_mock.assert_called_with(foo="bar")
def test_select_step(self):
processor = market.Processor(self.m)
processor = market.Processor(m)
method, arguments = processor.method_arguments("wait_for_recent")
- self.assertEqual(portfolio.Portfolio.wait_for_recent, method)
+ self.assertEqual(market.Portfolio.wait_for_recent, method)
self.assertEqual(["delta"], arguments)
method, arguments = processor.method_arguments("prepare_trades")
market = mock.Mock()
market.fetch_all_balances.return_value = fetch_balance
market.fetch_ticker.side_effect = fetch_ticker
- with mock.patch.object(portfolio.Portfolio, "repartition", return_value=repartition):
+ with mock.patch.object(market.Portfolio, "repartition", return_value=repartition):
# Action 1
helper.prepare_trades(market)
"amount": "10", "total": "1"
}
]
- with mock.patch.object(portfolio.time, "sleep") as sleep:
+ with mock.patch.object(market.time, "sleep") as sleep:
# Action 4
helper.follow_orders(verbose=False)
}
market.fetch_all_balances.return_value = fetch_balance
- with mock.patch.object(portfolio.Portfolio, "repartition", return_value=repartition):
+ with mock.patch.object(market.Portfolio, "repartition", return_value=repartition):
# Action 5
helper.prepare_trades(market, only="acquire", compute_value="average")
# TODO
# portfolio.TradeStore.run_orders()
- with mock.patch.object(portfolio.time, "sleep") as sleep:
+ with mock.patch.object(market.time, "sleep") as sleep:
# Action 8
helper.follow_orders(verbose=False)
projects / perso / Immae / Projets / Cryptomonnaies / Cryptoportfolio / Trader.git / blobdiff