Problem Description
Solution
To implement a trailing stop loss in QuantConnect, you generally have two options depending on your algorithm's structure:
- Manual Logic (Virtual Trailing Stop): You track the highest price (for longs) or lowest price (for shorts) since entry within
on_dataand liquidate if the price crosses your threshold. This gives you the most control. - Risk Management Model: If you are using the Algorithm Framework, you can use the built-in
TrailingStopRiskManagementModel.
Below is a complete, robust example using the Manual Logic approach. This method is preferred for standard algorithms as it does not require rewriting your strategy into the specific Framework structure and avoids spamming the brokerage with order updates.
Strategy: Manual Trailing Stop Implementation
This algorithm buys SPY and BTCUSDT. It tracks the "High Water Mark" (highest price seen while holding) for long positions and triggers a liquidation if the price drops 5% from that high.
# region imports
from AlgorithmImports import *
# endregion
class ManualTrailingStopAlgorithm(QCAlgorithm):
def initialize(self):
self.set_start_date(2023, 1, 1)
self.set_end_date(2024, 1, 1)
self.set_cash(100000)
# Add Equities and Crypto
self.spy = self.add_equity("SPY", Resolution.MINUTE).symbol
self.btc = self.add_crypto("BTCUSDT", Resolution.MINUTE).symbol
# Trailing Stop Settings
self.trailing_percent = 0.05 # 5% trailing stop
# Dictionary to store the highest price seen for each holding
# Key: Symbol, Value: Highest Price (float)
self.high_water_marks = {}
def on_data(self, data: Slice):
# 1. Entry Logic (Simple example: Buy if not invested)
if not self.portfolio.invested:
if self.spy in data and data[self.spy] is not None:
self.set_holdings(self.spy, 0.5)
# Initialize high water mark to current price on entry
self.high_water_marks[self.spy] = data[self.spy].close
if self.btc in data and data[self.btc] is not None:
self.set_holdings(self.btc, 0.5)
# Initialize high water mark to current price on entry
self.high_water_marks[self.btc] = data[self.btc].close
# 2. Trailing Stop Logic
self.manage_trailing_stops(data)
def manage_trailing_stops(self, data):
# Iterate over all invested securities
for symbol in self.portfolio.invested.keys():
# Ensure we have data for this symbol in the current slice
if symbol not in data or data[symbol] is None:
continue
current_price = data[symbol].close
position = self.portfolio[symbol]
# --- Handle Long Positions ---
if position.is_long:
# Initialize key if missing (safety check)
if symbol not in self.high_water_marks:
self.high_water_marks[symbol] = current_price
# Update High Water Mark if current price is higher
if current_price > self.high_water_marks[symbol]:
self.high_water_marks[symbol] = current_price
# Calculate Stop Price
stop_price = self.high_water_marks[symbol] * (1 - self.trailing_percent)
# Check if price hit the stop
if current_price < stop_price:
self.log(f"Trailing Stop Hit for {symbol}. High: {self.high_water_marks[symbol]}, Stop: {stop_price}, Current: {current_price}")
self.liquidate(symbol)
# Remove symbol from tracking dict
del self.high_water_marks[symbol]
# --- Handle Short Positions ---
elif position.is_short:
# For shorts, we track the "Low Water Mark"
if symbol not in self.high_water_marks:
self.high_water_marks[symbol] = current_price
# Update Low Water Mark if current price is lower
if current_price < self.high_water_marks[symbol]:
self.high_water_marks[symbol] = current_price
# Calculate Stop Price (Price rises X% above low)
stop_price = self.high_water_marks[symbol] * (1 + self.trailing_percent)
# Check if price hit the stop
if current_price > stop_price:
self.log(f"Trailing Stop Hit (Short) for {symbol}. Low: {self.high_water_marks[symbol]}, Stop: {stop_price}, Current: {current_price}")
self.liquidate(symbol)
del self.high_water_marks[symbol]
def on_order_event(self, order_event):
# Clean up dictionary if we manually liquidate or close positions elsewhere
if order_event.status == OrderStatus.FILLED:
# If the position is closed (quantity is 0), remove from tracking
if self.portfolio[order_event.symbol].quantity == 0:
if order_event.symbol in self.high_water_marks:
del self.high_water_marks[order_event.symbol]
Alternative: Using the Algorithm Framework
If you prefer using the built-in Risk Management models (which requires less code but offers less customization), you can add the TrailingStopRiskManagementModel in your initialize method.
# In Initialize:
self.add_risk_management(TrailingStopRiskManagementModel(0.05))
This one line will automatically liquidate any position that draws down 5% from its peak unrealized profit.
Q&A: Quantitative Trading & Implementation
Q: Why use a "virtual" stop loss in on_data instead of a TrailingStopOrder?
A: While QuantConnect supports TrailingStopOrder, using virtual logic in on_data is often safer for backtesting. It prevents "order spamming" (modifying an order every second the price changes), which can lead to API rate limits in live trading. It also allows you to base the stop on complex logic (e.g., ATR volatility) rather than just a fixed percentage.
Q: How does the resolution of data affect the trailing stop?
A: The code above runs inside on_data. If you use Resolution.DAILY, the stop is only checked once a day at market close. If you use Resolution.MINUTE or Resolution.SECOND, it is checked much more frequently, providing a "tighter" exit but potentially increasing transaction costs due to noise.
Q: Can I use Average True Range (ATR) for the trailing distance?
A: Yes. Instead of self.trailing_percent, you would initialize an ATR indicator (e.g., self.atr = self.ATR("SPY", 14)). In the logic, calculate stop_price = high_water_mark - (self.atr.current.value * multiplier).
Q: Does this logic work for Short positions?
A: Yes, the provided code includes logic for Short positions. It inverts the math: it tracks the lowest price seen (Low Water Mark) and triggers a liquidation if the price rises by the defined percentage above that low.