book
Checkout our new book! Hands on AI Trading with Python, QuantConnect, and AWS Learn More arrow

Live Trading

Commands

Introduction

Commands enable you to manually call methods in your live algorithm as it runs. You can command live projects to update your algorithm state, place orders, or run any other logic. Commands are different from notifications because notifications enable you to send information out of your algorithm while commands enable you to send information into your algorithm.

Basic Usage

To receive and handle generic commands in your algorithm, define the on_command method. This method receives the data you want to inject into your algorithm and returns a boolean that represents if the command was successful.

Select Language: 
class BasicCommandAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_benchmark(lambda x: 1) # So the algorithm needs no asset data.

    def on_command(self, data):
        self.log(f'Got command at {self.time} with data: {data}')
        return True

To invoke the on_command event handler, send a payload (for example, {"ticker": "AAPL", "quantity": 1}) through the QuantConnect REST API or LEAN CLI. The keys of the payload become the members of the data object.

Encapsulate Event Handlers

To encapsulate the command handler logic, define isolated classes. Command classes should extend the Command class. Extensions of the Command class must implement a run method, which receives the payload you send and returns a boolean that represents if the command was successful. To add the command to your algorithm, call the add_command method.

Select Language: 
class EncapsulatedCommandAlgorithm(QCAlgorithm):

    def initialize(self):
        MyCommand.ALGORITHM = self
        self.add_command(MyCommand)
        self.set_benchmark(lambda x: 1) # So the algorithm doesn’t need asset data.

    def do_something(self):
        self.log('Something was done!')

class MyCommand(Command):
    ALGORITHM = None
    ticker = None
    quantity = None
    parameters = {}

    def run(self, algorithm):
        algorithm.log(f"ticker: {self.ticker}; quantity: {self.quantity}; parameters: {self.parameters}")
        MyCommand.ALGORITHM.do_something()
        return True

To invoke the run method of the class you define, send a payload with a $type key (for example, {"ticker": "AAPL", "quantity": 1, "$type": "MyCommand"}). The value of the $type key must be the Command class you define. The other keys of the payload set the members of the class. If you don’t provide a $type key, LEAN runs the on_command method instead of the run method in your custom Command class.

To access your algorithm's attributes, you need to create a global static variable, and assign the reference to the global variable in the algorithm's initialize. Then, in the run merhod, you can acess your algorithm's attributes through the global.

Send Command Link

You can create the command in your algorithm and then send yourself an email notification with a link inside it. When you click the link, it then executes the event handler in your algorithm. To create the link, call the link method. The following algorithm demonstrates how send command requests for the on_command event handler:

Select Language: 
class BasicLinkedCommandAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_benchmark(lambda x: 1) # So the algorithm needs no asset data.
        link = self.link({"ticker": "AAPL", "quantity": 1})
        self.notify.email("email@address.com", "Run Command?", f"Click here to run: {link}")

    def on_command(self, data):
        self.log(f'Got command at {self.time} with data: {data}')
        return True

The following algorithm demonstrates how to send command requests for an event handler you encapsulate in a class:

Select Language: 
class EncapsulatedAndLinkedCommandAlgorithm(QCAlgorithm):

    def initialize(self):
        self.add_command(MyCommand)
        self.set_benchmark(lambda x: 1) # So the algorithm doesn’t need asset data.
        potential_command = MyCommand()
        potential_command.ticker = 'AAPL'
        potential_command.quantity = 1
        potential_command.parameters = {"tag": "Signal X"}
        link = self.link(potential_command)
        self.notify.email("email@address.com", "Run Command?", f"Click here to run: {link}")

class MyCommand(Command):
    ticker = None
    quantity = None
    parameters = {}

    def run(self, algorithm):
        algorithm.log(f"ticker: {self.ticker}; quantity: {self.quantity}; parameters: {self.parameters}")
        return True

Broadcast Command

You can create the command in your algorithm and then broadcast it to all live algorithms in your organization, except the source algorithm and commands of unregisted type name. The following algorithm demonstrates how broadcast a command to the on_command event handler in all your live algorithms:

Select Language: 
class BasicBroadcastCommandAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_benchmark(lambda x: 1) # So the algorithm needs no asset data.
        link = self.broadcast_command({"ticker": "AAPL", "quantity": 1})

    def on_command(self, data):
        self.log(f'Got command at {self.time} with data: {data}')
        return True

You can control the command data by encapsulating the event handler in a class. The following algorithm demonstrates how to broadcast a command for an event handler you encapsulate in a class:

Select Language: 
class EncapsulatedAndBroadcastCommandAlgorithm(QCAlgorithm):

    def initialize(self):
        self.add_command(MyCommand)
        self.set_benchmark(lambda x: 1) # So the algorithm doesn't need asset data.
        potential_command = MyCommand()
        potential_command.ticker = 'AAPL'
        potential_command.quantity = 1
        potential_command.parameters = {"tag": "Signal X"}
        broadcast_result = self.broadcast_command(potential_command)

