"""
Immutable entity for portfolio / benchmark weight allocations as PyArrow tables.
This module provides :class:`PortfolioWeights`, a frozen dataclass that
wraps a ``pa.Table`` of asset weights in the transposed attribution
format (one column per asset, rows indexed by date). Schema and data
integrity validations run automatically on construction.
"""
import datetime
from dataclasses import dataclass
import pyarrow as pa
import pyarrow.compute as pc
from kaxanuk.attribution_analysis.exceptions import (
DataIntegrityError,
DataValidationError,
MissingColumnError,
EntityTypeError,
NullError,
)
DATE_COLUMN = "date"
MIN_ANNUAL_OBSERVATIONS = 240
MAX_ANNUAL_OBSERVATIONS = 260
[docs]
@dataclass(frozen=True, slots=True)
class PortfolioWeights:
"""
Immutable wrapper around a PyArrow table of portfolio or benchmark weights.
The expected schema is::
[date: timestamp(ns), ASSET1: float64, ASSET2: float64, ...]
Validation is performed automatically on construction via
``__post_init__`` and includes:
* The table must contain at least one row.
* A ``"date"`` column of type ``timestamp(ns)`` must be present.
* Every non-date column must be ``float64``.
* No null values are allowed in asset columns.
* If the table spans at least a year, the observation density near
the start and end of the series must be consistent with a daily
series (roughly 240-260 rows per year).
Parameters
----------
table : pa.Table
PyArrow table whose first column is a ``timestamp(ns)`` date
column named ``"date"`` and whose remaining columns hold
per-asset weight allocations as ``float64``.
Raises
------
DataValidationError
If the table is empty.
MissingColumnError
If the ``"date"`` column is absent.
EntityTypeError
If the ``"date"`` column is not a timestamp or any asset column
is not ``float64``.
NullError
If any asset column contains null values.
DataIntegrityError
If the table spans at least a year but the number of
observations in the first or last 1-year window falls outside
``[MIN_ANNUAL_OBSERVATIONS, MAX_ANNUAL_OBSERVATIONS]``,
suggesting the series does not have daily granularity (e.g.
rebalance-only weights passed directly instead of a full daily
series).
Examples
--------
>>> import pyarrow as pa
>>> tbl = pa.table({
... "date": pa.array(["2024-01-02", "2024-01-03"]).cast(pa.timestamp("ns")),
... "AAPL": pa.array([0.6, 0.5], type=pa.float64()),
... "MSFT": pa.array([0.4, 0.5], type=pa.float64()),
... })
>>> pw = PortfolioWeights(table=tbl)
>>> pw.asset_names
['AAPL', 'MSFT']
"""
table: pa.Table
def __post_init__(self) -> None:
"""Validate table schema and data integrity after construction."""
self._validate_not_empty()
self._validate_date_column()
self._validate_asset_columns_dtype()
self._validate_no_nulls()
self._validate_annual_observation_density()
# self._validate_non_negative()
# self._validate_row_weight_sums()
# ------------------------------------------------------------------
# Validation helpers
# ------------------------------------------------------------------
def _validate_not_empty(self) -> None:
"""
Ensure the table contains at least one row.
Raises
------
DataValidationError
If ``self.table`` has zero rows.
"""
if self.table.num_rows == 0:
raise DataValidationError("PortfolioWeights cannot be empty")
def _validate_date_column(self) -> None:
"""
Ensure a ``"date"`` column exists and is a timestamp type.
Raises
------
MissingColumnError
If no column named ``"date"`` is present.
EntityTypeError
If the ``"date"`` column is not a PyArrow timestamp type.
"""
if DATE_COLUMN not in self.table.column_names:
raise MissingColumnError(
f"PortfolioWeights requires a '{DATE_COLUMN}' column"
)
date_type = self.table.schema.field(DATE_COLUMN).type
if not pa.types.is_timestamp(date_type):
raise EntityTypeError(
f"PortfolioWeights '{DATE_COLUMN}' column must be timestamp, "
f"found {date_type}"
)
def _validate_asset_columns_dtype(self) -> None:
"""
Ensure every non-date column is ``float64``.
Raises
------
EntityTypeError
If any asset column has a type other than ``float64``.
"""
for field in self.table.schema:
if field.name == DATE_COLUMN:
continue
if field.type != pa.float64():
raise EntityTypeError(
f"PortfolioWeights asset column '{field.name}' must be float64, "
f"found {field.type}"
)
def _validate_no_nulls(self) -> None:
"""
Ensure no asset column contains null values.
Raises
------
NullError
If any asset column has at least one null entry.
"""
for field in self.table.schema:
if field.name == DATE_COLUMN:
continue
column = self.table.column(field.name)
if pc.any(pc.is_null(column)).as_py():
raise NullError(
f"Null values found in PortfolioWeights column '{field.name}'"
)
def _validate_annual_observation_density(self) -> None:
"""
Ensure the series has roughly daily observation density.
Compares the number of rows in the first and last 1-year
windows of the series against ``[MIN_ANNUAL_OBSERVATIONS,
MAX_ANNUAL_OBSERVATIONS]``. Skipped entirely if the table spans
less than a year, since there isn't enough history to evaluate
density.
Raises
------
DataIntegrityError
If the observation count in the start or end 1-year window
falls outside the expected range, suggesting the weights
series does not have daily/backtested granularity (e.g.
rebalance-only weights passed directly instead of a full
daily series).
"""
date_column = self.table.column(DATE_COLUMN)
min_date = pc.min(date_column).as_py()
max_date = pc.max(date_column).as_py()
if max_date - min_date < datetime.timedelta(days=365):
return
one_year = datetime.timedelta(days=365)
windows = (
("start", min_date, min_date + one_year),
("end", max_date - one_year, max_date),
)
for label, window_start, window_end in windows:
count = pc.sum(
pc.and_(
pc.greater_equal(date_column, pa.scalar(window_start, type=date_column.type)),
pc.less_equal(date_column, pa.scalar(window_end, type=date_column.type)),
)
).as_py()
if count is None:
count = 0
if not (MIN_ANNUAL_OBSERVATIONS <= count <= MAX_ANNUAL_OBSERVATIONS):
raise DataIntegrityError(
f"PortfolioWeights has {count} observations in its {label} 1-year window "
f"(expected between {MIN_ANNUAL_OBSERVATIONS} and {MAX_ANNUAL_OBSERVATIONS}). "
"This suggests the weights series does not have daily/backtested "
"granularity, e.g. rebalance-only weights (monthly, quarterly, ...) "
"were passed directly instead of a full daily series."
)
# def _validate_non_negative(self):
# """
# Ensure all weight values are non-negative.
#
# Raises
# ------
# DataValidationError
# If any asset column contains a negative value.
# """
# for field in self.table.schema:
# if field.name == DATE_COLUMN:
# continue
# column = self.table.column(field.name)
# min_val = pc.min(column).as_py()
# if min_val is not None and min_val < 0:
# raise DataValidationError(
# f"Negative weight found in PortfolioWeights column '{field.name}': {min_val}"
# )
# def _validate_row_weight_sums(self):
# """
# Ensure per-row weight sums do not exceed 1.0.
#
# Uses a tolerance of ``1.000001`` to accommodate floating-point
# rounding errors.
#
# Raises
# ------
# DataValidationError
# If any row's asset weights sum to more than 1.0 (plus
# tolerance).
# """
# asset_columns = [
# self.table.column(f.name)
# for f in self.table.schema
# if f.name != DATE_COLUMN
# ]
# if not asset_columns:
# return
#
# row_sums = asset_columns[0]
# for col in asset_columns[1:]:
# row_sums = pc.add(row_sums, col)
#
# dates = self.table.column(DATE_COLUMN)
# for i in range(self.table.num_rows):
# total = row_sums[i].as_py()
# if total is not None and total > Decimal("1.000001"):
# raise DataValidationError(
# f"PortfolioWeights row at date {dates[i].as_py()} "
# f"has weights summing to {total:.6f}, exceeding 1.0"
# )
# ------------------------------------------------------------------
# Public helpers
# ------------------------------------------------------------------
@property
def asset_names(self) -> list[str]:
"""
Return the list of asset column names (excludes ``"date"``).
Returns
-------
list[str]
Ordered list of asset ticker names present in the table.
"""
return [n for n in self.table.column_names if n != DATE_COLUMN]
