| Overall Statistics |
|
Total Orders 1151 Average Win 0.44% Average Loss -0.58% Compounding Annual Return 43.137% Drawdown 13.500% Expectancy 0.574 Start Equity 100000 End Equity 600064.25 Net Profit 500.064% Sharpe Ratio 1.64 Sortino Ratio 2.134 Probabilistic Sharpe Ratio 96.537% Loss Rate 10% Win Rate 90% Profit-Loss Ratio 0.76 Alpha 0.214 Beta 0.726 Annual Standard Deviation 0.155 Annual Variance 0.024 Information Ratio 1.625 Tracking Error 0.122 Treynor Ratio 0.35 Total Fees $1202.18 Estimated Strategy Capacity $150000000.00 Lowest Capacity Asset FIX R735QTJ8XC9X Portfolio Turnover 2.10% Drawdown Recovery 245 |
# region imports
from AlgorithmImports import *
# endregion
class ConsistentGrowthMomentum(QCAlgorithm):
"""Macro momentum strategy with Monte Carlo portfolio optimization over a fixed, diversified universe."""
def initialize(self):
self.set_start_date(self.end_date - timedelta(5 * 365))
self.set_cash(100000)
self._lookback_days = 126
self._safety_buffer = 0.95
self._mc_simulations = 500
self._securities = []
for ticker in ["NVDA", "GLD", "COST", "LLY", "MCK", "FIX", "AXON"]:
self._securities.append(self.add_equity(ticker, Resolution.DAILY))
self.schedule.on(self.date_rules.week_start(), self.time_rules.at(8, 0), self._rebalance_portfolio)
self.set_warm_up(self._lookback_days + 15)
def on_warmup_finished(self):
self._rebalance_portfolio()
def _rebalance_portfolio(self):
if self.is_warming_up:
return
score_by_security = {}
returns_by_security = {}
for security in self._securities:
if not security.has_data or security.price <= 0:
continue
history = self.history(security, self._lookback_days + 5, Resolution.DAILY)
if history.empty or len(history) < self._lookback_days:
continue
closing_prices = history["close"]
daily_returns = closing_prices.pct_change().dropna()
volatility = daily_returns.std()
if volatility <= 0:
continue
historical_price = closing_prices.iloc[-self._lookback_days]
score = ((closing_prices.iloc[-1] - historical_price) / historical_price) / volatility
if score <= 0:
continue
score_by_security[security] = score
returns_by_security[security] = daily_returns
# Liquidate everything when no asset passes the absolute momentum gate.
if not score_by_security:
for security in self._securities:
if security.holdings.invested:
self.liquidate(security)
return
target_securities = list(score_by_security.keys())
weight_by_security = {security: self._safety_buffer / len(target_securities) for security in target_securities}
aligned_returns = pd.DataFrame(returns_by_security).dropna()
if not aligned_returns.empty:
returns_matrix = aligned_returns.values
best_sharpe = -99999.0
# Search random long-only weightings for the highest in-sample Sharpe ratio.
for _ in range(self._mc_simulations):
random_w = np.random.rand(len(target_securities))
normalized_w = random_w / np.sum(random_w)
sim_portfolio_returns = np.dot(returns_matrix, normalized_w)
volatility_return = np.std(sim_portfolio_returns)
if volatility_return > 0:
sim_sharpe = np.mean(sim_portfolio_returns) / volatility_return
if sim_sharpe > best_sharpe:
best_sharpe = sim_sharpe
for i, security in enumerate(target_securities):
weight_by_security[security] = normalized_w[i] * self._safety_buffer
# Liquidate holdings that are no longer targets, then size the survivors.
for security in self._securities:
if security.holdings.invested and security not in target_securities:
self.liquidate(security)
for security, weight in weight_by_security.items():
self.set_holdings(security, weight)