Overall Statistics
Total Trades
22
Average Win
1.68%
Average Loss
-0.22%
Compounding Annual Return
6.546%
Drawdown
7.900%
Expectancy
4.275
Net Profit
9.130%
Sharpe Ratio
0.559
Probabilistic Sharpe Ratio
22.082%
Loss Rate
40%
Win Rate
60%
Profit-Loss Ratio
7.79
Alpha
0.048
Beta
0.003
Annual Standard Deviation
0.087
Annual Variance
0.008
Information Ratio
-0.523
Tracking Error
0.259
Treynor Ratio
19.305
Total Fees
$64.18
Estimated Strategy Capacity
$3300000.00
Lowest Capacity Asset
TSLA UNU3P8Y3WFAD
from Execution.ImmediateExecutionModel import ImmediateExecutionModel
from Portfolio.EqualWeightingPortfolioConstructionModel import EqualWeightingPortfolioConstructionModel
from Risk.MaximumDrawdownPercentPerSecurity import MaximumDrawdownPercentPerSecurity
import operator
from datetime import timedelta, time

class DeterminedBlueSheep(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2020, 1, 1)  # Set Start Date
        self.SetCash(100000)  # Set Strategy Cash
        # self.AddEquity("SPY", Resolution.Minute)
        
        ################################### PARAMETRI ######################################################
        #Lista di avvio
        self.initialList = ["AMD", "TSLA", "M", "MSFT"]
        
        self.entryZscore = 0 #percentuale di sicurezza rispetto alla deviazione standard del periodo (1 = 100%)
       
        self.myStdDays = 90 #periodo di calcolo della deviazione standard mobile
        
        self.fastAverage = 5 #finestra in giorni per la media mobile veloce
        
        self.slowAverage = 30 #finestra in giorni per la media mobile lenta
        
        self.concurrentEquities = 10 #al massimo quanti titoli gestire assieme
        
        self.insightsDaysDuration = 100 #per quanti giorni dura al massimo un insight
        ####################################################################################################
        
        self.UniverseSettings.Resolution = Resolution.Hour
        
        self.SetExecution(ImmediateExecutionModel())

        self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())

        self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.01))
        
        #dizionario con gli indicatori e dati consolidati per ogni simbolo
        self.symDict = { };
        
        #subscribe to every ticker on the list before starting the filter
        for ticker in self.initialList:
            self.AddEquity(ticker, Resolution.Hour).Symbol
            self.symDict[ticker] = SymbolData(self, ticker, self.entryZscore, self.myStdDays, self.fastAverage, self.slowAverage)
            
        # per ora lo scheduling e il filtro grosso/fine non possono funzionare assieme https://github.com/QuantConnect/Lean/issues/3890
        self.SetUniverseSelection(ScheduledUniverseSelectionModel(
            self.DateRules.EveryDay(),
            self.TimeRules.At(10,1),
            self.SelectSymbols
        ))
        
        #lista dei simboli da tradare
        self.symbols = []
        
        self.AddAlpha(SimpleGapAlphaModel(self)) #AlphaModel needs the dictionary
        
        TimeSpan.FromDays(max([self.myStdDays, self.fastAverage, self.slowAverage])+ 1)
        self.SetWarmUp(TimeSpan.FromDays(max([self.myStdDays, self.fastAverage, self.slowAverage])+ 1))
        
        
        
    # Scheduled Universe Construction    
    
    def SelectSymbols(self, dateTime):
        
        createdSymbols = []
        
        # How the Universe is changing based on the list
        
        for sym in self.initialList:
            
            if self.symDict[sym].gapIndicator != 0 and sym not in self.symbols:
                
                self.symbols.append(sym)
                
            elif self.symDict[sym].gapIndicator == 0 and sym in self.symbols:
                
                self.symbols.remove(sym)
                
        subset = {key: self.symDict[key] for key in self.symbols}
        
        # This should sort The dictionary by gapLevel, so we can use the strongest gaps
        
        self.symbols = [sym.symbol for sym in (sorted(subset.values(), key=operator.attrgetter('gapLevel')))]
        
        
        if len(self.symbols) > self.concurrentEquities:
            
            del self.symbols[self.concurrentEquities:]
        
        for sym in self.symbols:
            
            createdSymbols.append(Symbol.Create(sym, SecurityType.Equity, Market.USA))
        
        return createdSymbols
        


    # I need to get Real Time data for some indicators

    def OnData(self, data):
        '''OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.
            Arguments:
                data: Slice object keyed by symbol containing the stock data
        '''
        if not (data is not None and data.Time.time() <= time(10,0)):
            
            return
            
        for symbol, symbol_data in self.symDict.items():
            
            if data.ContainsKey(symbol) and data[symbol] is not None and symbol_data.IsReady:
                
                symbol_data.Update(data[symbol])
            
            
        #for ticker in self.initialList:
            
        #    if data is None or (not data.Bars.ContainsKey(ticker)) or (not self.symDict[ticker]._retStd.IsReady): pass
            
        #    else:
            
        #        myTradeBar = data.Bars[ticker]
                
                # Check if it Gapes Up or Down
        #        self.Debug(f"Assert:{myTradeBar.Open >= self.symDict[ticker].daily_bar.High or myTradeBar.Open <= self.symDict[ticker].daily_bar.Low} - {self.Time.time()} - {ticker} ready? {self.symDict[ticker]._retStd.Current.Value} - prev Hi:{self.symDict[ticker].daily_bar.High} prev Lo:{self.symDict[ticker].daily_bar.Low} Op bar:{myTradeBar.Open} ")
                
        #        if myTradeBar.Open >= self.symDict[ticker].daily_bar.High*(1+self.symDict[ticker].entryZscore*self.symDict[ticker]._retStd.Current.Value):
                    
        #            self.symDict[ticker].gapIndicator = 1
                    
        #            self.symDict[ticker].gapLevel = (myTradeBar.Open - self.symDict[ticker].daily_bar.High)/self.symDict[ticker].daily_bar.High
                    
        #        elif myTradeBar.Open <= self.symDict[ticker].daily_bar.Low*(1-self.symDict[ticker].entryZscore*self.symDict[ticker]._retStd.Current.Value):
                    
        #            self.symDict[ticker].gapIndicator = -1
                    
        #            self.symDict[ticker].gapLevel = (self.symDict[ticker].daily_bar.Low - myTradeBar.Open)/self.symDict[ticker].daily_bar.Low
                
                
                # Check if the gape trend is still on
                
        #        if self.symDict[ticker].gapIndicator > 0 and self.symDict[ticker].fast.Current.Value < self.symDict[ticker].slow.Current.Value:
                    
        #            self.symDict[ticker].gapIndicator = 0
                    
        #        elif self.symDict[ticker].gapIndicator < 0 and self.symDict[ticker].fast.Current.Value > self.symDict[ticker].slow.Current.Value:
                    
        #            self.symDict[ticker].gapIndicator = 0
        
        
        
        
