| Overall Statistics |
|
Total Trades 117 Average Win 0.07% Average Loss -0.07% Compounding Annual Return 134.556% Drawdown 0.000% Expectancy 0.424 Net Profit 2.603% Sharpe Ratio 11.481 Probabilistic Sharpe Ratio 99.824% Loss Rate 31% Win Rate 69% Profit-Loss Ratio 1.06 Alpha 1.253 Beta -0.332 Annual Standard Deviation 0.092 Annual Variance 0.009 Information Ratio 4.161 Tracking Error 0.116 Treynor Ratio -3.195 Total Fees $117.30 Estimated Strategy Capacity $34000000.00 Lowest Capacity Asset MKL R735QTJ8XC9X |
import pandas as pd
class WellDressedSkyBlueSardine(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2020, 12, 21)
self.SetEndDate(2020, 12, 31)
self.SetCash(1000000)
self.SPY = self.AddEquity('SPY', Resolution.Daily).Symbol
self.AddUniverse(self.CoarseFilter, self.FineFilter)
self.UniverseSettings.Resolution = Resolution.Daily
self.UniverseSettings.SetDataNormalizationMode = DataNormalizationMode.SplitAdjusted
self.SCTRuniverse = []
self.potential = []
self.Data = {}
self.indicators = {}
self.percentagebuy = 0.05
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen('SPY', 30), self.daily_check)
def CoarseFilter(self, coarse):
sortedByDollarVolume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
return [x.Symbol for x in sortedByDollarVolume if x.Price > 10
and x.HasFundamentalData][:1000]
def FineFilter(self, fine):
sortedByPE = sorted(fine, key=lambda x: x.MarketCap)
self.universe = [x.Symbol for x in sortedByPE if x.MarketCap > 10e9][:500]
# self.Debug(self.universe)
return self.universe
def OnSecuritiesChanged(self, changes):
# close positions in removed securities
for x in changes.RemovedSecurities:
self.Liquidate()
if x.Symbol in self.Data:
del self.Data[x.Symbol]
# can't open positions here since data might not be added correctly yet
for x in changes.AddedSecurities:
self.Data[x.Symbol] = SymbolData(self, x.Symbol)
SCTR = {}
for symboldata in self.Data.values():
if symboldata.IsReady():
SCTR[symboldata.symbol] = symboldata.SCTR()
sctr_df = pd.DataFrame.from_dict(SCTR, orient='index')
sctr_df['Decile_rank'] = pd.qcut(sctr_df.iloc[:, 0], 10000, labels = False) /100
sctr_df = sctr_df.drop(sctr_df.columns[0], axis=1)
sctr_df = sctr_df.loc[sctr_df['Decile_rank'] >= 85]
self.SCTRuniverse = sctr_df
# self.Debug(SCTR)
return self.SCTRuniverse
def OnData(self, data):
for symboldata in self.SCTRuniverse.itertuples():
symbol = symboldata.Index
D_rank = symboldata.Decile_rank
if not symbol in self.potential:
self.potential.append(symbol)
for symbol in self.potential:
if not data.ContainsKey(symbol):
continue
self.indicators[symbol] = Entry_Indicator(self, symbol)
if self.IsWarmingUp: continue
tradeBars = data.Bars
symbolTradeBar = tradeBars[symbol]
symbolOpen = symbolTradeBar.Open ## Open price
symbolClose = symbolTradeBar.Close ## Close price
self.slow_stoch = self.indicators[symbol].get_sto()
self.sma = self.indicators[symbol].get_sma()
def daily_check(self):
for symbol in self.potential:
if not self.Securities.ContainsKey(symbol):
continue
if self.IsWarmingUp: continue
symbolClose = float(self.Securities[symbol].Close)
self.buyquantity = round((self.percentagebuy*self.Portfolio.TotalPortfolioValue)/symbolClose) if symbolClose != 0 else 0
if (not self.Portfolio[symbol].Invested) and (self.Portfolio.MarginRemaining > 0.9*self.percentagebuy*self.Portfolio.TotalPortfolioValue):
if ((symbolClose/self.sma)> 0 and self.slow_stoch<=30):
self.MarketOrder(symbol, self.buyquantity)
self.Log(str(symbol) + " -sto: " + str(self.slow_stoch))
self.Log(str(symbol) + " -sma: " + str(symbolClose/self.sma))
for kvp in self.Portfolio:
holdings = kvp.Value
symbol = holdings.Symbol
if holdings.Invested:
if not symbol in self.potential:
self.Liquidate(symbol)
self.Log(str(symbol) + " : " + str(holdings.Price))
class SymbolData:
def __init__(self, algo, symbol):
self.symbol = symbol
self.EMA200 = algo.EMA(symbol, 200, Resolution.Daily)
self.EMA50 = algo.EMA(symbol, 50, Resolution.Daily)
self.ROC125 = algo.ROC(symbol, 125, Resolution.Daily)
self.ROC20 = algo.ROC(symbol, 20, Resolution.Daily)
self.PPO = algo.PPO(symbol, 12, 26, MovingAverageType.Exponential, Resolution.Daily)
self.RSI14 = algo.RSI(symbol, 14, MovingAverageType.Simple, Resolution.Daily)
self.PPOWindow = RollingWindow[float](3)
history = algo.History(symbol, 200, Resolution.Daily)
for index, bar in history.loc[symbol].iterrows():
self.EMA200.Update(index, bar.close)
self.EMA50.Update(index, bar.close)
self.ROC125.Update(index, bar.close)
self.ROC20.Update(index, bar.close)
self.PPO.Update(index, bar.close)
self.RSI14.Update(index, bar.close)
self.PPOWindow.Add(self.PPO.Current.Value)
def SCTR(self):
return self.EMA200.Current.Value*0.3 + self.EMA50.Current.Value*0.15 + self.ROC125.Current.Value*0.3 + self.ROC20.Current.Value*0.15 \
+ (self.PPOWindow[0] - self.PPOWindow[2])/(3*self.PPOWindow[2]) *0.05 + self.RSI14.Current.Value*0.05
def IsReady(self):
return self.EMA200.IsReady and self.EMA50.IsReady \
and self.ROC125.IsReady and self.ROC20.IsReady\
and self.PPO.IsReady and self.RSI14.IsReady
class Entry_Indicator:
def __init__(self, algo, symbol):
self.SMA200 = algo.SMA(symbol, 200, Resolution.Daily)
self.Slow_Stoch5 = algo.STO(symbol, 5, Resolution.Daily)
history = algo.History(symbol, 200, Resolution.Daily)
for index, bar in history.loc[symbol].iterrows():
self.SMA200.Update(index, bar.close)
for bar in history.itertuples():
tradeBar = TradeBar(bar.Index[1], bar.Index[0], bar.open, bar.high, bar.low, bar.close, bar.volume, timedelta(1))
self.Slow_Stoch5.Update(tradeBar)
def get_sma(self):
return self.SMA200.Current.Value
def get_sto(self):
return self.Slow_Stoch5.Current.Value