257 lines
10 KiB
Python
257 lines
10 KiB
Python
import matplotlib.pyplot as plt
|
|
import pandas as pd
|
|
import yfinance as yf
|
|
import pandas_ta as ta
|
|
import matplotlib.dates as mdates
|
|
from datetime import datetime, timedelta
|
|
import logging
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
def calculate_technical_indicators(data: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Calculates a suite of technical indicators using pandas_ta.
|
|
|
|
Args:
|
|
data (pd.DataFrame): DataFrame containing historical stock price data.
|
|
|
|
Returns:
|
|
pd.DataFrame: DataFrame with added technical indicators.
|
|
"""
|
|
try:
|
|
# Moving Averages
|
|
data.ta.macd(append=True)
|
|
data.ta.sma(length=20, append=True)
|
|
data.ta.ema(length=50, append=True)
|
|
|
|
# Momentum Indicators
|
|
data.ta.rsi(append=True)
|
|
data.ta.stoch(append=True)
|
|
|
|
# Volatility Indicators
|
|
data.ta.bbands(append=True)
|
|
data.ta.adx(append=True)
|
|
|
|
# Other Indicators
|
|
data.ta.obv(append=True)
|
|
data.ta.willr(append=True)
|
|
data.ta.cmf(append=True)
|
|
data.ta.psar(append=True)
|
|
|
|
# Custom Calculations
|
|
data['OBV_in_million'] = data['OBV'] / 1e6
|
|
data['MACD_histogram_12_26_9'] = data['MACDh_12_26_9']
|
|
|
|
logging.info("Technical indicators calculated successfully.")
|
|
return data
|
|
except KeyError as e:
|
|
logging.error(f"Missing key in data: {e}")
|
|
except ValueError as e:
|
|
logging.error(f"Value error: {e}")
|
|
except Exception as e:
|
|
logging.error(f"Error during technical indicator calculation: {e}")
|
|
return None
|
|
|
|
def get_last_day_summary(data: pd.DataFrame) -> pd.Series:
|
|
"""
|
|
Extracts and summarizes technical indicators for the last trading day.
|
|
|
|
Args:
|
|
data (pd.DataFrame): DataFrame with calculated technical indicators.
|
|
|
|
Returns:
|
|
pd.Series: Summary of technical indicators for the last day.
|
|
"""
|
|
try:
|
|
last_day_summary = data.iloc[-1][[
|
|
'Adj Close', 'MACD_12_26_9', 'MACD_histogram_12_26_9', 'RSI_14',
|
|
'BBL_5_2.0', 'BBM_5_2.0', 'BBU_5_2.0', 'SMA_20', 'EMA_50',
|
|
'OBV_in_million', 'STOCHk_14_3_3', 'STOCHd_14_3_3', 'ADX_14',
|
|
'WILLR_14', 'CMF_20', 'PSARl_0.02_0.2', 'PSARs_0.02_0.2'
|
|
]]
|
|
logging.info("Last day summary extracted.")
|
|
return last_day_summary
|
|
except KeyError as e:
|
|
logging.error(f"Missing columns in data: {e}")
|
|
except Exception as e:
|
|
logging.error(f"Error extracting last day summary: {e}")
|
|
return None
|
|
|
|
def analyze_stock(ticker_symbol: str, start_date: datetime, end_date: datetime) -> pd.Series:
|
|
"""
|
|
Fetches stock data, calculates technical indicators, and provides a summary.
|
|
|
|
Args:
|
|
ticker_symbol (str): The stock symbol.
|
|
start_date (datetime): Start date for data retrieval.
|
|
end_date (datetime): End date for data retrieval.
|
|
|
|
Returns:
|
|
pd.Series: Summary of technical indicators for the last day.
|
|
"""
|
|
try:
|
|
# Fetch stock data
|
|
stock_data = yf.download(ticker_symbol, start=start_date, end=end_date)
|
|
logging.info(f"Stock data fetched for {ticker_symbol} from {start_date} to {end_date}")
|
|
|
|
# Calculate technical indicators
|
|
stock_data = calculate_technical_indicators(stock_data)
|
|
|
|
# Get last day summary
|
|
if stock_data is not None:
|
|
last_day_summary = get_last_day_summary(stock_data)
|
|
if last_day_summary is not None:
|
|
print("Summary of Technical Indicators for the Last Day:")
|
|
print(last_day_summary)
|
|
return last_day_summary
|
|
else:
|
|
logging.error("Stock data is None, unable to calculate indicators.")
|
|
except Exception as e:
|
|
logging.error(f"Error during analysis: {e}")
|
|
return None
|
|
|
|
def get_finance_data(symbol: str) -> pd.Series:
|
|
"""
|
|
Fetches financial data for a given stock symbol.
|
|
|
|
Args:
|
|
symbol (str): The stock symbol.
|
|
|
|
Returns:
|
|
pd.Series: Summary of technical indicators for the last day.
|
|
"""
|
|
end_date = datetime.today()
|
|
start_date = end_date - timedelta(days=120)
|
|
|
|
# Perform analysis
|
|
last_day_summary = analyze_stock(symbol, start_date, end_date)
|
|
return last_day_summary
|
|
|
|
def analyze_options_data(ticker: str, expiry_date: str) -> tuple:
|
|
"""
|
|
Analyzes option data for a given ticker and expiry date.
|
|
|
|
Args:
|
|
ticker (str): The stock ticker symbol.
|
|
expiry_date (str): The option expiry date.
|
|
|
|
Returns:
|
|
tuple: A tuple containing calculated metrics for call and put options.
|
|
"""
|
|
call_df = options.get_calls(ticker, expiry_date)
|
|
put_df = options.get_puts(ticker, expiry_date)
|
|
|
|
# Implied Volatility Analysis:
|
|
avg_call_iv = call_df["Implied Volatility"].str.rstrip("%").astype(float).mean()
|
|
avg_put_iv = put_df["Implied Volatility"].str.rstrip("%").astype(float).mean()
|
|
logging.info(f"Average Implied Volatility for Call Options: {avg_call_iv}%")
|
|
logging.info(f"Average Implied Volatility for Put Options: {avg_put_iv}%")
|
|
|
|
# Option Prices Analysis:
|
|
avg_call_last_price = call_df["Last Price"].mean()
|
|
avg_put_last_price = put_df["Last Price"].mean()
|
|
logging.info(f"Average Last Price for Call Options: {avg_call_last_price}")
|
|
logging.info(f"Average Last Price for Put Options: {avg_put_last_price}")
|
|
|
|
# Strike Price Analysis:
|
|
min_call_strike = call_df["Strike"].min()
|
|
max_call_strike = call_df["Strike"].max()
|
|
min_put_strike = put_df["Strike"].min()
|
|
max_put_strike = put_df["Strike"].max()
|
|
logging.info(f"Minimum Strike Price for Call Options: {min_call_strike}")
|
|
logging.info(f"Maximum Strike Price for Call Options: {max_call_strike}")
|
|
logging.info(f"Minimum Strike Price for Put Options: {min_put_strike}")
|
|
logging.info(f"Maximum Strike Price for Put Options: {max_put_strike}")
|
|
|
|
# Volume Analysis:
|
|
total_call_volume = call_df["Volume"].str.replace('-', '0').astype(float).sum()
|
|
total_put_volume = put_df["Volume"].str.replace('-', '0').astype(float).sum()
|
|
logging.info(f"Total Volume for Call Options: {total_call_volume}")
|
|
logging.info(f"Total Volume for Put Options: {total_put_volume}")
|
|
|
|
# Open Interest Analysis:
|
|
call_df['Open Interest'] = call_df['Open Interest'].str.replace('-', '0').astype(float)
|
|
put_df['Open Interest'] = put_df['Open Interest'].str.replace('-', '0').astype(float)
|
|
total_call_open_interest = call_df["Open Interest"].sum()
|
|
total_put_open_interest = put_df["Open Interest"].sum()
|
|
logging.info(f"Total Open Interest for Call Options: {total_call_open_interest}")
|
|
logging.info(f"Total Open Interest for Put Options: {total_put_open_interest}")
|
|
|
|
# Convert Implied Volatility to float
|
|
call_df['Implied Volatility'] = call_df['Implied Volatility'].str.replace('%', '').astype(float)
|
|
put_df['Implied Volatility'] = put_df['Implied Volatility'].str.replace('%', '').astype(float)
|
|
|
|
# Calculate Put-Call Ratio
|
|
put_call_ratio = total_put_volume / total_call_volume
|
|
logging.info(f"Put-Call Ratio: {put_call_ratio}")
|
|
|
|
# Calculate Implied Volatility Percentile
|
|
call_iv_percentile = (call_df['Implied Volatility'] > call_df['Implied Volatility'].mean()).mean() * 100
|
|
put_iv_percentile = (put_df['Implied Volatility'] > put_df['Implied Volatility'].mean()).mean() * 100
|
|
logging.info(f"Call Option Implied Volatility Percentile: {call_iv_percentile}")
|
|
logging.info(f"Put Option Implied Volatility Percentile: {put_iv_percentile}")
|
|
|
|
# Calculate Implied Volatility Skew
|
|
implied_vol_skew = call_df['Implied Volatility'].mean() - put_df['Implied Volatility'].mean()
|
|
logging.info(f"Implied Volatility Skew: {implied_vol_skew}")
|
|
|
|
# Determine market sentiment
|
|
is_bullish_sentiment = call_df['Implied Volatility'].mean() > put_df['Implied Volatility'].mean()
|
|
sentiment = "bullish" if is_bullish_sentiment else "bearish"
|
|
logging.info(f"The overall sentiment of {ticker} is {sentiment}.")
|
|
|
|
return (avg_call_iv, avg_put_iv, avg_call_last_price, avg_put_last_price,
|
|
min_call_strike, max_call_strike, min_put_strike, max_put_strike,
|
|
total_call_volume, total_put_volume, total_call_open_interest, total_put_open_interest,
|
|
put_call_ratio, call_iv_percentile, put_iv_percentile, implied_vol_skew, sentiment)
|
|
|
|
def get_fin_options_data(ticker: str) -> list:
|
|
"""
|
|
Fetches and analyzes options data for a given stock ticker.
|
|
|
|
Args:
|
|
ticker (str): The stock ticker symbol.
|
|
|
|
Returns:
|
|
list: A list of sentences summarizing the options data.
|
|
"""
|
|
current_price = round(stock_info.get_live_price(ticker), 3)
|
|
option_expiry_dates = options.get_expiration_dates(ticker)
|
|
nearest_expiry = option_expiry_dates[0]
|
|
|
|
results = analyze_options_data(ticker, nearest_expiry)
|
|
|
|
# Unpack the results tuple
|
|
(avg_call_iv, avg_put_iv, avg_call_last_price, avg_put_last_price,
|
|
min_call_strike, max_call_strike, min_put_strike, max_put_strike,
|
|
total_call_volume, total_put_volume, total_call_open_interest, total_put_open_interest,
|
|
put_call_ratio, call_iv_percentile, put_iv_percentile, implied_vol_skew, sentiment) = results
|
|
|
|
# Create a list of complete sentences with the results
|
|
results_sentences = [
|
|
f"Average Implied Volatility for Call Options: {avg_call_iv}%",
|
|
f"Average Implied Volatility for Put Options: {avg_put_iv}%",
|
|
f"Average Last Price for Call Options: {avg_call_last_price}",
|
|
f"Average Last Price for Put Options: {avg_put_last_price}",
|
|
f"Minimum Strike Price for Call Options: {min_call_strike}",
|
|
f"Maximum Strike Price for Call Options: {max_call_strike}",
|
|
f"Minimum Strike Price for Put Options: {min_put_strike}",
|
|
f"Maximum Strike Price for Put Options: {max_put_strike}",
|
|
f"Total Volume for Call Options: {total_call_volume}",
|
|
f"Total Volume for Put Options: {total_put_volume}",
|
|
f"Total Open Interest for Call Options: {total_call_open_interest}",
|
|
f"Total Open Interest for Put Options: {total_put_open_interest}",
|
|
f"Put-Call Ratio: {put_call_ratio}",
|
|
f"Call Option Implied Volatility Percentile: {call_iv_percentile}",
|
|
f"Put Option Implied Volatility Percentile: {put_iv_percentile}",
|
|
f"Implied Volatility Skew: {implied_vol_skew}",
|
|
f"The overall sentiment of {ticker} is {sentiment}."
|
|
]
|
|
|
|
# Print each sentence
|
|
for sentence in results_sentences:
|
|
logging.info(sentence)
|
|
|
|
return results_sentences
|