Model Volatility Contraction/Cup and Handle patterns

| |

Hi Members, I am wondering if we can model a Volatility Contraction Pattern or Cup-n-Handle patterns popularized by Mark Minervini and CANSLIM methodologies. Below are few charts on how they look like:

Volatility Contraction Pattern(VCP):

3 Contracting High-Low combinations before breakout(shown in blue)

In chart this is how it looks like

2 Contracting High-Low combinations before breakout in blue

To Summarize, VCP is characterized by tightening price action towards the right side of stock chart and it is coupled by drying up of volume

Cup-n-Handle:

Typical Cup-n-Handle setup

It is characterized by a U shaped price action coupled with a flat Handle type formations on the right side.

I am trying to achieve it with combination of rolling window+min,max period indicators, but would love to know better approaches from the forum. Thanks!

The material on this website is provided for informational purposes only and does not constitute an offer to sell, a solicitation to buy, or a recommendation or endorsement for any security or strategy, nor does it constitute an offer to provide investment advisory services by QuantConnect. In addition, the material offers no opinion with respect to the suitability of any security or specific investment. QuantConnect makes no guarantees as to the accuracy or completeness of the views expressed in the website. The views are subject to change, and may have become unreliable for various reasons, including changes in market conditions or economic circumstances. All investments involve risk, including loss of principal. You should consult with an investment professional before making any investment decisions.

Hi Log Up,

Consider trying to recognize the patterns with a neural network or with nonparametric kernel regression. The Journal of Finance has a paper on the latter: Foundations of Technical Analysis: Computational Algorithms, Statistical Inference, and Empirical Implementation.

Best,
Derek Melchin

Want to invest in QuantConnect as we build the Linux of quant finance? Checkout our Wefunder campaign to join the revolution.

