| Overall Statistics |
|
Total Orders 1012 Average Win 0.16% Average Loss -0.15% Compounding Annual Return 2.796% Drawdown 19.700% Expectancy 0.218 Start Equity 10000000 End Equity 11804463.42 Net Profit 18.045% Sharpe Ratio -0.216 Sortino Ratio -0.259 Probabilistic Sharpe Ratio 2.742% Loss Rate 42% Win Rate 58% Profit-Loss Ratio 1.10 Alpha -0.03 Beta 0.171 Annual Standard Deviation 0.059 Annual Variance 0.004 Information Ratio -0.887 Tracking Error 0.13 Treynor Ratio -0.075 Total Fees $34596.59 Estimated Strategy Capacity $39000000.00 Lowest Capacity Asset IEI TP8J6Z7L419H Portfolio Turnover 2.61% Drawdown Recovery 1458 |
#region imports
from AlgorithmImports import *
#endregion
import matplotlib.pyplot as plt
import scipy.stats
xs = scipy.stats.norm.rvs(5, 2, 10000)
fig, axes = plt.subplots(1, 2, figsize=(9, 3))
axes[0].hist(xs, bins=50)
axes[0].set_title("Samples")
axes[1].hist(
scipy.stats.norm.cdf(xs, 5, 2),
bins=50
)
axes[1].set_title("CDF(samples)")#region imports
from AlgorithmImports import *
#endregion
"""
COMPREHENSIVE STRATEGY EXPLANATION
This strategy uses the QuantConnect Algorithm Framework. It combines manual
universe selection, a composite technical alpha model, mean-variance portfolio
construction, immediate execution, risk management, and benchmark diagnostics.
The universe is a diversified ETF basket including equities, international
equities, emerging markets, bonds, real estate, gold, equity factor ETFs, sectors,
and industry ETFs.
The alpha model creates a composite technical score for each ETF. The score uses
three simple signals:
1. MACD confirmation:
One point if the MACD line is above the MACD signal line.
2. RSI confirmation:
One point if RSI is above 45.
3. Trend confirmation:
One point if price is above the 50-day simple moving average.
The score ranges from 0 to 3. This calibrated version is intentionally less strict:
an ETF only needs a score of at least 1 to receive an Up insight. This gives the
mean-variance optimizer a broader opportunity set.
The important implementation point is that the alpha model emits insights only
once per month. It does not emit daily signals and it does not emit repeated Flat
insights. This avoids the repeated insight / liquidation / re-entry loop that can
occur when the optimizer and risk model keep receiving conflicting instructions.
Only ETFs with qualifying composite alpha scores receive active Up insights. The
MeanVarianceOptimizationPortfolioConstructionModel then performs optimization only
on the active insight set.
Execution is handled by ImmediateExecutionModel. Risk management is handled by
MaximumDrawdownPercentPortfolio. The benchmark is 60% SPY and 40% AGG.
"""
class MonthlyCompositeTechnicalAlphaModel(AlphaModel):
def __init__(
self,
macd_fast=12,
macd_slow=26,
macd_signal=9,
rsi_period=14,
rsi_threshold=45,
sma_period=50,
minimum_score=1,
insight_duration_days=40
):
self.macd_fast = macd_fast
self.macd_slow = macd_slow
self.macd_signal = macd_signal
self.rsi_period = rsi_period
self.rsi_threshold = rsi_threshold
self.sma_period = sma_period
self.minimum_score = minimum_score
self.insight_duration = timedelta(days=insight_duration_days)
self.macd_by_symbol = {}
self.rsi_by_symbol = {}
self.sma_by_symbol = {}
self.last_emit_month = None
def Update(self, algorithm, data):
insights = []
if algorithm.IsWarmingUp:
return insights
current_month = (algorithm.Time.year, algorithm.Time.month)
# ------------------------------------------------------------
# HARD LOOP CONTROL
# ------------------------------------------------------------
# Emit insights only once per month.
# This prevents repeated optimization and re-entry loops.
if self.last_emit_month == current_month:
return insights
selected_count = 0
total_score = 0
scored_count = 0
for symbol in list(self.macd_by_symbol.keys()):
if not algorithm.Securities.ContainsKey(symbol):
continue
security = algorithm.Securities[symbol]
if not security.HasData:
continue
if symbol not in data:
continue
price = security.Price
if price <= 0:
continue
macd = self.macd_by_symbol[symbol]
rsi = self.rsi_by_symbol[symbol]
sma = self.sma_by_symbol[symbol]
if not macd.IsReady or not rsi.IsReady or not sma.IsReady:
continue
# --------------------------------------------------------
# COMPOSITE SCORE
# --------------------------------------------------------
score = 0
macd_strength = macd.Current.Value - macd.Signal.Current.Value
if macd_strength > 0:
score += 1
if rsi.Current.Value > self.rsi_threshold:
score += 1
if price > sma.Current.Value:
score += 1
total_score += score
scored_count += 1
# --------------------------------------------------------
# ONLY QUALIFIED ETFs RECEIVE INSIGHTS
# --------------------------------------------------------
if score >= self.minimum_score:
selected_count += 1
# Magnitude cannot be None for mean-variance optimization.
# Score 1 = small alpha
# Score 2 = medium alpha
# Score 3 = stronger alpha
magnitude = 0.005 + 0.0075 * score
confidence = 1.0
insights.append(
Insight.Price(
symbol,
self.insight_duration,
InsightDirection.Up,
magnitude,
confidence
)
)
self.last_emit_month = current_month
# ------------------------------------------------------------
# ALPHA DIAGNOSTICS
# ------------------------------------------------------------
algorithm.Plot(
"Composite Alpha",
"Qualified Securities",
selected_count
)
if scored_count > 0:
algorithm.Plot(
"Composite Alpha",
"Average Score",
total_score / scored_count
)
algorithm.Debug(
"Monthly composite alpha emitted "
+ str(len(insights))
+ " insights on "
+ str(algorithm.Time.date())
)
return insights
def OnSecuritiesChanged(self, algorithm, changes):
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.macd_by_symbol:
self.macd_by_symbol[symbol] = algorithm.MACD(
symbol,
self.macd_fast,
self.macd_slow,
self.macd_signal,
MovingAverageType.Exponential,
Resolution.Daily
)
self.rsi_by_symbol[symbol] = algorithm.RSI(
symbol,
self.rsi_period,
MovingAverageType.Wilders,
Resolution.Daily
)
self.sma_by_symbol[symbol] = algorithm.SMA(
symbol,
self.sma_period,
Resolution.Daily
)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.macd_by_symbol:
del self.macd_by_symbol[symbol]
if symbol in self.rsi_by_symbol:
del self.rsi_by_symbol[symbol]
if symbol in self.sma_by_symbol:
del self.sma_by_symbol[symbol]
class MeanVarianceOptimizationAlgorithm(QCAlgorithm):
def Initialize(self):
# ------------------------------------------------------------
# 1. BACKTEST SETTINGS
# ------------------------------------------------------------
self.SetStartDate(2020, 5, 1)
self.SetEndDate(2026, 5, 5)
self.initial_cash = 10000000
self.SetCash(self.initial_cash)
# ------------------------------------------------------------
# 2. UNIVERSE SELECTION
# ------------------------------------------------------------
self.UniverseSettings.Resolution = Resolution.Daily
tickers = [
# Aggregate indices
"IEFA", "AGG", "IWM", "EEM", "EWJ", "EPP",
# Fixed income and real estate
"IYR", "LQD", "EMB", "IEF", "IEI",
# Commodities
"IAU",
# Factors
"USMV", "DGRO", "QUAL", "DVY", "MTUM", "VLUE",
"EFAV", "EEMV", "IDV", "IQLT",
# Sectors and industries
"IBB", "IHI", "IYW", "IGF", "IYH", "IYF",
"IXC", "PICK", "IYE", "KXI", "WOOD",
# Broad U.S. equity anchor
"SPY"
]
symbols = [
Symbol.Create(ticker, SecurityType.Equity, Market.USA)
for ticker in tickers
]
self.SetUniverseSelection(
ManualUniverseSelectionModel(symbols)
)
# ------------------------------------------------------------
# 3. ALPHA MODEL
# ------------------------------------------------------------
self.AddAlpha(
MonthlyCompositeTechnicalAlphaModel(
macd_fast=12,
macd_slow=26,
macd_signal=9,
rsi_period=14,
rsi_threshold=45,
sma_period=50,
minimum_score=1,
insight_duration_days=40
)
)
# Warm up indicators.
self.SetWarmUp(120, Resolution.Daily)
# ------------------------------------------------------------
# 4. MEAN-VARIANCE PORTFOLIO CONSTRUCTION
# ------------------------------------------------------------
self.SetPortfolioConstruction(
MeanVarianceOptimizationPortfolioConstructionModel(
resolution=Resolution.Daily,
period=60
)
)
# ------------------------------------------------------------
# 5. EXECUTION
# ------------------------------------------------------------
self.SetExecution(
ImmediateExecutionModel()
)
# ------------------------------------------------------------
# 6. RISK MANAGEMENT
# ------------------------------------------------------------
self.SetRiskManagement(
MaximumDrawdownPercentPortfolio(0.10)
)
# ------------------------------------------------------------
# 7. BENCHMARK
# ------------------------------------------------------------
self._benchmark_spy = self.AddEquity(
"SPY",
Resolution.Daily,
Market.USA
).Symbol
self._benchmark_agg = self.AddEquity(
"AGG",
Resolution.Daily,
Market.USA
).Symbol
self.SetBenchmark(self._benchmark_spy)
self.initial_spy_price = None
self.initial_agg_price = None
self.benchmark_spy_weight = 0.60
self.benchmark_agg_weight = 0.40
self.strategy_peak = self.initial_cash
self.benchmark_peak = self.initial_cash
def OnData(self, data):
# ------------------------------------------------------------
# 1. CHECK BENCHMARK DATA
# ------------------------------------------------------------
if self._benchmark_spy not in data or data[self._benchmark_spy] is None:
return
if self._benchmark_agg not in data or data[self._benchmark_agg] is None:
return
spy_price = self.Securities[self._benchmark_spy].Price
agg_price = self.Securities[self._benchmark_agg].Price
if spy_price <= 0 or agg_price <= 0:
return
if self.initial_spy_price is None:
self.initial_spy_price = spy_price
if self.initial_agg_price is None:
self.initial_agg_price = agg_price
# ------------------------------------------------------------
# 2. CUSTOM BENCHMARK VALUE
# ------------------------------------------------------------
benchmark_value = (
self.initial_cash
* (
self.benchmark_spy_weight
* spy_price
/ self.initial_spy_price
+
self.benchmark_agg_weight
* agg_price
/ self.initial_agg_price
)
)
# ------------------------------------------------------------
# 3. PLOTS
# ------------------------------------------------------------
self.Plot(
"Strategy Equity",
"Portfolio Value",
self.Portfolio.TotalPortfolioValue
)
self.Plot(
"Strategy Equity",
"Benchmark 60 pct SPY 40 pct AGG",
benchmark_value
)
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 DIAGNOSTICS
# ------------------------------------------------------------
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
)