| Overall Statistics |
|
Total Orders 10001 Average Win 0.83% Average Loss -0.97% Compounding Annual Return -89.177% Drawdown 46.800% Expectancy -0.532 Net Profit -41.124% Sharpe Ratio -1.395 Sortino Ratio -1.408 Probabilistic Sharpe Ratio 1.620% Loss Rate 75% Win Rate 25% Profit-Loss Ratio 0.86 Alpha -0.303 Beta 1.024 Annual Standard Deviation 0.553 Annual Variance 0.306 Information Ratio -1.089 Tracking Error 0.289 Treynor Ratio -0.753 Total Fees $379.81 Estimated Strategy Capacity $44000000.00 Lowest Capacity Asset CMA R735QTJ8XC9X Portfolio Turnover 49.43% |
from AlgorithmImports import *
class FundamentalFactorAlphaModel(AlphaModel):
'''
An AlphaModel that generates insights based on fundamental factors.
It ranks stocks according to the Return on Equity (ROE), Net Margin, and the Price/Earnings Ratio (PERatio).
The model generates buy insights for the top 20% of each sector.
'''
def __init__(self):
self.rebalanceTime = datetime.min
# Dictionary containing set of securities in each sector
# e.g. {technology: set(AAPL, TSLA, ...), healthcare: set(XYZ, ABC, ...), ... }
self.sectors = {}
def Update(self, algorithm, data):
'''
Updates this alpha model with the latest data from the algorithm.
This is called each time the algorithm receives data for subscribed securities
Args:
algorithm: The algorithm instance
data: The new data available
Returns:
New insights
'''
if algorithm.Time <= self.rebalanceTime:
return []
# Set the rebalance time to match the insight expiry
self.rebalanceTime = Expiry.EndOfQuarter(algorithm.Time)
insights = []
# Loop through each sector to generate insights
for sector in self.sectors:
securities = self.sectors[sector]
# Sort securities by ROE, net margin, and PE ratio
sortedByROE = sorted(securities, key=lambda x: x.Fundamentals.OperationRatios.ROE.Value, reverse=True)
sortedByPM = sorted(securities, key=lambda x: x.Fundamentals.OperationRatios.NetMargin.Value, reverse=True)
sortedByPE = sorted(securities, key=lambda x: x.Fundamentals.ValuationRatios.PERatio, reverse=False)
# Dictionary holding a dictionary of scores for each security in the sector
scores = {}
for security in securities:
score = sum([sortedByROE.index(security), sortedByPM.index(security), sortedByPE.index(security)])
scores[security] = score
# Add best 20% of each sector to longs set (minimum 1)
length = max(int(len(scores)/5), 1)
for security in sorted(scores.items(), key=lambda x: x[1], reverse=False)[:length]:
symbol = security[0].Symbol
# Use Expiry.EndOfQuarter in this case to match Universe, Alpha and PCM
insights.append(Insight.Price(symbol, Expiry.EndOfQuarter, InsightDirection.Up))
return insights
def OnSecuritiesChanged(self, algorithm, changes):
'''
Event fired each time the we add/remove securities from the data feed
Args:
algorithm: The algorithm instance that experienced the change in securities
changes: The security additions and removals from the algorithm
'''
# Remove security from sector set
for security in changes.RemovedSecurities:
for sector in self.sectors:
if security in self.sectors[sector]:
self.sectors[sector].remove(security)
# Add security to corresponding sector set
for security in changes.AddedSecurities:
sector = security.Fundamentals.AssetClassification.MorningstarSectorCode
if sector not in self.sectors:
self.sectors[sector] = set()
self.sectors[sector].add(security)
from AlgorithmImports import *
class TechnicalAnalysisAlphaModel(AlphaModel):
'''
An AlphaModel that generates insights based on technical analysis.
It evaluates securities using a Simple Moving Average (SMA) crossover strategy to determine bullish or bearish signals.
'''
def __init__(self, short_window=50, long_window=200):
self.short_window = short_window
self.long_window = long_window
self.symbolDataBySymbol = {}
class SymbolData:
def __init__(self, symbol, algorithm):
self.Symbol = symbol
self.shortSMA = algorithm.SMA(symbol, 50, Resolution.Daily)
self.longSMA = algorithm.SMA(symbol, 200, Resolution.Daily)
self.IsBullish = False
def update(self, algorithm):
if not (self.shortSMA.IsReady and self.longSMA.IsReady):
return None
previousIsBullish = self.IsBullish
self.IsBullish = self.shortSMA.Current.Value > self.longSMA.Current.Value
if self.IsBullish and not previousIsBullish:
return Insight.Price(self.Symbol, timedelta(days=15), InsightDirection.Up)
elif not self.IsBullish and previousIsBullish:
return Insight.Price(self.Symbol, timedelta(days=15), InsightDirection.Down)
return None
def Update(self, algorithm, data):
insights = []
for symbol, symbolData in self.symbolDataBySymbol.items():
insight = symbolData.update(algorithm)
if insight is not None:
insights.append(insight)
return insights
def OnSecuritiesChanged(self, algorithm, changes):
for added in changes.AddedSecurities:
if added.Symbol not in self.symbolDataBySymbol:
self.symbolDataBySymbol[added.Symbol] = self.SymbolData(added.Symbol, algorithm)
for removed in changes.RemovedSecurities:
if removed.Symbol in self.symbolDataBySymbol:
del self.symbolDataBySymbol[removed.Symbol]
from AlgorithmImports import *
class PercentBAlphaModel(AlphaModel):
"""
Alpha model using Bollinger Bands %B value for generating insights.
Generates an 'Up' insight when %B crosses above 0.2 from below.
Generates a 'Flat' insight to liquidate when %B crosses below 0.8 from above.
"""
def __init__(self, period=20, k=2):
"""
Initializes the PercentBAlphaModel with Bollinger Bands parameters.
Args:
period (int): The period for the Bollinger Bands calculation.
k (float): The number of standard deviations specifying the distance
between the bands and the moving average.
"""
self.period = period
self.k = k
self.bbIndicators = {}
self.percentBWindows = {}
def Update(self, algorithm, data):
"""
Generates insights based on the %B value from the Bollinger Bands for each symbol.
Args:
algorithm (QCAlgorithm): The algorithm instance.
data (Slice): The current data slice.
Returns:
List[Insight]: A list of insights generated by this model.
"""
insights = []
for symbol, bb in self.bbIndicators.items():
if symbol in data and data[symbol] is not None:
# Update the Bollinger Bands with the latest price
price = data[symbol].Close
bb.Update(data.Time, price)
# Check if Bollinger Bands are ready to use
if bb.IsReady:
percentB = (price - bb.LowerBand.Current.Value) / \
(bb.UpperBand.Current.Value - bb.LowerBand.Current.Value)
# Store the %B values in a rolling window
percentBWindow = self.percentBWindows[symbol]
percentBWindow.Add(percentB)
if percentBWindow.IsReady:
# Generate insights based on the %B value
if percentB > 0.2 and percentBWindow[1] <= 0.2:
insights.append(Insight.Price(symbol, timedelta(days=1), InsightDirection.Up))
elif percentB < 0.8 and percentBWindow[1] >= 0.8:
insights.append(Insight.Price(symbol, timedelta(days=1), InsightDirection.Flat))
return insights
def OnSecuritiesChanged(self, algorithm, changes):
"""
Initializes or removes Bollinger Bands indicators and their associated rolling windows for added or removed securities.
"""
for added in changes.AddedSecurities:
symbol = added.Symbol
if symbol not in self.bbIndicators:
self.bbIndicators[symbol] = algorithm.BB(symbol, self.period, self.k, MovingAverageType.Simple, Resolution.Daily)
self.percentBWindows[symbol] = RollingWindow[float](2) # We need only the last two values for %B
for removed in changes.RemovedSecurities:
symbol = removed.Symbol
if symbol in self.bbIndicators:
del self.bbIndicators[symbol]
del self.percentBWindows[symbol]
from AlgorithmImports import *
import math
class FixedFractionPortfolioConstructionModel(PortfolioConstructionModel):
def __init__(self, fraction=0.01, atr_multiplier=2, atrIndicators=None, riskManagementModel=None):
"""
Initializes the Fixed Fraction Portfolio Construction Model.
Args:
fraction (float): The fixed fraction of the portfolio to allocate per security.
atrIndicators (Dictionary[Symbol, IndicatorBase]): A dictionary of ATR indicators for each security.
riskManagementModel (RiskManagementModel): An instance of the risk management model to be used for setting trailing stops.
"""
self.fraction = fraction
self.atr_multiplier = atr_multiplier
self.atrIndicators = atrIndicators
self.riskManagementModel = riskManagementModel
def CreateTargets(self, algorithm, insights):
"""
Create portfolio targets based on the provided insights and ATR values. Additionally,
calculates and sets the trailing stop percentage for each security.
Args:
algorithm (QCAlgorithm): The algorithm instance.
insights (List[Insight]): The list of insights provided by the Alpha model.
Returns:
List[PortfolioTarget]: A list of portfolio targets to try and achieve.
"""
targets = []
for insight in insights:
try:
if insight.IsActive(algorithm.UtcTime) and insight.Symbol in self.atrIndicators:
currentPrice = round(abs(algorithm.Securities[insight.Symbol].Price),2)
atrIndicator = self.atrIndicators[insight.Symbol]
if insight.Direction == InsightDirection.Flat:
targets.append(PortfolioTarget(insight.Symbol, 0)) # Liquidate position
continue # Move on to the next insight
if atrIndicator.IsReady:
scaledAtrValue = round(abs(atrIndicator.Current.Value * self.atr_multiplier), 2)
total_portfolio_value = algorithm.Portfolio.TotalPortfolioValue
stopLossPrice = currentPrice - scaledAtrValue
riskPerShare = round(abs(currentPrice - stopLossPrice), 2)
dollar_risk_per_trade = algorithm.Portfolio.TotalPortfolioValue * self.fraction
quantity = math.floor(dollar_risk_per_trade / riskPerShare)
# Enhanced debugging for ATR and position sizing
algorithm.Debug(f"Insight: {insight.Symbol}, Type: {insight.Direction}, " +
f"ATR: {atrIndicator.Current.Value:.2f}, Scaled ATR: {scaledAtrValue}, " +
f"Current Price: {currentPrice}, Stop Loss Price: {stopLossPrice:.2f}, Risk in chart points: {riskPerShare}," +
f"Portfolio Value: {total_portfolio_value}, Fraction: {self.fraction}, Dollar Risk Per Trade: ${dollar_risk_per_trade:.2f} " +
f"Quantity: {quantity:.2f} shares")
# Calculate and set the trailing stop percentage
self._setTrailingStop(algorithm, insight.Symbol, currentPrice, scaledAtrValue)
targets.append(PortfolioTarget(insight.Symbol, quantity))
except Exception as e:
algorithm.Error(f"Error in creating target for {insight.Symbol}: {str(e)}")
return targets
def _setTrailingStop(self, algorithm, symbol, currentPrice, scaledAtrValue):
try:
trailingStopPercentage = scaledAtrValue / currentPrice
if self.riskManagementModel:
# Pass the scaled ATR value to the risk management model
self.riskManagementModel.SetTrailingStop(algorithm, symbol, scaledAtrValue)
algorithm.Log(f"Trailing stop set for {symbol}: {trailingStopPercentage*100:.2f}%" +
f"ATR Multiplier: {self.atr_multiplier}, Scaled ATR: {scaledAtrValue}")
except Exception as e:
algorithm.Error(f"Error in setting trailing stop for {symbol}: {str(e)}")
#region imports
from AlgorithmImports import *
#endregion
class AtrBasedTrailingStopRiskManagementModel(RiskManagementModel):
'''
Evaluates each invested security against its trailing stop price, calculated using the initial ATR value. Generates liquidation targets for securities breaching their trailing stops.
This Risk Model adjusts the trailing stop based on a multiplier of the ATR value at the time of position initiation, and it maintains this initial ATR value for the life of the trade.
This approach aligns with the objective of using a consistent ATR value for risk management and stop-loss calculations, regardless of subsequent ATR fluctuations.
Ensures consistent risk management and stop-loss calculations using initial ATR values, irrespective of ATR fluctuations.
'''
def __init__(self,algorithm):
'''
The model initializes dictionaries for storing trailing stop percentages, peak prices, ATR multipliers, and initial ATR values.
This is a solid foundation for tracking the necessary data for each security.
'''
self.algorithm = algorithm
self.trailingStopPercentages = {}
self.peakPrices = {}
self.atrMultipliers = {}
# Store the initial ATR values for each symbol
self.initialAtrValues = {}
def ManageRisk(self, algorithm, targets):
'''
The ManageRisk method evaluates each invested security to determine if the current price breaches the trailing stop price, calculated using the initial ATR value.
This ensures that the stop adjusts based on the price movement relative to the entry point, using the ATR value at trade initiation.
Liquidation targets are generated for securities that breach their trailing stops, aligning with the objective of preserving capital by limiting losses.
'''
updatedTargets = []
for kvp in algorithm.Securities:
symbol = kvp.Key
security = kvp.Value
currentPrice = security.Price
holdings = security.Holdings
if symbol in self.trailingStopPercentages and security.Invested:
# Use the initial ATR value for the trailing stop calculation
initialAtrValue = self.initialAtrValues.get(symbol, 0)
trailingStopPrice = self.CalculateTrailingStopPriceUsingInitialATR(symbol, initialAtrValue, holdings)
if trailingStopPrice is None:
algorithm.Log(f"ManageRisk: No trailing stop price calculated for {symbol}.")
# Handle the None case, perhaps continue to the next loop iteration or log a warning
continue
if self.IsStopTriggered(self.algorithm, symbol, currentPrice, trailingStopPrice, holdings):
updatedTargets.append(PortfolioTarget(symbol, 0))
algorithm.Log(f"Trailing stop triggered for {symbol}, liquidating position at current price {currentPrice} with trailing stop at {trailingStopPrice}.")
self.RemoveSymbolData(symbol)
else:
algorithm.Log(f"Trailing stop not triggered for {symbol}. Current price: {currentPrice}, Trailing Stop: {trailingStopPrice}")
return updatedTargets
def SetTrailingStop(self, algorithm, symbol, scaledAtrValue):
"""
Sets the initial trailing stop values for a security using the scaled ATR value.
Args:
algorithm (QCAlgorithm): The algorithm instance.
symbol (Symbol): The symbol for the security.
scaledAtrValue (float): The scaled ATR value (already adjusted by the multiplier).
"""
initialPrice = algorithm.Securities[symbol].Price
atrPercentage = scaledAtrValue / initialPrice
self.initialAtrValues[symbol] = scaledAtrValue
self.trailingStopPercentages[symbol] = atrPercentage
self.peakPrices[symbol] = initialPrice # Initialize peak price at position entry
# self.peakPrices.pop(symbol, None) # Remove this line as it's not needed and incorrect here
algorithm.Log(f"Set initial trailing stop for {symbol} with ATR Percentage: {atrPercentage*100:.2f}%, based on scaled ATR Value: {scaledAtrValue} and initial price: {initialPrice}.")
##### Additional helper methods for clarity and organization ####
def UpdatePeakPrices(self, symbol, high, low):
"""
Updates the peak prices for a symbol based on new highs for long positions and new lows for short positions.
Args:
symbol (Symbol): The trading symbol.
high (float): The high price of the current bar.
low (float): The low price of the current bar.
"""
# Ensure there's an existing position to manage
if symbol in self.algorithm.Portfolio and self.algorithm.Portfolio[symbol].Invested:
# Get the SecurityHolding object
holdings = self.algorithm.Portfolio[symbol]
# Update peak price for long positions based on new highs
if holdings.IsLong:
if symbol not in self.peakPrices or high > self.peakPrices[symbol]:
self.peakPrices[symbol] = high
self.algorithm.Log(f"Updated peak high for {symbol}: {high}")
# Update peak price for short positions based on new lows
elif holdings.IsShort:
if symbol not in self.peakPrices or low < self.peakPrices[symbol]:
self.peakPrices[symbol] = low
self.algorithm.Log(f"Updated peak low for {symbol}: {low}")
def CalculateTrailingStopPriceUsingInitialATR(self, symbol, initialAtrValue, holdings):
# Retrieve the peak price and initial ATR value, safely handling missing keys
peak_price = self.peakPrices.get(symbol, None)
initial_atr_value = self.initialAtrValues.get(symbol, None)
# If either value is not found, handle it appropriately
if peak_price is None or initial_atr_value is None:
self.algorithm.Debug(f"CalculateTrailingStopPriceUsingInitialATR: Missing data for {symbol}")
return None # or some appropriate action
# Calculate the trailing stop price using the initial ATR value retrieved safely
if holdings.IsLong:
return peak_price - initial_atr_value
else:
return peak_price + initial_atr_value
def IsStopTriggered(self, algorithm, symbol, currentPrice, trailingStopPrice, holdings):
# Check if trailingStopPrice is None and handle the case appropriately
if trailingStopPrice is None:
algorithm.Debug(f"IsStopTriggered: Trailing stop price is None for {symbol}")
return False # Or handle it in a way that makes sense for your strategy
if (holdings.IsLong and currentPrice <= trailingStopPrice) or \
(holdings.IsShort and currentPrice >= trailingStopPrice):
return True
return False
def RemoveSymbolData(self, symbol):
'''
The RemoveSymbolData method cleans up data for securities that are no longer part of the portfolio, maintaining the model's efficiency and accuracy.
'''
# Remove the symbol's data from dictionaries after liquidation
self.trailingStopPercentages.pop(symbol, None)
self.peakPrices.pop(symbol, None)
self.initialAtrValues.pop(symbol, None)
self.algorithm.Log(f"Cleaned up data for {symbol} after liquidation.")#region imports
#region imports
from AlgorithmImports import *
#endregion
from AlphaModel import *
from PortfolioConstructionModel import FixedFractionPortfolioConstructionModel
from RiskManagement import AtrBasedTrailingStopRiskManagementModel
class SimpleAlgorithm(QCAlgorithm):
'''
A QuantConnect algorithm that uses both coarse and fine universe selection models to create a universe
of securities based on certain fundamental factors. The algorithm uses an alpha model for signal generation,
equal weighting portfolio construction model, and immediate execution model for trades.
'''
def Initialize(self):
'''Initialize the data and resolution required, as well as the cash and start-end dates for the algorithm.'''
self.SetStartDate(2020, 1, 1)
self.SetEndDate(2023, 1, 1)
self.SetCash(100000)
# Set the Universe Resolution
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.CoarseSelectionFunction)
# Alpha Model
self.AddAlpha(PercentBAlphaModel())
# Initialize and set the risk management model
self.riskManagementModel = AtrBasedTrailingStopRiskManagementModel(self)
self.SetRiskManagement(self.riskManagementModel)
# Portfolio construction model
self.atrIndicators = {}
# Initialize and set the portfolio construction model
self.SetPortfolioConstruction(FixedFractionPortfolioConstructionModel(
fraction=0.01,
atr_multiplier=2,
atrIndicators=self.atrIndicators,
riskManagementModel=self.riskManagementModel
))
# Execution model
self.SetExecution(ImmediateExecutionModel())
self.last_rebalance_day = 0
def IsRebalanceDue(self, time):
'''
Checks if a rebalance is due. A rebalance is due on the first trading day of each week.
'''
# Calculate the number of days since the start of the algorithm
days_since_start = (time - self.StartDate).days
# Calculate the day of the week (0=Monday, 6=Sunday)
day_of_week = time.weekday()
# Determine if today is Monday (or another chosen day of the week for rebalancing)
# and if at least a week has passed since the last rebalance
if day_of_week == 0 and (days_since_start - self.last_rebalance_day) >= 7:
self.last_rebalance_day = days_since_start
return time
return None
def CoarseSelectionFunction(self, coarse):
'''
The CoarseSelectionFunction, filters the universe of stocks to only include those whose current price is greater than $10
and whose average daily dollar volume (a proxy for liquidity) is greater than $1 million.
x.Price > 10: This condition filters out all stocks with a current price of $10 or less.
The idea here is to exclude stocks that may be considered too volatile or risky, which often includes lower-priced stocks.
x.DollarVolume > 1e6: This condition ensures that only stocks with an average daily dollar volume greater than $1 million are included.
Dollar volume is calculated by multiplying the average daily volume by the stock price, providing a measure of liquidity.
A higher dollar volume generally indicates that the stock is more liquid, making it easier to enter and exit positions without significantly impacting the stock's price.
'''
# Sort coarse universe by dollar volume, descending
sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
# Filter to select securities with a minimum price and volume threshold
# This example uses arbitrary thresholds for illustration
filtered = [x.Symbol for x in sorted_by_dollar_volume
if x.Price > 10
and x.DollarVolume > 1e6
and x.HasFundamentalData]
# Log the filtered symbol's ticker values
# self.Log(f"Filtered symbols: {[x.Value for x in filtered[:10]]}")
# Return a subset of securities, for example, the top 500 by dollar volume
return filtered[:500]
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
if security.Symbol not in self.atrIndicators:
# Create and store the ATR indicator for the added security
self.atrIndicators[security.Symbol] = self.ATR(security.Symbol, 14, MovingAverageType.Simple, Resolution.Daily)
for security in changes.RemovedSecurities:
if security.Symbol in self.atrIndicators:
# Remove the ATR indicator for the removed security
del self.atrIndicators[security.Symbol]