"""
Portfolio entity for managing asset weight allocations over time.
This module provides :class:`PortfolioEntity`, an immutable dataclass
that stores portfolio weights in a PyArrow table. The first column
holds ticker symbols and every subsequent column represents weights at
a rebalancing date. Comprehensive validation ensures date formats,
non-negative weights, per-date weight sums, and data-type correctness.
"""
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
import pyarrow as pa
import pyarrow.compute as pc
from kaxanuk.attribution_analysis.exceptions import PortfolioEntityError
from kaxanuk.attribution_analysis.portfolio_data_pipeline.date_utils import (
parse_date_with_formats,
DATE_FORMATS,
)
MIN_DATE_COLUMNS = 2
[docs]
@dataclass(frozen=True, slots=True)
class PortfolioEntity:
"""
A portfolio entity that manages asset weights over time using PyArrow tables.
This class represents a portfolio with asset allocations that can change over time.
It stores portfolio data in a PyArrow table format where the first column contains
ticker symbols and subsequent columns represent portfolio weights at different dates.
The class ensures data integrity through comprehensive validation including:
- Date format validation (YYYY-MM-DD or DD/MM/YYYY)
- Weight validation (non-negative, sum <= 1.0 per date)
- Data type validation for all columns
- Null value detection and prevention
Attributes
----------
portfolio_table : pa.Table
PyArrow table containing portfolio data with ticker symbols in the
first column and date-based weights in subsequent columns.
Examples
--------
>>> weights = {
... "2020-01-01": {"AAPL": 0.6, "MSFT": 0.4},
... "2020-06-01": {"AAPL": 0.5, "MSFT": 0.5}
... }
>>> portfolio = PortfolioEntity.from_dict(weights)
>>> tickers = portfolio.get_tickers()
>>> dates = portfolio.get_date_columns()
"""
portfolio_table: pa.Table
def __post_init__(self) -> None:
"""
Validate the portfolio table after initialization.
Performs comprehensive validation of the portfolio table structure and data
by delegating to specialized validation methods.
Raises
------
PortfolioEntityError
If any validation fails.
Notes
-----
This method is automatically called after dataclass initialization
due to the @dataclass decorator.
"""
self._validate_table_structure()
self._validate_ticker_column()
self._validate_date_columns_format()
self._validate_weights_data()
@staticmethod
def _check_null_values(
column: pa.Array,
col_name: str
) -> None:
"""
Check for null values in a column.
Parameters
----------
column : pa.Array
PyArrow array to check for null values.
col_name : str
Name of the column being validated (for error messages).
Raises
------
PortfolioEntityError
If any null values are found in the column.
"""
if pc.any(pc.is_null(column)).as_py():
raise PortfolioEntityError(
f"Null value found in column {col_name} (date). Nulls are not allowed."
)
@staticmethod
def _validate_date_format(
date_str: str
) -> None:
"""
Validate a single date string format.
Centralizes date format validation to avoid code duplication.
Used by both dictionary validation and table column validation.
Parameters
----------
date_str : str
Date string to validate.
Raises
------
PortfolioEntityError
If date format is invalid.
"""
parse_date_with_formats(
date_str=date_str,
formats=DATE_FORMATS
)
@staticmethod
def _convert_dict_to_pyarrow_table(
validated_dict: dict[str, dict[str, Decimal]]
) -> pa.Table:
"""
Convert a validated and normalized weights dictionary to PyArrow Table format.
This method transforms the validated dictionary structure into a PyArrow table
optimized for efficient storage and computation. The resulting table structure
facilitates fast column-based operations on portfolio data.
Table structure:
- First column ('ticker'): Contains all ticker symbols as strings
- Subsequent columns: Each date becomes a column with corresponding weights as float64
- Rows: Each row represents a ticker with its weights across all dates
- Column ordering: Dates are sorted alphabetically for consistency
Parameters
----------
validated_dict : dict[str, dict[str, Decimal]]
Pre-validated dictionary with date keys and ticker-weight mappings.
All tickers must be present for each date (normalized by
`_validate_and_normalize_weights_dict`).
Example: {"2020-01-01": {"AAPL": Decimal("0.6"), "MSFT": Decimal("0.4")}}
Returns
-------
pa.Table
PyArrow table with ticker column followed by date columns.
Weights are converted from Decimal to float64 for PyArrow compatibility.
Table schema: [ticker: string, date1: float64, date2: float64, ...]
Raises
------
PortfolioEntityError
If the dictionary is empty or PyArrow table creation fails.
Examples
--------
Input: {"2020-01-01": {"AAPL": Decimal("0.6"), "MSFT": Decimal("0.4")}}
Output: PyArrow table with columns ['ticker', '2020-01-01']
and rows [['AAPL', 0.6], ['MSFT', 0.4]]
Notes
-----
Decimal values are converted to float64 for PyArrow compatibility.
This conversion maintains sufficient precision for financial calculations.
"""
if not validated_dict:
raise PortfolioEntityError("Cannot create table from empty dictionary")
# Get sorted tickers and dates for consistent ordering
first_date_weights = next(iter(validated_dict.values()))
tickers = sorted(first_date_weights.keys())
dates = sorted(validated_dict.keys())
# Create the data structure for PyArrow
# First column: tickers
table_data = {"ticker": tickers}
# Subsequent columns: weights for each date
for date_str in dates:
weights_for_date = []
for ticker in tickers:
weight = validated_dict[date_str][ticker]
# Convert Decimal to float for PyArrow compatibility
weights_for_date.append(float(weight))
table_data[date_str] = weights_for_date
# Create PyArrow table
try:
portfolio_table = pa.table(table_data)
except (pa.ArrowInvalid, pa.ArrowTypeError, pa.ArrowNotImplementedError) as e:
raise PortfolioEntityError(f"Failed to create PyArrow table: {e!s}") from e
return portfolio_table
@staticmethod
def _validate_and_normalize_weights_dict(
weights_dict: dict[str, dict[str, float]]
) -> dict[str, dict[str, Decimal]]:
"""
Validate and normalize the input weights dictionary to a consistent format.
This method performs comprehensive validation and normalization of the input
dictionary before conversion to PyArrow table format. It ensures data integrity
and consistency across all portfolio dates and tickers.
Validation includes:
- Dictionary structure validation
- Date format validation (YYYY-MM-DD or YYYY/MM/DD)
- Ticker name validation and normalization (uppercase, trimmed)
- Weight value validation (non-negative, convertible to Decimal)
- Weight sum validation (≤ 1.0 per date)
- Duplicate date detection (case-insensitive)
- Ticker consistency across all dates (missing tickers filled with 0.0)
Parameters
----------
weights_dict : dict[str, dict[str, float]]
Dictionary with date keys and nested dictionaries of ticker-weight
pairs. Example: {"2020-01-01": {"AAPL": 0.6, "MSFT": 0.4}}
Returns
-------
dict[str, dict[str, Decimal]]
Validated and normalized dictionary with all tickers present for
each date (missing ones set to 0.0), ticker names normalized to
uppercase, and weights converted to Decimal.
Raises
------
PortfolioEntityError
If validation fails for any reason including:
- Empty or invalid dictionary structure
- Invalid date formats
- Invalid ticker names (empty or non-string)
- Invalid weight values (negative, non-numeric, or sum > 1.0)
- Duplicate dates (case-insensitive)
Notes
-----
This method performs the same core validations as `__post_init__`
but operates on dictionary format before PyArrow table conversion.
This validation is used when creating PortfolioEntity via `from_dict`.
For direct table creation, date validation occurs in
`_validate_date_columns_format`.
"""
if not isinstance(weights_dict, dict):
raise PortfolioEntityError("weights_dict must be a dictionary")
if len(weights_dict) == 0:
raise PortfolioEntityError("weights_dict cannot be empty")
# Validate date formats and collect all unique tickers
all_tickers = set()
validated_dates = {}
for date_str, ticker_weights in weights_dict.items():
# Validate date format
if not isinstance(date_str, str):
raise PortfolioEntityError(f"Date key must be string, found: {type(date_str)}")
# Use centralized date validation
PortfolioEntity._validate_date_format(
date_str=date_str
)
# Validate ticker weights
if not isinstance(ticker_weights, dict):
raise PortfolioEntityError(f"Weights for date '{date_str}' must be a dictionary")
if len(ticker_weights) == 0:
raise PortfolioEntityError(f"No tickers found for date '{date_str}'")
validated_weights = {}
total_weight = Decimal('0')
for raw_ticker, weight in ticker_weights.items():
# Validate ticker name
if not isinstance(raw_ticker, str) or not raw_ticker.strip():
raise PortfolioEntityError(f"Invalid ticker name: '{raw_ticker}' for date '{date_str}'")
ticker = raw_ticker.strip().upper() # Normalize ticker format
# Validate and convert weight
try:
decimal_weight = Decimal(str(weight))
except (InvalidOperation, PortfolioEntityError) as e:
raise PortfolioEntityError(
f"Invalid weight value '{weight}' for ticker '{ticker}' on date '{date_str}': {e!s}"
) from e
# Check for negative weights
if decimal_weight < 0:
raise PortfolioEntityError(
f"Negative weight detected for ticker '{ticker}' on date '{date_str}': {decimal_weight}"
)
validated_weights[ticker] = decimal_weight
total_weight += decimal_weight
all_tickers.add(ticker)
# Validate total weights for this date
if total_weight > Decimal("1.000001"):
raise PortfolioEntityError(
f"Total weights for date '{date_str}' exceed 1.0: {total_weight:.6f}"
)
validated_dates[date_str] = validated_weights
# Ensure all dates have the same tickers (fill missing ones with 0)
normalized_dict = {}
sorted_tickers = sorted(all_tickers) # Consistent ordering
for date_str, ticker_weights in validated_dates.items():
normalized_weights = {}
for ticker in sorted_tickers:
normalized_weights[ticker] = ticker_weights.get(ticker, Decimal('0'))
normalized_dict[date_str] = normalized_weights
# Check for duplicate dates (case-insensitive)
date_check = set()
for date_str in normalized_dict:
date_lower = date_str.lower()
if date_lower in date_check:
raise PortfolioEntityError(f"Duplicate date found: '{date_str}'")
date_check.add(date_lower)
return normalized_dict
@staticmethod
def _validate_column_weights(
column: pa.Array,
col_name: str
) -> None:
"""
Validate weights: convertible to Decimal, non-negative, sum <= 1.0.
Parameters
----------
column : pa.Array
PyArrow array containing weight values to validate.
col_name : str
Name of the column being validated (for error messages).
Raises
------
PortfolioEntityError
If weights are invalid (non-numeric, negative, or sum exceeds 1.0).
"""
try:
values = column.to_pylist()
decimal_weights = [Decimal(str(val)) for val in values]
except (InvalidOperation, PortfolioEntityError) as e:
raise PortfolioEntityError(
f"Invalid value found in column {col_name} for Decimal conversion: {e!s}"
) from e
for w in decimal_weights:
if w < 0:
raise PortfolioEntityError(
f"Negative weight detected in column {col_name}: {w}"
)
total = sum(decimal_weights)
if total > Decimal("1.000001"):
raise PortfolioEntityError(
f"Total weights for {col_name} exceed 1.0: {total:.6f}"
)
def _validate_date_columns_format(self) -> None:
"""
Validate date column formats.
Ensures all date columns follow the accepted formats (YYYY-MM-DD or YYYY/MM/DD).
Raises
------
PortfolioEntityError
If any date column has an invalid format.
Notes
-----
This validation is necessary when PortfolioEntity is created directly
with a pa.Table from files. When using `from_dict`, dates are pre-validated
in `_validate_and_normalize_weights_dict` using the same centralized validation.
"""
date_columns = self.portfolio_table.column_names[1:]
for col in date_columns:
# Use centralized date validation
self._validate_date_format(col)
def _validate_table_structure(self):
"""Validate the portfolio table has the minimum required date columns.
Raises
------
PortfolioEntityError
If fewer than ``MIN_DATE_COLUMNS`` rebalancing-date columns
are present (excluding the ticker column).
"""
date_column_count = self.portfolio_table.num_columns - 1
if date_column_count < MIN_DATE_COLUMNS:
raise PortfolioEntityError(
f"Portfolio table must have at least {MIN_DATE_COLUMNS} rebalance dates. "
f"Found {date_column_count}."
)
def _validate_ticker_column(self) -> None:
"""
Validate ticker column data type.
Ensures the first column contains string data for ticker symbols.
Raises
------
PortfolioEntityError
If the first column is not of string type.
"""
ticker_field = self.portfolio_table.schema.field(0)
if not pa.types.is_string(ticker_field.type):
raise PortfolioEntityError(
f"First column must contain tickers (string), found {ticker_field.type}."
)
def _validate_weights_data(self) -> None:
"""
Validate weight values and check for nulls.
Validates all date columns to ensure they contain valid weight data:
no null values, valid numeric values, and proper weight constraints.
Raises
------
PortfolioEntityError
If any weight data validation fails.
"""
date_columns = self.portfolio_table.column_names[1:]
for idx, col_name in enumerate(date_columns, start=1):
column = self.portfolio_table.column(idx)
self._check_null_values(column, col_name)
self._validate_column_weights(column, col_name)
[docs]
@classmethod
def from_dict(
cls,
weights_dict: dict[str, dict[str, float]]
) -> 'PortfolioEntity':
"""
Create PortfolioEntity from dictionary of weights with full validation.
Parameters
----------
weights_dict : dict[str, dict[str, float]]
Dictionary with date keys and ticker-weight mappings.
Example:
{
"2020-01-01": {"AAPL": 0.6, "MSFT": 0.4},
"2020-06-01": {"AAPL": 0.5, "MSFT": 0.5},
"2021-01-01": {"AAPL": 0.7, "MSFT": 0.3}
}
Returns
-------
PortfolioEntity
Validated PortfolioEntity instance.
Raises
------
PortfolioEntityError
If validation fails.
"""
if not weights_dict:
raise PortfolioEntityError("weights_dict cannot be empty")
# Validate and normalize input dictionary
validated_dict = cls._validate_and_normalize_weights_dict(
weights_dict=weights_dict
)
# Convert to PyArrow Table format
portfolio_table = cls._convert_dict_to_pyarrow_table(
validated_dict=validated_dict
)
# Create entity (triggers __post_init__ validation)
return cls(
portfolio_table=portfolio_table
)
[docs]
def get_date_columns(self) -> list[str]:
"""
Retrieve the list of date column names in the portfolio.
Returns all date column names from the portfolio table, excluding the first
column which contains ticker symbols. Each date column represents a specific
rebalancing date with corresponding portfolio weights.
Returns
-------
list[str]
A list of date column names in string format
(e.g., ['2020-01-01', '2020-06-01', '2021-01-01']).
The dates are in the format YYYY-MM-DD or DD/MM/YYYY as
originally provided during portfolio creation.
Examples
--------
>>> portfolio = PortfolioEntity.from_dict({
... "2020-01-01": {"AAPL": 0.6, "MSFT": 0.4},
... "2020-06-01": {"AAPL": 0.5, "MSFT": 0.5}
... })
>>> dates = portfolio.get_date_columns()
>>> print(dates)
['2020-01-01', '2020-06-01']
"""
return self.portfolio_table.column_names[1:]
[docs]
def get_tickers(self) -> list[str]:
"""
Retrieve the list of ticker symbols in the portfolio.
Returns all ticker symbols stored in the first column of the portfolio table.
The tickers are returned in the same order as they appear in the table.
Returns
-------
list[str]
A list of ticker symbols (e.g., ['AAPL', 'MSFT', 'GOOGL']).
The list maintains the original ordering from the portfolio table.
Examples
--------
>>> portfolio = PortfolioEntity.from_dict({
... "2020-01-01": {"AAPL": 0.6, "MSFT": 0.4}
... })
>>> tickers = portfolio.get_tickers()
>>> print(tickers)
['AAPL', 'MSFT']
"""
return self.portfolio_table.column(0).to_pylist()
[docs]
def get_weights_for_date_column(
self,
date_column: str
) -> list[float]:
"""
Retrieve portfolio weights for a specific date column.
Returns the portfolio weights for all tickers at the specified date.
The weights are returned in the same order as the tickers from get_tickers().
Parameters
----------
date_column : str
The date column name to retrieve weights for. Must be one of the
date columns returned by `get_date_columns`. Format should match
the original date format (YYYY-MM-DD or DD/MM/YYYY).
Returns
-------
list[float]
A list of portfolio weights corresponding to each ticker for the
specified date. The weights are in the same order as the tickers
returned by `get_tickers`.
Raises
------
PortfolioEntityError
If the specified date_column is not found in the portfolio.
Examples
--------
>>> portfolio = PortfolioEntity.from_dict({
... "2020-01-01": {"AAPL": 0.6, "MSFT": 0.4},
... "2020-06-01": {"AAPL": 0.5, "MSFT": 0.5}
... })
>>> weights = portfolio.get_weights_for_date_column("2020-01-01")
>>> tickers = portfolio.get_tickers()
>>> dict(zip(tickers, weights))
{'AAPL': 0.6, 'MSFT': 0.4}
"""
if date_column not in self.get_date_columns():
raise PortfolioEntityError(f"Date column '{date_column}' not found in portfolio")
col_index = self.portfolio_table.column_names.index(date_column)
return self.portfolio_table.column(col_index).to_pylist()
