Input Handlers#

Concrete input handlers for loading market data from CSV and Parquet files, normalizing raw data into standardized PyArrow tables, plus the contract they implement and the shared helper functions they use. See Data Formats for the expected file layouts.

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.

class CsvInput(*, input_dir: str)[source]#

Bases: 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.

input_dir#

The base directory path where input files are located.

Type:

str

file_exists(*, ticker: str, configuration: Configuration) → bool[source]#

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:

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

Return type:

bool

Notes

The file path is constructed as: {input_dir}/{ticker}.csv

input_data(*, ticker: str, configuration: Configuration, equivalence_dict: dict[str, str]) → Table[source]#

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:

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

Return type:

pa.Table

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.

Parquet input handler for market data files.

Read per-ticker Parquet files with efficient columnar loading, then delegate to the shared processing pipeline for type casting, date parsing, and date-range filtering.

class ParquetInput(*, input_dir: str)[source]#

Bases: InputHandlerInterface

A specialized input handler for reading and processing Parquet files containing market data.

This class implements the InputHandlerInterface to provide functionality for reading market data from Parquet files. It handles file existence checks, data loading, column mapping, type conversion, and date filtering operations. The class is designed to work with financial market data stored in Parquet format, providing efficient columnar data processing using Apache Arrow.

The ParquetInput handler supports: - Efficient columnar data reading from Parquet files - Automatic type inference and conversion to decimal types for financial data - Date column parsing and validation - Column name mapping between input files and standardized field names - Date range filtering based on configuration parameters - Comprehensive error handling for file operations

input_dir#

The directory path where input files are located.

Type:

str

file_exists(*, ticker: str, configuration: Configuration) → bool[source]#

Check if a Parquet file exists for the specified ticker without loading the data.

Parameters:

  • ticker (str) – The ticker symbol for which to check file existence.

  • configuration (Configuration) – The configuration object containing settings.

Returns:

True if the Parquet file exists for the given ticker, False otherwise.

Return type:

bool

input_data(*, ticker: str, configuration: Configuration, equivalence_dict: dict[str, str]) → Table[source]#

Read and process market data from a Parquet file for the specified ticker.

This method reads market data from a Parquet file and delegates all processing operations (type conversion, date parsing, decimal inference, validation, and filtering) to a common helper function. The Parquet-specific aspect is the efficient columnar reading with selective column loading.

Parameters:

  • ticker (str) – The ticker symbol for which to load market data.

  • 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 to actual Parquet column names used in the files.

Returns:

A PyArrow table containing the processed market data.

Return type:

pa.Table

Raises:

DataLoadingError – Raised in the following scenarios: - FileNotFoundError: When the Parquet file for the ticker doesn’t exist - pa.lib.ArrowInvalid: When the Parquet format is invalid or corrupted - KeyError: When column mapping fails due to missing keys - Exception: For any other unexpected errors during processing

Abstract base class interface for handling input data sources in the attribution analysis engine.

This module defines the contract that all input handlers must implement to provide data loading capabilities for different data sources (CSV, Parquet, Excel, etc.). Input handlers are responsible for loading and formatting ticker data.

class InputHandlerInterface[source]#

Bases: object

Abstract base class for input data handlers.

This interface defines the contract that all input handlers must implement to provide consistent data loading capabilities across different data sources. Input handlers are responsible for loading ticker data and checking data availability for the engine.

All concrete implementations must provide methods to: - Load data for a specific ticker - Check if data exists for a ticker

abstractmethod file_exists(*, ticker: str, configuration: Configuration) → bool[source]#

Check if data exists for a specific ticker.

Parameters:

  • ticker (str) – The ticker symbol to check for data availability.

  • configuration (Configuration) – Configuration object containing parameters and data source settings.

Returns:

True if data exists for the ticker, False otherwise.

Return type:

bool

abstractmethod input_data(*, ticker: str, configuration: Configuration, equivalence_dict: dict[str, str]) → Table[source]#

Load and return data for a specific ticker.

Parameters:

  • ticker (str) – The ticker symbol for which to load data.

  • configuration (Configuration) – Configuration object containing parameters and data source settings.

  • equivalence_dict (dict[str, str]) – Dictionary mapping column names to standardized column names for data consistency.

Returns:

PyArrow Table containing the loaded ticker data with standardized column names and proper data types.

Return type:

pa.Table

Helper functions for input handlers.

This module provides utility functions used by various input handlers (CSV, Parquet, Excel) for common data processing operations including:

  • Column name mapping and transformation

  • Data quality validation

  • Date range filtering

  • Decimal precision inference

  • Date parsing with ISO format support

All functions are designed to work with PyArrow tables and arrays for efficient columnar data processing.

