Merge pull request #144 from diogomatoschaves/issue#143

Refactor saving of portfolio value snapshot

data/tests/service/blueprints/test_dashboard.py

@@ -209,6 +209,8 @@ def test_get_resources_endpoint(
 
         res = client.get(f'{API_PREFIX}/resources{extra_url}')
 
+        print(res.json)
+
         assert res.json == response
 
     @pytest.mark.parametrize(

data/tests/setup/test_data/sample_data.py

@@ -816,7 +816,14 @@
             'quote_id': 'BTC',
             'price_precision': 2,
             'quantity_precision': 3
-        }
+        },
+        "ETHUSDT": {
+            "base_id": "ETH",
+            "name": "ETHUSDT",
+            "price_precision": 2,
+            "quantity_precision": 3,
+            "quote_id": "USDT",
+        },
     },
     "exchanges": {'binance': {'name': 'binance'}},
     "candleSizes": CANDLE_SIZES_ORDERED

docker-compose.yml

@@ -79,7 +79,7 @@ services:
       - APP_NAME=crypto-bot-execution
       - PORT=5000
       - LOGGER_LEVEL=DEBUG
-      - SNAPSHOTS_INTERVAL=900
+      - SNAPSHOTS_INTERVAL=180
     volumes:
       - ./database/model:/usr/src/app/database/model
   redis:

execution/exchanges/binance/futures/_trading.py

@@ -9,7 +9,7 @@
 from database.model.models import Pipeline
 from execution.exchanges.binance import BinanceTrader
 from execution.service.blueprints.market_data import filter_balances
-from execution.service.cron_jobs.save_pipelines_snapshot import save_pipelines_snapshot
+from execution.service.cron_jobs.save_pipelines_snapshot import save_portfolio_value_snapshot, save_pipeline_snapshot
 from execution.service.helpers.exceptions import SymbolAlreadyTraded, SymbolNotBeingTraded, NoUnits, NegativeEquity, \
     InsufficientBalance
 from execution.service.helpers.exceptions.leverage_setting_fail import LeverageSettingFail
@@ -95,8 +95,6 @@ def stop_symbol_trading(self, symbol, header='', **kwargs):
 
             self.close_pipeline(pipeline_id)
 
-            save_pipelines_snapshot([self, self], pipeline_id=pipeline_id)
-
             self.close_pos(symbol, date=datetime.now(tz=pytz.UTC), header=header, **kwargs)
         except NoUnits:
             logging.info(header + "There's no position to be closed.")
@@ -246,6 +244,9 @@ def _update_net_value(self, symbol, balance, units, pipeline, reducing=False):
         pipeline.current_equity = self.current_equity[symbol]
         pipeline.save()
 
+        if reducing:
+            save_pipeline_snapshot(pipeline_id=pipeline.id)
+
     def _get_symbol_info(self, symbol):
 
         symbol_obj = self.validate_symbol(symbol)

execution/service/app.py

@@ -46,7 +46,7 @@ def get_binance_trader_instance(paper_trading):
 
 def startup_task():
 
-    start_background_scheduler([binance_futures_mock_trader, binance_futures_trader], config_vars)
+    start_background_scheduler(config_vars)
 
     open_positions = Position.objects.filter(pipeline__active=True)
 

execution/service/blueprints/market_data.py

@@ -18,6 +18,7 @@
 testnet_client = BinanceHandler(paper_trading=True)
 
 
+@retry_failed_connection(num_times=2)
 def get_ticker(symbol, paper_trading=False):
     try:
         client.validate_symbol(symbol)
@@ -33,13 +34,22 @@ def filter_balances(balances, coins):
     return [balance for balance in balances if balance['asset'] in coins]
 
 
+@retry_failed_connection(num_times=2)
 def get_balances():
     testnet_balance = filter_balances(testnet_client.futures_account_balance(), ["USDT"])
     live_balance = filter_balances(client.futures_account_balance(), ["USDT"])
 
     return {"testnet": testnet_balance, "live": live_balance}
 
 
+@retry_failed_connection(num_times=2)
+def get_account_data():
+    testnet_balance = testnet_client.futures_account()
+    live_balance = client.futures_account()
+
+    return {"testnet": testnet_balance, "live": live_balance}
+
+
 @market_data.get('/prices')
 def get_current_price():
 

execution/service/cron_jobs/main.py

@@ -3,12 +3,12 @@
 
 from apscheduler.schedulers.background import BackgroundScheduler
 
-from execution.service.cron_jobs.save_pipelines_snapshot import save_pipelines_snapshot
+from execution.service.cron_jobs.save_pipelines_snapshot import save_portfolio_value_snapshot
 from shared.utils.decorators import handle_db_connection_error
 
 
 @handle_db_connection_error
-def start_background_scheduler(binance_trader_objects, config_vars):
+def start_background_scheduler(config_vars):
 
     logging.info('Starting scheduler.')
 
