| Overall Statistics |
|
Total Orders 244 Average Win 3.17% Average Loss -2.04% Compounding Annual Return 32.085% Drawdown 30.700% Expectancy 0.819 Start Equity 1000 End Equity 7270.14 Net Profit 627.014% Sharpe Ratio 0.918 Sortino Ratio 1.017 Probabilistic Sharpe Ratio 40.145% Loss Rate 29% Win Rate 71% Profit-Loss Ratio 1.55 Alpha 0.085 Beta 1.048 Annual Standard Deviation 0.24 Annual Variance 0.058 Information Ratio 0.772 Tracking Error 0.119 Treynor Ratio 0.21 Total Fees $244.00 Estimated Strategy Capacity $97000000.00 Lowest Capacity Asset AVGO UEW4IOBWVPT1 Portfolio Turnover 3.01% |
# 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 HEFactor:
def __init__(self, algorithm, symbol, lookback, maxLag):
self._he = algorithm.he(symbol, lookback, maxLag, resolution=Resolution.DAILY)
@property
def value(self):
return self._he.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 FCFDistributionFactor:
"""Measures how much FCF is distributed vs retained"""
def __init__(self, security, theAlgo):
self._security = security
self.algo = theAlgo
def _calculate_fcf(self):
cash_flow = self._security.fundamentals.financial_statements.cash_flow_statement
# cash_flow = fundamentals.financial_statements.cash_flow_statement
if hasattr(cash_flow, 'operating_cash_flow') and hasattr(cash_flow, 'capital_expenditure'):
return cash_flow.operating_cash_flow.value - abs(cash_flow.capital_expenditure.value)
return np.nan
@property
def value(self):
try:
cash_flow = self._security.fundamentals.financial_statements.cash_flow_statement
fcf = self._calculate_fcf()
if np.isnan(fcf) or fcf == 0:
return np.nan
distributions = 0
distribution_fields = [
'cash_dividends_paid',
'repurchase_of_capital_stock',
'long_term_debt_payments',
# 'net_long_term_debt_repayment'
]
# Sum up all distributions
for field in distribution_fields:
if hasattr(cash_flow, field):
val = abs(getattr(cash_flow, field).value)
distributions += val
else:
self.algo.Debug(f"Missing {field} field")
# Calculate distribution ratio
return distributions / abs(fcf)
except BaseException as e:
self.algo.Debug(f"FCFDistributionFactor error: {str(e)}")
return np.nan
class FCFCompositeFactor:
"""Free Cash Flow Composite Factor combining FCF Yield and Growth"""
def __init__(self, security):
self._security = security
def _get_fcf_yield(self, period='value'):
fundamentals = self._security.fundamentals
cash_flow = fundamentals.financial_statements.cash_flow_statement
balance_sheet = fundamentals.financial_statements.balance_sheet
# Get FCF for specified period
if hasattr(cash_flow, 'operating_cash_flow') and hasattr(cash_flow, 'capital_expenditure'):
fcf = getattr(cash_flow.operating_cash_flow, period) - abs(getattr(cash_flow.capital_expenditure, period))
else:
return np.nan
# For enterprise value, use current market cap but with quarterly cash/debt positions
period_for_bs = 'three_months' if period == 'three_months' else 'value'
if all(hasattr(balance_sheet, attr) for attr in ['cash_and_cash_equivalents', 'total_debt']) and hasattr(fundamentals, 'market_cap'):
enterprise_value = (fundamentals.market_cap +
getattr(balance_sheet.total_debt, period_for_bs) -
getattr(balance_sheet.cash_and_cash_equivalents, period_for_bs))
else:
return np.nan
if enterprise_value == 0:
return np.nan
return fcf/enterprise_value
@property
def value(self):
# Get FCF yields for available periods
current_yield = self._get_fcf_yield('value')
yield_3m = self._get_fcf_yield('three_months')
if any(np.isnan([current_yield, yield_3m])):
return np.nan
# Calculate growth rate
growth_3m = (current_yield - yield_3m) / yield_3m
# Simplified composite score
return 0.6 * current_yield + 0.4 * growth_3m
class FCFPerShareFactor:
"""Free Cash Flow Per Share (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
return self._security.fundamentals.valuation_ratios.FCFPerShare
except BaseException(e):
return np.nan
class SUEFactor:
"""Standardized Unexpected Earnings Factor - Measures earnings surprise relative to recent history"""
def __init__(self, security, algo):
self._security = security
self._algo = algo
@property
def value(self):
# try:
fundamentals = self._security.fundamentals
if not hasattr(fundamentals, 'financial_statements'):
self._algo.Log("No financial statements found")
return np.nan
earnings = fundamentals.earning_reports
if not hasattr(earnings, 'basic_eps'):
self._algo.Log("No EPS data found")
return np.nan
# Get individual quarter EPS values
q1 = earnings.basic_eps.three_months # Latest quarter
q2 = earnings.basic_eps.six_months
q3 = earnings.basic_eps.nine_months
q4 = earnings.basic_eps.value # Full year
if any(np.isnan([q1, q2, q3, q4])):
self._log.Debug("Missing EPS values in time series")
return np.nan
# Calculate expected EPS as average of last 3 quarters
expected_eps = np.mean([q2, q3, q4])/4 # Divide by 4 to get quarterly average
# Calculate standard deviation using all 4 quarters
eps_values = [q1/4, q2/4, q3/4, q4/4] # Convert to quarterly values
eps_std = np.std(eps_values)
if eps_std == 0:
self._log.Debug("Zero standard deviation in EPS")
return np.nan
# Calculate SUE
sue = (q1/4 - expected_eps) / eps_std
return sue
# except BaseException as e:
# self._algo.Log(f"Error calculating SUE: {str(e)}")
# return np.nan
# Get quarterly EPS values
# current_eps = earnings.basic_eps.value
# three_month_eps = earnings.basic_eps.three_months
# six_month_eps = earnings.basic_eps.six_months
# nine_month_eps = earnings.basic_eps.nine_months
# if any(np.isnan([current_eps, three_month_eps, six_month_eps, nine_month_eps])):
# self._algo.Log("Missing EPS values in time series")
# return np.nan
# # Calculate expected EPS as average of last 3 quarters
# expected_eps = np.mean([three_month_eps, six_month_eps, nine_month_eps])
# # Calculate standard deviation using all 4 quarters
# eps_values = [current_eps, three_month_eps, six_month_eps, nine_month_eps]
# eps_std = np.std(eps_values)
# if eps_std == 0:
# self._algo.Log("Zero standard deviation in EPS")
# return np.nan
# # Calculate SUE
# sue = (current_eps - expected_eps) / eps_std
# return sue
# except BaseException as e:
# self._algo.Log(f"Error calculating SUE: {str(e)}")
# return np.nan
class FCFYClaudeFactor:
"""Claude's Free Cash Flow Yield (FCF/Enterprise Value) factor"""
"""They subtract balance_sheet.cash_and_cash_equivalents.value from enterprise value"""
def __init__(self, security):
self._security = security
@property
def value(self):
fundamentals = self._security.fundamentals
cash_flow = fundamentals.financial_statements.cash_flow_statement
balance_sheet = fundamentals.financial_statements.balance_sheet
# Calculate FCF
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
# Enterprise Value = Market Cap + Total Debt - Cash/Equivalents
if all(hasattr(balance_sheet, attr) for attr in ['cash_and_cash_equivalents', 'total_debt']) and hasattr(fundamentals, 'market_cap'):
enterprise_value = (fundamentals.market_cap +
balance_sheet.total_debt.value -
balance_sheet.cash_and_cash_equivalents.value)
else:
return np.nan
if enterprise_value == 0:
return np.nan
# return fundamentals.valuation_ratios.FCFYield
return fcf/enterprise_value
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}")# region imports
from AlgorithmImports import *
from itertools import chain, combinations
from scipy import optimize
from scipy.optimize import Bounds
from factors import *
# endregion
"""
'Documentation'
Reddit Post: https://www.reddit.com/r/algotrading/comments/1h3pptt/seeking_feedback_on_rebalancing_strategy_using/
"""
class FactorWeightOptimizationAlgorithm(QCAlgorithm):
def initialize(self):
######## DATE
# self.SetStartDate(2023, 6, 1)
# self.SetEndDate(2020, 1, 1)
# self.SetStartDate(2010, 1, 1)
self.set_start_date(2018, 1, 1)
self.set_cash(1000)
self.settings.automatic_indicator_warm_up = True
# self.spy = Symbol.create('SPY', SecurityType.EQUITY, Market.USA)
ticker = "QQQ"
self.set_benchmark(ticker)
self.spy = self.add_equity(ticker).symbol
self._he = self.he(self.spy, 50, 10, resolution=Resolution.DAILY)
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)
# self.schedule.on(self.date_rules.week_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:]]
# return [c.symbol for c in sorted_constituents[:-100][-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)
self.Plot("HEE",'he',self._he.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))
# allFactors = [
# KERFactor(self, security.symbol, self._lookback), # Keeping KER for trend efficiency
# FCFPerShareFactor(security),
# FCFYieldFactor(security), # New value metric
# FCFDistributionFactor(security, self),
# FCFYClaudeFactor(security),
# SUEFactor(security,self)
# ]
## 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 = self.get_factor_subset_by_index(allFactors, self.get_parameter('factorUniverse', 32))
# '''
security.factors = [
KERFactor(self, security.symbol, self._lookback), # Keeping KER for trend efficiency
FCFYieldFactor(security),
# HEFactor(self, security.symbol, 50, 10),
# New value metric
# FCFPerShareFactor(security),
# SUEFactor(security,self),
# FCFDistributionFactor(security, self),
# FCFYClaudeFactor(security)
# Inconclusive i think it has errorrs. lots of nans --> # FCFCompositeFactor(security), <
# seems to be same or slighly worst --> FCFYClaudeFactor(security),
]
# '''
this = 0
# Hepful function for optimization
# where every combination of possible factors (the 'all_factors' array' )
# can eb reached per index.
#
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( int(self.get_parameter("useEMA", 0)) == 1):
self.liquidateIfEMABearish()
if (int(self.get_parameter("useStopPerStock", 0)) == 1):
for holding in self.Portfolio.Values:
if holding.UnrealizedProfitPercent < -0.10:
self.Liquidate(holding.symbol, tag="unrealized profit loss")
if (int(self.get_parameter("useEquityTrailStop",0)) == 1):
# trailing stop
if hasattr(self,"trailEquityStop"):
if self.Portfolio.TotalPortfolioValue <= self.trailEquityStop:
self.Liquidate(tag="trail equity 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
def isEMABearish(self):
# return (self.ema50.current.value < self.ema200.current.value )
return (self.securities['spy'].price < self.ema200.current.value )
# return self._he.current.value >= 0.5
# = algorithm.he(symbol, lookback, maxLag, resolution=Resolution.DAILY)
def liquidateIfEMABearish(self):
if( self.ema500.is_ready ):
if(self.isEMABearish()):
self.liquidate(tag="EMA is Bearish")
def _rebalance(self):
try:
self.trailEquityStop = self.Portfolio.TotalPortfolioValue * 0.80
# 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(f"{self.Time} 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")
self.c()
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(f"{self.Time} 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:
if (self.get_parameter('useEqualWeights', 0) == 1):
self.set_holdings([PortfolioTarget(symbol, 1/self.universe_size) for symbol in self._universe.selected], True)
else:
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)}")
return
def OnOrderEvent(self, orderEvent):
# Check if the order was submitted and it's a sell order
if orderEvent.Status == OrderStatus.Submitted and orderEvent.Direction == OrderDirection.Sell:
order = self.Transactions.GetOrderById(orderEvent.OrderId)
if order and self.Portfolio[order.Symbol].Invested:
# Calculate unrealized P&L percentage
upnl_pct = (self.Portfolio[order.Symbol].UnrealizedProfitPercent) * 100
# Determine tag as "WIN" or "LOSS" based on the unrealized P&L
tag = f"{'WIN' if upnl_pct > 0 else 'LOSS'} {upnl_pct:.2f}% for {order.Symbol}"
# self.Debug(tag)
# Attempt to set this custom tag on the order (demonstrative)
order.Tag = tag # This may or may not work, depending on Lean's constraints