apply_column_mapping(table: Table, equivalence_dict: dict[str, str]) → Table[source]#

Apply column name mapping to a PyArrow table using an equivalence dictionary.

This function transforms column names from source format to standardized format by applying the reverse mapping of the provided equivalence dictionary. If no equivalence dictionary is provided, the original table is returned unchanged.

Parameters:

  • table (pa.Table) – The PyArrow table whose column names need to be mapped.

  • equivalence_dict (dict[str, str]) – Dictionary mapping standard column names to source column names. The function will reverse this mapping internally.

Returns:

A new PyArrow table with mapped column names. Columns not found in the mapping retain their original names.

Return type:

pa.Table

Examples

>>> import pyarrow as pa
>>> table = pa.table({'old_name': [1, 2, 3], 'other_col': [4, 5, 6]})
>>> equiv_dict = {'new_name': 'old_name'}
>>> result = apply_column_mapping(table, equiv_dict)
>>> result.column_names
['new_name', 'other_col']
check_for_empty_values(col: ChunkedArray, column_name: str, ticker: str) → None[source]#

Validate a PyArrow column for data quality issues including empty and null values.

This function performs comprehensive data quality checks on a column, identifying empty strings, null values, and whitespace-only entries. If any issues are found, a detailed ValueError is raised with specific row indices and error descriptions.

Parameters:

  • col (pa.ChunkedArray) – The PyArrow chunked array column to validate.

  • column_name (str) – The name of the column being validated, used in error messages.

  • ticker (str) – The ticker symbol associated with the data, used in error messages for better context and debugging.

Returns:

This function doesn’t return a value but raises an exception if issues are found.

Return type:

None

Raises:

ValueError – If any data quality issues are detected, including: - Empty strings (‘’) - Null/NaN values - Whitespace-only strings (e.g., ‘ ‘) The exception message includes the ticker, column name, and specific row indices where issues were found.

Examples

>>> import pyarrow as pa
>>> col = pa.chunked_array([['valid', '', None, '  ']])
>>> check_for_empty_values(col, 'price', 'AAPL')
ValueError: Data quality issues in ticker 'AAPL', column 'price':
Empty strings at rows: [1]
Null values at rows: [2]
Whitespace-only at rows: [3]
filter_by_date_range(table: Table, start_date: date | Literal['auto'] | None, end_date: date | Literal['auto'] | None) → Table[source]#

Filter a PyArrow table to include only rows within the specified date range.

This function filters the input table based on a date column named ‘date’, keeping only rows where the date falls within the specified range (inclusive). If both start_date and end_date are None, the original table is returned unchanged.

Parameters:

  • table (pa.Table) – The PyArrow table to filter. Must contain a column named ‘date’.

  • start_date (datetime.date or {"auto"} or None) – The earliest date to include in the filtered result. Can be None or “auto” to indicate no lower bound. Should be compatible with pa.date32() type.

  • end_date (datetime.date or {"auto"} or None) – The latest date to include in the filtered result. Can be None or “auto” to indicate no upper bound. Should be compatible with pa.date32() type.

Returns:

A new PyArrow table containing only rows where the date column falls within the specified range (inclusive of both bounds).

Return type:

pa.Table

Notes

  • The table must have a column named ‘date’

  • Date comparisons are inclusive (>= start_date and <= end_date)

  • If only one bound is specified, filtering is applied only for that bound

  • Dates are converted to pa.date32() format for comparison

Examples

>>> import pyarrow as pa
>>> from datetime import date
>>> table = pa.table({
...     'date': [date(2023, 1, 1), date(2023, 6, 15), date(2023, 12, 31)],
...     'value': [100, 200, 300]
... })
>>> filtered = filter_by_date_range(table, date(2023, 6, 1), date(2023, 12, 1))
>>> len(filtered)
1  # Only the June 15 row remains
get_decimal_type(precision: int, scale: int, precision_limit: int) → DataType[source]#

Return the appropriate PyArrow decimal type for the given precision.

Select decimal128 when precision fits within precision_limit, otherwise fall back to decimal256.

Parameters:

  • precision (int) – Total number of significant digits.

  • scale (int) – Number of digits after the decimal point.

  • precision_limit (int) – Upper bound for decimal128; values above this threshold use decimal256.

Returns:

Either pa.decimal128(precision, scale) or pa.decimal256(precision, scale).

Return type:

pa.DataType

get_reverse_equivalence_dict(equivalence_dict: dict[str, str]) → dict[str, str][source]#

Create a reverse mapping dictionary by swapping keys and values.

This function takes an equivalence dictionary that maps standard column names to source column names and returns a new dictionary with the mapping reversed, allowing lookup from source names to standard names.

Parameters:

