# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class GapTradingAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1)
# self.SetEndDate(datetime.today())
# self.SetCash(10000000)
# self.UniverseSettings.Resolution = Resolution.Daily
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# self.gap_percent_threshold = 0.10
# self.atr_period = 14
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.atr = {}
# # Define thresholds
# self.buy_threshold = 9
# self.sell_threshold = 9
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# # Schedule the Rebalance function to run at the market open
# #self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen("SPY", 1), self.Rebalance)
# 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):
# for security in changes.AddedSecurities:
# self.atr[security.Symbol] = self.ATR(security.Symbol, self.atr_period, Resolution.Daily) # Create ATR indicator
# self.RegisterIndicator(security.Symbol, self.atr[security.Symbol]) # Register the indicator
# def OnData(self, data):
# for symbol in data.Bars.Keys:
# if symbol not in self.Securities:
# continue
# bar = data[symbol]
# if not bar:
# continue
# previous_day_close = self.Securities[symbol].Close
# current_open = bar.Open
# gap_percent = abs((current_open - previous_day_close) / previous_day_close)
# if gap_percent > self.gap_percent_threshold:
# atr = self.atr[symbol].Current.Value
# if current_open > previous_day_close:
# if symbol not in self.long_positions:
# self.long_positions.append(symbol)
# if symbol in self.short_positions:
# self.short_positions.remove(symbol)
# else:
# if symbol not in self.short_positions:
# self.short_positions.append(symbol)
# if symbol in self.long_positions:
# self.long_positions.remove(symbol)
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# current_quantity = self.Portfolio[symbol].Quantity
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# atr = self.atr[symbol].Current.Value
# self.SetStopLoss(symbol, atr)
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# current_quantity = self.Portfolio[symbol].Quantity
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# atr = self.atr[symbol].Current.Value
# self.SetStopLoss(symbol, atr)
# def SetStopLoss(self, symbol, atr):
# price = self.Securities[symbol].Price
# if symbol in self.long_positions:
# stop_price = price - atr
# else:
# stop_price = price + atr
# self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, stop_price)
####################################################################################################
#################################### Gap Strategy v2
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class GapTradingAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1)
# self.SetEndDate(datetime.today())
# self.SetCash(10000000)
# self.UniverseSettings.Resolution = Resolution.Daily
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# self.gap_percent_threshold = 0.10
# self.atr_period = 14
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.atr = {}
# self.stop_levels = {}
# # Define thresholds
# self.buy_threshold = 9
# self.sell_threshold = 9
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# # Schedule the Rebalance function to run at the market open
# #self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen("SPY", 1), self.Rebalance)
# 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):
# for security in changes.AddedSecurities:
# self.atr[security.Symbol] = self.ATR(security.Symbol, self.atr_period, Resolution.Daily) # Create ATR indicator
# self.RegisterIndicator(security.Symbol, self.atr[security.Symbol]) # Register the indicator
# def OnData(self, data):
# for symbol in data.Bars.Keys:
# if symbol not in self.Securities:
# continue
# bar = data[symbol]
# if not bar:
# continue
# previous_day_close = self.Securities[symbol].Close
# current_open = bar.Open
# gap_percent = abs((current_open - previous_day_close) / previous_day_close)
# if gap_percent > self.gap_percent_threshold:
# atr = self.atr[symbol].Current.Value
# if current_open > previous_day_close:
# if symbol not in self.long_positions:
# self.long_positions.append(symbol)
# if symbol in self.short_positions:
# self.short_positions.remove(symbol)
# else:
# if symbol not in self.short_positions:
# self.short_positions.append(symbol)
# if symbol in self.long_positions:
# self.long_positions.remove(symbol)
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# current_quantity = self.Portfolio[symbol].Quantity
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# atr = self.atr[symbol].Current.Value
# self.SetStopLoss(symbol, atr)
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# current_quantity = self.Portfolio[symbol].Quantity
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# atr = self.atr[symbol].Current.Value
# #self.SetStopLoss(symbol, atr)
# def SetStopLoss(self, symbol, atr):
# price = self.Securities[symbol].Price
# if symbol in self.long_positions:
# new_stop_price = price - atr
# if symbol in self.stop_levels:
# previous_stop_price = self.stop_levels[symbol]
# if new_stop_price > previous_stop_price:
# self.stop_levels[symbol] = new_stop_price
# self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
# else:
# new_stop_price = price + atr
# if symbol in self.stop_levels:
# previous_stop_price = self.stop_levels[symbol]
# if new_stop_price < previous_stop_price:
# self.stop_levels[symbol] = new_stop_price
# self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
# # else:
# # self.stop_levels[symbol] = new_stop_price
# # self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
#########################################################################################
############### Gap Trading with 2 day exit
# from AlgorithmImports import *
# from datetime import datetime, timedelta
# class GapTradingAlgorithm(QCAlgorithm):
# def Initialize(self):
# self.SetStartDate(2018, 1, 1)
# self.SetEndDate(datetime.today())
# self.SetCash(10000000)
# self.UniverseSettings.Resolution = Resolution.Daily
# self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
# self.gap_percent_threshold = 0.10
# self.atr_period = 14
# # Dictionary to store stock data including the rolling window
# self.data = {}
# self.atr = {}
# self.entry_dates = {}
# self.previous_day_close = {}
# # Lists to track long and short positions
# self.long_positions = []
# self.short_positions = []
# # Schedule the Rebalance function to run at the market open
# #self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen("SPY", 1), self.Rebalance)
# self.SetSecurityInitializer(lambda x: x.SetFeeModel(ConstantFeeModel(0)))
# 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):
# for security in changes.AddedSecurities:
# self.atr[security.Symbol] = self.ATR(security.Symbol, self.atr_period, Resolution.Daily) # Create ATR indicator
# self.RegisterIndicator(security.Symbol, self.atr[security.Symbol]) # Register the indicator
# #self.previous_day_close[security.Symbol] = self.History(security.Symbol, 2, Resolution.Daily).iloc[-2]['close']
# history = self.History(security.Symbol, 2, Resolution.Daily)
# if len(history) >= 2:
# self.previous_day_close[security.Symbol] = history.iloc[-2]['close']
# def OnData(self, data):
# for symbol in data.Bars.Keys:
# if symbol not in self.Securities:
# continue
# bar = data[symbol]
# if not bar:
# continue
# #previous_day_close = self.previous_day_close[symbol]
# current_open = bar.Open
# if symbol in self.previous_day_close:
# previous_day_close = self.previous_day_close[symbol]
# gap_percent = abs((current_open - previous_day_close) / previous_day_close)
# #gap_percent = abs((current_open - previous_day_close) / previous_day_close)
# if gap_percent > self.gap_percent_threshold:
# if current_open > previous_day_close:
# if symbol not in self.long_positions:
# self.long_positions.append(symbol)
# self.entry_dates[symbol] = self.Time
# if symbol in self.short_positions:
# self.short_positions.remove(symbol)
# else:
# if symbol not in self.short_positions:
# self.short_positions.append(symbol)
# self.entry_dates[symbol] = self.Time
# if symbol in self.long_positions:
# self.long_positions.remove(symbol)
# #self.previous_day_close[symbol] = bar.Close
# self.Rebalance()
# def Rebalance(self):
# num_longs = len(self.long_positions)
# num_shorts = len(self.short_positions)
# max_allocation = 0.05
# if num_longs > 0:
# long_allocation = min(max_allocation, 1.0 / num_longs)
# for symbol in self.long_positions:
# if (self.Time - self.entry_dates[symbol]).days >= 2:
# self.Liquidate(symbol)
# self.long_positions.remove(symbol)
# continue
# current_quantity = self.Portfolio[symbol].Quantity
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# if num_shorts > 0:
# short_allocation = -min(max_allocation, 1.0 / num_shorts)
# for symbol in self.short_positions:
# if (self.Time - self.entry_dates[symbol]).days >= 2:
# self.Liquidate(symbol)
# self.short_positions.remove(symbol)
# continue
# current_quantity = self.Portfolio[symbol].Quantity
# price = self.Securities[symbol].Price
# if price > 0:
# target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
# quantity_difference = target_quantity - current_quantity
# if quantity_difference != 0:
# self.MarketOrder(symbol, quantity_difference)
# def SetStopLoss(self, symbol, atr):
# price = self.Securities[symbol].Price
# if symbol in self.long_positions:
# new_stop_price = price - atr
# if symbol in self.stop_levels:
# previous_stop_price = self.stop_levels[symbol]
# if new_stop_price > previous_stop_price:
# self.stop_levels[symbol] = new_stop_price
# self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
# else:
# new_stop_price = price + atr
# if symbol in self.stop_levels:
# previous_stop_price = self.stop_levels[symbol]
# if new_stop_price < previous_stop_price:
# self.stop_levels[symbol] = new_stop_price
# self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
################################### Gap Strategy v2
from AlgorithmImports import *
from datetime import datetime, timedelta
class GapTradingAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2018, 1, 1)
self.SetEndDate(datetime.today())
self.SetCash(10000000)
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
self.gap_percent_threshold = 0.10
self.atr_period = 14
# Dictionary to store stock data including the rolling window
self.data = {}
self.atr = {}
self.stop_levels = {}
self.previous_day_close = {}
# Define thresholds
self.buy_threshold = 9
self.sell_threshold = 9
# Lists to track long and short positions
self.long_positions = []
self.short_positions = []
# Schedule the Rebalance function to run at the market open
#self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen("SPY", 1), self.Rebalance)
self.SetSecurityInitializer(lambda x: x.SetFeeModel(ConstantFeeModel(0)))
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):
for security in changes.AddedSecurities:
self.atr[security.Symbol] = self.ATR(security.Symbol, self.atr_period, Resolution.Daily) # Create ATR indicator
self.RegisterIndicator(security.Symbol, self.atr[security.Symbol]) # Register the indicator
#self.previous_day_close[security.Symbol] = self.History(security.Symbol, 2, Resolution.Daily).iloc[-2]['close']
history = self.History(security.Symbol, 2, Resolution.Daily)
if len(history) >= 2:
self.previous_day_close[security.Symbol] = history.iloc[-2]['close']
def OnData(self, data):
for symbol in data.Bars.Keys:
if symbol not in self.Securities:
continue
bar = data[symbol]
if not bar:
continue
#previous_day_close = self.Securities[symbol].Close
current_open = bar.Open
if symbol in self.previous_day_close:
previous_day_close = self.previous_day_close[symbol]
gap_percent = abs((current_open - previous_day_close) / previous_day_close)
#gap_percent = abs((current_open - previous_day_close) / previous_day_close)
if gap_percent > self.gap_percent_threshold:
atr = self.atr[symbol].Current.Value
if current_open < previous_day_close:
if symbol not in self.long_positions:
self.long_positions.append(symbol)
if symbol in self.short_positions:
self.short_positions.remove(symbol)
else:
if symbol not in self.short_positions:
self.short_positions.append(symbol)
if symbol in self.long_positions:
self.long_positions.remove(symbol)
self.Rebalance()
def Rebalance(self):
num_longs = len(self.long_positions)
num_shorts = len(self.short_positions)
max_allocation = 0.05
if num_longs > 0:
long_allocation = min(max_allocation, 1.0 / num_longs)
for symbol in self.long_positions:
current_quantity = self.Portfolio[symbol].Quantity
price = self.Securities[symbol].Price
if price > 0:
target_quantity = round(self.Portfolio.TotalPortfolioValue * long_allocation / price)
quantity_difference = target_quantity - current_quantity
if quantity_difference != 0:
self.MarketOrder(symbol, quantity_difference)
atr = self.atr[symbol].Current.Value
#self.SetStopLoss(symbol, atr)
if num_shorts > 0:
short_allocation = -min(max_allocation, 1.0 / num_shorts)
for symbol in self.short_positions:
current_quantity = self.Portfolio[symbol].Quantity
price = self.Securities[symbol].Price
if price > 0:
target_quantity = round(self.Portfolio.TotalPortfolioValue * short_allocation / price)
quantity_difference = target_quantity - current_quantity
if quantity_difference != 0:
self.MarketOrder(symbol, quantity_difference)
atr = self.atr[symbol].Current.Value
#self.SetStopLoss(symbol, atr)
def SetStopLoss(self, symbol, atr):
price = self.Securities[symbol].Price
if symbol in self.long_positions:
new_stop_price = price - atr
if symbol in self.stop_levels:
previous_stop_price = self.stop_levels[symbol]
if new_stop_price > previous_stop_price:
self.stop_levels[symbol] = new_stop_price
self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
else:
new_stop_price = price + atr
if symbol in self.stop_levels:
previous_stop_price = self.stop_levels[symbol]
if new_stop_price < previous_stop_price:
self.stop_levels[symbol] = new_stop_price
self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
# else:
# self.stop_levels[symbol] = new_stop_price
# self.StopMarketOrder(symbol, -self.Portfolio[symbol].Quantity, new_stop_price)