class SymbolData(object):
    
    def __init__(self, algorithm, symbol, entryZscore, myStDays, fastAverage, slowAverage):
        
        self.algo = algorithm
        
        self.symbol = symbol 
        
        self.entryZscore = entryZscore #percentuale di sicurezza rispetto alla deviazione standard del periodo
       
        self.myStdDays = myStDays #periodo di calcolo della deviazione standard mobile
        
        self.fastAverage = fastAverage #finestra in giorni per la media mobile veloce
        
        self.slowAverage = slowAverage #finestra in giorni per la media mobile lenta
        
        self.gapIndicator = 0 #0 fine trend, 1 gappa su e uptrend, -1 gappa giù e downtrend
        
        self.gapLevel = 0
        
        self.daily_bar = None
        
        self.prev_bar = None
        
        self.daily_consolidator = TradeBarConsolidator(timedelta(days = 1))    ## 1 Day TradeBar Consolidator
        self.daily_consolidator.DataConsolidated += self.DailyConsolidator  ## Add fuction to do what you want every day with your data
        self.algo.SubscriptionManager.AddConsolidator(self.symbol, self.daily_consolidator)
        
        self._retStd = StandardDeviation(self.symbol, self.myStdDays) #Deviazione Standard sui ritorni
        self.algo.RegisterIndicator(self.symbol, self._retStd, Resolution.Daily)
        
        self.fast = ExponentialMovingAverage(self.fastAverage)
        self.algo.RegisterIndicator(self.symbol, self.fast, Resolution.Daily)
        
        self.slow = ExponentialMovingAverage(self.slowAverage)
        self.algo.RegisterIndicator(self.symbol, self.slow, Resolution.Daily)
        
        
    def DailyConsolidator(self, sender, bar):
        
        self.daily_bar = bar
        
        if self.prev_bar is not None:
            
            ret = (self.daily_bar.Close -  self.prev_bar.Close) / self.prev_bar.Close
            
            self._retStd.Update(self.algo.Time, ret)
            
            self.fast.Update(self.algo.Time, self.daily_bar.Close)
            
            self.slow.Update(self.algo.Time, self.daily_bar.Close)
            
            self.prev_bar = bar
            
        else:
            
            self.prev_bar = bar
            
            
    @property
    def IsReady(self):
        
        return self._retStd.IsReady and self.fast.IsReady and self.slow.IsReady
    
    
    def Update(self, bar):
        
        # Check if it Gapes Up or Down
        if bar.Open >= self.daily_bar.High * (1+self.entryZscore * self._retStd.Current.Value):
            
            self.gapIndicator = 1
            
            self.gapLevel = (bar.Open - self.daily_bar.High)/self.daily_bar.High
            
        elif bar.Open <= self.daily_bar.Low*(1-self.entryZscore*self._retStd.Current.Value):
            
            self.gapIndicator = -1
            
            self.gapLevel = (self.daily_bar.Low - bar.Open)/self.daily_bar.Low
        
        
        # Check if the gape trend is still on
        if self.gapIndicator > 0 and self.fast.Current.Value < self.slow.Current.Value:
            
            self.gapIndicator = 0
            
        elif self.gapIndicator < 0 and self.fast.Current.Value > self.slow.Current.Value:
            
            self.gapIndicator = 0
            
            
            
