Covered Call Optimization: Algorithmic Income Generation
Covered calls generate income from stock holdings by selling call options. This guide covers systematic selection and optimization of covered call strategies.Covered Call Fundamentals
A covered call combines:- Long 100 shares of stock
- Short 1 call option per 100 shares
Covered Call Selection Algorithm
``python
import numpy as np
import pandas as pd
from scipy.stats import norm
from typing import Dict, List, Tuple
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CoveredCallOptimizer:
"""Optimize covered call strike selection"""
def __init__(self, stock_price: float, dividend_yield: float = 0.02,
risk_free_rate: float = 0.05):
self.S = stock_price
self.q = dividend_yield
self.r = risk_free_rate
def calculate_call_price(self, K: float, T: float, sigma: float) -> float:
"""Calculate call price using Black-Scholes"""
d1 = (np.log(self.S / K) + (self.r - self.q + 0.5 sigma 2) T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
return (self.S np.exp(-self.q T) * norm.cdf(d1) -
K np.exp(-self.r T) * norm.cdf(d2))
def calculate_assignment_probability(self, K: float, T: float, sigma: float) -> float:
"""Calculate probability of assignment"""
d2 = (np.log(self.S / K) + (self.r - self.q - 0.5 sigma 2) T) / (sigma * np.sqrt(T))
return norm.cdf(d2)
def evaluate_covered_call(self, K: float, T: float, sigma: float,
position_size: int = 100) -> Dict:
"""Evaluate covered call at specific strike"""
call_price = self.calculate_call_price(K, T, sigma)
assignment_prob = self.calculate_assignment_probability(K, T, sigma)
# Stock gain if assigned
stock_capital = self.S * position_size
profit_if_assigned = (K - self.S) position_size + call_price position_size
# Dividend income
dividend = self.S self.q T * position_size
# Total return scenarios
expected_return = (call_price + dividend) * position_size
return {
'strike': K,
'call_price': call_price,
'premium_percent': (call_price / self.S) * 100,
'assignment_probability': assignment_prob,
'profit_if_assigned': profit_if_assigned,
'profit_pct_if_assigned': (profit_if_assigned / stock_capital) * 100,
'expected_return': expected_return,
'annual_return': expected_return / stock_capital (365 / (T 365)),
'capital_requirement': stock_capital
}
def find_optimal_strike(self, T: float, sigma: float,
target_return: float = 0.05) -> Dict:
"""Find optimal strike for covered call"""
strikes = np.linspace(self.S 0.95, self.S 1.15, 20)
best_strike = None
best_metrics = None
for K in strikes:
metrics = self.evaluate_covered_call(K, T, sigma)
# Select strike closest to target annual return
if best_metrics is None or abs(metrics['annual_return'] - target_return) < abs(best_metrics['annual_return'] - target_return):
best_metrics = metrics
best_strike = K
best_metrics['optimal_strike'] = best_strike
return best_metrics
def rank_covered_calls(self, expirations: List[float], sigma: float,
position_size: int = 100) -> pd.DataFrame:
"""Rank covered calls across different strikes and expirations"""
results = []
strikes = np.linspace(self.S 0.95, self.S 1.15, 10)
for T in expirations:
for K in strikes:
metrics = self.evaluate_covered_call(K, T, sigma, position_size)
metrics['expiration_days'] = T * 365
results.append(metrics)
df = pd.DataFrame(results)
# Rank by annual return
df['rank'] = df['annual_return'].rank(ascending=False)
return df.sort_values('rank')
`
Multi-Position Covered Call Strategy
`python
class CoveredCallPortfolio:
"""Manage portfolio of covered call positions"""
def __init__(self, total_capital: float):
self.capital = total_capital
self.positions = []
self.allocation = {}
def add_covered_call(self, ticker: str, stock_price: float,
num_shares: int, call_strike: float,
call_price: float, days_to_exp: int):
"""Add covered call position"""
position_value = stock_price * num_shares
annual_income = call_price 100 (365 / days_to_exp)
self.positions.append({
'ticker': ticker,
'stock_price': stock_price,
'shares': num_shares,
'call_strike': call_strike,
'call_price': call_price,
'days_to_expiry': days_to_exp,
'position_value': position_value,
'annual_income': annual_income,
'annual_yield': annual_income / position_value * 100
})
self.allocation[ticker] = position_value / self.capital
def portfolio_metrics(self) -> Dict:
"""Calculate portfolio-level metrics"""
if not self.positions:
return {}
total_income = sum(p['annual_income'] for p in self.positions)
total_value = sum(p['position_value'] for p in self.positions)
avg_yield = np.mean([p['annual_yield'] for p in self.positions])
return {
'num_positions': len(self.positions),
'total_value': total_value,
'annual_income': total_income,
'portfolio_yield': (total_income / total_value) * 100 if total_value > 0 else 0,
'avg_position_yield': avg_yield,
'capital_utilization': total_value / self.capital
}
def identify_assignments_needed(self) -> List[Dict]:
"""Identify which positions need assignment planning"""
assignments_needed = []
for position in self.positions:
if position['days_to_expiry'] < 7:
assignments_needed.append({
'ticker': position['ticker'],
'days_remaining': position['days_to_expiry'],
'action': 'prepare_for_assignment_or_roll'
})
return assignments_needed
def optimize_allocations(self, target_yield: float = 0.08) -> Dict:
"""Optimize position sizes to reach target yield"""
current_yield = self.portfolio_metrics()['portfolio_yield']
if current_yield < target_yield:
# Increase positions with higher yields
rebalancing = {
'action': 'increase_high_yield_positions',
'target_yield': target_yield,
'current_yield': current_yield,
'shortfall': target_yield - current_yield
}
else:
rebalancing = {
'action': 'rebalance_down',
'reason': 'yield target reached',
'current_yield': current_yield
}
return rebalancing
`
Rolling and Management
`python
class CoveredCallRoller:
"""Manage rolling of covered calls at expiration"""
def __init__(self):
self.rolls = []
def should_roll_or_sell(self, position: Dict, current_S: float) -> str:
"""Decide whether to roll or allow assignment"""
call_strike = position['call_strike']
stock_price = position['stock_price']
# If stock above strike, likely to be assigned
if current_S > call_strike * 0.99:
return 'let_assign'
# If stock below strike, roll to collect more premium
if current_S < call_strike * 0.95:
return 'roll_up_and_out'
return 'roll_same_strike'
def calculate_roll_profit(self, position: Dict, new_call_price: float,
current_call_price: float) -> float:
"""Calculate additional profit from roll"""
# Profit from closing current call + premium from new call
close_profit = current_call_price
new_premium = new_call_price
net_credit = close_profit + new_premium
return net_credit
def execute_roll(self, position: Dict, new_strike: float,
new_expiration: float, new_call_price: float) -> Dict:
"""Execute covered call roll"""
roll_record = {
'ticker': position['ticker'],
'from_strike': position['call_strike'],
'to_strike': new_strike,
'new_call_price': new_call_price,
'new_expiration': new_expiration,
'additional_income': new_call_price * 100,
'timestamp': pd.Timestamp.now()
}
self.rolls.append(roll_record)
return roll_record
def analyze_roll_series(self, ticker: str) -> Dict:
"""Analyze history of rolls for a position"""
ticker_rolls = [r for r in self.rolls if r['ticker'] == ticker]
if not ticker_rolls:
return {'rolls': 0}
total_additional_income = sum(r['additional_income'] for r in ticker_rolls)
return {
'rolls': len(ticker_rolls),
'total_additional_income': total_additional_income,
'avg_income_per_roll': total_additional_income / len(ticker_rolls),
'roll_efficiency': 'high' if total_additional_income > 0 else 'low'
}
`
Risk Management for Covered Calls
`python
class CoveredCallRiskManager:
"""Manage risks in covered call strategy"""
def __init__(self, max_upside_cap_percent: float = 0.15):
self.max_upside_cap = max_upside_cap_percent
def calculate_max_loss(self, stock_price: float, purchase_price: float) -> float:
"""Calculate max loss (stock drops to zero)"""
return purchase_price * 100
def calculate_capped_upside(self, stock_price: float, call_strike: float) -> float:
"""Calculate max profit (stock called away)"""
stock_gain = (call_strike - stock_price) * 100
premium = 0 # Already received
return stock_gain + premium
def check_concentration_risk(self, portfolio: CoveredCallPortfolio,
max_single_position: float = 0.20) -> Dict:
"""Check if portfolio is too concentrated"""
metrics = portfolio.portfolio_metrics()
for ticker, allocation in portfolio.allocation.items():
if allocation > max_single_position:
return {
'risk': 'concentration',
'ticker': ticker,
'allocation': allocation,
'recommendation': 'reduce position size'
}
return {'risk': 'none', 'concentration': 'acceptable'}
def check_opportunity_cost(self, stock_price: float, annual_return: float,
expected_stock_return: float) -> Dict:
"""Compare covered call return to opportunity cost"""
covered_call_return = annual_return / stock_price
opportunity_cost = expected_stock_return - covered_call_return
return {
'covered_call_return': covered_call_return,
'expected_stock_return': expected_stock_return,
'opportunity_cost': opportunity_cost,
'recommended': 'consider_closing' if opportunity_cost > 0.10 else 'continue_position'
}
``