Source code for kaxanuk.attribution_analysis.input_handlers.csv_input

"""
CSV input handler for market data files.

Read per-ticker CSV files, parse columns as strings for consistent type
inference, then delegate to the shared processing pipeline for decimal
casting, date parsing, and date-range filtering.
"""

import logging
from pathlib import Path

import pyarrow as pa
import pyarrow.csv as pv

from kaxanuk.attribution_analysis.entities.configuration import Configuration
from kaxanuk.attribution_analysis.exceptions import ReadingCSVError
from kaxanuk.attribution_analysis.input_handlers import input_handlers_helpers
from kaxanuk.attribution_analysis.input_handlers.input_handler_interface import InputHandlerInterface
from kaxanuk.attribution_analysis.market_data_pipeline.field_definitions import STANDARD_FIELDS_ORDER

logger = logging.getLogger(__name__)




class CsvInput(InputHandlerInterface):
    """
    A CSV input handler for reading and processing market data from CSV files.

    This class implements the InputHandlerInterface to provide functionality for
    reading CSV files containing market data, validating their existence, and
    processing them into PyArrow tables with proper data type conversions and
    date range filtering.

    The handler supports column mapping through equivalence dictionaries and
    performs data validation including empty value checks and decimal precision
    inference for numerical columns.

    Attributes
    ----------
    input_dir : str
        The base directory path where input files are located.
    """

    def __init__(
        self,
        *,
        input_dir: str,
    ) -> None:
        """
        Initialize the CSV input handler with directory configuration.

        Parameters
        ----------
        input_dir : str
            The base directory path where input files are stored.
            Must be a valid directory path accessible by the application.

        Notes
        -----
        The complete path to CSV files will be constructed as:
        {input_dir}/{ticker}.csv
        """
        super().__init__()
        self.input_dir = input_dir



    def file_exists(
        self,
        *,
        ticker: str,
        configuration: Configuration,
    ) -> bool:
        """
        Check if a CSV file exists for the specified ticker without loading it into memory.

        This method performs a lightweight existence check by constructing the expected
        file path and verifying if the file exists on the filesystem. It does not
        validate the file content or structure.

        Parameters
        ----------
        ticker : str
            The ticker symbol for which to check file existence.
            The method will look for a file named '{ticker}.csv' in the
            configured input directory.
        configuration : Configuration
            The configuration object containing parameters. While not directly
            used in this method, it maintains interface consistency.

        Returns
        -------
        bool
            True if the CSV file exists at the expected path, False otherwise.

        Notes
        -----
        The file path is constructed as: {input_dir}/{ticker}.csv
        """
        input_file_path = Path(self.input_dir) / f"{ticker}.csv"
        return input_file_path.exists()




    def input_data(
            self,
            *,
            ticker: str,
            configuration: Configuration,
            equivalence_dict: dict[str, str],
    ) -> pa.Table:
        """
        Read and process CSV input data for a specified ticker symbol.

        This method reads market data from a CSV file and delegates all processing
        operations (type conversion, date parsing, decimal inference, validation,
        and filtering) to a common helper function. The CSV-specific aspect is
        reading all columns as strings initially to allow for flexible parsing.

        Parameters
        ----------
        ticker : str
            The ticker symbol for which to load market data.
            The method will look for a file named '{ticker}.csv' in the
            configured input directory.
        configuration : Configuration
            Configuration object containing:

            - start_date: Start date for data filtering
            - end_date: End date for data filtering
        equivalence_dict : dict[str, str]
            Dictionary mapping STANDARD field names (e.g. 'vwap',
            'adjusted_close') to actual CSV column names used in the files.

        Returns
        -------
        pa.Table
            A PyArrow table containing the processed market data with:

            - Properly typed columns (dates as timestamps, numbers as decimals)
            - Filtered data within the specified date range
            - Standardized column names as per STANDARD_FIELDS_ORDER
            - Validated data with no empty values in critical columns

        Raises
        ------
        ReadingCSVError
            Raised in the following scenarios:

            - FileNotFoundError: When the CSV file for the ticker doesn't exist
            - pa.lib.ArrowInvalid: When the CSV format is invalid or corrupted
            - KeyError: When column mapping fails due to missing keys
            - Exception: For any other unexpected errors during processing

        Notes
        -----
        Unlike ParquetInput which can rely on stored types, CSV files are read
        with all columns as strings to ensure consistent parsing behavior before
        type conversion.
        """
        input_file_path = f"{self.input_dir}/{ticker}.csv"

        try:
            # Get columns to read in standard order
            csv_columns_in_order = [
                equivalence_dict[std_name] for std_name in STANDARD_FIELDS_ORDER
            ]
            unique_csv_columns = list(dict.fromkeys(csv_columns_in_order))

            # Read CSV with all columns as strings for consistent parsing
            column_types = {col: pa.string() for col in unique_csv_columns}
            table = pv.read_csv(
                input_file_path,
                convert_options=pv.ConvertOptions(
                    include_columns=unique_csv_columns,
                    column_types=column_types,
                ),
            )

            # Delegate all processing to common helper
            return input_handlers_helpers.process_table_columns(
                table=table,
                ticker=ticker,
                configuration=configuration,
                equivalence_dict=equivalence_dict,
            )

        except FileNotFoundError as error:
            raise ReadingCSVError(
                f"File not found for ticker '{ticker}': {input_file_path}"
            ) from error

        except pa.lib.ArrowInvalid as error:
            raise ReadingCSVError(
                f"Invalid CSV format for ticker '{ticker}': {error}"
            ) from error

        except pa.lib.ArrowTypeError as error:
            raise ReadingCSVError(
                f"Type conversion error for ticker '{ticker}': {error}"
            ) from error

        except KeyError as error:
            raise ReadingCSVError(
                f"Column mapping error for ticker '{ticker}': {error}. "
                f"Check your equivalence_dict and STANDARD_FIELDS_ORDER."
            ) from error

        except ValueError as error:
            raise ReadingCSVError(
                f"Value error reading CSV for ticker '{ticker}': {error}"
            ) from error

        except OSError as error:
            raise ReadingCSVError(
                f"I/O error reading CSV for ticker '{ticker}': {error}"
            ) from error