| Overall Statistics |
|
Total Orders 2145 Average Win 0.02% Average Loss -0.01% Compounding Annual Return 10.901% Drawdown 8.700% Expectancy 0.465 Start Equity 1000000 End Equity 1108378.88 Net Profit 10.838% Sharpe Ratio 0.34 Sortino Ratio 0.477 Probabilistic Sharpe Ratio 50.162% Loss Rate 41% Win Rate 59% Profit-Loss Ratio 1.47 Alpha -0.054 Beta 0.604 Annual Standard Deviation 0.074 Annual Variance 0.005 Information Ratio -1.912 Tracking Error 0.055 Treynor Ratio 0.042 Total Fees $1529.91 Estimated Strategy Capacity $54000000.00 Lowest Capacity Asset MRK R735QTJ8XC9X Portfolio Turnover 1.41% Drawdown Recovery 150 |
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
# from AlgorithmImports import *
# class PureMorningstarTest(QCAlgorithm):
# """
# Simple test to verify Morningstar fundamentals are working
# No technical indicators, no complex logic - just fundamental data testing
# """
# def initialize(self):
# self.set_start_date(2023, 1, 1)
# self.set_end_date(2023, 6, 1) # Short test period
# self.set_cash(100000)
# # Test with just a few well-known stocks
# self.test_symbols = ["AAPL", "MSFT", "GOOGL", "JPM", "JNJ"]
# self.symbols = []
# for ticker in self.test_symbols:
# symbol = self.add_equity(ticker, Resolution.DAILY).symbol
# self.symbols.append(symbol)
# self.debug(f"Added {ticker}")
# # Schedule weekly fundamental data check
# self.schedule.on(
# self.date_rules.week_start(),
# self.time_rules.at(10, 0),
# self.test_fundamentals
# )
# self.debug("Morningstar fundamentals test initialized")
# def test_fundamentals(self):
# """Test fundamental data access and log results"""
# self.debug(f"=== TESTING FUNDAMENTALS AT {self.time} ===")
# for symbol in self.symbols:
# ticker = str(symbol).split(' ')[0]
# try:
# fund = self.securities[symbol].fundamentals
# if fund.has_fundamental_data:
# # Test basic valuation ratios
# pe_ratio = fund.valuation_ratios.pe_ratio
# pb_ratio = fund.valuation_ratios.pb_ratio
# ps_ratio = fund.valuation_ratios.ps_ratio
# # Test profitability ratios
# roe = fund.operation_ratios.roe.value
# roa = fund.operation_ratios.roa.value
# # Test growth metrics
# revenue_growth = fund.operation_ratios.revenue_growth.one_year
# # Test financial statement data
# market_cap = fund.market_cap
# total_revenue = fund.financial_statements.income_statement.total_revenue.value
# # Test sector/industry classification
# sector_code = fund.asset_classification.morningstar_sector_code
# industry_code = fund.asset_classification.morningstar_industry_group_code
# # Log all the data
# self.debug(f"{ticker} FUNDAMENTALS:")
# self.debug(f" Market Cap: ${market_cap:,.0f}")
# self.debug(f" P/E Ratio: {pe_ratio:.2f}")
# self.debug(f" P/B Ratio: {pb_ratio:.2f}")
# self.debug(f" P/S Ratio: {ps_ratio:.2f}")
# self.debug(f" ROE: {roe:.2%}")
# self.debug(f" ROA: {roa:.2%}")
# self.debug(f" Revenue Growth: {revenue_growth:.2%}")
# self.debug(f" Total Revenue: ${total_revenue:,.0f}")
# self.debug(f" Sector Code: {sector_code}")
# self.debug(f" Industry Code: {industry_code}")
# # Test if we can use this data for scoring
# if pe_ratio > 0 and roe > 0:
# value_score = pe_ratio / (roe * 100) # Simple PEG-like ratio
# self.debug(f" Value Score: {value_score:.2f}")
# # Test growth scoring
# if revenue_growth > 0:
# growth_score = revenue_growth * 100 + roe * 50
# self.debug(f" Growth Score: {growth_score:.1f}")
# else:
# self.debug(f"{ticker}: No fundamental data available")
# except Exception as e:
# self.debug(f"{ticker}: Error accessing fundamentals - {str(e)}")
# self.debug("=== END FUNDAMENTALS TEST ===")
# def on_data(self, data):
# """Simple data handler - no trading, just testing"""
# pass