| Overall Statistics |
|
Total Orders 451 Average Win 2.77% Average Loss -1.69% Compounding Annual Return 33.269% Drawdown 34.200% Expectancy 0.846 Start Equity 30000 End Equity 690619.07 Net Profit 2202.064% Sharpe Ratio 1.045 Sortino Ratio 1.18 Probabilistic Sharpe Ratio 52.657% Loss Rate 30% Win Rate 70% Profit-Loss Ratio 1.64 Alpha 0.131 Beta 1.191 Annual Standard Deviation 0.219 Annual Variance 0.048 Information Ratio 1.046 Tracking Error 0.141 Treynor Ratio 0.192 Total Fees $1270.92 Estimated Strategy Capacity $150000000.00 Lowest Capacity Asset MSFT R735QTJ8XC9X Portfolio Turnover 3.59% |
# region imports
from AlgorithmImports import *
# endregion
class MarketCapFactor:
def __init__(self, security):
self._security = security
@property
def value(self):
return self._security.fundamentals.market_cap
class SortinoFactor:
def __init__(self, algorithm, symbol, lookback):
self._sortino = algorithm.sortino(symbol, lookback, resolution=Resolution.DAILY)
@property
def value(self):
return self._sortino.current.value
class KERFactor:
def __init__(self, algorithm, symbol, lookback):
self._ker = algorithm.ker(symbol, lookback, resolution=Resolution.DAILY)
@property
def value(self):
return self._ker.current.value
class CorrFactor:
def __init__(self, algorithm, symbol, reference, lookback):
self._c = algorithm.c(symbol, reference, lookback, correlation_type=CorrelationType.Pearson, resolution=Resolution.DAILY)
@property
def value(self):
return 1 - abs(self._c.current.value)
class ROCFactor:
def __init__(self, algorithm, symbol, lookback):
self._roc = algorithm.roc(symbol, lookback, resolution=Resolution.DAILY)
@property
def value(self):
return self._roc.current.value
class QualityFactor:
"""Test quality/profitability factor"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
metrics = []
fundamentals = self._security.fundamentals
# Gross margin
gross_margin = fundamentals.operation_ratios.gross_margin
if hasattr(gross_margin, 'value'):
metrics.append(gross_margin.value)
# Operating margin
op_margin = fundamentals.operation_ratios.operation_margin
if hasattr(op_margin, 'value'):
metrics.append(op_margin.value)
# ROE
roe = fundamentals.operation_ratios.roe
if hasattr(roe, 'value'):
metrics.append(roe.value)
# ROA
roa = fundamentals.operation_ratios.roa
if hasattr(roa, 'value'):
metrics.append(roa.value)
# Return average if we have valid metrics
return np.mean(metrics) if metrics else np.nan
except:
return np.nan
class ValueFactor:
"""Test composite value factor"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
ratios = []
valuation = self._security.fundamentals.valuation_ratios
# Price/Book
ratios.append(1/valuation.pb_ratio)
# Price/Earnings
ratios.append(1/valuation.pe_ratio)
# Price/Sales
ratios.append(1/valuation.ps_ratio)
# Price/Cash Flow
ratios.append(1/valuation.pcf_ratio)
return np.mean(ratios) if ratios else np.nan
except:
return np.nan
class MomentumFactor:
"""Price momentum factor"""
def __init__(self, algorithm, security, lookback=252):
self._algorithm = algorithm
self._security = security
self._lookback = lookback
@property
def value(self):
try:
# Get the price history
history = self._algorithm.History(
self._security.Symbol,
self._lookback + 21,
Resolution.Daily
)
if len(history) < self._lookback:
return np.nan
# Calculate 12-1 month momentum (skip most recent month)
return history.close[-21]/history.close[0] - 1
except:
return np.nan
class GrowthFactor:
"""Growth metrics"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
metrics = []
# Revenue growth
if not np.isnan(self._security.fundamentals.operation_ratios.revenue_growth.value):
metrics.append(self._security.fundamentals.operation_ratios.revenue_growth.value)
# Net income growth
if not np.isnan(self._security.fundamentals.operation_ratios.net_income_growth.value):
metrics.append(self._security.fundamentals.operation_ratios.net_income_growth.value)
# Operating income growth
if not np.isnan(self._security.fundamentals.operation_ratios.operation_income_growth.value):
metrics.append(self._security.fundamentals.operation_ratios.operation_income_growth.value)
return np.nanmean(metrics) if metrics else np.nan
except:
return np.nan
class OCFConversionFactor:
"""Operating Cash Flow Conversion (OCF/EBITDA) factor"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
fundamentals = self._security.fundamentals
cash_flow = fundamentals.financial_statements.cash_flow_statement
income = fundamentals.financial_statements.income_statement
# Get operating cash flow
if hasattr(cash_flow, 'operating_cash_flow'):
ocf = cash_flow.operating_cash_flow.value
else:
return np.nan
# Calculate EBITDA
if hasattr(income, 'ebitda'):
ebitda = income.ebitda.value
else:
return np.nan
# Avoid division by zero
if ebitda == 0:
return np.nan
return ocf/ebitda
except:
return np.nan
class FCFYieldFactor:
"""Free Cash Flow Yield (FCF/Enterprise Value) factor"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
fundamentals = self._security.fundamentals
cash_flow = fundamentals.financial_statements.cash_flow_statement
# Calculate FCF (Operating Cash Flow - CapEx)
if hasattr(cash_flow, 'operating_cash_flow') and hasattr(cash_flow, 'capital_expenditure'):
fcf = cash_flow.operating_cash_flow.value - abs(cash_flow.capital_expenditure.value)
else:
return np.nan
# Get enterprise value
if hasattr(fundamentals, 'market_cap') and hasattr(fundamentals.financial_statements.balance_sheet, 'total_debt'):
enterprise_value = fundamentals.market_cap + fundamentals.financial_statements.balance_sheet.total_debt.value
else:
return np.nan
# Avoid division by zero
if enterprise_value == 0:
return np.nan
return fcf/enterprise_value
except:
return np.nan
class InstitutionalOwnershipFactor:
"""Institutional Ownership Changes factor"""
def __init__(self, algorithm, security, lookback=63): # ~3 months
self._algorithm = algorithm
self._security = security
self._lookback = lookback
@property
def value(self):
try:
# Get fundamental history
history = list(self._algorithm.History[Fundamental](
self._security.Symbol,
self._lookback
))
if len(history) < 2: # Need at least 2 points for change
return np.nan
# Get number of institutional holders from security reference data
current_inst = history[-1].security_reference.institutional_holders
previous_inst = history[0].security_reference.institutional_holders
if previous_inst == 0:
return np.nan
return (current_inst - previous_inst) / previous_inst
except:
return np.nan
class BuybackYieldFactor:
"""Buy-back Yield (Net Stock Repurchases/Market Cap) factor"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
fundamentals = self._security.fundamentals
cash_flow = fundamentals.financial_statements.cash_flow_statement
# Get stock repurchases
if hasattr(cash_flow, 'repurchase_of_capital_stock'):
repurchases = abs(cash_flow.repurchase_of_capital_stock.value) # Make positive
else:
return np.nan
# Get market cap
if hasattr(fundamentals, 'market_cap'):
market_cap = fundamentals.market_cap
else:
return np.nan
# Avoid division by zero
if market_cap == 0:
return np.nan
return repurchases/market_cap
except:
return np.nan
class BetaAdjustedVolatilityFactor:
"""Beta-adjusted Volatility Ratio (recent vs historical volatility) factor"""
def __init__(self, algorithm, security, recent_window=21, historical_window=252):
self._algorithm = algorithm
self._security = security
self._recent_window = recent_window # ~1 month
self._historical_window = historical_window # ~1 year
@property
def value(self):
try:
# Get price history
history = self._algorithm.History(
self._security.Symbol,
self._historical_window,
Resolution.Daily
)
if len(history) < self._historical_window:
return np.nan
# Calculate recent and historical volatility
recent_returns = history.close[-self._recent_window:].pct_change().dropna()
historical_returns = history.close.pct_change().dropna()
recent_vol = recent_returns.std() * np.sqrt(252) # Annualize
historical_vol = historical_returns.std() * np.sqrt(252)
# Avoid division by zero
if historical_vol == 0:
return np.nan
return recent_vol/historical_vol
except:
return np.nan
# except Exception as e:
# # Print exception details
# print(f"Exception type: {type(e).__name__}")
# print(f"Exception message: {e}")from AlgorithmImports import *
import numpy as np
class MarketCapFactor:
def __init__(self, security):
self._security = security
@property
def value(self):
try:
return self._security.fundamentals.market_cap
except:
return np.nan
def validate_value(value):
"""Helper to validate numeric values"""
if value is None:
return False
if not isinstance(value, (int, float)):
return False
try:
return not np.isnan(value) and not np.isinf(value)
except:
return False
class QualityFactor:
"""Test quality/profitability factor"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
metrics = []
fundamentals = self._security.fundamentals
# Gross margin
gross_margin = fundamentals.operation_ratios.gross_margin
if hasattr(gross_margin, 'value'):
metrics.append(gross_margin.value)
# Operating margin
op_margin = fundamentals.operation_ratios.operation_margin
if hasattr(op_margin, 'value'):
metrics.append(op_margin.value)
# ROE
roe = fundamentals.operation_ratios.roe
if hasattr(roe, 'value'):
metrics.append(roe.value)
# ROA
roa = fundamentals.operation_ratios.roa
if hasattr(roa, 'value'):
metrics.append(roa.value)
# Return average if we have valid metrics
return np.mean(metrics) if metrics else np.nan
except:
return np.nan
class ValueFactor:
"""Test composite value factor"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
ratios = []
valuation = self._security.fundamentals.valuation_ratios
# Price/Book
pb = valuation.pb_ratio
if hasattr(pb, 'value') and pb.value > 0:
ratios.append(1/pb.value)
# Price/Earnings
pe = valuation.pe_ratio
if hasattr(pe, 'value') and pe.value > 0:
ratios.append(1/pe.value)
# Price/Sales
ps = valuation.ps_ratio
if hasattr(ps, 'value') and ps.value > 0:
ratios.append(1/ps.value)
# Price/Cash Flow
pcf = valuation.pcf_ratio
if hasattr(pcf, 'value') and pcf.value > 0:
ratios.append(1/pcf.value)
return np.mean(ratios) if ratios else np.nan
except:
return np.nan
class MomentumFactor:
"""Price momentum factor"""
def __init__(self, algorithm, security, lookback=252):
self._algorithm = algorithm
self._security = security
self._lookback = lookback
@property
def value(self):
try:
# Get the price history
history = self._algorithm.History(
self._security.Symbol,
self._lookback + 21,
Resolution.Daily
)
if len(history) < self._lookback:
return np.nan
# Calculate 12-1 month momentum (skip most recent month)
return history.close[-21]/history.close[0] - 1
except:
return np.nan
class GrowthFactor:
"""Growth metrics"""
def __init__(self, security):
self._security = security
@property
def value(self):
try:
metrics = []
# Revenue growth
if not np.isnan(self._security.fundamentals.operation_ratios.revenue_growth):
metrics.append(self._security.fundamentals.operation_ratios.revenue_growth)
# Net income growth
if not np.isnan(self._security.fundamentals.operation_ratios.net_income_growth):
metrics.append(self._security.fundamentals.operation_ratios.net_income_growth)
# Operating income growth
if not np.isnan(self._security.fundamentals.operation_ratios.operation_income_growth):
metrics.append(self._security.fundamentals.operation_ratios.operation_income_growth)
return np.nanmean(metrics) if metrics else np.nan
except:
return np.nan
class SortinoFactor:
"""Keep original Sortino factor for comparison"""
def __init__(self, algorithm, symbol, lookback):
self._sortino = algorithm.sortino(symbol, lookback, resolution=Resolution.DAILY)
@property
def value(self):
try:
return self._sortino.current.value
except:
return np.nan# region imports
from AlgorithmImports import *
from itertools import chain, combinations
from scipy import optimize
from scipy.optimize import Bounds
from factors import *
# endregion
"""
SPY Constituents
-----------------
Best Sharpe - doenst beat market
--------------
Top 20
universe_size = 20 # we pick top 20 weights
portfolio_size = 20 # we ultimately select 20 top scores
factorUniverse = 34 // UseEMA = 1
SortinoFactor, ValueFactor,QualityFactor
Best Profit - flogs market after
--------------
Top 20
universe_size = 20 # we pick top 20 weights
portfolio_size = 20 # we ultimately select 20 top scores
factorUniverse = 16 // UseEMA = 0
KERFactor, QualityFactor
Great - GO live
--------------
Top 5
universe_size = 5 # we pick top 5 weights
portfolio_size = 5 # we ultimately select 5 top scores
factorUniverse = 32 // UseEMA = 1
SortinoFactor, KERFactor, QualityFactor
########################################
##### GO LIVE
########################################
##
## Great - KER + FCFYield
## --------------
## Top 5
## universe_size = 5 # we pick top 5 weights
## portfolio_size = 5 # we ultimately select 5 top scores
## use EMA = 1
## KERFactor, FCFYieldFactor
## (factor universe = 10 -- with second batch of factors)
##
################################################################################
Great - BuyBackYield + KER
--------------
Top 5
universe_size = 5 # we pick top 5 weights
portfolio_size = 5 # we ultimately select 5 top scores
use EMA = 0
KERFactor, BuyBackYieldFactor
(factor universe = 11 -- with second batch of factors)
"""
class FactorWeightOptimizationAlgorithm(QCAlgorithm):
def initialize(self):
######## DATE
self.SetStartDate(2014, 1, 1)
# self.SetEndDate(2021, 12, 20)
# self.SetStartDate(2010, 1, 1)
self.set_cash(30000)
self.settings.automatic_indicator_warm_up = True
# self.spy = Symbol.create('SPY', SecurityType.EQUITY, Market.USA)
ticker = "SPY"
self.set_benchmark(ticker)
self.spy = self.add_equity(ticker).symbol
self.ema50 = self.ema(self.spy,50, Resolution.Daily)
self.ema200 = self.ema(self.spy,200, Resolution.Daily)
self.ema500 = self.ema(self.spy,500, Resolution.Daily)
self.ema200.updated += self.OnEMAUpdated
# Add a universe of hourly data.
self.universe_settings.resolution = Resolution.HOUR
self.universe_size = self.get_parameter('universe_size', 5)
# Add variable to track last portfolio value
self.last_rebalance_portfolio_value = None
# Use the separate universe filter function.
self._universe = self.add_universe(self.universe.etf(self.spy, universe_filter_func=self.universe_filter))
self._lookback = self.get_parameter('lookback', 21) # Set a 21-day trading lookback.
# Create a Schedule Event to rebalance the portfolio.
self.schedule.on(self.date_rules.month_start(self.spy), self.time_rules.after_market_open(self.spy, 31), self._rebalance)
def universe_filter(self, constituents):
"""
Filters and sorts the given constituents based on their weight, returning the top symbols.
"""
filtered_constituents = [c for c in constituents if c.weight]
sorted_constituents = sorted(filtered_constituents, key=lambda c: c.weight)
return [c.symbol for c in sorted_constituents[-self.universe_size:]]
def OnEMAUpdated(self, sender, bar):
# self.plot("series", self.ema50.current.value)
# self.plot("series", self.ema200.current.value)
self.Plot('EMAx', 'SPY50', self.ema50.current.value)
self.Plot('EMAx', 'SPY200', self.ema200.current.value)
self.Plot('EMAx', 'SPY500', self.ema500.current.value)
def on_securities_changed(self, changes):
for security in changes.added_securities: # Create factors for assets that enter the universe.
# security.factors = [MarketCapFactor(security), SortinoFactor(self, security.symbol, self._lookback)]
# Go Live CAMO Factors
# security.factors = [
# SortinoFactor(self, security.symbol, self._lookback),
# KERFactor(self, security.symbol, self._lookback),
# QualityFactor(security),
# ]
# allFactors = [
# MarketCapFactor(security),
# SortinoFactor(self, security.symbol, self._lookback),
# KERFactor(self, security.symbol, self._lookback),
# ValueFactor(security),
# QualityFactor(security),
# GrowthFactor(security)
# ]
#
# security.factors = self.get_factor_subset_by_index(allFactors, self.get_parameter('factorUniverse', 32))
allFactors = [
KERFactor(self, security.symbol, self._lookback), # Keeping KER for trend efficiency
SortinoFactor(self, security.symbol, self._lookback), # Keeping Sortino for downside risk
QualityFactor(security), # Keeping core Quality metrics
OCFConversionFactor(security), # New cash flow quality metric
FCFYieldFactor(security), # New value metric
BuybackYieldFactor(security), # New management confidence metric
BetaAdjustedVolatilityFactor(self, security, 21, 252) # New risk regime metric
]
#
security.factors = self.get_factor_subset_by_index(allFactors, self.get_parameter('factorUniverse', 32))
## DEBUG : Print Subsets
############################################################
# all_subsets = list(chain.from_iterable(combinations(allFactors, r) for r in range(1, len(allFactors) + 1)))
# for index, content in enumerate(all_subsets):
# self.Log(f"{index} - position {index} {content}")
# self.quit("message")
# quit
############################################################
# security.factors = [
# KERFactor(self, security.symbol, self._lookback), # Keeping KER for trend efficiency
# SortinoFactor(self, security.symbol, self._lookback), # Keeping Sortino for downside risk
# QualityFactor(security), # Keeping core Quality metrics
# OCFConversionFactor(security), # New cash flow quality metric
# FCFYieldFactor(security), # New value metric
# # InstitutionalOwnershipFactor(self, security, 63), # New smart money flow metric
# BuybackYieldFactor(security), # New management confidence metric
# BetaAdjustedVolatilityFactor(self, security, 21, 252) # New risk regime metric
# ]
this = 0
def get_factor_subset_by_index(self, array, subset_param):
# Generate all subsets
all_subsets = list(chain.from_iterable(combinations(array, r) for r in range(1, len(array) + 1)))
# Handle out-of-bounds subset_param
if subset_param < 0 or subset_param >= len(all_subsets):
raise ValueError(f"subset_param must be in range 0 to {len(all_subsets) - 1}.")
# Return the subset at the specified index
return list(all_subsets[subset_param])
def OnData(self,slice):
if( self.get_parameter("useEMA", 1) == 1):
self.liquidateIfEMABearish()
if (int(self.get_parameter("useEquityTrailStop",1)) == 1):
# trailing stop
if not hasattr(self,"trailEquityStop"):
self.trailEquityStop = self.Portfolio.TotalPortfolioValue * 0.80
else:
if self.Portfolio.TotalPortfolioValue <= self.trailEquityStop:
self.Liquidate(tag="trail equity stop")
self.trailEquityStop = self.Portfolio.TotalPortfolioValue * 0.80
return
self.trailEquityStop = max(self.trailEquityStop,self.Portfolio.TotalPortfolioValue * 0.90)
def isEMABearish(self):
return (self.ema50.current.value < self.ema500.current.value )
def liquidateIfEMABearish(self):
if( self.ema500.is_ready ):
if(self.isEMABearish()):
self.liquidate(tag="EMA is Bearish")
def _rebalance(self):
try:
# Calculate and log profit since last rebalance
current_value = self.Portfolio.TotalPortfolioValue
if self.last_rebalance_portfolio_value is not None:
profit_pct = ((current_value - self.last_rebalance_portfolio_value) / self.last_rebalance_portfolio_value) * 100
self.Log(f"Profit since last rebalance: {profit_pct:.2f}%")
# Update the last rebalance value
self.last_rebalance_portfolio_value = current_value
if( self.get_parameter("useEMA", 1) == 1):
if(self.isEMABearish()):
self.liquidateIfEMABearish()
return
# Get raw factor values of the universe constituents.
factors_df = pd.DataFrame()
for symbol in self._universe.selected:
for i, factors in enumerate(self.securities[symbol].factors):
factors_df.loc[symbol, i] = factors.value
if factors_df.empty:
self.Debug("No factor data available for rebalancing")
return
# Calculate the factor z-scores.
factor_zscores = (factors_df - factors_df.mean()) / factors_df.std()
# Safely get historical data and calculate returns
try:
history_df = self.history(list(self._universe.selected), self._lookback, Resolution.DAILY)
# Check if we have the expected data structure
if not isinstance(history_df, pd.DataFrame) or 'close' not in history_df:
self.Debug(f"Historical data format unexpected: {type(history_df)}")
return
# Unstack and calculate returns, handling any missing data
price_df = history_df.close.unstack(0)
if price_df.empty:
self.Debug("No price data available for calculation")
return
# Calculate returns and handle any missing values
trailing_return = price_df.pct_change(self._lookback-1).iloc[-1]
trailing_return = trailing_return.fillna(0) # Replace any NaN values with 0
if trailing_return.empty:
self.Debug("No valid return data available")
return
except Exception as e:
self.Debug(f"Error calculating returns: {str(e)}")
return
# Run optimization only if we have valid data
num_factors = factors_df.shape[1]
try:
factor_weights = optimize.minimize(
lambda weights: -(np.dot(factor_zscores, weights) * trailing_return).sum(),
x0=np.array([1.0/num_factors] * num_factors),
method='Nelder-Mead',
bounds=Bounds([0] * num_factors, [1] * num_factors),
options={'maxiter': 10}
).x
except Exception as e:
self.Debug(f"Optimization failed: {str(e)}")
return
# Calculate the portfolio weights
portfolio_weights = (factor_zscores * factor_weights).sum(axis=1)
portfolio_weights = portfolio_weights[portfolio_weights > 0]
portfolio_weights = portfolio_weights.nlargest(self.get_parameter('portfolio_size', 5))
if portfolio_weights.empty:
self.Debug("No valid portfolio weights calculated")
return
# Log the date without time
# Log portfolio weights, one per line
formatted_weights = "\n\t".join([f"{x.Value} - {round((y/portfolio_weights.sum())*100, 2)}%" for x, y in portfolio_weights.items()])
self.Log(f"{self.Time.strftime('%Y-%m-%d')} -- {formatted_weights}")
# Set holdings only if we have valid weights
if not portfolio_weights.empty:
self.set_holdings([PortfolioTarget(symbol, weight/portfolio_weights.sum()) for symbol, weight in portfolio_weights.items()], True)
except BaseException as e:
self.Debug(f"Error During: {str(e)}")
returnfrom AlgorithmImports import *
from scipy import optimize
from scipy.optimize import Bounds
from collections import defaultdict
import numpy as np
class FactorTestAlgorithm(QCAlgorithm):
def initialize(self):
self.SetStartDate(2014, 12, 31)
self.SetCash(10_000_000)
self.settings.automatic_indicator_warm_up = True
spy = Symbol.create('SPY', SecurityType.EQUITY, Market.USA)
# Add a universe of hourly data
self.universe_settings.resolution = Resolution.HOUR
universe_size = self.get_parameter('universe_size', 50) # Reduced for testing
self._universe = self.add_universe(self.universe.etf(spy,
universe_filter_func=lambda constituents: [c.symbol for c in sorted(
[c for c in constituents if c.weight],
key=lambda c: c.weight
)[-universe_size:]]
))
self._lookback = self.get_parameter('lookback', 21)
# Create a Schedule Event to rebalance the portfolio
self.schedule.on(self.date_rules.month_start(spy),
self.time_rules.after_market_open(spy, 31),
self._rebalance)
def log_fundamental_data(self, symbol, data_checks):
"""
Helper to log fundamental data values
data_checks: list of tuples (name, path to check)
"""
security = self.securities[symbol]
self.Debug(f"\nLogging fundamental data for {symbol}:")
for name, path in data_checks:
try:
# Get the fundamental field
field = eval(f"security.fundamentals.{path}")
# Access the numeric value
value = field.value if hasattr(field, 'value') else field
if isinstance(value, (int, float)):
self.Debug(f"{name}: {value:.4f}")
else:
self.Debug(f"{name}: No valid value")
except Exception as e:
self.Debug(f"{name}: Not Available - {str(e)}")
def is_valid_numeric(value):
"""Helper to check if a value is valid for calculations"""
if value is None:
return False
if not isinstance(value, (int, float)):
return False
try:
return not np.isnan(value) and not np.isinf(value)
except:
return False
def validate_value_factors(self, symbol):
"""Test value factor data availability"""
self.Debug(f"\nValidating Value Factors for {symbol}")
self.log_fundamental_data(symbol, [
("Price/Book", "valuation_ratios.pb_ratio"),
("Price/Earnings", "valuation_ratios.pe_ratio"),
("Price/Sales", "valuation_ratios.ps_ratio"),
("Price/CFO", "valuation_ratios.pcf_ratio")
])
def validate_quality_factors(self, symbol):
"""Test quality/profitability factor data availability"""
self.Debug(f"\nValidating Quality Factors for {symbol}")
self.log_fundamental_data(symbol, [
("Gross Margin", "operation_ratios.gross_margin"),
("Operating Margin", "operation_ratios.operation_margin"),
("ROA", "operation_ratios.roa"),
("ROE", "operation_ratios.roe"),
("Net Income", "financial_statements.income_statement.net_income")
])
def validate_growth_factors(self, symbol):
"""Test growth factor data availability"""
self.Debug(f"\nValidating Growth Factors for {symbol}")
self.log_fundamental_data(symbol, [
("Revenue Growth", "operation_ratios.revenue_growth"),
("Net Income Growth", "operation_ratios.net_income_growth"),
("Operating Income Growth", "operation_ratios.operation_income_growth")
])
def _rebalance(self):
"""Monthly data quality analysis"""
self.Debug("\n=== Monthly Factor Data Analysis ===")
# Collect factor values
factor_data = defaultdict(list)
for symbol in self._universe.selected:
security = self.securities[symbol]
# Test Value Metrics
try:
pb = security.fundamentals.valuation_ratios.pb_ratio
if hasattr(pb, 'value'):
factor_data["P/B"].append(pb.value)
except: pass
try:
op_margin = security.fundamentals.operation_ratios.operation_margin
if hasattr(op_margin, 'value'):
factor_data["Operating Margin"].append(op_margin.value)
except: pass
try:
roe = security.fundamentals.operation_ratios.roe
if hasattr(roe, 'value'):
factor_data["ROE"].append(roe.value)
except: pass
# Log statistics for each factor
self.Debug("\nFactor Statistics:")
for factor_name, values in factor_data.items():
if values: # Only process if we have valid values
values = [v for v in values if isinstance(v, (int, float)) and not np.isnan(v)]
if values:
self.Debug(f"\n{factor_name}:")
self.Debug(f"Available data: {len(values)}/{len(self._universe.selected)}")
self.Debug(f"Mean: {np.mean(values):.4f}")
self.Debug(f"Std: {np.std(values):.4f}")
self.Debug(f"Range: [{np.min(values):.4f}, {np.max(values):.4f}]")
else:
self.Debug(f"\n{factor_name}: No valid numeric values")
else:
self.Debug(f"\n{factor_name}: No data collected")
def validate_momentum_data(self, symbol):
"""Test price momentum data availability"""
try:
history = self.History(symbol, self._lookback + 21, Resolution.Daily)
if len(history) >= self._lookback:
mom = history.close[-21]/history.close[0] - 1
self.Debug(f"\n{symbol} 12-1m Momentum: {mom:.2%}")
else:
self.Debug(f"\n{symbol} Insufficient price history")
except Exception as e:
self.Debug(f"\n{symbol} Momentum calculation failed: {str(e)}")
def on_securities_changed(self, changes):
"""Handle security additions and removals"""
for security in changes.added_securities:
# Run validation tests
self.validate_value_factors(security.symbol)
self.validate_quality_factors(security.symbol)
self.validate_growth_factors(security.symbol)
self.validate_momentum_data(security.symbol)