Skip to content

Instantly share code, notes, and snippets.

@ninenine

ninenine/black_scholes.py

Created February 29, 2024 19:04
Show Gist options
  • Save ninenine/b02bb254f11d256a161c70274dd9b4a3 to your computer and use it in GitHub Desktop.
Save ninenine/b02bb254f11d256a161c70274dd9b4a3 to your computer and use it in GitHub Desktop.
Download ZIP
Option Pricing with the Black–Scholes Model and Python
Raw
black_scholes.py
import asyncio
import logging
from typing import Literal
import yfinance as yf
import numpy as np
from scipy.stats import norm
import coloredlogs
from concurrent.futures import ThreadPoolExecutor
# Define a custom logging format
log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
# Configure coloredlogs for better logging visualization
coloredlogs.install(
level="INFO",
fmt=log_format,
datefmt="%Y-%m-%d %H:%M:%S %Z", # Date format including timezone
# Customize the log level styles
level_styles={
"debug": {"color": "green"},
"info": {"color": "cyan"},
"warning": {"color": "yellow"},
"error": {"color": "red"},
"critical": {"color": "red", "bold": True},
},
# Customize the field styles
field_styles={
"asctime": {"color": "green"},
"hostname": {"color": "magenta"},
"levelname": {"color": "black", "bold": True},
"name": {"color": "blue"},
"programname": {"color": "cyan"},
},
)
# Create a ThreadPoolExecutor for running blocking operations in separate threads
executor = ThreadPoolExecutor(max_workers=10)
# Market Data Fetcher class for fetching stock prices
class MarketDataFetcher:
def __init__(self, ticker: str):
self.ticker = ticker
# Asynchronous method to get stock price
async def get_stock_price(self) -> float:
loop = asyncio.get_running_loop()
# Run the blocking _fetch_price method in a separate thread
return await loop.run_in_executor(executor, self._fetch_price)
# Method to fetch stock price using yfinance
def _fetch_price(self) -> float:
try:
stock = yf.Ticker(self.ticker)
# Fetch the stock price for the last 5 days
todays_data = stock.history(period="5d")
return float(todays_data["Close"].iloc[-1])
except Exception as e:
logging.error(f"🚨 Error fetching stock price for {self.ticker}: {e}")
raise
# Black-Scholes Calculator class for calculating option prices
class BlackScholesCalculator:
def __init__(self, S: float, K: float, T: float, r: float, sigma: float):
"""
Calculate the Black-Scholes option price.
Parameters:
- S: Current stock price
- K: Strike price
- T: Time to expiration in years
- r: Risk-free interest rate
- sigma: Volatility of the stock
"""
self.S = S
self.K = K
self.T = T
self.r = r
self.sigma = sigma
# Method to calculate option price based on Black-Scholes formula
def calculate_option_price(self, option_type: Literal["call", "put"]) -> float:
try:
# Calculate d1 and d2
d1 = (
np.log(self.S / self.K) + (self.r + 0.5 * self.sigma**2) * self.T
) / (self.sigma * np.sqrt(self.T))
d2 = d1 - self.sigma * np.sqrt(self.T)
if option_type == "call":
# Calculate and return the call option price
return self.S * norm.cdf(d1) - self.K * np.exp(
-self.r * self.T
) * norm.cdf(d2)
elif option_type == "put":
# Calculate and return the put option price
return self.K * np.exp(-self.r * self.T) * norm.cdf(
-d2
) - self.S * norm.cdf(-d1)
else:
raise ValueError("🚨 Invalid option type. Use 'call' or 'put'.")
except Exception as e:
logging.error(f"🚨 Error calculating option price: {e}")
raise
# Main function to fetch stock price and calculate option prices
async def main(ticker: str, K: float, T: float, r: float, sigma: float) -> None:
try:
logging.info(f"Fetching stock price for {ticker}...")
fetcher = MarketDataFetcher(ticker)
S = await fetcher.get_stock_price()
logging.info(f"Calculating option prices for {ticker}...")
calculator = BlackScholesCalculator(S, K, T, r, sigma)
call_price = calculator.calculate_option_price("call")
put_price = calculator.calculate_option_price("put")
# Log the results
logging.info(f"Stock Price for {ticker}: {S:.2f}")
logging.info(f"Call Option Price: {call_price:.2f}")
logging.info(f"Put Option Price: {put_price:.2f}")
logging.info("=" * 50)
except Exception as e:
logging.error(f"🚨 Error in main function: {e}")
raise
# Function to run the main function for all tickers
async def run_all(tickers):
tasks = [main(ticker, K, T, r, sigma) for ticker, K, T, r, sigma in tickers]
await asyncio.gather(*tasks)
if __name__ == "__main__":
# List of tickers with their respective parameters
tickers = [
("AAPL", 150, 1, 0.05, 0.2), # Apple Inc. - Technology
("MSFT", 300, 1, 0.05, 0.2), # Microsoft Corp. - Technology
("AMZN", 3500, 1, 0.05, 0.3), # Amazon.com Inc. - Consumer Retail
("GOOGL", 2500, 1, 0.05, 0.25), # Alphabet Inc. (Google) - Technology
("META", 350, 1, 0.05, 0.25), # Meta Platforms Inc. (Facebook) - Technology
("TSLA", 700, 0.5, 0.03, 0.3), # Tesla Inc. - Automotive & Energy
("JNJ", 165, 1, 0.05, 0.2), # Johnson & Johnson - Healthcare
("JPM", 125, 1, 0.05, 0.2), # JPMorgan Chase & Co. - Finance
("V", 220, 1, 0.05, 0.2), # Visa Inc. - Financial Services
("PG", 140, 1, 0.05, 0.2), # Procter & Gamble Co. - Consumer Goods
("UNH", 500, 1, 0.05, 0.2), # UnitedHealth Group Inc. - Healthcare
("MA", 360, 1, 0.05, 0.2), # Mastercard Inc. - Financial Services
("NVDA", 200, 1, 0.05, 0.25), # NVIDIA Corp. - Semiconductors
("HD", 330, 1, 0.05, 0.2), # Home Depot Inc. - Retail
("BABA", 220, 1, 0.05, 0.3), # Alibaba Group Holding Ltd - Technology & Retail
("XOM", 90, 1, 0.05, 0.2), # Exxon Mobil Corp. - Energy
("WMT", 140, 1, 0.05, 0.2), # Walmart Inc. - Retail
("BP", 36, 1, 0.05, 0.25), # BP plc - Energy
("T", 28, 1, 0.05, 0.2), # AT&T Inc. - Telecommunications
("VZ", 58, 1, 0.05, 0.2), # Verizon Communications Inc. - Telecommunications
("PFE", 42, 1, 0.05, 0.2), # Pfizer Inc. - Healthcare
("NFLX", 600, 1, 0.05, 0.3), # Netflix Inc. - Entertainment
("DIS", 130, 1, 0.05, 0.25), # Walt Disney Co. - Entertainment
]
# Run the main function for all tickers
asyncio.run(run_all(tickers))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment