| Overall Statistics |
|
Total Trades 10 Average Win 17.53% Average Loss -17.89% Compounding Annual Return -15.838% Drawdown 89.000% Expectancy -0.152 Net Profit -29.167% Sharpe Ratio 0.24 Probabilistic Sharpe Ratio 12.016% Loss Rate 57% Win Rate 43% Profit-Loss Ratio 0.98 Alpha -0.029 Beta 1.769 Annual Standard Deviation 0.646 Annual Variance 0.418 Information Ratio 0.084 Tracking Error 0.61 Treynor Ratio 0.088 Total Fees $136.90 Estimated Strategy Capacity $220000000.00 Lowest Capacity Asset NQ WSVU0MELFS3L |
from typing import Dict
# CONFIGS
RSI_period = 14
RSI_upper = 30
RSI_lower = 27
bar_size = timedelta(minutes=5)
portfolio_pct = .6
stoploss_dist = 20
takeprofit_dist = 20
max_losses = 2 # max number of losses in a day
debug = True # turn off to reduce logging
# END CONFIGS
f = False
if f:
from AlgorithmImports import *
class Consulting(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2018, 1, 1)
self.SetEndDate(2020, 1, 1)
self.SetCash(100000)
future = self.AddFuture(Futures.Indices.NASDAQ100EMini, Resolution.Minute)
future.SetFilter(lambda x: x.FrontMonth().OnlyApplyFilterAtMarketOpen())
self.rsi = RelativeStrengthIndex(RSI_period)
self.consolidators: Dict[Symbol, QuoteBarConsolidator] = {}
self.stoploss = None
self.takeprofit = None
self.last_rsi = None
self.loss_count = 0
self.curr_day = -1
def OnData(self, data:Slice):
# new day
if self.curr_day != self.Time.day:
self.curr_day = self.Time.day
self.Reset()
price = self.Securities[self.GetSymbol()].Price
if self.takeprofit and price > self.takeprofit:
self.Liquidate()
elif self.stoploss and price < self.stoploss:
self.loss_count += 1
self.Liquidate()
def GetSymbol(self) -> Symbol:
'''
get current front month contract
'''
return list(self.consolidators.keys())[0]
def Reset(self):
self.stoploss = None
self.takeprofit = None
self.last_rsi = None
self.loss_count = 0
def Print(self, msg):
if debug:
self.Log(msg)
def OnDataConsolidated(self, sender, quoteBar:QuoteBar):
'''
5 minute consolidator
Update RSI, SetHoldings
'''
self.rsi.Update(self.Time, quoteBar.Close)
if not self.rsi.IsReady:
return
curr_rsi = self.rsi.Current.Value
if self.loss_count > max_losses:
return
self.Plot('RSI', 'Value', curr_rsi)
if (not self.Portfolio.Invested and self.last_rsi
and self.last_rsi < RSI_lower and curr_rsi > RSI_upper):
if quoteBar.High - stoploss_dist < quoteBar.Close:
self.stoploss = quoteBar.High - stoploss_dist
self.takeprofit = quoteBar.High + stoploss_dist
self.SetHoldings(self.GetSymbol(), portfolio_pct)
self.last_rsi = curr_rsi
def OnSecuritiesChanged(self, changes):
'''
register consolidator for frontmonth contract, deregister consolidator for expired contract
'''
for security in changes.AddedSecurities:
consolidator = QuoteBarConsolidator(bar_size)
consolidator.DataConsolidated += self.OnDataConsolidated
self.SubscriptionManager.AddConsolidator(security.Symbol, consolidator)
self.consolidators[security.Symbol] = consolidator
for security in changes.RemovedSecurities:
consolidator = self.consolidators.pop(security.Symbol)
self.SubscriptionManager.RemoveConsolidator(security.Symbol, consolidator)
consolidator.DataConsolidated -= self.OnDataConsolidated