class SimpleGapAlphaModel(AlphaModel):
    
    def __init__(self, algorithm):
        
        self.algo = algorithm
        
        self.addedSym = []
        
        self.removedSym = []
        
        
        
    def OnSecuritiesChanged(self, algorithm, changes):
        
        addedEquities = [x for x in changes.AddedSecurities]
        
        self.addedSym = [x.Symbol for x in addedEquities]
        
        removedEquities = [x for x in changes.RemovedSecurities]
        
        self.removedSym = [x.Symbol for x in removedEquities]
        
        
        
    def Update(self, algorithm, data):
        
        insights = []
        
        if self.algo.IsWarmingUp:
            
            return insights
            
        
        if self.removedSym is not None:
            
            for x in self.removedSym:
                
                insights.append(Insight.Price(x, timedelta(days=self.algo.insightsDaysDuration), InsightDirection.Flat))
                
            self.removedSym = None
            
            
        if self.addedSym is not None:
            
            for x in self.addedSym:
                
                if self.algo.symDict[x.Value].gapIndicator > 0:
                    
                    insights.append(Insight.Price(x,timedelta(days=self.algo.insightsDaysDuration), InsightDirection.Up))
                
                elif self.algo.symDict[x.Value].gapIndicator < 0:
                    
                    insights.append(Insight.Price(x,timedelta(days=self.algo.insightsDaysDuration), InsightDirection.Down))
            
            self.addedSym = None
        
        return insights