Write tests for crossfilter sub-functions

Makefile

@@ -11,7 +11,7 @@ log:
 	docker-compose --env-file=stack.env logs -f
 
 test:
-	docker-compose --env-file=stack.env run --rm app sh -c ". /venv/bin/activate && python manage.py test $(MODULES)"
+	docker-compose --env-file=stack.env run --rm app sh -c ". /venv/bin/activate && python manage.py test $(MODULES) -v 2"
 
 migrate:
 	docker-compose --env-file=stack.env run --rm app sh -c ". /venv/bin/activate && python manage.py migrate"

core/lib/market.py

@@ -8,7 +8,82 @@ from core.util import logs
 log = logs.get_logger(__name__)
 
 
-def crossfilter(account, symbol, direction, func):
+def check_existing_position(
+    func: str,
+    position: dict,
+    open_side: str,
+    open_symbol: str,
+    open_units: str,
+    new_side: str,
+    new_symbol: str,
+    trade_side_opposite: str,
+):
+    # Check if we already have a position for the symbol
+    if open_symbol == new_symbol:
+        # If the live side is the inverse of what we want to do,
+        # we can't open a position
+        if open_side == trade_side_opposite:
+            # If there is a position open, we can't open a new one in the opposite
+            # direction
+            if open_units != "0":
+                # If we have a short on GBP/AUD, we can only place more shorts on
+                # GBP/AUD.
+                if func == "entry":
+                    log.debug(
+                        f"Refusing to open new {new_side} position on {new_symbol} due "
+                        f"to {open_side} position on {open_symbol}"
+                    )
+                    return {
+                        "action": "rejected",
+                        "positions": position,
+                    }
+                elif func == "exit":
+                    log.debug(
+                        (
+                            f"Found {open_units} units of "
+                            f"{open_symbol} on side {trade_side_opposite}"
+                        )
+                    )
+                    # Pass back opposing side so we can close it
+                    return {
+                        "action": "close",
+                        "side": trade_side_opposite,
+                        "positions": position,
+                    }
+    return False
+
+
+def check_conflicting_position(
+    func: str,
+    position: dict,
+    open_base: str,
+    open_quote: str,
+    open_side: str,
+    open_symbol: str,
+    open_units: str,
+    new_base: str,
+    new_quote: str,
+    new_side: str,
+    new_symbol: str,
+    trade_side_opposite: str,
+):
+    if open_base == new_quote or open_quote == new_base:
+        # If we have a long on GBP/AUD, we can only place shorts on XAU/GBP.
+        if open_side != trade_side_opposite:
+            if open_units != "0":
+                # Only do this for entries
+                if func == "entry":
+                    log.debug(
+                        f"Refusing to open {new_side} position on {new_symbol} due to "
+                        f"{open_side} position on {open_symbol}"
+                    )
+                    return {
+                        "action": "rejected",
+                        "positions": position,
+                    }
+
+
+def crossfilter(account, new_symbol, new_direction, func):
     """
     Determine if we are betting against ourselves.
     Checks open positions for the account, rejecting the trade if there is one
@@ -20,7 +95,11 @@ def crossfilter(account, symbol, direction, func):
     :return: dict of action and opposing position, or False
     """
     try:
-        position_info = account.client.get_position_info(symbol)
+        # Only get the data we need
+        if func == "entry":
+            all_positions = account.client.get_all_positions()
+        else:
+            all_positions = [account.client.get_position_info(new_symbol)]
     except GenericAPIError as e:
         if "No position exists for the specified instrument" in str(e):
             log.debug("No position exists for this symbol")
@@ -28,28 +107,61 @@ def crossfilter(account, symbol, direction, func):
         else:
             log.error(f"Error getting position info: {e}")
             return None
-    if direction == "buy":
+    if new_direction == "buy":
         opposing_side = "short"
-    elif direction == "sell":
+        new_side = "long"
+    elif new_direction == "sell":
         opposing_side = "long"
