Merge pull request #178 from diogomatoschaves/fix-pipeline-restart

Refactor restarting pipeline logic

data/service/app.py

@@ -55,10 +55,9 @@ def startup_task(app):
         cache.set("bearer_token", bearer_token)
 
     for pipeline in active_pipelines:
+        response = start_symbol_trading(pipeline, restart=True)
 
-        response = start_symbol_trading(pipeline)
-
-        if not response["success"] and response["code"] != "SYMBOL_ALREADY_TRADED":
+        if not response["success"]:
             logging.info(f"Pipeline {pipeline.id} could not be started. {response['message']}")
 
     while any(is_pipeline_loading(cache, pipeline.id) for pipeline in active_pipelines):

data/service/blueprints/bots_api/_helpers.py

@@ -87,7 +87,7 @@ def stop_pipeline(pipeline_id, header='', raise_exception=False, nr_retries=3, f
             time.sleep(60 * retries)
 
 
-def start_symbol_trading(pipeline):
+def start_symbol_trading(pipeline, restart=False):
 
     add_pipeline_loading(cache, pipeline.id)
 
@@ -102,15 +102,16 @@ def start_symbol_trading(pipeline):
 
     response = start_stop_symbol_trading(payload, 'start')
 
-    if response["success"]:
-        pipeline.last_entry = None
-        pipeline.save()
-    else:
+    if not response["success"] and not (response["code"] == "SYMBOL_ALREADY_TRADED" and restart):
         pipeline.active = False
         pipeline.open_time = None
         pipeline.save()
 
         return response
+    else:
+        response["success"] = True
+        pipeline.last_entry = None
+        pipeline.save()
 
     executor.submit(
         initialize_data_collection,

data/tests/setup/fixtures/internal_modules.py

@@ -144,7 +144,7 @@ def fake_start_stop_trading(pipeline_id, start_or_stop):
     if start_or_stop == 'stop':
         Pipeline.objects.get(id=pipeline_id).update(active=False)
 
-    return {"success": True, "message": ''}
+    return {"success": True, "message": '', "code": ""}
 
 
 @pytest.fixture
@@ -166,7 +166,7 @@ def mock_start_stop_symbol_trading_success_true_binance_handler(mocker):
     mocker.patch.object(
         data.sources.binance._binance,
         'start_stop_symbol_trading',
-        lambda pipeline_id, start_or_stop: {"success": True, "message": ''},
+        lambda pipeline_id, start_or_stop: {"success": True, "message": '', "code": ""},
     )
 
 
@@ -180,7 +180,7 @@ def mock_start_stop_symbol_trading_success_false(mocker):
     return mocker.patch.object(
         data.service.blueprints.bots_api._helpers,
         'start_stop_symbol_trading',
-        lambda payload, start_or_stop: {"success": False, "message": "Pipeline could not be started."},
+        lambda payload, start_or_stop: {"success": False, "message": "Pipeline could not be started.", "code": ""},
     )