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
The strategy limits upside while generating income from call premium. Success depends on strike selection and assignment probability.
Covered Call Selection Algorithm
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
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
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
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'
}
Conclusion
Covered calls provide steady income from stock holdings but cap upside. Success requires disciplined strike selection, regular rolling, and appropriate position sizing. The strategy works best for stocks in stable ranges or when you're comfortable with assignment.