| 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 -4.374 Tracking Error 0.137 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.SetEndDate(2020, 1, 6)
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
symbol = self.AddEquity('SPY', Resolution.Hour).Symbol
self.Schedule.On(self.DateRules.EveryDay(symbol), self.TimeRules.At(10, 0), 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
#dizionario con gli indicatori e dati consolidati per ogni simbolo
self.symDict = { };
for ticker in self.initialList:
symbol = self.AddEquity(ticker, Resolution.Hour).Symbol
self.symDict[symbol] = SymbolData(self, symbol, self.entryZscore, self.myStdDays, self.fastAverage, self.slowAverage)
self.symbols = []
self.SetWarmUp(150, Resolution.Daily)
# Scheduled Universe Construction
def SelectSymbols(self):
symbols = []
for symbol, symbol_data in self.symDict.items():
if symbol_data.gapIndicator != 0:
symbols.append(symbol)
subset = {symbol: self.symDict[symbol] for symbol in symbols}
symbols = [sym.symbol for sym in (sorted(subset.values(), key=operator.attrgetter('gapLevel')))]
self.symbols = symbols[self.concurrentEquities:]
return self.symbols
def OnData(self, data):
if not (data is not None and data.Time.time() <= time(10,1)):
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])
# Emit insights for `self.symbols`
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.fast = ExponentialMovingAverage(self.fastAverage)
self.slow = ExponentialMovingAverage(self.slowAverage)
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):
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