tools/chart_expert.py

from pydantic.v1 import BaseModel, Field
from langchain.tools import BaseTool
from typing import Optional, Type
from langchain.tools import StructuredTool
import yfinance as yf
from typing import List
from datetime import datetime,timedelta
import matplotlib.pyplot as plt
import chainlit as cl
import plotly.graph_objects as go
import pandas as pd
import yfinance as yf
from plotly.subplots import make_subplots

def chart_expert_tools(): 

    def historical_stock_prices(stockticker, days_ago):
        """Upload accurate data to accurate dates from yahoo finance.
        Receive data on the last week and give them to forecasting experts.
        Receive data on the last 90 days and give them to visualization expert."""
        ticker = yf.Ticker(stockticker)
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days_ago)
        start_date = start_date.strftime('%Y-%m-%d')
        end_date = end_date.strftime('%Y-%m-%d')
        historical_data = ticker.history(start=start_date, end=end_date)
        return historical_data

    class HistoricalStockPricesInput(BaseModel):
        """Input for Stock ticker check."""

        stockticker: str = Field(..., description="Ticker symbol for stock or index")
        days_ago: int = Field(..., description="Int number of days to look back")

    class HistoricalStockPricesTool(BaseTool):
        name = "historical_stock_prices"
        description = "Useful for when you need to find out the historical stock prices. Use Yahoo Finance API to find the correct stockticker."

        def _run(self, stockticker: str, days_ago: int):
            historical_prices = historical_stock_prices(stockticker, days_ago)

            return {"historical prices":  historical_prices}

        def _arun(self, stockticker: str, days_ago: int):
            raise NotImplementedError("This tool does not support async")

        args_schema: Optional[Type[BaseModel]] = HistoricalStockPricesInput

    def calculate_MACD(historical_data, fast_period=12, slow_period=26, signal_period=9):
        """
        Calculates the MACD (Moving Average Convergence Divergence) and related indicators.

        Parameters:
            df (DataFrame): A pandas DataFrame containing at least a 'Close' column with closing prices.
            fast_period (int): The period for the fast EMA (default is 12).
            slow_period (int): The period for the slow EMA (default is 26).
            signal_period (int): The period for the signal line EMA (default is 9).

        Returns:
            DataFrame: A pandas DataFrame with the original data and added columns for MACD, Signal Line, and MACD Histogram.
        """
        df=historical_data[['Close','Open','High','Low']]
        df['EMA_fast'] = df['Close'].ewm(span=fast_period, adjust=False).mean()
        df['EMA_slow'] = df['Close'].ewm(span=slow_period, adjust=False).mean()
        df['MACD'] = df['EMA_fast'] - df['EMA_slow']

        df['Signal_Line'] = df['MACD'].ewm(span=signal_period, adjust=False).mean()
        df['MACD_Histogram'] = df['MACD'] - df['Signal_Line']
        return df


    class MACDCalculateInput(BaseModel):
        """Input for Stock ticker check."""
        stockticker: str = Field(..., description="Ticker symbol for stock or index")

    class MACDCalculateTool(BaseTool):
        name = "macd_calculate"
        description = "Useful for calculating MACD as input for MACD plot."

        def _run(self, stockticker: str, historical_data: float):
            df = calculate_MACD(historical_data)

            return df

        def _arun(self, stockticker: str, historical_data: float):
            raise NotImplementedError("This tool does not support async")

        args_schema: Optional[Type[BaseModel]] = MACDCalculateInput

    def plot_macd(df):

        # Create Figure
        fig = make_subplots(rows=2, cols=1, shared_xaxes=True, row_heights=[0.7, 0.3],
                            vertical_spacing=0.15,  # Adjust vertical spacing between subplots
                            subplot_titles=("Candlestick Chart", "MACD"))  # Add subplot titles


        # Subplot 1: Plot candlestick chart
        fig.add_trace(go.Candlestick(
            x=df.index,
            open=df['Open'],
            high=df['High'],
            low=df['Low'],
            close=df['Close'],
            increasing_line_color='#00cc96',  # Green for increasing
            decreasing_line_color='#ff3e3e',  # Red for decreasing
            showlegend=False
        ), row=1, col=1)  # Specify row and column indices


        # Subplot 2: Plot MACD
        fig.add_trace(
            go.Scatter(
                x=df.index,
                y=df['MACD'],
                mode='lines',
                name='MACD',
                line=dict(color='blue')
            ),
            row=2, col=1
        )

        fig.add_trace(
            go.Scatter(
                x=df.index,
                y=df['Signal_Line'],
                mode='lines',
                name='Signal Line',
                line=dict(color='red')
            ),
            row=2, col=1
        )

        # Plot MACD Histogram with different colors for positive and negative values
        histogram_colors = ['green' if val >= 0 else 'red' for val in df['MACD_Histogram']]

        fig.add_trace(
            go.Bar(
                x=df.index,
                y=df['MACD_Histogram'],
                name='MACD Histogram',
                marker_color=histogram_colors
            ),
            row=2, col=1
        )

        # Update layout with zoom and pan tools enabled
        layout = go.Layout(
            title='MSFT Candlestick Chart and MACD Subplots',
            title_font=dict(size=25),  # Adjust title font size
            plot_bgcolor='#f2f2f2',  # Light gray background
            height=800,
            width=1500,
            xaxis_rangeslider=dict(visible=True, thickness=0.03),
        )

        # Update the layout of the entire figure
        fig.update_layout(layout)
        fig.update_yaxes(fixedrange=False, row=1, col=1)
        fig.update_yaxes(fixedrange=True, row=2, col=1)
        fig.update_xaxes(type='category', row=1, col=1)
        fig.update_xaxes(type='category', nticks=10, row=2, col=1)

        fig.show()

    class PlotMACDInput(BaseModel):
        """Input for Stock ticker check."""

        stockticker: str = Field(..., description="Ticker symbol for stock or index")
        df: List = Field(..., description="List of historical price values")
        days_ago: int = Field(..., description="Int number of days to look back")

    class PlotMACDTool(BaseTool):
        name = "plot_macd"
        description = "Useful for creating beautiful candle stick plot for MACD for a stock price."

        def _run(self, df: List[float]):
            historical_prices = plot_macd(df)

            return {"historical prices":  historical_prices}

        def _arun(self, df: List[float]):
            raise NotImplementedError("This tool does not support async")

        args_schema: Optional[Type[BaseModel]] = PlotMACDInput



    tools_chart_expert = [StructuredTool.from_function(
            func=HistoricalStockPricesTool,
            args_schema=HistoricalStockPricesInput,
            description="Function to get historical stock prices.",
        ),
        StructuredTool.from_function(
            func=MACDCalculateTool,
            args_schema=MACDCalculateInput,
            description="Calculate MACD as  input for MACD plot.",
        ),
        StructuredTool.from_function(
            func=PlotMACDTool,
            args_schema=PlotMACDInput,
            description="Plot MACD.",
        ),

    ]
    return tools_chart_expert