Overall Statistics |
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -0.85 Tracking Error 0.345 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset Portfolio Turnover 0% |
# 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 AlgorithmImports import * class CustomHistoricalReturnsAlphaModel(AlphaModel): '''Uses Historical returns to create insights.''' def __init__(self, *args, **kwargs): '''Initializes a new default instance of the HistoricalReturnsAlphaModel class. Args: lookback(int): Historical return lookback period resolution: The resolution of historical data''' self.lookback = kwargs['lookback'] if 'lookback' in kwargs else 1 self.resolution = kwargs['resolution'] if 'resolution' in kwargs else Resolution.Daily self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), self.lookback) self.symbolDataBySymbol = {} 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(): if symbolData.CanEmit: direction = InsightDirection.Flat magnitude = symbolData.Return if magnitude > 0: direction = InsightDirection.Up if magnitude < 0: direction = InsightDirection.Down insights.append(Insight.Price(symbol, self.predictionInterval, direction, magnitude, None)) 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''' # clean up data for removed securities for removed in changes.RemovedSecurities: symbolData = self.symbolDataBySymbol.pop(removed.Symbol, None) if symbolData is not None: symbolData.RemoveConsolidators(algorithm) # initialize data for added securities symbols = [ x.Symbol for x in changes.AddedSecurities ] history = algorithm.History(symbols, self.lookback, self.resolution) if history.empty: return tickers = history.index.levels[0] for ticker in tickers: symbol = SymbolCache.GetSymbol(ticker) if symbol not in self.symbolDataBySymbol: symbolData = SymbolData(symbol, self.lookback) self.symbolDataBySymbol[symbol] = symbolData symbolData.RegisterIndicators(algorithm, self.resolution) symbolData.WarmUpIndicators(history.loc[ticker]) class SymbolData: '''Contains data specific to a symbol required by this model''' def __init__(self, symbol, lookback): self.Symbol = symbol self.ROC = RateOfChange('{}.ROC({})'.format(symbol, lookback), lookback) self.Consolidator = None self.previous = 0 def RegisterIndicators(self, algorithm, resolution): self.Consolidator = algorithm.ResolveConsolidator(self.Symbol, resolution) algorithm.RegisterIndicator(self.Symbol, self.ROC, self.Consolidator) def RemoveConsolidators(self, algorithm): if self.Consolidator is not None: algorithm.SubscriptionManager.RemoveConsolidator(self.Symbol, self.Consolidator) def WarmUpIndicators(self, history): for tuple in history.itertuples(): self.ROC.Update(tuple.Index, tuple.close) @property def Return(self): return float(self.ROC.Current.Value) @property def CanEmit(self): if self.previous == self.ROC.Samples: return False self.previous = self.ROC.Samples return self.ROC.IsReady def __str__(self, **kwargs): return '{}: {:.2%}'.format(self.ROC.Name, (1 + self.Return)**252 - 1)
# 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. ###### TODO: make this calculate sortino instead and make decisions on that. from AlgorithmImports import * import quantstats as qs class StatBasedAlphaModel(AlphaModel): '''Uses Historical returns to create insights.''' def __init__(self, *args, **kwargs): '''Initializes a new default instance of the HistoricalReturnsAlphaModel class. Args: lookback(int): Historical return lookback period resolution: The resolution of historical data''' self.lookback = kwargs['lookback'] if 'lookback' in kwargs else 1 self.resolution = kwargs['resolution'] if 'resolution' in kwargs else Resolution.Daily self.predictionInterval = Time.Multiply(Extensions.ToTimeSpan(self.resolution), self.lookback) self.symbolDataBySymbol = {} 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(): if symbolData.CanEmit: direction = InsightDirection.Flat magnitude = symbolData.Return if magnitude > 0: direction = InsightDirection.Up if magnitude < 0: direction = InsightDirection.Down insights.append(Insight.Price(symbol, self.predictionInterval, direction, magnitude, None)) 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''' # clean up data for removed securities for removed in changes.RemovedSecurities: symbolData = self.symbolDataBySymbol.pop(removed.Symbol, None) if symbolData is not None: symbolData.RemoveConsolidators(algorithm) # initialize data for added securities symbols = [ x.Symbol for x in changes.AddedSecurities ] history = algorithm.History(symbols, self.lookback, self.resolution) if history.empty: return tickers = history.index.levels[0] for ticker in tickers: symbol = SymbolCache.GetSymbol(ticker) if symbol not in self.symbolDataBySymbol: symbolData = SymbolData(symbol, self.lookback) self.symbolDataBySymbol[symbol] = symbolData symbolData.RegisterIndicators(algorithm, self.resolution) symbolData.WarmUpIndicators(history.loc[ticker]) class SymbolData: '''Contains data specific to a symbol required by this model''' def __init__(self, symbol, lookback): self.Symbol = symbol self.SORTINO = SORTINO('{}.SORTINO({})'.format(symbol, lookback), lookback) self.Consolidator = None self.previous = 0 def RegisterIndicators(self, algorithm, resolution): self.Consolidator = algorithm.ResolveConsolidator(self.Symbol, resolution) algorithm.RegisterIndicator(self.Symbol, self.SORTINO, self.Consolidator) def RemoveConsolidators(self, algorithm): if self.Consolidator is not None: algorithm.SubscriptionManager.RemoveConsolidator(self.Symbol, self.Consolidator) def WarmUpIndicators(self, history): for tuple in history.itertuples(): self.SORTINO.Update(tuple.Index, tuple.close) @property def Return(self): self.Log(f"Sortino {self.SORTINO.Current.Value}") return float(self.SORTINO.Current.Value) @property def CanEmit(self): if self.previous == self.SORTINO.Samples: return False self.previous = self.SORTINO.Samples return self.SORTINO.IsReady def __str__(self, **kwargs): return '{}: {:.2%}'.format(self.SORTINO.Name, (1 + self.Return)**252 - 1)
# region imports from AlgorithmImports import * # endregion #from StatBasedAlphaModel import StatBasedAlphaModel from HistoricalReturnsAlphaModel import CustomHistoricalReturnsAlphaModel class FormalVioletGaur(QCAlgorithm): # def Initialize(self): # self.SetStartDate(1995, 1, 1) # Set Start Date # self.SetCash(100000) # Set Strategy Cash # self.AddEquity("SPY", Resolution.Minute) # class MyFrameworkAlgorithm(QCAlgorithm): def Initialize(self) -> None: # universe tickers = ["AAPL", "COKE", "GOOGL"] symbols = [ Symbol.Create(ticker, SecurityType.Equity, Market.USA) for ticker in tickers] self.AddUniverseSelection(ManualUniverseSelectionModel(symbols)) self.SetBenchmark(SecurityType.Equity, "AAPL") self.SetAlpha(CustomHistoricalReturnsAlphaModel(resolution = Resolution.Daily, lookback = 22)) #self.SetAlpha(HistoricalReturnsAlphaModel(resolution = Resolution.Daily, lookback = 22)) self.SetPortfolioConstruction(MeanVarianceOptimizationPortfolioConstructionModel()) self.SetExecution(ImmediateExecutionModel()) self.SetRiskManagement(NullRiskManagementModel()) #self.SetExecution(ImmediateExecutionModel()) #self.AddRiskManagement() # def OnData(self, data: Slice): # if not self.Portfolio.Invested: # self.SetHoldings("SPY", 0.33)