#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 datetime import datetime from System.Collections.Generic import List #from AlphaModel import * from MyAlpha import * ### <summary> ### In this algorithm we demonstrate how to perform some technical analysis as ### part of your coarse fundamental universe selection ### </summary> ### <meta name="tag" content="using data" /> ### <meta name="tag" content="indicators" /> ### <meta name="tag" content="universes" /> ### <meta name="tag" content="coarse universes" /> class EmaCrossUniverseSelectionAlgorithm(QCAlgorithm): def Initialize(self): '''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.''' self.SetStartDate(date.today()-timedelta(days=365*5)) #Set Start Date #self.SetStartDate(2013,12,1) #Set Start Date self.SetEndDate(date.today()) #Set End Date #self.SetEndDate(2020,1,1) #Set End Date self.SetCash(100000) #Set Strategy Cash self.UniverseSettings.Resolution = Resolution.Daily self.UniverseSettings.Leverage = 2 self.month = 0 self.coarse_count = 1 #self.SetBenchmark("IJH") self.priorBenchmarkPrice=-1.0 self.qb = QuantBook() #self.universeSelected=False #reshuffle quarterly self.rebalanceTime = self.Time # this add universe method accepts two parameters: # - coarse selection function: accepts an IEnumerable<CoarseFundamental> and returns an IEnumerable<Symbol> self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) self.AddAlpha(MyAlphaModel()) #use below if you dont want portfolio construction to rebalance portfolio when universe is changed and security is removed self.Settings.RebalancePortfolioOnSecurityChanges = False self.Settings.RebalancePortfolioOnInsightChanges = False self.SetPortfolioConstruction(MyCustomWeightingPortfolioConstructionModel(self.shallRebalance)) #self.SetPortfolioConstruction(NullPortfolioConstructionModel()) #performing middle #self.SetRiskManagement(NullRiskManagementModel()) #performing best self.AddRiskManagement(TrailingStopRiskWithMinProfitManagementModel(0.08,0.2)) #for this its saying algo needs to be manually restarted... #self.AddRiskManagement(MaximumDrawdownPercentPortfolio(0.08)) #performing worst #self.AddRiskManagement(TrailingStopRiskManagementModel(0.1)) self.SetExecution(ImmediateExecutionModel()) #liquidate portfolio on end date self.Schedule.On(self.DateRules.On(self.EndDate.year, self.EndDate.month, self.EndDate.day), self.TimeRules.At(0, 0), self.LiquidateAll) def LiquidateAll(self): self.Liquidate() self.Debug("liquidated on the last day") def OnData(self,slice): #if end date tomorrow, liquidate everything # if(self.Time.date()==((self.EndDate - timedelta(days = 1)).date())): # self.Liquidate() #prior benchmark is not set if self.priorBenchmarkPrice==-1.0: self.priorBenchmarkPrice = self.Benchmark.Evaluate(self.Time) if self.Time < self.rebalanceTime: return else: #till found better place, liquidated from here #self.Liquidate() #try to liquidate laggers, here issue is happening if u liquidate recent entry candidates which havent gotten chance to gain much currentBenchmarkPrice = self.Benchmark.Evaluate(self.Time) percBenchmarkChg = (currentBenchmarkPrice - self.priorBenchmarkPrice)/self.priorBenchmarkPrice for securityMap in [securityMap for securityMap in self.Portfolio if securityMap.Value.Invested]: #bypassing it as of now noOfDaysSinceBought=0.0 filledSecurityOrders = self.Transactions.GetOrders(lambda o: o.Status == OrderStatus.Filled and o.Symbol.Value == securityMap.Key.Value) # [x for x in self.Transactions.GetOrders(None) if (x.Symbol.Value == "securityMap.Key.Value") and (x.Status == OrderStatus.Filled)] latestOrder= next(security for security in filledSecurityOrders) if latestOrder is not None: noOfDaysSinceBought = (self.Time.date() - latestOrder.Time.date()).days if securityMap.Value.UnrealizedProfitPercent<percBenchmarkChg and noOfDaysSinceBought>=70: #returns lagging benchmark self.Liquidate(securityMap.Key) #in this above code self.rebalanceTime = Expiry.EndOfQuarter(self.Time) self.priorBenchmarkPrice = currentBenchmarkPrice #set current benchmark price as prior price for next rebalancing # pass to fine selection function for market cap def CoarseSelectionFunction(self, coarse): #self.Debug('coarse universe selection called') #return portfolio unchanged if not rebalanceTime if self.Time < self.rebalanceTime: return Universe.Unchanged filteredByVolume = list(filter(lambda x: x.Volume >= 400000 and x.HasFundamentalData, coarse)) #and x.Symbol.Value=='WNC' return [ x.Symbol for x in filteredByVolume ] # liquidate and rerank everything to see # if each month actual orders are being placed # self.Liquidate() # Filter the values of the dict: we only want up-trending securities from ones which has indicator values def FineSelectionFunction(self, fine): self.Debug('fine universe selection called: ') #return portfolio unchanged if not rebalanceTime if self.Time < self.rebalanceTime: return Universe.Unchanged filteredByMarketCap = list(filter(lambda x: 300000000 < x.MarketCap < 50000000000, fine)) #check if atleast last 1.5 year old ipo filteredByFunda = list(filter(lambda x: x.SecurityReference.IPODate+timedelta(550)<self.Time , filteredByMarketCap)) #filteredByMarketCap = list(filter(lambda x: 300000000 < x.MarketCap < 2000000000 , fine)) #take top n stop for analysis sortedByMarketCap = sorted(filteredByFunda, key=lambda x: x.MarketCap, reverse=True)[:2000] # will keep the indicators for all mid caps self.Debug(str(self.Time)+' Market cap length'+ str(len(sortedByMarketCap))) #self.universeSelected = True return [ x.Symbol for x in sortedByMarketCap ] # Share the same rebalance function for Universe and PCM for clarity def shallRebalance(self, time): self.Debug(str(time)+'*********Rebalncing called*********') if time.month == self.month or time.month not in [1, 4, 7, 10]: return None self.month = time.month return time class MyCustomWeightingPortfolioConstructionModel(PortfolioConstructionModel): def __init__(self,rebalance=None): #super().__init__(rebalance) #this function is for rebalancing #tried calling but not workinh # if rebalance is not None: # self.SetRebalancingFunc(rebalance) self.month=0 def CreateTargets(self, algorithm, insights): self.percent=0.1 targets=[] capacity=10-len([x.Key for x in algorithm.Portfolio if x.Value.Invested]) #added check here to prevent unnecessary loops if capacity>0: for insight in insights: #added check here to prevent more addition when current target exhaust capacity if capacity>0: #****IMP this does not work with normal cash/leverage=1 account. only works in margin account. dont know why!*** targets.append(PortfolioTarget.Percent(algorithm,insight.Symbol, 0.1)) capacity=capacity-1 return targets # Determines if the portfolio should be rebalanced base on the provided rebalancing func # def IsRebalanceDue(self, insights, algorithmUtc): # if self.rebalance is not None: # return self.rebalance(algorithmUtc) # return True def OnSecuritiesChanged(self, algorithm, changes): pass class TrailingStopRiskWithMinProfitManagementModel(RiskManagementModel): '''Provides an implementation of IRiskManagementModel that limits the maximum possible loss measured from the highest unrealized profit, also it will be triggered once profit level is crossed''' def __init__(self, maximumDrawdownPercent = 0.05, minProfitBeforeTrail=0.1): '''Initializes a new instance of the TrailingStopRiskManagementModel class Args: maximumDrawdownPercent: The maximum percentage drawdown allowed for algorithm portfolio compared with the highest unrealized profit, defaults to 5% drawdown''' self.maximumDrawdownPercent = abs(maximumDrawdownPercent) self.minProfitBeforeTrail = abs(minProfitBeforeTrail) self.trailingAbsoluteHoldingsState = dict() def ManageRisk(self, algorithm, targets): '''Manages the algorithm's risk at each time step Args: algorithm: The algorithm instance targets: The current portfolio targets to be assessed for risk''' riskAdjustedTargets = list() for kvp in algorithm.Securities: symbol = kvp.Key security = kvp.Value # Remove if not invested if not security.Invested: self.trailingAbsoluteHoldingsState.pop(symbol, None) continue if symbol.Value=='CRUS': algorithm.Debug('here') position = PositionSide.Long if security.Holdings.IsLong else PositionSide.Short absoluteHoldingsValue = security.Holdings.AbsoluteHoldingsValue trailingAbsoluteHoldingsState = self.trailingAbsoluteHoldingsState.get(symbol) unrealizedProfitPerc=security.Holdings.UnrealizedProfitPercent # Add newly invested security (if doesn't exist) or reset holdings state (if position changed) if trailingAbsoluteHoldingsState == None or position != trailingAbsoluteHoldingsState.position: self.trailingAbsoluteHoldingsState[symbol] = trailingAbsoluteHoldingsState = self.HoldingsState(position, security.Holdings.AbsoluteHoldingsCost) trailingAbsoluteHoldingsValue = trailingAbsoluteHoldingsState.absoluteHoldingsValue minProfitTrailPointCrossed = trailingAbsoluteHoldingsState.minProfitTrailPointCrossed #dont update trailing point once u noted initial entry point till min profit level is reached #BUT you need to check if it has crossed min profit threshold already or not as after say u in 22% profit ur stoploss should be 14%(8% down), so here if u dont check it, below condition will never set trailing stop when u in profit if(unrealizedProfitPerc> 0.0 and unrealizedProfitPerc<self.minProfitBeforeTrail and minProfitTrailPointCrossed == False): continue # first cross above min profit trailing point observed, now let your algorithm trail from this point if(unrealizedProfitPerc> 0.0 and unrealizedProfitPerc>=self.minProfitBeforeTrail and minProfitTrailPointCrossed == False): self.trailingAbsoluteHoldingsState[symbol].minProfitTrailPointCrossed = True # Check for new max (for long position) or min (for short position) absolute holdings value if ((position == PositionSide.Long and trailingAbsoluteHoldingsValue < absoluteHoldingsValue) or (position == PositionSide.Short and trailingAbsoluteHoldingsValue > absoluteHoldingsValue)): self.trailingAbsoluteHoldingsState[symbol].absoluteHoldingsValue = absoluteHoldingsValue continue drawdown = abs((trailingAbsoluteHoldingsValue - absoluteHoldingsValue) / trailingAbsoluteHoldingsValue) if self.maximumDrawdownPercent < drawdown: # liquidate riskAdjustedTargets.append(PortfolioTarget(symbol, 0)) return riskAdjustedTargets class HoldingsState: def __init__(self, position, absoluteHoldingsValue): self.position = position self.absoluteHoldingsValue = absoluteHoldingsValue self.minProfitTrailPointCrossed = False

