| Overall Statistics |
|
Total Orders 39 Average Win 5.53% Average Loss -3.61% Compounding Annual Return 148.906% Drawdown 18.900% Expectancy 0.864 Start Equity 50000 End Equity 92359.27 Net Profit 84.719% Sharpe Ratio 2.306 Sortino Ratio 4.679 Probabilistic Sharpe Ratio 87.166% Loss Rate 26% Win Rate 74% Profit-Loss Ratio 1.53 Alpha 0.87 Beta 1.461 Annual Standard Deviation 0.423 Annual Variance 0.179 Information Ratio 2.689 Tracking Error 0.336 Treynor Ratio 0.667 Total Fees $105.27 Estimated Strategy Capacity $0 Lowest Capacity Asset QQQ RIWIV7K5Z9LX Portfolio Turnover 15.64% Drawdown Recovery 23 |
from AlgorithmImports import *
from datetime import timedelta
import numpy as np
mom_safe_asset_period = 125
class DualMomentumInOut(QCAlgorithm):
def initialize(self):
self.set_start_date(2025, 1, 1)
self.cap = 50000
self.set_cash(self.cap)
self.set_warm_up(timedelta(days=760))
# Asset esistenti
self.TQQQ = self.add_equity('TQQQ', Resolution.DAILY).Symbol
self.UVXY = self.add_equity('UVXY', Resolution.DAILY).Symbol
self.QQQ = self.add_equity('QQQ', Resolution.DAILY).Symbol
self.safe_asset = None
# RSI per le logiche già esistenti
self.RSI_PERIOD = 8
self.rsi_tqqq = self.rsi(self.TQQQ, self.RSI_PERIOD, MovingAverageType.WILDERS, Resolution.DAILY)
self.rsi_qqq = self.rsi(self.QQQ, self.RSI_PERIOD, MovingAverageType.WILDERS, Resolution.DAILY)
# Asset per regime bullish
self.STK = self.add_equity('QQQ', Resolution.DAILY).Symbol
# Altri asset usati per le logiche di exit
self.SLV = self.add_equity('SLV', Resolution.DAILY).Symbol
self.GLD = self.add_equity('GLD', Resolution.DAILY).Symbol
self.XLI = self.add_equity('XLI', Resolution.DAILY).Symbol
self.XLU = self.add_equity('XLU', Resolution.DAILY).Symbol
self.DBB = self.add_equity('DBB', Resolution.DAILY).Symbol
self.UUP = self.add_equity('UUP', Resolution.DAILY).Symbol # già usato in risk off
self.MKT = self.add_equity('QQQ', Resolution.DAILY).Symbol
self.BNCH = self.add_equity('QQQ', Resolution.DAILY).Symbol
self.pairs = [self.XLI, self.XLU, self.GLD, self.SLV, self.DBB, self.UUP]
# Aggiungo gli extra per la selezione tramite momentum
# Asset per regime risk off: TLT e BND2
self.TLT = self.add_equity('TLT', Resolution.DAILY).Symbol
self.BND2 = self.add_equity('UUP', Resolution.DAILY).Symbol
self.IEF = self.add_equity('IEF', Resolution.DAILY).Symbol
self.SHY = self.add_equity('SHY', Resolution.DAILY).Symbol
self.USMV = self.add_equity('USMV', Resolution.DAILY).Symbol
self.SPHD = self.add_equity('SPHD', Resolution.DAILY).Symbol
self.VDC = self.add_equity('VDC', Resolution.DAILY).Symbol
self.BSV = self.add_equity('BSV', Resolution.DAILY).Symbol
self.DBC = self.add_equity('DBC', Resolution.DAILY).Symbol
self.REZ = self.add_equity('REZ', Resolution.DAILY).Symbol
self.ASSETS = [self.STK, self.TLT, self.BND2, self.IEF, self.SHY, self.GLD, self.BSV, self.DBC, self.REZ, self.TLT]
self.RISK_OFF_MOMENTUM_ASSETS = [self.TLT, self.BND2, self.IEF, self.SHY, self.GLD, self.BSV, self.DBC, self.REZ, self.TLT] # USMV SPHD VDC BSV DBC REZ
# Variabili di controllo
self.bull = True
self.count = 0
self.outday = 0
self.wt = {}
self.real_wt = {}
self.mkt = []
self.current_inflation = 0
self.fredcode = "MEDCPIM158SFRBCLE"
self.cpi = self.add_data(Fred, self.fredcode, Resolution.DAILY).Symbol
self.schedule.on(self.date_rules.every_day(),
self.time_rules.after_market_open(self.STK, 100),
self.daily_check)
# Recupero storico: includo anche i safe asset e gli extra per momentum
symbols = list(set([self.MKT] + self.pairs + [self.TLT, self.BND2] + self.RISK_OFF_MOMENTUM_ASSETS))
for symbol in symbols:
consolidator = TradeBarConsolidator(timedelta(days=1))
self.subscription_manager.add_consolidator(symbol, consolidator)
history = self.history(symbols, 127, Resolution.DAILY)
if history.empty or 'close' not in history.columns:
return
self.history_data = history['close'].unstack(level=0).dropna()
self.last_trade_date = None
def select_momentum_safe_asset(self):
"""Seleziona il safe asset tra quelli in RISK_OFF_MOMENTUM_ASSETS basandosi sul momentum"""
momentum = {}
vola = self.history_data[self.MKT].pct_change().std() * np.sqrt(252)
period = mom_safe_asset_period #int((1.0 - vola) * 85)
if period <= 0:
period = 20
for asset in self.RISK_OFF_MOMENTUM_ASSETS:
try:
mom = self.history_data[asset].pct_change(period).iloc[-1]
momentum[asset] = mom
self.Debug(f"Asset {asset} - Period: {period} - Momentum: {mom}")
except Exception as e:
momentum[asset] = -99999
self.Debug(f"Asset {asset} - Errore nel calcolo del momentum: {e}")
best_asset = max(momentum, key=momentum.get)
self.Debug(f"Momentum safe asset: {momentum}, best: {best_asset}")
return best_asset
def daily_check(self):
if self.last_trade_date and (self.time - self.last_trade_date).days < 1:
return
livinfla = 6
symbols = list(set([self.MKT] + self.pairs + [self.TLT, self.BND2] + self.RISK_OFF_MOMENTUM_ASSETS))
self.history_data = self.history(symbols, 127, Resolution.DAILY)['close'].unstack(level=0).dropna()
current_rsi = self.rsi_tqqq.current.value
qqq_rsi = self.rsi_qqq.current.value
# Logica RSI per TQQQ e UVXY
if current_rsi >= 84: #84
self.set_holdings(self.UVXY, 1, liquidate_existing_holdings=True)
elif current_rsi < 75: #75
self.liquidate(self.UVXY)
if current_rsi <= 25: #25
self.set_holdings(self.TQQQ, 1, liquidate_existing_holdings=True)
elif current_rsi >= 30: #30
self.liquidate(self.TQQQ)
current_inflation = self.securities[self.cpi].price
vola = self.history_data[self.MKT].pct_change().std() * np.sqrt(252)
wait_days = 10
period = int((((1.0 - vola) * 85)))
r = self.history_data.pct_change(period).iloc[-1]
exit1 = (r[self.XLI] < r[self.XLU]) and (r[self.SLV] < r[self.GLD])
exit2 = (r[self.XLI] < r[self.XLU]) and (r[self.SLV] < r[self.GLD]) and (r[self.DBB] < r[self.UUP])
exit = exit1 if current_inflation > livinfla else exit2
if exit:
self.bull = False
self.outday = self.count
if self.count >= self.outday + wait_days:
self.bull = True
self.count += 1
if not self.bull:
# Se non abbiamo ancora scelto il safe asset, lo scegliamo in base all'inflazione/momentum:
if self.safe_asset is None:
if current_inflation > livinfla:
self.safe_asset = self.BND2
self.Debug("Inflazione alta: safe asset impostato su BND2")
else:
self.safe_asset = self.select_momentum_safe_asset()
self.Debug(f"Safe asset scelto tramite momentum: {self.safe_asset}")
# Manteniamo la scelta: investiamo interamente sul safe asset già fissato
for sec in self.ASSETS:
self.wt[sec] = 1 if sec == self.safe_asset else 0
else:
# Se siamo in regime bullish, resettiamo safe_asset
self.safe_asset = None
for sec in self.ASSETS:
self.wt[sec] = 1 if sec == self.STK else 0
self.trade()
if self.portfolio[self.TQQQ].invested or self.portfolio[self.UVXY].invested:
self.last_trade_date = self.time
def trade(self):
for sec, weight in self.wt.items():
if weight == 0 and self.portfolio[sec].is_long:
self.liquidate(sec)
elif weight > 0:
if not self.portfolio[sec].invested:
buying_power = self.portfolio.get_buying_power(sec, OrderDirection.BUY)
if buying_power >= self.portfolio.total_portfolio_value * weight:
self.set_holdings(sec, weight)
def on_end_of_day(self):
vola = self.history_data[self.MKT].pct_change().std() * np.sqrt(252)
period = int((1.0 - vola) * 85)
r = self.history_data.pct_change(period).iloc[-1]
r_g_l_d = round(((r[self.GLD] - r[self.SLV]) * 50), 100)
r_x_l_u = round(((r[self.XLU] - r[self.XLI]) * 50), 100)
r_u_u_p = round(((r[self.UUP] - r[self.DBB]) * 50), 100)
self.plot('ROC', 'GOLD/SLV', r_g_l_d)
self.plot('ROC', 'XLU/XLI', r_x_l_u)
self.plot('ROC', 'UUP/DBB', r_u_u_p)
current_inflation = self.securities[self.cpi].price
self.plot('CPI', 'CPI', current_inflation)
wait_days = 10 #int(vola * 85)
self.plot('Wait_days', 'Days', wait_days)
account_leverage = self.portfolio.total_holdings_value / self.portfolio.total_portfolio_value
self.plot('Holdings', 'leverage', round(account_leverage, 1))