path: root/openbb_terminal/stocks/fundamental_analysis/av_model.py

"""Alpha Vantage Model"""
__docformat__ = "numpy"

import logging
from typing import Dict, List

import numpy as np
import pandas as pd
from alpha_vantage.fundamentaldata import FundamentalData

from openbb_terminal.core.session.current_user import get_current_user
from openbb_terminal.decorators import log_start_end
from openbb_terminal.helper_funcs import lambda_long_number_format, request
from openbb_terminal.rich_config import console
from openbb_terminal.stocks.fundamental_analysis import yahoo_finance_model
from openbb_terminal.stocks.fundamental_analysis.fa_helper import clean_df_index
from openbb_terminal.stocks.stocks_helper import clean_fraction

logger = logging.getLogger(__name__)


def check_premium_key(json_response: Dict) -> bool:
    """Checks if the response is the premium endpoint"""
    if json_response == {
        "Information": "Thank you for using Alpha Vantage! This is a premium endpoint. You may subscribe to "
        "any of the premium plans at https://www.alphavantage.co/premium/ to instantly unlock all premium endpoints"
    }:
        console.print(
            "This is a premium endpoint for AlphaVantage. Please use a premium key.\n"
        )
        return True
    return False


@log_start_end(log=logger)
def get_overview(symbol: str) -> pd.DataFrame:
    """Get alpha vantage company overview

    Parameters
    ----------
    symbol : str
        Stock ticker symbol

    Returns
    -------
    pd.DataFrame
        Dataframe of fundamentals
    """
    # Request OVERVIEW data from Alpha Vantage API
    s_req = (
        f"https://www.alphavantage.co/query?function=OVERVIEW&symbol={symbol}&apikey"
        f"={get_current_user().credentials.API_KEY_ALPHAVANTAGE}"
    )
    result = request(s_req, stream=True)
    result_json = result.json()

    df_fa = pd.DataFrame()

    # If the returned data was unsuccessful
    if "Error Message" in result_json:
        console.print(result_json["Error Message"])
    # check if json is empty
    elif not result_json:
        console.print("No data found from Alpha Vantage\n")
    # Parse json data to dataframe
    elif "Note" in result_json:
        console.print(result_json["Note"], "\n")
    else:
        df_fa = pd.json_normalize(result_json)

        # Keep json data sorting in dataframe
        df_fa = df_fa[list(result_json.keys())].T
        df_fa.iloc[5:] = df_fa.iloc[5:].applymap(lambda x: lambda_long_number_format(x))
        df_fa.columns = [" "]

    return df_fa


@log_start_end(log=logger)
def get_key_metrics(symbol: str) -> pd.DataFrame:
    """Get key metrics from overview

    Parameters
    ----------
    symbol : str
        Stock ticker symbol

    Returns
    -------
    pd.DataFrame
        Dataframe of key metrics
    """
    # Request OVERVIEW data
    s_req = (
        f"https://www.alphavantage.co/query?function=OVERVIEW&symbol={symbol}"
        f"&apikey={get_current_user().credentials.API_KEY_ALPHAVANTAGE}"
    )
    result = request(s_req, stream=True)
    result_json = result.json()

    # If the returned data was unsuccessful
    if "Error Message" in result_json:
        console.print(result_json["Error Message"])
    else:
        # check if json is empty
        if not result_json or len(result_json) < 2:
            console.print("No data found from Alpha Vantage\n")
            return pd.DataFrame()

        df_fa = pd.json_normalize(result_json)
        df_fa = df_fa[list(result_json.keys())].T

        if not get_current_user().preferences.USE_INTERACTIVE_DF:
            df_fa = df_fa.applymap(lambda x: lambda_long_number_format(x))
        clean_df_index(df_fa)
        df_fa = df_fa.rename(
            index={
                "E b i t d a": "EBITDA",
                "P e ratio": "PE ratio",
                "P e g ratio": "PEG ratio",
                "E p s": "EPS",
                "Return on equity t t m": "Return on equity TTM",
                "Price to sales ratio t t m": "Price to sales ratio TTM",
            }
        )
        as_key_metrics = [
            "Market capitalization",
            "EBITDA",
            "EPS",
            "PE ratio",
            "PEG ratio",
            "Price to book ratio",
            "Return on equity TTM",
            "Price to sales ratio TTM",
            "Dividend yield",
            "50 day moving average",
            "Analyst target price",
            "Beta",
        ]
        return df_fa.loc[as_key_metrics]

    return pd.DataFrame()


@log_start_end(log=logger)
def get_income_statements(
    symbol: str,
    limit: int = 5,
    quarterly: bool = False,
    ratios: bool = False,
    plot: bool = False,
) -> pd.DataFrame:
    """Get income statements for company

    Parameters
    ----------
    symbol : str
        Stock ticker symbol
    limit : int
        Number of past to get
    quarterly : bool, optional
        Flag to get quarterly instead of annual, by default False
    ratios: bool
        Shows percentage change, by default False
    plot: bool
        If the data shall be formatted ready to plot

    Returns
    -------
    pd.DataFrame
        DataFrame of income statements
    """
    url