equivalence_dict (dict[str, str]) – A dictionary mapping standard column names (keys) to source column names (values).

Returns:

A new dictionary with reversed key-value pairs, mapping source column names (keys) to standard column names (values).

Return type:

dict[str, str]

Examples

>>> equiv_dict = {'standard_name': 'source_name', 'price': 'Price_USD'}
>>> reversed_dict = get_reverse_equivalence_dict(equiv_dict)
>>> reversed_dict
{'source_name': 'standard_name', 'Price_USD': 'price'}

Notes

This function assumes that all values in the input dictionary are unique. If duplicate values exist, only the last key-value pair will be preserved in the reversed dictionary.

infer_decimal_precision_and_scale(col: ChunkedArray) → DecimalPrecisionScale[source]#

Infer the optimal precision and scale for decimal representation of numeric data.

This function analyzes a PyArrow column containing numeric data (as strings) to determine the appropriate precision (total number of digits) and scale (number of decimal places) for decimal type conversion. It handles edge cases such as maximum precision limits and ensures valid decimal type constraints.

Parameters:

col (pa.ChunkedArray) – A PyArrow chunked array containing numeric data as strings (e.g., “123.45”, “67.890”). The function expects string representation of numbers with optional decimal points.

Returns:

A result object containing: - precision: Total number of significant digits (1 to MAX_PRECISION) - scale: Number of digits after the decimal point (0 to precision)

Return type:

DecimalPrecisionScale

Notes

  • MAX_PRECISION constant defines the maximum allowed precision (76)

  • If inferred precision exceeds MAX_PRECISION, scale is reduced accordingly

  • Minimum precision is 1, and scale cannot exceed precision

  • Numbers without decimal points are treated as having scale 0

  • The function uses PyArrow compute functions for efficient processing

Examples

>>> import pyarrow as pa
>>> col = pa.chunked_array([["123.45", "67.890", "1000"]])
>>> result = infer_decimal_precision_and_scale(col)
>>> result.precision, result.scale
(6, 3)  # 6 total digits, 3 decimal places (from "67.890")

>>> # Using result to create decimal type
>>> decimal_type = pa.decimal128(result.precision, result.scale)
Raises:

pyarrow.ArrowInvalid – If PyArrow compute operations fail on invalid input data.

parse_date_column(date_col: ChunkedArray) → ChunkedArray[source]#

Parse a date column with strict ISO format validation and comprehensive error handling.

This function attempts to parse date strings using only ISO-compliant formats: YYYY-MM-DD and YYYY/MM/DD. It employs multiple parsing strategies and provides detailed error messages with sample data when parsing fails, making it easier to diagnose data format issues.

Parameters:

date_col (pa.ChunkedArray) – A PyArrow chunked array containing date strings to be parsed. Expected formats are ISO-compliant date strings.

Returns:

A PyArrow chunked array with dates converted to pa.date32() type, suitable for date-based operations and filtering.

Return type:

pa.ChunkedArray

Raises:

DateParsingError – If the date column cannot be parsed using any of the supported ISO formats. The exception includes: - Data type information - Array length - Sample values from the column - List of supported formats - Specific parsing error details

Notes

  • Only ISO date formats are accepted for data consistency:

    • YYYY-MM-DD (ISO format with hyphens, e.g., “2023-12-25”)

    • YYYY/MM/DD (ISO format with slashes, e.g., “2023/12/25”)

  • The function logs debug information during parsing attempts

  • Error messages include sample data to help identify format issues

  • Parsed dates are returned as pa.date32() type for memory efficiency

Examples

>>> import pyarrow as pa
>>> date_col = pa.chunked_array([["2023-01-15", "2023/06/30", "2023-12-25"]])
>>> parsed_dates = parse_date_column(date_col)
>>> parsed_dates.type
date32[day]
process_table_columns(*, table: Table, ticker: str, configuration: Configuration, equivalence_dict: dict[str, str]) → Table[source]#

Process raw table columns into standardized, typed, and date-filtered output.

Consolidate the common processing pipeline shared by CSV, Parquet, and other input handlers: parse date columns, validate and cast numeric columns to decimal types, rename to standard field names, and filter by date range.

Parameters:

  • table (pa.Table) – Raw input table with columns as strings.

  • ticker (str) – Ticker symbol, used in validation error messages.

  • configuration (Configuration) – Configuration containing start_date and end_date for date range filtering.

  • equivalence_dict (dict[str, str]) – Mapping from standard field names to the actual column names present in table.

Returns:

Processed table with standard column names, typed columns (dates as date32, numerics as decimal256), and rows filtered to the configured date range.

Return type:

pa.Table

Raises:

  • DateParsingError – If the date column cannot be parsed.

  • ValueError – If data quality checks fail on any column.