From c6a5550cadf4876b66f57bd820dcaee4ad5ef7da Mon Sep 17 00:00:00 2001 From: cardosofede Date: Fri, 12 Apr 2024 19:59:38 -0300 Subject: [PATCH] (feat) add backend api client --- utils/backend_api_client.py | 161 ++++++++++++++++++++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 utils/backend_api_client.py diff --git a/utils/backend_api_client.py b/utils/backend_api_client.py new file mode 100644 index 0000000..8cb7daf --- /dev/null +++ b/utils/backend_api_client.py @@ -0,0 +1,161 @@ +import requests + + +class BackendAPIClient: + """ + This class is a client to interact with the backend API. The Backend API is a REST API that provides endpoints to + create new Hummingbot instances, start and stop them, add new script and controller config files, and get the status + of the active bots. + """ + _shared_instance = None + + @classmethod + def get_instance(cls, *args, **kwargs) -> "MarketsRecorder": + if cls._shared_instance is None: + cls._shared_instance = BackendAPIClient(*args, **kwargs) + return cls._shared_instance + + def __init__(self, host: str = "localhost", port: int = 8000): + self.host = host + self.port = port + self.base_url = f"http://{self.host}:{self.port}" + + def is_docker_running(self): + """Check if Docker is running.""" + url = f"{self.base_url}/is-docker-running" + response = requests.get(url) + return response.json() + + def pull_image(self, image_name: str): + """Pull a Docker image.""" + url = f"{self.base_url}/pull-image/" + payload = {"image_name": image_name} + response = requests.post(url, json=payload) + return response.json() + + def list_available_images(self, image_name: str): + """List available images by name.""" + url = f"{self.base_url}/available-images/{image_name}" + response = requests.get(url) + return response.json() + + def list_active_containers(self): + """List all active containers.""" + url = f"{self.base_url}/active-containers" + response = requests.get(url) + return response.json() + + def list_exited_containers(self): + """List all exited containers.""" + url = f"{self.base_url}/exited-containers" + response = requests.get(url) + return response.json() + + def clean_exited_containers(self): + """Clean up exited containers.""" + url = f"{self.base_url}/clean-exited-containers" + response = requests.post(url) + return response.json() + + def remove_container(self, container_name: str, archive_locally: bool = True, s3_bucket: str = None): + """Remove a specific container.""" + url = f"{self.base_url}/remove-container/{container_name}" + params = {"archive_locally": archive_locally} + if s3_bucket: + params["s3_bucket"] = s3_bucket + response = requests.post(url, params=params) + return response.json() + + def stop_container(self, container_name: str): + """Stop a specific container.""" + url = f"{self.base_url}/stop-container/{container_name}" + response = requests.post(url) + return response.json() + + def start_container(self, container_name: str): + """Start a specific container.""" + url = f"{self.base_url}/start-container/{container_name}" + response = requests.post(url) + return response.json() + + def create_hummingbot_instance(self, instance_config: dict): + """Create a new Hummingbot instance.""" + url = f"{self.base_url}/create-hummingbot-instance" + response = requests.post(url, json=instance_config) + return response.json() + + def start_bot(self, start_bot_config: dict): + """Start a Hummingbot bot.""" + url = f"{self.base_url}/start-bot" + response = requests.post(url, json=start_bot_config) + return response.json() + + def stop_bot(self, bot_name: str, skip_order_cancellation: bool = False, async_backend: bool = True): + """Stop a Hummingbot bot.""" + url = f"{self.base_url}/stop-bot" + response = requests.post(url, json={"bot_name": bot_name, "skip_order_cancellation": skip_order_cancellation, "async_backend": async_backend}) + return response.json() + + def import_strategy(self, strategy_config: dict): + """Import a trading strategy to a bot.""" + url = f"{self.base_url}/import-strategy" + response = requests.post(url, json=strategy_config) + return response.json() + + def get_bot_status(self, bot_name: str): + """Get the status of a bot.""" + url = f"{self.base_url}/get-bot-status/{bot_name}" + response = requests.get(url) + return response.json() + + def get_bot_history(self, bot_name: str): + """Get the historical data of a bot.""" + url = f"{self.base_url}/get-bot-history/{bot_name}" + response = requests.get(url) + return response.json() + + def get_active_bots_status(self): + """ + Retrieve the cached status of all active bots. + Returns a JSON response with the status and data of active bots. + """ + url = f"{self.base_url}/get-active-bots-status" + response = requests.get(url) + if response.status_code == 200: + return response.json() # Successful request + else: + return response.json() # Handle errors or no data found + + def get_all_controllers_config(self): + """Get all controller configurations.""" + url = f"{self.base_url}/all-controller-configs" + response = requests.get(url) + return response.json() + + def get_available_credentials(self): + """Get available credentials.""" + url = f"{self.base_url}/list-credentials" + response = requests.get(url) + return response.json() + + def get_available_images(self, image_name: str = "hummingbot"): + """Get available images.""" + url = f"{self.base_url}/available-images/{image_name}" + response = requests.get(url) + return response.json()["available_images"] + + def add_script_config(self, script_config: dict): + """Add a new script configuration.""" + url = f"{self.base_url}/add-script-config" + response = requests.post(url, json=script_config) + return response.json() + + def add_controller_config(self, controller_config: dict): + """Add a new controller configuration.""" + url = f"{self.base_url}/add-controller-config" + config = { + "name": controller_config["id"], + "content": controller_config + } + response = requests.post(url, json=config) + return response.json()