| Overall Statistics |
|
Total Orders 2359 Average Win 0.43% Average Loss -0.56% Compounding Annual Return 10.725% Drawdown 32.900% Expectancy 0.080 Start Equity 100000 End Equity 172368.82 Net Profit 72.369% Sharpe Ratio 0.296 Sortino Ratio 0.32 Probabilistic Sharpe Ratio 7.975% Loss Rate 39% Win Rate 61% Profit-Loss Ratio 0.77 Alpha 0.003 Beta 0.67 Annual Standard Deviation 0.174 Annual Variance 0.03 Information Ratio -0.136 Tracking Error 0.153 Treynor Ratio 0.077 Total Fees $3274.62 Estimated Strategy Capacity $180000000.00 Lowest Capacity Asset EFA S79U6IHK5HLX Portfolio Turnover 26.01% Drawdown Recovery 580 |
#region imports
from AlgorithmImports import *
#endregion
from datetime import timedelta
"""
PARAMETERIZED QC FRAMEWORK MOMENTUM MODEL
This strategy demonstrates how QuantConnect Framework algorithms can be made easy
to adjust through parameters.
The model uses a manually defined multi-asset universe containing broad equity
ETFs, bonds, credit, gold, real estate, emerging markets, large-cap technology,
and selected individual stocks.
The alpha model ranks the universe by 14-day momentum. Momentum is measured using
QuantConnect's MOM indicator. Securities with the strongest positive momentum
receive Up insights. Securities with weak or negative momentum receive Flat
insights. The portfolio construction model then equal-weights the active Up
insights.
The risk model uses MaximumDrawdownPercentPerSecurity. The drawdown limit is read
from the QuantConnect parameter called risk_tolerance. For example, if the
parameter is set to 10, the model uses a 10% maximum drawdown threshold per
security.
Framework structure:
1. ManualUniverseSelectionModel defines the tradable universe.
2. MOMAlphaModel ranks securities by momentum.
3. EqualWeightingPortfolioConstructionModel equal-weights active Up insights.
4. MaximumDrawdownPercentPerSecurity controls individual security drawdowns.
5. ImmediateExecutionModel submits orders immediately.
This version is pedagogical and intentionally simple, but safer than the original:
it checks indicator readiness, avoids fixed list-index errors, provides parameter
fallbacks, and plots the strategy against SPY.
"""
class FrameworkAlgorithm(QCAlgorithm):
def Initialize(self):
# ------------------------------------------------------------
# 1. BACKTEST SETTINGS
# ------------------------------------------------------------
self.SetStartDate(2021, 1, 1)
self.SetEndDate(2026, 5, 5)
self.initial_cash = 100000
self.SetCash(self.initial_cash)
# ------------------------------------------------------------
# 2. PARAMETERS
# ------------------------------------------------------------
# QuantConnect parameters are strings.
# risk_tolerance is expected as a percentage, for example:
# 5 means 5%
# 10 means 10%
risk_tolerance_parameter = self.GetParameter("risk_tolerance")
if risk_tolerance_parameter is None or risk_tolerance_parameter == "":
risk_tolerance_parameter = "10"
self.max_risk = float(risk_tolerance_parameter) / 100.0
# Optional parameters.
# These make the model easier to tune without changing the code.
momentum_period_parameter = self.GetParameter("momentum_period")
if momentum_period_parameter is None or momentum_period_parameter == "":
momentum_period_parameter = "14"
self.momentum_period = int(momentum_period_parameter)
top_count_parameter = self.GetParameter("top_count")
if top_count_parameter is None or top_count_parameter == "":
top_count_parameter = "3"
self.top_count = int(top_count_parameter)
insight_days_parameter = self.GetParameter("insight_days")
if insight_days_parameter is None or insight_days_parameter == "":
insight_days_parameter = "5"
self.insight_days = int(insight_days_parameter)
# ------------------------------------------------------------
# 3. UNIVERSE SELECTION MODEL
# ------------------------------------------------------------
symbols = [
Symbol.Create("SPY", SecurityType.Equity, Market.USA),
Symbol.Create("GLD", SecurityType.Equity, Market.USA),
Symbol.Create("HYG", SecurityType.Equity, Market.USA),
Symbol.Create("BND", SecurityType.Equity, Market.USA),
Symbol.Create("EEM", SecurityType.Equity, Market.USA),
Symbol.Create("EFA", SecurityType.Equity, Market.USA),
Symbol.Create("IYR", SecurityType.Equity, Market.USA),
Symbol.Create("QQQ", SecurityType.Equity, Market.USA),
Symbol.Create("TSLA", SecurityType.Equity, Market.USA),
Symbol.Create("AAPL", SecurityType.Equity, Market.USA),
Symbol.Create("TLT", SecurityType.Equity, Market.USA)
]
self.UniverseSettings.Resolution = Resolution.Daily
self.SetUniverseSelection(
ManualUniverseSelectionModel(symbols)
)
# ------------------------------------------------------------
# 4. ALPHA MODEL
# ------------------------------------------------------------
self.AddAlpha(
MOMAlphaModel(
momentum_period=self.momentum_period,
top_count=self.top_count,
insight_days=self.insight_days
)
)
# ------------------------------------------------------------
# 5. PORTFOLIO CONSTRUCTION MODEL
# ------------------------------------------------------------
self.SetPortfolioConstruction(
EqualWeightingPortfolioConstructionModel()
)
# ------------------------------------------------------------
# 6. RISK MANAGEMENT
# ------------------------------------------------------------
self.SetRiskManagement(
MaximumDrawdownPercentPerSecurity(self.max_risk)
)
# ------------------------------------------------------------
# 7. EXECUTION
# ------------------------------------------------------------
self.SetExecution(
ImmediateExecutionModel()
)
# ------------------------------------------------------------
# 8. BENCHMARK
# ------------------------------------------------------------
self._benchmark = self.AddEquity(
"SPY",
Resolution.Daily,
Market.USA
).Symbol
self.SetBenchmark(self._benchmark)
self.initial_benchmark_price = None
# ------------------------------------------------------------
# 9. DIAGNOSTICS
# ------------------------------------------------------------
self.strategy_peak = self.initial_cash
self.benchmark_peak = self.initial_cash
self.Debug(
"Parameters "
+ "risk_tolerance="
+ str(self.max_risk)
+ " momentum_period="
+ str(self.momentum_period)
+ " top_count="
+ str(self.top_count)
+ " insight_days="
+ str(self.insight_days)
)
def OnData(self, data):
# ------------------------------------------------------------
# 1. BENCHMARK DATA
# ------------------------------------------------------------
if self._benchmark not in data or data[self._benchmark] is None:
return
benchmark_price = self.Securities[self._benchmark].Price
if benchmark_price <= 0:
return
if self.initial_benchmark_price is None:
self.initial_benchmark_price = benchmark_price
benchmark_value = (
self.initial_cash
* benchmark_price
/ self.initial_benchmark_price
)
# ------------------------------------------------------------
# 2. EQUITY CURVE PLOTS
# ------------------------------------------------------------
self.Plot(
"Strategy Equity",
"Portfolio Value",
self.Portfolio.TotalPortfolioValue
)
self.Plot(
"Strategy Equity",
"Buy Hold SPY",
benchmark_value
)
# ------------------------------------------------------------
# 3. PORTFOLIO STATE PLOTS
# ------------------------------------------------------------
invested_value = 0
active_holdings = 0
for holding in self.Portfolio.Values:
if holding.Invested:
invested_value += abs(holding.HoldingsValue)
active_holdings += 1
if self.Portfolio.TotalPortfolioValue > 0:
invested_weight = (
invested_value
/ self.Portfolio.TotalPortfolioValue
)
cash_weight = 1 - invested_weight
self.Plot(
"Portfolio State",
"Invested Weight",
invested_weight
)
self.Plot(
"Portfolio State",
"Cash Weight",
cash_weight
)
self.Plot(
"Portfolio Diagnostics",
"Active Holdings",
active_holdings
)
# ------------------------------------------------------------
# 4. DRAWDOWN PLOTS
# ------------------------------------------------------------
self.strategy_peak = max(
self.strategy_peak,
self.Portfolio.TotalPortfolioValue
)
self.benchmark_peak = max(
self.benchmark_peak,
benchmark_value
)
strategy_drawdown = (
self.Portfolio.TotalPortfolioValue
/ self.strategy_peak
- 1
)
benchmark_drawdown = (
benchmark_value
/ self.benchmark_peak
- 1
)
self.Plot(
"Drawdown",
"Strategy Drawdown",
strategy_drawdown
)
self.Plot(
"Drawdown",
"Benchmark Drawdown",
benchmark_drawdown
)
self.Plot(
"Risk Management",
"Risk Limit",
-self.max_risk
)
class MOMAlphaModel(AlphaModel):
def __init__(
self,
momentum_period=14,
top_count=3,
insight_days=5
):
self.momentum_period = momentum_period
self.top_count = top_count
self.insight_period = timedelta(days=insight_days)
self.momentum_by_symbol = {}
self.last_emit_date = None
def OnSecuritiesChanged(self, algorithm, changes):
# ------------------------------------------------------------
# ADD INDICATORS FOR NEW SECURITIES
# ------------------------------------------------------------
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.momentum_by_symbol:
self.momentum_by_symbol[symbol] = algorithm.MOM(
symbol,
self.momentum_period,
Resolution.Daily
)
# ------------------------------------------------------------
# REMOVE INDICATORS FOR REMOVED SECURITIES
# ------------------------------------------------------------
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.momentum_by_symbol:
del self.momentum_by_symbol[symbol]
def Update(self, algorithm, data):
insights = []
if algorithm.IsWarmingUp:
return insights
current_date = algorithm.Time.date()
# Emit insights once per day.
# This prevents repeated emissions inside the same daily bar.
if self.last_emit_date == current_date:
return insights
ready_items = []
for symbol, indicator in self.momentum_by_symbol.items():
if not indicator.IsReady:
continue
if not algorithm.Securities.ContainsKey(symbol):
continue
if not algorithm.Securities[symbol].HasData:
continue
ready_items.append(
{
"symbol": symbol,
"momentum": indicator.Current.Value
}
)
if len(ready_items) == 0:
return insights
# Rank from strongest momentum to weakest momentum.
ranked = sorted(
ready_items,
key=lambda item: item["momentum"],
reverse=True
)
# Select strongest positive momentum names.
selected_symbols = []
for item in ranked:
if item["momentum"] <= 0:
continue
if len(selected_symbols) >= self.top_count:
break
selected_symbols.append(item["symbol"])
# Emit Up insights for selected securities.
for symbol in selected_symbols:
insights.append(
Insight.Price(
symbol,
self.insight_period,
InsightDirection.Up,
0.01,
1.0
)
)
# Emit Flat insights for ready securities not selected.
# This helps the framework exit names that fall out of the top group.
for item in ranked:
symbol = item["symbol"]
if symbol not in selected_symbols:
insights.append(
Insight.Price(
symbol,
self.insight_period,
InsightDirection.Flat,
0.0,
1.0
)
)
self.last_emit_date = current_date
# ------------------------------------------------------------
# DIAGNOSTICS
# ------------------------------------------------------------
algorithm.Plot(
"Alpha Diagnostics",
"Selected Momentum Names",
len(selected_symbols)
)
if len(ranked) > 0:
algorithm.Plot(
"Alpha Diagnostics",
"Top Momentum",
ranked[0]["momentum"]
)
return Insight.Group(insights)