kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AI finance TA writer, yfinance, pandas_ta, WIP

Browse Source
This commit is contained in:
ajaysi
2024-05-15 15:53:15 +05:30
parent 417183a6d2
commit 26a35ee355
5 changed files with 457 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
alwrity.py
View File

@@ -17,7 +17,7 @@ print("Loading, required libraries..")
app = typer.Typer()
from lib.utils.alwrity_utils import blog_from_audio, blog_from_keyword, do_web_research, do_web_research, ai_news_writer
from lib.utils.alwrity_utils import write_story, essay_writer, blog_tools
from lib.utils.alwrity_utils import write_story, essay_writer, blog_tools, ai_finance_ta_writer
from lib.utils.alwrity_utils import content_planning, competitor_analysis, image_to_text_writer, image_generator
@@ -37,9 +37,9 @@ def write_blog_options():
("AI Story Writer", "AI Story Writer"),
("AI Essay Writer", "AI Essay writer"),
("AI News Articles", "News - AI News article writer, factual trusted sources"),
("AI Finance TA report", "AI TA report - Write stocks Techincal Analysis report."),
("Programming", "Programming - Write technical blogs on latest topics"),
("Scholar", "Scholar - Research Reports from google scholar, arxiv articles."),
("Finance/TBD", "Finance/TBD"),
("Quit", "Quit")
]
selected_blog_type = radiolist_dialog(title="Choose a blog type:", values=choices).run()
@@ -62,7 +62,7 @@ def start_interactive_mode():
("Online Blog Tools/Apps", "Online AI Apps - Content & Digital marketing"),
("Create Blog Images", "Create Images - Stability, Dalle3"),
("AI Social Media(TBD)", "AI Social Media(TBD)"),
("AI Code Writer(TBD)", "AI Code Writer(TBD)"),
("AI CopyWriter(TBD)", "AI CopyWriter(TBD)"),
("Quit", "Quit")
]
mode = radiolist_dialog(title="Choose an option:", values=choices).run()
@@ -178,6 +178,8 @@ def write_blog():
blog_from_audio()
elif blog_type == 'AI News Articles':
ai_news_writer()
elif blog_type == 'AI Finance TA report':
ai_finance_ta_writer()
elif blog_type == 'GitHub':
github = prompt("Enter GitHub URL, CSV file, or topic:")
print(f"Write blog based on GitHub: {github}")

335
lib/ai_web_researcher/finance_data_researcher.py Normal file
View File

