Key Concepts
Globals and Statics
API Access by Algorithm Instance
You can access the QCAlgorithm API by passing the self
object into a constructor of your target class.
The class constructor receives it as a variable you can use to initialize your algorithm.
The following algorithm demonstrates this process:
class CustomPartialFillModelAlgorithm(QCAlgorithm): def initialize(self) -> None: self.set_start_date(2019, 1, 1) self.set_end_date(2019, 3, 1) # Request SPY data for trading. equity = self.add_equity("SPY", Resolution.HOUR) self._spy = equity.symbol self._holdings = equity.holdings # Set the fill model with the algorithm instance. equity.set_fill_model(CustomPartialFillModel(self)) def on_data(self, data: Slice) -> None: # To get all SPY open orders, use the transaction manager. open_orders = self.transactions.get_open_orders(self.spy) if len(open_orders) != 0: return # Hold SPY during 10th-20th in each month. if self.time.day > 10 and self._holdings.quantity <= 0: self.market_order(self._spy, 105, True) elif self.time.day > 20 and self._holdings.quantity >= 0: self.market_order(self._spy, -100, True) class CustomPartialFillModel(FillModel): def __init__(self, algorithm: QCAlgorithm) -> None: self._algorithm = algorithm self._absolute_remaining_by_order_id = {} def market_fill(self, asset: Security, order: Order) -> OrderEvent: absolute_remaining = self._absolute_remaining_by_order_id.get(order.id, order.absolute_quantity) fill = super().market_fill(asset, order) # Allow at most 10 shares to be filled at once. fill.fill_quantity = np.sign(order.quantity) * 10 # If the remaining shares are less than or equal to 10, the order will be fully filled. if (min(abs(fill.fill_quantity), absolute_remaining) == absolute_remaining): fill.fill_quantity = np.sign(order.quantity) * absolute_remaining fill.status = OrderStatus.FILLED self._absolute_remaining_by_order_id.pop(order.id, None) # Otherwise, the order is only partially filled out, and we need to update the remaining quantity. else: fill.status = OrderStatus.PARTIALLY_FILLED self._absolute_remaining_by_order_id[order.id] = absolute_remaining - abs(fill.fill_quantity) price = fill.fill_price # Use the instance of the algorithm to log the information self._algorithm.debug(f"{self._algorithm.time} - Partial Fill - Remaining {self._absolute_remaining_by_order_id[order.id]} Price - {price}") return fill
API Access by Global Static Variable
Occasionally, passing the QCAlgorithm object to your class constructor is impossible, so you need to use global static variables to access the API for debugging and initialization purposes. The most common case is custom data implementations, where the LEAN Engine is creating the objects, and you're unable to specify a custom constructor.
To create a global static variable, assign the reference to the global variable in the algorithm's initialize
method.
This assignment sets the active instance of the variable to the global static.
Then, in your custom class, you can access the QCAlgorithm API through the global.
The following algorithm demonstrates this process:
class MyAlgorithmInstance(QCAlgorithm): def initialize(self) -> None: self.set_cash(100000) self.set_start_date(1998, 1, 1) self.set_end_date(2014, 6, 1) # Instantiate a custom type as a global variable in the QCAlgorithm instance. # Assign self to Static Cape Variable. Cape.algorithm = self # Request data for trading or signal. self.add_data(Cape, "CAPE") def on_data(self, data: Slice) -> None: self.plot("CAPE", "Value", data["CAPE"].value) class Cape(PythonData): def get_source(self, config: SubscriptionDataConfig, date:datetime, isLiveMode: bool) -> SubscriptionDataSource: Cape.algorithm.debug("Test Static: GetSource") return SubscriptionDataSource("https://www.dropbox.com/scl/fi/mqbnfb7ll88nne7b8ymy7/CAPE.csv?rlkey=mnu0ax1d8lcj3gzkdw79z0pm8&dl=1", SubscriptionTransportMedium.REMOTE_FILE) def reader(self, config: SubscriptionDataConfig, line: str, date:datetime, isLiveMode: bool) -> BaseData: if not (line.strip() and line[0].isdigit()): return None index = Cape() try: # Construct the custom data type with input data entry. data = line.split(',') index.symbol = config.symbol index.time = datetime.strptime(data[0], "%Y-%m") index.value = float(data[10]) except ValueError: Cape.algorithm.debug("Test Static 2: ValueError") return None return index