Overall Statistics |
Total Trades 5455 Average Win 0.54% Average Loss -0.55% Compounding Annual Return -0.096% Drawdown 51.000% Expectancy 0.015 Net Profit -0.480% Sharpe Ratio 0.156 Probabilistic Sharpe Ratio 1.078% Loss Rate 49% Win Rate 51% Profit-Loss Ratio 0.98 Alpha 0.056 Beta -0.081 Annual Standard Deviation 0.315 Annual Variance 0.099 Information Ratio -0.085 Tracking Error 0.369 Treynor Ratio -0.607 Total Fees $101442.70 Estimated Strategy Capacity $34000000.00 Lowest Capacity Asset CRM SZQUJUA9SVOL Portfolio Turnover 55.44% |
#region imports from AlgorithmImports import * #endregion # QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from clr import AddReference AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") AddReference("QuantConnect.Indicators") from QuantConnect import * from QuantConnect.Indicators import * from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Alphas import * import numpy as np class EmaCrossAlphaModel(AlphaModel): '''Alpha model that uses an EMA cross to create insights''' def __init__(self, fastPeriod = 12, slowPeriod = 26, resolution = Resolution.Daily): '''Initializes a new instance of the EmaCrossAlphaModel class Args: fastPeriod: The fast EMA period slowPeriod: The slow EMA period''' self.fastPeriod = fastPeriod self.slowPeriod = slowPeriod self.resolution = resolution self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(resolution), fastPeriod) self.symbolDataBySymbol = {} resolutionString = Extensions.GetEnumString(resolution, Resolution) self.Name = '{}({},{},{})'.format(self.__class__.__name__, fastPeriod, slowPeriod, resolutionString) def Update(self, algorithm, data): '''Updates this alpha model with the latest data from the algorithm. This is called each time the algorithm receives data for subscribed securities Args: algorithm: The algorithm instance data: The new data available Returns: The new insights generated''' insights = [] pos = 0 neg = 0.0 for symbol, symbolData in self.symbolDataBySymbol.items(): if symbolData.Fast.IsReady and symbolData.Slow.IsReady: mag = (symbolData.Fast.Current.Value-symbolData.Slow.Current.Value )/\ np.mean([symbolData.Fast.Current.Value,symbolData.Slow.Current.Value]) if symbolData.FastIsOverSlow: if symbolData.Slow > symbolData.Fast: neg += -mag elif symbolData.SlowIsOverFast: if symbolData.Fast > symbolData.Slow: pos += mag tot = pos + neg for symbol, symbolData in self.symbolDataBySymbol.items(): if symbolData.Fast.IsReady and symbolData.Slow.IsReady: mag = (symbolData.Fast.Current.Value-symbolData.Slow.Current.Value )/\ np.mean([symbolData.Fast.Current.Value,symbolData.Slow.Current.Value]) if symbolData.FastIsOverSlow: if symbolData.Slow > symbolData.Fast: insights.append(Insight.Price(symbolData.Symbol, self.predictionInterval, InsightDirection.Down, mag,-mag, "MyAlphaModel",-mag/tot)) elif symbolData.SlowIsOverFast: if symbolData.Fast > symbolData.Slow: insights.append(Insight.Price(symbolData.Symbol, self.predictionInterval, InsightDirection.Up, mag, mag, "MyAlphaModel",mag/tot)) symbolData.FastIsOverSlow = symbolData.Fast > symbolData.Slow return insights def OnSecuritiesChanged(self, algorithm, changes): '''Event fired each time the we add/remove securities from the data feed Args: algorithm: The algorithm instance that experienced the change in securities changes: The security additions and removals from the algorithm''' for added in changes.AddedSecurities: symbolData = self.symbolDataBySymbol.get(added.Symbol) if symbolData is None: # create fast/slow EMAs symbolData = SymbolData(added) symbolData.Fast = algorithm.EMA(added.Symbol, self.fastPeriod, self.resolution) symbolData.Slow = algorithm.EMA(added.Symbol, self.slowPeriod, self.resolution) self.symbolDataBySymbol[added.Symbol] = symbolData else: # a security that was already initialized was re-added, reset the indicators symbolData.Fast.Reset() symbolData.Slow.Reset() class SymbolData: '''Contains data specific to a symbol required by this model''' def __init__(self, security): self.Security = security self.Symbol = security.Symbol self.Fast = None self.Slow = None # True if the fast is above the slow, otherwise false. # This is used to prevent emitting the same signal repeatedly self.FastIsOverSlow = False @property def SlowIsOverFast(self): return not self.FastIsOverSlow
#region imports from AlgorithmImports import * #endregion class HighPERatioUniverseSelectionModel(FineFundamentalUniverseSelectionModel): def __init__(self, universe_settings: UniverseSettings = None) -> None: super().__init__(self.SelectCoarse, self.SelectFine, universe_settings) def SelectCoarse(self, coarse: List[CoarseFundamental]) -> List[Symbol]: selected = [c for c in coarse if c.HasFundamentalData] sorted_by_dollar_volume = sorted(selected, key=lambda c: c.DollarVolume, reverse=True) return [c.Symbol for c in sorted_by_dollar_volume[:100]] def SelectFine(self, fine: List[FineFundamental]) -> List[Symbol]: sorted_by_pe_ratio = sorted(fine, key=lambda x: x.ValuationRatios.PERatio, reverse=True) return [c.Symbol for c in sorted_by_pe_ratio[:10]]
#region imports from AlgorithmImports import * #endregion class LowPERatioUniverseSelectionModel(FineFundamentalUniverseSelectionModel): def __init__(self, universe_settings: UniverseSettings = None) -> None: super().__init__(self.SelectCoarse, self.SelectFine, universe_settings) def SelectCoarse(self, coarse: List[CoarseFundamental]) -> List[Symbol]: selected = [c for c in coarse if c.HasFundamentalData] sorted_by_dollar_volume = sorted(selected, key=lambda c: c.DollarVolume, reverse=True) return [c.Symbol for c in sorted_by_dollar_volume[:100]] def SelectFine(self, fine: List[FineFundamental]) -> List[Symbol]: sorted_by_pe_ratio = sorted(fine, key=lambda x: x.ValuationRatios.PERatio, reverse=False) return [c.Symbol for c in sorted_by_pe_ratio[:10]]
#region imports from AlgorithmImports import * #endregion # QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from clr import AddReference AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") AddReference("QuantConnect.Indicators") from QuantConnect import * from QuantConnect.Indicators import * from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Alphas import * class MacdAlphaModel(AlphaModel): '''Defines a custom alpha model that uses MACD crossovers. The MACD signal line is used to generate up/down insights if it's stronger than the bounce threshold. If the MACD signal is within the bounce threshold then a flat price insight is returned.''' def __init__(self, fastPeriod = 12, slowPeriod = 26, signalPeriod = 9, movingAverageType = MovingAverageType.Exponential, resolution = Resolution.Daily): ''' Initializes a new instance of the MacdAlphaModel class Args: fastPeriod: The MACD fast period slowPeriod: The MACD slow period</param> signalPeriod: The smoothing period for the MACD signal movingAverageType: The type of moving average to use in the MACD''' self.fastPeriod = fastPeriod self.slowPeriod = slowPeriod self.signalPeriod = signalPeriod self.movingAverageType = movingAverageType self.resolution = resolution self.insightPeriod = Time.Multiply(Extensions.ToTimeSpan(resolution), fastPeriod) self.bounceThresholdPercent = 0.01 self.symbolData = {} resolutionString = Extensions.GetEnumString(resolution, Resolution) movingAverageTypeString = Extensions.GetEnumString(movingAverageType, MovingAverageType) self.Name = '{}({},{},{},{},{})'.format(self.__class__.__name__, fastPeriod, slowPeriod, signalPeriod, movingAverageTypeString, resolutionString) def Update(self, algorithm, data): ''' Determines an insight for each security based on it's current MACD signal Args: algorithm: The algorithm instance data: The new data available Returns: The new insights generated''' insights = [] for key, sd in self.symbolData.items(): if sd.Security.Price == 0: continue direction = InsightDirection.Flat normalized_signal = sd.MACD.Signal.Current.Value / sd.Security.Price if normalized_signal > self.bounceThresholdPercent: direction = InsightDirection.Up elif normalized_signal < -self.bounceThresholdPercent: direction = InsightDirection.Down # ignore signal for same direction as previous signal if direction == sd.PreviousDirection: continue insight = Insight.Price(sd.Security.Symbol, self.insightPeriod, direction,normalized_signal) sd.PreviousDirection = insight.Direction insights.append(insight) return insights def OnSecuritiesChanged(self, algorithm, changes): '''Event fired each time the we add/remove securities from the data feed. This initializes the MACD for each added security and cleans up the indicator for each removed security. Args: algorithm: The algorithm instance that experienced the change in securities changes: The security additions and removals from the algorithm''' for added in changes.AddedSecurities: self.symbolData[added.Symbol] = SymbolData(algorithm, added, self.fastPeriod, self.slowPeriod, self.signalPeriod, self.movingAverageType, self.resolution) for removed in changes.RemovedSecurities: data = self.symbolData.pop(removed.Symbol, None) if data is not None: # clean up our consolidator algorithm.SubscriptionManager.RemoveConsolidator(removed.Symbol, data.Consolidator) class SymbolData: def __init__(self, algorithm, security, fastPeriod, slowPeriod, signalPeriod, movingAverageType, resolution): self.Security = security self.MACD = MovingAverageConvergenceDivergence(fastPeriod, slowPeriod, signalPeriod, movingAverageType) self.Consolidator = algorithm.ResolveConsolidator(security.Symbol, resolution) algorithm.RegisterIndicator(security.Symbol, self.MACD, self.Consolidator) self.PreviousDirection = None
#region imports from AlgorithmImports import * #endregion # QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. # Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from clr import AddReference AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Algorithm.Framework") AddReference("QuantConnect.Indicators") AddReference("QuantConnect.Logging") AddReference("QuantConnect.Common") from QuantConnect import * from QuantConnect.Indicators import * from QuantConnect.Logging import Log from QuantConnect.Algorithm import * from QuantConnect.Algorithm.Framework import * from QuantConnect.Algorithm.Framework.Alphas import * from datetime import timedelta from enum import Enum class RsiAlphaModel(AlphaModel): '''Uses Wilder's RSI to create insights. Using default settings, a cross over below 30 or above 70 will trigger a new insight.''' def __init__(self, period = 14, resolution = Resolution.Daily): '''Initializes a new instance of the RsiAlphaModel class Args: period: The RSI indicator period''' self.period = period self.resolution = resolution self.insightPeriod = Time.Multiply(Extensions.ToTimeSpan(resolution), period) self.symbolDataBySymbol ={} self.lowcutoff = 30.0 self.highcutoff = 70.0 self.buffer = 5.0 resolutionString = Extensions.GetEnumString(resolution, Resolution) self.Name = '{}({},{})'.format(self.__class__.__name__, period, resolutionString) def Update(self, algorithm, data): '''Updates this alpha model with the latest data from the algorithm. This is called each time the algorithm receives data for subscribed securities Args: algorithm: The algorithm instance data: The new data available Returns: The new insights generated''' insights = [] for symbol, symbolData in self.symbolDataBySymbol.items(): rsi = symbolData.RSI previous_state = symbolData.State state = self.GetState(rsi, previous_state) if state != previous_state and rsi.IsReady: if state == State.TrippedLow: insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Up, 0.01*(self.lowcutoff-rsi.Current.Value)/self.lowcutoff)) if state == State.TrippedHigh: insights.append(Insight.Price(symbol, self.insightPeriod, InsightDirection.Down, 0.01*(rsi.Current.Value-self.highcutoff)/(100-self.highcutoff))) symbolData.State = state return insights def OnSecuritiesChanged(self, algorithm, changes): '''Cleans out old security data and initializes the RSI for any newly added securities. Event fired each time the we add/remove securities from the data feed Args: algorithm: The algorithm instance that experienced the change in securities changes: The security additions and removals from the algorithm''' # clean up data for removed securities symbols = [ x.Symbol for x in changes.RemovedSecurities ] if len(symbols) > 0: for subscription in algorithm.SubscriptionManager.Subscriptions: if subscription.Symbol in symbols: self.symbolDataBySymbol.pop(subscription.Symbol, None) subscription.Consolidators.Clear() # initialize data for added securities addedSymbols = [ x.Symbol for x in changes.AddedSecurities if x.Symbol not in self.symbolDataBySymbol] if len(addedSymbols) == 0: return history = algorithm.History(addedSymbols, self.period, self.resolution) for symbol in addedSymbols: rsi = algorithm.RSI(symbol, self.period, MovingAverageType.Wilders, self.resolution) if not history.empty: ticker = SymbolCache.GetTicker(symbol) if ticker not in history.index.levels[0]: Log.Trace(f'RsiAlphaModel.OnSecuritiesChanged: {ticker} not found in history data frame.') continue for tuple in history.loc[ticker].itertuples(): rsi.Update(tuple.Index, tuple.close) self.symbolDataBySymbol[symbol] = SymbolData(symbol, rsi) def GetState(self, rsi, previous): ''' Determines the new state. This is basically cross-over detection logic that includes considerations for bouncing using the configured bounce tolerance.''' if rsi.Current.Value > self.highcutoff : return State.TrippedHigh if rsi.Current.Value < self.lowcutoff : return State.TrippedLow if previous == State.TrippedLow: if rsi.Current.Value > (self.lowcutoff+self.buffer): return State.Middle if previous == State.TrippedHigh: if rsi.Current.Value < (self.highcutoff-self.buffer): return State.Middle return previous class SymbolData: '''Contains data specific to a symbol required by this model''' def __init__(self, symbol, rsi): self.Symbol = symbol self.RSI = rsi self.State = State.Middle class State(Enum): '''Defines the state. This is used to prevent signal spamming and aid in bounce detection.''' TrippedLow = 0 Middle = 1 TrippedHigh = 2
#region imports from AlgorithmImports import * #endregion from datetime import timedelta from QuantConnect.Data.UniverseSelection import * from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity from HighPERatioUnivese import HighPERatioUniverseSelectionModel from LowPERatioUniverse import LowPERatioUniverseSelectionModel from MacdAlphaModel import MacdAlphaModel from RsiAlphaModel import RsiAlphaModel class MakretNeutralAlgoFramework(QCAlgorithm): def Initialize(self): self.SetStartDate(2018, 1, 1) self.SetEndDate(2023, 1, 1) self.SetCash(1000000) self.UniverseSettings.Resolution = Resolution.Hour #self.SetSecurityInitializer(lambda x: x.SetMarketPrice(self.GetLastKnownPrice(x))) self.AddUniverseSelection(HighPERatioUniverseSelectionModel()) #Short Universe self.AddUniverseSelection(LowPERatioUniverseSelectionModel()) #Long Universe self.AddAlpha(MacdAlphaModel(12, 26, 9, MovingAverageType.Simple, Resolution.Daily)) #Long Alpha self.AddAlpha(RsiAlphaModel(10, Resolution.Daily)) #Short Alpha self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel()) self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.01)) self.SetExecution(ImmediateExecutionModel()) self.SetWarmUp(300, Resolution.Hour)