class MyCommand(Command):
    ticker = None
    quantity = None
    parameters = {}

    def run(self, algorithm):
        algorithm.log(f"ticker: {self.ticker}; quantity: {self.quantity}; parameters: {self.parameters}")
        return True

Send Commands by API

To send a command to your algorithm, send a payload through the QuantConnect REST API or LEAN CLI. The following script demonstrates how to send a command with the REST API. In this case, it calls the on_command method. 

from base64 import b64encode
from hashlib import sha256
from time import time
from requests import get, post

# Edit these values:
USER_ID = 0
API_TOKEN = '_____'

BASE_URL = 'https://www.quantconnect.com/api/v2/'

def get_headers():
    # Get timestamp
    timestamp = f'{int(time())}'
    time_stamped_token = f'{API_TOKEN}:{timestamp}'.encode('utf-8')

    # Get hased API token
    hashed_token = sha256(time_stamped_token).hexdigest()
    authentication = f'{USER_ID}:{hashed_token}'.encode('utf-8')
    authentication = b64encode(authentication).decode('ascii')

    # Create headers dictionary.
    return {
        'Authorization': f'Basic {authentication}',
        'Timestamp': timestamp
    }

def create_command(project_id, command):
    return post(
        f'{BASE_URL}/live/commands/create', 
        headers=get_headers(), 
        json={"projectId": project_id, "command": command}
    ).json()

# Provide your project Id and command here:
print(create_command(<project_id>, {'ticker': 'AAPL', 'quantity': 1}))

To get your user Id and API token, see Request API Token.

Broadcast Commands by API

To broadcast a command to your live algorithms, send a payload through the QuantConnect REST API or LEAN CLI. The following script demonstrates how to broadcast a command to all projects in your organization with the REST API. 

from base64 import b64encode
from hashlib import sha256
from time import time
from requests import get, post

# Edit these values:
USER_ID = 0
API_TOKEN = '_____'
ORGANIZATION_ID = '_____'

BASE_URL = 'https://www.quantconnect.com/api/v2/'

def get_headers():
    # Get timestamp
    timestamp = f'{int(time())}'
    time_stamped_token = f'{API_TOKEN}:{timestamp}'.encode('utf-8')

    # Get hased API token
    hashed_token = sha256(time_stamped_token).hexdigest()
    authentication = f'{USER_ID}:{hashed_token}'.encode('utf-8')
    authentication = b64encode(authentication).decode('ascii')

    # Create headers dictionary.
    return {
        'Authorization': f'Basic {authentication}',
        'Timestamp': timestamp
    }

def broadcast_command(command):
    return post(
        f'{BASE_URL}/live/commands/broadcast', 
        headers=get_headers(), 
        json={"organizationId": ORGANIZATION_ID, "command": command}
    ).json()

# Provide the command to call the on_command event handler:
print(broadcast_command({'ticker': 'AAPL', 'quantity': 1}))
# Provide the command with the $type to call encapsulated event handlers:
print(broadcast_command({'$type': 'MyCommand', 'ticker': 'AAPL', 'quantity': 1}))

To get your user Id, API token, and organization Id, see Request API Token and Get Organization Id.

Examples

The following examples demonstrate common practices for implementing live commands.

Example 1: Grey Box

The following algorithm trades a simple EMA cross strategy on SPY. However, instead of direct ordering by the algorithm as a "black box", it sends an email to ask for clicking a confirmation link before ordering, making each order decision an informed manual decision. The algorithm is called a "grey box".

Select Language: 
class LiveCommandAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2024, 8, 12)
        self.set_end_date(2024, 9, 1)
        self.set_cash(1000000)

        # Request SPY data to trade.
        self.spy = self.add_equity("SPY").symbol
        # Create an EMA indicator to generate trade signals.
        self._ema = self.ema(self.spy, 20, Resolution.DAILY)
        # Warm-up indicator for immediate readiness.
        self.warm_up_indicator(self.spy, self._ema, Resolution.DAILY)

    def on_data(self, slice: Slice) -> None:
        bar = slice.bars.get(self.spy)
        if bar and self.live_mode:
            # Trend-following strategy using price and EMA.
            # If the price is above EMA, SPY is in an uptrend, and we buy it.
            # We sent a link to our email address and await confirmation.
            if bar.close > self._ema.current.value and not self.portfolio[self.spy].is_long:
                link = self.link({"ticker": "SPY", "size": 1})
                self.notify.email("email@address.com", "Trade Confirmation Needed", f"Click here to run: {link}")
            elif bar.close < self._ema.current.value and not self.portfolio[self.spy].is_short:
                link = self.link({"ticker": "SPY", "size": -1})
                self.notify.email("email@address.com", "Trade Confirmation Needed", f"Click here to run: {link}")

    def on_command(self, data: DynamicData) -> Optional[bool]:
        # The algorithm will place the order if we click the email link to confirm the trade.
        self.set_holdings(data["ticker"], data["size"])
        return True

Other Examples

For more examples, see the following algorithms:

Demonstration Algorithm
CallbackCommandRegressionAlgorithm.py Python

You can also see our Videos. You can also get in touch with us via Discord.

Did you find this page helpful?

Contribute to the documentation: