| Overall Statistics |
|
Total Orders 168 Average Win 0.54% Average Loss -3.59% Compounding Annual Return -1.811% Drawdown 29.200% Expectancy -0.002 Start Equity 1000000 End Equity 926427.73 Net Profit -7.357% Sharpe Ratio -0.432 Sortino Ratio -0.549 Probabilistic Sharpe Ratio 0.495% Loss Rate 13% Win Rate 87% Profit-Loss Ratio 0.15 Alpha -0.061 Beta 0.113 Annual Standard Deviation 0.124 Annual Variance 0.015 Information Ratio -0.661 Tracking Error 0.179 Treynor Ratio -0.472 Total Fees $976.06 Estimated Strategy Capacity $170000000.00 Lowest Capacity Asset TLT SGNKIKYGE9NP Portfolio Turnover 1.33% Drawdown Recovery 0 |
#region imports
from AlgorithmImports import *
#endregion
"""
BLACK-LITTERMAN PORTFOLIO CONSTRUCTION WITH A MORE FLEXIBLE ALPHA MODEL
This strategy demonstrates a flexible Black-Litterman portfolio construction model
inside the QuantConnect Algorithm Framework.
The framework structure is preserved:
1. Universe Selection:
The model uses a manual ETF universe containing factor ETFs, SPY, and TLT.
2. Alpha Model:
The alpha model creates a composite technical score for every ETF. The score
combines four indicators:
A. Trend:
Price above the 50-day simple moving average.
B. MACD confirmation:
MACD line above the MACD signal line, using a faster 8 / 21 / 5 MACD.
C. RSI confirmation:
RSI above 42. This is more flexible than the standard 50 threshold and allows
recovering assets to qualify earlier.
D. Momentum:
21-day rate of change above zero.
Each positive condition adds one point. The composite score ranges from 0 to 4.
This flexible version requires only one positive condition for an ETF to receive
an Up insight. This gives the Black-Litterman optimizer a broader opportunity
set and avoids over-filtering.
The score controls the strength of the view. Higher scores receive higher
expected-return magnitudes and higher confidence. These values are passed into
Insight objects, which the Black-Litterman model uses as investor views.
3. Portfolio Construction:
The BlackLittermanOptimizationPortfolioConstructionModel blends equilibrium
market assumptions with the alpha model's views. The alpha model decides which
ETFs have positive views and how strong those views are. The Black-Litterman
model then converts those views into portfolio targets.
4. Execution:
The ImmediateExecutionModel submits orders when portfolio targets are created.
5. Risk Management:
The MaximumDrawdownPercentPortfolio model applies a portfolio-level trailing
drawdown control. This version uses a 35% trailing drawdown limit, so risk
management remains visible but does not dominate the strategy too early.
The benchmark is 60% SPY and 40% TLT.
"""
class FlexibleBlackLittermanAlphaModel(AlphaModel):
def __init__(
self,
sma_period=50,
rsi_period=14,
rsi_threshold=42,
roc_period=21,
macd_fast=8,
macd_slow=21,
macd_signal=5,
minimum_score=1,
insight_duration_days=25
):
self.sma_period = sma_period
self.rsi_period = rsi_period
self.rsi_threshold = rsi_threshold
self.roc_period = roc_period
self.macd_fast = macd_fast
self.macd_slow = macd_slow
self.macd_signal = macd_signal
self.minimum_score = minimum_score
self.insight_duration = timedelta(days=insight_duration_days)
self.sma_by_symbol = {}
self.rsi_by_symbol = {}
self.roc_by_symbol = {}
self.macd_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)
# Emit insights only once per month.
# This avoids repeated insight churn and prevents loop-like behavior.
if self.last_emit_month == current_month:
return insights
qualified_count = 0
total_score = 0
scored_count = 0
for symbol in list(self.sma_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
sma = self.sma_by_symbol[symbol]
rsi = self.rsi_by_symbol[symbol]
roc = self.roc_by_symbol[symbol]
macd = self.macd_by_symbol[symbol]
if not sma.IsReady:
continue
if not rsi.IsReady:
continue
if not roc.IsReady:
continue
if not macd.IsReady:
continue
# --------------------------------------------------------
# 1. COMPOSITE SCORE
# --------------------------------------------------------
score = 0
# Faster trend filter.
if price > sma.Current.Value:
score += 1
# Faster MACD confirmation.
macd_strength = macd.Current.Value - macd.Signal.Current.Value
if macd_strength > 0:
score += 1
# Relaxed RSI confirmation.
if rsi.Current.Value > self.rsi_threshold:
score += 1
# Shorter-term momentum confirmation.
roc_value = roc.Current.Value
if roc_value > 0:
score += 1
total_score += score
scored_count += 1
# --------------------------------------------------------
# 2. ONLY QUALIFIED ETFs RECEIVE INSIGHTS
# --------------------------------------------------------
if score < self.minimum_score:
continue
qualified_count += 1
# --------------------------------------------------------
# 3. VIEW INTENSITY FOR BLACK-LITTERMAN
# --------------------------------------------------------
# Score 1 = weak positive view
# Score 2 = moderate positive view
# Score 3 = strong positive view
# Score 4 = very strong positive view
momentum_component = max(0, roc_value) * 0.15
magnitude = (
0.006 * score
+ momentum_component
)
# Keep expected-return views bounded.
magnitude = min(magnitude, 0.07)
confidence = (
0.35
+ 0.10 * score
)
confidence = min(confidence, 0.80)
insights.append(
Insight.Price(
symbol,
self.insight_duration,
InsightDirection.Up,
magnitude,
confidence
)
)
self.last_emit_month = current_month
# ------------------------------------------------------------
# 4. ALPHA DIAGNOSTICS
# ------------------------------------------------------------
algorithm.Plot(
"Composite Alpha",
"Qualified Securities",
qualified_count
)
if scored_count > 0:
algorithm.Plot(
"Composite Alpha",
"Average Score",
total_score / scored_count
)
algorithm.Debug(
"Flexible Black-Litterman 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.sma_by_symbol:
self.sma_by_symbol[symbol] = algorithm.SMA(
symbol,
self.sma_period,
Resolution.Daily
)
self.rsi_by_symbol[symbol] = algorithm.RSI(
symbol,
self.rsi_period,
MovingAverageType.Wilders,
Resolution.Daily
)
self.roc_by_symbol[symbol] = algorithm.ROC(
symbol,
self.roc_period,
Resolution.Daily
)
self.macd_by_symbol[symbol] = algorithm.MACD(
symbol,
self.macd_fast,
self.macd_slow,
self.macd_signal,
MovingAverageType.Exponential,
Resolution.Daily
)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.sma_by_symbol:
del self.sma_by_symbol[symbol]
if symbol in self.rsi_by_symbol:
del self.rsi_by_symbol[symbol]
if symbol in self.roc_by_symbol:
del self.roc_by_symbol[symbol]
if symbol in self.macd_by_symbol:
del self.macd_by_symbol[symbol]
class EnergeticSkyBlueFrog(QCAlgorithm):
def Initialize(self):
# ------------------------------------------------------------
# 1. BACKTEST SETTINGS
# ------------------------------------------------------------
self.SetStartDate(2022, 3, 1)
self.SetEndDate(2026, 5, 5)
self.initial_cash = 1000000
self.SetCash(self.initial_cash)
# ------------------------------------------------------------
# 2. UNIVERSE SELECTION
# ------------------------------------------------------------
self.UniverseSettings.Resolution = Resolution.Daily
tickers = [
"USMV",
"DGRO",
"QUAL",
"DVY",
"MTUM",
"VLUE",
"TLT",
"SPY"
]
symbols = [
Symbol.Create(ticker, SecurityType.Equity, Market.USA)
for ticker in tickers
]
self.SetUniverseSelection(
ManualUniverseSelectionModel(symbols)
)
# Warm up enough data for the 50-day SMA and indicators.
self.SetWarmUp(80, Resolution.Daily)
# ------------------------------------------------------------
# 3. MORE FLEXIBLE BLACK-LITTERMAN ALPHA MODEL
# ------------------------------------------------------------
self.AddAlpha(
FlexibleBlackLittermanAlphaModel(
sma_period=50,
rsi_period=14,
rsi_threshold=42,
roc_period=21,
macd_fast=8,
macd_slow=21,
macd_signal=5,
minimum_score=1,
insight_duration_days=25
)
)
# ------------------------------------------------------------
# 4. BLACK-LITTERMAN PORTFOLIO CONSTRUCTION
# ------------------------------------------------------------
self.SetPortfolioConstruction(
BlackLittermanOptimizationPortfolioConstructionModel()
)
# ------------------------------------------------------------
# 5. EXECUTION MODEL
# ------------------------------------------------------------
self.SetExecution(
ImmediateExecutionModel()
)
# ------------------------------------------------------------
# 6. FLEXIBLE RISK MANAGEMENT MODEL
# ------------------------------------------------------------
self.risk_drawdown_limit = 0.35
self.SetRiskManagement(
MaximumDrawdownPercentPortfolio(
self.risk_drawdown_limit,
isTrailing=True
)
)
# ------------------------------------------------------------
# 7. BENCHMARK
# ------------------------------------------------------------
self._benchmark_spy = self.AddEquity(
"SPY",
Resolution.Daily,
Market.USA
).Symbol
self._benchmark_tlt = self.AddEquity(
"TLT",
Resolution.Daily,
Market.USA
).Symbol
self.SetBenchmark(self._benchmark_spy)
self.initial_spy_price = None
self.initial_tlt_price = None
self.benchmark_spy_weight = 0.60
self.benchmark_tlt_weight = 0.40
# ------------------------------------------------------------
# 8. DIAGNOSTIC STATE VARIABLES
# ------------------------------------------------------------
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_tlt not in data or data[self._benchmark_tlt] is None:
return
spy_price = self.Securities[self._benchmark_spy].Price
tlt_price = self.Securities[self._benchmark_tlt].Price
if spy_price <= 0 or tlt_price <= 0:
return
if self.initial_spy_price is None:
self.initial_spy_price = spy_price
if self.initial_tlt_price is None:
self.initial_tlt_price = tlt_price
# ------------------------------------------------------------
# 2. CUSTOM BENCHMARK VALUE
# ------------------------------------------------------------
benchmark_value = (
self.initial_cash
* (
self.benchmark_spy_weight
* spy_price
/ self.initial_spy_price
+
self.benchmark_tlt_weight
* tlt_price
/ self.initial_tlt_price
)
)
# ------------------------------------------------------------
# 3. STRATEGY VS BENCHMARK
# ------------------------------------------------------------
self.Plot(
"Strategy Equity",
"Portfolio Value",
self.Portfolio.TotalPortfolioValue
)
self.Plot(
"Strategy Equity",
"Benchmark 60 pct SPY 40 pct TLT",
benchmark_value
)
# ------------------------------------------------------------
# 4. PORTFOLIO STATE
# ------------------------------------------------------------
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
)
# ------------------------------------------------------------
# 5. 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
)
# ------------------------------------------------------------
# 6. RISK LIMIT VISUALIZATION
# ------------------------------------------------------------
self.Plot(
"Risk Management",
"Drawdown Limit",
-self.risk_drawdown_limit
)
risk_triggered_marker = 0
if strategy_drawdown <= -self.risk_drawdown_limit:
risk_triggered_marker = 1
self.Plot(
"Risk Management",
"Risk Triggered",
risk_triggered_marker
)