+        new_side = "short"
 
-    opposing_position_info = position_info[opposing_side]
+    quotes = []
-    if opposing_position_info["units"] != "0":
+    new_base, new_quote = new_symbol.split("_")
-        if func == "entry":
+    for position in all_positions:
-            return {"action": "rejected", "positions": opposing_position_info}
+        # For Forex, get a list of all the quotes.
-        elif func == "exit":
+        # This is to prevent betting against ourselves.
-            log.debug(
+        # Consider we have a long position open, EUR/USD, and we want to open a
-                (
+        # long position on USD/JPY. If the first goes up, the second one will go
-                    f"Found {opposing_position_info['units']} units of "
+        # down just as much. We won't make any money.
-                    f"{symbol} on side {opposing_side}"
+        if "_" in position["symbol"]:
+            open_base, open_quote = position["symbol"].split("_")
+            quotes.append(open_quote)
+
+        open_symbol = position["symbol"]
+        open_side = position["side"]
+        open_base, open_quote = open_symbol.split("_")
+
+        # Check if we already have a position
+        existing_position_check = check_existing_position(
+            func=func,
+            position=position,
+            open_side=open_side,
+            open_symbol=open_symbol,
+            open_units=position["units"],
+            new_side=new_side,
+            new_symbol=new_symbol,
+            trade_side_opposite=opposing_side,
         )
+        if existing_position_check:
+            return existing_position_check
+
+        # Check if we are betting against ourselves
+        conflicting_position_check = check_conflicting_position(
+            func=func,
+            position=position,
+            open_base=open_base,
+            open_quote=open_quote,
+            open_side=open_side,
+            open_symbol=open_symbol,
+            open_units=position["units"],
+            new_base=new_base,
+            new_quote=new_quote,
+            new_side=new_side,
+            new_symbol=new_symbol,
+            trade_side_opposite=opposing_side,
         )
-            # Pass back opposing side so we can close it
+        if conflicting_position_check:
-            return {
+            return conflicting_position_check
-                "action": "close",
+
-                "side": opposing_side,
-                "positions": opposing_position_info,
-            }
     return False
 
 

core/tests/lib/test_market.py

@@ -0,0 +1,248 @@
+from django.test import TestCase
+
+from core.lib.market import check_conflicting_position, check_existing_position
+
+
+class MarketTestCase(TestCase):
+    def test_conflict_position(self):
+        position = {
+            "symbol": "EUR_USD",
+            "base": "EUR",
+            "quote": "USD",
+            "side": "long",
+            "units": "1",
+        }
+        trade_cast = {
+            "symbol": "GBP_EUR",
+            "base": "GBP",
+            "quote": "EUR",
+            "side": "long",
+            "units": "1",
+        }
+        opposite = "short"
+
+        check = check_conflicting_position(
+            func="entry",
+            position=position,
+            open_base=position["base"],
+            open_quote=position["quote"],
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_base=trade_cast["base"],
+            new_quote=trade_cast["quote"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+        self.assertEqual(check["action"], "rejected")
+
+    def test_conflict_position_quote(self):
+        position = {
+            "symbol": "EUR_USD",
+            "base": "EUR",
+            "quote": "USD",
+            "side": "long",
+            "units": "1",
+        }
+        trade_cast = {
+            "symbol": "USD_JPY",
+            "base": "USD",
+            "quote": "JPY",
+            "side": "long",
+            "units": "1",
+        }
+        opposite = "short"
+
+        check = check_conflicting_position(
+            func="entry",
+            position=position,
+            open_base=position["base"],
+            open_quote=position["quote"],
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_base=trade_cast["base"],
+            new_quote=trade_cast["quote"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+        self.assertEqual(check["action"], "rejected")
+
+    def test_conflict_position_invert_allowed(self):
+        position = {
+            "symbol": "EUR_USD",
+            "base": "EUR",
+            "quote": "USD",
+            "side": "long",
+            "units": "1",
+        }
+        trade_cast = {
+            "symbol": "GBP_EUR",
+            "base": "GBP",
+            "quote": "EUR",
+            "side": "short",
+            "units": "1",
+        }
+        opposite = "long"
+
+        check = check_conflicting_position(
+            func="entry",
+            position=position,
+            open_base=position["base"],
+            open_quote=position["quote"],
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_base=trade_cast["base"],
+            new_quote=trade_cast["quote"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+        self.assertFalse(check)
+
+    def test_conflict_position_quote_invert_allowed(self):
+        position = {
+            "symbol": "EUR_USD",
+            "base": "EUR",
+            "quote": "USD",
+            "side": "short",
+            "units": "1",
+        }
+        trade_cast = {
+            "symbol": "USD_JPY",
+            "base": "USD",
+            "quote": "JPY",
+            "side": "long",
+            "units": "1",
+        }
+        opposite = "short"
+
+        check = check_conflicting_position(
+            func="entry",
+            position=position,
+            open_base=position["base"],
+            open_quote=position["quote"],
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_base=trade_cast["base"],
+            new_quote=trade_cast["quote"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+        self.assertFalse(check)
+
+    def test_conflict_position_identical_allowed(self):
+        position = {
+            "symbol": "EUR_USD",
+            "base": "EUR",
+            "quote": "USD",
+            "side": "long",
+            "units": "1",
+        }
+        trade_cast = {
+            "symbol": "EUR_USD",
+            "base": "EUR",
+            "quote": "USD",
+            "side": "long",
+            "units": "1",
+        }
+        opposite = "short"
+
+        check = check_conflicting_position(
+            func="entry",
+            position=position,
+            open_base=position["base"],
+            open_quote=position["quote"],
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_base=trade_cast["base"],
+            new_quote=trade_cast["quote"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+        self.assertFalse(check)
+
+    def test_existing_position_fail(self):
+        """
+        Check we cannot enter a short if we already have a long.
+        """
+        position = {"symbol": "EUR_USD", "side": "long", "units": "1"}
+        trade_cast = {"symbol": "EUR_USD", "side": "short", "units": "1"}
+        opposite = "long"
+
+        check = check_existing_position(
+            func="entry",
+            position=position,
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+        self.assertEqual(check["action"], "rejected")
+
+    def test_existing_position_flip_fail(self):
+        """
+        Check we cannot enter a long if we already have a short.
+        """
+        position = {"symbol": "EUR_USD", "side": "short", "units": "1"}
+        trade_cast = {"symbol": "EUR_USD", "side": "long", "units": "1"}
+        opposite = "short"
+
+        check = check_existing_position(
+            func="entry",
+            position=position,
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+        self.assertEqual(check["action"], "rejected")
+
+    def test_existing_position_exit(self):
+        position = {"symbol": "EUR_USD", "side": "long", "units": "1"}
+        trade_cast = {"symbol": "EUR_USD", "side": "short", "units": "1"}
+        opposite = "long"
+
+        check = check_existing_position(
+            func="exit",
+            position=position,
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+
+        self.assertEqual(check["action"], "close")
+        self.assertEqual(check["positions"], position)
+        self.assertEqual(check["side"], opposite)
+
+    def test_existing_position_exit_no_units(self):
+        position = {"symbol": "EUR_USD", "side": "long", "units": "0"}
+        trade_cast = {"symbol": "EUR_USD", "side": "short", "units": "1"}
+        opposite = "long"
+
+        check = check_existing_position(
+            func="exit",
+            position=position,
+            open_side=position["side"],
+            open_symbol=position["symbol"],
+            open_units=position["units"],
+            new_side=trade_cast["side"],
+            new_symbol=trade_cast["symbol"],
+            trade_side_opposite=opposite,
+        )
+
+        self.assertFalse(check)