| Overall Statistics |
|
Total Trades 0 Average Win 0% Average Loss 0% Compounding Annual Return 0% Drawdown 0% Expectancy 0 Net Profit 0% Sharpe Ratio 0 Probabilistic Sharpe Ratio 0% Loss Rate 0% Win Rate 0% Profit-Loss Ratio 0 Alpha 0 Beta 0 Annual Standard Deviation 0 Annual Variance 0 Information Ratio -0.725 Tracking Error 0.246 Treynor Ratio 0 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset |
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)
self.UniverseSettings.Resolution = Resolution.Hour
self.SetExecution(ImmediateExecutionModel())
self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())
self.SetRiskManagement(MaximumDrawdownPercentPerSecurity(0.01))
# 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,30),
self.SelectSymbols
))
################################### 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
####################################################################################################
#subscribe to every ticker on the list
for sym in self.initialList:
self.AddEquity(sym, Resolution.Hour)
#dizionario con gli indicatori e dati consolidati per ogni simbolo
self.symDict = { };
for sym in self.initialList:
self.symDict[sym] = SymbolData(self, sym, self.entryZscore, self.myStdDays, self.fastAverage, self.slowAverage)
#lista dei simboli da tradare
self.symbols = []
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
# Emitting alphas on Securities changed
def OnSecuritiesChanged(self, changes):
insights = []
addedEquities = [x for x in changes.AddedSecurities if x.Type == SecurityType.Equity]
addedSym = [x.Symbol for x in addedEquities]
removedEquities = [x for x in changes.RemovedSecurities]
removedSym = [x.Symbol for x in removedEquities]
for x in removedSym:
insights.append(Insight.Price(x,timedelta(days=self.insightsDaysDuration), InsightDirection.Flat))
for x in addedSym:
if self.symDict[x.Value].gapIndicator > 0:
insights.append(Insight.Price(x,timedelta(days=self.insightsDaysDuration), InsightDirection.Up))
elif self.symDict[x.Value].gapIndicator < 0:
insights.append(Insight.Price(x,timedelta(days=self.insightsDaysDuration), InsightDirection.Down))
return insights
# 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 data is not None and data.Time.time() <= time(10,1):
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