From 7a522e603a5c763753524ce3b6bdedb9dba4674e Mon Sep 17 00:00:00 2001 From: Alexandre Catarino Date: Wed, 12 Aug 2026 20:02:37 +0100 Subject: [PATCH] feature: add cloud live broadcast command The API exposes live/commands/broadcast to send a command to every live algorithm in an organization, but the CLI could only target a single project through live/commands/create. Add `lean cloud live broadcast`, which sends the command given by --data to all live deployments of an organization. The organization defaults to the one of the current Lean CLI directory and can be overridden with --organization, and --exclude-project leaves a single project out of the broadcast. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 24 ++++++ lean/commands/cloud/live/__init__.py | 2 + lean/commands/cloud/live/broadcast.py | 56 +++++++++++++ lean/components/api/live_client.py | 21 +++++ .../cloud/live/test_cloud_live_commands.py | 78 ++++++++++++++++++- 5 files changed, 180 insertions(+), 1 deletion(-) create mode 100644 lean/commands/cloud/live/broadcast.py diff --git a/README.md b/README.md index f3ba5734..2ce9acad 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,7 @@ The following CLI configurations are available. Use the [`lean config list`](#le - [`lean build`](#lean-build) - [`lean cloud backtest`](#lean-cloud-backtest) - [`lean cloud live`](#lean-cloud-live) +- [`lean cloud live broadcast`](#lean-cloud-live-broadcast) - [`lean cloud live command`](#lean-cloud-live-command) - [`lean cloud live deploy`](#lean-cloud-live-deploy) - [`lean cloud live liquidate`](#lean-cloud-live-liquidate) @@ -352,12 +353,35 @@ Options: --help Show this message and exit. Commands: + broadcast Broadcast a command to all running cloud live trading projects in an organization. command Send a command to a running cloud live trading project. deploy Start live trading for a project in the cloud. liquidate Stops live trading and liquidates existing positions for a certain project. stop Stops live trading for a certain project without liquidating existing positions. ``` +### `lean cloud live broadcast` + +Broadcast a command to all running cloud live trading projects in an organization. + +``` +Usage: lean cloud live broadcast [OPTIONS] + + Broadcast a command to all running cloud live trading projects in an organization. + +Options: + --data TEXT The command to send, 'str' representation of a 'dict' e.g. "{ \"target\": \"BTCUSD\", + \"$type\":\"MyCommand\" }" [required] + --organization TEXT The name or id of the organization to broadcast the command to, defaults to the organization + of the current Lean CLI directory + --exclude-project TEXT The name or id of the project to exclude from the broadcast, by default all projects are + included + --verbose Enable debug logging + --help Show this message and exit. +``` + +_See code: [lean/commands/cloud/live/broadcast.py](lean/commands/cloud/live/broadcast.py)_ + ### `lean cloud live command` Send a command to a running cloud live trading project. diff --git a/lean/commands/cloud/live/__init__.py b/lean/commands/cloud/live/__init__.py index 697bab85..6f2cb64e 100644 --- a/lean/commands/cloud/live/__init__.py +++ b/lean/commands/cloud/live/__init__.py @@ -15,10 +15,12 @@ from lean.commands.cloud.live.deploy import deploy from lean.commands.cloud.live.stop import stop from lean.commands.cloud.live.command import command +from lean.commands.cloud.live.broadcast import broadcast from lean.commands.cloud.live.liquidate import liquidate live.add_command(deploy) live.add_command(stop) live.add_command(command) +live.add_command(broadcast) live.add_command(liquidate) diff --git a/lean/commands/cloud/live/broadcast.py b/lean/commands/cloud/live/broadcast.py new file mode 100644 index 00000000..4d8a3a28 --- /dev/null +++ b/lean/commands/cloud/live/broadcast.py @@ -0,0 +1,56 @@ +# QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. +# Lean CLI v1.0. Copyright 2021 QuantConnect Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +from typing import Optional + +from lean.container import container +from click import command, option +from lean.click import LeanCommand + + +@command(cls=LeanCommand, name="broadcast") +@option("--data", type=str, required=True, + help="The command to send, 'str' representation of a 'dict' e.g. " + "\"{ \\\"target\\\": \\\"BTCUSD\\\", \\\"$type\\\":\\\"MyCommand\\\" }\"") +@option("--organization", type=str, + help="The name or id of the organization to broadcast the command to, " + "defaults to the organization of the current Lean CLI directory") +@option("--exclude-project", type=str, + help="The name or id of the project to exclude from the broadcast, by default all projects are included") +def broadcast(data: str, organization: Optional[str], exclude_project: Optional[str]) -> None: + """ + Broadcast a command to all running cloud live trading projects in an organization. + """ + data = eval(data) + + logger = container.logger + api_client = container.api_client + + if organization is not None: + from lean.commands.init import _get_organization_id + organization_id, _ = _get_organization_id(organization) + else: + organization_id = container.organization_manager.try_get_working_organization_id() + + exclude_project_id = None + if exclude_project is not None: + cloud_project_manager = container.cloud_project_manager + exclude_project_id = cloud_project_manager.get_cloud_project(exclude_project, False).projectId + + logger.info(f"cloud.live.broadcast(): broadcasting command.") + response = api_client.live.broadcast_command(organization_id, data, exclude_project_id) + if response.success: + logger.info(f"cloud.live.broadcast(): command broadcasted successfully.") + else: + raise Exception("cloud.live.broadcast(): Failed: to broadcast the command successfully.") diff --git a/lean/components/api/live_client.py b/lean/components/api/live_client.py index 9e6b6bc9..c8bbaaf4 100644 --- a/lean/components/api/live_client.py +++ b/lean/components/api/live_client.py @@ -134,3 +134,24 @@ def command_create(self, project_id: int, command: dict) -> QCRestResponse: "command": command }) return QCRestResponse(**data) + + def broadcast_command(self, + organization_id: str, + command: dict, + exclude_project_id: Optional[int] = None) -> QCRestResponse: + """Broadcasts a command to all live trading deployments in an organization + + :param organization_id: the id of the organization to broadcast the command to + :param command: the command to send + :param exclude_project_id: the id of the project to exclude from the broadcast, None to include all projects + """ + parameters = { + "organizationId": organization_id, + "command": command + } + + if exclude_project_id is not None: + parameters["excludeProjectId"] = exclude_project_id + + data = self._api.post("live/commands/broadcast", parameters) + return QCRestResponse(**data) diff --git a/tests/commands/cloud/live/test_cloud_live_commands.py b/tests/commands/cloud/live/test_cloud_live_commands.py index 2a4bd04e..c6740e27 100644 --- a/tests/commands/cloud/live/test_cloud_live_commands.py +++ b/tests/commands/cloud/live/test_cloud_live_commands.py @@ -20,7 +20,7 @@ from lean.container import container from lean.models.api import QCEmailNotificationMethod, QCWebhookNotificationMethod, QCSMSNotificationMethod, \ QCTelegramNotificationMethod, QCAuth0Authorization -from tests.test_helpers import create_fake_lean_cli_directory, create_qc_nodes +from tests.test_helpers import create_fake_lean_cli_directory, create_qc_nodes, create_api_organization from tests.commands.test_live import brokerage_required_options brokerage_required_options = { @@ -63,6 +63,82 @@ def test_cloud_live_liquidate() -> None: assert result.exit_code == 0 +def test_cloud_live_command() -> None: + create_fake_lean_cli_directory() + + api_client = mock.Mock() + container.api_client = api_client + + cloud_project_manager = mock.Mock(get_cloud_project=mock.Mock(return_value=mock.Mock(projectId=123))) + container.cloud_project_manager = cloud_project_manager + + result = CliRunner().invoke(lean, ["cloud", "live", "command", "Python Project", + "--data", '{ "$type": "MyCommand" }']) + + assert result.exit_code == 0 + + api_client.live.command_create.assert_called_once_with(123, {"$type": "MyCommand"}) + +def test_cloud_live_broadcast() -> None: + create_fake_lean_cli_directory() + + api_client = mock.Mock() + container.api_client = api_client + + result = CliRunner().invoke(lean, ["cloud", "live", "broadcast", "--data", '{ "$type": "MyCommand" }']) + + assert result.exit_code == 0 + + # "abc" is the organization id of the fake Lean CLI directory + api_client.live.broadcast_command.assert_called_once_with("abc", {"$type": "MyCommand"}, None) + +def test_cloud_live_broadcast_excludes_given_project() -> None: + create_fake_lean_cli_directory() + + api_client = mock.Mock() + container.api_client = api_client + + cloud_project_manager = mock.Mock(get_cloud_project=mock.Mock(return_value=mock.Mock(projectId=123))) + container.cloud_project_manager = cloud_project_manager + + result = CliRunner().invoke(lean, ["cloud", "live", "broadcast", "--data", '{ "$type": "MyCommand" }', + "--exclude-project", "Python Project"]) + + assert result.exit_code == 0 + + cloud_project_manager.get_cloud_project.assert_called_once_with("Python Project", False) + api_client.live.broadcast_command.assert_called_once_with("abc", {"$type": "MyCommand"}, 123) + +def test_cloud_live_broadcast_uses_given_organization() -> None: + create_fake_lean_cli_directory() + + organization = create_api_organization() + + api_client = mock.Mock() + api_client.organizations.get_all.return_value = [organization] + container.api_client = api_client + + result = CliRunner().invoke(lean, ["cloud", "live", "broadcast", "--data", '{ "$type": "MyCommand" }', + "--organization", organization.name]) + + assert result.exit_code == 0 + + api_client.live.broadcast_command.assert_called_once_with(organization.id, {"$type": "MyCommand"}, None) + +def test_cloud_live_broadcast_aborts_when_organization_not_found() -> None: + create_fake_lean_cli_directory() + + api_client = mock.Mock() + api_client.organizations.get_all.return_value = [create_api_organization()] + container.api_client = api_client + + result = CliRunner().invoke(lean, ["cloud", "live", "broadcast", "--data", '{ "$type": "MyCommand" }', + "--organization", "not-a-member"]) + + assert result.exit_code != 0 + + api_client.live.broadcast_command.assert_not_called() + def test_cloud_live_deploy() -> None: create_fake_lean_cli_directory()