#region imports from AlgorithmImports import * from collections import deque import functools as ft #endregion # Your New Python File #you have access to fundamental data in alphamodel #quareterly check added in both files due to separation of concerns methodology class MyAlphaModel(AlphaModel): def __init__(self): self.averages = { } #reshuffle quarterly self.rebalanceTime = datetime.min self.qb = QuantBook() self.errorMap={} self.errorMap['count']=0 self.errorMap['symbols']=[] #every time there is new info for alpha model which creates insights which it passes to portfolio contruction model #you have access to fundamental data in alphamodel def Update(self,algorithm,slice): '''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: New insights''' #Expiry.EndOfQuarter(algorithm.Time) #self.Debug(str(self.Time)+' On data called') #here by Bars, u get only when there is price data for security, # cause OnData is called for dividend etc. types things also insights = [] keys = slice.Bars.Keys for key in keys: if key in self.averages: stock_price = slice[key].Price volume=slice[key].Volume time=slice[key].Time asset = self.averages[key] # warm up stock's indicators if not ready if not asset.areIndsReady(): # for slow ma as we need to check 52 min below 52 low 200ma, we need 200ma avaiable on 52 week before date, so for that we need 253 more data points history = algorithm.History(asset.symbol, 506) for index, row in history.loc[asset.symbol].iterrows(): asset.slow.Update(index, row["close"]) asset.ma200FiftyTwoLow.Update(index, asset.slow.Current.Value) history = algorithm.History(asset.symbol, 253) for index, row in history.loc[asset.symbol].iterrows(): asset.updateInds(index, row["close"]) tradeBar = TradeBar(index, asset.symbol, row.open, row.high, row.low, row.close, row.volume, timedelta(1)) asset.updateBarInds(tradeBar) asset.updateVolInds(index, row.volume) asset.todayVolume=row.volume #update close window asset._closeWindow.Add(row["close"]) # now register indicators for automatic daily updates asset.registerIndsForUpdates(algorithm) vcpHistory = algorithm.History(asset.symbol, 50) for index, row in vcpHistory.loc[asset.symbol].iterrows(): asset.checkVCP(row["close"],index) #manually updating the indicator till found better way asset.ma200FiftyTwoLow.Update(time, asset.slow.Current.Value) asset.updateForSelection(time, slice[key],algorithm) # if(slice[key].Symbol.Value=="ACAD" and asset.areIndsReady()): # algorithm.Plot("Debug",'Price', stock_price) # algorithm.Plot("Debug",'slow', asset.slow.Current.Value) # algorithm.Plot("Debug",'slow52Low', asset.ma200FiftyTwoLow.Current.Value) # check if portfolio is full, otherwise invest if algorithm.Time < self.rebalanceTime: #******* commented this 'and not x.isHugeGapUp' values = list(filter(lambda x: x.lowBelow200Low and x.is_uptrend and x.vcpFormed and x.todayVolume>x.avgVolume.Current.Value*2 , self.averages.values())) # Sorts the values of the dict: we want those with greater difference between the moving averages # candidates=sorted(values,key=lambda x: (x.downFromHigh, -x.upFromLow))[:10] for value in values: if not value.areFundaReady(self.qb) and value.symbol.Value not in self.errorMap['symbols']: value.populateInitialFundas(algorithm.Time,self.qb,self.errorMap) algorithm.Debug('errorMap Count: '+str(self.errorMap['count'])) valuesFilterdOnFunda=list(filter(lambda x: x.areFundaSuitable(self), values)) candidates=sorted(valuesFilterdOnFunda,key=lambda x: (x.upFromLow,x.volumeMaDict['yestMa']/x.volumeMaDict['maAtTrend']))[:10] for candidate in candidates: #if candidate not in porfolio already, BUT SHUD THIS BE IN PORTFOLIO CONSTRUCTION? if not algorithm.Portfolio[candidate.symbol].Invested: #algorithm.SetHoldings(candidate.symbol, 0.1) #Insight.Price(candidate.symbol, self.rebalanceTime, InsightDirection.Up, None, None, None, 0.1) insights.append(Insight.Price(candidate.symbol, self.rebalanceTime, InsightDirection.Up, None, None, None, 0.1)) #month changed, liquidate everything and reenter in top 10 else: self.rebalanceTime = Expiry.EndOfQuarter(algorithm.Time) #reset error map in case next time proper data comes for error map stocks self.errorMap['count']=0 self.errorMap['symbols']=[] # month changed, liquidate portfolio #algorithm.Liquidate() #dint use it as alpha Insight has expiry added #you can create sort of weighted score of securities by index of it in sorted eps.revenue,margin growth arrays #like 4+2+1=7 return insights def OnSecuritiesChanged(self, algorithm, changes): algorithm.Debug('On security changed called') # liquidate removed securities for security in changes.RemovedSecurities: if security.Symbol in self.averages: self.averages.pop(security.Symbol) # we want 20% allocation in each security in our universe for security in changes.AddedSecurities: self.averages[security.Symbol] = SymbolData(security.Symbol) #self.SetHoldings(security.Symbol, 0.1) class SymbolData(object): def __init__(self, symbol): #self.ma200FiftyTwoLowChained = IndicatorExtensions.MIN(self.slow, 200) #self.ma200FiftyTwoLowChained = IndicatorExtensions.Of(Minimum(253), self.slow) self._closeWindow = RollingWindow[float](2) self.length = 14 self.adrPerc = NormalizedAverageTrueRange(self.length) self.adrPercThreshold = 1.0 self.avgVolume = SimpleMovingAverage(20) self.colState="up" self.downFromHigh=1 self.downTrend=0.0 self.fast = ExponentialMovingAverage(50) self.fiftyTwoHigh = Maximum(253) self.fiftyTwoLow = Minimum(253) self.is_uptrend = False self.isHugeGapUp=False self.lowBelow200Low = False self.ma200FiftyTwoLow = Minimum(253) self.molt = 1 self.naturalRally = 0.0 self.naturalReaction = 0.0 self.pivot1=0.0 #high prie pivot self.pivot2=0.0 #low price pivot self.scale = 0 self.secondaryRally = 0.0 self.secondaryReaction = 0.0 self.slow = ExponentialMovingAverage(200) self.slow = ExponentialMovingAverage(200) self.symbol = symbol self.todayVolume = 0.1 self.tolerance = 1.05 self.trend="up" self.upFromLow=0 self.upTrend = 0.0 self.vcpFormed=False #this will hold last 7 quarters funda data of 3Ds self.threeDs = deque(maxlen=7) #this map will hold growt of 3Ds self.threeDGrowthMap = {} self.threeDGrowthMap['net_income']=RollingWindow[float](3) self.threeDGrowthMap['revenue']=RollingWindow[float](3) self.threeDGrowthMap['margin']=RollingWindow[float](3) self.threeQoQDGrowthMap = {} self.threeQoQDGrowthMap['net_income']=RollingWindow[float](3) self.threeQoQDGrowthMap['revenue']=RollingWindow[float](3) self.threeQoQDGrowthMap['margin']=RollingWindow[float](3) self.volumeMaDict={ 'maAtTrend':0.0, 'yestMa':0.0 } def areFundaReady(self,qb): indsReady=True indsReady=self.threeDGrowthMap['net_income'].IsReady and indsReady indsReady=self.threeDGrowthMap['revenue'].IsReady and indsReady indsReady=self.threeDGrowthMap['margin'].IsReady and indsReady return indsReady #endTime is to prevent lookahead bias and see only past data from algo running time def populateInitialFundas(self,endTime,qb,errorMap): net_income_history = qb.GetFundamental(self.symbol, "FinancialStatements.IncomeStatement.NetIncomeContinuousOperations.ThreeMonths", endTime.date() - timedelta(days=650), endTime) revenue_history = qb.GetFundamental(self.symbol, "FinancialStatements.IncomeStatement.OperatingRevenue.ThreeMonths", endTime.date() - timedelta(days=650), endTime) ebit_history = qb.GetFundamental(self.symbol, "FinancialStatements.IncomeStatement.EBIT.ThreeMonths", endTime.date() - timedelta(days=650), endTime) net_income_history_filtered = net_income_history.loc[net_income_history.shift().values[:,0]!=net_income_history.values[:,0]] revenue_history_filtered = revenue_history.loc[revenue_history.shift().values[:,0]!=revenue_history.values[:,0]] ebit_history_filtered= ebit_history.loc[ebit_history.shift().values[:,0]!=ebit_history.values[:,0]] #margin_history_filtered = margin_history.drop_duplicates() net_income_history_filtered_sortred = net_income_history_filtered.sort_index(ascending=False).head(7) revenue_history_filtered_sortred = revenue_history_filtered.sort_index(ascending=False).head(7) ebit_history_filtered_sortred = ebit_history_filtered.sort_index(ascending=False).head(7) #margin_history_filtered_sortred = margin_history_filtered.sort_index(ascending=False).head(5) margin_history_filtered_sortred = (ebit_history_filtered_sortred / revenue_history_filtered_sortred) dfs = [net_income_history_filtered_sortred, revenue_history_filtered_sortred, margin_history_filtered_sortred] final_df = ft.reduce(lambda left, right: pd.merge(left, right, left_index=True,right_index=True), dfs) #history_filtered.sort_index(ascending=False).index[0]).date() #history_filtered_sortred.iloc[0,column_index] #history_filtered_sortred for index, row in final_df.iterrows(): self.threeDs.append(threeDValue(index,row[0],row[1],row[2])) if len(self.threeDs)<7 : errorMap['count']=errorMap['count']+1 errorMap['symbols'].append(self.symbol.Value) return #update QoQ data counter=3 while counter>0: self.threeQoQDGrowthMap['net_income'].Add((self.threeDs[counter-1].net_income - self.threeDs[counter].net_income)/abs(self.threeDs[counter].net_income)) self.threeQoQDGrowthMap['revenue'].Add((self.threeDs[counter-1].revenue - self.threeDs[counter].revenue)/abs(self.threeDs[counter].revenue)) self.threeQoQDGrowthMap['margin'].Add((self.threeDs[counter-1].margin - self.threeDs[counter].margin)/abs(self.threeDs[counter].margin)) counter=counter-1 #update YoY data counter=2 while counter>=0: self.threeDGrowthMap['net_income'].Add((self.threeDs[counter].net_income - self.threeDs[counter+4].net_income)/abs(self.threeDs[counter+4].net_income)) self.threeDGrowthMap['revenue'].Add((self.threeDs[counter].revenue - self.threeDs[counter+4].revenue)/abs(self.threeDs[counter+4].revenue)) self.threeDGrowthMap['margin'].Add((self.threeDs[counter].margin - self.threeDs[counter+4].margin)/abs(self.threeDs[counter+4].margin)) counter=counter-1 def areFundaSuitable(self,context): compareThreshold=0.7 fundaSuitable=True #0 position is latest in rollingwindow counter=0 #if funda not available, then return not suitable funda if self.symbol.Value in context.errorMap['symbols']: fundaSuitable=False return fundaSuitable while counter<2: #min YoY growth threshold if(self.threeDGrowthMap['net_income'][counter]<0.2 or self.threeDGrowthMap['revenue'][counter]<0.1): fundaSuitable=False return fundaSuitable #rising YoY growth if(self.threeDGrowthMap['net_income'][counter]<compareThreshold*self.threeDGrowthMap['net_income'][counter+1]): fundaSuitable=False return fundaSuitable if(self.threeDGrowthMap['revenue'][counter]<compareThreshold*self.threeDGrowthMap['revenue'][counter+1]): fundaSuitable=False return fundaSuitable if(self.threeDGrowthMap['margin'][counter]<compareThreshold*self.threeDGrowthMap['margin'][counter+1]): fundaSuitable=False return fundaSuitable #rising QoQ growth if(self.threeQoQDGrowthMap['net_income'][counter]<compareThreshold*self.threeQoQDGrowthMap['net_income'][counter+1]): fundaSuitable=False return fundaSuitable if(self.threeQoQDGrowthMap['revenue'][counter]<compareThreshold*self.threeQoQDGrowthMap['revenue'][counter+1]): fundaSuitable=False return fundaSuitable if(self.threeQoQDGrowthMap['margin'][counter]<compareThreshold*self.threeQoQDGrowthMap['margin'][counter+1]): fundaSuitable=False return fundaSuitable counter=counter+1 return fundaSuitable def areIndsReady(self): indsReady=True indsReady=self.fiftyTwoHigh.IsReady and indsReady indsReady=self.fiftyTwoLow.IsReady and indsReady indsReady=self.ma200FiftyTwoLow.IsReady and indsReady indsReady=self.fast.IsReady and indsReady indsReady=self.slow.IsReady and indsReady indsReady=self.avgVolume.IsReady and indsReady indsReady=self._closeWindow.IsReady and indsReady return indsReady def updateForSelection(self, time, currentSlice,context): self.resetVariables() stockPrice=currentSlice.Price #update volume self.todayVolume=currentSlice.Volume #add current Close to rolling window to access it tomorrow self._closeWindow.Add(currentSlice.Close) if self.areIndsReady(): fast = self.fast.Current.Value slow = self.slow.Current.Value self.downFromHigh=1-(stockPrice/self.fiftyTwoHigh.Current.Value) self.upFromLow=(stockPrice/self.fiftyTwoLow.Current.Value)-1 if(self.symbol.Value == 'DDD'): context.Debug('here') #keep updating pivots daily as even when down 200ma, u need the pivots later man self.checkVCP(stockPrice,time) if stockPrice>slow: self.is_uptrend = stockPrice>slow and stockPrice>fast and fast > slow * self.tolerance and self.upFromLow>=0.8 and self.downFromHigh<=0.2 #we want to check today open with yesterday close to see huge gap up opening if((currentSlice.Open-self._closeWindow[1])/self._closeWindow[1])>=0.10: #(self.adrPerc.Current.Value/100): self.isHugeGapUp=True if self.is_uptrend: #self.checkVCP(stockPrice) # if in uptrend and vcp then only enter self.scale = (fast - slow) / ((fast + slow) / 2.0) if self.fiftyTwoLow.Current.Value < self.ma200FiftyTwoLow.Current.Value: self.lowBelow200Low=True def updateInds(self, time, value): self.fiftyTwoHigh.Update(time, value) self.fiftyTwoLow.Update(time, value) self.fast.Update(time, value) #moved slow ind update out as we need it while checking the close below 52 week prior 200ma #self.slow.Update(time, value) #see if you really need it #self.ma200FiftyTwoLow.Update(time, self.slow.Current.Value) # if(self.symbol.Value=="ACAD" and self.areIndsReady()): # pass def updateBarInds(self,barData): self.adrPerc.Update(barData) def updateVolInds(self,time,volume): #update yesterday volume ma with current volume ma value first self.volumeMaDict['yestMa']=self.avgVolume.Current.Value self.avgVolume.Update(time,volume) def registerIndsForUpdates(self,context): context.RegisterIndicator(self.symbol,self.fiftyTwoHigh,Resolution.Daily) context.RegisterIndicator(self.symbol,self.fiftyTwoLow,Resolution.Daily) context.RegisterIndicator(self.symbol,self.fast,Resolution.Daily) context.RegisterIndicator(self.symbol,self.slow,Resolution.Daily) context.RegisterIndicator(self.symbol,self.adrPerc,Resolution.Daily) context.RegisterIndicator(self.symbol,self.avgVolume,Resolution.Daily,Field.Volume) #context.RegisterIndicator(self.symbol, self.ma200FiftyTwoLowChained, Resolution.Daily) #this method is used to reset all signal variables so that in current iteration we get proper true signal and not prior stale one def resetVariables(self): self.vcpFormed=False self.is_uptrend=False self.scale=0.0 self.isHugeGapUp=False self.lowBelow200Low=False def checkVCP(self,close,time): # self.vcpFormed=True # return True # make default as false self.vcpFormed=False if(self.trend=="down"): # add your logic here for downtrend noting if(self.pivot1==0.0): #no upward pivot set yet self.upTrend= close+close*0.05 #stoploss/up trend marker self.pivot1=self.upTrend self.vcpFormed=False #in pivot setup make vcp false if (close>self.upTrend): #rose above downtrend so downtrend over, mark upTrend instead of natural reaction self.upTrend=close self.trend="up" self.colState="up" #strategy.entry("startBuy",strategy.long) //*** entered long here in scrip self.vcpFormed=True #as we want to go long on stock which broke above downtrend on rising 200 else: if(close>=(self.pivot1+self.pivot1*0.02)): #broke above natural rally pivot, mark uptrend, enter again self.trend="up" # change trend self.colState="up" #change note down column self.upTrend=close #note down price as upTrend #strategy.entry("startBuy",strategy.long, when=strategy.position_size <= 0) //start buying *** entered long here in scrip self.vcpFormed=True #as we want to go long on stock which broke above pivot 1. this u can remove later if u want to enter on 3-C types formation if(close<self.downTrend): #price less than downTrend IS ALWAYS DOWNTREND if(self.trend=="down"): #more downtrend self.downTrend=close self.vcpFormed=False #downtrend continues #note down volume to keep track of volume drying when vcp happens self.volumeMaDict['maAtTrend']=self.avgVolume.Current.Value if(self.colState=="naturalRally"): #direct downtrend from natural rally instead of natural reaction self.downTrend=close self.pivot1=self.naturalRally self.vcpFormed=False #direct downtrend from natural rally, #so vol expanded and dont enter #note down volume to keep track of volume drying when vcp happens self.volumeMaDict['maAtTrend']=self.avgVolume.Current.Value self.pivot2=self.downTrend self.colState="down" else: if(self.colState=="down"): # if noting in downtrend col, price is more than downtrend, check for SIGNIFICANT rally if(((close-self.downTrend)/self.downTrend)>=((self.adrPerc.Current.Value*self.adrPercThreshold)/100)): # first significant rally from downtrend, mark downtrend as lower pivot(pivot2), note colState as naturalRally self.pivot2=self.downTrend self.naturalRally=close self.colState="naturalRally" self.vcpFormed=False #stock going through natural price cycle still if(self.colState=="naturalRally"): if(close>self.naturalRally): self.naturalRally=close #keep updating natural rally with higher values self.vcpFormed=False #stock going through natural price cycle still else: # close less than natural rally if(((self.naturalRally-close)/self.naturalRally)>=((self.adrPerc.Current.Value*self.adrPercThreshold)/100)): # first significant reaction from natural rally, mark natural rally as higher pivot(pivot1), note colState as naturalReaction self.colState="naturalReaction" self.pivot1=self.naturalRally self.vcpFormed=False #stock going through natural price cycle still #self.vcpFormed=True if(self.trend=="up"): if(self.pivot2 == 0.0): #no downward pivot set yet self.downTrend= self.slow.Current.Value-self.slow.Current.Value*0.05 #stoploss self.pivot2=self.downTrend self.vcpFormed=False #in pivot setup make vcp false if(close>self.slow.Current.Value and self.upTrend == 0): #if uptrend is not initialized first time self.trend="up" self.upTrend=close self.pivot1=self.upTrend self.colState="up" self.vcpFormed=False #in pivot setup make vcp false #note down volume to keep track of volume drying when vcp happens self.volumeMaDict['maAtTrend']=self.avgVolume.Current.Value if (close<self.downTrend): #uptrend over, exit strategy, mark downTrend instead of natural reaction self.downTrend=close self.trend="down" self.colState="down" self.vcpFormed=False #broke pivot so vcp false #*****in script closed strategy here**** if(close<=(self.pivot2-self.pivot2*0.02)): # broke natural reaction, mark downtrend self.trend="down" self.downTrend=close self.colState="down" self.vcpFormed=False #broke pivot so vcp false #*****in script closed strategy here**** if(close>self.upTrend): if(self.trend=="up" and self.upTrend != 0):#uptrend is set and prices rising self.upTrend=close self.vcpFormed=False #in constant uptrending and no vcp formed yet #keep latest uptrend as starting point for volume drying up logic self.volumeMaDict['maAtTrend']=self.avgVolume.Current.Value if(self.colState=="naturalReaction"): #direct uptrend from natural reaction instead of natural rally self.upTrend=close self.pivot2=self.naturalReaction self.vcpFormed=True #direct uptrend from natural reaction, #so broke pivot and eligible for entry self.pivot1=self.upTrend #???????READ THIS????? here u assigning pivot when prices rising, shudnt u be assigning it when first reaction from uptrend? if(self.colState=="naturalRally"): #you added it later, broke above natural rally, means from reaction came to natural rally and now uptrend self.vcpFormed=True self.colState="up" if(close<self.upTrend and self.trend=="up"): # only when in uptrend if(self.colState == "up" and ((self.upTrend-close)/(close))>=((self.adrPerc.Current.Value*self.adrPercThreshold)/100)): #first reaction from uptrend self.naturalReaction=close self.colState="naturalReaction" #enter into natural reaction colState self.vcpFormed=False #in normal price behavior, wait for vcp now if(close<self.naturalReaction and self.colState=="naturalReaction"): self.naturalReaction=close #keep updating natural reaction with lower values self.vcpFormed=False #in normal price behavior, wait for vcp now if(self.colState=="naturalReaction" and ((close-self.naturalReaction)/(self.naturalReaction))>=((self.adrPerc.Current.Value*self.adrPercThreshold)/100)): #natural rally happened, mark natural reaction as pivot 2 self.colState="naturalRally" self.naturalRally=close self.pivot2=self.naturalReaction self.vcpFormed=False #natural rally is normal price behavior, wait for vcp now # if(self.colState=="naturalRally" and false): #if were noting in natural rally, # if(((abs(upTrend-high))/(upTrend))<=0.01) // stock high was just short of uptrend, # if(((high-close)/(close))>=((adrPerc-1)/2))//but stock reacts heavily from uptrend point, danger signal, sell # strategy.close("startBuy",when=time_cond) //exit here class threeDValue(object): def __init__(self,date,net_income,revenue,margin): self.date = date self.net_income=net_income self.revenue=revenue self.margin=margin