@@ -0,0 +1,335 @@
import matplotlib.pyplot as plt
import pandas as pd
import yfinance as yf
#from yahoo_fin import options, stock_info
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):
"""
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 Exception as e:
logging.error(f"Error during technical indicator calculation: {e}")
return None
def get_last_day_summary(data):
"""
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}")
return None
except Exception as e:
logging.error(f"Error extracting last day summary: {e}")
return None
def analyze_stock(ticker_symbol, start_date, end_date):
"""
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)
# Plot the technical indicators
plt.figure(figsize=(14, 8))
# Price Trend Chart
plt.subplot(3, 3, 1)
plt.plot(stock_data.index, stock_data['Adj Close'], label='Adj Close', color='blue')
plt.plot(stock_data.index, stock_data['EMA_50'], label='EMA 50', color='green')
plt.plot(stock_data.index, stock_data['SMA_20'], label='SMA_20', color='orange')
plt.title("Price Trend")
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
plt.legend()
plt.show()
# On-Balance Volume Chart
plt.subplot(3, 3, 2)
plt.plot(stock_data['OBV'], label='On-Balance Volume')
plt.title('On-Balance Volume (OBV) Indicator')
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
plt.legend()
# MACD Plot
plt.subplot(3, 3, 3)
plt.plot(stock_data['MACD_12_26_9'], label='MACD')
plt.plot(stock_data['MACDh_12_26_9'], label='MACD Histogram')
plt.title('MACD Indicator')
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
plt.title("MACD")
plt.legend()
# RSI Plot
plt.subplot(3, 3, 4)
plt.plot(stock_data['RSI_14'], label='RSI')
plt.axhline(y=70, color='r', linestyle='--', label='Overbought (70)')
plt.axhline(y=30, color='g', linestyle='--', label='Oversold (30)')
plt.legend()
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
plt.title('RSI Indicator')
# Bollinger Bands Plot
plt.subplot(3, 3, 5)
plt.plot(stock_data.index, stock_data['BBU_5_2.0'], label='Upper BB')
plt.plot(stock_data.index, stock_data['BBM_5_2.0'], label='Middle BB')
plt.plot(stock_data.index, stock_data['BBL_5_2.0'], label='Lower BB')
plt.plot(stock_data.index, stock_data['Adj Close'], label='Adj Close', color='brown')
plt.title("Bollinger Bands")
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
plt.legend()
# Stochastic Oscillator Plot
plt.subplot(3, 3, 6)
plt.plot(stock_data.index, stock_data['STOCHk_14_3_3'], label='Stoch %K')
plt.plot(stock_data.index, stock_data['STOCHd_14_3_3'], label='Stoch %D')
plt.title("Stochastic Oscillator")
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
plt.legend()
# Williams %R Plot
plt.subplot(3, 3, 7)
plt.plot(stock_data.index, stock_data['WILLR_14'])
plt.title("Williams %R")
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
# ADX Plot
plt.subplot(3, 3, 8)
plt.plot(stock_data.index, stock_data['ADX_14'])
plt.title("Average Directional Index (ADX)")
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
# CMF Plot
plt.subplot(3, 3, 9)
plt.plot(stock_data.index, stock_data['CMF_20'])
plt.title("Chaikin Money Flow (CMF)")
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%b%d')) # Format date as "Jun14"
plt.xticks(rotation=45, fontsize=8) # Adjust font size
# Show the plots
plt.tight_layout()
plt.show()
return last_day_summary
else:
logging.error("Stock data is None, unable to calculate indicators.")
return None
except Exception as e:
logging.error(f"Error during analysis: {e}")
return None
def get_finance_data(symbol):
# FIXME: Expose them to end users.
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, expiry_date):
"""
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):
""" Function to get information for Options & Futures Trading """
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

91
lib/ai_writers/ai_financial_writer.py Normal file
View File

