| Overall Statistics |
|
Total Orders 90 Average Win 11.19% Average Loss -7.51% Compounding Annual Return 815.347% Drawdown 7.700% Expectancy 0.611 Start Equity 1000 End Equity 13145.48 Net Profit 1214.548% Sharpe Ratio 5.133 Sortino Ratio 6.823 Probabilistic Sharpe Ratio 99.997% Loss Rate 35% Win Rate 65% Profit-Loss Ratio 1.49 Alpha 0 Beta 0 Annual Standard Deviation 0.167 Annual Variance 0.028 Information Ratio 5.463 Tracking Error 0.167 Treynor Ratio 0 Total Fees $93.49 Estimated Strategy Capacity $4100000.00 Lowest Capacity Asset QLD TJNNZWL5I4IT Portfolio Turnover 15.91% |
from AlgorithmImports import *
import time
class RSIRebalanceStrategy(QCAlgorithm):
def Initialize(self):
self.set_start_date(2024, 1, 1)
if not self.live_mode:
self.set_warmup(timedelta(days=250))
self.set_cash(1000)
self.settings.free_portfolio_value_percentage = 0.01
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
self.qqq = self.add_equity("QQQ", Resolution.MINUTE, data_normalization_mode=DataNormalizationMode.RAW).Symbol
self.lqqq = self.add_equity("QLD", Resolution.MINUTE, data_normalization_mode=DataNormalizationMode.RAW).Symbol
self.spy = self.add_equity("SPY", Resolution.MINUTE, data_normalization_mode=DataNormalizationMode.RAW).Symbol
self.xlu = self.add_equity("XLU", Resolution.MINUTE).Symbol
self.bil = self.add_equity("BIL", Resolution.MINUTE).Symbol
self.uvxy = self.add_equity("UVXY", Resolution.MINUTE).Symbol
self.strat_symbols = [self.qqq, self.lqqq, self.spy, self.xlu, self.bil, self.uvxy]
self.schedule.on(self.date_rules.every_day(), self.time_rules.before_market_close(self.spy, 5), self.rebalance)
if not self.live_mode:
self.schedule.on(self.date_rules.week_end(), self.time_rules.after_market_close(self.spy, 0), self.add_cash)
# self.schedule.on(self.date_rules.month_end(0), self.time_rules.after_market_close(self.spy, 0), self.add_cash)
# self.schedule.on(self.date_rules.month_end(15), self.time_rules.after_market_close(self.spy, 0), self.add_cash)
self.cashInvested = self.portfolio.cash_book["USD"].amount
# def on_data(self, slice: Slice) -> None:
# # Obtain the mapped TradeBar of the symbol if any
# # if not self.portfolio[self.spy].invested:
# # self.set_holdings(self.spy, 1, True)
# self.set_holdings_2(self.xlu, 1)
def add_cash(self):
dcaCash = 98
self.portfolio.cash_book["USD"].add_amount(dcaCash)
self.cashInvested += dcaCash
def rsi_2(self,equity,period):
extension = min(period*5,250)
r_w = RollingWindow[float](extension)
history = self.history(equity,extension - 1,Resolution.DAILY)
for historical_bar in history:
r_w.add(historical_bar.close)
while r_w.count < extension:
current_price = self.securities[equity].price
r_w.add(current_price)
if r_w.is_ready:
average_gain = 0
average_loss = 0
gain = 0
loss = 0
for i in range(extension - 1,extension - period -1,-1):
gain += max(r_w[i-1] - r_w[i],0)
loss += abs(min(r_w[i-1] - r_w[i],0))
average_gain = gain/period
average_loss = loss/period
for i in range(extension - period - 1,0,-1):
average_gain = (average_gain*(period-1) + max(r_w[i-1] - r_w[i],0))/period
average_loss = (average_loss*(period-1) + abs(min(r_w[i-1] - r_w[i],0)))/period
if average_loss == 0:
return 100
else:
rsi = 100 - (100/(1 + average_gain / average_loss))
return rsi
else:
return None
def sma_2(self,equity,period):
r_w = RollingWindow[float](period)
history = self.history(equity,period - 1,Resolution.DAILY)
for historical_bar in history:
r_w.add(historical_bar.close)
while r_w.count < period:
current_price = self.securities[equity].price
r_w.add(current_price)
if r_w.is_ready:
sma = sum(r_w) / period
return sma
else:
return 0
def get_strat_positions(self):
current_positions = []
for strat_symbol in self.strat_symbols:
if self.portfolio[strat_symbol].invested:
current_positions.append(strat_symbol)
return current_positions
def set_holdings_2(self, symbol, portion):
"""IBKR Brokage Model somehow doesn't wait till liquidation finishes in set_holdings(symbol, 1, True)
So we liquidate explicitly first and set_holdings after
"""
liquidate_others = not self.portfolio[symbol].invested
current_positions = self.get_strat_positions()
# liquidate any other strategy's symbols when switching
if liquidate_others and len(current_positions) > 0:
for curr_pos in current_positions:
self.liquidate(curr_pos)
while liquidate_others and len(self.get_strat_positions()) > 0:
time.sleep(5) # Pause execution for 5 seconds
self.debug("Waiting to liquidate")
self.debug(f"Liqudated, buying {symbol} now")
self.set_holdings(symbol, portion)
def rebalance(self):
if self.time < self.start_date:
return
if not self.securities[self.spy].has_data:
return
rsi_value = self.rsi_2(self.qqq.value, 10)
spy_price = self.securities[self.spy].price
spy_sma_200 = self.sma_2(self.spy.value, 200)
spy_sma_30 = self.sma_2(self.spy.value, 30)
if rsi_value > 79:
self.set_holdings_2(self.uvxy, 1)
elif rsi_value < 31:
self.set_holdings_2(self.lqqq, 1)
elif spy_price > spy_sma_200:
if spy_price > spy_sma_30:
self.set_holdings_2(self.qqq, 1)
else:
self.set_holdings_2(self.xlu, 1)
else:
if spy_price > spy_sma_30:
self.set_holdings_2(self.xlu, 1)
else:
self.set_holdings_2(self.bil, 1)
def on_end_of_algorithm(self) -> None:
self.debug(f"Cash invested: {self.cashInvested}")