Source code for kaxanuk.attribution_analysis.input_handlers.excel_portfolio_input_handler
"""
Excel portfolio input handler for reading .xlsx portfolio allocation files.
Read portfolio weight files in Excel format, supporting both horizontal
(Ticker | Date1 | Date2 | ...) and vertical (Date | Ticker1 | Ticker2 | ...)
layouts with automatic format detection, transposition, and decimal precision
inference.
"""
from pathlib import Path
import logging
import pandas as pd
import pyarrow as pa
import pyarrow.compute as pc
from kaxanuk.attribution_analysis.exceptions import MissingPortfolioError, ReadingExcelPortfolioError
from kaxanuk.attribution_analysis.input_handlers import input_handlers_helpers
from kaxanuk.attribution_analysis.portfolio_data_pipeline import portfolio_transformer
from kaxanuk.attribution_analysis.input_handlers.portfolio_input_handler_interface import PortfolioInputHandlerInterface
logger = logging.getLogger(__name__)
PRECISION = 38
[docs]
class ExcelPortfolioInputHandler(PortfolioInputHandlerInterface):
"""
Professional Excel portfolio file handler for financial systems.
This class provides a robust interface for reading and processing Excel-based portfolio
files (.xlsx format). It automatically handles data type inference, decimal precision
detection for financial data, and converts portfolio information into optimized
PyArrow tables for high-performance analytics.
Supports both horizontal and vertical formats:
- Horizontal: Ticker | Date1 | Date2 | ...
- Vertical: Date | Ticker1 | Ticker2 | ... (auto-transposes to horizontal)
Attributes
----------
base_dir : pathlib.Path
The base directory path where Excel portfolio files are stored.
"""
def __init__(
self,
base_dir: str
) -> None:
"""
Initialize the Excel portfolio input handler with specified base directory.
Parameters
----------
base_dir : str
The absolute or relative path to the base directory containing Excel
portfolio files.
"""
self.base_dir = Path(base_dir)
[docs]
def read_portfolio(
self,
*,
file_name: str
) -> pa.Table:
"""
Read and process an Excel portfolio file into a typed PyArrow table.
This method performs comprehensive processing of Excel portfolio files, including
automatic file extension handling, data type inference, decimal precision detection,
and conversion to an optimized PyArrow table format.
Automatically detects horizontal vs vertical format and returns horizontal format.
Parameters
----------
file_name : str
The name of the Excel portfolio file to read. Can be provided with or
without the .xlsx extension.
Returns
-------
pyarrow.Table
A typed PyArrow table containing the processed portfolio data in horizontal
format.
Raises
------
MissingPortfolioError
If the file does not exist in the specified base directory.
ReadingExcelPortfolioError
If the Excel file cannot be read or processed correctly.
"""
file_path = (
self.base_dir / file_name
if file_name.lower().endswith(".xlsx")
else self.base_dir / f"{file_name}.xlsx"
)
if not file_path.exists():
raise MissingPortfolioError(
f"Portfolio file not found in {self.base_dir}: {file_path.name}"
)
new_columns: list[pa.ChunkedArray] = []
csv_column_names: list[str] = []
try:
# Read every cell as a string directly (nulls are preserved as
# nulls, not the literal "nan"), so no pandas type inference is
# wasted on values only to be cast back to string afterwards.
df = pd.read_excel(file_path, dtype=str)
# Detect format using helper
first_col = df.columns[0]
# Handle Timestamp columns (could be dates)
if isinstance(first_col, pd.Timestamp):
first_col = "Date"
format_type = portfolio_transformer.detect_portfolio_format(
first_column_name=str(first_col),
error_class=ReadingExcelPortfolioError
)
logger.info("Detected %s format for %s", format_type, file_path.name)
prem_table = pa.Table.from_pandas(df, preserve_index=False)
if format_type == "vertical":
# Transpose to horizontal format
prem_table = portfolio_transformer.transpose_vertical_to_horizontal(
table=prem_table
)
elif prem_table.column_names[0].lower() in ["ticker"]:
# Normalize first column to "Ticker" for horizontal format
old_names = prem_table.column_names
new_names = ["Ticker", *old_names[1:]]
prem_table = prem_table.rename_columns(new_names)
# Guard: after format detection and normalization, first column must be "Ticker"
if prem_table.column_names[0] != "Ticker":
raise ReadingExcelPortfolioError(
f"First column of portfolio must be named 'Ticker' "
f"(found: '{prem_table.column_names[0]}')"
)
# Process columns with decimal precision
for column_name in prem_table.column_names:
csv_column_names.append(column_name)
col = prem_table[column_name]
if column_name == "Ticker":
# pa.Table.from_pandas yields large_string; normalize to string
new_columns.append(pc.cast(col, pa.string()))
else:
decimal_spec = input_handlers_helpers.infer_decimal_precision_and_scale(col=col)
target_type = input_handlers_helpers.get_decimal_type(
precision=decimal_spec.precision,
scale=decimal_spec.scale,
precision_limit=PRECISION,
)
new_columns.append(pc.cast(col, target_type))
return pa.table(new_columns, names=csv_column_names)
except ReadingExcelPortfolioError:
raise
except FileNotFoundError as error:
raise ReadingExcelPortfolioError(
f"Portfolio Excel file not found: {file_path}"
) from error
except pa.lib.ArrowInvalid as error:
raise ReadingExcelPortfolioError(
f"Invalid data format in Excel file {file_path.name}: {error}"
) from error
except (ValueError, KeyError, TypeError) as error:
raise ReadingExcelPortfolioError(
f"Error parsing Excel portfolio file {file_path.name}: {error}"
) from error
except OSError as error:
raise ReadingExcelPortfolioError(
f"I/O error reading Excel file {file_path.name}: {error}"
) from error