Source code for kaxanuk.attribution_analysis.input_handlers.portfolio_weights_input_handler

import pandas as pd

from kaxanuk.attribution_analysis.interfaces.input_handler_interface import InputHandlerInterface
from kaxanuk.attribution_analysis.exceptions import MissingCSVReadError
from kaxanuk.attribution_analysis.entities import PortfolioDailyWeights



class WeightsInputHandler(InputHandlerInterface):
    """
    Read portfolio weights input.

    Parameters
    ----------
    input_base_dir : str
        The path that will contain all the input subdirectories
    input_subdir: str
        The subdirectory to which the input data files will be readed

    """

    def __init__(
            self,
            input_base_dir: str,
            input_subdir: str,
            file_name: str,
            dates_column_name: str = 'm_date',
    ):
        self.input_base_dir = input_base_dir
        self.input_subdir = input_subdir
        self.file_name = file_name
        self.dates_column_name = dates_column_name
        # HAcer metodo privado que ejecute el leer las cosas
        # Siempre que tengan attributos que vaya a jalar el usuario que sean publicos para que no jale los privados
        # hacer interfaz para reeder tool, que lo recibe como input y ejecuta los metodos publicos del



    def input_data(
            self,
            *,
            ticker: str,
    ) -> PortfolioDailyWeights:
        """


        Parameters
        ----------
        ticker : str
            The ticker of the stock, on which the output csv file will be named.

        Returns
        -------
        Pandas DataFrame
        """
        input_subdir_path = f'{self.input_base_dir}/{self.input_subdir}'

        try:
            input_file_path = f'{input_subdir_path}/{self.file_name}.csv'
        except KeyError as error:
            raise MissingCSVReadError from error

        table_df = pd.read_csv(
            input_file_path,
            index_col=self.dates_column_name,
            parse_dates=[0]

        )
        table_df = table_df.sort_index()
        table_df.index = pd.to_datetime(table_df.index, format='mixed')

        return PortfolioDailyWeights(table_df)






def read_csv_intersection(path, wanted_cols, index_col=0, **kwargs):
    # Read only column names (no data)
    available = pd.read_csv(path, nrows=0).columns

    # Keep only columns that exist
    cols = [c for c in wanted_cols if c in set(available)]

    # Ensure the index column is included so index_col works
    idx_name = available[index_col] if isinstance(index_col, int) else index_col
    if idx_name not in cols:
        cols = [idx_name, *cols]

    return pd.read_csv(
        path,
        usecols=cols,
        index_col=index_col,
        parse_dates=[index_col],
        **kwargs
    )