| Overall Statistics |
|
Total Orders 1257 Average Win 0.15% Average Loss -0.13% Compounding Annual Return 40.932% Drawdown 6.400% Expectancy 0.297 Start Equity 100000 End Equity 130232.02 Net Profit 30.232% Sharpe Ratio 2.093 Sortino Ratio 2.998 Probabilistic Sharpe Ratio 91.984% Loss Rate 41% Win Rate 59% Profit-Loss Ratio 1.19 Alpha 0.119 Beta 0.764 Annual Standard Deviation 0.108 Annual Variance 0.012 Information Ratio 1.046 Tracking Error 0.083 Treynor Ratio 0.295 Total Fees $1286.74 Estimated Strategy Capacity $3000000.00 Lowest Capacity Asset MRK R735QTJ8XC9X Portfolio Turnover 25.68% |
#region imports
from AlgorithmImports import *
#endregion
class DualMomentumAlphaModel(AlphaModel):
def __init__(self):
self.sectors = {}
self.securities_list = []
self.day = -1
def update(self, algorithm, data):
insights = []
for symbol in set(data.splits.keys() + data.dividends.keys()):
security = algorithm.securities[symbol]
if security in self.securities_list:
security.indicator.reset()
algorithm.subscription_manager.remove_consolidator(security.symbol, security.consolidator)
self._register_indicator(algorithm, security)
history = algorithm.history[TradeBar](security.symbol, 7,
Resolution.DAILY,
data_normalization_mode=DataNormalizationMode.SCALED_RAW)
for bar in history:
security.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 = {}
for sector in self.sectors:
securities = self.sectors[sector]
security_momentum[sector] = {security: security.indicator.current.value
for security in securities if
security.symbol in data.quote_bars and security.indicator.is_ready}
momentum_by_sector[sector] = sum(list(security_momentum[sector].values())) / len(self.sectors[sector])
target_sectors = [sector for sector in self.sectors if momentum_by_sector[sector] > 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 security in target_securities:
insights.append(Insight.price(security.symbol, Expiry.END_OF_DAY, InsightDirection.UP))
return insights
def on_securities_changed(self, algorithm, changes):
security_by_symbol = {}
for security in changes.RemovedSecurities:
if security in self.securities_list:
algorithm.subscription_manager.remove_consolidator(security.symbol, security.consolidator)
self.securities_list.remove(security)
for sector in self.sectors:
if security in self.sectors[sector]:
self.sectors[sector].remove(security)
for security in changes.AddedSecurities:
sector = security.Fundamentals.AssetClassification.MorningstarSectorCode
security_by_symbol[security.symbol] = security
security.indicator = MomentumPercent(1)
self._register_indicator(algorithm, security)
self.securities_list.append(security)
if sector not in self.sectors:
self.sectors[sector] = set()
self.sectors[sector].add(security)
if security_by_symbol:
history = algorithm.history[TradeBar](list(security_by_symbol.keys()), 7,
Resolution.DAILY,
data_normalization_mode=DataNormalizationMode.SCALED_RAW)
for trade_bars in history:
for bar in trade_bars.values():
security_by_symbol[bar.symbol].consolidator.update(bar)
def _register_indicator(self, algorithm, security):
security.consolidator = TradeBarConsolidator(Calendar.WEEKLY)
algorithm.subscription_manager.add_consolidator(security.symbol, security.consolidator)
algorithm.register_indicator(security.symbol, security.indicator, security.consolidator)
# 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_alpha(DualMomentumAlphaModel())
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))
self.set_benchmark("SPY")
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)