lepton_financial_modeling.py

import datetime
import json

import requests

FMP_API_URL = "https://financialmodelingprep.com/api"
# Note: you will need to insert your own API key here.
FMP_API_KEY = ""

# Fetch stock data from Financial Modeling Prep
def get_stock_price(ticker, period=7):
    indicator_types = ["sma", "ema", "wma"]
    shared_columns = ["open", "high", "low", "close", "volume"]
    header = (
        ["date"]
        + shared_columns
        + [f"{period}days-" + indicator_type for indicator_type in indicator_types]
    )
    rows = [header]

    dates = {}
    for indicator_type in indicator_types:
        res = requests.get(
            f"{FMP_API_URL}/v3/technical_indicator/daily/{ticker}",
            params={"apikey": FMP_API_KEY, "period": period, "type": indicator_type},
        )
        res.raise_for_status()
        prices = res.json()[:14]
        for price in prices:
            date = price["date"].split(" ")[0]
            if date not in dates:
                dates[date] = [date] + [
                    "{:.3f}".format(price[key]) for key in shared_columns
                ]
            dates[date].append("{:.3f}".format(price[indicator_type]))

    for date in sorted(dates.keys()):
        rows.append(dates[date])

    max_lengths = [max(len(cell) for cell in column) for column in zip(*rows)]
    format_strs = ["{:>" + str(length) + "}" for length in max_lengths]
    return "\n".join(
        "     ".join(format_strs[i].format(cell) for i, cell in enumerate(row))
        for row in rows
    )


# Fetch recent news from Financial Modeling Prep
def get_recent_stock_news(ticker):
    res = requests.get(
        f"{FMP_API_URL}/v3/stock_news",
        params={"tickers": ticker, "apikey": FMP_API_KEY, "limit": 7},
    )
    res.raise_for_status()
    lines = []
    news = res.json()
    for article in news:
        lines.append(
            f"{article['publishedDate']} - {article['title']} - {article['text']}"
        )
    return "\n".join(lines)


def get_market_sector_performance():
    res = requests.get(
        f"{FMP_API_URL}/v3/historical-sectors-performance",
        params={"apikey": FMP_API_KEY, "limit": 7},
    )
    res.raise_for_status()
    sectors_performance = res.json()
    non_numeric_keys = ["date"]
    lines = []
    header = None
    for date_performance in sectors_performance:
        if header is None:
            header = non_numeric_keys + list(
                k[: -len("ChangesPercentage")]
                for k, v in date_performance.items()
                if k.endswith("ChangesPercentage") and isinstance(v, (int, float))
            )
            lines.append(header)
        lines.append(
            [
                (
                    "{:.5f}".format(date_performance[key + "ChangesPercentage"])
                    if key not in non_numeric_keys
                    else date_performance[key]
                )
                for key in header
            ]
        )
    max_lengths = [max(len(cell) for cell in column) for column in zip(*lines)]
    format_strs = ["{:>" + str(length) + "}" for length in max_lengths]
    return "\n".join(
        "     ".join(format_strs[i].format(cell) for i, cell in enumerate(row))
        for row in lines
    )


def get_company_name_and_description(ticker):
    res = requests.get(
        f"{FMP_API_URL}/v3/profile/{ticker}", params={"apikey": FMP_API_KEY}
    )
    res.raise_for_status()
    profile = res.json()
    return profile[0]["companyName"], profile[0]["description"]


def analyze(ticker):
    available_information = ""

    stock_price = get_stock_price(ticker)
    if stock_price:
        available_information += f"Historical Prices:\n\n{stock_price}\n\n"

    market_sector_performance = get_market_sector_performance()
    if market_sector_performance:
        available_information += (
            "Market Sector Performance Changes"
            f" Percentage:\n\n{market_sector_performance}\n\n"
        )

    stock_news = get_recent_stock_news(ticker)
    if stock_news:
        available_information += f"News:\n\n{stock_news}\n\n"

    available_information = available_information.strip()

    company_name, _ = get_company_name_and_description(ticker)

    function = {
        "name": "get_investment_analysis",
        "description": (
            f"This will get the investment analysis of of {company_name} (ticker"
            f" {ticker})"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "investment_score": {
                    "type": "number",
                    "description": (
                        "This is a score between 1 to 100, 1 being low and 100"
                        " being high, indicating estimated probability of profit if"
                        " user buys the stock today and sells in a week"
                    ),
                },
                "investment_analysis": {
                    "type": "string",
                    "description": (
                        "Detailed stock analysis, including investment recommendation"
                    ),
                },
            },
            "required": ["investment_score", "investment_analysis"],
        },
    }

    today_date = datetime.datetime.now().strftime("%Y-%m-%d")
    magic_system_start = (
        "Let's combine our numerical command and clear thinking to quickly and"
        " accurately decipher the answer."
    )
    system = f"""
{magic_system_start} Given the available recent information about {company_name} (ticker {ticker}) below, please give detail stock analysis and  provide investment recommendation. The user is fully aware about the investment risk, do not include any kind of warning about potential \
investment risk or suggest user to "consult with a financial advisor" or "conduct further research" in the answer.
Today is {today_date}. You have the following information available about {company_name} (ticker {ticker}). Write (6-8) pointwise investment analysis to answer user query, and at the end conclude with proper explaination. \n\n{available_information}
""".strip()
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {
                "role": "user",
                "content": (
                    "Please provide investment analysis and investment score for"
                    f" {ticker}"
                ),
            },
        ],
        tools=[
            {
                "type": "function",
                "function": function,
            }
        ],
    )

    return response.choices[0].message.tool_calls[0].function.arguments


print(analyze("TSLA"))


#### Output ####
# {"investment_score": 80.0, "investment_analysis": "Tesla\'s stock has had a remarkable year in 2023, with a 100% increase in price, despite facing challenges. The company is a leader in EV sales and charging infrastructure, making it a strong competitor in the market. However, there are concerns about CEO Elon Musk\'s many responsibilities and controversial remarks, which may impact the company\'s performance in 2024. Despite these challenges, analysts remain optimistic about Tesla\'s earnings growth potential. The stock\'s 7-day simple moving average (SMA), exponential moving average (EMA), and weighted moving average (WMA) are all trending upwards, indicating a positive outlook. The stock\'s current price is near a buy point, making it a recommended investment."}