@@ -17,7 +17,7 @@ def start_background_scheduler(binance_trader_objects, config_vars):
     # Create an instance of scheduler and add function.
     scheduler = BackgroundScheduler()
     scheduler.add_job(
-        lambda: save_pipelines_snapshot(binance_trader_objects),
+        lambda: save_portfolio_value_snapshot(),
         "interval",
         seconds=int(interval_between_snapshots)
     )

execution/service/cron_jobs/save_pipelines_snapshot/__init__.py

@@ -1 +1,3 @@
-from execution.service.cron_jobs.save_pipelines_snapshot._save_pipelines_snapshot import save_pipelines_snapshot
+from execution.service.cron_jobs.save_pipelines_snapshot._save_pipelines_snapshot import (
+    save_portfolio_value_snapshot, save_pipeline_snapshot
+)

execution/service/cron_jobs/save_pipelines_snapshot/_save_pipelines_snapshot.py

@@ -5,63 +5,50 @@
 import django
 import pytz
 
-from execution.service.blueprints.market_data import get_ticker, get_balances
+from execution.service.blueprints.market_data import get_account_data
 
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "database.settings")
 django.setup()
 
-from database.model.models import PortfolioTimeSeries, Position
+from database.model.models import PortfolioTimeSeries, Pipeline, Position
 
 
-def save_pipelines_snapshot(binance_trader_objects, pipeline_id=None):
-
+def save_portfolio_value_snapshot():
     logging.debug('Saving pipelines snapshot...')
 
-    filter_dict = {
-        "pipeline__active": True
-    }
-
-    if pipeline_id is not None:
-        filter_dict["pipeline__id"] = pipeline_id
+    open_positions_test = Position.objects.filter(pipeline__active=True, paper_trading=True)
+    open_positions_live = Position.objects.filter(pipeline__active=True, paper_trading=False)
 
-    open_positions = Position.objects.filter(**filter_dict)
+    if len(open_positions_live) == 0 and len(open_positions_test) == 0:
+        return
 
     time = datetime.now(pytz.utc)
 
-    total_current_value = {
-        'live': 0,
-        'testnet': 0
-    }
-    for position in open_positions:
+    symbols = dict()
 
-        binance_obj = binance_trader_objects[0] if position.paper_trading else binance_trader_objects[1]
+    symbols["testnet"] = [{"symbol": position.symbol.name, "pipeline_id": position.pipeline.id} for position in open_positions_test]
+    symbols["live"] = [{"symbol": position.symbol.name, "pipeline_id": position.pipeline.id} for position in open_positions_live]
 
-        symbol = position.symbol.name
+    balances = get_account_data()
 
-        response = get_ticker(position.symbol.name, position.paper_trading)
+    for account_type, account_balances in balances.items():
 
-        if response is None:
-            continue
+        net_account_balance = float(account_balances["totalWalletBalance"]) - float(account_balances["totalUnrealizedProfit"])
+        PortfolioTimeSeries.objects.create(time=time, value=net_account_balance, type=account_type)
 
-        try:
-            current_portfolio_value = position.pipeline.current_equity
+        for symbol in symbols[account_type]:
 
-            PortfolioTimeSeries.objects.create(pipeline=position.pipeline, time=time, value=current_portfolio_value)
+            position = [balance for balance in account_balances["positions"] if balance["symbol"] == symbol["symbol"]][0]
 
-            account_type = 'testnet' if position.pipeline.paper_trading else 'live'
+            save_pipeline_snapshot(symbol["pipeline_id"], float(position["unrealizedProfit"]))
 
-            total_current_value[account_type] += current_portfolio_value
 
-        except TypeError:
-            continue
+def save_pipeline_snapshot(pipeline_id, unrealized_profit=0):
 
-    if pipeline_id is None and len(open_positions) > 0:
-        balances = get_balances()
+    pipeline = Pipeline.objects.get(id=pipeline_id)
 
-        for account_type in balances:
-            for asset in balances[account_type]:
-                if asset['asset'] == 'USDT':
+    time = datetime.now(pytz.utc)
 
-                    current_asset_value = float(asset["availableBalance"]) + total_current_value[account_type]
+    equity = pipeline.current_equity + unrealized_profit
 
-                    PortfolioTimeSeries.objects.create(time=time, value=current_asset_value, type=account_type)
+    PortfolioTimeSeries.objects.create(pipeline=pipeline, time=time, value=equity)

execution/tests/exchanges/binance/futures/test_binance_futures_trader.py

@@ -3,13 +3,14 @@
 import django
 import pytest
 
+from execution.tests.setup.test_data.binance_api_responses import margin_order_creation
+
 with pytest.MonkeyPatch().context() as ctx:
     ctx.setenv("TEST", True)
     from execution.service.helpers.exceptions import SymbolNotBeingTraded, SymbolAlreadyTraded, NegativeEquity, \
     InsufficientBalance
     from execution.exchanges.binance.futures import BinanceFuturesTrader
     from execution.tests.setup.fixtures.external_modules import *
