| Overall Statistics |
|
Total Trades 80 Average Win 2.17% Average Loss -0.94% Compounding Annual Return 16.913% Drawdown 7.600% Expectancy 0.252 Net Profit 13.897% Sharpe Ratio 0.86 Probabilistic Sharpe Ratio 41.877% Loss Rate 62% Win Rate 38% Profit-Loss Ratio 2.31 Alpha 0.082 Beta 0.493 Annual Standard Deviation 0.147 Annual Variance 0.022 Information Ratio 0.246 Tracking Error 0.148 Treynor Ratio 0.256 Total Fees $260.68 Estimated Strategy Capacity $1400000.00 Lowest Capacity Asset SWCH WOHB1KIF49K5 |
#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
#this contains bar date for which the data is passed on
time=slice[key].Time
asset = self.averages[key]
if asset.symbol.Value=='AXDX':
algorithm.Debug('here')
# warm up stock's indicators if not ready
if not asset.areIndsReady():
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
# 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)
if asset.symbol.Value=='AXDX':
algorithm.Debug('here')
asset.updateForSelection(time, slice[key],algorithm)
# self.Plot("Debug",'P1', asset.pivot1)
# self.Plot("Debug",'Price', stock_price)
# self.Plot("Debug",'P2', asset.pivot2)
# self.Plot("Debug",'ema200', asset.slow.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.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]
candidates=sorted(values,key=lambda x: (x.upFromLow,x.volumeMaDict['todayMA']/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.todayVolume = 0.1
self.yesterdayClose = 0.0
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.downFromHigh=1
self.upFromLow=0
self.is_uptrend = 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.vcpFormed=False
self.volumeMaDict={
'maAtTrend':0.0,
'todayMA':0.0
}
def areIndsReady(self):
indsReady=True
indsReady=self.fiftyTwoHigh.IsReady and indsReady
indsReady=self.fiftyTwoLow.IsReady and indsReady
indsReady=self.fast.IsReady and indsReady
indsReady=self.slow.IsReady and indsReady
indsReady=self.avgVolume.IsReady and indsReady
return indsReady
def updateForSelection(self, time, currentSlice,context):
self.resetVariables()
stockPrice=currentSlice.Price
#update volume
self.todayVolume=currentSlice.Volume
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
if self.is_uptrend:
#self.checkVCP(stockPrice) # if in uptrend and vcp then only enter
self.scale = (fast - slow) / ((fast + slow) / 2.0)
def updateInds(self, time, value):
self.fiftyTwoHigh.Update(time, value)
self.fiftyTwoLow.Update(time, value)
self.fast.Update(time, value)
self.slow.Update(time, value)
def updateBarInds(self,barData):
self.adrPerc.Update(barData)
def updateVolInds(self,time,volume):
# update today volume ma with current volume ma value first, as quantconnect runs daily midnight, latest candle is of today. need some logic for yest candle
self.volumeMaDict['todayMA']=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)
#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
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/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/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/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/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(2018,12,1) #Set Start 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
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]:
if securityMap.Value.UnrealizedProfitPercent<percBenchmarkChg: #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))
#take top n stop for analysis
sortedByMarketCap = sorted(filteredByMarketCap, key=lambda x: x.MarketCap, reverse=True)[:3000]
# 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
#dont update trailing point once u noted initial entry point till min profit level is reached
if(unrealizedProfitPerc> 0.0 and unrealizedProfitPerc<self.minProfitBeforeTrail):
continue
# 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# 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)