| Overall Statistics |
|
Total Orders 912 Average Win 0.03% Average Loss -0.02% Compounding Annual Return 13.447% Drawdown 8.400% Expectancy 0.406 Start Equity 1000000 End Equity 1133685.16 Net Profit 13.369% Sharpe Ratio 0.525 Sortino Ratio 0.726 Probabilistic Sharpe Ratio 55.211% Loss Rate 40% Win Rate 60% Profit-Loss Ratio 1.33 Alpha -0.047 Beta 0.687 Annual Standard Deviation 0.082 Annual Variance 0.007 Information Ratio -1.846 Tracking Error 0.048 Treynor Ratio 0.063 Total Fees $935.64 Estimated Strategy Capacity $58000000.00 Lowest Capacity Asset REG R735QTJ8XC9X Portfolio Turnover 0.75% Drawdown Recovery 147 |
from AlgorithmImports import *
class SP500FundamentalsOnlyStrategy(QCAlgorithm):
"""
S&P 500 Growth vs Value Strategy - Fundamentals Only Version
Growth Score (Higher = Better):
- YoY Revenue Growth: 70% weight
- ROE: 30% weight (quality growth proxy)
Value Score (Lower = Better):
- P/E Ratio: 75% weight
- ROE: 25% weight (inverted - higher ROE = lower score)
Portfolio: Top 200 Growth + Top 200 Value with tier weighting
"""
def initialize(self):
self.set_start_date(2023, 1, 1)
self.set_end_date(2023, 12, 31)
self.set_cash(1000000)
self.set_benchmark("SPY")
# S&P 500 universe via SPY ETF constituents
self.spy_symbol = self.add_equity("SPY", Resolution.DAILY).symbol
self.add_universe(self.universe.etf(self.spy_symbol, self.universe_settings))
# Strategy parameters
self.growth_bucket_weight = 0.5
self.value_bucket_weight = 0.5
self.rebalance_frequency = timedelta(days=7) # Weekly rebalancing (more practical)
# Data tracking
self.universe_symbols = []
self.last_rebalance = datetime.min
# Schedule weekly rebalancing
self.schedule.on(
self.date_rules.week_start("SPY"),
self.time_rules.at(10, 0),
self.rebalance_strategy
)
self.debug("S&P 500 Fundamentals-Only strategy initialized")
def on_securities_changed(self, changes):
"""Handle universe changes - no indicators to create"""
# Remove old securities
for removed in changes.removed_securities:
symbol = removed.symbol
if symbol in self.universe_symbols:
self.universe_symbols.remove(symbol)
# Add new securities
for added in changes.added_securities:
symbol = added.symbol
if symbol != self.spy_symbol:
self.universe_symbols.append(symbol)
self.debug(f"Universe updated: {len(self.universe_symbols)} stocks")
def rebalance_strategy(self):
"""Main rebalancing logic using only fundamentals"""
if (self.time - self.last_rebalance) < self.rebalance_frequency:
return
self.debug(f"Rebalancing at {self.time}")
# Calculate scores for all stocks
growth_scores = {}
value_scores = {}
processed_count = 0
for symbol in self.universe_symbols:
try:
fund = self.securities[symbol].fundamentals
if fund.has_fundamental_data:
processed_count += 1
# Calculate growth score
growth_score = self.calculate_growth_score_fundamental(fund)
if growth_score is not None:
growth_scores[symbol] = growth_score
# Calculate value score
value_score = self.calculate_value_score_fundamental(fund)
if value_score is not None:
value_scores[symbol] = value_score
except Exception:
continue
self.debug(f"Processed {processed_count} stocks with fundamentals")
self.debug(f"Valid scores - Growth: {len(growth_scores)}, Value: {len(value_scores)}")
if len(growth_scores) < 50 or len(value_scores) < 50:
self.debug("Insufficient data for rebalancing")
return
# Rank stocks
growth_rankings = self.rank_stocks(growth_scores, ascending=False)
value_rankings = self.rank_stocks(value_scores, ascending=True)
# Select top 200 from each strategy
top_200_growth = growth_rankings[:200]
top_200_value = value_rankings[:200]
# Calculate position weights with tier structure
growth_weights = self.calculate_tier_weights(top_200_growth, self.growth_bucket_weight)
value_weights = self.calculate_tier_weights(top_200_value, self.value_bucket_weight)
# Combine all target positions
all_target_weights = {**growth_weights, **value_weights}
# Execute rebalancing
successful_trades = self.rebalance_to_targets(all_target_weights)
# Enhanced logging
self.debug(f"Rebalanced: {len(growth_weights)} growth + {len(value_weights)} value positions")
self.debug(f"Successfully executed {successful_trades}/{len(all_target_weights)} trades")
# Log top picks with scores for verification
if len(growth_rankings) >= 10:
self.debug("=== TOP 5 GROWTH PICKS ===")
for i, symbol in enumerate(growth_rankings[:5]):
ticker = self.get_ticker(symbol)
score = growth_scores[symbol]
weight = growth_weights.get(symbol, 0)
self.debug(f"{i+1}. {ticker}: Score={score:.1f}, Weight={weight:.2%}")
if len(value_rankings) >= 10:
self.debug("=== TOP 5 VALUE PICKS ===")
for i, symbol in enumerate(value_rankings[:5]):
ticker = self.get_ticker(symbol)
score = value_scores[symbol]
weight = value_weights.get(symbol, 0)
self.debug(f"{i+1}. {ticker}: Score={score:.1f}, Weight={weight:.2%}")
self.last_rebalance = self.time
def calculate_growth_score_fundamental(self, fund):
"""
Calculate growth score using only fundamentals (higher = better)
70% YoY Revenue Growth + 30% ROE (quality growth proxy)
"""
try:
# Revenue growth component (70% weight)
revenue_growth = fund.operation_ratios.revenue_growth.one_year
if revenue_growth <= 0:
growth_component = 0
else:
growth_component = min(revenue_growth * 100, 50) # Cap at 50% for scoring
# ROE component (30% weight) - higher ROE = better quality growth
roe = fund.operation_ratios.roe.value
if roe <= 0:
roe_component = 0
else:
roe_component = min(roe * 100, 50) # Cap at 50% ROE for scoring
# Calculate weighted growth score
growth_score = (growth_component * 0.70) + (roe_component * 0.30)
return growth_score
except Exception:
return None
def calculate_value_score_fundamental(self, fund):
"""
Calculate value score using only fundamentals (lower = better)
75% P/E Ratio + 25% ROE (inverted)
"""
try:
# P/E Ratio component (75% weight) - lower is better
pe_ratio = fund.valuation_ratios.pe_ratio
if pe_ratio <= 0 or pe_ratio > 100:
pe_component = 100 # High penalty
else:
pe_component = pe_ratio
# ROE component (25% weight) - higher ROE is better, so invert
roe = fund.operation_ratios.roe.value
if roe <= 0:
roe_component = 50 # Penalty for negative ROE
else:
roe_component = max(0, 50 - min(roe * 100, 50)) # Invert and cap
# Calculate weighted value score (lower = better value)
value_score = (pe_component * 0.75) + (roe_component * 0.25)
return value_score
except Exception:
return None
def rank_stocks(self, scores_dict, ascending=True):
"""Rank stocks by scores"""
sorted_items = sorted(scores_dict.items(), key=lambda x: x[1], reverse=not ascending)
return [symbol for symbol, score in sorted_items]
def calculate_tier_weights(self, ranked_symbols, total_bucket_weight):
"""Calculate tier-based weights within a bucket"""
weights = {}
# Tier 1: Ranks 1-50 get 50% of bucket weight
tier1_weight = (total_bucket_weight * 0.50) / 50
for i in range(min(50, len(ranked_symbols))):
weights[ranked_symbols[i]] = tier1_weight
# Tier 2: Ranks 51-100 get 25% of bucket weight
tier2_weight = (total_bucket_weight * 0.25) / 50
for i in range(50, min(100, len(ranked_symbols))):
weights[ranked_symbols[i]] = tier2_weight
# Tier 3: Ranks 101-200 get 25% of bucket weight
tier3_weight = (total_bucket_weight * 0.25) / 100
for i in range(100, min(200, len(ranked_symbols))):
weights[ranked_symbols[i]] = tier3_weight
return weights
def rebalance_to_targets(self, target_weights):
"""Execute rebalancing to target weights"""
# Liquidate positions not in target
current_holdings = {kvp.key for kvp in self.portfolio if kvp.value.invested}
for symbol in current_holdings:
if symbol not in target_weights:
self.liquidate(symbol)
# Set target positions
successful_trades = 0
failed_trades = 0
for symbol, target_weight in target_weights.items():
if target_weight > 0.0001: # Minimum position size
try:
# Check if security has valid price data
if symbol in self.securities and self.securities[symbol].price > 0:
self.set_holdings(symbol, target_weight)
successful_trades += 1
else:
failed_trades += 1
except Exception:
failed_trades += 1
continue
if failed_trades > 0:
self.debug(f"Failed to place {failed_trades} orders (likely due to missing price data)")
return successful_trades
def get_ticker(self, symbol):
"""Extract ticker from symbol"""
return str(symbol).split(' ')[0]
def on_data(self, data):
"""Handle incoming data"""
pass
# from AlgorithmImports import *
# class SP500GrowthValueStrategy(QCAlgorithm):
# """
# Complete S&P 500 Growth vs Value Strategy
# Growth Score (Higher = Better):
# - YoY Revenue Growth: 50% weight
# - RSI Level: 30% weight (higher RSI = momentum)
# - Price vs 20-day MA: 20% weight
# Value Score (Lower = Better):
# - P/E Ratio: 75% weight
# - ROE: 25% weight (inverted - higher ROE = lower score)
# Portfolio: Top 200 Growth + Top 200 Value
# Tier Weighting:
# - Ranks 1-50: 50% of bucket weight
# - Ranks 51-100: 25% of bucket weight
# - Ranks 101-200: 25% of bucket weight
# """
# def initialize(self):
# self.set_start_date(2023, 1, 1)
# self.set_end_date(2023, 12, 31)
# self.set_cash(1000000)
# self.set_benchmark("SPY")
# # S&P 500 universe via SPY ETF constituents
# self.spy_symbol = self.add_equity("SPY", Resolution.DAILY).symbol
# self.add_universe(self.universe.etf(self.spy_symbol, self.universe_settings))
# # Strategy parameters
# self.growth_bucket_weight = 0.5 # 50% to growth
# self.value_bucket_weight = 0.5 # 50% to value
# self.rebalance_frequency = timedelta(hours=1) # Hourly rebalancing as requested
# # Data tracking
# self.universe_symbols = []
# self.technical_indicators = {}
# self.last_rebalance = datetime.min
# # Schedule rebalancing every hour during market hours
# self.schedule.on(
# self.date_rules.every_day("SPY"),
# self.time_rules.every(timedelta(hours=1)),
# self.rebalance_strategy
# )
# self.debug("S&P 500 Growth vs Value strategy initialized")
# def on_securities_changed(self, changes):
# """Handle universe changes"""
# # Remove old securities and indicators
# for removed in changes.removed_securities:
# symbol = removed.symbol
# if symbol in self.universe_symbols:
# self.universe_symbols.remove(symbol)
# if symbol in self.technical_indicators:
# del self.technical_indicators[symbol]
# # Add new securities and create indicators
# for added in changes.added_securities:
# symbol = added.symbol
# if symbol != self.spy_symbol: # Don't include SPY itself
# self.universe_symbols.append(symbol)
# # Create technical indicators using string ticker (to avoid RSI syntax error)
# ticker = str(symbol).split(' ')[0]
# try:
# self.technical_indicators[symbol] = {
# 'rsi': self.rsi(ticker, 14),
# 'sma': self.sma(ticker, 20),
# 'ticker': ticker
# }
# except Exception as e:
# self.debug(f"Failed to create indicators for {ticker}: {e}")
# self.debug(f"Universe updated: {len(self.universe_symbols)} stocks")
# def rebalance_strategy(self):
# """Main rebalancing logic"""
# if (self.time - self.last_rebalance) < self.rebalance_frequency:
# return
# self.debug(f"Rebalancing at {self.time}")
# # Calculate scores for all stocks
# growth_scores = {}
# value_scores = {}
# for symbol in self.universe_symbols:
# try:
# growth_score = self.calculate_growth_score(symbol)
# value_score = self.calculate_value_score(symbol)
# if growth_score is not None:
# growth_scores[symbol] = growth_score
# if value_score is not None:
# value_scores[symbol] = value_score
# except Exception as e:
# continue # Skip problematic stocks
# if len(growth_scores) < 50 or len(value_scores) < 50:
# self.debug(f"Insufficient data: {len(growth_scores)} growth, {len(value_scores)} value")
# return
# # Rank stocks (1-500 for each strategy)
# growth_rankings = self.rank_stocks(growth_scores, ascending=False) # Higher score = better
# value_rankings = self.rank_stocks(value_scores, ascending=True) # Lower score = better
# # Select top 200 from each strategy
# top_200_growth = growth_rankings[:200]
# top_200_value = value_rankings[:200]
# # Calculate position weights with tier structure
# growth_weights = self.calculate_tier_weights(top_200_growth, self.growth_bucket_weight)
# value_weights = self.calculate_tier_weights(top_200_value, self.value_bucket_weight)
# # Combine all target positions
# all_target_weights = {**growth_weights, **value_weights}
# # Execute rebalancing
# self.rebalance_to_targets(all_target_weights)
# # Log summary
# self.debug(f"Rebalanced: {len(growth_weights)} growth + {len(value_weights)} value positions")
# # Log top picks for verification
# if len(growth_rankings) >= 10:
# top_growth_tickers = [self.get_ticker(s) for s in growth_rankings[:5]]
# self.debug(f"Top 5 Growth: {top_growth_tickers}")
# if len(value_rankings) >= 10:
# top_value_tickers = [self.get_ticker(s) for s in value_rankings[:5]]
# self.debug(f"Top 5 Value: {top_value_tickers}")
# self.last_rebalance = self.time
# def calculate_growth_score(self, symbol):
# """
# Calculate growth score (higher = better)
# 50% YoY Revenue Growth + 30% RSI + 20% Price vs 20-day MA
# """
# try:
# fund = self.securities[symbol].fundamentals
# if not fund.has_fundamental_data:
# return None
# # Component 1: YoY Revenue Growth (50% weight)
# revenue_growth = fund.operation_ratios.revenue_growth.one_year
# if revenue_growth <= 0:
# growth_component = 0
# else:
# growth_component = min(revenue_growth * 100, 100) # Cap at 100% for scoring
# # Component 2: RSI Level (30% weight) - Higher RSI = momentum
# rsi_component = 0
# if symbol in self.technical_indicators:
# rsi_indicator = self.technical_indicators[symbol]['rsi']
# if rsi_indicator.is_ready:
# rsi_component = rsi_indicator.current.value # Already 0-100 scale
# # Component 3: Price vs 20-day MA (20% weight)
# ma_component = 0
# if symbol in self.technical_indicators:
# sma_indicator = self.technical_indicators[symbol]['sma']
# if sma_indicator.is_ready:
# current_price = self.securities[symbol].price
# ma_value = sma_indicator.current.value
# if ma_value > 0:
# price_ma_ratio = ((current_price / ma_value) - 1) * 100
# ma_component = max(0, price_ma_ratio + 10) # Shift to make positive
# # Calculate weighted growth score
# growth_score = (
# growth_component * 0.50 +
# rsi_component * 0.30 +
# ma_component * 0.20
# )
# return growth_score
# except Exception:
# return None
# def calculate_value_score(self, symbol):
# """
# Calculate value score (lower = better)
# 75% P/E Ratio + 25% ROE (inverted)
# """
# try:
# fund = self.securities[symbol].fundamentals
# if not fund.has_fundamental_data:
# return None
# # Component 1: P/E Ratio (75% weight) - Lower is better
# pe_ratio = fund.valuation_ratios.pe_ratio
# if pe_ratio <= 0 or pe_ratio > 150: # Filter extreme P/E ratios
# pe_component = 150 # High penalty
# else:
# pe_component = pe_ratio
# # Component 2: ROE (25% weight) - Higher ROE is better, so invert for scoring
# roe = fund.operation_ratios.roe.value
# if roe <= 0:
# roe_component = 100 # Penalty for negative/zero ROE
# else:
# # Invert ROE: higher ROE = lower component score
# roe_component = max(0, 100 - min(roe * 100, 100))
# # Calculate weighted value score (lower = better value)
# value_score = (
# pe_component * 0.75 +
# roe_component * 0.25
# )
# return value_score
# except Exception:
# return None
# def rank_stocks(self, scores_dict, ascending=True):
# """Rank stocks by scores and return ordered list"""
# sorted_items = sorted(scores_dict.items(), key=lambda x: x[1], reverse=not ascending)
# return [symbol for symbol, score in sorted_items]
# def calculate_tier_weights(self, ranked_symbols, total_bucket_weight):
# """
# Calculate tier-based weights within a bucket
# Ranks 1-50: 50% of bucket weight (1% each)
# Ranks 51-100: 25% of bucket weight (0.5% each)
# Ranks 101-200: 25% of bucket weight (0.25% each)
# """
# weights = {}
# # Tier 1: Ranks 1-50 get 50% of bucket weight
# tier1_individual_weight = (total_bucket_weight * 0.50) / 50 # 0.5% each if 50% bucket
# for i in range(min(50, len(ranked_symbols))):
# weights[ranked_symbols[i]] = tier1_individual_weight
# # Tier 2: Ranks 51-100 get 25% of bucket weight
# tier2_individual_weight = (total_bucket_weight * 0.25) / 50 # 0.25% each if 50% bucket
# for i in range(50, min(100, len(ranked_symbols))):
# weights[ranked_symbols[i]] = tier2_individual_weight
# # Tier 3: Ranks 101-200 get 25% of bucket weight
# tier3_individual_weight = (total_bucket_weight * 0.25) / 100 # 0.125% each if 50% bucket
# for i in range(100, min(200, len(ranked_symbols))):
# weights[ranked_symbols[i]] = tier3_individual_weight
# return weights
# def rebalance_to_targets(self, target_weights):
# """Execute rebalancing to target weights"""
# # Get current holdings
# current_holdings = {kvp.key: kvp.value.holdings_value / self.portfolio.total_portfolio_value
# for kvp in self.portfolio if kvp.value.invested}
# # Liquidate positions not in target
# for symbol in current_holdings:
# if symbol not in target_weights:
# self.liquidate(symbol)
# # Set target positions
# successful_trades = 0
# for symbol, target_weight in target_weights.items():
# if target_weight > 0.0001: # Minimum 0.01% position
# try:
# self.set_holdings(symbol, target_weight)
# successful_trades += 1
# except Exception:
# continue # Skip if order fails
# self.debug(f"Successfully placed {successful_trades} orders out of {len(target_weights)} targets")
# def get_ticker(self, symbol):
# """Extract ticker from symbol"""
# return str(symbol).split(' ')[0]
# def on_data(self, data):
# """Handle incoming data - indicators auto-update"""
# pass