Calendar Spread Strategy Guide: Exploit Time Decay Differences
Calendar spreads (also called time spreads) profit from differential time decay between options at different expirations. This guide covers building and managing calendar spread portfolios.Calendar Spread Fundamentals
A calendar spread involves:- Selling near-term option (generates premium)
- Buying longer-term option (decays slower)
- Profit from faster decay of near-term option
- Benefits from time decay (positive theta)
- Profits when underlying stays near strike
- Benefits from rising implied volatility
- Losses from large underlying moves
Calendar Spread Implementation
``python
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple
from scipy.stats import norm
import logging
from datetime import datetime, timedelta
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CalendarSpreadBuilder:
"""Build and manage calendar spreads"""
def __init__(self, S: float, r: float = 0.05):
self.S = S
self.r = r
def black_scholes_call(self, S: float, K: float, T: float, sigma: float) -> float:
"""Calculate call price"""
if T <= 0:
return max(S - K, 0)
d1 = (np.log(S / K) + (self.r + 0.5 sigma 2) T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
return S norm.cdf(d1) - K np.exp(-self.r T) norm.cdf(d2)
def design_calendar_spread(self, K: float, T_short: float, T_long: float,
sigma_short: float, sigma_long: float) -> Dict:
"""Design calendar spread at same strike"""
# Sell near-term, buy far-term
short_price = self.black_scholes_call(self.S, K, T_short, sigma_short)
long_price = self.black_scholes_call(self.S, K, T_long, sigma_long)
net_debit = long_price - short_price
# Max profit occurs at expiration of short option
max_profit_scenario = long_price
return {
'strike': K,
'short_expiration': T_short,
'long_expiration': T_long,
'short_price': short_price,
'long_price': long_price,
'net_debit': net_debit,
'max_profit_at_short_exp': long_price - net_debit,
'theta_daily': (short_price / T_short - long_price / T_long) / 365,
'setup_cost': net_debit
}
def calculate_calendar_spread_pnl(self, current_S: float,
current_T_short: float,
current_T_long: float,
K: float, sigma: float,
initial_setup: Dict) -> float:
"""Calculate current P&L for calendar spread"""
short_value = self.black_scholes_call(current_S, K, current_T_short, sigma)
long_value = self.black_scholes_call(current_S, K, current_T_long, sigma)
current_value = long_value - short_value
pnl = current_value - initial_setup['net_debit']
return pnl
def analyze_calendar_spread_risks(self, spread: Dict,
current_S: float) -> Dict:
"""Analyze risks for calendar spread"""
K = spread['strike']
distance_from_strike = abs(current_S - K) / K
risks = {
'price_move_risk': 'HIGH' if distance_from_strike > 0.05 else 'MEDIUM' if distance_from_strike > 0.02 else 'LOW',
'volatility_risk': 'HIGH if IV drops after sale',
'assignment_risk': 'SHORT call may be assigned early',
'recommendations': []
}
if distance_from_strike > 0.1:
risks['recommendations'].append('Consider rolling to new strike')
if current_S > K:
risks['recommendations'].append('Monitor for early assignment on short call')
return risks
def identify_optimal_calendar_spreads(self, S: float, T_short: float,
T_long: float, sigma: float) -> Dict:
"""Identify which strike gives best calendar spread"""
strikes = [S (0.95 + 0.05 i) for i in range(5)] # Test 5 strikes
best_spread = None
best_theta = 0
for K in strikes:
spread = self.design_calendar_spread(K, T_short, T_long, sigma, sigma)
if spread['theta_daily'] > best_theta:
best_theta = spread['theta_daily']
best_spread = spread
best_spread['strike'] = K
return best_spread
def generate_calendar_pnl_surface(self, K: float, T_short: float,
T_long: float, sigma: float,
spot_range: Tuple[float, float] = None,
days_range: Tuple[float, float] = None) -> np.ndarray:
"""Generate 3D P&L surface for calendar spread"""
if spot_range is None:
spot_range = (self.S 0.9, self.S 1.1)
if days_range is None:
days_range = (0, T_short * 365)
spots = np.linspace(spot_range[0], spot_range[1], 30)
days = np.linspace(days_range[0], days_range[1], 30)
pnl_surface = np.zeros((len(days), len(spots)))
initial_spread = self.design_calendar_spread(K, T_short, T_long, sigma, sigma)
for i, day in enumerate(days):
new_T_short = T_short - (day / 365)
new_T_long = T_long - (day / 365)
for j, spot in enumerate(spots):
if new_T_short > 0:
pnl = self.calculate_calendar_spread_pnl(spot, new_T_short, new_T_long,
K, sigma, initial_spread)
else:
pnl = max(spot - K, 0) - initial_spread['net_debit']
pnl_surface[i, j] = pnl
return pnl_surface
`
Diagonal Spread Variations
`python
class DiagonalSpreadBuilder:
"""Build diagonal spreads (different strikes and expirations)"""
def __init__(self, S: float, r: float = 0.05):
self.S = S
self.r = r
self.base_builder = CalendarSpreadBuilder(S, r)
def build_diagonal_call_spread(self, K_short: float, K_long: float,
T_short: float, T_long: float,
sigma: float) -> Dict:
"""Build diagonal call spread (lower long strike, higher short strike)"""
short_price = self.base_builder.black_scholes_call(self.S, K_short, T_short, sigma)
long_price = self.base_builder.black_scholes_call(self.S, K_long, T_long, sigma)
net_credit = short_price - long_price
return {
'type': 'diagonal_call',
'short_strike': K_short,
'short_expiration': T_short,
'long_strike': K_long,
'long_expiration': T_long,
'net_credit': net_credit,
'max_profit': K_short - K_long + net_credit,
'max_loss': abs(net_credit) - (K_short - K_long) if net_credit < K_short - K_long else net_credit
}
def build_diagonal_put_spread(self, K_short: float, K_long: float,
T_short: float, T_long: float,
sigma: float) -> Dict:
"""Build diagonal put spread"""
# Similar logic but for puts
return {}
`
Rolling Calendar Spreads
`python
class CalendarSpreadRoller:
"""Manage rolling and adjusting calendar spreads"""
def __init__(self):
self.open_spreads = []
self.roll_history = []
def monitor_for_roll(self, spread: Dict, current_date: datetime,
current_S: float) -> Dict:
"""Determine if spread should be rolled"""
days_to_short_exp = (spread['short_expiration'] - datetime.now()).days
# Roll decisions
should_roll = False
roll_reason = ""
# Roll if short option expires in 7 days or less
if days_to_short_exp <= 7:
should_roll = True
roll_reason = "short option near expiration"
# Roll if underlying moved significantly
if abs(current_S - spread['strike']) / spread['strike'] > 0.1:
should_roll = True
roll_reason = "underlying moved > 10% from strike"
return {
'should_roll': should_roll,
'reason': roll_reason,
'days_to_expiration': days_to_short_exp
}
def calculate_roll_cost(self, current_spread: Dict,
new_spread: Dict) -> float:
"""Calculate cost to roll to new spread"""
# Cost = buy back short option + sell new short option
buy_back_cost = current_spread['short_price']
new_premium = new_spread['short_price']
net_cost = buy_back_cost - new_premium
return net_cost
def execute_calendar_roll(self, current_spread: Dict,
roll_target: Dict,
new_T_short: float,
new_T_long: float) -> Dict:
"""Execute roll of calendar spread"""
# Calculate new prices
builder = CalendarSpreadBuilder(current_spread['strike'])
new_spread = builder.design_calendar_spread(
current_spread['strike'],
new_T_short,
new_T_long,
0.2, # Use current sigma
0.2
)
roll_details = {
'timestamp': datetime.now(),
'from_expiration': current_spread['short_expiration'],
'to_expiration': new_T_short,
'roll_cost': self.calculate_roll_cost(current_spread, new_spread),
'new_spread': new_spread,
'cumulative_credit': current_spread['net_debit'] - new_spread['net_debit']
}
self.roll_history.append(roll_details)
return roll_details
`
Calendar Spread Portfolio Management
`python
class CalendarSpreadPortfolio:
"""Manage portfolio of calendar spreads"""
def __init__(self, total_capital: float):
self.capital = total_capital
self.spreads = []
self.portfolio_theta = 0
def add_calendar_spread(self, spread: Dict, quantity: int = 1):
"""Add calendar spread to portfolio"""
self.spreads.append({
'spread': spread,
'quantity': quantity,
'entry_date': datetime.now(),
'status': 'open'
})
def calculate_portfolio_metrics(self) -> Dict:
"""Calculate portfolio-level metrics"""
total_debit = sum(s['spread']['net_debit'] * s['quantity'] for s in self.spreads)
total_theta = sum(s['spread']['theta_daily'] * s['quantity'] for s in self.spreads)
return {
'num_spreads': len(self.spreads),
'total_debit': total_debit,
'total_daily_theta': total_theta,
'annual_theta': total_theta * 365,
'roi_at_max_profit': (total_theta * 365) / total_debit if total_debit > 0 else 0,
'capital_utilization': total_debit / self.capital
}
def identify_portfolio_risks(self) -> List[Dict]:
"""Identify risks across portfolio"""
risks = []
# Check concentration
strikes = [s['spread']['strike'] for s in self.spreads]
if len(set(strikes)) < len(strikes) / 2:
risks.append({
'type': 'concentration',
'message': 'Portfolio concentrated at few strikes',
'recommendation': 'diversify strikes'
})
# Check gamma exposure
total_gamma = sum(0.01 * s['quantity'] for s in self.spreads) # Simplified
if abs(total_gamma) > 10:
risks.append({
'type': 'gamma_imbalance',
'message': f'High gamma exposure: {total_gamma:.2f}',
'recommendation': 'consider hedging'
})
return risks
``