| Overall Statistics |
|
Total Orders 1473 Average Win 0.36% Average Loss -0.41% Compounding Annual Return 7.120% Drawdown 17.500% Expectancy 0.115 Start Equity 100000 End Equity 142233.52 Net Profit 42.234% Sharpe Ratio 0.162 Sortino Ratio 0.169 Probabilistic Sharpe Ratio 7.732% Loss Rate 41% Win Rate 59% Profit-Loss Ratio 0.88 Alpha 0.046 Beta -0.4 Annual Standard Deviation 0.115 Annual Variance 0.013 Information Ratio -0.225 Tracking Error 0.221 Treynor Ratio -0.046 Total Fees $0.00 Estimated Strategy Capacity $0 Lowest Capacity Asset NVDA RHM8UTD8DT2D Portfolio Turnover 3.86% Drawdown Recovery 779 |
from AlgorithmImports import *
class TechMomentumWithHedges(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 1, 1)
self.SetCash(100_000)
# ALPHA ASSETS
self.tech_stocks = ["NVDA", "AMD", "PLTR", "TSLA", "MSFT", "GOOGL", "META", "AVGO", "ARM", "SNOW"]
# HEDGE ASSETS
self.hedge_assets = {"GLD": 0.12, "TLT": 0.12, "XLU": 0.08, "VNQ": 0.04, "SQQQ": 0.14}
# INDICATORS
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_sma200 = self.SMA(self.spy, 200)
self.spy_mom = self.MOMP(self.spy, 21)
self.symbols = {}
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
self.symbols[ticker] = self.AddEquity(ticker, Resolution.Daily).Symbol
self.indicators = {}
for ticker in self.tech_stocks:
s = self.symbols[ticker]
self.indicators[ticker] = {
'rsi': self.RSI(s, 14),
'macd': self.MACD(s, 12, 26, 9),
'sma20': self.SMA(s, 20),
'sma50': self.SMA(s, 50),
'mom': self.MOMP(s, 21),
'std': self.STD(s, 20)
}
self.high_water_marks = {}
self.penalty_box = {} # Smart Re-entry Tracker
self.Schedule.On(self.DateRules.WeekStart(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 30), self.Rebalance)
self.Schedule.On(self.DateRules.EveryDay(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 60), self.RiskManagement)
self.SetWarmUp(200)
self.set_brokerage_model(BrokerageName.ALPACA)
@property
def is_market_bullish(self):
return self.Securities[self.spy].Price > self.spy_sma200.Current.Value
def CalculateMomentumScore(self, ticker):
if not self.indicators[ticker]['macd'].IsReady: return 0
ind = self.indicators[ticker]
price = self.Securities[self.symbols[ticker]].Price
score = 0
if price > ind['sma20'].Current.Value: score += 20
if ind['sma20'].Current.Value > ind['sma50'].Current.Value: score += 20
if ind['macd'].Current.Value > ind['macd'].Signal.Current.Value: score += 20
if 45 < ind['rsi'].Current.Value < 75: score += 20
if ind['mom'].Current.Value > self.spy_mom.Current.Value: score += 20
return score
def Rebalance(self):
if self.IsWarmingUp or not self.spy_sma200.IsReady: return
target_weights = {}
if not self.is_market_bullish:
self.Debug("BEAR MARKET: Scaling into 100% Hedges")
base_hedge_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / base_hedge_sum)
else:
# 1. Determine Volatility Regime
vols = [self.indicators[t]['std'].Current.Value / self.Securities[self.symbols[t]].Price
for t in self.tech_stocks if self.indicators[t]['std'].IsReady]
avg_vol = sum(vols) / len(vols) if vols else 0.02
t_ratio = 0.60 if avg_vol < 0.025 else (0.50 if avg_vol < 0.035 else 0.35)
h_ratio = 1.0 - t_ratio
# 2. Score and Filter Tech (The Penalty Box Logic)
tech_scores = {t: self.CalculateMomentumScore(t) for t in self.tech_stocks}
sorted_tech = [t for t, s in sorted(tech_scores.items(), key=lambda x: x[1], reverse=True) if s > 50]
valid_tech = []
for t in sorted_tech:
if t in self.penalty_box:
exit_data = self.penalty_box[t]
days_passed = (self.Time - exit_data['time']).days
if self.Securities[self.symbols[t]].Price > exit_data['price'] or days_passed > 21:
valid_tech.append(t)
self.penalty_box.pop(t, None)
else: continue # Skip if price hasn't recovered or time hasn't passed
else:
valid_tech.append(t)
top_tech = valid_tech[:6] # Select top 6 validated stocks
# 3. Assign Weights with Concentration Cap
if top_tech:
total_s = sum(tech_scores[t] for t in top_tech)
for t in top_tech:
target_weights[t] = min((tech_scores[t] / total_s) * t_ratio, t_ratio * 0.30)
# Rescale Tech Bucket
actual_t_sum = sum([v for k, v in target_weights.items() if k in self.tech_stocks])
if actual_t_sum > 0:
scale = t_ratio / actual_t_sum
for t in top_tech: target_weights[t] *= scale
# 4. Assign Hedge Weights
h_base_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / h_base_sum) * h_ratio
# Execute Trades
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
symbol = self.symbols[ticker]
weight = target_weights.get(ticker, 0)
if abs(self.Portfolio[symbol].HoldingsValue / self.Portfolio.TotalPortfolioValue - weight) > 0.01:
self.SetHoldings(symbol, weight)
def RiskManagement(self):
if self.IsWarmingUp: return
for ticker in self.tech_stocks:
symbol = self.symbols[ticker]
if not self.Portfolio[symbol].Invested:
self.high_water_marks.pop(ticker, None)
continue
price = self.Securities[symbol].Price
cost = self.Portfolio[symbol].AveragePrice
if ticker not in self.high_water_marks: self.high_water_marks[ticker] = price
self.high_water_marks[ticker] = max(self.high_water_marks[ticker], price)
peak = self.high_water_marks[ticker]
pnl, drawdown = (price - cost) / cost, (price - peak) / peak
# Exit Conditions
trigger_exit = False
if pnl > 0.30 and drawdown < -0.06: trigger_exit = True
elif pnl > 0.15 and drawdown < -0.10: trigger_exit = True
elif pnl < -0.08: trigger_exit = True
if trigger_exit:
self.Liquidate(symbol)
# Put in Penalty Box: Store exit price and time
self.penalty_box[ticker] = {'price': price, 'time': self.Time}
self.Debug(f"EXIT & PENALTY: {ticker} at {pnl:.1%}")from AlgorithmImports import *
class TechMomentumWithHedges(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 1, 1)
self.SetCash(100_000)
self.tech_stocks = ["NVDA", "AMD", "PLTR", "TSLA", "MSFT", "GOOGL", "META", "AVGO", "ARM", "SNOW"]
self.hedge_assets = {"GLD": 0.12, "TLT": 0.12, "XLU": 0.08, "VNQ": 0.04, "SQQQ": 0.14}
# Market Health Filter: 200-Day SMA
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_sma200 = self.SMA(self.spy, 200)
self.spy_mom = self.MOMP(self.spy, 21)
self.symbols = {}
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
self.symbols[ticker] = self.AddEquity(ticker, Resolution.Daily).Symbol
self.indicators = {}
for ticker in self.tech_stocks:
s = self.symbols[ticker]
self.indicators[ticker] = {
'rsi': self.RSI(s, 14),
'macd': self.MACD(s, 12, 26, 9),
'sma20': self.SMA(s, 20),
'sma50': self.SMA(s, 50),
'mom': self.MOMP(s, 21),
'std': self.STD(s, 20)
}
self.Schedule.On(self.DateRules.WeekStart(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 30), self.Rebalance)
self.Schedule.On(self.DateRules.EveryDay(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 60), self.RiskManagement)
self.SetWarmUp(200) # Increased for SMA200
self.set_brokerage_model(BrokerageName.ALPACA)
@property
def is_market_bullish(self):
"""Returns True if SPY is above its 200-day trend line."""
return self.Securities[self.spy].Price > self.spy_sma200.Current.Value
def CalculateMomentumScore(self, ticker):
if not self.indicators[ticker]['macd'].IsReady: return 0
price = self.Securities[self.symbols[ticker]].Price
ind = self.indicators[ticker]
score = 0
# Momentum & Trend Logic
if price > ind['sma20'].Current.Value: score += 20
if ind['sma20'].Current.Value > ind['sma50'].Current.Value: score += 20
if ind['macd'].Current.Value > ind['macd'].Signal.Current.Value: score += 20
if 45 < ind['rsi'].Current.Value < 75: score += 20
if ind['mom'].Current.Value > self.spy_mom.Current.Value: score += 20
return score
def Rebalance(self):
if self.IsWarmingUp or not self.spy_sma200.IsReady: return
target_weights = {}
# 1. CHECK MARKET STRESS FILTER
if not self.is_market_bullish:
self.Debug("Market Stress Detected: Moving to 100% Hedges")
# Liquidate Tech
for ticker in self.tech_stocks:
if self.Portfolio[self.symbols[ticker]].Invested:
self.Liquidate(self.symbols[ticker])
# Reallocate Hedges to fill 100%
base_hedge_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / base_hedge_sum)
else:
# 2. NORMAL DYNAMIC ALLOCATION
tech_vols = [self.indicators[t]['std'].Current.Value / self.Securities[self.symbols[t]].Price
for t in self.tech_stocks if self.indicators[t]['std'].IsReady]
avg_vol = sum(tech_vols) / len(tech_vols) if tech_vols else 0.02
# Decide Bucket Ratios
if avg_vol > 0.035: tech_ratio, hedge_ratio = 0.35, 0.65
elif avg_vol > 0.025: tech_ratio, hedge_ratio = 0.50, 0.50
else: tech_ratio, hedge_ratio = 0.60, 0.40
# Tech Weights
tech_scores = {t: self.CalculateMomentumScore(t) for t in self.tech_stocks}
top_tech = [t for t, s in sorted(tech_scores.items(), key=lambda x: x[1], reverse=True)[:6] if s > 50]
if top_tech:
total_s = sum(tech_scores[t] for t in top_tech)
for t in top_tech:
target_weights[t] = min((tech_scores[t] / total_s) * tech_ratio, 0.15)
# Re-scale to tech_ratio
scale = tech_ratio / sum([v for k, v in target_weights.items() if k in self.tech_stocks])
for t in top_tech: target_weights[t] *= scale
# Hedge Weights
base_hedge_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / base_hedge_sum) * hedge_ratio
# 3. EXECUTE
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
symbol = self.symbols[ticker]
weight = target_weights.get(ticker, 0)
if abs(self.Portfolio[symbol].HoldingsValue / self.Portfolio.TotalPortfolioValue - weight) > 0.01:
self.SetHoldings(symbol, weight)
def RiskManagement(self):
if self.IsWarmingUp: return
for ticker in self.tech_stocks:
symbol = self.symbols[ticker]
if not self.Portfolio[symbol].Invested: continue
pnl = (self.Securities[symbol].Price - self.Portfolio[symbol].AveragePrice) / self.Portfolio[symbol].AveragePrice
# Fixed Bug: Ensure quantity > 0 for partial profit
if pnl > 0.30:
qty_to_sell = int(self.Portfolio[symbol].Quantity * 0.5)
if qty_to_sell > 0:
self.MarketOrder(symbol, -qty_to_sell)
self.Debug(f"Harvested 50% profit on {ticker}")
elif pnl < -0.08:
self.Liquidate(symbol)
self.Debug(f"Stop-loss triggered for {ticker}")from AlgorithmImports import *
class TechMomentumWithHedges(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 1, 1)
self.SetCash(100_000)
# ALPHA ASSETS
self.tech_stocks = ["NVDA", "AMD", "PLTR", "TSLA", "MSFT", "GOOGL", "META", "AVGO", "ARM", "SNOW"]
# HEDGE ASSETS
self.hedge_assets = {"GLD": 0.12, "TLT": 0.12, "XLU": 0.08, "VNQ": 0.04, "SQQQ": 0.14}
# INDICATORS
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_sma200 = self.SMA(self.spy, 200)
self.spy_mom = self.MOMP(self.spy, 21)
self.symbols = {}
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
self.symbols[ticker] = self.AddEquity(ticker, Resolution.Daily).Symbol
self.indicators = {}
for ticker in self.tech_stocks:
s = self.symbols[ticker]
self.indicators[ticker] = {
'rsi': self.RSI(s, 14),
'macd': self.MACD(s, 12, 26, 9),
'sma20': self.SMA(s, 20),
'sma50': self.SMA(s, 50),
'mom': self.MOMP(s, 21),
'std': self.STD(s, 20)
}
self.high_water_marks = {}
self.Schedule.On(self.DateRules.WeekStart(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 30), self.Rebalance)
self.Schedule.On(self.DateRules.EveryDay(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 60), self.RiskManagement)
self.SetWarmUp(200)
self.set_brokerage_model(BrokerageName.ALPACA)
@property
def is_market_bullish(self):
return self.Securities[self.spy].Price > self.spy_sma200.Current.Value
def CalculateMomentumScore(self, ticker):
if not self.indicators[ticker]['macd'].IsReady: return 0
ind = self.indicators[ticker]
price = self.Securities[self.symbols[ticker]].Price
score = 0
if price > ind['sma20'].Current.Value: score += 20
if ind['sma20'].Current.Value > ind['sma50'].Current.Value: score += 20
if ind['macd'].Current.Value > ind['macd'].Signal.Current.Value: score += 20
if 45 < ind['rsi'].Current.Value < 75: score += 20
if ind['mom'].Current.Value > self.spy_mom.Current.Value: score += 20
return score
def Rebalance(self):
if self.IsWarmingUp or not self.spy_sma200.IsReady: return
target_weights = {}
if not self.is_market_bullish:
self.Debug("BEAR MARKET: Scaling into 100% Hedges")
base_hedge_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / base_hedge_sum)
else:
# 1. Vol-Adjusted Bucket Split
vols = [self.indicators[t]['std'].Current.Value / self.Securities[self.symbols[t]].Price
for t in self.tech_stocks if self.indicators[t]['std'].IsReady]
avg_vol = sum(vols) / len(vols) if vols else 0.02
if avg_vol > 0.035: t_ratio, h_ratio = 0.35, 0.65
elif avg_vol > 0.025: t_ratio, h_ratio = 0.50, 0.50
else: t_ratio, h_ratio = 0.60, 0.40
# 2. Tech Scoring & Diversification Cap
tech_scores = {t: self.CalculateMomentumScore(t) for t in self.tech_stocks}
top_tech = [t for t, s in sorted(tech_scores.items(), key=lambda x: x[1], reverse=True)[:6] if s > 50]
if top_tech:
total_s = sum(tech_scores[t] for t in top_tech)
for t in top_tech:
# Weight based on score, but capped at 30% of the TECH bucket (which is t_ratio * 0.3)
score_weight = (tech_scores[t] / total_s) * t_ratio
target_weights[t] = min(score_weight, t_ratio * 0.30)
# Re-normalize to ensure we hit the full t_ratio exactly
actual_t_sum = sum([v for k, v in target_weights.items() if k in self.tech_stocks])
if actual_t_sum > 0:
scale = t_ratio / actual_t_sum
for t in top_tech: target_weights[t] *= scale
# 3. Hedge Allocation
h_base_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / h_base_sum) * h_ratio
# Execute trades
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
symbol = self.symbols[ticker]
weight = target_weights.get(ticker, 0)
if abs(self.Portfolio[symbol].HoldingsValue / self.Portfolio.TotalPortfolioValue - weight) > 0.01:
self.SetHoldings(symbol, weight)
def RiskManagement(self):
if self.IsWarmingUp: return
for ticker in self.tech_stocks:
symbol = self.symbols[ticker]
if not self.Portfolio[symbol].Invested:
self.high_water_marks.pop(ticker, None)
continue
price = self.Securities[symbol].Price
cost = self.Portfolio[symbol].AveragePrice
# Update High Water Mark
if ticker not in self.high_water_marks:
self.high_water_marks[ticker] = price
self.high_water_marks[ticker] = max(self.high_water_marks[ticker], price)
peak = self.high_water_marks[ticker]
pnl_pct = (price - cost) / cost
drawdown_from_peak = (price - peak) / peak
# Aggressive Trail (>30% gain -> 6% drop exit)
if pnl_pct > 0.30:
if drawdown_from_peak < -0.06:
self.Liquidate(symbol)
self.Debug(f"PROFIT LOCK: {ticker} exit at {pnl_pct:.1%}")
# Standard Trail (>15% gain -> 10% drop exit)
elif pnl_pct > 0.15:
if drawdown_from_peak < -0.10:
self.Liquidate(symbol)
self.Debug(f"TRAIL STOP: {ticker} exit at {pnl_pct:.1%}")
# Safety Stop
elif pnl_pct < -0.08:
self.Liquidate(symbol)
self.Debug(f"HARD STOP: {ticker} exit at {pnl_pct:.1%}")from AlgorithmImports import *
class TechMomentumWithHedges(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 1, 1)
self.SetCash(100_000)
# ALPHA ASSETS
self.tech_stocks = ["NVDA", "AMD", "PLTR", "TSLA", "MSFT", "GOOGL", "META", "AVGO", "ARM", "SNOW"]
# HEDGE ASSETS
self.hedge_assets = {"GLD": 0.12, "TLT": 0.12, "XLU": 0.08, "VNQ": 0.04, "SQQQ": 0.14}
# INDICATORS
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_sma200 = self.SMA(self.spy, 200)
self.spy_mom = self.MOMP(self.spy, 21)
self.symbols = {}
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
self.symbols[ticker] = self.AddEquity(ticker, Resolution.Daily).Symbol
self.indicators = {}
for ticker in self.tech_stocks:
s = self.symbols[ticker]
self.indicators[ticker] = {
'rsi': self.RSI(s, 14),
'macd': self.MACD(s, 12, 26, 9),
'sma20': self.SMA(s, 20),
'sma50': self.SMA(s, 50),
'mom': self.MOMP(s, 21),
'std': self.STD(s, 20)
}
self.high_water_marks = {}
self.penalty_box = {} # Smart Re-entry Tracker
self.Schedule.On(self.DateRules.WeekStart(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 30), self.Rebalance)
self.Schedule.On(self.DateRules.EveryDay(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 60), self.RiskManagement)
self.SetWarmUp(200)
self.set_brokerage_model(BrokerageName.ALPACA)
@property
def is_market_bullish(self):
return self.Securities[self.spy].Price > self.spy_sma200.Current.Value
def CalculateMomentumScore(self, ticker):
if not self.indicators[ticker]['macd'].IsReady: return 0
ind = self.indicators[ticker]
price = self.Securities[self.symbols[ticker]].Price
score = 0
if price > ind['sma20'].Current.Value: score += 20
if ind['sma20'].Current.Value > ind['sma50'].Current.Value: score += 20
if ind['macd'].Current.Value > ind['macd'].Signal.Current.Value: score += 20
if 45 < ind['rsi'].Current.Value < 75: score += 20
if ind['mom'].Current.Value > self.spy_mom.Current.Value: score += 20
return score
def Rebalance(self):
if self.IsWarmingUp or not self.spy_sma200.IsReady: return
target_weights = {}
if not self.is_market_bullish:
self.Debug("BEAR MARKET: Scaling into 100% Hedges")
base_hedge_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / base_hedge_sum)
else:
# 1. Determine Volatility Regime
vols = [self.indicators[t]['std'].Current.Value / self.Securities[self.symbols[t]].Price
for t in self.tech_stocks if self.indicators[t]['std'].IsReady]
avg_vol = sum(vols) / len(vols) if vols else 0.02
t_ratio = 0.60 if avg_vol < 0.025 else (0.50 if avg_vol < 0.035 else 0.35)
h_ratio = 1.0 - t_ratio
# 2. Score and Filter Tech (The Penalty Box Logic)
tech_scores = {t: self.CalculateMomentumScore(t) for t in self.tech_stocks}
sorted_tech = [t for t, s in sorted(tech_scores.items(), key=lambda x: x[1], reverse=True) if s > 50]
valid_tech = []
for t in sorted_tech:
if t in self.penalty_box:
exit_data = self.penalty_box[t]
days_passed = (self.Time - exit_data['time']).days
if self.Securities[self.symbols[t]].Price > exit_data['price'] or days_passed > 21:
valid_tech.append(t)
self.penalty_box.pop(t, None)
else: continue # Skip if price hasn't recovered or time hasn't passed
else:
valid_tech.append(t)
top_tech = valid_tech[:6] # Select top 6 validated stocks
# 3. Assign Weights with Concentration Cap
if top_tech:
total_s = sum(tech_scores[t] for t in top_tech)
for t in top_tech:
target_weights[t] = min((tech_scores[t] / total_s) * t_ratio, t_ratio * 0.30)
# Rescale Tech Bucket
actual_t_sum = sum([v for k, v in target_weights.items() if k in self.tech_stocks])
if actual_t_sum > 0:
scale = t_ratio / actual_t_sum
for t in top_tech: target_weights[t] *= scale
# 4. Assign Hedge Weights
h_base_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / h_base_sum) * h_ratio
# Execute Trades
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
symbol = self.symbols[ticker]
weight = target_weights.get(ticker, 0)
if abs(self.Portfolio[symbol].HoldingsValue / self.Portfolio.TotalPortfolioValue - weight) > 0.01:
self.SetHoldings(symbol, weight)
def RiskManagement(self):
if self.IsWarmingUp: return
for ticker in self.tech_stocks:
symbol = self.symbols[ticker]
if not self.Portfolio[symbol].Invested:
self.high_water_marks.pop(ticker, None)
continue
price = self.Securities[symbol].Price
cost = self.Portfolio[symbol].AveragePrice
if ticker not in self.high_water_marks: self.high_water_marks[ticker] = price
self.high_water_marks[ticker] = max(self.high_water_marks[ticker], price)
peak = self.high_water_marks[ticker]
pnl, drawdown = (price - cost) / cost, (price - peak) / peak
# Exit Conditions
trigger_exit = False
if pnl > 0.30 and drawdown < -0.06: trigger_exit = True
elif pnl > 0.15 and drawdown < -0.10: trigger_exit = True
elif pnl < -0.08: trigger_exit = True
if trigger_exit:
self.Liquidate(symbol)
# Put in Penalty Box: Store exit price and time
self.penalty_box[ticker] = {'price': price, 'time': self.Time}
self.Debug(f"EXIT & PENALTY: {ticker} at {pnl:.1%}")from AlgorithmImports import *
class TechMomentumWithHedges(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 1, 1)
self.SetCash(100_000)
# ALPHA ASSETS
self.tech_stocks = ["NVDA", "AMD", "PLTR", "TSLA", "MSFT", "GOOGL", "META", "AVGO", "ARM", "SNOW"]
# HEDGE ASSETS (Base weights sum to 0.50)
self.hedge_assets = {"GLD": 0.12, "TLT": 0.12, "XLU": 0.08, "VNQ": 0.04, "SQQQ": 0.14}
# INDICATORS
self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
self.spy_sma200 = self.SMA(self.spy, 200)
self.spy_mom = self.MOMP(self.spy, 21)
self.symbols = {}
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
self.symbols[ticker] = self.AddEquity(ticker, Resolution.Daily).Symbol
self.indicators = {}
for ticker in self.tech_stocks:
s = self.symbols[ticker]
self.indicators[ticker] = {
'rsi': self.RSI(s, 14),
'macd': self.MACD(s, 12, 26, 9),
'sma20': self.SMA(s, 20),
'sma50': self.SMA(s, 50),
'mom': self.MOMP(s, 21),
'std': self.STD(s, 20)
}
# Track peak prices for trailing stops
self.high_water_marks = {}
self.Schedule.On(self.DateRules.WeekStart(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 30), self.Rebalance)
self.Schedule.On(self.DateRules.EveryDay(self.spy), self.TimeRules.AfterMarketOpen(self.spy, 60), self.RiskManagement)
self.SetWarmUp(200)
self.set_brokerage_model(BrokerageName.ALPACA)
@property
def is_market_bullish(self):
return self.Securities[self.spy].Price > self.spy_sma200.Current.Value
def CalculateMomentumScore(self, ticker):
if not self.indicators[ticker]['macd'].IsReady: return 0
ind = self.indicators[ticker]
price = self.Securities[self.symbols[ticker]].Price
score = 0
if price > ind['sma20'].Current.Value: score += 20
if ind['sma20'].Current.Value > ind['sma50'].Current.Value: score += 20
if ind['macd'].Current.Value > ind['macd'].Signal.Current.Value: score += 20
if 45 < ind['rsi'].Current.Value < 75: score += 20
if ind['mom'].Current.Value > self.spy_mom.Current.Value: score += 20
return score
def Rebalance(self):
if self.IsWarmingUp or not self.spy_sma200.IsReady: return
target_weights = {}
if not self.is_market_bullish:
self.Debug("BEAR MARKET FILTER ACTIVE: High Cash/Hedge Position")
base_hedge_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / base_hedge_sum)
else:
# Determine Volatility Regime
vols = [self.indicators[t]['std'].Current.Value / self.Securities[self.symbols[t]].Price
for t in self.tech_stocks if self.indicators[t]['std'].IsReady]
avg_vol = sum(vols) / len(vols) if vols else 0.02
if avg_vol > 0.035: t_ratio, h_ratio = 0.35, 0.65
elif avg_vol > 0.025: t_ratio, h_ratio = 0.50, 0.50
else: t_ratio, h_ratio = 0.60, 0.40
# Rank Tech
tech_scores = {t: self.CalculateMomentumScore(t) for t in self.tech_stocks}
top_tech = [t for t, s in sorted(tech_scores.items(), key=lambda x: x[1], reverse=True)[:6] if s > 50]
if top_tech:
total_s = sum(tech_scores[t] for t in top_tech)
for t in top_tech:
target_weights[t] = min((tech_scores[t] / total_s) * t_ratio, 0.15)
# Normalize tech bucket
actual_t_sum = sum([v for k, v in target_weights.items() if k in self.tech_stocks])
scale = t_ratio / actual_t_sum
for t in top_tech: target_weights[t] *= scale
# Fill Hedge bucket
h_base_sum = sum(self.hedge_assets.values())
for ticker, base_w in self.hedge_assets.items():
target_weights[ticker] = (base_w / h_base_sum) * h_ratio
# Execute Trades
for ticker in self.tech_stocks + list(self.hedge_assets.keys()):
symbol = self.symbols[ticker]
weight = target_weights.get(ticker, 0)
current_w = self.Portfolio[symbol].HoldingsValue / self.Portfolio.TotalPortfolioValue
if abs(current_w - weight) > 0.01:
self.SetHoldings(symbol, weight)
def RiskManagement(self):
if self.IsWarmingUp: return
for ticker in self.tech_stocks:
symbol = self.symbols[ticker]
if not self.Portfolio[symbol].Invested:
self.high_water_marks.pop(ticker, None)
continue
price = self.Securities[symbol].Price
cost = self.Portfolio[symbol].AveragePrice
# Update High Water Mark
if ticker not in self.high_water_marks:
self.high_water_marks[ticker] = price
self.high_water_marks[ticker] = max(self.high_water_marks[ticker], price)
peak = self.high_water_marks[ticker]
pnl_pct = (price - cost) / cost
drawdown_from_peak = (price - peak) / peak
# 1. TIGHT TRAILING STOP (Tier 2: Gains > 30%)
if pnl_pct > 0.30:
if drawdown_from_peak < -0.06:
self.Liquidate(symbol)
self.Debug(f"TRIGHT TRAIL EXIT: {ticker} at {pnl_pct:.2%}")
continue
# 2. STANDARD TRAILING STOP (Tier 1: Gains > 15%)
elif pnl_pct > 0.15:
if drawdown_from_peak < -0.10:
self.Liquidate(symbol)
self.Debug(f"STD TRAIL EXIT: {ticker} at {pnl_pct:.2%}")
continue
# 3. INITIAL STOP LOSS
if pnl_pct < -0.08:
self.Liquidate(symbol)
self.Debug(f"HARD STOP EXIT: {ticker} at {pnl_pct:.2%}")