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
OnCommand
method.
This method receives the data you want to inject into your algorithm and returns a boolean that represents if the command was successful.
public class BasicCommandAlgorithm : QCAlgorithm { public override void Initialize() { SetBenchmark(x => 1); // So the algorithm needs no asset data. } public override bool? OnCommand(dynamic data) { Log($"Got command at {Time} with data: {data}"); return true; } }
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
OnCommand
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
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
AddCommand
method.
public class EncapsulatedCommandAlgorithm : QCAlgorithm { public override void Initialize() { AddCommand<MyCommand>(); SetBenchmark(x => 1); // So the algorithm doesn’t need asset data. } } public class MyCommand : Command { public string Ticker { get; set; } public int Quantity { get; set; } public Dictionary<string, string> Parameters { get; set; } public override bool? Run(IAlgorithm algorithm) { ((QCAlgorithm)algorithm).Log($"ticker: {Ticker}; quantity: {Quantity}; parameters: {Parameters}"); return true; } }
class EncapsulatedCommandAlgorithm(QCAlgorithm): def initialize(self): self.add_command(MyCommand) self.set_benchmark(lambda x: 1) # So the algorithm doesn’t need asset data. 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
To invoke the run
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
OnCommand
method instead of the run
Run
method in your custom Command
class.
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
Link
method.
The following algorithm demonstrates how send command requests for the on_command
event handler:
public class BasicLinkedCommandAlgorithm : QCAlgorithm { public override void Initialize() { SetBenchmark(x => 1); // So the algorithm needs no asset data. var link = Link(new {Ticker = "AAPL", Quantity = 1}); Notify.Email("email@address.com", "Run Command?", $"Click here to run: {link}"); } public override bool? OnCommand(dynamic data) { Log($"Got command at {Time} with data: {data}"); return true; } }
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:
public class EncapsulatedAndLinkedCommandAlgorithm : QCAlgorithm { public override void Initialize() { AddCommand<MyCommand>(); SetBenchmark(x => 1); // So the algorithm doesn’t need asset data. var potentialCommand = new MyCommand { Ticker = "AAPL", Quantity = 1, Parameters = new() { {"tag", "Signal X"} } }; var link = Link(potentialCommand); Notify.Email("email@address.com", "Run Command?", $"Click here to run: {link}"); } } public class MyCommand : Command { public string Ticker { get; set; } public int Quantity { get; set; } public Dictionary<string, string> Parameters { get; set; } public override bool? Run(IAlgorithm algorithm) { ((QCAlgorithm)algorithm).Log($"ticker: {Ticker}; quantity: {Quantity}; parameters: {Parameters}"); return true; } }
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
Send Commands by API
To send a command to your algorithm, send a payload through the QuantConnect REST API or LEAN CLI.
LEAN converts the first character in each key of the payload to uppercase before injecting the data into your algorithm.
The following script demonstrates how to send a command with the REST API.
In this case, it calls the on_command
OnCommand
method.
from base64 import b64encode from hashlib import sha256 from time import time from requests import get, post # Edit these values: USER_ID = _____ 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(), data={"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.