from decimal import Decimal as D from unittest.mock import patch from django.test import TestCase from core.exchanges.convert import convert_trades from core.models import RiskModel, Trade from core.tests.helpers import ElasticMock, LiveBase from core.trading import market, risk class LiveTradingTestCase(ElasticMock, LiveBase, TestCase): def setUp(self): super(LiveTradingTestCase, self).setUp() self.trade = Trade.objects.create( user=self.user, account=self.account, symbol="EUR_USD", time_in_force="FOK", type="market", amount=10, direction="buy", ) self.commission = 0.025 self.risk_model = RiskModel.objects.create( user=self.user, name="Test Risk Model", max_loss_percent=4, max_risk_percent=2, max_open_trades=3, max_open_trades_per_symbol=2, ) def test_account_functional(self): """ Test that the account is functional. """ balance = self.account.client.get_balance() # We need some money to place trades self.assertTrue(balance > 1000) def open_trade(self, trade=None): if trade: posted = trade.post() else: trade = self.trade posted = self.trade.post() # Check the opened trade self.assertEqual(posted["type"], "MARKET_ORDER") self.assertEqual(posted["symbol"], trade.symbol) self.assertEqual(posted["units"], str(trade.amount)) self.assertEqual(posted["timeInForce"], "FOK") return posted def close_trade(self, trade=None): if trade: trade.refresh_from_db() closed = self.account.client.close_trade(trade.order_id) else: trade = self.trade # refresh the trade to get the trade id self.trade.refresh_from_db() closed = self.account.client.close_trade(self.trade.order_id) # Check the feedback from closing the trade self.assertEqual(closed["type"], "MARKET_ORDER") self.assertEqual(closed["symbol"], trade.symbol) self.assertEqual(closed["units"], str(0 - int(trade.amount))) self.assertEqual(closed["timeInForce"], "FOK") self.assertEqual(closed["reason"], "TRADE_CLOSE") return closed def test_place_close_trade(self): """ Test placing a trade. """ self.open_trade() self.close_trade() def test_get_all_open_trades(self): """ Test getting all open trades. """ self.open_trade() trades = self.account.client.get_all_open_trades() self.trade.refresh_from_db() found = False for trade in trades: if trade["id"] == self.trade.order_id: self.assertEqual(trade["symbol"], "EUR_USD") self.assertEqual(trade["currentUnits"], "10") self.assertEqual(trade["initialUnits"], "10") self.assertEqual(trade["state"], "OPEN") found = True break self.close_trade() if not found: self.fail("Could not find the trade in the list of open trades") def create_complex_trade(self, direction, amount, symbol, tp_percent, sl_percent): eur_usd_price = market.get_price(self.account, direction, symbol) trade_tp = market.get_tp(direction, tp_percent, eur_usd_price) trade_sl = market.get_sl(direction, sl_percent, eur_usd_price) # trade_tsl = market.get_sl("buy", 1, eur_usd_price, return_var=True) # # TP 1% profit # trade_tp = eur_usd_price * D(1.01) # # SL 2% loss # trade_sl = eur_usd_price * D(0.98) # # TSL 1% loss # trade_tsl = eur_usd_price * D(0.99) trade_precision, display_precision = market.get_precision(self.account, symbol) # Round everything to the display precision trade_tp = round(trade_tp, display_precision) trade_sl = round(trade_sl, display_precision) # trade_tsl = round(trade_tsl, display_precision) complex_trade = Trade.objects.create( user=self.user, account=self.account, symbol=symbol, time_in_force="FOK", type="market", amount=amount, direction=direction, take_profit=trade_tp, stop_loss=trade_sl, # trailing_stop_loss=trade_tsl, ) return complex_trade @patch("core.exchanges.oanda.OANDAExchange.get_balance", return_value=100) def test_check_risk_max_risk_pass(self, mock_balance): # SL of 19% on a 10 trade on a 100 account is 1.8 loss # Should be comfortably under 2% risk trade = self.create_complex_trade("buy", 10, "EUR_USD", 1, 18) allowed = risk.check_risk(self.risk_model, self.account, trade) self.assertTrue(allowed["allowed"]) @patch("core.exchanges.oanda.OANDAExchange.get_balance", return_value=100) def test_check_risk_max_risk_fail(self, mock_balance): # SL of 21% on a 10 trade on a 100 account is 2.2 loss # Should be over 2% risk trade = self.create_complex_trade("buy", 10, "EUR_USD", 1, 22) allowed = risk.check_risk(self.risk_model, self.account, trade) self.assertFalse(allowed["allowed"]) self.assertEqual(allowed["reason"], "Maximum risk exceeded.") @patch("core.exchanges.oanda.OANDAExchange.get_balance", return_value=94) # We have lost 6% of our account def test_check_risk_max_loss_fail(self, mock_balance): # Doesn't matter, shouldn't get as far as the trade trade = self.create_complex_trade("buy", 1, "EUR_USD", 1, 1) allowed = risk.check_risk(self.risk_model, self.account, trade) self.assertFalse(allowed["allowed"]) self.assertEqual(allowed["reason"], "Maximum loss exceeded.") @patch("core.exchanges.oanda.OANDAExchange.get_balance", return_value=100) def test_check_risk_max_open_trades_fail(self, mock_balance): # The maximum open trades is 3. Let's open 2 trades # This would not be allowed by the risk model but we're doing it # manually # Don't be confused by the next test. The max open trades check # fails before the symbol one is run, but yes, they would both # fail. trade1 = self.create_complex_trade("buy", 1, "EUR_USD", 1, 1) self.open_trade(trade1) trade2 = self.create_complex_trade("buy", 1, "EUR_USD", 1, 1) self.open_trade(trade2) trade3 = self.create_complex_trade("buy", 1, "EUR_USD", 1, 1) allowed = risk.check_risk(self.risk_model, self.account, trade3) self.assertFalse(allowed["allowed"]) self.assertEqual(allowed["reason"], "Maximum open trades exceeded.") self.close_trade(trade1) self.close_trade(trade2) @patch("core.exchanges.oanda.OANDAExchange.get_balance", return_value=100) def test_check_risk_max_open_trades_per_symbol_fail(self, mock_balance): trade1 = self.create_complex_trade("buy", 1, "EUR_USD", 1, 1) self.open_trade(trade1) trade2 = self.create_complex_trade("buy", 1, "EUR_USD", 1, 1) allowed = risk.check_risk(self.risk_model, self.account, trade2) self.assertFalse(allowed["allowed"]) self.assertEqual(allowed["reason"], "Maximum open trades per symbol exceeded.") self.close_trade(trade1) def test_convert_trades(self): """ Test converting open trades response to Trade-like format. """ complex_trade = self.create_complex_trade("buy", 10, "EUR_USD", 1, 2) self.open_trade(complex_trade) # Get and annotate the trades trades = self.account.client.get_all_open_trades() trades_converted = convert_trades(trades) # Check the converted trades self.assertEqual(len(trades_converted), 1) expected_tp_percent = D(1 - self.commission) expected_sl_percent = D(2 - self.commission) actual_tp_percent = trades_converted[0]["take_profit_percent"] actual_sl_percent = trades_converted[0]["stop_loss_percent"] tp_percent_difference = abs(expected_tp_percent - actual_tp_percent) sl_percent_difference = abs(expected_sl_percent - actual_sl_percent) max_difference = D(0.08) # depends on market conditions self.assertLess(tp_percent_difference, max_difference) self.assertLess(sl_percent_difference, max_difference) # Convert the trades to USD trades_usd = market.convert_trades_to_usd(self.account, trades_converted) # Convert the trade to USD ourselves trade_in_usd = D(trades_usd[0]["amount"]) * D(trades_usd[0]["current_price"]) # It will never be perfect, but let's check it's at least close trade_usd_conversion_difference = ( trades_usd[0]["trade_amount_usd"] - trade_in_usd ) self.assertLess(trade_usd_conversion_difference, D(0.01)) # Check the converted TP and SL values trade_usd_tp_difference = trades_usd[0]["take_profit_usd"] - D(0.1) trade_usd_sl_difference = trades_usd[0]["stop_loss_usd"] - D(0.2) self.assertLess(trade_usd_tp_difference, D(0.01)) self.assertLess(trade_usd_sl_difference, D(0.02)) self.close_trade(complex_trade)