# 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 * from System.Collections.Generic import List ### <summary> ### In this algorithm we demonstrate how to perform some technical analysis as ### part of your coarse fundamental universe selection ### </summary> ### <meta name="tag" content="using data" /> ### <meta name="tag" content="indicators" /> ### <meta name="tag" content="universes" /> ### <meta name="tag" content="coarse universes" /> class EmaCrossUniverseSelectionAlgorithm(QCAlgorithm): def Initialize(self): '''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.''' self.SetStartDate(2015,1,1) #Set Start Date self.SetEndDate(2019,1,1) #Set End Date self.SetCash(100000) #Set Strategy Cash self.UniverseSettings.Resolution = Resolution.Daily self.UniverseSettings.Leverage = 2 self.coarse_count = 10 self.averages = { } # this add universe method accepts two parameters: # - coarse selection function: accepts an IEnumerable<CoarseFundamental> and returns an IEnumerable<Symbol> self.AddUniverse(self.CoarseSelectionFunction) # sort the data by daily dollar volume and take the top 'NumberOfSymbols' def CoarseSelectionFunction(self, coarse): # We are going to use a dictionary to refer the object that will keep the moving averages for cf in coarse: if cf.Symbol not in self.averages: self.averages[cf.Symbol] = SymbolData(cf.Symbol) # Updates the SymbolData object with current EOD price avg = self.averages[cf.Symbol] avg.update(cf.EndTime, cf.AdjustedPrice) # Filter the values of the dict: we only want up-trending securities values = list(filter(lambda x: x.is_uptrend, self.averages.values())) # Sorts the values of the dict: we want those with greater difference between the moving averages values.sort(key=lambda x: x.scale, reverse=True) for x in values[:self.coarse_count]: self.Log('symbol: ' + str(x.symbol.Value) + ' scale: ' + str(x.scale)) # we need to return only the symbol objects return [ x.symbol for x in values[:self.coarse_count] ] # this event fires whenever we have changes to our universe def OnSecuritiesChanged(self, changes): # liquidate removed securities for security in changes.RemovedSecurities: if security.Invested: self.Liquidate(security.Symbol) # we want 20% allocation in each security in our universe for security in changes.AddedSecurities: self.SetHoldings(security.Symbol, 0.1) class SymbolData(object): def __init__(self, symbol): self.symbol = symbol self.tolerance = 1.01 self.fast = ExponentialMovingAverage(50) self.slow = ExponentialMovingAverage(200) self.is_uptrend = False self.scale = 0 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)

Platform

Radically Open-Source Algorithmic Trading Engine

Join Our Discord Channel

Alpha League Competition: $1,000 Weekly Prize Pool

Draft Discussions

Bookmarked Discussions

SEARCH DISCUSSIONS

TOP 5 Research PUblications

283,800 Quants.

VOTE FOR UPCOMING FEATURES

Model Volatility Contraction/Cup and Handle patterns

IN THIS RESEARCH

PARTICIPANTS

Discussion Awards

Actions

Join QuantConnect for Free

Platform

SIGN IN

Radically Open-Source Algorithmic Trading Engine

Join Our Discord Channel

Alpha League Competition: $1,000 Weekly Prize Pool

Draft Discussions

Bookmarked Discussions

SEARCH DISCUSSIONS

TOP 5 Research PUblications

283,800 Quants.

VOTE FOR UPCOMING FEATURES

Model Volatility Contraction/Cup and Handle patterns

IN THIS RESEARCH

PARTICIPANTS

Discussion Awards

SHARE RESEARCH

SHARE DISCUSSION

SHARE ARTICLE

Actions

Join QuantConnect for Free