| Overall Statistics |
|
Total Orders 258 Average Win 2.60% Average Loss -1.64% Compounding Annual Return 42.790% Drawdown 17.900% Expectancy 0.497 Start Equity 100000000 End Equity 291135266.42 Net Profit 191.135% Sharpe Ratio 1.367 Sortino Ratio 1.372 Probabilistic Sharpe Ratio 75.371% Loss Rate 42% Win Rate 58% Profit-Loss Ratio 1.58 Alpha 0 Beta 0 Annual Standard Deviation 0.207 Annual Variance 0.043 Information Ratio 1.477 Tracking Error 0.207 Treynor Ratio 0 Total Fees $7891055.25 Estimated Strategy Capacity $34000000.00 Lowest Capacity Asset TSLA UNU3P8Y3WFAD Portfolio Turnover 19.06% |
from AlgorithmImports import *
class EldersTripleScreenAlpha(QCAlgorithm):
def Initialize(self):
########### Strategy Params ###########
self.SetStartDate(2017, 1, 1)
self.SetEndDate(2019, 12, 31)
self.SetCash(100000000)
# Warmup the algorithm with prior data for backtesting
self.set_warmup(timedelta(7))
########### Universe Selection ###########
self.UniverseSettings.Resolution = Resolution.HOUR
self.AddUniverse(self.CoarseSelectionFunction)#, self.FineSelectionFunction)
self.symbols = []
self.data = {}
########### Elders Triple Screen Signals ###########
self.ema_period = 20
self.rsi_period = 14
self.macd_fast_period = 12
self.macd_slow_period = 26
self.macd_signal_period = 6
########### Risk Management Model ###########
self.add_risk_management(MaximumDrawdownPercentPerSecurity())
########### Order Execution Model ###########
self.set_execution(VolumeWeightedAveragePriceExecutionModel())
########### Portfolio Construction Model ###########
self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())#rebalance=self.date_rules.week_start("SPY")))
########### Scheduler ###########
# daily trading option
#self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.at(11,30,0, TimeZones.NEW_YORK), self.EvaluateIndicators)
###************* ADD IN MARKET ON CLOSE ORDER HERE****************#################
def CoarseSelectionFunction(self, coarse):
# Filter to Russell 1000 constituents with a minimum daily trading volume of 500 million
selected = [x for x in coarse if x.HasFundamentalData and x.DollarVolume > 500000000\
and 5e6 < x.market_cap] # < 1e10]
# Log the top 10 and bottom 10 by dollar volume
sorted_selected = sorted(selected, key=lambda x: x.DollarVolume, reverse=True)
# Slicing to only return the top 20 to increase concentration and reduce trading frequency
top20 = sorted_selected[:20]
############### Evaluate Indicators #########################
''' Elder's Triple Screen Alpha Logic'''
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
symbol = security.Symbol
self.data[symbol] = {
'ema': ExponentialMovingAverage(self.ema_period),
'rsi': RelativeStrengthIndex(self.rsi_period, MovingAverageType.Wilders),
'macd': MovingAverageConvergenceDivergence(self.macd_fast_period, self.macd_slow_period, self.macd_signal_period, MovingAverageType.Wilders)
}
self.RegisterIndicator(symbol, self.data[symbol]['ema'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['rsi'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['macd'], Resolution.Daily)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.data:
del self.data[symbol]
def EvaluateIndicators(self):
for symbol, indicators in self.data.items():
ema = indicators['ema']
rsi = indicators['rsi']
macd = indicators['macd']
if not (ema.IsReady and rsi.IsReady and macd.IsReady):
continue
ema_value = ema.Current.Value
rsi_value = rsi.Current.Value
macd_value = macd.Current.Value
macd_signal_value = macd.Signal.Current.Value
price = self.Securities[symbol].Price
direction = InsightDirection.Flat
confidence = 0
magnitude = 0
if rsi_value < 30 and macd_value > macd_signal_value: #and price > ema_value:
direction = InsightDirection.Up
confidence = (70 - rsi_value) / 100
magnitude = macd_value - macd_signal_value
elif rsi_value > 70 or macd_value < macd_signal_value:# or price < ema_value:
direction = InsightDirection.Down
confidence = (rsi_value - 30) / 100
magnitude = macd_signal_value - macd_value
insight = Insight.Price(symbol, timedelta(days=1), direction, magnitude, confidence)
#self.EmitInsights(insight)
#self.Debug(f"Generated Insight: {insight}")
if confidence > 0.55:
self.emit_insights(insight)
self.Debug(f"Generated Insight: {insight}")
return insight
def OnData(self, data):
pass
'''
if not self.Portfolio.Invested:
return
###### Max Security Weight ######
# Calculate the target weight for each security (e.g., 5%)
target_weight = 0.05
# Get the symbols of the invested securities
invested_symbols = [x.Symbol for x in self.Portfolio.Values if x.Invested]
# Loop through each invested security
for symbol in invested_symbols:
if symbol in data:
bar = data[symbol]
if bar is not None:
security = self.Portfolio[symbol]
# Calculate the current weight of the security in the portfolio
if bar.Close is not None:
current_weight = security.Quantity * bar.Close / self.Portfolio.TotalPortfolioValue
# If the current weight exceeds the target weight, trim the position
if current_weight > target_weight:
# Calculate the amount to trim
trim_amount = (current_weight - target_weight) * self.Portfolio.TotalPortfolioValue
# Reduce the position to maintain the target weight
self.SetHoldings(symbol, target_weight)
self.Debug(f"Reduced position for {symbol} to maintain 5% weight, amount: {trim_amount}")
###### Sell Orders ######
# Iterate through portfolio holdings and check for securities to sell
for security in self.Portfolio.Values:
symbol = security.Symbol
if security.Invested and symbol not in invested_symbols:
# Security is invested but not in the current universe, sell it
self.Debug(symbol, "Invalid sell order: Security not in current universe")
self.Debug(f"Invalid sell order: Sold {symbol} as it's not in the current universe")
###### Buy Orders ######
# Check available buying power before submitting orders
for kvp in data.Bars:
symbol = kvp.Key
bar = kvp.Value
# Calculate the order size based on available buying power
available_buying_power = self.Portfolio.Cash
if available_buying_power <= 0:
return # Skip if there's no buying power
price = bar.Close
quantity = int(available_buying_power / price)
# Submit the order only if there's sufficient buying power
if quantity > 0:
self.SetHoldings(symbol, 0.05)
self.Debug(f"Order placed for {symbol}: Quantity={quantity}, Price={price}, TotalValue={quantity*price}")
else:
self.Debug(f"Not enough buying power to place order for {symbol}")
'''
from AlgorithmImports import *
from symbol_calcs import *
class EldersTripleScreenAlpha(QCAlgorithm):
def Initialize(self):
''' Initial Algo parameters and QuantConnect methods'''
########### Strategy Params ###########
self.SetStartDate(2017, 1, 1)
self.SetEndDate(2019, 12, 31)
# Out of Sample (OoS) testing date
#self.SetEndDate(2024, 3, 31)
self.SetCash(100000)
# Define boolean for trading permission for the algorithm. Stops it from trading until the next trading period.
#self.tradingperm = True
# Warmup the algorithm with prior data for backtesting
self.set_warmup(timedelta(30))
########### Universe Selection ###########
self.universe_settings.resolution = Resolution.HOUR
self.universe_settings.leverage = 2
#self.add_universe(self.CoarseFilter)
self.add_universe(self._fundamental_selection_function)
# Variables for universe selection model
self.coarse_count = 10
self.averages = { }
self.symbols = []
self.data = {}
'''
########### Elders Triple Screen Signals ###########
self.ema_period = int(self.get_parameter("ema_period"))
self.rsi_period = int(self.get_parameter("rsi_period"))
self.macd_fast_period = int(self.get_parameter("macd_period"))
self.macd_slow_period = int(self.macd_fast_period*2)
self.macd_signal_period = int(self.macd_fast_period/2)
'''
########### Elders Triple Screen Signals ###########
self.ema_period = 20
self.rsi_period = 14
self.macd_fast_period = 12
self.macd_slow_period = 26
self.macd_signal_period = 6
########### Risk Management Model ###########
self.add_risk_management(MaximumDrawdownPercentPerSecurity(0.05))
########### Order Execution Model ###########
#self.set_execution(ImmediateExecutionModel())
# VWAP Execution Model, to later be changed to the EMA Model
self.set_execution(VolumeWeightedAveragePriceExecutionModel())
########### Portfolio Construction Model ###########
#self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())#rebalance=self.date_rules.week_start("SPY")))
#self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(rebalance=timedelta(7), portfolioBias=PortfolioBias.LONG))
#self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(portfolioBias=PortfolioBias.LONG))
self.set_portfolio_construction(RiskParityPortfolioConstructionModel(portfolioBias=PortfolioBias.LONG))
#self.settings.rebalance_portfolio_on_insight_changes = False
# Required portfolio free cash value
self.Settings.FreePortfolioValuePercentage=0.05
########### Reality Modeling Parameters ###########
# Brokerage Simulation
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
# Risk Free Interest Rate Modeling
self.set_risk_free_interest_rate_model(InterestRateProvider())
# Margin Call Modeling
self.portfolio.margin_call_model = DefaultMarginCallModel(self.portfolio, self.default_order_properties)
########### Scheduler ###########
# Run 30 mins after open to collect signals
# Schedule insight generation every 4 hours
#self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.every(timedelta(minutes=90)), self.EvaluateIndicators)
# daily trading option
self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.at(11,30,0, TimeZones.NEW_YORK), self.EvaluateIndicators)
def _fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
''' Fundamental Filter to produce a list of securities within the universe'''
filtered = [f for f in fundamental if f.price > 10 and f.has_fundamental_data and not np.isnan(f.valuation_ratios.pe_ratio)]
sorted_by_dollar_volume = sorted(filtered, key=lambda f: f.dollar_volume, reverse=True)[:100]
sorted_by_pe_ratio = sorted(sorted_by_dollar_volume, key=lambda f: f.valuation_ratios.pe_ratio, reverse=False)[:20]
return [f.symbol for f in sorted_by_pe_ratio]
def FineFilter(self, fine):
''' Placeholder for further filtering of the universe '''
pass
''' Simpler changes to universe
# this event fires whenever we have changes to our universe
def on_securities_changed(self, changes):
# liquidate removed securities
for security in changes.removed_securities:
if security.invested:
self.liquidate(security.symbol)
self.Debug(f"Liquidated {security.symbol}")
# we want 20% allocation in each security in our universe - This gives 10% allocation
for security in changes.added_securities:
self.set_holdings(security.symbol, 0.1)
self.Debug(f"Bought {security.symbol}")
'''
############### Evaluate Indicators #########################
''' Elder's Triple Screen Alpha Logic'''
def OnSecuritiesChanged(self, changes):
''' Add indicators whenever securities are added to or removed from the universe'''
for security in changes.AddedSecurities:
symbol = security.Symbol
self.data[symbol] = {
'ema': ExponentialMovingAverage(self.ema_period),
'rsi': RelativeStrengthIndex(self.rsi_period, MovingAverageType.Wilders),
'macd': MovingAverageConvergenceDivergence(self.macd_fast_period, self.macd_slow_period, self.macd_signal_period, MovingAverageType.Wilders)
}
self.RegisterIndicator(symbol, self.data[symbol]['ema'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['rsi'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['macd'], Resolution.Daily)
# Reality Modeling for slippage
#security.set_slippage_model(VolumeShareSlippageModel(0.025, 0.1))
#security.set_slippage_model(MarketImpactSlippageModel(self))
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.data:
del self.data[symbol]
def EvaluateIndicators(self):
''' Evaluate the indicators for the securities '''
# Maybe change this self.data.items() to something else. Rn its not iterating through all the securities like it should
for symbol, indicators in self.data.items():
ema = indicators['ema']
rsi = indicators['rsi']
macd = indicators['macd']
if not (ema.IsReady and rsi.IsReady and macd.IsReady):
continue
ema_value = ema.Current.Value
rsi_value = rsi.Current.Value
macd_value = macd.Current.Value
macd_signal_value = macd.Signal.Current.Value
price = self.Securities[symbol].Price
direction = InsightDirection.Flat
confidence = 0
magnitude = 0
if rsi_value < 30 and macd_value > macd_signal_value: #and price > ema_value:
direction = InsightDirection.Up
confidence = (70 - rsi_value) / 100
magnitude = macd_value - macd_signal_value
# Consider changing this to an AND inistead of an or, since both need to be satisfied
elif rsi_value > 70 or macd_value < macd_signal_value:# or price < ema_value:
direction = InsightDirection.Down
confidence = (rsi_value - 30) / 100
magnitude = macd_signal_value - macd_value
insight = Insight.Price(symbol, timedelta(days=1), direction, magnitude, confidence)
#self.EmitInsights(insight)
#self.Debug(f"Generated Insight: {insight}")
if confidence > 0.2:
self.emit_insights(insight)
self.Debug(f"Generated Insight: {insight}")
return insightfrom AlgorithmImports import *
from symbol_calcs import *
class EldersTripleScreenAlpha(QCAlgorithm):
def Initialize(self):
########### Strategy Params ###########
self.SetStartDate(2017, 1, 1)
#self.SetEndDate(2019, 12, 31)
self.SetEndDate(2024, 3, 31)
self.SetCash(100000)
# Define boolean for trading permission for the algorithm. Stops it from trading until the next trading period.
#self.tradingperm = True
# Warmup the algorithm with prior data for backtesting
self.set_warmup(timedelta(30))
########### Universe Selection ###########
self.universe_settings.resolution = Resolution.HOUR
self.universe_settings.leverage = 2
#self.add_universe(self.CoarseFilter)
self.add_universe(self._fundamental_selection_function)
# Variables for universe selection model
self.coarse_count = 10
self.averages = { }
self.symbols = []
self.data = {}
'''
########### Elders Triple Screen Signals ###########
self.ema_period = int(self.get_parameter("ema_period"))
self.rsi_period = int(self.get_parameter("rsi_period"))
self.macd_fast_period = int(self.get_parameter("macd_period"))
self.macd_slow_period = int(self.macd_fast_period*2)
self.macd_signal_period = int(self.macd_fast_period/2)
'''
########### Elders Triple Screen Signals ###########
self.ema_period = 20
self.rsi_period = 14
self.macd_fast_period = 12
self.macd_slow_period = 26
self.macd_signal_period = 6
########### Risk Management Model ###########
self.add_risk_management(MaximumDrawdownPercentPerSecurity(0.05))
########### Order Execution Model ###########
self.set_execution(ImmediateExecutionModel())
########### Portfolio Construction Model ###########
#self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())#rebalance=self.date_rules.week_start("SPY")))
#self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(rebalance=timedelta(7), portfolioBias=PortfolioBias.LONG))
#self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(portfolioBias=PortfolioBias.LONG))
self.set_portfolio_construction(RiskParityPortfolioConstructionModel(portfolioBias=PortfolioBias.LONG))
#self.settings.rebalance_portfolio_on_insight_changes = False
# Required portfolio free cash value
self.Settings.FreePortfolioValuePercentage=0.05
########### Reality Modeling Parameters ###########
# Brokerage Simulation
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
# Risk Free Interest Rate Modeling
self.set_risk_free_interest_rate_model(InterestRateProvider())
# Margin Call Modeling
self.portfolio.margin_call_model = DefaultMarginCallModel(self.portfolio, self.default_order_properties)
########### Scheduler ###########
# Run 30 mins after open to collect signals
# Schedule insight generation every 4 hours
#self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.every(timedelta(minutes=90)), self.EvaluateIndicators)
# daily trading option
self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.at(11,30,0, TimeZones.NEW_YORK), self.EvaluateIndicators)
def _fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
''' Fundamental Filter to produce a list of securities within the universe'''
filtered = [f for f in fundamental if f.price > 10 and f.has_fundamental_data and not np.isnan(f.valuation_ratios.pe_ratio)]
sorted_by_dollar_volume = sorted(filtered, key=lambda f: f.dollar_volume, reverse=True)[:100]
sorted_by_pe_ratio = sorted(sorted_by_dollar_volume, key=lambda f: f.valuation_ratios.pe_ratio, reverse=False)[:20]
return [f.symbol for f in sorted_by_pe_ratio]
def FineFilter(self, fine):
''' Plaseholder for further filtering of the universe '''
pass
''' Simpler changes to universe
# this event fires whenever we have changes to our universe
def on_securities_changed(self, changes):
# liquidate removed securities
for security in changes.removed_securities:
if security.invested:
self.liquidate(security.symbol)
self.Debug(f"Liquidated {security.symbol}")
# we want 20% allocation in each security in our universe - This gives 10% allocation
for security in changes.added_securities:
self.set_holdings(security.symbol, 0.1)
self.Debug(f"Bought {security.symbol}")
'''
############### Evaluate Indicators #########################
''' Elder's Triple Screen Alpha Logic'''
def OnSecuritiesChanged(self, changes):
''' Add indicators whenever securities are added to or removed from the universe'''
for security in changes.AddedSecurities:
symbol = security.Symbol
self.data[symbol] = {
'ema': ExponentialMovingAverage(self.ema_period),
'rsi': RelativeStrengthIndex(self.rsi_period, MovingAverageType.Wilders),
'macd': MovingAverageConvergenceDivergence(self.macd_fast_period, self.macd_slow_period, self.macd_signal_period, MovingAverageType.Wilders)
}
self.RegisterIndicator(symbol, self.data[symbol]['ema'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['rsi'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['macd'], Resolution.Daily)
# Reality Modeling for slippage
security.set_slippage_model(VolumeShareSlippageModel(0.025, 0.1))
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.data:
del self.data[symbol]
def EvaluateIndicators(self):
''' Evaluate the indicators for the securities '''
# Maybe change this self.data.items() to something else. Rn its not iterating through all the securities like it should
for symbol, indicators in self.data.items():
ema = indicators['ema']
rsi = indicators['rsi']
macd = indicators['macd']
if not (ema.IsReady and rsi.IsReady and macd.IsReady):
continue
ema_value = ema.Current.Value
rsi_value = rsi.Current.Value
macd_value = macd.Current.Value
macd_signal_value = macd.Signal.Current.Value
price = self.Securities[symbol].Price
direction = InsightDirection.Flat
confidence = 0
magnitude = 0
# Requires both to be true to be conservative and only enter positions where we are reasonably certain its a good time
if rsi_value < 30 and macd_value > macd_signal_value: #and price > ema_value:
direction = InsightDirection.Up
confidence = (70 - rsi_value) / 100
magnitude = macd_value - macd_signal_value
# Uses or instead of and to be more conservative and exit faster, where if one of the signals is below then liquidate.
elif rsi_value > 70 or macd_value < macd_signal_value:# or price < ema_value:
direction = InsightDirection.Down
confidence = (rsi_value - 30) / 100
magnitude = macd_signal_value - macd_value
insight = Insight.Price(symbol, timedelta(days=1), direction, magnitude, confidence)
#self.EmitInsights(insight)
#self.Debug(f"Generated Insight: {insight}")
if confidence > 0.2:
self.emit_insights(insight)
self.Debug(f"Generated Insight: {insight}")
return insightfrom AlgorithmImports import *
#from symbol_calcs import *
class EldersTripleScreenAlpha(QCAlgorithm):
def Initialize(self):
########### Strategy Params ###########
self.SetStartDate(2017, 1, 1)
self.SetEndDate(2019, 12, 31)
self.SetCash(100000)
# Define boolean for trading permission for the algorithm. Stops it from trading until the next trading period.
#self.tradingperm = True
# Warmup the algorithm with prior data for backtesting
#self.set_warmup(timedelta(7))
self.set_warmup(30)
########### Universe Selection ###########
self.universe_settings.resolution = Resolution.MINUTE
self.universe_settings.leverage = 2
#self.add_universe(self.CoarseFilter)
self.add_universe(self._fundamental_selection_function)
# Variables for universe selection model
self.coarse_count = 10
self.averages = { }
self.symbols = []
self.data = {}
########### Elders Triple Screen Signals ###########
self.ema_period = int(self.get_parameter("ema_period"))
self.rsi_period = int(self.get_parameter("rsi_period"))
self.macd_fast_period = int(self.get_parameter("macd_period"))
self.macd_slow_period = int(self.macd_fast_period*2)
self.macd_signal_period = int(self.macd_fast_period/2)
########### Risk Management Model ###########
self.add_risk_management(MaximumDrawdownPercentPerSecurity(0.10))
########### Order Execution Model ###########
self.set_execution(ImmediateExecutionModel())
########### Portfolio Construction Model ###########
#self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())#rebalance=self.date_rules.week_start("SPY")))
self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())
#self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())
self.Settings.FreePortfolioValuePercentage=0.05
########### Scheduler ###########
# Run 30 mins after open to collect signals
# Schedule insight generation every 4 hours
#self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.every(timedelta(minutes=180)), self.EvaluateIndicators)
# daily trading option
self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.at(11,30,0, TimeZones.NEW_YORK), self.EvaluateIndicators)
########## Alpha Models ############
'''
self.add_alpha(self.EvaluateEMA())
self.add_alpha(self.EvaluateRSI())
self.add_alpha(self.EvaluateMACD())
'''
def _fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
filtered = [f for f in fundamental if f.price > 10 and f.has_fundamental_data and not np.isnan(f.valuation_ratios.pe_ratio)]
sorted_by_dollar_volume = sorted(filtered, key=lambda f: f.dollar_volume, reverse=True)[:100]
sorted_by_pe_ratio = sorted(sorted_by_dollar_volume, key=lambda f: f.valuation_ratios.pe_ratio, reverse=False)[:self.coarse_count]
return [f.symbol for f in sorted_by_pe_ratio]
def FineFilter(self, fine):
pass
''' Simpler changes to universe '''
# this event fires whenever we have changes to our universe
def on_securities_changed(self, changes):
# liquidate removed securities
for security in changes.removed_securities:
if security.invested:
self.liquidate(security.symbol)
self.Debug(f"Liquidated {security.symbol}")
# we want 20% allocation in each security in our universe - This gives 10% allocation
for security in changes.added_securities:
self.set_holdings(security.symbol, 0.1)
self.Debug(f"Bought {security.symbol}")
############### Evaluate Indicators #########################
''' Elder's Triple Screen Alpha Logic'''
''' Simpler combined method for insights. Either keep these seperate or consolidate into a "Momentum" indicator '''
def EvaluateIndicators(self):
for symbol, indicators in self.data.items():
ema = indicators['ema']
rsi = indicators['rsi']
macd = indicators['macd']
if not (ema.IsReady and rsi.IsReady and macd.IsReady):
continue
ema_value = ema.Current.Value
rsi_value = rsi.Current.Value
macd_value = macd.Current.Value
macd_signal_value = macd.Signal.Current.Value
price = self.Securities[symbol].Price
direction = InsightDirection.Flat
confidence = 0
magnitude = 0
if rsi_value < 45 or macd_value > macd_signal_value: #and price > ema_value:
direction = InsightDirection.Up
confidence = (70 - rsi_value) / 100
magnitude = macd_value - macd_signal_value
elif rsi_value > 70 and macd_value < macd_signal_value:# or price < ema_value:
direction = InsightDirection.Down
confidence = (rsi_value - 30) / 100
magnitude = macd_signal_value - macd_value
insight = Insight.Price(symbol, timedelta(days=1), direction, magnitude, confidence)
#self.EmitInsights(insight)
#self.Debug(f"Generated Insight: {insight}")
if confidence > 0.55:
self.emit_insights(insight)
self.Debug(f"Generated Insight: {insight}")
return insightfrom AlgorithmImports import *
#from symbol_calcs import *
class EldersTripleScreenAlpha(QCAlgorithm):
def Initialize(self):
########### Strategy Params ###########
self.SetStartDate(2017, 1, 1)
self.SetEndDate(2019, 12, 31)
self.SetCash(100000)
# Define boolean for trading permission for the algorithm. Stops it from trading until the next trading period.
#self.tradingperm = True
# Warmup the algorithm with prior data for backtesting
#self.set_warmup(timedelta(7))
self.set_warmup(30)
########### Universe Selection ###########
self.universe_settings.resolution = Resolution.MINUTE
self.universe_settings.leverage = 2
#self.add_universe(self.CoarseFilter)
self.add_universe(self._fundamental_selection_function)
# Variables for universe selection model
self.coarse_count = 10
self.averages = { }
self.symbols = []
self.data = {}
########### Elders Triple Screen Signals ###########
self.ema_period = int(self.get_parameter("ema_period"))
self.rsi_period = int(self.get_parameter("rsi_period"))
self.macd_fast_period = int(self.get_parameter("macd_period"))
self.macd_slow_period = int(self.macd_fast_period*2)
self.macd_signal_period = int(self.macd_fast_period/2)
########### Risk Management Model ###########
self.add_risk_management(MaximumDrawdownPercentPerSecurity(0.10))
########### Order Execution Model ###########
self.set_execution(ImmediateExecutionModel())
########### Portfolio Construction Model ###########
#self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())#rebalance=self.date_rules.week_start("SPY")))
self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())
#self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())
self.Settings.FreePortfolioValuePercentage=0.05
########### Scheduler ###########
# Run 30 mins after open to collect signals
# Schedule insight generation every 4 hours
#self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.every(timedelta(minutes=180)), self.EvaluateIndicators)
# daily trading option
self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.at(11,30,0, TimeZones.NEW_YORK), self.EvaluateIndicators)
########## Alpha Models ############
'''
self.add_alpha(self.EvaluateEMA())
self.add_alpha(self.EvaluateRSI())
self.add_alpha(self.EvaluateMACD())
'''
def _fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
filtered = [f for f in fundamental if f.price > 10 and f.has_fundamental_data and not np.isnan(f.valuation_ratios.pe_ratio)]
sorted_by_dollar_volume = sorted(filtered, key=lambda f: f.dollar_volume, reverse=True)[:100]
sorted_by_pe_ratio = sorted(sorted_by_dollar_volume, key=lambda f: f.valuation_ratios.pe_ratio, reverse=False)[:self.coarse_count]
return [f.symbol for f in sorted_by_pe_ratio]
def FineFilter(self, fine):
pass
''' Simpler changes to universe '''
# this event fires whenever we have changes to our universe
def on_securities_changed(self, changes):
# liquidate removed securities
for security in changes.removed_securities:
if security.invested:
self.liquidate(security.symbol)
self.Debug(f"Liquidated {security.symbol}")
# we want 20% allocation in each security in our universe - This gives 10% allocation
for security in changes.added_securities:
self.set_holdings(security.symbol, 0.1)
self.Debug(f"Bought {security.symbol}")
############### Evaluate Indicators #########################
''' Elder's Triple Screen Alpha Logic'''
''' Simpler combined method for insights. Either keep these seperate or consolidate into a "Momentum" indicator '''
def EvaluateIndicators(self):
for symbol, indicators in self.data.items():
ema = indicators['ema']
rsi = indicators['rsi']
macd = indicators['macd']
if not (ema.IsReady and rsi.IsReady and macd.IsReady):
continue
ema_value = ema.Current.Value
rsi_value = rsi.Current.Value
macd_value = macd.Current.Value
macd_signal_value = macd.Signal.Current.Value
price = self.Securities[symbol].Price
direction = InsightDirection.Flat
confidence = 0
magnitude = 0
if rsi_value < 45 or macd_value > macd_signal_value: #and price > ema_value:
direction = InsightDirection.Up
confidence = (70 - rsi_value) / 100
magnitude = macd_value - macd_signal_value
elif rsi_value > 70 and macd_value < macd_signal_value:# or price < ema_value:
direction = InsightDirection.Down
confidence = (rsi_value - 30) / 100
magnitude = macd_signal_value - macd_value
insight = Insight.Price(symbol, timedelta(days=1), direction, magnitude, confidence)
#self.EmitInsights(insight)
#self.Debug(f"Generated Insight: {insight}")
if confidence > 0.55:
self.emit_insights(insight)
self.Debug(f"Generated Insight: {insight}")
return insight
from AlgorithmImports import *
from symbol_calcs import *
class EldersTripleScreenAlpha(QCAlgorithm):
def Initialize(self):
''' Initial Algo parameters and QuantConnect methods'''
########### Strategy Params ###########
self.SetStartDate(2017, 1, 1)
self.SetEndDate(2019, 12, 31)
# Out of Sample (OoS) testing date
#self.SetEndDate(2024, 3, 31)
#self.SetCash(100000)
self.SetCash(100000000)
# Define boolean for trading permission for the algorithm. Stops it from trading until the next trading period.
#self.tradingperm = True
# Warmup the algorithm with prior data for backtesting
self.set_warmup(timedelta(30))
########### Universe Selection ###########
self.universe_settings.resolution = Resolution.HOUR
self.universe_settings.leverage = 2
#self.add_universe(self.CoarseFilter)
self.add_universe(self._fundamental_selection_function)
# Variables for universe selection model
self.coarse_count = 10
self.averages = { }
self.symbols = []
self.data = {}
'''
########### Elders Triple Screen Signals ###########
self.ema_period = int(self.get_parameter("ema_period"))
self.rsi_period = int(self.get_parameter("rsi_period"))
self.macd_fast_period = int(self.get_parameter("macd_period"))
self.macd_slow_period = int(self.macd_fast_period*2)
self.macd_signal_period = int(self.macd_fast_period/2)
'''
########### Elders Triple Screen Signals ###########
self.ema_period = 20
self.rsi_period = 14
self.macd_fast_period = 12
self.macd_slow_period = 26
self.macd_signal_period = 6
########### Risk Management Model ###########
self.add_risk_management(MaximumDrawdownPercentPerSecurity(0.05))
########### Order Execution Model ###########
# Immediat execution model - so trades right at 1130 each day
self.set_execution(ImmediateExecutionModel())
# VWAP Execution Model, to later be changed to the EMA Model
#self.set_execution(VolumeWeightedAveragePriceExecutionModel())
########### Portfolio Construction Model ###########
#self.set_portfolio_construction(AccumulativeInsightPortfolioConstructionModel())#rebalance=self.date_rules.week_start("SPY")))
#self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(rebalance=timedelta(7), portfolioBias=PortfolioBias.LONG))
#self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(portfolioBias=PortfolioBias.LONG))
#self.set_portfolio_construction(RiskParityPortfolioConstructionModel(portfolioBias=PortfolioBias.LONG))
self.set_portfolio_construction(RiskParityPortfolioConstructionModel(portfolioBias=PortfolioBias.LONG))
#self.settings.rebalance_portfolio_on_insight_changes = False
# Required portfolio free cash value
self.Settings.FreePortfolioValuePercentage=0.05
########### Reality Modeling Parameters ###########
# Brokerage Simulation
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
# Risk Free Interest Rate Modeling
self.set_risk_free_interest_rate_model(InterestRateProvider())
# Margin Call Modeling
self.portfolio.margin_call_model = DefaultMarginCallModel(self.portfolio, self.default_order_properties)
########### Scheduler ###########
# Run 30 mins after open to collect signals
# Schedule insight generation every 4 hours
#self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.every(timedelta(minutes=90)), self.EvaluateIndicators)
# daily trading option
self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.at(11,30,0, TimeZones.NEW_YORK), self.EvaluateIndicators)
########### Object Store ##########
self.universe_store = ''
self.EMA_store = ''
self.RSI_store = ''
self.MACD_store = ''
########### Load Existing Universe ###########
if self.IsLiveEnvironment():
self.LoadExistingUniverse()
def IsLiveEnvironment(self):
''' Check if the algorithm is running in a live environment '''
return self.LiveMode
def LoadExistingUniverse(self):
''' Load the existing universe from the live brokerage account holdings '''
self.Debug("Loading existing universe from brokerage account...")
# Fetch existing holdings
holdings = [x.Symbol for x in self.Portfolio.Values if x.Invested]
self.Debug(f"Found {len(holdings)} existing holdings.")
# Register indicators for the existing holdings
for symbol in holdings:
self.data[symbol] = {
'ema': ExponentialMovingAverage(self.ema_period),
'rsi': RelativeStrengthIndex(self.rsi_period, MovingAverageType.Wilders),
'macd': MovingAverageConvergenceDivergence(self.macd_fast_period, self.macd_slow_period, self.macd_signal_period, MovingAverageType.Wilders)
}
self.RegisterIndicator(symbol, self.data[symbol]['ema'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['rsi'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['macd'], Resolution.Daily)
return holdings
def _fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
''' Fundamental Filter to produce a list of securities within the universe'''
filtered = [f for f in fundamental if f.price > 10 and \
f.has_fundamental_data and \
not np.isnan(f.valuation_ratios.pe_ratio) and \
f.valuation_ratios.forward_pe_ratio > 5 and \
f.market_cap > 500000]
self.Debug(f"Filtered securities: {', '.join([f.symbol.Value for f in filtered])}")
sorted_by_dollar_volume = sorted(filtered, key=lambda f: f.dollar_volume, reverse=True)[:100]
self.Debug(f"Top 100 by dollar volume: {', '.join([f.symbol.Value for f in sorted_by_dollar_volume])}")
sorted_by_pe_ratio = sorted(sorted_by_dollar_volume, key=lambda f: f.valuation_ratios.pe_ratio, reverse=False)[:20]
self.Debug(f"Final selected securities: {', '.join([f.symbol.Value for f in sorted_by_pe_ratio])}")
self.Debug(f"Final selected securities: {sorted_by_pe_ratio[:20]}")
return [f.symbol for f in sorted_by_pe_ratio[:20]]
def FineFilter(self, fine):
''' Placeholder for further filtering of the universe '''
pass
''' Simpler changes to universe
# this event fires whenever we have changes to our universe
def on_securities_changed(self, changes):
# liquidate removed securities
for security in changes.removed_securities:
if security.invested:
self.liquidate(security.symbol)
self.Debug(f"Liquidated {security.symbol}")
# we want 20% allocation in each security in our universe - This gives 10% allocation
for security in changes.added_securities:
self.set_holdings(security.symbol, 0.1)
self.Debug(f"Bought {security.symbol}")
'''
############### Evaluate Indicators #########################
''' Elder's Triple Screen Alpha Logic'''
def OnSecuritiesChanged(self, changes):
''' Add indicators whenever securities are added to or removed from the universe'''
for security in changes.AddedSecurities:
symbol = security.Symbol
self.data[symbol] = {
'ema': ExponentialMovingAverage(self.ema_period),
'rsi': RelativeStrengthIndex(self.rsi_period, MovingAverageType.Wilders),
'macd': MovingAverageConvergenceDivergence(self.macd_fast_period, self.macd_slow_period, self.macd_signal_period, MovingAverageType.Wilders)
}
self.RegisterIndicator(symbol, self.data[symbol]['ema'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['rsi'], Resolution.HOUR)
self.RegisterIndicator(symbol, self.data[symbol]['macd'], Resolution.Daily)
# Reality Modeling for slippage
#security.set_slippage_model(VolumeShareSlippageModel(0.025, 0.1))
#security.set_slippage_model(MarketImpactSlippageModel(self))
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.data:
del self.data[symbol]
def EvaluateIndicators(self):
''' Evaluate the indicators for the securities '''
# Maybe change this self.data.items() to something else. Rn its not iterating through all the securities like it should
for symbol, indicators in self.data.items():
ema = indicators['ema']
rsi = indicators['rsi']
macd = indicators['macd']
if not (ema.IsReady and rsi.IsReady and macd.IsReady):
continue
ema_value = ema.Current.Value
rsi_value = rsi.Current.Value
macd_value = macd.Current.Value
macd_signal_value = macd.Signal.Current.Value
price = self.Securities[symbol].Price
direction = InsightDirection.Flat
confidence = 0
magnitude = 0
if rsi_value < 30 and macd_value > macd_signal_value: #and price > ema_value:
direction = InsightDirection.Up
confidence = (70 - rsi_value) / 100
magnitude = macd_value - macd_signal_value
# Consider changing this to an AND inistead of an or, since both need to be satisfied
elif rsi_value > 70 or macd_value < macd_signal_value:# or price < ema_value:
direction = InsightDirection.Down
confidence = (rsi_value - 30) / 100
magnitude = macd_signal_value - macd_value
insight = Insight.Price(symbol, timedelta(days=1), direction, magnitude, confidence)
#self.EmitInsights(insight)
#self.Debug(f"Generated Insight: {insight}")
if confidence > 0.2:
self.emit_insights(insight)
self.Debug(f"Generated Insight: {insight}")
# Store indicator values in an object
#self.plot('EMA', 'Value', self.ema.current.value)
#self.EMA_store += f'{self.ema.current.end_time},{self.sma.current.value}\n'
#self.RSI_store += f'{self.rsi.current.end_time},{self.rsi.current.value}\n'
#self.MACD_store += f'{self.macd.current.end_time},{self.macd.current.value}\n'
return insight
def on_end_of_algorithm(self):
pass #placeholder for now, will infact do clean endings for the algo where it saves the status of all the indicators
#self.object_store.save('EMA_values_python', self.EMA_store)
#self.object_store.save('RSI_values_python', self.RSI_store)
#self.object_store.save('MACD_values_python', self.MACD_store)#region imports
from AlgorithmImports import *
#endregion
# Note this class was copied from bottom of main.py, it can technically be run seperately as long as the import is there
class SymbolData(object):
def __init__(self, symbol):
self._symbol = symbol
self.tolerance = 1.01
self.fast = ExponentialMovingAverage(100)
self.slow = ExponentialMovingAverage(300)
self.is_uptrend = False
self.scale = 1
print("SymbolData Init Complete")
def update(self, time, value):
if self.fast.update(time, value) and self.slow.update(time, value):
fast = self.fast.current.value
slow = self.slow.current.value
self.is_uptrend = fast > slow * self.tolerance
if self.is_uptrend:
self.scale = (fast - slow) / ((fast + slow) / 2.0)
return
print("Update method ran successfully")#region imports from AlgorithmImports import * #endregion # Your New Python File