feature: add cloud live broadcast command

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.

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)

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.")

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)

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()