-    from execution.tests.setup.fixtures.internal_modules import mock_futures_symbol_ticker
 
 from shared.utils.exceptions import SymbolInvalid
 
@@ -52,7 +53,6 @@ def test_mock_setup(
     futures_change_leverage,
     futures_create_order,
     futures_account_balance,
-    mock_futures_symbol_ticker
 ):
     return
 

execution/tests/service/cron_jobs/test_execution_cron_jobs.py

@@ -1,6 +1,6 @@
 import pytest
 
-from execution.service.cron_jobs.save_pipelines_snapshot import save_pipelines_snapshot
+from execution.service.cron_jobs.save_pipelines_snapshot import save_portfolio_value_snapshot, save_pipeline_snapshot
 from execution.tests.setup.fixtures.external_modules import binance_handler_market_data_factory
 from shared.utils.tests.fixtures.models import *
 
@@ -15,6 +15,7 @@ def inject_fixture(mock_name, method):
     ("futures_symbol_ticker", "futures_symbol_ticker"),
     ("futures_account_balance", "futures_account_balance"),
     ("futures_position_information", "futures_position_information"),
+    ("futures_account", "futures_account"),
 ]
 
 for method in METHODS:
@@ -23,14 +24,13 @@ def inject_fixture(mock_name, method):
 
 @pytest.fixture
 def test_mock_setup(
-    mocker,
-    create_open_position,
-    create_open_position_paper_trading_pipeline,
+    create_exchange,
     ping,
     init_session,
     futures_symbol_ticker,
     futures_account_balance,
-    futures_position_information
+    futures_position_information,
+    futures_account
 ):
     return
 
@@ -56,37 +56,85 @@ class TestCronJobs:
         [
             pytest.param(
                 [2, 11],
-                {"portfolio_timeseries": 4, "values": [100, 500]},
-                id="no_pipeline_id",
-            ),
-            pytest.param(
-                [2],
-                {"portfolio_timeseries": 1, "values": [100]},
-                id="existent_pipeline_2",
+                {"portfolio_timeseries": 4, "values": [85, 530]},
+                id="pipelines=[2, 11]",
             ),
+        ],
+    )
+    def test_save_portfolio_value_snapshot(
+        self,
+        pipelines,
+        response,
+        test_mock_setup,
+        create_open_position,
+        create_open_position_paper_trading_pipeline,
+    ):
+        save_portfolio_value_snapshot()
+
+        assert PortfolioTimeSeries.objects.count() == response["portfolio_timeseries"]
+
+        for i, pipeline in enumerate(pipelines):
+            entries = PortfolioTimeSeries.objects.filter(pipeline_id=pipeline).last()
+            assert entries.value == response["values"][i]
+
+    @pytest.mark.parametrize(
+        "pipelines,response",
+        [
             pytest.param(
-                [11],
-                {"portfolio_timeseries": 1, "values": [500.0]},
-                id="existent_pipeline_11",
+                [],
+                0,
+                id="no_pipeline_id",
             ),
         ],
     )
-    def test_save_pipelines_snapshot(
+    def test_save_portfolio_value_snapshot_no_open_positions(
         self,
         pipelines,
         response,
         test_mock_setup,
     ):
-        binance_instances = [MockBinanceInstance(), MockBinanceInstance(paper_trading=True)]
+        save_portfolio_value_snapshot()
 
-        kwargs = {}
-        if len(pipelines) == 1:
-            kwargs['pipeline_id'] = pipelines[0]
+        assert PortfolioTimeSeries.objects.count() == response
 
-        save_pipelines_snapshot(binance_instances, **kwargs)
+    @pytest.mark.parametrize(
+        "pipeline,unrealized_profit",
+        [
+            pytest.param(
+                2,
+                0,
+                id="pipeline_id=2-unrealized_profit=0",
+            ),
+            pytest.param(
+                2,
+                -15,
+                id="pipeline_id=2-unrealized_profit=-15",
+            ),
+            pytest.param(
+                11,
+                0,
+                id="pipeline_id=11-unrealized_profit=0",
+            ),
+            pytest.param(
+                11,
+                50,
+                id="pipeline_id=11-unrealized_profit=50",
+            ),
+        ],
+    )
+    def test_save_pipeline_snapshot(
+        self,
+        pipeline,
+        unrealized_profit,
+        create_open_position,
+        create_open_position_paper_trading_pipeline,
+        test_mock_setup,
+    ):
+        save_pipeline_snapshot(pipeline, unrealized_profit)
 
-        assert PortfolioTimeSeries.objects.count() == response["portfolio_timeseries"]
+        assert PortfolioTimeSeries.objects.count() == 1
 
-        for i, pipeline in enumerate(pipelines):
-            entries = PortfolioTimeSeries.objects.filter(pipeline_id=pipeline)
-            assert entries[0].value == response["values"][i]
+        entry = PortfolioTimeSeries.objects.filter(pipeline_id=pipeline).last()
+
+        assert entry.pipeline.id == pipeline
+        assert entry.value == entry.pipeline.current_equity + unrealized_profit