| Overall Statistics |
|
Total Orders 222 Average Win 1.36% Average Loss -0.68% Compounding Annual Return 2.063% Drawdown 12.300% Expectancy 0.211 Start Equity 10000000 End Equity 11396096.38 Net Profit 13.961% Sharpe Ratio -0.124 Sortino Ratio -0.035 Probabilistic Sharpe Ratio 1.305% Loss Rate 60% Win Rate 40% Profit-Loss Ratio 2.00 Alpha -0.003 Beta -0.057 Annual Standard Deviation 0.062 Annual Variance 0.004 Information Ratio -0.473 Tracking Error 0.186 Treynor Ratio 0.135 Total Fees $219055.26 Estimated Strategy Capacity $16000000.00 Lowest Capacity Asset SBC R735QTJ8XC9X Portfolio Turnover 5.09% |
from AlgorithmImports import *
from datetime import datetime, timedelta
class OptimalStopStrategyAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
self.SetEndDate(datetime.today()) # Set the end date to today
self.SetCash(10000000) # Set the initial cash for the algorithm
self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# Add universe selection based on the constituents of SPY ETF
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# Dictionary to store stock data including the rolling window
self.data = {}
self.SetBenchmark("SPY")
# List to track long positions
self.long_positions = []
def CoarseSelectionFunction(self, coarse):
sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
return [x.Symbol for x in sorted_by_dollar_volume[:500]]
def FineSelectionFunction(self, fine):
# Filter fine universe to include only stocks with market cap > 10 billion
return [x.Symbol for x in fine if x.MarketCap > 1e10]
def OnSecuritiesChanged(self, changes):
# Remove data for securities that are no longer in the universe
for security in changes.RemovedSecurities:
if security.Symbol in self.data:
del self.data[security.Symbol]
# Add data for new securities in the universe
for security in changes.AddedSecurities:
symbol = security.Symbol
self.data[symbol] = {
'high_window': RollingWindow[float](30) ,
'low_window': RollingWindow[float](30) ,
'atr': self.ATR(symbol, 14, Resolution.Daily),
'stop_price': None,
'long_position': False,
'stop_order_ticket': None
}
self.AddEquity(symbol, Resolution.Daily)
def OnData(self, data):
for symbol, stock_data in list(self.data.items()):
if symbol in data.Bars:
bar = data.Bars[symbol]
stock_data['high_window'].Add(bar.High)
stock_data['low_window'].Add(bar.Low)
if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
current_price = bar.Close
highest_high = max(stock_data['high_window'])
atr = stock_data['atr'].Current.Value
# Long Entry Condition
if current_price >= highest_high and not stock_data['long_position']:
stock_data['long_position'] = True
stock_data['stop_price'] = current_price - 2 * atr # Set initial stop price
self.long_positions.append(symbol)
self.Debug(f"Long Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
if current_price < highest_high and stock_data['long_position']:
stock_data['long_position'] = False
self.long_positions.remove(symbol)
self.Debug(f"Long Entry: {symbol} liquidated at {current_price}")
self.Liquidate()
self.Rebalance()
def Rebalance(self):
num_longs = len(self.long_positions)
max_allocation = 1.0 / num_longs if num_longs > 0 else 0
# Adjust long positions
for symbol in self.long_positions:
if symbol in self.Securities and symbol in self.data:
price = self.Securities[symbol].Price
if price > 0:
target_quantity = round(self.Portfolio.TotalPortfolioValue * max_allocation / price)
current_quantity = self.Portfolio[symbol].Quantity
quantity_difference = target_quantity - current_quantity
if quantity_difference != 0:
self.MarketOrder(symbol, quantity_difference)
#Update or create trailing stop order
atr = self.data[symbol]['atr'].Current.Value
new_stop_price = price - 2 * atr
if self.data[symbol]['stop_order_ticket'] is None:
self.data[symbol]['stop_order_ticket'] = self.StopMarketOrder(symbol, -target_quantity, new_stop_price)
self.data[symbol]['stop_price'] = new_stop_price
elif new_stop_price > self.data[symbol]['stop_price']:
self.data[symbol]['stop_price'] = new_stop_price
self.data[symbol]['stop_order_ticket'].UpdateStopPrice(new_stop_price)
self.Debug(f"Updated Long Stop: {symbol} to {new_stop_price}")
def OnEndOfAlgorithm(self):
self.Liquidate()
self.Debug("Liquidated all positions at the end of the algorithm.")
def LiquidateMostProfitable(self):
max_profit = 0
max_symbol = None
for symbol, holding in self.Portfolio.items():
profit = holding.UnrealizedProfit
if profit > max_profit:
max_profit = profit
max_symbol = symbol
if max_symbol is not None:
self.Liquidate(max_symbol)
self.Debug(f"Liquidated {max_symbol} to free up funds")
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class OptimalStopStrategyAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
# self.SetEndDate(datetime.today()) # Set the end date to today
# self.SetCash(10000000) # Set the initial cash for the algorithm
# self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# # Add universe selection based on the constituents of SPY ETF
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.SetBenchmark("SPY")
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# def CoarseSelectionFunction(self, coarse):
# sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# return [x.Symbol for x in sorted_by_dollar_volume[:500]]
# def FineSelectionFunction(self, fine):
# # Filter fine universe to include only stocks with market cap > 10 billion
# return [x.Symbol for x in fine if x.MarketCap > 1e10]
# def OnSecuritiesChanged(self, changes):
# # Remove data for securities that are no longer in the universe
# for security in changes.RemovedSecurities:
# if security.Symbol in self.data:
# del self.data[security.Symbol]
# # Add data for new securities in the universe
# for security in changes.AddedSecurities:
# symbol = security.Symbol
# self.data[symbol] = {
# 'high_window': RollingWindow[Decimal](30) ,
# 'low_window': RollingWindow[Decimal](30) ,
# 'atr': self.ATR(symbol, 14, Resolution.Daily),
# 'stop_price': None,
# 'long_position': False,
# 'short_position': False
# }
# self.AddEquity(symbol, Resolution.Daily)
# def OnData(self, data):
# for symbol, stock_data in list(self.data.items()):
# if symbol in data.Bars:
# bar = data.Bars[symbol]
# stock_data['high_window'].Add(bar.High)
# stock_data['low_window'].Add(bar.Low)
# atr = stock_data['atr'].Current.Value
# if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
# current_price = bar.Close
# highest_high = max(stock_data['high_window'])
# lowest_low = min(stock_data['low_window'])
# # Long Entry Condition
# if current_price >= highest_high and not stock_data['long_position'] and not stock_data['short_position']:
# stock_data['long_position'] = True
# stock_data['short_position'] = False
# stock_data['stop_price'] = current_price - 2 * atr
# self.long_positions.append(symbol)
# self.Debug(f"Long Entry: {symbol} at {current_price} with stop at {stock_data['stop_price']}")
# # Short Entry Condition
# if current_price <= lowest_low and not stock_data['short_position'] and not stock_data['long_position']:
# stock_data['short_position'] = True
# stock_data['long_position'] = False
# stock_data['stop_price'] = current_price + (current_price * 0.05)
# self.short_positions.append(symbol)
# self.Debug(f"Short Entry: {symbol} at {current_price} with stop at {stock_data['stop_price']}")
# # Long Stop Condition
# if stock_data['long_position'] and current_price <= stock_data['stop_price']:
# self.long_positions.remove(symbol)
# stock_data['long_position'] = False
# self.Liquidate(symbol)
# self.Debug(f"Long Exit: {symbol} at {current_price}")
# # Short Stop Condition
# if stock_data['short_position'] and current_price >= stock_data['stop_price']:
# self.short_positions.remove(symbol)
# stock_data['short_position'] = False
# self.Liquidate(symbol)
# self.Debug(f"Short Exit: {symbol} at {current_price}")
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# # Adjust long positions
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# if symbol in self.Securities:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Adjust short positions
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# if symbol in self.Securities:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# def OnEndOfAlgorithm(self):
# self.Liquidate()
# self.Debug("Liquidated all positions at the end of the algorithm.")
# def LiquidateMostProfitable(self):
# max_profit = 0
# max_symbol = None
# for symbol, holding in self.Portfolio.items():
# profit = holding.UnrealizedProfit
# if profit > max_profit:
# max_profit = profit
# max_symbol = symbol
# if max_symbol is not None:
# self.Liquidate(max_symbol)
# self.Debug(f"Liquidated {max_symbol} to free up funds")
##########################################################################
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class OptimalStopStrategyAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
# self.SetEndDate(datetime.today()) # Set the end date to today
# self.SetCash(10000000) # Set the initial cash for the algorithm
# self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# # Add universe selection based on the constituents of SPY ETF
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.SetBenchmark("SPY")
# # Define thresholds
# self.buy_threshold = 9
# self.sell_threshold = 9
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# def CoarseSelectionFunction(self, coarse):
# sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# return [x.Symbol for x in sorted_by_dollar_volume[:500]]
# def FineSelectionFunction(self, fine):
# # Filter fine universe to include only stocks with market cap > 10 billion
# return [x.Symbol for x in fine if x.MarketCap > 1e10]
# def OnSecuritiesChanged(self, changes):
# # Remove data for securities that are no longer in the universe
# for security in changes.RemovedSecurities:
# if security.Symbol in self.data:
# del self.data[security.Symbol]
# # Add data for new securities in the universe
# for security in changes.AddedSecurities:
# symbol = security.Symbol
# self.data[symbol] = {
# 'high_window': RollingWindow[Decimal](30) ,
# 'low_window': RollingWindow[Decimal](30) ,
# 'stop_price': None,
# 'long_position': False,
# 'short_position': False
# }
# self.AddEquity(symbol, Resolution.Daily)
# def OnData(self, data):
# for symbol, stock_data in list(self.data.items()):
# if symbol in data.Bars:
# bar = data.Bars[symbol]
# stock_data['high_window'].Add(bar.High)
# stock_data['low_window'].Add(bar.Low)
# if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
# current_price = bar.Close
# highest_high = max(stock_data['high_window'])
# lowest_low = min(stock_data['low_window'])
# # Long Entry Condition
# if current_price >= highest_high and not stock_data['long_position'] and not stock_data['short_position']:
# stock_data['long_position'] = True
# stock_data['short_position'] = False
# stock_data['stop_price'] = current_price * 0.80 # 20% below the entry price
# self.long_positions.append(symbol)
# self.Debug(f"Long Entry: {symbol} at {current_price} with stop at {stock_data['stop_price']}")
# # Short Entry Condition
# if current_price <= lowest_low and not stock_data['short_position'] and not stock_data['long_position']:
# stock_data['short_position'] = True
# stock_data['long_position'] = False
# stock_data['stop_price'] = current_price * 1.20 # 20% above the entry price
# self.short_positions.append(symbol)
# self.Debug(f"Short Entry: {symbol} at {current_price} with stop at {stock_data['stop_price']}")
# # Long Stop Condition
# if stock_data['long_position'] and current_price <= stock_data['stop_price']:
# self.long_positions.remove(symbol)
# stock_data['long_position'] = False
# self.Liquidate(symbol)
# self.Debug(f"Long Exit: {symbol} at {current_price}")
# # Short Stop Condition
# if stock_data['short_position'] and current_price >= stock_data['stop_price']:
# self.short_positions.remove(symbol)
# stock_data['short_position'] = False
# self.Liquidate(symbol)
# self.Debug(f"Short Exit: {symbol} at {current_price}")
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# # Adjust long positions
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# if symbol in self.Securities:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Adjust short positions
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# if symbol in self.Securities:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# def OnEndOfAlgorithm(self):
# self.Liquidate()
# self.Debug("Liquidated all positions at the end of the algorithm.")
# def LiquidateMostProfitable(self):
# max_profit = 0
# max_symbol = None
# for symbol, holding in self.Portfolio.items():
# profit = holding.UnrealizedProfit
# if profit > max_profit:
# max_profit = profit
# max_symbol = symbol
# if max_symbol is not None:
# self.Liquidate(max_symbol)
# self.Debug(f"Liquidated {max_symbol} to free up funds")
##############################################################################
# ### Trailing stops Try 1
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class OptimalStopStrategyAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
# self.SetEndDate(datetime.today()) # Set the end date to today
# self.SetCash(10000000) # Set the initial cash for the algorithm
# self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# # Add universe selection based on the constituents of SPY ETF
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.SetBenchmark("SPY")
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# def CoarseSelectionFunction(self, coarse):
# sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# return [x.Symbol for x in sorted_by_dollar_volume[:500]]
# def FineSelectionFunction(self, fine):
# # Filter fine universe to include only stocks with market cap > 10 billion
# return [x.Symbol for x in fine if x.MarketCap > 1e10]
# def OnSecuritiesChanged(self, changes):
# # Remove data for securities that are no longer in the universe
# for security in changes.RemovedSecurities:
# if security.Symbol in self.data:
# del self.data[security.Symbol]
# # Add data for new securities in the universe
# for security in changes.AddedSecurities:
# symbol = security.Symbol
# self.data[symbol] = {
# 'high_window': RollingWindow[Decimal](30) ,
# 'low_window': RollingWindow[Decimal](30) ,
# 'atr': self.ATR(symbol, 14, Resolution.Daily),
# 'stop_price': None,
# 'long_position': False,
# 'short_position': False
# }
# self.AddEquity(symbol, Resolution.Daily)
# def OnData(self, data):
# for symbol, stock_data in list(self.data.items()):
# if symbol in data.Bars:
# bar = data.Bars[symbol]
# stock_data['high_window'].Add(bar.High)
# stock_data['low_window'].Add(bar.Low)
# if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
# current_price = bar.Close
# highest_high = max(stock_data['high_window'])
# lowest_low = min(stock_data['low_window'])
# atr = stock_data['atr'].Current.Value
# # Long Entry Condition
# if current_price >= highest_high and not stock_data['long_position'] and not stock_data['short_position']:
# stock_data['long_position'] = True
# stock_data['short_position'] = False
# stock_data['stop_price'] = current_price - 2 * atr # Set trailing stop
# self.long_positions.append(symbol)
# self.Debug(f"Long Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# # Short Entry Condition
# if current_price <= lowest_low and not stock_data['short_position'] and not stock_data['long_position']:
# stock_data['short_position'] = True
# stock_data['long_position'] = False
# stock_data['stop_price'] = current_price + 2 * atr # Set trailing stop
# self.short_positions.append(symbol)
# self.Debug(f"Short Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# # Update Long Trailing Stop
# if stock_data['long_position']:
# new_stop_price = current_price - 2 * atr
# if new_stop_price > stock_data['stop_price']:
# stock_data['stop_price'] = new_stop_price
# self.Debug(f"Updated Long Stop: {symbol} to {stock_data['stop_price']} at {current_price}")
# # Update Short Trailing Stop
# if stock_data['short_position']:
# new_stop_price = current_price + 2 * atr
# if new_stop_price < stock_data['stop_price']:
# stock_data['stop_price'] = new_stop_price
# self.Debug(f"Updated Short Stop: {symbol} to {stock_data['stop_price']} at {current_price}")
# # Long Stop Condition
# if stock_data['long_position'] and current_price <= stock_data['stop_price']:
# self.long_positions.remove(symbol)
# stock_data['long_position'] = False
# self.Liquidate(symbol)
# self.Debug(f"Long Exit: {symbol} at {current_price}")
# # Short Stop Condition
# if stock_data['short_position'] and current_price >= stock_data['stop_price']:
# self.short_positions.remove(symbol)
# stock_data['short_position'] = False
# self.Liquidate(symbol)
# self.Debug(f"Short Exit: {symbol} at {current_price}")
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# # Adjust long positions
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# if symbol in self.Securities:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Adjust short positions
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# if symbol in self.Securities:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# def OnEndOfAlgorithm(self):
# self.Liquidate()
# self.Debug("Liquidated all positions at the end of the algorithm.")
# def LiquidateMostProfitable(self):
# max_profit = 0
# max_symbol = None
# for symbol, holding in self.Portfolio.items():
# profit = holding.UnrealizedProfit
# if profit > max_profit:
# max_profit = profit
# max_symbol = symbol
# if max_symbol is not None:
# self.Liquidate(max_symbol)
# self.Debug(f"Liquidated {max_symbol} to free up funds")
########################################################################################
# Trailing Stop v2 RollingWindow[Decimal](30)
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class OptimalStopStrategyAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
# self.SetEndDate(datetime.today()) # Set the end date to today
# self.SetCash(10000000) # Set the initial cash for the algorithm
# self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# # Add universe selection based on the constituents of SPY ETF
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.SetBenchmark("SPY")
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# def CoarseSelectionFunction(self, coarse):
# sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# return [x.Symbol for x in sorted_by_dollar_volume[:500]]
# def FineSelectionFunction(self, fine):
# # Filter fine universe to include only stocks with market cap > 10 billion
# return [x.Symbol for x in fine if x.MarketCap > 1e10]
# def OnSecuritiesChanged(self, changes):
# # Remove data for securities that are no longer in the universe
# for security in changes.RemovedSecurities:
# if security.Symbol in self.data:
# del self.data[security.Symbol]
# # Add data for new securities in the universe
# for security in changes.AddedSecurities:
# symbol = security.Symbol
# self.data[symbol] = {
# 'high_window': RollingWindow[Decimal](30) ,
# 'low_window': RollingWindow[Decimal](30) ,
# 'atr': self.ATR(symbol, 14, Resolution.Daily),
# 'stop_price': None,
# 'long_position': False,
# 'short_position': False,
# 'stop_order_ticket': None
# }
# self.AddEquity(symbol, Resolution.Daily)
# def OnData(self, data):
# for symbol, stock_data in list(self.data.items()):
# if symbol in data.Bars:
# bar = data.Bars[symbol]
# stock_data['high_window'].Add(bar.High)
# stock_data['low_window'].Add(bar.Low)
# if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
# current_price = bar.Close
# highest_high = max(stock_data['high_window'])
# lowest_low = min(stock_data['low_window'])
# atr = stock_data['atr'].Current.Value
# # Long Entry Condition
# if current_price >= highest_high and not stock_data['long_position'] and not stock_data['short_position']:
# stock_data['long_position'] = True
# stock_data['short_position'] = False
# stock_data['stop_price'] = current_price - 2 * atr # Set initial stop price
# self.long_positions.append(symbol)
# self.Debug(f"Long Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# # Short Entry Condition
# if current_price <= lowest_low and not stock_data['short_position'] and not stock_data['long_position']:
# stock_data['short_position'] = True
# stock_data['long_position'] = False
# stock_data['stop_price'] = current_price + 2 * atr # Set initial stop price
# self.short_positions.append(symbol)
# self.Debug(f"Short Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# # Adjust long positions
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# if symbol in self.Securities and symbol in self.data:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Update or create trailing stop order
# atr = self.data[symbol]['atr'].Current.Value
# new_stop_price = price - 2 * atr
# if self.data[symbol]['stop_order_ticket'] is None:
# self.data[symbol]['stop_order_ticket'] = self.StopMarketOrder(symbol, -target_quantity, new_stop_price)
# self.data[symbol]['stop_price'] = new_stop_price
# elif new_stop_price > self.data[symbol]['stop_price']:
# self.data[symbol]['stop_price'] = new_stop_price
# self.data[symbol]['stop_order_ticket'].UpdateStopPrice(new_stop_price)
# self.Debug(f"Updated Long Stop: {symbol} to {new_stop_price}")
# # Adjust short positions
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# if symbol in self.Securities and symbol in self.data:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Update or create trailing stop order
# atr = self.data[symbol]['atr'].Current.Value
# new_stop_price = price + 2 * atr
# if self.data[symbol]['stop_order_ticket'] is None:
# self.data[symbol]['stop_order_ticket'] = self.StopMarketOrder(symbol, target_quantity, new_stop_price)
# self.data[symbol]['stop_price'] = new_stop_price
# elif new_stop_price < self.data[symbol]['stop_price']:
# self.data[symbol]['stop_price'] = new_stop_price
# self.data[symbol]['stop_order_ticket'].UpdateStopPrice(new_stop_price)
# self.Debug(f"Updated Short Stop: {symbol} to {new_stop_price}")
# def OnEndOfAlgorithm(self):
# self.Liquidate()
# self.Debug("Liquidated all positions at the end of the algorithm.")
# def LiquidateMostProfitable(self):
# max_profit = 0
# max_symbol = None
# for symbol, holding in self.Portfolio.items():
# profit = holding.UnrealizedProfit
# if profit > max_profit:
# max_profit = profit
# max_symbol = symbol
# if max_symbol is not None:
# self.Liquidate(max_symbol)
# self.Debug(f"Liquidated {max_symbol} to free up funds")
###########################################################################################################
##### Base Case
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class OptimalStopStrategyAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
# self.SetEndDate(datetime.today()) # Set the end date to today
# self.SetCash(10000000) # Set the initial cash for the algorithm
# self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# # Add universe selection based on the constituents of SPY ETF
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.SetBenchmark("SPY")
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# def CoarseSelectionFunction(self, coarse):
# sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# return [x.Symbol for x in sorted_by_dollar_volume[:500]]
# def FineSelectionFunction(self, fine):
# # Filter fine universe to include only stocks with market cap > 10 billion
# return [x.Symbol for x in fine if x.MarketCap > 1e10]
# def OnSecuritiesChanged(self, changes):
# # Remove data for securities that are no longer in the universe
# for security in changes.RemovedSecurities:
# if security.Symbol in self.data:
# del self.data[security.Symbol]
# # Add data for new securities in the universe
# for security in changes.AddedSecurities:
# symbol = security.Symbol
# self.data[symbol] = {
# 'high_window': RollingWindow[Decimal](30) ,
# 'low_window': RollingWindow[Decimal](30) ,
# 'atr': self.ATR(symbol, 14, Resolution.Daily),
# 'stop_price': None,
# 'long_position': False,
# 'short_position': False,
# 'stop_order_ticket': None
# }
# self.AddEquity(symbol, Resolution.Daily)
# def OnData(self, data):
# for symbol, stock_data in list(self.data.items()):
# if symbol in data.Bars:
# bar = data.Bars[symbol]
# stock_data['high_window'].Add(bar.High)
# stock_data['low_window'].Add(bar.Low)
# if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
# current_price = bar.Close
# highest_high = max(stock_data['high_window'])
# lowest_low = min(stock_data['low_window'])
# atr = stock_data['atr'].Current.Value
# # Long Entry Condition
# if current_price >= highest_high and not stock_data['long_position'] and not stock_data['short_position']:
# stock_data['long_position'] = True
# stock_data['short_position'] = False
# stock_data['stop_price'] = current_price - 2 * atr # Set initial stop price
# self.long_positions.append(symbol)
# self.Debug(f"Long Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# # Short Entry Condition
# if current_price <= lowest_low and not stock_data['short_position'] and not stock_data['long_position']:
# stock_data['short_position'] = True
# stock_data['long_position'] = False
# stock_data['stop_price'] = current_price + 2 * atr # Set initial stop price
# self.short_positions.append(symbol)
# self.Debug(f"Short Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# # Adjust long positions
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# if symbol in self.Securities and symbol in self.data:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Update or create trailing stop order
# atr = self.data[symbol]['atr'].Current.Value
# new_stop_price = price - 2 * atr
# # if self.data[symbol]['stop_order_ticket'] is None:
# # self.data[symbol]['stop_order_ticket'] = self.StopMarketOrder(symbol, -target_quantity, new_stop_price)
# # self.data[symbol]['stop_price'] = new_stop_price
# # elif new_stop_price > self.data[symbol]['stop_price']:
# # self.data[symbol]['stop_price'] = new_stop_price
# # self.data[symbol]['stop_order_ticket'].UpdateStopPrice(new_stop_price)
# # self.Debug(f"Updated Long Stop: {symbol} to {new_stop_price}")
# # Adjust short positions
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# if symbol in self.Securities and symbol in self.data:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Update or create trailing stop order
# atr = self.data[symbol]['atr'].Current.Value
# new_stop_price = price + 2 * atr
# # if self.data[symbol]['stop_order_ticket'] is None:
# # self.data[symbol]['stop_order_ticket'] = self.StopMarketOrder(symbol, target_quantity, new_stop_price)
# # self.data[symbol]['stop_price'] = new_stop_price
# # elif new_stop_price < self.data[symbol]['stop_price']:
# # self.data[symbol]['stop_price'] = new_stop_price
# # self.data[symbol]['stop_order_ticket'].UpdateStopPrice(new_stop_price)
# # self.Debug(f"Updated Short Stop: {symbol} to {new_stop_price}")
# def OnEndOfAlgorithm(self):
# self.Liquidate()
# self.Debug("Liquidated all positions at the end of the algorithm.")
# def LiquidateMostProfitable(self):
# max_profit = 0
# max_symbol = None
# for symbol, holding in self.Portfolio.items():
# profit = holding.UnrealizedProfit
# if profit > max_profit:
# max_profit = profit
# max_symbol = symbol
# if max_symbol is not None:
# self.Liquidate(max_symbol)
# self.Debug(f"Liquidated {max_symbol} to free up funds")
#######################################################################################
###### Base case long only without stop
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class OptimalStopStrategyAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
# self.SetEndDate(datetime.today()) # Set the end date to today
# self.SetCash(10000000) # Set the initial cash for the algorithm
# self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# # Add universe selection based on the constituents of SPY ETF
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.SetBenchmark("SPY")
# # List to track long positions
# self.long_positions = []
# def CoarseSelectionFunction(self, coarse):
# sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# return [x.Symbol for x in sorted_by_dollar_volume[:500]]
# def FineSelectionFunction(self, fine):
# # Filter fine universe to include only stocks with market cap > 10 billion
# return [x.Symbol for x in fine if x.MarketCap > 1e10]
# def OnSecuritiesChanged(self, changes):
# # Remove data for securities that are no longer in the universe
# for security in changes.RemovedSecurities:
# if security.Symbol in self.data:
# del self.data[security.Symbol]
# # Add data for new securities in the universe
# for security in changes.AddedSecurities:
# symbol = security.Symbol
# self.data[symbol] = {
# 'high_window': RollingWindow[float](30) ,
# 'low_window': RollingWindow[float](30) ,
# 'atr': self.ATR(symbol, 14, Resolution.Daily),
# 'stop_price': None,
# 'long_position': False,
# 'stop_order_ticket': None
# }
# self.AddEquity(symbol, Resolution.Daily)
# def OnData(self, data):
# for symbol, stock_data in list(self.data.items()):
# if symbol in data.Bars:
# bar = data.Bars[symbol]
# stock_data['high_window'].Add(bar.High)
# stock_data['low_window'].Add(bar.Low)
# if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
# current_price = bar.Close
# highest_high = max(stock_data['high_window'])
# atr = stock_data['atr'].Current.Value
# # Long Entry Condition
# if current_price >= highest_high and not stock_data['long_position']:
# stock_data['long_position'] = True
# stock_data['stop_price'] = current_price - 2 * atr # Set initial stop price
# self.long_positions.append(symbol)
# self.Debug(f"Long Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# if current_price < highest_high and stock_data['long_position']:
# stock_data['long_position'] = False
# self.long_positions.remove(symbol)
# self.Debug(f"Long Entry: {symbol} liquidated at {current_price}")
# self.Liquidate()
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# max_allocation = 1.0 / num_longs if num_longs > 0 else 0
# # Adjust long positions
# for symbol in self.long_positions:
# if symbol in self.Securities and symbol in self.data:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * max_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# # Update or create trailing stop order
# # atr = self.data[symbol]['atr'].Current.Value
# # new_stop_price = price - 2 * atr
# # if self.data[symbol]['stop_order_ticket'] is None:
# # self.data[symbol]['stop_order_ticket'] = self.StopMarketOrder(symbol, -target_quantity, new_stop_price)
# # self.data[symbol]['stop_price'] = new_stop_price
# # elif new_stop_price > self.data[symbol]['stop_price']:
# # self.data[symbol]['stop_price'] = new_stop_price
# # self.data[symbol]['stop_order_ticket'].UpdateStopPrice(new_stop_price)
# # self.Debug(f"Updated Long Stop: {symbol} to {new_stop_price}")
# def OnEndOfAlgorithm(self):
# self.Liquidate()
# self.Debug("Liquidated all positions at the end of the algorithm.")
# def LiquidateMostProfitable(self):
# max_profit = 0
# max_symbol = None
# for symbol, holding in self.Portfolio.items():
# profit = holding.UnrealizedProfit
# if profit > max_profit:
# max_profit = profit
# max_symbol = symbol
# if max_symbol is not None:
# self.Liquidate(max_symbol)
# self.Debug(f"Liquidated {max_symbol} to free up funds")
##########################################################################################
##### Base CAse long only with stops
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class OptimalStopStrategyAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
# self.SetEndDate(datetime.today()) # Set the end date to today
# self.SetCash(10000000) # Set the initial cash for the algorithm
# self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# # Add universe selection based on the constituents of SPY ETF
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.SetBenchmark("SPY")
# # List to track long positions
# self.long_positions = []
# def CoarseSelectionFunction(self, coarse):
# sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# return [x.Symbol for x in sorted_by_dollar_volume[:500]]
# def FineSelectionFunction(self, fine):
# # Filter fine universe to include only stocks with market cap > 10 billion
# return [x.Symbol for x in fine if x.MarketCap > 1e10]
# def OnSecuritiesChanged(self, changes):
# # Remove data for securities that are no longer in the universe
# for security in changes.RemovedSecurities:
# if security.Symbol in self.data:
# del self.data[security.Symbol]
# # Add data for new securities in the universe
# for security in changes.AddedSecurities:
# symbol = security.Symbol
# self.data[symbol] = {
# 'high_window': RollingWindow[float](30) ,
# 'low_window': RollingWindow[float](30) ,
# 'atr': self.ATR(symbol, 14, Resolution.Daily),
# 'stop_price': None,
# 'long_position': False,
# 'stop_order_ticket': None
# }
# self.AddEquity(symbol, Resolution.Daily)
# def OnData(self, data):
# for symbol, stock_data in list(self.data.items()):
# if symbol in data.Bars:
# bar = data.Bars[symbol]
# stock_data['high_window'].Add(bar.High)
# stock_data['low_window'].Add(bar.Low)
# if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
# current_price = bar.Close
# highest_high = max(stock_data['high_window'])
# atr = stock_data['atr'].Current.Value
# # Long Entry Condition
# if current_price >= highest_high and not stock_data['long_position']:
# stock_data['long_position'] = True
# stock_data['stop_price'] = current_price - 2 * atr # Set initial stop price
# self.long_positions.append(symbol)
# self.Debug(f"Long Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
# if current_price < highest_high and stock_data['long_position']:
# stock_data['long_position'] = False
# self.long_positions.remove(symbol)
# self.Debug(f"Long Entry: {symbol} liquidated at {current_price}")
# self.Liquidate()
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# max_allocation = 1.0 / num_longs if num_longs > 0 else 0
# # Adjust long positions
# for symbol in self.long_positions:
# if symbol in self.Securities and symbol in self.data:
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * max_allocation / price)
# current_quantity = self.Portfolio[symbol].Quantity
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# #Update or create trailing stop order
# atr = self.data[symbol]['atr'].Current.Value
# new_stop_price = price - 2 * atr
# if self.data[symbol]['stop_order_ticket'] is None:
# self.data[symbol]['stop_order_ticket'] = self.StopMarketOrder(symbol, -target_quantity, new_stop_price)
# self.data[symbol]['stop_price'] = new_stop_price
# elif new_stop_price > self.data[symbol]['stop_price']:
# self.data[symbol]['stop_price'] = new_stop_price
# self.data[symbol]['stop_order_ticket'].UpdateStopPrice(new_stop_price)
# self.Debug(f"Updated Long Stop: {symbol} to {new_stop_price}")
# def OnEndOfAlgorithm(self):
# self.Liquidate()
# self.Debug("Liquidated all positions at the end of the algorithm.")
# def LiquidateMostProfitable(self):
# max_profit = 0
# max_symbol = None
# for symbol, holding in self.Portfolio.items():
# profit = holding.UnrealizedProfit
# if profit > max_profit:
# max_profit = profit
# max_symbol = symbol
# if max_symbol is not None:
# self.Liquidate(max_symbol)
# self.Debug(f"Liquidated {max_symbol} to free up funds")
####################################################################################################
#### Base Case Short only without Stops
from AlgorithmImports import *
from datetime import datetime, timedelta
class OptimalStopStrategyAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2018, 1, 1) # Set the start date for the algorithm
self.SetEndDate(datetime.today()) # Set the end date to today
self.SetCash(10000000) # Set the initial cash for the algorithm
self.UniverseSettings.Resolution = Resolution.Daily # Set the universe selection resolution to daily
# Add universe selection based on the constituents of SPY ETF
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# Dictionary to store stock data including the rolling window
self.data = {}
self.SetBenchmark("SPY")
# List to track short positions
self.short_positions = []
def CoarseSelectionFunction(self, coarse):
sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
return [x.Symbol for x in sorted_by_dollar_volume[:500]]
def FineSelectionFunction(self, fine):
# Filter fine universe to include only stocks with market cap > 10 billion
return [x.Symbol for x in fine if x.MarketCap > 1e10]
def OnSecuritiesChanged(self, changes):
# Remove data for securities that are no longer in the universe
for security in changes.RemovedSecurities:
if security.Symbol in self.data:
del self.data[security.Symbol]
# Add data for new securities in the universe
for security in changes.AddedSecurities:
symbol = security.Symbol
self.data[symbol] = {
'high_window': RollingWindow[float](30) ,
'low_window': RollingWindow[float](30) ,
'atr': self.ATR(symbol, 14, Resolution.Daily),
'stop_price': None,
'short_position': False
}
self.AddEquity(symbol, Resolution.Daily)
def OnData(self, data):
for symbol, stock_data in list(self.data.items()):
if symbol in data.Bars:
bar = data.Bars[symbol]
stock_data['high_window'].Add(bar.High)
stock_data['low_window'].Add(bar.Low)
if stock_data['high_window'].IsReady and stock_data['low_window'].IsReady:
current_price = bar.Close
lowest_low = min(stock_data['low_window'])
atr = stock_data['atr'].Current.Value
# Short Entry Condition
if current_price <= lowest_low and not stock_data['short_position']:
stock_data['short_position'] = True
stock_data['stop_price'] = current_price + 2 * atr # Set initial stop price
self.short_positions.append(symbol)
self.Debug(f"Short Entry: {symbol} at {current_price} with initial stop at {stock_data['stop_price']}")
if current_price > lowest_low and stock_data['short_position']:
stock_data['short_position'] = False
self.short_positions.remove(symbol)
self.Debug(f"Short Exit: {symbol} liquidated at {current_price}")
self.Liquidate(symbol)
self.Rebalance()
def Rebalance(self):
num_shorts = len(self.short_positions)
max_allocation = 1.0 / num_shorts if num_shorts > 0 else 0
# Adjust short positions
for symbol in self.short_positions:
if symbol in self.Securities and symbol in self.data:
price = self.Securities[symbol].Price
if price > 0:
target_quantity = round(self.Portfolio.TotalPortfolioValue * max_allocation / price)
current_quantity = self.Portfolio[symbol].Quantity
quantity_difference = -target_quantity - current_quantity
if quantity_difference != 0:
self.MarketOrder(symbol, quantity_difference)
def OnEndOfAlgorithm(self):
self.Liquidate()
self.Debug("Liquidated all positions at the end of the algorithm.")
def LiquidateMostProfitable(self):
max_profit = 0
max_symbol = None
for symbol, holding in self.Portfolio.items():
profit = holding.UnrealizedProfit
if profit > max_profit:
max_profit = profit
max_symbol = symbol
if max_symbol is not None:
self.Liquidate(max_symbol)
self.Debug(f"Liquidated {max_symbol} to free up funds")