"""
Loads and returns a Configuration entity from an Excel file
Functions
---------
init:
Loads and returns Configuration
"""
import logging
import pathlib
import sys
import types
import typing
import openpyxl
import openpyxl.cell
import openpyxl.utils.exceptions
import openpyxl.worksheet.cell_range
import openpyxl.worksheet.worksheet
import packaging.version
from kaxanuk.attribution_analysis import __parameters_format_version__
from kaxanuk.attribution_analysis.config_handlers.configurator_interface import ConfiguratorInterface
from kaxanuk.attribution_analysis.entities.configuration import Configuration
from kaxanuk.attribution_analysis.exceptions import (
ConfigurationError,
ConfigurationHandlerError,
)
[docs]
class ExcelConfigurator(ConfiguratorInterface):
"""
Build a `Configuration` and dashboard port from an Excel parameters file.
On construction, the workbook at `file_path` is loaded and its
``General`` sheet is parsed according to `SHEET_KEY_VALUES`. The
``parameters_format_version`` cell is checked against the package's
`__parameters_format_version__`, method-flag cells (e.g.
``brinson_fachler_method``) are coerced to booleans, and the
``dashboard_port`` cell is validated as an integer in range. Any
`ConfigurationError` or `ConfigurationHandlerError` raised while
parsing is logged as critical and terminates the process via
`sys.exit`, rather than propagating to the caller.
Parameters
----------
file_path : str
Path to the Excel configuration file.
logger_format : str
Format string injected into the temporary bootstrap logger used
while parsing `file_path`.
Attributes
----------
METHOD_FLAG_VALUES : Mapping[str, bool]
Accepted case-insensitive text values (``'yes'``/``'no'``) for
method-flag fields, mapped to their boolean equivalent.
SHEET_KEY_VALUES : Mapping[str, tuple[str, ...]]
Schema of required sheets and the key values to extract from
each, passed to `_extract_workbook_key_values_by_schema`.
"""
METHOD_FLAG_VALUES = types.MappingProxyType({
'yes': True,
'no': False,
})
SHEET_KEY_VALUES = types.MappingProxyType({
'General': (
'input_directory',
'weights_portfolio_directory',
'weights_benchmark_directory',
'benchmark_returns_directory',
'investable_assets_directory',
'factor_returns_by_factor_directory',
'user_column_date',
'user_column_price',
'market_data_input_format',
'portfolio_input_format',
'start_date',
'end_date',
'portfolio_file_name',
'benchmark_file_name',
'benchmark_return_file_name',
'brinson_fachler_method',
'factor_model_method',
'dashboard_port',
'parameters_format_version',
),
})
def __init__(
self,
file_path: str,
logger_format: str = "[%(levelname)s] %(message)s",
):
"""
Initialize configuration from a performance attribution Excel parameters file.
Parameters
----------
file_path
Path to the Excel configuration file
logger_format
Format string injected into the temporary bootstrap logger
"""
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(logger_format))
logger.addHandler(handler)
try:
workbook = self._load_file(file_path)
sheet_key_values = self._extract_workbook_key_values_by_schema(
workbook,
self.SHEET_KEY_VALUES,
)
general = sheet_key_values['General']
current_format_version = self._validate_parameters_format_version(
general['parameters_format_version']
)
if (
packaging.version.parse(current_format_version)
> packaging.version.parse(__parameters_format_version__)
):
logger.warning(
"Excel configuration is NEWER than this package supports "
"(Excel: %s, package expects: %s). Unexpected behavior may occur "
"— consider upgrading kaxanuk-attribution_analysis.",
current_format_version,
__parameters_format_version__,
)
else:
logger.info("Excel parameters_format_version: %s", current_format_version)
self._configuration = Configuration(
input_directory=general['input_directory'],
weights_portfolio_directory=general['weights_portfolio_directory'],
weights_benchmark_directory=general['weights_benchmark_directory'],
benchmark_returns_directory=general['benchmark_returns_directory'],
investable_assets_directory=general['investable_assets_directory'],
factor_returns_by_factor_directory=general['factor_returns_by_factor_directory'],
user_column_date=general['user_column_date'],
user_column_price=general['user_column_price'],
market_data_input_format=general['market_data_input_format'],
portfolio_input_format=general['portfolio_input_format'],
start_date=general['start_date'],
end_date=general['end_date'],
portfolio_file_name=general['portfolio_file_name'],
benchmark_file_name=general['benchmark_file_name'],
benchmark_return_file_name=general['benchmark_return_file_name'],
brinson_fachler_method=self._parse_method_flag(
general['brinson_fachler_method'],
'brinson_fachler_method',
),
factor_model_method=self._parse_method_flag(
general['factor_model_method'],
'factor_model_method',
),
)
self._dashboard_port = self._validate_dashboard_port(
general['dashboard_port']
)
logger.handlers.clear()
except (
ConfigurationError,
ConfigurationHandlerError,
) as error:
msg = f"An error was encountered when parsing your configuration file: {error!s}"
logging.getLogger(__name__).critical(msg)
sys.exit()
[docs]
def get_configuration(self) -> Configuration:
"""
Return the configuration parsed from the Excel parameters file.
Returns
-------
Configuration
The configuration built during `__init__`.
"""
return self._configuration
[docs]
def get_dashboard_port(self) -> int:
"""
Return the dashboard port parsed from the Excel configuration.
Returns
-------
int
A valid port number (1-9999).
"""
return self._dashboard_port
@staticmethod
def _extract_cell_value(cell: openpyxl.cell.cell.Cell) -> str | None:
"""
Extract the value of an openpyxl cell as a string.
Parameters
----------
cell
The cell object to extract the value from.
Returns
-------
The stripped value of the cell. If the cell is empty returns None.
"""
if (
cell is not None
and cell.value is not None
):
return str(cell.value).strip()
else:
return None
@classmethod
def _extract_column_values(
cls,
column: tuple[openpyxl.cell.cell.Cell, ...]
) -> list[str]:
"""
Extract the values of an openpyxl column as a string list.
Parameters
----------
column : openpyxl.worksheet.cell_range.CellRange
The column object to extract the values from.
Returns
-------
list[str]
The values as strings.
"""
values = filter(
None,
(
cls._extract_cell_value(i)
for i in column
)
)
return list(values)
@classmethod
def _extract_workbook_key_values_by_schema(
cls,
workbook: openpyxl.workbook.workbook.Workbook,
schema: typing.Mapping[str, tuple[str, ...]],
key_column: str = 'A',
) -> dict[str, dict[str, typing.Any]]:
"""
Extract the values of particular key values specified in the schema from a workbook's sheets.
Parameters
----------
workbook
The workbook to search
schema
The schema indicating the required sheets and the key values to be extracted from them
key_column
The letter identifier of the column to search for the key
Returns
-------
The extracted values, in the same arrangement as the schema
Raises
------
ConfigurationHandlerError
"""
value_column = cls._increment_column_identifier(key_column)
missing_sheets = set(schema.keys()).difference(set(workbook.sheetnames))
if len(missing_sheets) > 0:
msg = " ".join([
"The following sheets are missing from the Configuration file:",
", ".join(missing_sheets)
])
raise ConfigurationHandlerError(msg)
sheets = {}
missing_sheet_fields = {}
for sheet_name in schema:
sheets[sheet_name] = {}
missing_fields = []
for field_name in schema[sheet_name]:
row = cls._find_sheet_row_by_column_value(
workbook[sheet_name],
key_column,
field_name,
)
if row is None:
missing_fields.append(field_name)
else:
sheets[sheet_name][field_name] = workbook[sheet_name][f'{value_column}{row}'].value
if len(missing_fields) > 0:
missing_sheet_fields[sheet_name] = missing_fields
if len(missing_sheet_fields) > 0:
msg = " ".join([
"The following sheet params are missing from the Configuration file:",
"; ".join([
f"{sheet}: " + ", ".join(fields)
for sheet, fields in missing_sheet_fields.items()
])
])
raise ConfigurationHandlerError(msg)
return sheets
@staticmethod
def _find_sheet_row_by_column_value(
sheet: openpyxl.worksheet.worksheet.Worksheet,
column: str,
search_value: str,
) -> int | None:
"""
Find the row location of the search_value in the column.
Parameters
----------
sheet
The sheet in which to search
column
The letter identifier of the column where to search
search_value
the value to be searched for in the column
Returns
-------
The number of the row where the value was found, or None if not found
"""
found_row = None
for row in range(1, sheet.max_row + 1):
cell_name = f"{column}{row}"
if (
sheet[cell_name].value is not None
and sheet[cell_name].value.strip() == search_value
):
found_row = row
break
return found_row
@classmethod
def _increment_column_identifier(cls, column_identifier: str) -> str:
"""
Get the letter identifier of the next column after column_identifier.
Uses Excel's column identifier scheme of successive uppercase letters.
AA follows Z, and so on.
Parameters
----------
column_identifier
The column identifier to increment
Returns
-------
The next column letter identifier
"""
if column_identifier == 'Z':
return 'AA'
elif column_identifier[-1] == 'Z':
return cls._increment_column_identifier(column_identifier[:-1]) + 'A'
else:
next_character = chr(ord(column_identifier[-1]) + 1)
return column_identifier[:-1] + next_character
@classmethod
def _parse_method_flag(cls, raw: typing.Any, field_name: str) -> bool:
"""
Coerce a method-flag cell value into a boolean.
Accepts a real Excel boolean (``VERDADERO``/``FALSO`` -> ``True``/``False``)
for backward compatibility, or the case-insensitive text ``'yes'``/``'no'``.
Any other value raises an error.
Parameters
----------
raw
The raw cell value read from the workbook.
field_name
The parameter name, used in error messages.
Returns
-------
bool
``True`` for ``VERDADERO``/``'yes'``, ``False`` for ``FALSO``/``'no'``.
Raises
------
ConfigurationHandlerError
If the value is neither a boolean nor ``'yes'``/``'no'``.
"""
if isinstance(raw, bool):
return raw
key = str(raw).strip().lower() if raw is not None else ''
if key not in cls.METHOD_FLAG_VALUES:
msg = f"'{field_name}' must be 'yes' or 'no' (got: {raw!r})"
raise ConfigurationHandlerError(msg)
return cls.METHOD_FLAG_VALUES[key]
@staticmethod
def _validate_parameters_format_version(raw: typing.Any) -> str:
"""
Validate the Excel ``parameters_format_version`` against the package.
Parameters
----------
raw
The raw value read from the ``parameters_format_version`` cell.
Returns
-------
str
The (stripped) version string found in the Excel file.
Raises
------
ConfigurationHandlerError
If the value is missing/empty, or older than the version this
package supports (``__parameters_format_version__``).
"""
current = str(raw).strip() if raw is not None else ""
expected = __parameters_format_version__
if not current:
msg = (
"Excel configuration file is missing the 'parameters_format_version' "
f"value in the General sheet. Expected version: {expected}. "
"Please run 'kaxanuk.attribution_analysis update excel' to get the latest template."
)
raise ConfigurationHandlerError(msg)
if packaging.version.parse(current) < packaging.version.parse(expected):
msg = (
f"Excel configuration file uses an outdated format "
f"(found: {current}, expected: {expected}). "
"Please run 'kaxanuk.attribution_analysis update excel' to get the latest template."
)
raise ConfigurationHandlerError(msg)
return current
@classmethod
def _validate_dashboard_port(cls, port_value: typing.Any) -> int:
"""
Validate and convert the dashboard port value.
Parameters
----------
port_value
The raw port value from the Excel configuration.
Returns
-------
int
The validated port number.
Raises
------
ConfigurationHandlerError
If the port is not a valid integer or outside the range
``DASHBOARD_PORT_MIN``-``DASHBOARD_PORT_MAX``.
"""
try:
port = int(port_value)
except (ValueError, TypeError) as err:
msg = f"Invalid dashboard_port '{port_value}' in parameters file. Must be an integer."
raise ConfigurationHandlerError(msg) from err
if port < cls.DASHBOARD_PORT_MIN or port > cls.DASHBOARD_PORT_MAX:
msg = (
f"Invalid dashboard_port {port} in parameters file. "
f"Must be between {cls.DASHBOARD_PORT_MIN} and {cls.DASHBOARD_PORT_MAX}."
)
raise ConfigurationHandlerError(msg)
return port
@staticmethod
def _load_file(file_path: str) -> openpyxl.Workbook:
"""
Load an Excel file.
Parameters
----------
file_path
The path of the Excel file to load
Returns
-------
The openpyxl Workbook corresponding to the file
Raises
------
ConfigurationHandlerError
"""
if not pathlib.Path(file_path).is_file():
msg = f"Configuration file not found in path: {file_path}"
raise ConfigurationHandlerError(msg)
try:
workbook = openpyxl.load_workbook(file_path)
except openpyxl.utils.exceptions.InvalidFileException as error:
msg = f"Invalid Configuration file in path: {file_path}"
raise ConfigurationHandlerError(msg) from error
return workbook
