| Overall Statistics |
|
Total Orders 3820 Average Win 0.15% Average Loss -0.11% Compounding Annual Return 14.430% Drawdown 42.400% Expectancy 0.588 Start Equity 1000000 End Equity 1963413.20 Net Profit 96.341% Sharpe Ratio 0.438 Sortino Ratio 0.458 Probabilistic Sharpe Ratio 10.193% Loss Rate 33% Win Rate 67% Profit-Loss Ratio 1.36 Alpha -0.004 Beta 1.186 Annual Standard Deviation 0.228 Annual Variance 0.052 Information Ratio 0.12 Tracking Error 0.101 Treynor Ratio 0.084 Total Fees $2831.13 Estimated Strategy Capacity $48000000.00 Lowest Capacity Asset ERIE R735QTJ8XC9X Portfolio Turnover 0.69% |
#region imports from AlgorithmImports import * #endregion def CalculateTrendIndicators(self): """ Calculates momentum-based trading signals using historical price data. Trading Logic: 1. Signal Generation: - Analyzes price history over specified lookback period - Calculates compound returns for each security - Filters for only positive-returning securities 2. Security Selection: - Ranks securities by compound return performance - Selects top 10% of performers - Implements momentum factor strategy 3. Performance Monitoring: - Tracks average, minimum, and maximum returns - Monitors selection pool size and characteristics - Provides detailed logging of selection metrics Risk Management: - Only considers securities with positive returns - Uses compound returns to capture consistent performers - Implements percentage-based selection for diversification Returns: list: Top performing securities selected for potential investment, filtered by positive returns and ranked by performance. """ # Define the percentage of top performers to select top_pct = 0.1 # Select top 10% of securities - momentum concentration control # Calculate compounded returns for each security compounded_returns = {} for symbol, prices in self.historical_data.items(): if len(prices) >= self.lookback: # Ensure sufficient data daily_returns = prices.pct_change().dropna() compounded_return = (1 + daily_returns).prod() - 1 # Risk management: Only include positive returns if compounded_return > 0: compounded_returns[symbol] = compounded_return # Calculate and log performance statistics if compounded_returns: returns_values = list(compounded_returns.values()) avg_return = sum(returns_values) / len(returns_values) min_return = min(returns_values) max_return = max(returns_values) # Performance monitoring logs self.Debug(f"**Compounded Returns Statistics:**") self.Debug(f"Number of symbols analyzed (positive returns only): {len(compounded_returns)}") self.Debug(f"Population average return: {avg_return:.2%}") self.Debug(f"Population min return: {min_return:.2%}") self.Debug(f"Population max return: {max_return:.2%}") # Select top performing symbols using momentum strategy top_symbols = sorted(compounded_returns, key=compounded_returns.get, reverse=True) top_symbols = top_symbols[:int(len(compounded_returns) * top_pct)] # Log selection statistics self.Debug(f"**Selected {len(top_symbols)} top symbols out of {len(compounded_returns)} total symbols (all positive returns)**") if top_symbols: top_returns = [compounded_returns[symbol] for symbol in top_symbols] self.Debug(f"Selected symbols average return: {sum(top_returns) / len(top_returns):.2%}") self.Debug(f"Selected symbols return range: {min(top_returns):.2%} to {max(top_returns):.2%}") return top_symbols
#region imports from AlgorithmImports import * from pypfopt import BlackLittermanModel import pandas as pd import numpy as np #endregion def OptimizePortfolio(self, total_returns, volatilities): """ Implements a sophisticated portfolio optimization strategy using risk-adjusted returns. Trading Logic: 1. Weight Calculation: - Uses return-to-volatility ratios for position sizing - Implements risk-adjusted allocation strategy - Normalizes weights to respect leverage constraints 2. Risk Management: - Monitors for negative returns/volatilities - Implements position size limits - Ensures portfolio leverage constraints 3. Portfolio Construction: - Calculates optimal position sizes - Applies leverage and concentration limits - Normalizes allocations to meet constraints 4. Performance Monitoring: - Tracks allocation statistics - Monitors position concentration - Logs portfolio characteristics Args: total_returns (pd.Series): Historical returns for each asset volatilities (pd.Series): Calculated volatilities for each asset Returns: dict: Optimized portfolio weights for each symbol, respecting all risk and leverage constraints """ # Monitor for data quality issues for symbol in total_returns.index: if total_returns[symbol] < 0: self.Debug(f"Warning: Negative return detected for {symbol}: {total_returns[symbol]:.4f}") if volatilities[symbol] < 0: self.Debug(f"Warning: Negative volatility detected for {symbol}: {volatilities[symbol]:.4f}") # Calculate risk-adjusted return scores epsilon = 1e-8 # Numerical stability factor return_to_vol = total_returns / (volatilities + epsilon) # Calculate initial weights using risk-adjusted scores total_score = return_to_vol.sum() weights = return_to_vol / total_score if total_score != 0 else pd.Series(0, index=return_to_vol.index) # Apply leverage constraints for risk management max_allocation = self.universe_settings.leverage # Scale weights to meet leverage constraints total_weight = sum(abs(weight) for weight in weights) scaling_factor = max_allocation / total_weight if total_weight > max_allocation else 1.0 # Create final weight dictionary with constraints applied normalized_weights = {symbol: weight * scaling_factor for symbol, weight in weights.items()} # Log portfolio construction metrics self.Debug(f"**Portfolio Allocation Statistics:**") self.Debug(f"Number of positions: {len(normalized_weights)}") self.Debug(f"Average position size: {np.mean(list(normalized_weights.values())):.2%}") self.Debug(f"Position size range: {min(normalized_weights.values()):.2%} to {max(normalized_weights.values()):.2%}") self.Debug(f"Total allocation: {sum(normalized_weights.values()):.2%}") return normalized_weights
#region imports from pypfopt import risk_models, expected_returns from AlgorithmImports import * import numpy as np import pandas as pd #endregion def CalculateRiskParameters(self, top_symbols): """ Calculates comprehensive risk metrics for portfolio management. Trading Logic: 1. Risk Assessment: - Calculates individual security volatilities - Computes historical returns for risk-adjusted metrics - Analyzes return distributions 2. Data Validation: - Ensures data availability for each symbol - Handles missing data scenarios - Provides warning logs for data issues 3. Risk Metrics: - Calculates daily returns and volatilities - Annualizes volatility metrics (252 trading days) - Computes total period returns 4. Risk Monitoring: - Tracks average volatility levels - Monitors volatility extremes - Reports data quality metrics Args: top_symbols (list): List of symbols to analyze for risk metrics Returns: tuple: (returns, volatilities) containing: - returns: pandas Series of total period returns - volatilities: pandas Series of annualized volatilities Used for portfolio optimization and position sizing """ # Filter and validate historical data selected_data = {} for symbol in top_symbols: if symbol in self.historical_data: selected_data[symbol] = self.historical_data[symbol] else: self.Debug(f"Warning: No historical data found for {symbol}") # Data validation check if not selected_data: self.Debug("Error: No valid historical data found for any symbols") return pd.Series(), pd.Series() # Convert to DataFrame for calculations selected_history = pd.DataFrame(selected_data) # Calculate risk metrics daily_returns = selected_history.pct_change().dropna() total_returns = (1 + daily_returns).prod() - 1 # Annualize volatility for risk scaling volatilities = daily_returns.std() * np.sqrt(252) # Annualization factor # Risk monitoring logs self.Debug(f"**Volatility Statistics:**") self.Debug(f"Average Annualized Volatility: {volatilities.mean():.2%}") self.Debug(f"Min Volatility: {volatilities.min():.2%}") self.Debug(f"Max Volatility: {volatilities.max():.2%}") # Data quality metrics self.Debug(f"Number of symbols analyzed: {len(selected_data)}") self.Debug(f"Data points per symbol: {len(daily_returns)}") return total_returns, volatilities
#region imports from AlgorithmImports import * #endregion def Execute_Trades(self, position_list): """ Implements an intelligent trade execution strategy with capital efficiency. Trading Logic: 1. Trade Sequencing: - Prioritizes position reductions before increases - Orders increases by size (smallest first) - Optimizes capital utilization 2. Position Management: - Separates increase/decrease trades - Tracks current vs target positions - Manages position adjustments efficiently 3. Risk Management: - Executes reductions first to free capital - Controls position sizing - Implements smart order routing 4. Execution Strategy: - Uses SetHoldings for position management - Implements size-based execution ordering - Optimizes trade sequence Args: position_list (dict): Target portfolio weights for each symbol Format: {'AAPL': 0.25, 'GOOGL': 0.25} """ self.Debug("***Placing Trades***") # Separate trades by direction for optimal execution reduce_trades = {} increase_trades = {} # Analyze current vs target positions for symbol, target_weight in position_list.items(): holding = self.Portfolio[symbol] current_weight = holding.HoldingsValue / self.Portfolio.TotalPortfolioValue if holding.Invested else 0 # Categorize trades by direction if current_weight > target_weight: reduce_trades[symbol] = target_weight elif current_weight < target_weight: increase_trades[symbol] = target_weight # Execute reductions first to free up capital for symbol, weight in reduce_trades.items(): self.SetHoldings(symbol, weight) # Execute increases in order of size (smallest first) sorted_increases = sorted(increase_trades.items(), key=lambda x: x[1]) for symbol, weight in sorted_increases: self.SetHoldings(symbol, weight) def Exit_Positions(self, position_list): """ Manages complete position exits and portfolio cleanup. Trading Logic: 1. Position Analysis: - Compares current holdings to target portfolio - Identifies positions for complete exit - Maintains portfolio alignment 2. Exit Management: - Executes complete liquidations - Cleans up legacy positions - Ensures portfolio accuracy 3. Risk Management: - Removes unwanted exposure - Maintains clean portfolio structure - Implements efficient liquidation Args: position_list (dict): Dictionary of target positions Any holding not in this list will be liquidated """ # Review current holdings against target portfolio for holding in self.Portfolio.Values: # Liquidate positions not in target portfolio if holding.Symbol not in position_list and holding.Invested: self.Liquidate(holding.Symbol)
# region imports from AlgorithmImports import * from Alpha_Models import CalculateTrendIndicators from Risk_Models import CalculateRiskParameters from Portfolio_Construction import OptimizePortfolio from Trade_Execution import Execute_Trades, Exit_Positions # endregion class NCSU_Strategy_2025_Q2(QCAlgorithm): """ A sophisticated quantitative trading strategy implementing a trend-following approach with comprehensive risk management. Trading Logic Overview: 1. Universe Selection: Uses SPY ETF constituents to focus on liquid, large-cap stocks 2. Signal Generation: Implements trend following by analyzing compound returns over a 63-day lookback period 3. Risk Management: - Employs position sizing limits - Implements drawdown protection - Uses volatility-adjusted position sizing 4. Portfolio Construction: - Optimizes weights using return-to-volatility ratios - Applies leverage constraints (max 2.0x) - Manages concentration risk 5. Trade Execution: - Prioritizes reducing positions before increasing - Implements transaction cost consideration - Uses smart order routing with SetHoldings Key Parameters: - Lookback Period: 63 days - Rebalance Threshold: 4% - Maximum Leverage: 2.0x - Initial Capital: $1,000,000 """ def Initialize(self): """ Initializes the strategy with core settings and parameters. This method: 1. Sets the backtest period and initial capital 2. Configures the investment universe using SPY constituents 3. Establishes risk parameters and trading thresholds 4. Initializes tracking variables for portfolio management The initialization ensures all components are properly set up before trading begins. """ # Set basic algorithm parameters self.SetStartDate(2020, 1, 1) self.SetEndDate(2024, 12, 31) self.SetCash(1000000) # Define the universe using SPY ETF constituents self.etf = "SPY" # Set maximum leverage for the strategy self.universe_settings.leverage = 2.0 # Add ETF and set up universe selection self.AddEquity(self.etf, Resolution.Daily) self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverseSelection(ETFConstituentsUniverseSelectionModel(self.etf)) # Initialize strategy parameters self.historical_data = {} # Store historical prices self.lookback = 63 # Lookback period for calculations # Risk management parameters self.equity_high_water_mark = self.Portfolio.TotalPortfolioValue self.rebalance_threshold = 0.04 # 4% rebalance threshold self.rebalance = True self.current_equity = self.Portfolio.TotalPortfolioValue def OnSecuritiesChanged(self, changes): """ Manages the dynamic universe of tradeable securities. Trading Logic: 1. For new securities: - Retrieves historical data for signal generation - Adds to tracking system for ongoing monitoring 2. For removed securities: - Liquidates any existing positions - Triggers rebalance to reallocate capital - Cleans up historical data tracking Args: changes (SecurityChanges): Contains lists of added and removed securities """ # Add new securities to historical data tracking for security in changes.AddedSecurities: try: history = self.History(security.Symbol, self.lookback, Resolution.Daily) if not history.empty and 'close' in history: self.historical_data[security.Symbol] = history['close'] else: self.Debug(f"No historical data available for {security.Symbol}") continue except Exception as e: self.Debug(f"Error getting historical data for {security.Symbol}: {str(e)}") continue # Clean up removed securities for security in changes.RemovedSecurities: if security.Symbol in self.historical_data: del self.historical_data[security.Symbol] # Liquidate positions in removed securities if self.Portfolio[security.Symbol].Invested: self.Liquidate(security.Symbol) self.Debug(f"Liquidating {security.Symbol} as it is removed from the ETF") # Rebalance portfolio when positions are removed self.Debug(f"Reblancing due to postion liquidation") self.Rebalance() def OnData(self, data): """ Primary market data handler and portfolio monitoring system. Trading Logic: 1. Monitors portfolio value against high-water mark 2. Triggers rebalancing when: - Portfolio value deviates by more than 4% from high-water mark - Manual rebalance flag is set 3. Updates portfolio tracking metrics 4. Implements drawdown protection by monitoring equity ratios Args: data: Current market data slice """ # Update current equity value self.current_equity = self.Portfolio.TotalPortfolioValue # Calculate portfolio rebalance percentage abs_equity_ratio = abs(self.current_equity / self.equity_high_water_mark) # Log equity values and ratios self.Debug(f"**Portfolio Status:**") self.Debug(f"Current Equity: ${self.current_equity:,.2f}") self.Debug(f"High Water Mark: ${self.equity_high_water_mark:,.2f}") self.Debug(f"Absolute Equity Ratio: {abs_equity_ratio:.4f}") # Check for rebalance condition if abs(abs_equity_ratio - 1) >= self.rebalance_threshold or self.rebalance==True: self.Debug(f"**ALERT: Rebalance Threshold {self.rebalance_threshold:.2%}. Rebalancing...**") self.rebalance=True self.Rebalance() self.equity_high_water_mark = self.current_equity # Reset high water mark def Rebalance(self): """ Executes the complete portfolio rebalancing process. Trading Logic Flow: 1. Alpha Signal Generation: - Calculates trend indicators using compound returns - Identifies top-performing securities 2. Risk Assessment: - Calculates volatility metrics - Evaluates return patterns 3. Portfolio Optimization: - Adjusts returns for transaction costs - Optimizes weights based on risk-adjusted returns 4. Trade Execution: - Exits unnecessary positions - Executes new position targets efficiently """ self.Debug(f"Rebalancing on {self.Time}") # Get alpha signals sorted_symbols = CalculateTrendIndicators(self) # Calculate risk parameters #mu, S = CalculateRiskParameters(self, top_symbols=sorted_symbols) total_returns, volatilities = CalculateRiskParameters(self, top_symbols=sorted_symbols) # Adjust expected returns for transaction costs transaction_cost = 0.001 # 0.1% per trade for symbol in sorted_symbols: if symbol in total_returns: total_returns[symbol] -= transaction_cost else: self.Debug(f"Symbol {symbol} not found in total_returns") # Optimize portfolio target_positions = OptimizePortfolio(self, total_returns=total_returns, volatilities=volatilities) # Execute trades Exit_Positions(self, position_list=target_positions) Execute_Trades(self, position_list=target_positions) def OnOrderEvent(self, orderEvent): """ Handles order execution feedback and portfolio state management. Trading Logic: 1. Monitors order execution status 2. Resets rebalancing flag after trades complete 3. Enables tracking of portfolio changes Args: orderEvent (OrderEvent): Details of the executed order including fill price, quantity, and execution time """ self.rebalance = False # Reset rebalanced flag after trades execute
