| Overall Statistics |
|
Total Trades 541 Average Win 2.11% Average Loss -1.06% Compounding Annual Return 3.388% Drawdown 27.000% Expectancy 0.049 Net Profit 22.489% Sharpe Ratio 0.237 Probabilistic Sharpe Ratio 1.179% Loss Rate 65% Win Rate 35% Profit-Loss Ratio 2.00 Alpha -0.027 Beta 0.679 Annual Standard Deviation 0.142 Annual Variance 0.02 Information Ratio -0.44 Tracking Error 0.126 Treynor Ratio 0.05 Total Fees $2037.89 Estimated Strategy Capacity $2400000.00 Lowest Capacity Asset MODN VF1TI9X213Z9 |
#region imports
from AlgorithmImports import *
#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
#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:
# if not invested, then only try to invest, otherwise let portfolio run till monnth change
# for x in algorithm.Portfolio:
# algorithm.Debug(x)
# capacity=10-len([x.Key for x in algorithm.Portfolio if x.Value.Invested])
# if capacity>0:
# Filter the values of the dict: we only want up-trending securities from ones which has indicator values
values = list(filter(lambda x: x.lowBelow200Low and x.is_uptrend and x.vcpFormed and x.todayVolume>x.avgVolume.Current.Value*2 and not x.isHugeGapUp, 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]
candidates=sorted(values,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)
# 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.symbol = symbol
self.isHugeGapUp=False
self._closeWindow = RollingWindow[float](2)
self.todayVolume = 0.1
self.tolerance = 1.05
self.fast = ExponentialMovingAverage(50)
self.avgVolume = SimpleMovingAverage(20)
self.slow = ExponentialMovingAverage(200)
self.fiftyTwoHigh = Maximum(253)
self.fiftyTwoLow = Minimum(253)
self.ma200FiftyTwoLow = Minimum(253)
#self.ma200FiftyTwoLowChained = IndicatorExtensions.Of(Minimum(253), self.slow)
#self.ma200FiftyTwoLowChained = IndicatorExtensions.MIN(self.slow, 200)
self.downFromHigh=1
self.upFromLow=0
self.is_uptrend = False
self.lowBelow200Low = False
self.scale = 0
self.secondaryRally = 0.0
self.naturalRally = 0.0
self.pivot1=0.0 #high prie pivot
self.pivot2=0.0 #low price pivot
self.upTrend = 0.0
self.naturalReaction = 0.0
self.secondaryReaction = 0.0
self.trend="up"
self.colState="up"
self.downTrend=0.0
self.length = 14
self.molt = 1
self.slow = ExponentialMovingAverage(200)
self.adrPerc = NormalizedAverageTrueRange(self.length)
self.adrPercThreshold = 1.0
self.vcpFormed=False
self.volumeMaDict={
'maAtTrend':0.0,
'yestMa':0.0
}
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.12: #(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 #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(2013,12,1) #Set Start Date
self.SetEndDate(2020,1,1) #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.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>=30: #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 >= 500000 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
# Filter for market cap less than large caps
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 < 10000000000 , fine))
#filteredByMarketCap = list(filter(lambda x: 300000000 < x.MarketCap < 2000000000 , fine))
#take top n stop for analysis
sortedByMarketCap = sorted(filteredByMarketCap, 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 # 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)