fisk/core/tests/trading/test_assetfilter.py

from django.test import TestCase

from core.models import AssetGroup, AssetRule, User
from core.trading import assetfilter


class AssetfilterTestCase(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="testuser",
            email="test@example.com",
        )
        self.group = AssetGroup.objects.create(
            user=self.user,
            name="Group1",
            description="Test group",
            when_no_data=6,
        )

    def test_get_allowed_prohibited(self):
        """
        Test that the asset filter works on negative aggregations.
        """
        # We have negative news about EUR
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=3,
        )

        # This means that:
        # * base == EUR: long is not allowed, short is allowed
        # * quote == EUR: long is allowed, short is not allowed
        # We are betting on it going down, so we short base, and long quote.

        # Test that short on base of EUR is allowed
        self.assertTrue(assetfilter.get_allowed(self.group, "EUR", "USD", "short"))

        # Test that long on quote of EUR is allowed
        self.assertTrue(assetfilter.get_allowed(self.group, "USD", "EUR", "long"))

        # Test that long on base of EUR is not allowed
        self.assertFalse(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

        # Test that short on quote of EUR is not allowed
        self.assertFalse(assetfilter.get_allowed(self.group, "USD", "EUR", "short"))

    def test_get_allowed_base_not_listed(self):
        self.group.when_no_data = 7  # Always deny
        self.group.save()
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="JPY",
            status=6,  # Always allow
        )

        self.assertFalse(assetfilter.get_allowed(self.group, "XAG", "JPY", "long"))

    def test_get_allowed_quote_not_listed(self):
        self.group.when_no_data = 7
        self.group.save()

        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="JPY",
            status=6,
        )

        self.assertFalse(assetfilter.get_allowed(self.group, "JPY", "XAG", "long"))

    def test_get_allowed_paranoid_success(self):
        """
        Test that when we request explicit checking of both base and quote,
        it works.
        """
        self.group.when_no_data = 7
        self.group.save()

        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="JPY",
            status=6,
        )
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="XAG",
            status=6,
        )

        self.assertTrue(assetfilter.get_allowed(self.group, "JPY", "XAG", "long"))

    def test_get_allowed_paranoid_success_direction(self):
        """
        Test that when we request explicit checking of both base and quote,
        and specify a direction, it works.
        """
        self.group.when_no_data = 7
        self.group.save()

        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="JPY",
            status=2,  # Positive
        )
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="XAG",
            status=3,  # Negative
        )

        self.assertTrue(assetfilter.get_allowed(self.group, "JPY", "XAG", "long"))

    def test_get_allowed_paranoid_fail_direction(self):
        """
        Test that when we request explicit checking of both base and quote,
        and specify a direction, it works.
        """
        self.group.when_no_data = 7
        self.group.save()

        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="JPY",
            status=3,  # Negative
        )
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="XAG",
            status=2,  # Positive
        )

        self.assertFalse(assetfilter.get_allowed(self.group, "JPY", "XAG", "long"))

    def test_get_allowed_paranoid_fail_direction_short(self):
        """
        Test that when we request explicit checking of both base and quote,
        and specify a direction, it works.
        """
        self.group.when_no_data = 7
        self.group.save()

        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="JPY",
            status=2,  # Positive
        )
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="XAG",
            status=3,  # Negative
        )

        self.assertFalse(assetfilter.get_allowed(self.group, "JPY", "XAG", "short"))

    def test_get_allowed_mixed_sentiment(self):
        """
        Test that deny is hit before allow.
        """
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="JPY",
            status=6,  # Always allow
        )
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="XAG",
            status=7,  # Always deny
        )

        self.assertFalse(assetfilter.get_allowed(self.group, "JPY", "XAG", "short"))
        self.assertFalse(assetfilter.get_allowed(self.group, "XAG", "JPY", "short"))
        self.assertFalse(assetfilter.get_allowed(self.group, "JPY", "XAG", "long"))
        self.assertFalse(assetfilter.get_allowed(self.group, "XAG", "JPY", "long"))

    def test_get_allowed_permitted(self):
        """
        Test that the asset filter works on positive aggregations.
        """
        # We have positive news about EUR
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=2,
        )

        # This means that:
        # * base == EUR: long is allowed, short is not allowed
        # * quote == EUR: long is not allowed, short is allowed
        # We are betting on it going up, so we long base, and short quote.

        # Test that long on base of EUR is allowed
        self.assertTrue(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

        # Test that short on quote of EUR is allowed
        self.assertTrue(assetfilter.get_allowed(self.group, "USD", "EUR", "short"))

        # Test that short on base of EUR is not allowed
        self.assertFalse(assetfilter.get_allowed(self.group, "EUR", "USD", "short"))

        # Test that long on quote of EUR is not allowed
        self.assertFalse(assetfilter.get_allowed(self.group, "USD", "EUR", "long"))

    def test_get_allowed_no_data(self):
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=0,
        )

        self.assertTrue(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

    def test_get_allowed_no_data_prohibited(self):
        self.group.when_no_data = 7
        self.group.save()
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=0,
        )

        self.assertFalse(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

    def test_get_allowed_no_match(self):
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=1,
        )

        self.assertTrue(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

    def test_get_allowed_no_aggregation(self):
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=4,
        )

        self.assertTrue(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

    def test_get_allowed_not_in_bounds(self):
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=5,
        )

        self.assertTrue(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

    def test_get_allowed_always_allow(self):
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=6,
        )

        self.assertTrue(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

    def test_get_allowed_always_deny(self):
        AssetRule.objects.create(
            user=self.user,
            group=self.group,
            asset="EUR",
            status=7,
        )

        self.assertFalse(assetfilter.get_allowed(self.group, "EUR", "USD", "long"))

    def test_check_asset_aggregation(self):
        """
        Test that the asset aggregation works.
        """
        # Test within lower bound
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                1.0, trigger_above=None, trigger_below=2.0
            ),
            3,  # Prohibited
        )

        # Test within upper bound
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                1.0, trigger_above=0.0, trigger_below=None
            ),
            2,  # Permitted
        )

        # Test within bounds
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                1.0, trigger_above=0.0, trigger_below=2.0
            ),
            3,  # Prohibited
        )

        # Test outside lower bound
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                1.0, trigger_above=None, trigger_below=0.0
            ),
            5,  # Not in bounds
        )

        # Test outside upper bound
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                1.0, trigger_above=2.0, trigger_below=None
            ),
            5,  # Not in bounds
        )

        # Test no bounds, just to be sure
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                1.0, trigger_above=None, trigger_below=None
            ),
            4,  # No aggregation
        )

        # Test both bounds, but inverted
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                1.0, trigger_above=2.0, trigger_below=0.0
            ),
            5,  # Not in bounds
        )

        # Test within negative and positive bounds
        self.assertEqual(
            assetfilter.check_asset_aggregation(
                -1.0, trigger_above=-2.0, trigger_below=0.0
            ),
            3,  # Prohibited
        )

    def test_update_status_from_mappings_when_no_data(self):
        self.group.when_no_data = 2  # Bullish
        self.group.save()

        self.assertEqual(
            assetfilter.update_status_from_mappings(0, self.group),
            2,  # Bullish
        )

    def test_update_status_from_mappings_when_no_match(self):
        self.group.when_no_match = 2
        self.group.save()

        self.assertEqual(
            assetfilter.update_status_from_mappings(1, self.group),
            2,  # Bullish
        )

    def test_update_status_from_mappings_when_no_aggregation(self):
        self.group.when_no_aggregation = 2
        self.group.save()

        self.assertEqual(
            assetfilter.update_status_from_mappings(4, self.group),
            2,  # Bullish
        )

    def test_update_status_from_mappings_when_not_in_bounds(self):
        self.group.when_not_in_bounds = 2
        self.group.save()

        self.assertEqual(
            assetfilter.update_status_from_mappings(5, self.group),
            2,  # Bullish
        )

    def test_update_status_from_mappings_when_bullish(self):
        self.group.when_bullish = 3  # Bearish
        self.group.save()

        self.assertEqual(
            assetfilter.update_status_from_mappings(2, self.group),
            3,  # Bearish
        )

    def test_update_status_from_mappings_when_bearish(self):
        self.group.when_bearish = 2  # Bullish
        self.group.save()

        self.assertEqual(
            assetfilter.update_status_from_mappings(3, self.group),
            2,  # Bullish
        )