src/fetch_forecast.py

import logging
import os

import dotenv
import pandas as pd
import requests


class FetchForecast:
    def __init__(self, ticker: str, debug=False) -> None:
        if debug:
            self.logger_level = logging.DEBUG
        else:
            self.logger_level = logging.INFO
        self.logger = logging.getLogger(__name__)
        logging.basicConfig(level=self.logger_level)

        # args
        self.ticker = ticker
        # constants
        self.past_horizon = 5  # number of past business days
        # build the api-url based on env variables
        self.api_env = os.environ.get("FORECAST_API_ENV")
        api_url_temp = os.environ.get("API_URL_TEMPLATE")
        self.api_url = api_url_temp.replace("ENV", self.api_env)

    def run(self):
        past_df, fcst_df = self.call_api()
        return past_df, fcst_df

    def call_api(self) -> tuple:
        pl_in = {"ticker": self.ticker, "past_horizon": self.past_horizon}
        self.logger.info(f"Calling forecast API ({self.api_env}) for ticker: {self.ticker}")
        resp = requests.post(f"{self.api_url}/forecast", json=pl_in, timeout=30)
        if resp.status_code == 200:
            data = resp.json()
            past_df, fcst_df = self.transform_data(data)
        else:
            self.logger.error(f"Error (status: {resp.status_code}) fetching stock info for {self.ticker}.")
            past_df, fcst_df = None, None
        return past_df, fcst_df
    
    def transform_data(self, data) -> tuple:
        past_df = pd.DataFrame(data["past"]).rename(columns={"index": "Date"})
        fcst_df = pd.DataFrame(data["forecast"]).rename(columns={"index": "Date"})
        # move Date to the front
        past_df = past_df[["Date"] + [col for col in past_df.columns if col != "Date"]]
        fcst_df = fcst_df[["Date"] + [col for col in fcst_df.columns if col != "Date"]]
        return past_df, fcst_df
    

if __name__ == "__main__":
    # load the env variables fom .env file
    dotenv.load_dotenv(dotenv.find_dotenv())
    past_df, fcst_df = FetchForecast("AAPL").run()
    print("Last available price:\n", past_df.tail(1))
    print("Forecasts:\n", fcst_df.head())