In The Money Covered Call Option Scanner
I was here posting a lot more frequently last month about my credit spread scanner which is pretty much done now if you want to check it out: https://spreadfinder.com/index
https://www.reddit.com/r/ClaudeAI/comments/1eb0xbq/one_month_of_coding_with_claude/
And recently I was browsing around reddit and youtube and learned about ITM covered calls which seemed interesting, so I figured I'd also make another tool to help assist in the discovery process of these trades!
You can check out the new tool here: https://spreadfinder.com/cc
Learn about in the money covered calls at the below links:
https://www.investopedia.com/articles/optioninvestor/06/inthemoneycallwrite.asp
https://www.youtube.com/watch?v=6dUzuGZTUZU
https://www.reddit.com/r/thetagang/comments/iknbv5/thanks_theta_gang_52_in_7_months_from_selling/
This was a lot easier to finish than the credit spread finder. Probably only took me 2-3 days. Some of the work I've already completed for the credit spread finder in terms of having a ready to go database of all earnings related information on all companies definitely helped.
The credit spread scanner was probably around 10k lines of code and took me about a month and a half of no sleep to complete. But I guess I'm also just better at coding now, which is great. I've also made loads of little improvements to the credit spread finder and continue to make adjustments when I have new ideas, so definitely have a look if you haven't seen it in a while.
The profit margins tend to be really thin for this method, and doing manual calculations is kind of annoying, so perhaps this tool will help save some people some time!
Enjoy!
Here's my code!
import math
import time
from datetime import datetime, timedelta
from scipy.stats import norm
from services.market_data import fetch_stock_price, fetch_options_chain, fetch_expiration_dates
import logging
from functools import lru_cache
import argparse
import sys
from contextlib import contextmanager
from services.earnings_data import get_tickers_with_upcoming_earnings_for_period
import sqlite3
from config import Config
from functools import lru_cache
import json
import os
from tabulate import tabulate as tabulate_func
from services.safety_score import calculate_safety_score
DB_NAME = Config.EARNINGS_DB_PATH
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def get_db_connection():
conn = sqlite3.connect(DB_NAME, timeout=20)
conn.execute('PRAGMA journal_mode=WAL')
return conn
CACHE_DIR = "cache"
CACHE_DURATION = timedelta(hours=72) # Cache data for 4 hours
def cache_key(func, *args, **kwargs):
return f"{func.__name__}_{args}_{kwargs}"
@lru_cache(maxsize=100)
def cached_fetch_stock_price(ticker):
key = cache_key(cached_fetch_stock_price, ticker)
cached_result = read_cache(key)
if cached_result is not None:
return cached_result
result = fetch_stock_price(ticker)
write_cache(key, result)
return result
@lru_cache(maxsize=100)
def cached_fetch_options_chain(ticker, expiration_date):
key = cache_key(cached_fetch_options_chain, ticker, expiration_date)
cached_result = read_cache(key)
if cached_result is not None:
return cached_result
result = fetch_options_chain(ticker, expiration_date)
write_cache(key, result)
return result
@lru_cache(maxsize=1000)
def cached_get_next_earnings_date(ticker):
key = cache_key(cached_get_next_earnings_date, ticker)
cached_result = read_cache(key)
if cached_result is not None:
return cached_result
result = get_next_earnings_date(ticker)
write_cache(key, result)
return result
def read_cache(key):
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
if os.path.exists(cache_file):
with open(cache_file, 'r') as f:
data = json.load(f)
if datetime.now() - datetime.fromisoformat(data['timestamp']) < CACHE_DURATION:
return data['value']
return None
def write_cache(key, value):
os.makedirs(CACHE_DIR, exist_ok=True)
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
with open(cache_file, 'w') as f:
json.dump({'timestamp': datetime.now().isoformat(), 'value': value}, f)
@contextmanager
def db_connection():
conn = get_db_connection()
try:
yield conn
finally:
conn.close()
def get_upcoming_er_tickers(days=7):
today = datetime.now().date()
end_date = today + timedelta(days=days)
with db_connection() as conn:
c = conn.cursor()
c.execute('''
SELECT DISTINCT ticker FROM earnings
WHERE earnings_date IS NOT NULL
AND date(earnings_date) BETWEEN ? AND ?
AND is_historical = 0
ORDER BY date(earnings_date)
''', (today.isoformat(), end_date.isoformat()))
tickers = [row[0] for row in c.fetchall()]
return tickers
def get_next_earnings_date(ticker):
max_retries = 3
retry_delay = 0.1
for attempt in range(max_retries):
try:
with db_connection() as conn:
c = conn.cursor()
c.execute('''
SELECT earnings_date, report_time FROM earnings
WHERE ticker = ? AND earnings_date >= date('now')
ORDER BY earnings_date ASC
LIMIT 1
''', (ticker,))
result = c.fetchone()
if result:
er_date = datetime.strptime(result[0], '%Y-%m-%d')
er_time = 'BMO' if result[1] == 'before open' else 'AMC'
return f"ER: {er_date.strftime('%m/%d/%y')} - {er_time}"
return "No ER data"
except sqlite3.OperationalError as e:
if "database is locked" in str(e) and attempt < max_retries - 1:
time.sleep(retry_delay * (2 ** attempt)) # Exponential backoff
else:
logging.error(f"Error fetching earnings date for {ticker}: {str(e)}")
return "Error fetching ER data"
except Exception as e:
logging.error(f"Unexpected error fetching earnings date for {ticker}: {str(e)}")
return "Error fetching ER data"
def calculate_ev(stock_price, strike, premium, iv, tte):
std_dev = stock_price * iv * math.sqrt(tte)
price_points = [
max(0, stock_price - 2*std_dev),
max(0, stock_price - std_dev),
stock_price,
stock_price + std_dev,
stock_price + 2*std_dev
]
probabilities = [
norm.cdf(-2),
norm.cdf(-1) - norm.cdf(-2),
norm.cdf(1) - norm.cdf(-1),
norm.cdf(2) - norm.cdf(1),
1 - norm.cdf(2)
]
payoffs = []
for price in price_points:
if price >= strike:
payoff = (strike - stock_price + premium) * 100 # Stock called away
else:
payoff = (price - stock_price + premium) * 100 # Stock not called away
payoffs.append(payoff)
ev = sum(payoff * prob for payoff, prob in zip(payoffs, probabilities))
return ev
def calculate_aror(ror, days_to_expiration):
"""Calculate the Annualized Rate of Return (AROR)."""
if days_to_expiration == 0:
return float('inf') # Avoid division by zero
trading_days_per_year = 252 # Approximate number of trading days in a year
times_per_year = trading_days_per_year / days_to_expiration
aror = ((1 + ror/100) ** times_per_year - 1) * 100
return aror
def get_margin_rate(borrowed_amount):
if borrowed_amount <= 50000:
return 0.0675
elif borrowed_amount <= 100000:
return 0.0655
elif borrowed_amount <= 1000000:
return 0.0625
elif borrowed_amount <= 10000000:
return 0.0600
elif borrowed_amount <= 50000000:
return 0.0595
else:
return 0.0570
def calculate_composite_score(stock_price, short_strike, max_pain_strike, side, transformed_safety_score, debug=False):
if max_pain_strike is None or stock_price is None or transformed_safety_score is None:
return None
dist_to_max_pain = (stock_price - max_pain_strike) / max_pain_strike * 100
dist_of_short_strike_to_max_pain = (short_strike - max_pain_strike) / max_pain_strike * 100
if side.lower() == 'call':
risk_factor = (dist_to_max_pain - dist_of_short_strike_to_max_pain) * -1
elif side.lower() == 'put':
risk_factor = (dist_of_short_strike_to_max_pain - dist_to_max_pain) * -1
else:
return None
composite_score = -risk_factor
final_score = composite_score * transformed_safety_score
if debug:
print(f"\nComposite Score Calculation Debug:")
print(f"Stock Price: ${stock_price:.2f}")
print(f"Short Strike: ${short_strike:.2f}")
print(f"Max Pain Strike: ${max_pain_strike:.2f}")
print(f"Side: {side}")
print(f"Transformed Safety Score: {transformed_safety_score:.4f}")
print(f"Distance to Max Pain: {dist_to_max_pain:.2f}%")
print(f"Distance of Short Strike to Max Pain: {dist_of_short_strike_to_max_pain:.2f}%")
print(f"Risk Factor: {risk_factor:.2f}")
print(f"Composite Score (before safety score multiplication): {composite_score:.2f}")
print(f"Final Score (after safety score multiplication): {final_score:.2f}")
return final_score
def calculate_max_pain(options_chain):
strikes = sorted(set(float(strike) for strike in options_chain['Strike']))
max_pain = None
min_total_loss = float('inf')
for strike in strikes:
total_loss = 0
for i, option_strike in enumerate(options_chain['Strike']):
option_strike = float(option_strike)
option_type = options_chain['Option Side'][i]
open_interest = float(options_chain['Open Interest'][i])
if option_type.lower() == 'call':
if strike > option_strike:
total_loss += (strike - option_strike) * open_interest
elif option_type.lower() == 'put':
if strike < option_strike:
total_loss += (option_strike - strike) * open_interest
if total_loss < min_total_loss:
min_total_loss = total_loss
max_pain = strike
return max_pain
def calculate_covered_call_opportunities(tickers, expiration_date, min_ror=1, max_ror=None, min_pop=50, min_ev=0, moneyness='both', trades_per_ticker=1, max_iv=None, min_safety_score=None, min_market_cap=None):
best_opportunities = {ticker: [] for ticker in tickers}
total_trades_considered = 0
filtered_reasons_summary = {}
for ticker in tickers:
stock_price, _ = fetch_stock_price(ticker, use_cached=False)
if stock_price is None:
logging.warning(f"Could not fetch stock price for {ticker}")
continue
options_result = cached_fetch_options_chain(ticker, expiration_date)
if not options_result or len(options_result) < 2:
logging.warning(f"Could not fetch options chain for {ticker}")
continue
options_chain, headers = options_result[:2]
max_pain_strike = calculate_max_pain(options_chain)
safety_data = calculate_safety_score(ticker)
transformed_safety_score = safety_data['transformed_score'] if safety_data else None
market_cap = safety_data.get('market_cap') if safety_data else None
if min_market_cap is not None and (market_cap is None or market_cap < min_market_cap):
continue
if not isinstance(options_chain, dict):
logging.warning(f"Options chain for {ticker} is not a dictionary. Type: {type(options_chain)}")
continue
if 'Symbol' not in options_chain or 'Option Side' not in options_chain or 'Strike' not in options_chain or 'Bid' not in options_chain or 'IV' not in options_chain:
logging.warning(f"Missing required keys in options chain for {ticker}")
continue
current_date = datetime.now().date()
expiration_datetime = datetime.strptime(expiration_date, '%Y-%m-%d').date()
days_to_expiration = (expiration_datetime - current_date).days
if current_date == expiration_datetime:
tte = 1 / 1440 # Use 1 minute as the minimum time to expiration
else:
tte = days_to_expiration / 252 # Use trading days instead of calendar days
try:
atm_option = min(options_chain['Strike'], key=lambda x: abs(float(x) - stock_price))
atm_index = options_chain['Strike'].index(atm_option)
iv = float(options_chain['IV'][atm_index])
logging.info(f"ATM IV for {ticker}: {iv*100}% (Strike: {atm_option})")
if max_iv is not None and iv * 100 > max_iv:
logging.info(f"Skipping {ticker} due to high IV: {iv*100}% > {max_iv}%")
continue
except Exception as e:
logging.error(f"Error calculating ATM IV for {ticker}: {str(e)}")
continue
if iv <= 0 or tte <= 0:
logging.warning(f"Invalid IV ({iv*100}%) or TTE ({tte}) for {ticker}")
continue
for i, option in enumerate(zip(options_chain['Symbol'], options_chain['Option Side'], options_chain['Strike'], options_chain['Bid'])):
symbol, side, strike_str, bid_str = option
if side.lower() != 'call':
continue
strike = float(strike_str)
if strike >= stock_price:
continue
max_retries = 3
retry_delay = 0.1
for attempt in range(max_retries):
try:
strike = float(strike_str)
premium = float(bid_str)
if moneyness == 'ITM' and strike >= stock_price:
filtered_reasons_summary["Wrong moneyness (ITM required)"] = filtered_reasons_summary.get("Wrong moneyness (ITM required)", 0) + 1
continue
elif moneyness == 'OTM' and strike <= stock_price:
filtered_reasons_summary["Wrong moneyness (OTM required)"] = filtered_reasons_summary.get("Wrong moneyness (OTM required)", 0) + 1
continue
total_trades_considered += 1
max_profit = (strike - stock_price) + premium
max_loss = stock_price - premium
margin_rate = get_margin_rate(stock_price * 100)
annual_margin_interest = stock_price * margin_rate
daily_margin_interest = annual_margin_interest / 365
total_margin_interest = daily_margin_interest * days_to_expiration
max_profit -= total_margin_interest
max_loss += total_margin_interest
if max_profit <= 0:
filtered_reasons_summary["Negative or zero max profit"] = filtered_reasons_summary.get("Negative or zero max profit", 0) + 1
continue
ror = (max_profit / max_loss) * 100
aror = calculate_aror(ror, days_to_expiration)
break_even = strike + premium
pop = norm.cdf((math.log(stock_price/strike) + (0.5 * iv**2) * tte) / (iv * math.sqrt(tte))) * 100
ev = calculate_ev(stock_price, strike, premium, iv, tte)
composite_score = calculate_composite_score(stock_price, strike, max_pain_strike, "call", transformed_safety_score)
if composite_score is None or (min_safety_score is not None and composite_score < min_safety_score):
filtered_reasons_summary["Low composite score"] = filtered_reasons_summary.get("Low composite score", 0) + 1
continue
if (min_ror is None or ror >= min_ror) and \
(max_ror is None or ror <= max_ror) and \
(min_pop is None or pop >= min_pop) and \
(min_ev is None or ev >= min_ev):
opportunity = {
'ticker': ticker,
'stock_price': stock_price,
'side': "CC",
'short_strike': strike,
'long_strike': "",
'premium': premium,
'max_profit': max_profit,
'max_loss': max_loss,
'ror': ror,
'market_cap': market_cap,
'aror': aror,
'pop': pop,
'ev': ev / 100,
'expiration_date': expiration_date,
'days_to_expiration': days_to_expiration,
'itm_otm': 'OTM' if strike > stock_price else 'ITM',
'atm_iv': iv * 100,
'er_info': cached_get_next_earnings_date(ticker),
'margin_interest': total_margin_interest,
'composite_score': composite_score,
'transformed_safety_score': transformed_safety_score,
'max_pain_strike': max_pain_strike
}
best_opportunities[ticker].append(opportunity)
best_opportunities[ticker] = sorted(best_opportunities[ticker], key=lambda x: x['ev'], reverse=True)[:trades_per_ticker]
else:
filtered_reasons_summary["Did not meet ROR, POP, or EV criteria"] = filtered_reasons_summary.get("Did not meet ROR, POP, or EV criteria", 0) + 1
break # If successful, break out of the retry loop
except sqlite3.OperationalError as e:
if "database is locked" in str(e) and attempt < max_retries - 1:
time.sleep(retry_delay * (2 ** attempt)) # Exponential backoff
else:
logging.error(f"Unexpected error processing option for {ticker} at index {i}: {str(e)}")
break
except Exception as e:
logging.error(f"Unexpected error processing option for {ticker} at index {i}: {str(e)}")
break
# Flatten the list of opportunities
all_opportunities = [opp for ticker_opps in best_opportunities.values() for opp in ticker_opps]
sorted_opportunities = sorted(all_opportunities, key=lambda x: x['ev'], reverse=True)
logging.info(f"\nTotal trades considered: {total_trades_considered}")
logging.info(f"Best opportunities found: {len(sorted_opportunities)}")
return sorted_opportunities, total_trades_considered, filtered_reasons_summary
@lru_cache(maxsize=100)
def get_expiration_dates(ticker):
"""Fetch and return expiration dates for a given ticker."""
key = cache_key(get_expiration_dates, ticker)
cached_result = read_cache(key)
if cached_result is not None:
return cached_result
try:
dates = fetch_expiration_dates(ticker)
result = [date for date in dates if datetime.strptime(date, '%Y-%m-%d').date() >= datetime.now().date()]
write_cache(key, result)
return result
except Exception as e:
logging.error(f"Error fetching expiration dates for {ticker}: {str(e)}")
return []
def analyze_all_expirations(ticker, min_ror=1, max_ror=None, min_pop=50, max_pop=None, min_ev=0, moneyness='both', max_iv=None):
"""Analyze trades for all available expiration dates for a given ticker."""
expiration_dates = get_expiration_dates(ticker)
best_trades = []
for expiration_date in expiration_dates:
opportunities, _ = calculate_covered_call_opportunities(
[ticker], expiration_date, min_ror, max_ror, min_pop, max_pop, min_ev, moneyness, max_iv=max_iv
)
if opportunities:
best_trade = max(opportunities, key=lambda x: x['ev'])
best_trades.append(best_trade)
return best_trades
@lru_cache(maxsize=100)
def cached_analyze_all_expirations(ticker, min_ror, max_ror, min_pop, max_pop, min_ev, moneyness, max_iv, min_safety_score, min_market_cap):
"""Cached version of analyze_all_expirations."""
return analyze_all_expirations(ticker, min_ror, max_ror, min_pop, max_pop, min_ev, moneyness, max_iv, min_safety_score, min_market_cap)
def print_best_trades(tickers, min_ror=1, max_ror=None, min_pop=50, max_pop=None, min_ev=0, moneyness='both', expiration_date=None, trades_per_ticker=1, max_iv=None, max_results=None, min_safety_score=None, min_market_cap=None, debug=False):
"""Print the best trades for each ticker and expiration date."""
all_best_trades = []
if expiration_date:
# Single expiration date mode
opportunities, total_trades, filtered_reasons_summary = calculate_covered_call_opportunities(
tickers, expiration_date, min_ror, max_ror, min_pop, min_ev, moneyness, trades_per_ticker, max_iv, min_safety_score, min_market_cap
)
all_best_trades = opportunities
else:
# Multi-date mode
for ticker in tickers:
best_trades = cached_analyze_all_expirations(ticker, min_ror, max_ror, min_pop, max_pop, min_ev, moneyness, max_iv, min_safety_score, min_market_cap)
all_best_trades.extend(best_trades[:trades_per_ticker]) # Only take the top N trades per ticker
# Sort all trades by EV
all_best_trades.sort(key=lambda x: x['ev'], reverse=True)
# Limit the number of results if max_results is specified
if max_results is not None:
all_best_trades = all_best_trades[:max_results]
# Prepare data for tabulate
table_data = []
for i, trade in enumerate(all_best_trades):
if trade['er_info'] != "No ER data":
er_date_str = trade['er_info'].split(': ')[1].split(' - ')[0]
er_date = datetime.strptime(er_date_str, '%m/%d/%y')
days_to_er = (er_date.date() - datetime.now().date()).days
er_info = f"{trade['er_info']} ({days_to_er} days)"
else:
er_info = "-"
# Recalculate composite score with debug output for the top trade
if i == 0 and debug:
composite_score = calculate_composite_score(
trade['stock_price'],
trade['short_strike'],
trade['max_pain_strike'],
trade['side'],
trade['transformed_safety_score'],
debug=True
)
else:
composite_score = trade['composite_score']
# Format market cap
market_cap = trade['market_cap']
if market_cap is not None:
if market_cap >= 1e12:
market_cap_str = f"${market_cap/1e12:.2f}T"
elif market_cap >= 1e9:
market_cap_str = f"${market_cap/1e9:.2f}B"
elif market_cap >= 1e6:
market_cap_str = f"${market_cap/1e6:.2f}M"
else:
market_cap_str = f"${market_cap:.2f}"
else:
market_cap_str = "N/A"
table_data.append([
trade['ticker'],
f"${trade['stock_price']:.2f}",
f"${trade['short_strike']:.2f}",
er_info,
f"${trade['premium']:.2f}",
f"${trade['max_profit']:.2f}",
f"${trade['max_loss']:.2f}",
f"{trade['ror']:.1f}%",
f"{round(trade['aror'])}%",
f"{trade['pop']:.2f}%",
f"${trade['ev']:.2f}",
f"{round(trade['atm_iv'])}%",
f"{composite_score:.1f}" if composite_score is not None else "N/A",
f"${trade['max_pain_strike']:.1f}" if trade['max_pain_strike'] is not None else "N/A",
market_cap_str
])
# Print the expiration date
if expiration_date:
formatted_exp_date = datetime.strptime(expiration_date, '%Y-%m-%d').strftime('%m/%d/%y')
print(f"\nExpiration Date: {formatted_exp_date}")
# Print the table
headers = ["Ticker", "Stock", "Strike", "ER Info", "Credit", "Profit", "Risk", "ROR", "AROR", "POP", "EV", "IV", "Safety", "Max Pain", "Market Cap"]
print("\nBest Trades:")
print(tabulate_func(table_data, headers=headers, tablefmt="grid"))
return all_best_trades, filtered_reasons_summary
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Calculate covered call opportunities")
parser.add_argument("--use_upcoming_er", action="store_true", help="Use tickers with upcoming earnings reports")
parser.add_argument("--expiration_date", type=str, help="Expiration date (YYYY-MM-DD)")
parser.add_argument("--min_ror", type=float, help="Minimum Rate of Return")
parser.add_argument("--max_ror", type=float, help="Maximum Rate of Return")
parser.add_argument("--min_pop", type=float, help="Minimum Probability of Profit")
parser.add_argument("--max_pop", type=float, help="Maximum Probability of Profit")
parser.add_argument("--min_ev", type=float, help="Minimum Expected Value")
parser.add_argument("--moneyness", choices=['ITM', 'OTM', 'both'], help="Option moneyness")
parser.add_argument("--trades_per_ticker", type=int, help="Number of trades to show per ticker")
parser.add_argument("--er_days", type=int, help="Number of days to look ahead for earnings reports")
parser.add_argument("--max_iv", type=float, help="Maximum IV allowed")
parser.add_argument("--max_results", type=int, help="Maximum number of results to display")
parser.add_argument("--min_safety_score", type=float, help="Minimum safety score")
parser.add_argument("--min_market_cap", type=float, help="Minimum market cap in billions")
parser.add_argument("--debug", action="store_true", help="Enable debug output")
args = parser.parse_args()
# Set default values
moneyness = 'both' # Set your desired default moneyness
expiration_date = "2024-08-23" # Set your desired default expiration date here
er_days = 14 # Set your desired default number of days to look ahead for earnings reports
max_iv = None # Set your desired default maximum IV (as a percentage)
max_results = None # Default to showing all results
min_ror = 1 # Set your desired default minimum Rate of Return
max_ror = None # Set your desired default maximum Rate of Return
min_pop = 20 # Set your desired default minimum Probability of Profit
max_pop = None # Set your desired default maximum Probability of Profit
min_ev = None # Set your desired default minimum Expected Value
trades_per_ticker = 20 # Set your desired default number of trades per ticker
min_safety_score = args.min_safety_score or None # Add this line
min_market_cap = args.min_market_cap * 1e9 if args.min_market_cap else None # Convert to actual value
# Define your list of tickers here
tickers = ["NVDL"] # Add or remove tickers as needed
# Override defaults with command-line arguments if provided
if args.expiration_date:
expiration_date = args.expiration_date
if args.er_days:
er_days = args.er_days
if args.max_iv:
max_iv = args.max_iv
if args.max_results:
max_results = args.max_results
if args.min_ror:
min_ror = args.min_ror
if args.max_ror:
max_ror = args.max_ror
if args.min_pop:
min_pop = args.min_pop
if args.max_pop:
max_pop = args.max_pop
if args.min_ev:
min_ev = args.min_ev
if args.moneyness:
moneyness = args.moneyness
if args.trades_per_ticker:
trades_per_ticker = args.trades_per_ticker
if args.use_upcoming_er:
tickers = get_upcoming_er_tickers(days=er_days)
print(f"Analyzing {len(tickers)} tickers with upcoming earnings reports in the next {er_days} days")
else:
print(f"Analyzing {len(tickers)} predefined tickers")
best_trades, filtered_reasons = print_best_trades(
tickers,
min_ror,
max_ror,
min_pop,
max_pop,
min_ev,
moneyness,
expiration_date,
trades_per_ticker,
max_iv,
max_results,
min_safety_score,
min_market_cap,
args.debug
)