| Overall Statistics |
|
Total Orders 867 Average Win 0.12% Average Loss -0.10% Compounding Annual Return 9.701% Drawdown 11.600% Expectancy 0.102 Start Equity 100000 End Equity 108153.48 Net Profit 8.153% Sharpe Ratio 0.201 Sortino Ratio 0.216 Probabilistic Sharpe Ratio 46.166% Loss Rate 51% Win Rate 49% Profit-Loss Ratio 1.23 Alpha -0.059 Beta 0.536 Annual Standard Deviation 0.073 Annual Variance 0.005 Information Ratio -1.77 Tracking Error 0.069 Treynor Ratio 0.028 Total Fees $963.94 Estimated Strategy Capacity $2200000.00 Lowest Capacity Asset VOX T2FCD04TATET Portfolio Turnover 22.10% |
#region imports
from AlgorithmImports import *
#endregion
class DualMomentumAlphaModel(AlphaModel):
def __init__(self, algorithm, fast_period = 10, slow_period = 20, moving_average_type = MovingAverageType.EXPONENTIAL, resolution = Resolution.DAILY):
self.fast_period = fast_period
self.slow_period = slow_period
self.moving_average_type = moving_average_type
self.resolution = resolution
self.day = -1
self.symbol_data = {}
self.ETF_data = {}
self.etf_map = {
algorithm.Securities["VGT"] : MorningstarSectorCode.TECHNOLOGY,
algorithm.Securities["XLB"] : MorningstarSectorCode.BASIC_MATERIALS,
algorithm.Securities["XLY"] : MorningstarSectorCode.CONSUMER_CYCLICAL,
algorithm.Securities["XLF"] : MorningstarSectorCode.FINANCIAL_SERVICES,
algorithm.Securities["VNQ"] : MorningstarSectorCode.REAL_ESTATE,
algorithm.Securities["XLV"] : MorningstarSectorCode.HEALTHCARE,
algorithm.Securities["XLP"] : MorningstarSectorCode.CONSUMER_DEFENSIVE,
algorithm.Securities["XLU"] : MorningstarSectorCode.UTILITIES,
algorithm.Securities["VOX"] : MorningstarSectorCode.COMMUNICATION_SERVICES,
algorithm.Securities["XLE"] : MorningstarSectorCode.ENERGY,
algorithm.Securities["XLI"] : MorningstarSectorCode.INDUSTRIALS
}
self.insight_collection = InsightCollection()
def update(self, algorithm, data):
insights = []
for symbol in set(data.splits.keys() + data.dividends.keys()):
if symbol in self.symbol_data.keys():
self.symbol_data[symbol].ppo.reset()
algorithm.subscription_manager.remove_consolidator(symbol, self.symbol_data[symbol].Consolidator)
self._register_indicator(algorithm, self.symbol_data[symbol].security)
history = algorithm.history[TradeBar](symbol, 20,
Resolution.DAILY,
data_normalization_mode=DataNormalizationMode.SCALED_RAW)
for bar in history:
self.symbol_data[symbol].Consolidator.update(bar)
if symbol in self.ETF_data.keys():
self.ETF_data[symbol].ppo.reset()
algorithm.subscription_manager.remove_consolidator(symbol, self.ETF_data[symbol].Consolidator)
self._register_indicator(algorithm, self.ETF_data[symbol].security)
history = algorithm.history[TradeBar](symbol, 20,
Resolution.DAILY,
data_normalization_mode=DataNormalizationMode.SCALED_RAW)
for bar in history:
self.ETF_data[symbol].Consolidator.update(bar)
if data.quote_bars.count == 0:
return []
if self.day == algorithm.time.day:
return []
self.day = algorithm.time.day
momentum_by_sector = {}
security_momentum = {}
#target_sectors = [self.etf_map[sector_etf.security] for sector_etf in self.etf_data.values()
#if sector_etf.security.symbol in data.quote_bars and sector_etf.ppo.is_ready
#and sector_etf.ppo.current.value > 0]
target_sectors = [sector_etf.security for sector_etf in self.ETF_data.values()
if sector_etf.security.symbol in data.quote_bars and sector_etf.ppo.is_ready
and sector_etf.ppo.current.value > 0]
#target_securities = []
#for sector in target_sectors:
#for security in security_momentum[sector]:
#if security_momentum[sector][security] > 0:
#target_securities.append(security)
#target_securities = sorted(target_securities, key = lambda x: algorithm.securities[x.symbol].Fundamentals.MarketCap, reverse=True)[:10]
#for sector in target_securities:
#insights.append(Insight.price(security.symbol, Expiry.END_OF_DAY, InsightDirection.UP))
for sector_security in target_sectors:
insights.append(Insight.price(sector_security.symbol, Expiry.END_OF_DAY, InsightDirection.UP))
return insights
def on_securities_changed(self, algorithm, changes):
security_by_symbol = {}
ETF_by_symbol = {}
for added in changes.added_securities:
if added in self.etf_map.keys():
self.ETF_data[added.symbol] = ETF(algorithm, added, self.fast_period, self.slow_period, self.moving_average_type, self.resolution)
else:
self.symbol_data[added.symbol] = SymbolData(algorithm, added, self.fast_period, self.slow_period, self.moving_average_type, self.resolution)
if security_by_symbol:
history = algorithm.history[TradeBar](list(security_by_symbol.keys()), 20,
Resolution.DAILY,
data_normalization_mode=DataNormalizationMode.SCALED_RAW)
for trade_bars in history:
for bar in trade_bars.values():
symbol_data[bar.symbol].consolidator.update(bar)
if ETF_by_symbol:
history = algorithm.history[TradeBar](list(ETF_by_symbol.keys()), 20,
Resolution.DAILY,
data_normalization_mode=DataNormalizationMode.SCALED_RAW)
for trade_bars in history:
for bar in trade_bars.values():
ETF_data[bar.symbol].consolidator.update(bar)
for removed in changes.removed_securities:
symbol = removed.Symbol
if removed in self.etf_map.keys():
data = self.ETF_data.pop(symbol, None)
if data is not None:
algorithm.SubscriptionManager.RemoveConsolidator(symbol, data.Consolidator)
else:
data = self.symbol_data.pop(symbol, None)
if data is not None:
algorithm.SubscriptionManager.RemoveConsolidator(symbol, data.Consolidator)
def _register_indicator(self, algorithm, security):
if security.symbol in self.symbol_data.keys():
self.symbol_data[security.symbol].Consolidator = TradeBarConsolidator(timedelta(days = 1))
algorithm.subscription_manager.add_consolidator(security.symbol, self.symbol_data[security.symbol].Consolidator)
algorithm.RegisterIndicator(security.symbol, self.symbol_data[security.symbol].ppo, self.symbol_data[security.symbol].Consolidator)
if security.symbol in self.ETF_data.keys():
self.ETF_data[security.symbol].Consolidator = TradeBarConsolidator(timedelta(days=1))
algorithm.subscription_manager.add_consolidator(security.symbol, self.ETF_data[security.symbol].Consolidator)
algorithm.RegisterIndicator(security.symbol, self.ETF_data[security.symbol].ppo, self.ETF_data[security.symbol].Consolidator)
class SymbolData:
def __init__(self, algorithm, security, fast_period, slow_period, moving_average_type, resolution):
self.security = security
self.sector = security.Fundamentals.AssetClassification.MorningstarSectorCode
self.ppo = PercentagePriceOscillator(security.symbol, fast_period, slow_period, moving_average_type)
self.Consolidator = algorithm.ResolveConsolidator(security.symbol, resolution)
algorithm.RegisterIndicator(security.symbol, self.ppo, self.Consolidator)
algorithm.WarmUpIndicator(security.symbol, self.ppo, resolution)
class ETF:
def __init__(self, algorithm, security, fast_period, slow_period, moving_average_type, resolution):
etf_map = {
algorithm.Securities["VGT"] : MorningstarSectorCode.TECHNOLOGY,
algorithm.Securities["XLB"] : MorningstarSectorCode.BASIC_MATERIALS,
algorithm.Securities["XLY"] : MorningstarSectorCode.CONSUMER_CYCLICAL,
algorithm.Securities["XLF"] : MorningstarSectorCode.FINANCIAL_SERVICES,
algorithm.Securities["VNQ"] : MorningstarSectorCode.REAL_ESTATE,
algorithm.Securities["XLV"] : MorningstarSectorCode.HEALTHCARE,
algorithm.Securities["XLP"] : MorningstarSectorCode.CONSUMER_DEFENSIVE,
algorithm.Securities["XLU"] : MorningstarSectorCode.UTILITIES,
algorithm.Securities["VOX"] : MorningstarSectorCode.COMMUNICATION_SERVICES,
algorithm.Securities["XLE"] : MorningstarSectorCode.ENERGY,
algorithm.Securities["XLI"] : MorningstarSectorCode.INDUSTRIALS
}
self.security = security
self.sector = etf_map[security]
self.ppo = PercentagePriceOscillator(security.symbol, fast_period, slow_period, moving_average_type)
self.Consolidator = algorithm.ResolveConsolidator(security.symbol, resolution)
algorithm.RegisterIndicator(security.symbol, self.ppo, self.Consolidator)
algorithm.WarmUpIndicator(security.symbol, self.ppo, resolution)# region imports
from AlgorithmImports import *
from DualMomentumAlphaModel import *
# endregion
class SectorDualMomentumStrategy(QCAlgorithm):
undesired_symbols_from_previous_deployment = []
checked_symbols_from_previous_deployment = False
def initialize(self):
self.set_start_date(2023, 6, 5)
self.set_end_date(2024, 6, 5)
self.set_cash(100000)
#self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
self.settings.minimum_order_margin_portfolio_percentage = 0
self.universe_settings.data_normalization_mode = DataNormalizationMode.RAW
self.universe_settings.asynchronous = True
self.add_universe(self.universe.etf("SPY", self.universe_settings, self._etf_constituents_filter))
self.add_equity("VGT")
self.add_equity("XLB")
self.add_equity("XLY")
self.add_equity("XLF")
self.add_equity("VNQ")
self.add_equity("XLP")
self.add_equity("XLV")
self.add_equity("XLU")
self.add_equity("VOX")
self.add_equity("XLE")
self.add_equity("XLI")
self.add_alpha(DualMomentumAlphaModel(self))
self.settings.rebalance_portfolio_on_security_changes = False
self.settings.rebalance_portfolio_on_insight_changes = False
self.day = -1
self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(self._rebalance_func))
self.add_risk_management(TrailingStopRiskManagementModel())
self.set_execution(ImmediateExecutionModel())
self.set_warm_up(timedelta(7))
def _etf_constituents_filter(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]:
selected = sorted([c for c in constituents if c.weight],
key=lambda c: c.weight, reverse=True)[:200]
return [c.symbol for c in selected]
def _rebalance_func(self, time):
if self.day != self.time.day and not self.is_warming_up and self.current_slice.quote_bars.count > 0:
self.day = self.time.day
return time
return None
def on_data(self, data):
if not self.is_warming_up and not self.checked_symbols_from_previous_deployment:
for security_holding in self.portfolio.values():
if not security_holding.invested:
continue
symbol = security_holding.symbol
if not self.insights.has_active_insights(symbol, self.utc_time):
self.undesired_symbols_from_previous_deployment.append(symbol)
self.checked_symbols_from_previous_deployment = True
for symbol in self.undesired_symbols_from_previous_deployment:
if self.is_market_open(symbol):
self.liquidate(symbol, tag="Not backed up by current insights")
self.undesired_symbols_from_previous_deployment.remove(symbol)