from AlgorithmImports import *
from datetime import datetime
import math
from scipy.stats import kurtosis
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler, LabelEncoder
from xgboost import XGBClassifier
from collections import deque
class MySecurityInitializer(BrokerageModelSecurityInitializer):
def __init__(self, brokerage_model: IBrokerageModel, security_seeder: ISecuritySeeder):
super().__init__(brokerage_model, security_seeder)
def Initialize(self, security: Security):
# First call the base class initialization
super().Initialize(security)
class CombinedOptionsAlpha(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2022, 5, 1)
self.SetEndDate(2022, 8, 1)
self.SetCash(2000000)
self.SetTimeZone(TimeZones.NewYork)
self.SetWarmup(30)
# Initialize strategies
self.symbols = []
self.straddle_alpha = DeltaHedgedStraddleAlpha(self)
self.condor_alpha = IronCondorAlpha(self)
self.factor_alpha = FactorAlpha(self)
# Set Brokerage Model
self.SetSecurityInitializer(MySecurityInitializer(
self.BrokerageModel,
FuncSecuritySeeder(self.GetLastKnownPrice)
))
# Initialize performance tracking
self.strategy_performance = {
self.straddle_alpha: deque(maxlen=10), # Store last 10 trades
self.condor_alpha: deque(maxlen=10),
self.factor_alpha: deque(maxlen=10)
}
# Initialize strategy returns tracking
self.strategy_returns = {
self.straddle_alpha: [],
self.condor_alpha: [],
self.factor_alpha: []
}
# Initial equal weights for each alpha
self.alpha_weights = {
self.straddle_alpha: 1/3,
self.condor_alpha: 1/3,
self.factor_alpha: 1/3,
}
# Track enabled status of strategies
self.strategy_enabled = {
self.straddle_alpha: True,
self.condor_alpha: True,
self.factor_alpha: True
}
# Performance evaluation parameters
self.evaluation_period = timedelta(days=14) # 2 weeks
self.last_evaluation = None
self.min_allocation = 0.20 # Minimum allocation per strategy
self.baseline_allocation = 1/3 # Default equal weight
# Risk management parameters
self.profit_target = 1.5
self.stop_loss = 0.75
self.Log(f"[{self.Time}] Initialized CombinedOptionsAlpha with 3 strategies.")
def OnData(self, slice):
if self.IsWarmingUp:
return
self.ManagePositions()
# Update strategy returns and adjust allocations every 2 weeks
self.UpdateStrategyPerformance()
# Pass option chain data to alpha models
if slice.OptionChains:
if self.strategy_enabled[self.condor_alpha]:
self.condor_alpha.OnOptionChainChanged(slice)
if self.strategy_enabled[self.straddle_alpha]:
self.straddle_alpha.OnOptionChainChanged(slice)
if self.strategy_enabled[self.factor_alpha]:
if not self.symbols:
return
self.ExecuteStrategyOrders(self.factor_alpha, slice)
current_time = self.Time
# Straddle at 11:30
if (current_time.hour == 11 and
current_time.minute == 30 and
self.strategy_enabled[self.straddle_alpha]):
if self.straddle_alpha.ShouldTrade(slice):
self.Log("Executing Straddle Strategy")
self.ExecuteStrategyOrders(self.straddle_alpha, slice)
# Iron Condor between 15:00 and 15:05
if (current_time.hour == 15 and
0 <= current_time.minute <= 5 and
self.strategy_enabled[self.condor_alpha]):
if self.condor_alpha.ShouldTrade(slice):
self.ExecuteStrategyOrders(self.condor_alpha, slice)
def UpdateStrategyPerformance(self):
"""Update strategy performance and adjust allocations"""
current_time = self.Time
if (self.last_evaluation is None or
(current_time - self.last_evaluation) >= self.evaluation_period):
self.Log("Updating strategy performance and allocations...")
# Calculate returns for each strategy
strategy_metrics = {}
for strategy in [self.straddle_alpha, self.condor_alpha, self.factor_alpha]:
returns = self.CalculateStrategyReturns(strategy)
sharpe = self.CalculateStrategySharpe(returns)
strategy_metrics[strategy] = {
'returns': returns,
'sharpe': sharpe
}
self.Log(f"{strategy.__class__.__name__} - Returns: {returns:.2%}, Sharpe: {sharpe:.2f}")
# Adjust allocations based on performance
self.AdjustAllocations(strategy_metrics)
self.last_evaluation = current_time
def CalculateStrategyReturns(self, strategy):
"""Calculate returns for a strategy over the evaluation period"""
if not self.strategy_returns[strategy]:
return 0.0
returns = self.strategy_returns[strategy]
if len(returns) < 2:
return 0.0
# Calculate return over the period
start_value = returns[0]
end_value = returns[-1]
if start_value == 0:
return 0.0
return (end_value - start_value) / start_value
def CalculateStrategySharpe(self, returns):
"""Calculate Sharpe ratio for a strategy"""
if not returns:
return 0.0
returns_series = pd.Series(returns)
if len(returns_series) < 2:
return 0.0
return_mean = returns_series.mean()
return_std = returns_series.std()
if return_std == 0:
return 0.0
# Assuming risk-free rate of 0 for simplicity
sharpe = np.sqrt(252) * (return_mean / return_std) # Annualized
return sharpe
def AdjustAllocations(self, strategy_metrics):
"""Adjust strategy allocations based on performance metrics"""
total_score = 0
scores = {}
# Calculate scores based on both returns and Sharpe ratio
for strategy, metrics in strategy_metrics.items():
# Combine returns and Sharpe into a single score
score = (metrics['returns'] + metrics['sharpe']) / 2
score = max(score, 0) # Ensure non-negative score
scores[strategy] = score
total_score += score
# Calculate new weights ensuring minimum allocations
if total_score > 0:
new_weights = {}
remaining_allocation = 1.0
strategies_below_min = []
# First pass: identify strategies below minimum allocation
for strategy, score in scores.items():
weight = score / total_score
if weight < self.min_allocation:
new_weights[strategy] = self.min_allocation
remaining_allocation -= self.min_allocation
strategies_below_min.append(strategy)
# Second pass: allocate remaining capital proportionally
remaining_score = sum(scores[s] for s in scores if s not in strategies_below_min)
if remaining_score > 0:
for strategy, score in scores.items():
if strategy not in strategies_below_min:
weight = (score / remaining_score) * remaining_allocation
new_weights[strategy] = weight
else:
# If no remaining score, distribute equally
remaining_strategies = len(scores) - len(strategies_below_min)
if remaining_strategies > 0:
equal_weight = remaining_allocation / remaining_strategies
for strategy in scores:
if strategy not in strategies_below_min:
new_weights[strategy] = equal_weight
# Update alpha weights
for strategy, weight in new_weights.items():
self.alpha_weights[strategy] = weight
self.Log(f"New allocation for {strategy.__class__.__name__}: {weight:.2%}")
else:
# If no positive scores, revert to baseline equal allocation
for strategy in strategy_metrics:
self.alpha_weights[strategy] = self.baseline_allocation
self.Log(f"Reverting {strategy.__class__.__name__} to baseline allocation: {self.baseline_allocation:.2%}")
def ExecuteStrategyOrders(self, strategy, slice):
"""Execute orders for a specific strategy with weight applied"""
trade_orders = strategy.GenerateOrders(slice)
if trade_orders:
weight = self.alpha_weights[strategy]
self.Log(f"Executing orders for {strategy.__class__.__name__} with {weight:.2%} allocation")
weighted_orders = self.WeightOrders(trade_orders, weight)
if weighted_orders:
self.ExecuteOrders(weighted_orders)
self.Log(f"Executed {len(weighted_orders)} orders for {strategy.__class__.__name__}")
def WeightOrders(self, orders, weight):
"""Apply strategy weight to order quantities"""
self.Log(f"Weighting orders with weight: {weight}")
weighted_orders = []
try:
for order in orders:
if len(order) == 2: # Iron Condor case
strategy, quantity = order
weighted_quantity = max(1, int(quantity * weight))
weighted_orders.append((strategy, weighted_quantity))
self.Log(f"Weighted Iron Condor order: {weighted_quantity} contracts")
else: # Straddle and Factor Alpha case
symbol, quantity, is_buy = order
if isinstance(symbol, str): # Factor Alpha case
weighted_quantity = int(quantity * weight)
else: # Straddle case
weighted_quantity = max(1, int(quantity * weight))
if weighted_quantity > 0:
weighted_orders.append((symbol, weighted_quantity, is_buy))
self.Log(f"Weighted {'long' if is_buy else 'short'} order for {symbol}: {weighted_quantity}")
return weighted_orders
except Exception as e:
self.Error(f"Error in WeightOrders: {str(e)}")
return []
def ExecuteOrders(self, orders):
"""Execute the weighted orders"""
for order in orders:
try:
if len(order) == 2: # Iron Condor case
strategy, quantity = order
self.Buy(strategy, quantity)
self.Log(f"Executing Iron Condor order: {quantity} contracts")
else: # Straddle or Factor Alpha case
symbol, quantity, is_buy = order
if is_buy:
self.Buy(symbol, quantity)
self.Log(f"Buying {quantity} of {symbol}")
else:
self.Sell(symbol, quantity)
self.Log(f"Selling {quantity} of {symbol}")
except Exception as e:
self.Error(f"Order execution failed: {str(e)}")
def OnOrderEvent(self, orderEvent):
"""Track trade performance for each strategy"""
if orderEvent.Status == OrderStatus.Filled:
# Determine which strategy the order belongs to
for strategy in [self.straddle_alpha, self.condor_alpha, self.factor_alpha]:
if strategy.trade_open:
# Calculate trade P&L
trade_pnl = orderEvent.FillPrice * orderEvent.FillQuantity
self.strategy_performance[strategy].append(trade_pnl)
# Update strategy returns
current_return = trade_pnl / self.Portfolio.TotalPortfolioValue
self.strategy_returns[strategy].append(current_return)
self.Log(f"{strategy.__class__.__name__} trade P&L: ${trade_pnl:.2f}")
def ManagePositions(self):
"""Centralized position management for all strategies"""
if not self.Portfolio.Invested:
return
total_pnl = sum([holding.UnrealizedProfit
for holding in self.Portfolio.Values
if holding.Invested])
# For each strategy, check if its positions need management
for alpha in [self.straddle_alpha, self.condor_alpha, self.factor_alpha]:
if hasattr(alpha, 'trade_open') and alpha.trade_open:
if hasattr(alpha, 'initial_credit'): # Iron Condor case
if total_pnl >= alpha.initial_credit * self.profit_target:
self.Liquidate()
alpha.trade_open = False
self.Log(f"Closed position at profit target on {self.Time}")
elif total_pnl <= -alpha.max_potential_loss * self.stop_loss:
self.Liquidate()
alpha.trade_open = False
self.Log(f"Closed position at stop loss on {self.Time}")
elif hasattr(alpha, 'max_potential_loss'): # Straddle case
if total_pnl >= alpha.max_potential_loss * self.profit_target:
self.Liquidate()
alpha.trade_open = False
elif total_pnl <= -alpha.max_potential_loss * self.stop_loss:
self.Liquidate()
alpha.trade_open = False
class FactorAlpha:
def __init__(self, algorithm):
self.algorithm = algorithm
self.Initialize()
def Initialize(self):
self.algorithm.Log("Initializing FactorAlpha")
self.algorithm.UniverseSettings.Resolution = Resolution.Daily
self.algorithm.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
self.num_stocks = 500
self.trade_open = False
self.num_groups = 10
self.current_month = -1
self.model = None
self.last_month_features = pd.DataFrame()
self.label_encoder = LabelEncoder()
self.predicted_stocks = {'long': [], 'short': []}
self.position_size = 0.1 # 10% of portfolio per position
# Initialize XGBoost model
self.model = XGBClassifier(
n_estimators=100,
learning_rate=0.1,
max_depth=5,
random_state=42,
objective='multi:softprob'
)
self.algorithm.Log("FactorAlpha initialization complete")
def CoarseSelectionFunction(self, coarse):
self.algorithm.Log(f"Running CoarseSelectionFunction at {self.algorithm.Time}")
if self.algorithm.Time.month == self.current_month:
self.algorithm.Log("Same month - returning unchanged universe")
return Universe.Unchanged
self.current_month = self.algorithm.Time.month
try:
sorted_by_volume = sorted(
[x for x in coarse if x.HasFundamentalData],
key=lambda x: x.Market,
reverse=True
)
self.algorithm.Log(f"Found {len(sorted_by_volume)} stocks with fundamental data")
selected_symbols = [x.Symbol for x in sorted_by_volume[:self.num_stocks]]
self.algorithm.Log(f"Selected {len(selected_symbols)} symbols in coarse selection")
return selected_symbols
except Exception as e:
self.algorithm.Error(f"Error in CoarseSelectionFunction: {str(e)}")
return []
def FineSelectionFunction(self, fine):
self.algorithm.Log(f"Running FineSelectionFunction at {self.algorithm.Time}")
fine_list = list(fine)
if not fine_list:
self.algorithm.Log("Empty fine data received")
return []
try:
current_month_features = pd.DataFrame()
current_month_returns = pd.DataFrame()
for stock in fine_list:
try:
symbol = str(stock.Symbol)
# Get historical data
history = self.algorithm.History(stock.Symbol, 20, Resolution.Daily)
if len(history) < 20:
continue
# Calculate features
daily_returns = history['close'].pct_change().dropna()
volatility = daily_returns.std() * np.sqrt(252)
momentum = stock.ValuationRatios.PriceChange1M
# Value calculation
if stock.ValuationRatios.PERatio > 0 and stock.ValuationRatios.PERatio < 100:
value = 1 / stock.ValuationRatios.PERatio
else:
continue
size = np.log(stock.MarketCap) if stock.MarketCap > 0 else np.nan
quality = stock.OperationRatios.ROE.Value
pb = stock.ValuationRatios.PBRatio
margin = stock.OperationRatios.GrossMargin.OneMonth
# Store features
current_month_features.loc[symbol, 'Momentum'] = momentum
current_month_features.loc[symbol, 'Value'] = value
current_month_features.loc[symbol, 'Size'] = size
current_month_features.loc[symbol, 'Quality'] = quality
current_month_features.loc[symbol, 'Volatility'] = volatility
current_month_features.loc[symbol, 'PB'] = pb
current_month_features.loc[symbol, 'Margin'] = margin
# Calculate returns
first_price = history['close'].iloc[0]
last_price = history['close'].iloc[-1]
log_return = np.log(last_price / first_price)
current_month_returns.loc[symbol, 'Returns'] = log_return
except Exception as e:
self.algorithm.Log(f"Error processing individual stock {symbol}: {str(e)}")
continue
if current_month_features.empty:
self.algorithm.Log("No features collected this month")
return []
if self.last_month_features.empty:
self.algorithm.Log("Storing first month's features")
self.last_month_features = current_month_features
return []
self.algorithm.Log("Training model with previous month's data")
# Prepare training data
X_train = self.last_month_features
y_train = current_month_returns
common_symbols = X_train.index.intersection(y_train.index)
X_train = X_train.loc[common_symbols]
y_train = y_train.loc[common_symbols]
# Process features
X_train = X_train.fillna(X_train.median())
y_classes = pd.qcut(y_train['Returns'], q=self.num_groups, labels=False)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
# Train model
self.model.fit(X_train_scaled, y_classes)
self.algorithm.Log("Model training completed")
# Make predictions
predictions = self.PredictGroups(current_month_features)
if predictions.empty:
self.algorithm.Log("No predictions generated")
return []
# Update predicted stocks for trading
self.predicted_stocks['long'] = list(predictions[predictions['predicted_group'] == self.num_groups - 1].index)
self.predicted_stocks['short'] = list(predictions[predictions['predicted_group'] == 0].index)
self.algorithm.Log(f"Selected {len(self.predicted_stocks['long'])} long and {len(self.predicted_stocks['short'])} short positions")
# Convert string symbols back to Symbol objects
selected_symbols = []
for symbol_str in self.predicted_stocks['long'] + self.predicted_stocks['short']:
for stock in fine_list:
if str(stock.Symbol) == symbol_str:
selected_symbols.append(stock.Symbol)
break
self.algorithm.symbols = selected_symbols
self.last_month_features = current_month_features
return selected_symbols
except Exception as e:
self.algorithm.Error(f"Error in FineSelectionFunction: {str(e)}")
return []
def PredictGroups(self, features):
self.algorithm.Log("Making predictions for current month")
try:
features = features.fillna(features.mean())
scaler = StandardScaler()
features_scaled = scaler.fit_transform(features)
class_probs = self.model.predict_proba(features_scaled)
predicted_classes = np.argmax(class_probs, axis=1)
predictions = pd.DataFrame({
'predicted_group': predicted_classes,
'confidence': np.max(class_probs, axis=1)
}, index=features.index)
self.algorithm.Log(f"Generated predictions for {len(predictions)} stocks")
return predictions
except Exception as e:
self.algorithm.Error(f"Error in PredictGroups: {str(e)}")
return pd.DataFrame()
def GenerateOrders(self, slice):
"""Generate orders based on predictions"""
self.algorithm.Log("Generating orders for Factor Alpha")
if not hasattr(self.algorithm, 'symbols') or not self.algorithm.symbols:
self.algorithm.Log("No symbols available for trading")
return []
try:
orders = []
portfolio_value = self.algorithm.Portfolio.TotalPortfolioValue
position_value = portfolio_value * self.position_size
# Process long positions
for symbol_str in self.predicted_stocks['long']:
symbol = None
for s in self.algorithm.symbols:
if str(s) == symbol_str:
symbol = s
break
if symbol is None:
continue
# Get current price
security = self.algorithm.Securities[symbol]
if security.Price == 0:
continue
# Calculate position size
quantity = int(position_value / security.Price)
if quantity > 0:
orders.append((symbol, quantity, True)) # True for buy
self.algorithm.Log(f"Generated long order for {symbol}: {quantity} shares")
# Process short positions
for symbol_str in self.predicted_stocks['short']:
symbol = None
for s in self.algorithm.symbols:
if str(s) == symbol_str:
symbol = s
break
if symbol is None:
continue
# Get current price
security = self.algorithm.Securities[symbol]
if security.Price == 0:
continue
# Calculate position size
quantity = int(position_value / security.Price)
if quantity > 0:
orders.append((symbol, quantity, False)) # False for sell/short
self.algorithm.Log(f"Generated short order for {symbol}: {quantity} shares")
self.trade_open = True
return orders
except Exception as e:
self.algorithm.Error(f"Error generating orders in FactorAlpha: {str(e)}")
return []
class IronCondorAlpha:
def __init__(self, algorithm):
self.algorithm = algorithm # Store reference to main algorithm
self.Initialize()
def Initialize(self):
# Add SPX index
self.index = self.algorithm.AddIndex("SPX")
# Universe 1 (option1): Wide filter for kurtosis calculations
self.option1 = self.algorithm.AddIndexOption(self.index.Symbol, "SPXW")
self.option1.SetFilter(lambda universe: universe.IncludeWeeklys()
.Strikes(-30,30).Expiration(0, 0))
self._symbol1 = self.option1.Symbol
# Universe 2 (option2): Iron Condor filter for placing trades
self.option2 = self.algorithm.AddIndexOption(self.index.Symbol, "SPXW")
self.option2.SetFilter(lambda x: x.IncludeWeeklys().IronCondor(0, 20, 40))
self._symbol2 = self.option2.Symbol
# Risk and trade management parameters
self.max_portfolio_risk = 0.05
self.profit_target = 1.5
self.stop_loss = 0.75
self.trade_open = False
self.initial_credit = 0
self.max_potential_loss = 0
self.target_delta = 0.25
self.kurtosis_threshold = 2 # Changed to match original
self.current_date = None
self.kurtosis_condition_met = False
self.computed_kurtosis_today = False
def OnOptionChainChanged(self, slice):
# Check if a new day has started
if self.current_date != self.algorithm.Time.date():
self.current_date = self.algorithm.Time.date()
self.trade_open = False
self.kurtosis_condition_met = False
self.computed_kurtosis_today = False
self.algorithm.Log(f"New day reset for Iron Condor at {self.algorithm.Time}")
# Compute kurtosis at 9:31-9:36 AM
if (not self.computed_kurtosis_today and
self.algorithm.Time.hour == 9 and
self.algorithm.Time.minute >= 31 and
self.algorithm.Time.minute <= 36):
chain1 = slice.OptionChains.get(self._symbol1)
if chain1:
iv_values = [x.ImpliedVolatility for x in chain1
if x.ImpliedVolatility and 0 < x.ImpliedVolatility < 5]
if len(iv_values) > 10: # Using 10 as in original
daily_kurtosis = kurtosis(iv_values)
self.algorithm.Log(f"Iron Condor Kurtosis: {daily_kurtosis}")
if daily_kurtosis > self.kurtosis_threshold:
self.kurtosis_condition_met = True
self.algorithm.Log("Iron Condor Kurtosis condition met")
self.computed_kurtosis_today = True
def ShouldTrade(self, slice):
# Only check if we should trade based on conditions, not time
return (not self.algorithm.Portfolio.Invested and
self.kurtosis_condition_met)
def GenerateOrders(self, slice):
chain2 = slice.OptionChains.get(self._symbol2)
if not chain2:
return None
expiry = max([x.Expiry for x in chain2])
chain2 = sorted([x for x in chain2 if x.Expiry == expiry],
key=lambda x: x.Strike)
put_contracts = [x for x in chain2
if x.Right == OptionRight.PUT and
abs(x.Greeks.Delta) <= self.target_delta]
call_contracts = [x for x in chain2
if x.Right == OptionRight.CALL and
abs(x.Greeks.Delta) <= self.target_delta]
if len(call_contracts) < 2 or len(put_contracts) < 2:
return None
near_call = min(call_contracts,
key=lambda x: abs(x.Greeks.Delta - self.target_delta))
far_call = min([x for x in call_contracts if x.Strike > near_call.Strike],
key=lambda x: abs(x.Greeks.Delta - self.target_delta))
near_put = min(put_contracts,
key=lambda x: abs(x.Greeks.Delta + self.target_delta))
far_put = min([x for x in put_contracts if x.Strike < near_put.Strike],
key=lambda x: abs(x.Greeks.Delta + self.target_delta))
credit = (near_call.BidPrice - far_call.AskPrice) + (near_put.BidPrice - far_put.AskPrice)
spread_width = max(far_call.Strike - near_call.Strike,
near_put.Strike - far_put.Strike)
max_potential_loss = spread_width * 100 - credit * 100
total_portfolio_value = self.algorithm.Portfolio.TotalPortfolioValue
max_trade_risk = total_portfolio_value * self.max_portfolio_risk
contracts = int(max_trade_risk / max_potential_loss)
if contracts > 0:
iron_condor = OptionStrategies.IronCondor(
self._symbol2,
far_put.Strike,
near_put.Strike,
near_call.Strike,
far_call.Strike,
expiry
)
# Store trade parameters for position management
self.initial_credit = credit * 100 * contracts
self.max_potential_loss = max_potential_loss * contracts
self.trade_open = True
self.algorithm.Log(f"Generated iron condor at {self.algorithm.Time}, "
f"Contracts: {contracts}, Credit: ${self.initial_credit:.2f}")
return [(iron_condor, contracts)]
return None
class DeltaHedgedStraddleAlpha:
def __init__(self, algorithm):
self.algorithm = algorithm
self.Initialize()
def Initialize(self):
# Add SPX index
self.index = self.algorithm.AddIndex("SPX")
# Add SPY for Delta Hedging
self.spy = self.algorithm.AddEquity("SPY", Resolution.Minute)
self.spy.SetLeverage(1)
self.spy.SetDataNormalizationMode(DataNormalizationMode.Raw)
self.spy_symbol = self.spy.Symbol
# Add SPX options
self.option = self.algorithm.AddIndexOption(self.index.Symbol, "SPXW")
self.option.SetFilter(lambda universe: universe.IncludeWeeklys()
.Strikes(-30, 30).Expiration(0, 0))
self.option_symbol = self.option.Symbol
# Risk and trade management parameters
self.max_portfolio_risk = 0.05
self.profit_target = 1.5
self.stop_loss = 0.75
self.trade_open = False
# Kurtosis calculation variables
self.kurtosis_threshold = 0
self.kurtosis_condition_met = False
self.computed_kurtosis_today = False
self.current_date = None
# Variables for delta hedging
self.hedge_order_ticket = None
self.net_delta = 0.0
self.max_potential_loss = 0.0
def OnOptionChainChanged(self, slice):
# Check if a new day has started
if self.current_date != self.algorithm.Time.date():
self.current_date = self.algorithm.Time.date()
self.trade_open = False
self.kurtosis_condition_met = False
self.computed_kurtosis_today = False
self.algorithm.Log(f"New day reset for Straddle at {self.algorithm.Time}")
# Liquidate any existing hedge at the start of a new day
if self.hedge_order_ticket and self.hedge_order_ticket.Status not in [OrderStatus.Filled, OrderStatus.Canceled]:
self.algorithm.CancelOrder(self.hedge_order_ticket.OrderId)
self.algorithm.Liquidate(self.spy_symbol)
self.algorithm.Liquidate(self.option_symbol)
# Compute kurtosis from option chain at 9:31-9:36 AM
if (not self.computed_kurtosis_today and
self.algorithm.Time.hour == 9 and
self.algorithm.Time.minute >= 31 and
self.algorithm.Time.minute <= 36):
chain = slice.OptionChains.get(self.option_symbol)
if chain:
iv_values = [x.ImpliedVolatility for x in chain
if x.ImpliedVolatility and 0 < x.ImpliedVolatility < 5]
if len(iv_values) > 3:
daily_kurtosis = kurtosis(iv_values)
if daily_kurtosis > self.kurtosis_threshold:
self.kurtosis_condition_met = True
self.algorithm.Log(f"Straddle Kurtosis met: {daily_kurtosis}")
self.computed_kurtosis_today = True
def ShouldTrade(self, slice):
return (not self.trade_open and
self.kurtosis_condition_met)
def GenerateOrders(self, slice):
chain = slice.OptionChains.get(self.option_symbol)
if not chain:
return None
# Find ATM strike
atm_strike = self.index.Price
closest_option = min(chain, key=lambda x: abs(x.Strike - atm_strike))
atm_strike = closest_option.Strike
# Filter for ATM call and put contracts with the highest Vega
atm_call_candidates = [x for x in chain
if x.Strike == atm_strike and
x.Right == OptionRight.CALL]
atm_put_candidates = [x for x in chain
if x.Strike == atm_strike and
x.Right == OptionRight.PUT]
if not atm_call_candidates or not atm_put_candidates:
return None
# Select contracts with highest Vega
atm_call = max(atm_call_candidates, key=lambda x: x.Greeks.Vega)
atm_put = max(atm_put_candidates, key=lambda x: x.Greeks.Vega)
# Calculate credit received from selling the straddle
credit = atm_call.BidPrice + atm_put.BidPrice
max_loss = abs(atm_call.Strike - self.index.Price) * 100 + credit * 100
if max_loss <= 0:
return None
# Position size calculation
total_portfolio_value = self.algorithm.Portfolio.TotalPortfolioValue
max_trade_risk = total_portfolio_value * self.max_portfolio_risk
contracts = int(max_trade_risk / max_loss)
if contracts <= 0:
return None
# Calculate delta hedge - Converting SPX delta to SPY (dividing by 10)
net_delta = (atm_call.Greeks.Delta + atm_put.Greeks.Delta) * contracts
required_spy_position = int(-net_delta * 10) # Multiply by 10 as SPX/SPY ratio is roughly 10:1
# Store trade parameters
self.trade_open = True
self.max_potential_loss = max_loss * contracts
self.net_delta = net_delta
# Log the hedge calculation
self.algorithm.Log(f"Delta calculation: Call Delta={atm_call.Greeks.Delta}, "
f"Put Delta={atm_put.Greeks.Delta}, "
f"Contracts={contracts}, "
f"Net Delta={net_delta}, "
f"Required SPY Position={required_spy_position}")
# Return orders as tuples: (symbol, quantity, is_buy)
orders = [
(atm_call.Symbol, contracts, False), # Sell call
(atm_put.Symbol, contracts, False), # Sell put
(self.spy_symbol, abs(required_spy_position), required_spy_position > 0) # Hedge
]
self.algorithm.Log(f"Generated straddle orders: Straddle Contracts={contracts}, "
f"Delta Hedge Size={abs(required_spy_position)}, "
f"Hedge Direction={'Long' if required_spy_position > 0 else 'Short'}")
return orders