@@ -0,0 +1,91 @@
import sys
import os
from textwrap import dedent
from pathlib import Path
from datetime import datetime
from loguru import logger
logger.remove()
logger.add(sys.stdout,
colorize=True,
format="<level>{level}</level>|<green>{file}:{line}:{function}</green>| {message}"
)
from ..ai_web_researcher.finance_data_researcher import get_finance_data, get_fin_options_data
from ..gpt_providers.text_generation.main_text_generation import llm_text_gen
def write_basic_ta_report(symbol):
""" Write financial TA for given ticker symbol """
try:
symbol_fin_data = get_finance_data(symbol)
#get_visual_reports
fin_report = gen_finta_report(symbol_fin_data, symbol)
logger.info(f"Done: Final Technical Analysis for {symbol}:\n\n")
except Exception as err:
logger.error(f"Error: Failed to generate Financial report: {err}")
#fin_options_data = get_fin_options_data(symbol)
#options_report = gen_options_report(fin_options_data, symbol)
def gen_options_report(results_sentences, ticker):
""" Call LLM to generate options report """
prompt = f"""
You are a financial expert specializing in options trading and market sentiment analysis.
You have been provided with the following technical analysis of options data for the ticker symbol {ticker} with the nearest expiry date:
{chr(10).join(results_sentences)}
Based on this data, provide a comprehensive analysis of the options market for {ticker}.
Your analysis should include:
1. **Implied Volatility Interpretation:** Discuss the significance of the average implied volatility for both call and put options. What does it suggest about market expectations of future price movements?
2. **Volume and Open Interest Insights:** Analyze the volume and open interest for call and put options. What does this data reveal about current market positioning and potential future trading activity?
3. **Sentiment Analysis:** Evaluate the put-call ratio, implied volatility skew, and overall market sentiment. What do these indicators suggest about trader sentiment and potential future price direction?
4. **Potential Trading Strategies:** Based on your analysis, suggest potential options trading strategies that could be employed for {ticker}, considering the current market conditions and sentiment.
Please provide your analysis in a clear and concise manner, suitable for someone with a good understanding of options trading.
"""
logger.info("Generating Financial Technical report..")
try:
response = llm_text_gen(prompt)
return response
except Exception as err:
logger.error(f"Exit: Failed to get response from LLM: {err}")
exit(1)
def gen_finta_report(last_day_summary, symbol):
""" Get AI to write TA report from given data """
prompt = f"""
You are a seasoned Technical Analysis (TA) expert, rivaling legends like Charles Dow, John Bollinger, and Alan Andrews.
Your deep understanding of market dynamics, coupled with mastery of technical indicators,
allows you to decipher complex patterns and offer precise predictions.
Your expertise extends to practical tools like the pandas_ta module, enabling you to extract valuable insights from raw data.
**Objective:**
Analyze the provided technical indicators for {symbol} on its last trading day and predict its price movement over the next few trading sessions.
**Instructions:**
1. **Identify Potential Trading Signals:** Highlight specific indicators suggesting bullish, bearish, or neutral signals. Explain the rationale behind each signal, referencing historical patterns or comparable market scenarios.
2. **Detect Patterns and Divergences:** Analyze the interplay between different indicators. Detect patterns like moving average crossovers, candlestick formations, or divergences between price action and indicators. Explain the significance of each pattern.
3. **Price Movement Prediction:** Based on your analysis, provide a clear prediction for {symbol}'s price movement in the next few days. State the expected direction (up, down, sideways) and potential price targets if identifiable.
4. **Risk Assessment:** Briefly discuss any potential risks or factors that could invalidate your predictions, promoting a balanced and informed perspective.
**Technical Indicators for {symbol} on the Last Trading Day:**
{last_day_summary}
Remember, your analysis should be detailed, insightful, and actionable for traders seeking to capitalize on market movements.
"""
logger.info("Generating Financial Technical report..")
try:
response = llm_text_gen(prompt)
return response
except Exception as err:
logger.error(f"Exit: Failed to get response from LLM: {err}")
exit(1)

24
lib/utils/alwrity_utils.py
View File

@@ -16,6 +16,7 @@ from lib.ai_writers.speech_to_blog.main_audio_to_blog import generate_audio_blog
from lib.ai_writers.long_form_ai_writer import long_form_generator
from lib.ai_writers.ai_news_article_writer import ai_news_generation
from lib.ai_writers.ai_agents_crew_writer import ai_agents_writers
from lib.ai_writers.ai_financial_writer import write_basic_ta_report
from lib.gpt_providers.text_generation.ai_story_writer import ai_story_generator
from lib.gpt_providers.text_generation.ai_essay_writer import ai_essay_generator
from lib.gpt_providers.text_to_image_generation.main_generate_image_from_prompt import generate_image
@@ -46,6 +47,29 @@ def blog_from_audio():
break
def ai_finance_ta_writer():
""" Call upon AI finance writer with user inputs. """
print("________________________________________________________________")
content_keywords = input_dialog(
title='Enter Ticker Symbol For TA.',
text='👋 Be sure of ticker symbol, Else no results:(Examples:IBM, BABA, HDFCBANK.NS, TATAMOTORS.NS etc)',
).run()
# If the user cancels, exit the loop
if content_keywords.strip():
try:
write_basic_ta_report(content_keywords)
except Exception as err:
print(f"🚫 Check ticker symbol: Failed to write Financial Technical Analysis.")
exit(1)
else:
message_dialog(
title='Error',
text='🚫 Provide Symbol Ticker. Dont waste my time.'
).run()
exit(1)
def blog_from_keyword():
""" Input blog keywords, research and write a factual blog."""
while True:

2
requirements.txt
View File

@@ -26,3 +26,5 @@ prompt_toolkit
ipython
html2image
lxml_html_clean
yfinance
pandas_ta