| Overall Statistics |
|
Total Trades 25 Average Win 17.16% Average Loss -4.30% Compounding Annual Return 9.012% Drawdown 31.500% Expectancy 0.663 Net Profit 32.269% Sharpe Ratio 0.47 Probabilistic Sharpe Ratio 11.778% Loss Rate 67% Win Rate 33% Profit-Loss Ratio 3.99 Alpha 0.066 Beta 0.116 Annual Standard Deviation 0.16 Annual Variance 0.025 Information Ratio -0.015 Tracking Error 0.239 Treynor Ratio 0.646 Total Fees $25.00 Estimated Strategy Capacity $540000000.00 Lowest Capacity Asset SPY R735QTJ8XC9X Portfolio Turnover 2.09% |
#region imports
from AlgorithmImports import *
#endregion
class CPIData(PythonData):
# 12-month unadjusted CPI data
# Source: https://www.bls.gov/charts/consumer-price-index/consumer-price-index-by-category-line-chart.htm
# Release dates source: https://www.bls.gov/bls/news-release/cpi.htm
def GetSource(self,
config: SubscriptionDataConfig,
date: datetime,
isLive: bool) -> SubscriptionDataSource:
return SubscriptionDataSource("https://www.dropbox.com/s/f02a9htg6pyhf9p/CPI%20data%201.csv?dl=1", SubscriptionTransportMedium.RemoteFile)
def Reader(self,
config: SubscriptionDataConfig,
line: str,
date: datetime,
isLive: bool) -> BaseData:
if not (line.strip()):
return None
cpi = CPIData()
cpi.Symbol = config.Symbol
try:
def parse(pct):
return float(pct[:-1]) / 100
data = line.split(',')
cpi.EndTime = datetime.strptime(data[0], "%m%d%Y %H:%M %p")
cpi["month"] = data[1]
cpi['all-items'] = parse(data[2])
cpi['food'] = parse(data[3])
cpi['food-at-home'] = parse(data[4])
cpi['food-away-from-home'] = parse(data[5])
cpi['energy'] = parse(data[6])
cpi['gasoline'] = parse(data[7])
cpi['electricity'] = parse(data[8])
cpi['natural-gas'] = parse(data[9])
cpi['all-items-less-food-and-energy'] = parse(data[10])
cpi['commodities-less-food-and-energy-commodities'] = parse(data[11])
cpi['apparel'] = parse(data[12])
cpi['new-vehicles'] = parse(data[13])
cpi['medical-car-commodities'] = parse(data[14])
cpi['services-less-energy-services'] = parse(data[15])
cpi['shelter'] = parse(data[16])
cpi['medical-care-services'] = parse(data[17])
cpi['education-and-communication'] = parse(data[18])
cpi.Value = cpi['all-items-less-food-and-energy']
except ValueError:
# Do nothing
return None
return cpi#region imports
from AlgorithmImports import *
#endregion
import numpy as np
from datetime import datetime
import matplotlib.pyplot as plt
from AlgorithmImports import *
from CPI import CPIData
# -------------------------------------------------------------------------
STK = ['QQQ']; BND = ['TLT']; VOLA = 126; BASE_RET = 85; DAY = 85 ;LEV = 1.00 #855
LEV = 1.00 #85
LEV = 1.00; #85
PAIRS = ['SLV', 'GLD', 'XLI', 'XLU', 'DBB', 'UUP'] ; res = Resolution.Daily
# -------------------------------------------------------------------------
class DualMomentumInOut(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2020,1,1)
# self.SetEndDate(2010,1,1)
self.cap = 10000 #Settare il Capitale Iniziale
self.SetCash(self.cap)
self.AddEquity('SPY', res).Symbol
self.SetBenchmark('SPY')
self.STK = self.AddEquity('SPY', res).Symbol
self.BND = self.AddEquity('TLT', res).Symbol
self.ASSETS = [self.STK, self.BND]
self.SLV = self.AddEquity('SLV', res).Symbol
self.GLD = self.AddEquity('GLD', res).Symbol
self.XLI = self.AddEquity('XLI', res).Symbol
self.XLU = self.AddEquity('XLU', res).Symbol
self.DBB = self.AddEquity('DBB', res).Symbol
self.UUP = self.AddEquity('UUP', res).Symbol
# self.SPY = self.AddEquity('SPY', res).Symbol
# self.TLT = self.AddEquity('TLT', res).Symbol
self.MKT = self.AddEquity('SPY', res).Symbol
self.BNCH = self.AddEquity('SPY', res).Symbol
self.pairs = [self.XLI, self.XLU, self.GLD, self.SLV, self.DBB, self.UUP] #self.TVC, self.TIP
self.bull = 1
self.count = 0
self.outday = 0
self.wt = {}
self.real_wt = {}
self.mkt = []
self.SetWarmUp(timedelta(350))
self.cpi = self.AddData(CPIData, "CPI")
# Aggiungi un grafico per plottare il CPI
chart = Chart("CPI")
series = Series("CPI", SeriesType.Line)
chart.AddSeries(series)
self.AddChart(chart)
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen('SPY', 100), #100), #1000), #1005 0), #100), #1000), #100100), #100), #1000), #1005 0), #100), #1000), #100
self.daily_check)
symbols = [self.MKT] + self.pairs
for symbol in symbols:
self.consolidator = TradeBarConsolidator(timedelta(days=1))
self.consolidator.DataConsolidated += self.consolidation_handler
self.SubscriptionManager.AddConsolidator(symbol, self.consolidator)
self.history = self.History(symbols, VOLA + 1, Resolution.Daily)
if self.history.empty or 'close' not in self.history.columns:
return
self.history = self.history['close'].unstack(level=0).dropna()
def consolidation_handler(self, sender, consolidated):
self.history.loc[consolidated.EndTime, consolidated.Symbol] = consolidated.Close
self.history = self.history.iloc[-(VOLA + 1):]
def daily_check(self):
vola =0.7*( self.history[[self.MKT]].pct_change().std() * np.sqrt(252))
wait_days = int(vola * DAY)
self.Debug('{}'.format(wait_days))
period = int((1.0 - vola) * BASE_RET)
r = self.history.pct_change(period).iloc[-1]
exit = (r[self.XLI] < r[self.XLU]) and (r[self.SLV] < r[self.GLD]) and (r[self.DBB] < r[self.UUP])
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:
for sec in self.ASSETS:
self.wt[sec] = LEV if sec is self.BND else 0
self.trade()
elif self.bull:
for sec in self.ASSETS:
self.wt[sec] = LEV if sec is self.STK else 0
self.trade()
def trade(self):
for sec, weight in self.wt.items():
if weight == 0 and self.Portfolio[sec].IsLong:
self.Liquidate(sec)
cond1 = weight == 0 and self.Portfolio[sec].IsLong
cond2 = weight > 0 and not self.Portfolio[sec].Invested
if cond1 or cond2:
self.SetHoldings(sec, weight)
def OnEndOfDay(self):
vola = self.history[[self.MKT]].pct_change().std() * np.sqrt(252)
period = int((1.0 - vola) * (BASE_RET))
r = self.history.pct_change(period).iloc[-1]
rGLD = round(((r[self.GLD] - r[self.SLV]) * 50), 100)
rXLU = round(((r[self.XLU] - r[self.XLI]) * 50), 100)
rUUP = round(((r[self.UUP] - r[self.DBB]) * 50), 100)
# rI = round(((r[self.RINF] - r[self.TLT]) * 50), 100)
self.Plot('ROC', 'GOLD/SLV', rGLD)
self.Plot('ROC', 'XLU/XLI', rXLU)
self.Plot('ROC', 'UUP/DBB', rUUP)
# self.Plot('ROC', 'RINF/TLT', rI)
vola = self.history[[self.MKT]].pct_change().std() * np.sqrt(252)
wait_days = int(vola * DAY)
self.Plot('Wait_days', 'Days', wait_days)
# mkt_price = self.Securities[self.BNCH].Close
#self.mkt.append(mkt_price)
#mkt_perf = self.mkt[-1] / self.mkt[0] * self.cap
#self.Plot('Strategy Equity', 'SPY', mkt_perf)
account_leverage = self.Portfolio.TotalHoldingsValue / self.Portfolio.TotalPortfolioValue
self.Plot('Holdings', 'leverage', round(account_leverage, 1))
# Crea una istanza della classe CPIData come simbolo personalizzato
def OnData(self, data):
if not data.ContainsKey("CPI"):
return
# Ottieni il valore del CPI dal simbolo personalizzato
cpi = data["CPI"].Value
# Aggiorna il valore del CPI sul grafico
self.Plot("CPI", "CPI", cpi)