# region imports
from AlgorithmImports import *
# endregion
# ===========================================================================
# MULTI-ALPHA FRAMEWORK - STATUS REPORT
# ===========================================================================
# Projekt: 25806232 - Multi-Alpha Framework Defense+Seasonal
# Datum: 2025-10-23
# Session: Fortsetzung (Context-Limit)
# ===========================================================================
"""
PROJEKTZIEL:
Kombination von Defense (60%) + Seasonal (40%) Strategien im Framework
ERWARTET (Original Non-Framework):
- Profit: 475%, DD: 20.3%, Sharpe: 0.907, Trades: 1,183
BESTE VERSION: v2 (35ca7aca)
- Profit: 44.8%, DD: 21.7% ✅, Sharpe: 0.043, Trades: 806
AKTUELL: v4 (c1cbd4194)
- Profit: 73%, DD: 41.5%, Sharpe: 0.164, Trades: 25,986 ❌
HAUPTPROBLEM: 20x zu viele Trades!
NÄCHSTE TESTS:
1. Insight Period = Forever (statt 30 Tage)
2. Simplified PCM (ohne Per-Alpha Tracking)
3. Call-Frequency Logging
4. Separate Alpha Tests
ALTERNATIVEN:
- Option B: Hybrid (Alpha Models + Manual Scheduling)
- Option C: Original Code verwenden (funktioniert perfekt)
DETAILS: Siehe Docstring unten
"""
pass
from AlgorithmImports import *
from datetime import timedelta
from models.universe.multi_alpha_universe_model import MultiAlphaUniverseSelectionModel
from models.alpha.defense_alpha_model import DefenseAlphaModel
from models.alpha.seasonal_alpha_model import SeasonalAlphaModel
from models.alpha.intraday_momentum_alpha_model import IntradayMomentumAlphaModel
from models.portfolio.multi_alpha_pcm import MultiAlphaPortfolioConstructionModel
from models.risk.multi_alpha_risk_model import MultiAlphaRiskManagementModel
from models.execution.tagged_execution_model import TaggedImmediateExecutionModel
class MultiAlphaFramework(QCAlgorithm):
"""
Multi-Alpha Framework: Defense (40%) + Seasonal (40%) + Intraday (20%)
Architecture:
- Tag-based insights (no weight in alphas)
- Forever Period for Defense + Seasonal (explicit FLAT on exits)
- Short Period (30 min) for Intraday + EOD liquidation
- PCM handles all allocation (40/40/20 budget + rank/equal/volatility logic)
- Simple risk filters (leverage, position size, drawdown)
Backtest: 2015-2025 (10 years)
Target Trades: ~2,500 (vs original 1,183)
"""
def Initialize(self) -> None:
# Backtest Period - MATCHING ORIGINAL (2015-2025)
self.SetStartDate(2015, 1, 2)
self.SetEndDate(2025, 5, 16)
self.SetCash(100_000)
# Track which Alpha owns each symbol (for Capital Allocation Chart)
# Updated in OnOrderEvent, read in OnData - much more efficient!
self.symbol_to_alpha = {} # {Symbol: "Defense" or "Seasonal"}
# Brokerage
self.SetBrokerageModel(
BrokerageName.InteractiveBrokersBrokerage,
AccountType.Margin
)
# Warm-up (Defense needs 252 days, Intraday needs 90 days)
self.SetWarmUp(timedelta(days=260))
self.Debug("=== Multi-Alpha Framework Initialization ===")
# CRITICAL FIX: Defense assets need MINUTE resolution for Update() to run at 14:40!
# With Daily resolution, Update() only runs once per day at midnight
defense_symbols = ["TLT", "GLD", "DBC", "UUP", "UPRO"]
for symbol_str in defense_symbols:
security = self.AddEquity(symbol_str, Resolution.Minute)
security.SetDataNormalizationMode(DataNormalizationMode.Adjusted)
self.Debug(f" Added {symbol_str} with MINUTE resolution for Defense timing")
# Universe Selection (for Seasonal assets - Daily is fine)
universe = MultiAlphaUniverseSelectionModel()
self.SetUniverseSelection(universe)
# SPY: Special handling (Minute Resolution for Seasonal)
spy_symbol = universe.add_spy_with_minute_resolution(self)
self.Debug(" Universe: Defense (5 Minute) + SPY (Minute) + Seasonal (20 Daily)")
# Alpha Models (NO weights - tags only!)
defense_alpha = DefenseAlphaModel()
seasonal_alpha = SeasonalAlphaModel()
# TEMPORARILY DISABLED FOR TESTING
# intraday_alpha = IntradayMomentumAlphaModel(
# period=90,
# min_volatility=0.008,
# consolidation_period=30
# )
# Composite Alpha Model - WITHOUT INTRADAY FOR NOW
self.SetAlpha(CompositeAlphaModel(defense_alpha, seasonal_alpha))
self.Debug(" Alpha Models:")
self.Debug(" - DefenseAlphaModel (60% budget) - MATCHING ORIGINAL")
self.Debug(" - SeasonalAlphaModel (40% budget) - MATCHING ORIGINAL")
self.Debug(" - IntradayMomentumAlphaModel (0% budget) - DISABLED FOR COMPARISON")
# Portfolio Construction Model (60/40/0 budget allocation - matching original)
alpha_budgets = {
"DefenseAlphaModel": 0.6,
"SeasonalAlphaModel": 0.4,
"IntradayMomentumAlphaModel": 0.0
}
pcm = MultiAlphaPortfolioConstructionModel(alpha_budgets=alpha_budgets)
self.SetPortfolioConstruction(pcm)
self.Debug(" PCM Budget: 60% Defense / 40% Seasonal / 0% Intraday (Original allocation)")
# Risk Management Model
risk = MultiAlphaRiskManagementModel(
max_total_leverage=2.0,
max_single_position=0.25,
max_drawdown=0.30
)
self.SetRiskManagement(risk)
self.Debug(" Risk: Max Leverage 2.0x, Max Position 25%, Max DD 30%")
# Execution Model with Order Tags
self.SetExecution(TaggedImmediateExecutionModel())
self.Debug(" Execution: Tagged Immediate (preserves Alpha source in order tags)")
# Schedule EOD Liquidation for Intraday positions
# Important: Only liquidate SPY if NOT in Seasonal hold period (Oct 1 - Jan 31)
self.Schedule.On(
self.DateRules.EveryDay("SPY"),
self.TimeRules.BeforeMarketClose("SPY", 1),
self._close_intraday_positions
)
self.Debug(" Scheduled: EOD Liquidation (Intraday only, respects Seasonal hold period)")
# Chart: Capital Allocation by Alpha Strategy
capital_chart = Chart("Capital Allocation")
capital_chart.AddSeries(Series("Defense", SeriesType.Line, "%", Color.Blue))
capital_chart.AddSeries(Series("Seasonal", SeriesType.Line, "%", Color.Green))
capital_chart.AddSeries(Series("Total", SeriesType.Line, "%", Color.White))
self.AddChart(capital_chart)
self.Debug(" Charts: Capital Allocation (Defense/Seasonal/Total)")
self.Debug("=== Framework Initialization Complete ===")
def _close_intraday_positions(self) -> None:
"""
End-of-day liquidation for Intraday positions
CRITICAL: Only close SPY if NOT in Seasonal hold period (Oct 1 - Jan 31)
- During Oct-Jan: Seasonal owns SPY, don't liquidate
- Outside Oct-Jan: Intraday can be liquidated
"""
if self.IsWarmingUp:
return
# Check if SPY is in Seasonal hold period
if self._is_spy_seasonal_active():
# Seasonal period active - don't touch SPY
self.Debug("[EOD] SPY in Seasonal hold period (Oct-Jan), skipping liquidation")
return
# Outside seasonal period - liquidate SPY if held
if self.Portfolio["SPY"].Invested:
self.Liquidate("SPY")
self.Debug("[EOD] Liquidated Intraday SPY position")
def _is_spy_seasonal_active(self) -> bool:
"""
Check if SPY seasonal hold period is active (Oct 1 - Jan 31)
"""
month = self.Time.month
return month >= 10 or month == 1
def OnOrderEvent(self, orderEvent: OrderEvent) -> None:
"""
Log order events with timing information and track Alpha ownership
"""
if orderEvent.Status == OrderStatus.Filled:
# Get the order to check its tag
order = self.Transactions.GetOrderById(orderEvent.OrderId)
if order and order.Tag:
# Cache which Alpha owns this symbol
self.symbol_to_alpha[orderEvent.Symbol] = order.Tag
self.Debug(f"[TIMING|Order] {self.Time}: Order FILLED - {orderEvent.Symbol.Value}, "
f"Qty: {orderEvent.FillQuantity}, Tag: {order.Tag if order else 'N/A'}")
def OnEndOfDay(self, symbol: Symbol) -> None:
"""
Track capital allocation by Alpha strategy - ONLY ONCE PER DAY
Called at end of each trading day (much more efficient than OnData)
"""
if self.IsWarmingUp:
return
# Only process for SPY (one call per day instead of 390!)
if symbol.Value != "SPY":
return
# Calculate capital allocation by Alpha source (using cached tags)
portfolio_value = self.Portfolio.TotalPortfolioValue
if portfolio_value <= 0:
return
defense_value = 0.0
seasonal_value = 0.0
# Iterate through all holdings - use cached Alpha ownership
for sym, holding in self.Portfolio.items():
if not holding.Invested:
continue
# Use cached tag from OnOrderEvent (MUCH faster than searching orders!)
alpha_tag = self.symbol_to_alpha.get(sym, "Unknown")
if alpha_tag == "Defense":
defense_value += holding.HoldingsValue
elif alpha_tag == "Seasonal":
seasonal_value += holding.HoldingsValue
# Calculate percentages
defense_pct = defense_value / portfolio_value
seasonal_pct = seasonal_value / portfolio_value
total_invested_pct = (defense_value + seasonal_value) / portfolio_value
# Plot to chart (daily)
self.Plot("Capital Allocation", "Defense", defense_pct)
self.Plot("Capital Allocation", "Seasonal", seasonal_pct)
self.Plot("Capital Allocation", "Total", total_invested_pct)
# Log every Monday
if self.Time.weekday() == 0:
positions = len([h for h in self.Portfolio.Values if h.Invested])
self.Debug(f"[{self.Time.strftime('%Y-%m-%d')}] Portfolio: ${portfolio_value:,.0f}, "
f"Defense: {defense_pct:.1%}, Seasonal: {seasonal_pct:.1%}, "
f"Total: {total_invested_pct:.1%}, Positions: {positions}")
def OnEndOfAlgorithm(self) -> None:
"""
Summary at end of backtest
"""
portfolio_value = self.Portfolio.TotalPortfolioValue
starting_capital = 100_000
total_return = (portfolio_value - starting_capital) / starting_capital
# Count total trades
total_trades = len([order for order in self.Transactions.GetOrders() if order.Status == OrderStatus.Filled])
self.Debug("=== Backtest Complete ===")
self.Debug(f" Final Portfolio Value: ${portfolio_value:,.2f}")
self.Debug(f" Total Return: {total_return:.2%}")
self.Debug(f" Total Trades: {total_trades}")
# Statistics will be shown in QuantConnect UI
self.Debug("===================")
# region imports
from AlgorithmImports import *
# endregion
# Alpha Models Module
from .defense_alpha_model import DefenseAlphaModel
from .seasonal_alpha_model import SeasonalAlphaModel
from .intraday_momentum_alpha_model import IntradayMomentumAlphaModel
__all__ = [
'DefenseAlphaModel',
'SeasonalAlphaModel',
'IntradayMomentumAlphaModel'
]
from AlgorithmImports import *
from datetime import timedelta
from typing import List, Dict, Set
class DefenseAlphaModel(AlphaModel):
"""
Defense-First Alpha Model with UPRO
Strategy:
- Ranks defensive assets (TLT, GLD, DBC, UUP) by multi-period momentum
- Top 4 get ranks 1-4 (allocated 40%, 30%, 20%, 10% by PCM via tag parsing)
- If momentum < risk-free rate, that rank's weight goes to UPRO instead
- Rebalances monthly on first trading day
CRITICAL FIX (Daily Consolidator):
- Uses Daily consolidator to ensure history windows receive DAILY bars
- Works regardless of subscription resolution (Minute or Daily)
- Each AlphaModel manages its own consolidators independently
"""
def __init__(self):
self.defensive_symbols = ["TLT", "GLD", "DBC", "UUP"]
self.upro_symbol = "UPRO"
self.symbols = []
self.momentum_periods = [21, 63, 126, 252]
self.history_windows = {}
self.last_rebalance_month = -1
self.risk_free_monthly = 0.045 / 12 # ~4.5% annual
self.defensive_weights = [0.4, 0.3, 0.2, 0.1] # For info only, PCM uses ranks
# Track current allocation to emit FLAT insights on exit
self.current_allocation: Set[Symbol] = set()
# Track consolidators for cleanup
self.consolidators = {}
# Debug tracking
self.update_call_count = 0
self.last_debug_date = None
def Update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]:
if algorithm.IsWarmingUp:
return []
# Debug: Track Update() calls (once per day)
current_date = algorithm.Time.date()
if current_date != self.last_debug_date:
self.update_call_count += 1
self.last_debug_date = current_date
# Log Update() frequency
if self.update_call_count <= 5 or self.update_call_count % 20 == 0:
algorithm.Debug(f"[DefenseAlpha|DEBUG] Update() called {self.update_call_count} times total, date={current_date}")
# Rebalance on first trading day of month
current_month = algorithm.Time.month
if current_month != self.last_rebalance_month:
self.last_rebalance_month = current_month
# Debug: Check window state before rebalancing
self._debug_window_state(algorithm)
return self._generate_insights(algorithm)
return []
def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
for added in changes.AddedSecurities:
symbol = added.Symbol
# Only track Defense symbols
if symbol.Value not in self.defensive_symbols and symbol.Value != self.upro_symbol:
continue
self.symbols.append(symbol)
# Create rolling window for momentum calculation (253 days = max period + 1)
self.history_windows[symbol] = RollingWindow[float](max(self.momentum_periods) + 1)
# Warm up history with DAILY bars
history = algorithm.History(symbol, max(self.momentum_periods) + 1, Resolution.Daily)
if not history.empty:
for time, row in history.iterrows():
self.history_windows[symbol].Add(float(row['close']))
algorithm.Debug(f"[DefenseAlpha] {symbol.Value}: Warmed up with {len(history)} daily bars")
# Create Daily consolidator for this symbol
# This ensures we get DAILY bars regardless of subscription resolution
consolidator = TradeBarConsolidator(timedelta(days=1))
# Use closure to capture symbol
def create_handler(sym):
return lambda sender, bar: self._on_daily_bar(algorithm, sym, bar)
consolidator.DataConsolidated += create_handler(symbol)
algorithm.SubscriptionManager.AddConsolidator(symbol, consolidator)
self.consolidators[symbol] = consolidator
algorithm.Debug(f"[DefenseAlpha] {symbol.Value}: Daily consolidator added")
for removed in changes.RemovedSecurities:
symbol = removed.Symbol
if symbol in self.symbols:
self.symbols.remove(symbol)
if symbol in self.history_windows:
del self.history_windows[symbol]
if symbol in self.consolidators:
# Cleanup consolidator
algorithm.SubscriptionManager.RemoveConsolidator(symbol, self.consolidators[symbol])
del self.consolidators[symbol]
def _on_daily_bar(self, algorithm: QCAlgorithm, symbol: Symbol, bar: TradeBar) -> None:
"""
Called when a daily bar is consolidated
Updates history window with DAILY close price
"""
if symbol in self.history_windows:
self.history_windows[symbol].Add(float(bar.Close))
# Debug: Log first few daily bar updates
if not hasattr(self, '_daily_bar_count'):
self._daily_bar_count = {}
if symbol not in self._daily_bar_count:
self._daily_bar_count[symbol] = 0
self._daily_bar_count[symbol] += 1
if self._daily_bar_count[symbol] <= 3:
algorithm.Debug(f"[DefenseAlpha|DEBUG] {symbol.Value}: Daily bar #{self._daily_bar_count[symbol]} @ {bar.Time}, Close={bar.Close:.2f}")
def _debug_window_state(self, algorithm: QCAlgorithm) -> None:
"""
Debug: Log window state at rebalance time
"""
algorithm.Debug(f"[DefenseAlpha|DEBUG] === WINDOW STATE AT REBALANCE ({algorithm.Time.date()}) ===")
for symbol in self.symbols:
if symbol.Value in self.defensive_symbols:
if symbol in self.history_windows:
window = self.history_windows[symbol]
if window.IsReady and window.Count >= 252:
algorithm.Debug(f"[DefenseAlpha|DEBUG] {symbol.Value}: "
f"window[0]=${window[0]:.2f}, "
f"window[21]=${window[21]:.2f}, "
f"window[252]=${window[252]:.2f}, "
f"Count={window.Count}")
else:
algorithm.Debug(f"[DefenseAlpha|DEBUG] {symbol.Value}: NOT READY (Count={window.Count})")
else:
algorithm.Debug(f"[DefenseAlpha|DEBUG] {symbol.Value}: NO WINDOW")
def _compute_momentum_score(self, symbol: Symbol) -> float:
"""
Compute average momentum across multiple periods (21, 63, 126, 252 days)
Returns None if not ready
"""
if symbol not in self.history_windows:
return None
window = self.history_windows[symbol]
if not window.IsReady:
return None
current_price = window[0]
momentum_values = []
for period in self.momentum_periods:
if window.Count > period:
past_price = window[period]
if past_price > 0:
momentum = (current_price - past_price) / past_price
momentum_values.append(momentum)
if momentum_values:
return sum(momentum_values) / len(momentum_values)
return None
def _generate_insights(self, algorithm: QCAlgorithm) -> List[Insight]:
"""
Generate insights with tags for PCM to parse
Tag format: "Defense|Rank{N}|Score{X}"
"""
insights = []
period = timedelta(days=365) # Long period - FLAT insights will close positions
# Step 1: Calculate momentum for all defensive assets
momentum_scores = {}
for symbol in self.symbols:
if symbol.Value in self.defensive_symbols:
score = self._compute_momentum_score(symbol)
if score is not None:
momentum_scores[symbol] = score
if not momentum_scores:
algorithm.Debug("[DefenseAlpha] No momentum scores available, skipping rebalance")
return insights
# Step 2: Rank by momentum (best to worst)
sorted_assets = sorted(momentum_scores.items(), key=lambda x: x[1], reverse=True)
# Debug: Log all momentum scores
algorithm.Debug(f"[DefenseAlpha] === MOMENTUM SCORES (Month {algorithm.Time.month}) ===")
for symbol, score in sorted_assets:
pass_fail = "✓ PASS" if score > self.risk_free_monthly else "✗ FAIL"
algorithm.Debug(f"[DefenseAlpha] {symbol.Value}: {score:.6f} ({pass_fail}, threshold={self.risk_free_monthly:.6f})")
# Step 3: Track new allocation
new_allocation: Set[Symbol] = set()
upro_ranks = [] # Collect ranks that should go to UPRO
# Step 4: Generate insights for top 4 defensive assets
for i, (symbol, momentum_score) in enumerate(sorted_assets[:4]):
rank = i + 1 # Rank 1-4
if momentum_score > self.risk_free_monthly:
# Defensive asset has good momentum - allocate
tag = f"Defense|Rank{rank}|Score{momentum_score:.4f}"
insight = Insight.Price(symbol, period, InsightDirection.Up)
insight.Tag = tag
insights.append(insight)
new_allocation.add(symbol)
algorithm.Debug(f"[DefenseAlpha] {symbol.Value}: Rank {rank}, Score {momentum_score:.4f} (UP)")
else:
# Momentum weak - this rank goes to UPRO
upro_ranks.append(rank)
algorithm.Debug(f"[DefenseAlpha] {symbol.Value}: Rank {rank}, Score {momentum_score:.4f} (WEAK → UPRO)")
# Step 5: Allocate UPRO if any ranks were weak
if upro_ranks:
upro_symbol = None
for symbol in self.symbols:
if symbol.Value == self.upro_symbol:
upro_symbol = symbol
break
if upro_symbol:
# Tag with all ranks going to UPRO
ranks_str = ",".join([f"Rank{r}" for r in upro_ranks])
tag = f"Defense|{ranks_str}|Fallback"
insight = Insight.Price(upro_symbol, period, InsightDirection.Up)
insight.Tag = tag
insights.append(insight)
new_allocation.add(upro_symbol)
algorithm.Debug(f"[DefenseAlpha] UPRO: {ranks_str} (Fallback)")
# Step 6: Emit FLAT insights for positions we're exiting
for symbol in self.current_allocation:
if symbol not in new_allocation:
# Position being closed
flat_insight = Insight.Price(symbol, period, InsightDirection.Flat)
flat_insight.Tag = "Defense|Exit"
insights.append(flat_insight)
algorithm.Debug(f"[DefenseAlpha] {symbol.Value}: FLAT (Exit)")
# Update tracking
self.current_allocation = new_allocation
algorithm.Debug(f"[DefenseAlpha] Rebalance complete: {len(insights)} insights "
f"({len([i for i in insights if i.Direction == InsightDirection.Up])} UP, "
f"{len([i for i in insights if i.Direction == InsightDirection.Flat])} FLAT)")
return insights
from AlgorithmImports import *
from typing import List, Dict
from collections import OrderedDict
import numpy as np
from datetime import timedelta
from models.utils.fixed_size_dict import FixedSizeDict
from models.utils.vwap_calculator import VWAPCalculator
class IntradayMomentumAlphaModel(AlphaModel):
"""
Intraday Momentum Alpha Model - Copied from IntradayMomentumSP500Framework
Strategy:
- VWAP-based breakout detection
- Historical move analysis (90-day lookback)
- 30-minute consolidation period
- Max 1 trade per day
- Tag-based insights for PCM integration
ONLY CHANGE vs Original: Added tag to insights (Line ~249)
"""
def __init__(self,
period: int = 90,
min_volatility: float = 0.008,
consolidation_period: int = 30):
self.period = period
self.min_volatility = min_volatility
self.consolidation_period = consolidation_period
self.abs_move: Dict[datetime.date, List] = FixedSizeDict(max_size=period)
self.volatility_window = RollingWindow[float](10)
self.daily_returns = RollingWindow[float](period)
self.consolidators = {}
self.vwap_calculators = {}
self.current_opens = {}
self.prev_closes = {}
self.is_ready = False
self.trades_today = {}
def Update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]:
insights = []
if not self.is_ready:
return insights
for symbol in data.Keys:
if symbol in self.consolidators:
bar = data[symbol]
if bar is not None and bar.Volume > 0:
self.vwap_calculators[symbol].update(bar.Close, bar.Volume)
return insights
def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
symbol = security.Symbol
# Only process SPY for intraday trading
if symbol.Value != "SPY":
continue
self.vwap_calculators[symbol] = VWAPCalculator()
self.current_opens[symbol] = None
self.prev_closes[symbol] = None
self.trades_today[symbol] = 0
consolidator = TradeBarConsolidator(timedelta(minutes=self.consolidation_period))
consolidator.DataConsolidated += lambda sender, bar, s=symbol: self._on_data_consolidated(algorithm, s, bar)
algorithm.SubscriptionManager.AddConsolidator(symbol, consolidator)
self.consolidators[symbol] = consolidator
daily_consolidator = TradeBarConsolidator(timedelta(days=1))
daily_consolidator.DataConsolidated += lambda sender, bar, s=symbol: self._on_daily_data(algorithm, s, bar)
algorithm.SubscriptionManager.AddConsolidator(symbol, daily_consolidator)
algorithm.Schedule.On(
algorithm.DateRules.EveryDay(symbol),
algorithm.TimeRules.At(10, 0),
lambda s=symbol: self._after_open(algorithm, s)
)
algorithm.Schedule.On(
algorithm.DateRules.EveryDay(symbol),
algorithm.TimeRules.BeforeMarketOpen(symbol, 1),
lambda s=symbol: self._before_market_open(algorithm, s)
)
self._warm_up(algorithm, symbol)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.consolidators:
algorithm.SubscriptionManager.RemoveConsolidator(symbol, self.consolidators[symbol])
del self.consolidators[symbol]
del self.vwap_calculators[symbol]
del self.current_opens[symbol]
del self.prev_closes[symbol]
del self.trades_today[symbol]
def _warm_up(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
try:
history = algorithm.History(symbol, self.period + 10, Resolution.DAILY)
if not history.empty:
for row in history.itertuples():
if row.open != 0:
daily_return = row.close / row.open - 1
daily_volatility = (row.high - row.low) / row.open
self.daily_returns.Add(daily_return)
self.volatility_window.Add(daily_volatility)
except Exception as e:
algorithm.Debug(f"Fehler bei Tagesdaten-Warm-up für {symbol}: {e}")
self._warm_up_abs_move(algorithm, symbol)
self.is_ready = self.daily_returns.IsReady and len(self.abs_move) >= self.period
if self.is_ready:
algorithm.Debug("[IntradayAlpha] Warm-up complete")
def _warm_up_abs_move(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
try:
history = algorithm.History(symbol, timedelta(days=self.period * 2), Resolution.Minute)
if history.empty:
algorithm.Debug(f"[IntradayAlpha] No minute data for abs_move warm-up: {symbol}")
return
except Exception as e:
algorithm.Debug(f"[IntradayAlpha] Error in abs_move warm-up for {symbol}: {e}")
return
grouped_by_day = {}
for time_index, row in history.loc[symbol].iterrows():
dt = time_index.to_pydatetime()
date = dt.date()
if date not in grouped_by_day:
grouped_by_day[date] = []
grouped_by_day[date].append((dt, row["close"]))
abs_move_complete = OrderedDict()
for date, data in grouped_by_day.items():
data.sort()
if len(data) < 30:
continue
session_open = data[0][1]
current_bar_time = None
current_window = []
moves_for_day = []
for dt, price in data:
rounded_time = dt.replace(minute=(dt.minute // self.consolidation_period) * self.consolidation_period, second=0, microsecond=0)
if current_bar_time is None:
current_bar_time = rounded_time
if rounded_time == current_bar_time:
current_window.append(price)
else:
if current_window:
close_price = current_window[-1]
move = abs(close_price / session_open - 1)
moves_for_day.append((current_bar_time, move))
current_bar_time = rounded_time
current_window = [price]
if current_window:
close_price = current_window[-1]
move = abs(close_price / session_open - 1)
moves_for_day.append((current_bar_time, move))
if moves_for_day:
abs_move_complete[date] = moves_for_day
if len(abs_move_complete) >= self.period:
break
for date, move_list in abs_move_complete.items():
self.abs_move[date] = move_list
algorithm.Debug(f"[IntradayAlpha] Warm-up complete: {len(self.abs_move)} days loaded")
def _after_open(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
self.trades_today[symbol] = 0
if symbol in algorithm.Securities:
self.current_opens[symbol] = algorithm.Securities[symbol].Open
def _before_market_open(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
if symbol in algorithm.Securities:
self.prev_closes[symbol] = algorithm.Securities[symbol].Close
if symbol in self.vwap_calculators:
self.vwap_calculators[symbol].reset()
def _on_daily_data(self, algorithm: QCAlgorithm, symbol: Symbol, bar: TradeBar) -> None:
if bar.Open != 0:
daily_volatility = (bar.High - bar.Low) / bar.Open
self.volatility_window.Add(daily_volatility)
def _on_data_consolidated(self, algorithm: QCAlgorithm, symbol: Symbol, consolidated_bar: TradeBar) -> None:
if not self.is_ready:
return
if not self.current_opens.get(symbol):
return
if self.volatility_window.Count < 10:
return
vol10 = sum(self.volatility_window) / self.volatility_window.Count
if vol10 < self.min_volatility:
return
current_time = algorithm.Time
if current_time.date() not in self.abs_move:
self.abs_move[current_time.date()] = []
self.abs_move[current_time.date()].append((
current_time,
abs(consolidated_bar.Close / self.current_opens[symbol] - 1)
))
if len(self.abs_move) < self.period:
return
avg_move = [
move for date, moves in self.abs_move.items()
if date != current_time.date()
for time, move in moves
if time.hour == current_time.hour and time.minute == current_time.minute
]
if not avg_move:
return
mean_move = np.mean(avg_move)
prev_close = self.prev_closes.get(symbol)
session_open = self.current_opens.get(symbol)
if not prev_close or not session_open:
return
upper_bound = max(session_open, prev_close) * (1 + mean_move)
lower_bound = min(session_open, prev_close) * (1 - mean_move)
vwap_value = self.vwap_calculators[symbol].get_vwap()
if vwap_value is None:
vwap_value = lower_bound
if self.trades_today.get(symbol, 0) >= 1:
return
trade_direction = None
confidence = 0.5
duration = timedelta(minutes=self.consolidation_period)
if consolidated_bar.Close > upper_bound:
trade_direction = InsightDirection.Up
confidence = min(1.0, (consolidated_bar.Close - upper_bound) / upper_bound)
direction_str = "Upper"
elif consolidated_bar.Close < lower_bound:
trade_direction = InsightDirection.Down
confidence = min(1.0, (lower_bound - consolidated_bar.Close) / lower_bound)
direction_str = "Lower"
if trade_direction is not None:
# **ONLY CHANGE vs Original: Add Tag to Insight**
tag = f"Intraday|Breakout|{direction_str}|Strength{confidence:.2f}"
insight = Insight.Price(symbol, duration, trade_direction, confidence)
insight.Tag = tag # Tag added for Multi-Alpha PCM
algorithm.EmitInsights([insight])
self.trades_today[symbol] = self.trades_today.get(symbol, 0) + 1
algorithm.Debug(f"[IntradayAlpha] {symbol.Value}: {direction_str} breakout, confidence {confidence:.2f}")
from AlgorithmImports import *
from datetime import date, timedelta
from typing import List, Set
class SeasonalAlphaModel(AlphaModel):
"""
Seasonal Alpha Model - Calendar-based equity rotation
Strategy:
- Hold specific assets during their favorable calendar periods
- Equal-weight allocation (PCM handles via counting UP insights)
- SHY fallback when no assets in hold period
- Forever Period: Insights never expire, explicit FLAT on exits
- Subscribes to SPY Daily Consolidator (from Universe Model)
"""
def __init__(self):
self.symbols_map = {}
# Hold periods: (start_month, start_day, end_month, end_day)
self.hold_periods = {
"SPY": [(10, 1, 1, 31)], # Oct 1 - Jan 31
"QQQ": [(3, 11, 7, 20)], # Mar 11 - Jul 20
"IWM": [(10, 28, 11, 25)], # Oct 28 - Nov 25
"XLY": [(5, 20, 7, 31), (11, 1, 11, 30)], # May 20 - Jul 31, Nov 1-30
"XLK": [(1, 1, 2, 15), (5, 16, 7, 22)], # Jan 1 - Feb 15, May 16 - Jul 22
# "GLD": [(1, 1, 1, 31), (2, 1, 2, 28)], # REMOVED TO TEST DRAWDOWN EFFECT - GLD only in Defense now
"GER40": [(3, 15, 6, 5), (11, 1, 11, 30)], # Mar 15 - Jun 5, Nov 1-30 (DAX - matching Original)
"MTUM": [], # No hold period
"FDN": [], # No hold period
"ARKK": [(5, 13, 7, 21), (11, 1, 11, 30)], # May 13 - Jul 21, Nov 1-30
"DBA": [], # No hold period
"IHF": [(4, 1, 4, 30), (5, 1, 5, 31)], # Apr 1-30, May 1-31
"SLV": [(1, 1, 2, 20), (6, 20, 9, 1)], # Jan 1 - Feb 20, Jun 20 - Sep 1
"EWQ": [], # No hold period
"EVP": [], # No hold period
"XLE": [(3, 20, 4, 30), (9, 20, 11, 20)], # Mar 20 - Apr 30, Sep 20 - Nov 20
"EWU": [], # No hold period
"XLB": [(6, 20, 7, 28), (10, 1, 11, 5)], # Jun 20 - Jul 28, Oct 1 - Nov 5
"XRT": [(11, 1, 11, 30)], # Nov 1-30
"EWC": [] # No hold period
}
self.fallback_symbol = "SHY"
# Track current holdings to emit FLAT on exits
self.current_holdings: Set[str] = set()
# Track last check to avoid duplicate processing
self.last_check_date = None
# Subscribe to SPY Daily Consolidator
self.spy_daily_ready = False
def Update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]:
"""
Check daily if holdings should change
Only emit insights when actual change detected
"""
insights = []
current_date = algorithm.Time.date()
if algorithm.IsWarmingUp:
return insights
# Only check once per day
if self.last_check_date == current_date:
return insights
self.last_check_date = current_date
# Determine which assets should be held today
assets_to_hold = self._get_assets_to_hold(current_date)
# Only generate insights if holdings changed
if assets_to_hold != self.current_holdings:
algorithm.Debug(f"[SeasonalAlpha] {current_date}: Holdings changing from {self.current_holdings} to {assets_to_hold}")
insights = self._generate_rebalance_insights(algorithm, assets_to_hold)
self.current_holdings = assets_to_hold
return insights
def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
"""
Map symbol objects to ticker strings
NOTE: SPY Daily Consolidator disabled - Seasonal works with Minute data directly
"""
for added in changes.AddedSecurities:
symbol = added.Symbol
symbol_str = symbol.Value
self.symbols_map[symbol_str] = symbol
for removed in changes.RemovedSecurities:
symbol = removed.Symbol
keys_to_remove = [k for k, v in self.symbols_map.items() if v == symbol]
for key in keys_to_remove:
del self.symbols_map[key]
def _get_assets_to_hold(self, current_date: date) -> Set[str]:
"""Determine which assets should be held on this date"""
assets_to_hold = set()
for asset_name, periods in self.hold_periods.items():
for start_month, start_day, end_month, end_day in periods:
if self._is_in_holding_period(current_date, start_month, start_day, end_month, end_day):
assets_to_hold.add(asset_name)
break
return assets_to_hold
def _is_in_holding_period(self, current_date: date, start_month: int, start_day: int,
end_month: int, end_day: int) -> bool:
"""
Check if current date falls within the holding period
IMPORTANT: Using ORIGINAL logic from Non-Framework code (Line 73-75)
This logic is BUGGY for year-spanning periods but we replicate it exactly
to match Original's behavior!
"""
try:
year = current_date.year
# ORIGINAL LOGIC (EXACT REPLICATION)
start_date = date(year, start_month, start_day)
end_date = date(year, end_month, end_day)
return start_date <= current_date <= end_date
except ValueError:
# Invalid date (e.g., Feb 30)
return False
def _generate_rebalance_insights(self, algorithm: QCAlgorithm, assets_to_hold: Set[str]) -> List[Insight]:
"""
Generate insights with tags for PCM to parse
Tag format: "Seasonal|HoldPeriod|{SYMBOL}|{DATES}"
Period: Forever (timedelta.max)
"""
algorithm.Debug(f"[TIMING|SeasonalAlpha] {algorithm.Time}: _generate_rebalance_insights STARTED")
insights = []
period = timedelta(days=365) # Long period - FLAT insights will close positions
# Step 1: Emit FLAT insights for symbols we're exiting
symbols_to_exit = self.current_holdings - assets_to_hold
for asset_name in symbols_to_exit:
if asset_name in self.symbols_map:
symbol = self.symbols_map[asset_name]
flat_insight = Insight.Price(symbol, period, InsightDirection.Flat)
flat_insight.Tag = "Seasonal|Exit"
insights.append(flat_insight)
algorithm.Debug(f"[SeasonalAlpha] FLAT: Exiting {asset_name}")
# Handle fallback exit
if len(self.current_holdings) == 0 and len(assets_to_hold) > 0:
# Exiting fallback (SHY)
if self.fallback_symbol in self.symbols_map:
fallback_symbol = self.symbols_map[self.fallback_symbol]
flat_insight = Insight.Price(fallback_symbol, period, InsightDirection.Flat)
flat_insight.Tag = "Seasonal|Exit|Fallback"
insights.append(flat_insight)
algorithm.Debug(f"[SeasonalAlpha] FLAT: Exiting fallback {self.fallback_symbol}")
# Step 2: Emit UP insights for new holdings
if assets_to_hold:
for asset_name in assets_to_hold:
if asset_name in self.symbols_map:
symbol = self.symbols_map[asset_name]
# Find which period we're in (for tag)
period_str = self._get_period_string(asset_name, algorithm.Time.date())
tag = f"Seasonal|HoldPeriod|{asset_name}|{period_str}"
insight = Insight.Price(symbol, period, InsightDirection.Up)
insight.Tag = tag
insights.append(insight)
algorithm.Debug(f"[SeasonalAlpha] UP: {asset_name} ({period_str})")
else:
algorithm.Debug(f"[SeasonalAlpha] WARNING: {asset_name} not in symbol map")
else:
# Fallback to SHY if no assets should be held
if self.fallback_symbol in self.symbols_map:
fallback_symbol = self.symbols_map[self.fallback_symbol]
tag = "Seasonal|Fallback|SHY"
insight = Insight.Price(fallback_symbol, period, InsightDirection.Up)
insight.Tag = tag
insights.append(insight)
algorithm.Debug(f"[SeasonalAlpha] UP: Fallback to {self.fallback_symbol}")
algorithm.Debug(f"[SeasonalAlpha] Generated {len(insights)} insights ({len([i for i in insights if i.Direction == InsightDirection.Up])} UP, {len([i for i in insights if i.Direction == InsightDirection.Flat])} FLAT)")
algorithm.Debug(f"[TIMING|SeasonalAlpha] {algorithm.Time}: _generate_rebalance_insights COMPLETED, returning {len(insights)} insights")
return insights
def _get_period_string(self, asset_name: str, current_date: date) -> str:
"""Get human-readable period string for tag"""
if asset_name not in self.hold_periods:
return "Unknown"
for start_month, start_day, end_month, end_day in self.hold_periods[asset_name]:
if self._is_in_holding_period(current_date, start_month, start_day, end_month, end_day):
return f"{start_month:02d}/{start_day:02d}-{end_month:02d}/{end_day:02d}"
return "Unknown"
from AlgorithmImports import *
from typing import List
class TaggedImmediateExecutionModel(ExecutionModel):
"""
OPTIMIZED Execution Model that preserves Alpha source tags in Order tags
Performance: O(1) instead of O(n) - no iteration through all insights!
Uses cached symbol_to_alpha mapping from main.py OnOrderEvent
This allows us to track which Alpha Model generated each order:
- Defense orders tagged with "Defense"
- Seasonal orders tagged with "Seasonal"
- Intraday orders tagged with "Intraday"
"""
def __init__(self):
self.targets_collection = PortfolioTargetCollection()
def Execute(self, algorithm: QCAlgorithm, targets: List[IPortfolioTarget]) -> None:
"""
Execute portfolio targets immediately with tags
OPTIMIZED: Uses algorithm.symbol_to_alpha (cached in OnOrderEvent)
instead of iterating through ALL insights every time!
Performance: O(targets) instead of O(all_insights)
"""
# Update targets collection
self.targets_collection.AddRange(targets)
# Execute orders with tags
for target in self.targets_collection.OrderByMarginImpact(algorithm):
# Get existing holdings
existing = algorithm.Securities[target.Symbol].Holdings.Quantity
# Calculate quantity to order
quantity = target.Quantity - existing
if quantity == 0:
continue
# Get alpha source tag from cached mapping (FAST!)
# If not in cache, determine from current insights (fallback)
alpha_tag = None
if hasattr(algorithm, 'symbol_to_alpha'):
alpha_tag = algorithm.symbol_to_alpha.get(target.Symbol, None)
if not alpha_tag:
# Fallback: Check current insights for this specific symbol only
for insight in algorithm.Insights:
if insight.Symbol == target.Symbol and not insight.IsExpired(algorithm.UtcTime):
if insight.Tag:
alpha_tag = insight.Tag.split('|')[0] if '|' in insight.Tag else insight.Tag
# Cache for next time (if dictionary exists)
if hasattr(algorithm, 'symbol_to_alpha'):
algorithm.symbol_to_alpha[target.Symbol] = alpha_tag
break
if not alpha_tag:
alpha_tag = "Unknown"
# Submit order with tag
ticket = algorithm.MarketOrder(target.Symbol, quantity, tag=alpha_tag)
if ticket:
algorithm.Debug(f"[Execution] {target.Symbol.Value}: {quantity} shares, Tag: {alpha_tag}")
# Clear processed targets
self.targets_collection.ClearFulfilled(algorithm)
# region imports
from AlgorithmImports import *
# endregion
# Portfolio Module
from .multi_alpha_pcm import MultiAlphaPortfolioConstructionModel
__all__ = ['MultiAlphaPortfolioConstructionModel']
from AlgorithmImports import *
from typing import Dict, List, Tuple
import numpy as np
class MultiAlphaPortfolioConstructionModel(PortfolioConstructionModel):
"""
Multi-Alpha Portfolio Construction Model
Budget Allocation:
- Defense: 40%
- Seasonal: 40%
- Intraday: 20%
Tag-based Allocation Logic:
- Defense: Parse Rank from tag, apply 40/30/20/10 weights (of 40% budget)
- Seasonal: Equal-weight across all UP insights (of 40% budget)
- Intraday: Volatility-based (of 20% budget)
SPY Conflict Resolution:
- If SPY in Seasonal Hold Period (Oct 1 - Jan 31): Seasonal takes priority
- Otherwise: Intraday can trade SPY
"""
def __init__(self, alpha_budgets: Dict[str, float] = None):
"""
Args:
alpha_budgets: Budget allocation per alpha model
Default: {"DefenseAlphaModel": 0.4, "SeasonalAlphaModel": 0.4, "IntradayMomentumAlphaModel": 0.2}
"""
if alpha_budgets is None:
self.alpha_budgets = {
"DefenseAlphaModel": 0.4,
"SeasonalAlphaModel": 0.4,
"IntradayMomentumAlphaModel": 0.2
}
else:
self.alpha_budgets = alpha_budgets
# Validate budgets sum to 1.0
budget_sum = sum(self.alpha_budgets.values())
if abs(budget_sum - 1.0) > 0.001:
raise ValueError(f"Alpha budgets must sum to 1.0, got {budget_sum}")
# Intraday: Volatility calculation (copied from Intraday PCM)
self.intraday_target_volatility = 0.02
self.intraday_max_leverage = 4.0
self.intraday_period = 90
self.daily_returns = {}
self.current_opens = {}
# Defense: Rank-based weights (40%, 30%, 20%, 10%)
self.defense_rank_weights = {1: 0.4, 2: 0.3, 3: 0.2, 4: 0.1}
def CreateTargets(self, algorithm: QCAlgorithm, insights: List[Insight]) -> List[PortfolioTarget]:
"""
Main entry point: Convert insights to portfolio targets
Steps:
1. Group insights by alpha model (via tag parsing)
2. Allocate Defense (rank-based)
3. **RISK-OFF MODE:** Check if Defense is fully defensive → Pause Seasonal
4. Allocate Seasonal (equal-weight) - only if NOT in Risk-Off Mode
5. Allocate Intraday (volatility-based)
6. Merge weights
7. **Normalize NET exposure** to ensure Total ≤ 100% (matching Original)
8. Resolve SPY conflicts
9. Convert weights to quantities
"""
algorithm.Debug(f"[TIMING|PCM] {algorithm.Time}: CreateTargets STARTED with {len(insights)} insights")
targets = []
if not insights:
algorithm.Debug(f"[TIMING|PCM] {algorithm.Time}: CreateTargets COMPLETED (no insights), returning 0 targets")
return targets
# Step 1: Group insights by alpha model
grouped = self._group_insights_by_alpha(insights)
# Debug
algorithm.Debug(f"[PCM] Processing {len(insights)} insights: Defense={len(grouped.get('Defense', []))}, Seasonal={len(grouped.get('Seasonal', []))}, Intraday={len(grouped.get('Intraday', []))}")
# Step 2: Calculate Defense weights
defense_weights = self._allocate_defense(algorithm, grouped.get('Defense', []))
# Step 3: **RISK-OFF MODE CHECK**
risk_off_mode = self._is_risk_off_mode(algorithm, defense_weights)
if risk_off_mode:
algorithm.Debug(f"[PCM] ⚠️ RISK-OFF MODE ACTIVE - Defense is fully defensive, Seasonal PAUSED")
# Step 4: Calculate Seasonal weights (skip if Risk-Off)
if risk_off_mode:
seasonal_weights = {} # PAUSE Seasonal during crisis
else:
seasonal_weights = self._allocate_seasonal(algorithm, grouped.get('Seasonal', []))
# Step 5: Calculate Intraday weights
intraday_weights = self._allocate_intraday(algorithm, grouped.get('Intraday', []))
# Step 3: Merge weights
all_weights = {}
for symbol, weight in defense_weights.items():
all_weights[symbol] = all_weights.get(symbol, 0.0) + weight
for symbol, weight in seasonal_weights.items():
all_weights[symbol] = all_weights.get(symbol, 0.0) + weight
for symbol, weight in intraday_weights.items():
all_weights[symbol] = all_weights.get(symbol, 0.0) + weight
# Step 3.5: Normalize to ensure NET exposure ≤ 100% (matching Original behavior)
all_weights = self._normalize_net_exposure(algorithm, all_weights)
# Step 4: Resolve SPY conflicts
all_weights = self._resolve_conflicts(algorithm, all_weights, seasonal_weights, intraday_weights)
# Step 5: Convert to portfolio targets
for symbol, weight in all_weights.items():
quantity = self._weight_to_quantity(algorithm, symbol, weight)
targets.append(PortfolioTarget(symbol, quantity))
algorithm.Debug(f"[PCM] Target: {symbol.Value} = {quantity} shares (weight={weight:.4f})")
algorithm.Debug(f"[TIMING|PCM] {algorithm.Time}: CreateTargets COMPLETED, returning {len(targets)} targets")
return targets
def OnSecuritiesChanged(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
"""
Initialize tracking for new securities
NOTE: Scheduling removed - was causing conflicts with main.py scheduling
"""
for security in changes.AddedSecurities:
symbol = security.Symbol
self._initialize_symbol(algorithm, symbol)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.daily_returns:
del self.daily_returns[symbol]
if symbol in self.current_opens:
del self.current_opens[symbol]
# ========================================================================
# HELPER METHODS
# ========================================================================
def _group_insights_by_alpha(self, insights: List[Insight]) -> Dict[str, List[Insight]]:
"""
Group insights by alpha model based on tag prefix
Tag format: "Defense|...", "Seasonal|...", "Intraday|..."
"""
grouped = {'Defense': [], 'Seasonal': [], 'Intraday': []}
for insight in insights:
if not insight.Tag:
continue
if insight.Tag.startswith('Defense'):
grouped['Defense'].append(insight)
elif insight.Tag.startswith('Seasonal'):
grouped['Seasonal'].append(insight)
elif insight.Tag.startswith('Intraday'):
grouped['Intraday'].append(insight)
return grouped
def _allocate_defense(self, algorithm: QCAlgorithm, insights: List[Insight]) -> Dict[Symbol, float]:
"""
Defense Allocation: Rank-based (40%, 30%, 20%, 10%)
Tag format: "Defense|Rank{N}|Score{X}" or "Defense|Rank1,Rank2,...|Fallback" (UPRO)
Budget: 40% of total portfolio
"""
weights = {}
defense_budget = self.alpha_budgets.get("DefenseAlphaModel", 0.4)
for insight in insights:
if insight.Direction == InsightDirection.Flat:
# Exit signal - set to 0
weights[insight.Symbol] = 0.0
continue
# Parse tag for rank
tag = insight.Tag
parts = tag.split('|')
if 'Fallback' in tag:
# UPRO fallback - sum all ranks that went to UPRO
# Tag format: "Defense|Rank1,Rank2|Fallback"
rank_part = parts[1] # "Rank1,Rank2"
ranks = rank_part.split(',') # ["Rank1", "Rank2"]
upro_weight = 0.0
for rank_str in ranks:
rank_num = int(rank_str.replace('Rank', ''))
upro_weight += self.defense_rank_weights.get(rank_num, 0.0)
weights[insight.Symbol] = defense_budget * upro_weight
algorithm.Debug(f"[PCM] Defense UPRO: {ranks} → weight={weights[insight.Symbol]:.4f}")
else:
# Normal defense asset
# Tag format: "Defense|Rank{N}|Score{X}"
rank_part = parts[1] # "Rank1"
rank = int(rank_part.replace('Rank', ''))
rank_weight = self.defense_rank_weights.get(rank, 0.0)
weights[insight.Symbol] = defense_budget * rank_weight
algorithm.Debug(f"[PCM] Defense {insight.Symbol.Value}: Rank {rank} → weight={weights[insight.Symbol]:.4f}")
return weights
def _allocate_seasonal(self, algorithm: QCAlgorithm, insights: List[Insight]) -> Dict[Symbol, float]:
"""
Seasonal Allocation: Equal-weight
Tag format: "Seasonal|HoldPeriod|{SYMBOL}|{DATES}" or "Seasonal|Fallback|SHY"
Budget: 40% of total portfolio
"""
weights = {}
seasonal_budget = self.alpha_budgets.get("SeasonalAlphaModel", 0.4)
# Count UP insights
up_insights = [i for i in insights if i.Direction == InsightDirection.Up]
flat_insights = [i for i in insights if i.Direction == InsightDirection.Flat]
# FLAT insights - exit positions
for insight in flat_insights:
weights[insight.Symbol] = 0.0
# UP insights - equal-weight
if up_insights:
equal_weight = seasonal_budget / len(up_insights)
for insight in up_insights:
weights[insight.Symbol] = equal_weight
algorithm.Debug(f"[PCM] Seasonal {insight.Symbol.Value}: equal-weight → {equal_weight:.4f}")
return weights
def _allocate_intraday(self, algorithm: QCAlgorithm, insights: List[Insight]) -> Dict[Symbol, float]:
"""
Intraday Allocation: Volatility-based (copied from Intraday PCM)
Tag format: "Intraday|Breakout|{Direction}|Strength{X}"
Budget: 20% of total portfolio
"""
weights = {}
intraday_budget = self.alpha_budgets.get("IntradayMomentumAlphaModel", 0.2)
for insight in insights:
symbol = insight.Symbol
if insight.Direction == InsightDirection.Flat:
weights[symbol] = 0.0
continue
# Initialize if needed
if symbol not in self.daily_returns:
self._initialize_symbol(algorithm, symbol)
if not self._is_ready(symbol):
algorithm.Debug(f"[PCM] Intraday {symbol.Value}: not ready (volatility)")
continue
current_open = self.current_opens.get(symbol)
if not current_open or current_open == 0:
continue
# Calculate volatility
returns = list(self.daily_returns[symbol])
mean_return = np.mean(returns)
volatility = np.std(returns)
if volatility == 0:
continue
# Volatility-based leverage (capped)
leverage = min(self.intraday_max_leverage, self.intraday_target_volatility / volatility)
# Apply intraday budget limit
leverage_with_budget = leverage * intraday_budget
# Calculate weight
portfolio_value = algorithm.Portfolio.TotalPortfolioValue
if portfolio_value <= 0:
continue
quantity = int(portfolio_value * leverage_with_budget / current_open)
weight = quantity * current_open / portfolio_value
if insight.Direction == InsightDirection.Up:
weights[symbol] = weight
elif insight.Direction == InsightDirection.Down:
weights[symbol] = -weight
else:
weights[symbol] = 0.0
algorithm.Debug(f"[PCM] Intraday {symbol.Value}: vol={volatility:.4f}, leverage={leverage:.2f}, weight={weights[symbol]:.4f}")
return weights
def _is_risk_off_mode(self, algorithm: QCAlgorithm, defense_weights: Dict[Symbol, float]) -> bool:
"""
Risk-Off Mode Detection:
Returns True if Defense is holding ONLY defensive assets (TLT, GLD, DBC, UUP)
and NO UPRO (leveraged S&P500).
When Risk-Off = True → Seasonal should PAUSE to avoid equity exposure during crisis
Defensive Assets: TLT (bonds), GLD (gold), DBC (commodities), UUP (US dollar)
Risk Asset: UPRO (3x leveraged S&P500)
"""
defensive_symbols = {"TLT", "GLD", "DBC", "UUP"}
risk_symbol = "UPRO"
# Check what Defense is holding
defense_holdings = set()
for symbol, weight in defense_weights.items():
if weight > 0: # Currently allocated
defense_holdings.add(symbol.Value)
# Risk-Off if:
# 1. Defense has positions AND
# 2. NO UPRO AND
# 3. Only defensive assets
if not defense_holdings:
return False # No positions = not risk-off
has_upro = risk_symbol in defense_holdings
only_defensive = defense_holdings.issubset(defensive_symbols)
risk_off = (not has_upro) and only_defensive
if risk_off:
algorithm.Debug(f"[PCM] Risk-Off Detection: Holdings={defense_holdings}, UPRO={has_upro}, OnlyDefensive={only_defensive}")
return risk_off
def _normalize_net_exposure(self, algorithm: QCAlgorithm, all_weights: Dict[Symbol, float]) -> Dict[Symbol, float]:
"""
Normalize weights to ensure Total Allocation ≤ 100%
WICHTIG: Beide Strategien (Defense + Seasonal) sind Long-only!
Daher: Total Allocation = Summe aller Weights
Defense Budget: 60%
Seasonal Budget: 40%
Total: 100%
Wenn durch Timing, Rundung oder andere Faktoren Total > 100%:
→ Scale down ALL weights proportional
"""
if not all_weights:
return all_weights
# Calculate Total Allocation (Summe aller Weights)
# Bei Long-only: Total = Summe aller positiven Weights
total_allocation = sum(w for w in all_weights.values() if w > 0)
algorithm.Debug(f"[PCM] Total Allocation Before Normalization: {total_allocation:.2%}")
# If total allocation exceeds 100%, scale down proportionally
if total_allocation > 1.0:
scale_factor = 1.0 / total_allocation
normalized_weights = {symbol: weight * scale_factor for symbol, weight in all_weights.items()}
# Recalculate
new_total = sum(w for w in normalized_weights.values() if w > 0)
algorithm.Debug(f"[PCM] ⚠️ Total Allocation exceeded 100%! Scaled by {scale_factor:.3f}")
algorithm.Debug(f"[PCM] Total Allocation After Normalization: {new_total:.2%}")
return normalized_weights
return all_weights
def _resolve_conflicts(self, algorithm: QCAlgorithm, all_weights: Dict[Symbol, float],
seasonal_weights: Dict[Symbol, float], intraday_weights: Dict[Symbol, float]) -> Dict[Symbol, float]:
"""
SPY Conflict Resolution:
- If SPY in Seasonal Hold Period (Oct 1 - Jan 31): Seasonal takes priority
- Otherwise: Intraday takes priority
"""
# Find SPY symbol
spy_symbol = None
for symbol in all_weights.keys():
if symbol.Value == "SPY":
spy_symbol = symbol
break
if not spy_symbol:
return all_weights # No SPY, no conflict
# Check if both Seasonal and Intraday want SPY
has_seasonal = spy_symbol in seasonal_weights and seasonal_weights[spy_symbol] != 0
has_intraday = spy_symbol in intraday_weights and intraday_weights[spy_symbol] != 0
if not (has_seasonal and has_intraday):
return all_weights # No conflict
# Conflict detected - check hold period
if self._is_spy_seasonal_active(algorithm):
# Seasonal period active (Oct 1 - Jan 31) - Seasonal wins
all_weights[spy_symbol] = seasonal_weights[spy_symbol]
algorithm.Debug(f"[PCM] SPY Conflict: Seasonal priority (Hold Period active)")
else:
# Outside seasonal period - Intraday wins
all_weights[spy_symbol] = intraday_weights[spy_symbol]
algorithm.Debug(f"[PCM] SPY Conflict: Intraday priority (Hold Period inactive)")
return all_weights
def _is_spy_seasonal_active(self, algorithm: QCAlgorithm) -> bool:
"""
Check if SPY seasonal hold period is active (Oct 1 - Jan 31)
"""
current_date = algorithm.Time.date()
month = current_date.month
day = current_date.day
# Oct 1 - Dec 31
if month >= 10:
return True
# Jan 1 - Jan 31
if month == 1:
return True
return False
def _weight_to_quantity(self, algorithm: QCAlgorithm, symbol: Symbol, weight: float) -> int:
"""Convert weight to share quantity"""
if weight == 0:
return 0
portfolio_value = algorithm.Portfolio.TotalPortfolioValue
if portfolio_value <= 0:
return 0
price = algorithm.Securities[symbol].Price
if price <= 0:
return 0
quantity = int((weight * portfolio_value) / price)
return quantity
# ========================================================================
# INTRADAY VOLATILITY TRACKING (copied from Intraday PCM)
# ========================================================================
def _initialize_symbol(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
"""Initialize volatility tracking for symbol"""
self.daily_returns[symbol] = RollingWindow[float](self.intraday_period)
self.current_opens[symbol] = None
try:
history = algorithm.History(symbol, self.intraday_period + 10, Resolution.DAILY)
if not history.empty:
for row in history.itertuples():
if row.open != 0:
daily_return = row.close / row.open - 1
self.daily_returns[symbol].Add(daily_return)
except Exception as e:
algorithm.Debug(f"[PCM] Error in volatility warm-up for {symbol}: {e}")
def _update_open(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
"""Update opening price for volatility calculation"""
if symbol in algorithm.Securities:
self.current_opens[symbol] = algorithm.Securities[symbol].Open
def _update_daily_return(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
"""Update daily return for volatility calculation"""
if symbol in algorithm.Securities and symbol in self.current_opens:
current_price = algorithm.Securities[symbol].Price
current_open = self.current_opens[symbol]
if current_open and current_open != 0:
daily_return = current_price / current_open - 1
self.daily_returns[symbol].Add(daily_return)
def _is_ready(self, symbol: Symbol) -> bool:
"""Check if volatility data is ready"""
return (symbol in self.daily_returns and
self.daily_returns[symbol].IsReady)
# region imports
from AlgorithmImports import *
# endregion
# Risk Module
from .multi_alpha_risk_model import MultiAlphaRiskManagementModel
__all__ = ['MultiAlphaRiskManagementModel']
from AlgorithmImports import *
from typing import List
class MultiAlphaRiskManagementModel(RiskManagementModel):
"""
Multi-Alpha Risk Management Model
Simple portfolio-level risk filters:
- Max Total Leverage: 2.0 (200%)
- Max Single Position: 0.25 (25% of portfolio)
- Max Drawdown: 0.30 (30% emergency stop)
Philosophy: Simple, broad filters
- NOT complex per-alpha limits like Intraday Framework
- Scale down proportionally when limits exceeded
"""
def __init__(self,
max_total_leverage: float = 2.0,
max_single_position: float = 0.25,
max_drawdown: float = 0.30):
"""
Args:
max_total_leverage: Maximum total portfolio leverage (default 2.0 = 200%)
max_single_position: Maximum single position size as % of portfolio (default 0.25 = 25%)
max_drawdown: Maximum drawdown before emergency liquidation (default 0.30 = 30%)
"""
self.max_total_leverage = max_total_leverage
self.max_single_position = max_single_position
self.max_drawdown = max_drawdown
# Track highest portfolio value for drawdown calculation
self.highest_portfolio_value = 0.0
def ManageRisk(self, algorithm: QCAlgorithm, targets: List[PortfolioTarget]) -> List[PortfolioTarget]:
"""
Apply risk filters to portfolio targets
Steps:
1. Check drawdown - liquidate everything if exceeded
2. Check single position limits - scale down oversized positions
3. Check total leverage - scale down entire portfolio if needed
"""
if not targets:
return targets
portfolio_value = algorithm.Portfolio.TotalPortfolioValue
# Update highest value
if portfolio_value > self.highest_portfolio_value:
self.highest_portfolio_value = portfolio_value
# Step 1: Check for maximum drawdown breach
if self.highest_portfolio_value > 0:
current_drawdown = (self.highest_portfolio_value - portfolio_value) / self.highest_portfolio_value
if current_drawdown > self.max_drawdown:
algorithm.Debug(f"[Risk] MAX DRAWDOWN EXCEEDED: {current_drawdown:.2%} > {self.max_drawdown:.2%}")
algorithm.Debug(f"[Risk] EMERGENCY LIQUIDATION: Closing all positions")
# Emergency liquidation - flatten everything
return [PortfolioTarget(target.Symbol, 0) for target in targets]
# Step 2: Check single position limits
targets = self._limit_single_positions(algorithm, targets, portfolio_value)
# Step 3: Check total leverage
targets = self._limit_total_leverage(algorithm, targets, portfolio_value)
return targets
def _limit_single_positions(self, algorithm: QCAlgorithm, targets: List[PortfolioTarget],
portfolio_value: float) -> List[PortfolioTarget]:
"""
Scale down any positions exceeding max_single_position
"""
adjusted_targets = []
for target in targets:
symbol = target.Symbol
quantity = target.Quantity
if quantity == 0:
adjusted_targets.append(target)
continue
price = algorithm.Securities[symbol].Price
if price <= 0:
adjusted_targets.append(target)
continue
# Calculate position value
position_value = abs(quantity * price)
position_pct = position_value / portfolio_value
# Check if exceeds limit
if position_pct > self.max_single_position:
# Scale down to max limit
scale_factor = self.max_single_position / position_pct
new_quantity = int(quantity * scale_factor)
algorithm.Debug(f"[Risk] Single position limit exceeded: {symbol.Value} {position_pct:.2%} > {self.max_single_position:.2%}")
algorithm.Debug(f"[Risk] Scaling down {symbol.Value}: {quantity} → {new_quantity} shares")
adjusted_targets.append(PortfolioTarget(symbol, new_quantity))
else:
adjusted_targets.append(target)
return adjusted_targets
def _limit_total_leverage(self, algorithm: QCAlgorithm, targets: List[PortfolioTarget],
portfolio_value: float) -> List[PortfolioTarget]:
"""
Scale down entire portfolio if total leverage exceeds limit
"""
# Calculate total exposure
total_exposure = 0.0
for target in targets:
symbol = target.Symbol
quantity = target.Quantity
if quantity == 0:
continue
price = algorithm.Securities[symbol].Price
if price <= 0:
continue
total_exposure += abs(quantity * price)
# Calculate leverage
leverage = total_exposure / portfolio_value if portfolio_value > 0 else 0.0
# Check if exceeds limit
if leverage > self.max_total_leverage:
# Scale down entire portfolio proportionally
scale_factor = self.max_total_leverage / leverage
algorithm.Debug(f"[Risk] Total leverage exceeded: {leverage:.2f}x > {self.max_total_leverage:.2f}x")
algorithm.Debug(f"[Risk] Scaling down entire portfolio by {scale_factor:.2%}")
scaled_targets = []
for target in targets:
if target.Quantity == 0:
scaled_targets.append(target)
else:
new_quantity = int(target.Quantity * scale_factor)
scaled_targets.append(PortfolioTarget(target.Symbol, new_quantity))
return scaled_targets
return targets
# region imports
from AlgorithmImports import *
# endregion
# Universe Module
from AlgorithmImports import *
from datetime import timedelta
class MultiAlphaUniverseSelectionModel(ManualUniverseSelectionModel):
"""
Multi-Alpha Universe Selection Model
Symbole:
- Defense: TLT, GLD, DBC, UUP, UPRO (Daily)
- Seasonal: SPY, QQQ, IWM, XLY, XLK, GLD, MTUM, FDN, ARKK, SHY, DBA, IHF, SLV, EWQ, EVP, XLE, EWU, XLB, XRT, EWC, GER40 (Daily)
- Intraday: SPY (Minute)
SPY: Minute Resolution (für Intraday + Daily Consolidator für Seasonal)
Rest: Daily Resolution
"""
def __init__(self):
# Seasonal symbols only (Daily Resolution)
# NOTE: Defense symbols (TLT, GLD, DBC, UUP, UPRO) are added in main.py with Minute resolution!
daily_symbols = [
# Seasonal (GLD is in both Defense and Seasonal - added by main.py)
"QQQ", "IWM", "XLY", "XLK", "MTUM", "FDN", "ARKK",
"SHY", "DBA", "IHF", "SLV", "EWQ", "EVP", "XLE",
"EWU", "XLB", "XRT", "EWC",
"GER40" # DAX - German stock index (matching Original)
]
# Erstelle Symbol Objects (wird in Initialize später zu Securities)
symbols = [Symbol.Create(ticker, SecurityType.Equity, Market.USA) for ticker in daily_symbols]
super().__init__(symbols)
def add_spy_with_minute_resolution(self, algorithm):
"""
SPY separat hinzufügen mit Minute Resolution
Wird von main.py aufgerufen
NOTE: Daily Consolidator temporarily removed - was causing FATAL ERROR
Seasonal Alpha can work with Minute data directly
"""
if not hasattr(algorithm, '_spy_added'):
spy = algorithm.AddEquity("SPY", Resolution.Minute)
spy.SetDataNormalizationMode(DataNormalizationMode.Adjusted)
algorithm._spy_added = True
algorithm.Debug("[Universe] SPY added with Minute resolution (Daily consolidator disabled)")
return spy.Symbol
return None
# region imports
from AlgorithmImports import *
# endregion
# Utils Module
# region imports
from AlgorithmImports import *
# endregion
from collections import OrderedDict, deque
from typing import Deque
import datetime
class FixedSizeDict(OrderedDict):
def __init__(self, max_size: int) -> None:
self.max_size: int = max_size
super().__init__()
def __setitem__(self, key: datetime.date, value: Deque) -> None:
if len(self) >= self.max_size:
self.popitem(last=False)
super().__setitem__(key, value)
# region imports
from AlgorithmImports import *
# endregion
class VWAPCalculator:
def __init__(self):
self.cumulative_price_volume = 0.0
self.cumulative_volume = 0.0
self.current_vwap = None
def update(self, price: float, volume: float) -> None:
if volume > 0:
self.cumulative_price_volume += (price * volume)
self.cumulative_volume += volume
if self.cumulative_volume > 0:
self.current_vwap = self.cumulative_price_volume / self.cumulative_volume
def reset(self) -> None:
self.cumulative_price_volume = 0.0
self.cumulative_volume = 0.0
self.current_vwap = None
def get_vwap(self) -> float:
return self.current_vwap
from AlgorithmImports import *
class CombinedStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetCash(100000)
# Verwende unterstütztes Brokerage-Modell
self.SetBrokerageModel(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
self.weight_defense = 0.6
self.weight_seasonal = 0.4
self.defense = DefenseFirstWithUPRO(self, self.weight_defense)
self.seasonal = SeasonalHoldPeriodStrategy(self, self.weight_seasonal)
chart = Chart("Capital")
chart.AddSeries(Series("Defense", SeriesType.Line, 0))
chart.AddSeries(Series("Seasonal", SeriesType.Line, 0))
chart.AddSeries(Series("Total", SeriesType.Line, 0))
self.AddChart(chart)
def OnData(self, data: Slice):
total_invested_ratio = self.Portfolio.TotalHoldingsValue / self.Portfolio.TotalPortfolioValue
self.Debug(f"{self.Time.date()}: Gesamt Investiert = {total_invested_ratio:.2%}")
for kvp in self.Portfolio:
if kvp.Value.Invested:
self.Debug(f" Holding: {kvp.Key} @ {kvp.Value.Quantity} (Value: {kvp.Value.HoldingsValue:.2f})")
self.defense.OnData(data)
self.seasonal.OnData(data)
self.Plot("Capital", "Total", total_invested_ratio)
self.Plot("Capital", "Defense", self.defense.GetCurrentExposure())
self.Plot("Capital", "Seasonal", self.seasonal.GetCurrentExposure())
def OnOrderEvent(self, orderEvent: OrderEvent):
if orderEvent.Status == OrderStatus.Filled:
self.Debug(f"[Order] {self.Time}: Filled {orderEvent.Symbol.Value}, Qty: {orderEvent.FillQuantity}, Price: {orderEvent.FillPrice}")
class SeasonalHoldPeriodStrategy:
def __init__(self, algo: QCAlgorithm, weight: float):
self.algo = algo
self.weight = weight
self.symbols = {
name: algo.AddEquity(name if name != "DAX" else "GER40", Resolution.Daily).Symbol
for name in [
"SPY", "QQQ", "IWM", "XLY", "XLK", "GLD", "MTUM", "FDN", "ARKK", "DAX",
"SHY", "DBA", "IHF", "SLV", "EWQ", "EVP", "XLE", "EWU", "XLB", "XRT", "EWC"
]
}
self.hold_periods = {
"SPY": [(10, 1, 1, 31)], "QQQ": [(3, 11, 7, 20)], "IWM": [(10, 28, 11, 25)],
"XLY": [(5, 20, 7, 31), (11, 1, 11, 30)], "XLK": [(1, 1, 2, 15), (5, 16, 7, 22)],
"GLD": [(1, 1, 1, 31), (2, 1, 2, 28)], "DAX": [(3, 15, 6, 5), (11, 1, 11, 30)],
"MTUM": [], "ARKK": [(5, 13, 7, 21), (11, 1, 11, 30)], "DBA": [], "IHF": [(4, 1, 4, 30), (5, 1, 5, 31)],
"SLV": [(1, 1, 2, 20), (6, 20, 9, 1)], "EWQ": [], "EVP": [], "XLE": [(3, 20, 4, 30), (9, 20, 11, 20)],
"EWU": [], "XLB": [(6, 20, 7, 28), (10, 1, 11, 5)], "XRT": [(11, 1, 11, 30)], "EWC": []
}
self.current_holdings = set()
def OnData(self, data: Slice):
today = self.algo.Time.date()
to_hold = set()
for name, periods in self.hold_periods.items():
for start_m, start_d, end_m, end_d in periods:
start = datetime(today.year, start_m, start_d).date()
end = datetime(today.year, end_m, end_d).date()
if start <= today <= end:
to_hold.add(name)
break
if to_hold != self.current_holdings:
for symbol in self.symbols.values():
if self.algo.Portfolio[symbol].Invested:
self.algo.Liquidate(symbol)
self.algo.Debug(f"[Seasonal] Liquidate {symbol.Value}")
if to_hold:
weight = self.weight / len(to_hold)
for name in to_hold:
symbol = self.symbols[name]
self.algo.SetHoldings(symbol, weight)
self.algo.Debug(f"[Seasonal] Holding {name} @ {weight:.2f}")
else:
fallback = self.symbols.get("SHY", None)
if fallback:
self.algo.SetHoldings(fallback, self.weight)
self.algo.Debug(f"[Seasonal] Fallback to SHY @ {self.weight:.2f}")
self.current_holdings = to_hold
def GetCurrentExposure(self):
exposure = sum([
self.algo.Portfolio[symbol].HoldingsValue
for symbol in self.symbols.values()
]) / self.algo.Portfolio.TotalPortfolioValue
return exposure
class DefenseFirstWithUPRO:
def __init__(self, algo: QCAlgorithm, weight: float):
self.algo = algo
self.weight = weight
self.defensive_symbols = ["TLT", "GLD", "DBC", "UUP"]
self.upro_symbol = "UPRO"
self.assets = [self.algo.AddEquity(ticker, Resolution.Daily).Symbol for ticker in self.defensive_symbols + [self.upro_symbol]]
self.defensive = self.assets[:4]
self.upro = self.assets[4]
self.momentum_periods = [21, 63, 126, 252]
self.history = {symbol: RollingWindow[float](max(self.momentum_periods) + 1) for symbol in self.assets}
self.last_rebalance_month = -1
history = self.algo.History(self.assets, max(self.momentum_periods) + 1, Resolution.Daily)
for symbol in self.assets:
if symbol in history.index.levels[0]:
for price in history.loc[symbol]["close"]:
self.history[symbol].Add(float(price))
self.algo.Schedule.On(self.algo.DateRules.MonthStart(self.upro),
self.algo.TimeRules.AfterMarketOpen(self.upro, 10),
self.Rebalance)
def OnData(self, data: Slice):
for symbol in self.assets:
if data.Bars.ContainsKey(symbol):
self.history[symbol].Add(float(data.Bars[symbol].Close))
def ComputeMomentumScore(self, symbol):
window = self.history[symbol]
if not window.IsReady:
return None
current = window[0]
momenta = [(current - window[p]) / window[p] for p in self.momentum_periods if window.Count > p]
return sum(momenta) / len(momenta) if momenta else None
def Rebalance(self):
if self.algo.Time.month == self.last_rebalance_month:
return
scores = {s: self.ComputeMomentumScore(s) for s in self.defensive}
scores = {k: v for k, v in scores.items() if v is not None}
sorted_assets = sorted(scores.items(), key=lambda x: x[1], reverse=True)
weights = [0.4, 0.3, 0.2, 0.1]
risk_free = 0.045 / 12
allocation = {}
for i, (symbol, score) in enumerate(sorted_assets):
if score > risk_free:
allocation[symbol] = weights[i]
else:
allocation[self.upro] = allocation.get(self.upro, 0) + weights[i]
total_weight = sum(allocation.values())
# Nur eigene Symbole liquidieren
for symbol in self.assets:
if self.algo.Portfolio[symbol].Invested and symbol not in allocation:
self.algo.Liquidate(symbol)
self.algo.Debug(f"[Defense] Liquidate {symbol.Value}")
for symbol, w in allocation.items():
target_pct = self.weight * (w / total_weight)
self.algo.Debug(f"[Defense] SetHoldings {symbol.Value} @ {target_pct:.2%}")
self.algo.SetHoldings(symbol, target_pct)
self.last_rebalance_month = self.algo.Time.month
def GetCurrentExposure(self):
exposure = sum([
self.algo.Portfolio[symbol].HoldingsValue
for symbol in self.assets
]) / self.algo.Portfolio.TotalPortfolioValue
return exposure
from AlgorithmImports import *
class MinimalTest(QCAlgorithm):
"""
Minimal test to check if basic framework works
"""
def Initialize(self):
self.SetStartDate(2020, 1, 1)
self.SetEndDate(2020, 12, 31)
self.SetCash(100000)
self.Debug("=== Minimal Test Started ===")
# Add single symbol
self.AddEquity("SPY", Resolution.Daily)
self.Debug("=== Minimal Test Initialized ===")
def OnData(self, data):
if not self.Portfolio.Invested:
self.SetHoldings("SPY", 1.0)
self.Debug("Bought SPY")
