"""Asset Panel Fields module."""
# Copyright (c) 2023-2026
# Author: Hugo Delatte <hugo.delatte@skfoliolabs.com>
# SPDX-License-Identifier: BSD-3-Clause
from __future__ import annotations
from abc import ABC
from dataclasses import dataclass, field
from dataclasses import fields as dataclass_fields
from enum import auto
import numpy as np
from skfolio.containers._utils import _as_pickle_safe_array, _validate_unique_labels
from skfolio.typing import AnyArray, ArrayLike, BoolArray, StrArray
from skfolio.utils.tools import AutoEnum
__all__ = [
"MISSING_CATEGORY_CODE",
"BaseField",
"Field2D",
"Field3D",
"FieldCategorical",
"InactivePolicy",
]
MISSING_CATEGORY_CODE: int = -1
[docs]
class InactivePolicy(AutoEnum):
"""Validation policy for values outside an `AssetPanel` active universe."""
MISSING = auto()
ZERO = auto()
IGNORE = auto()
[docs]
@dataclass(slots=True)
class BaseField(ABC):
"""Base class for fields stored in an `AssetPanel`.
A field stores a NumPy array for one named variable. The first two axes are always
observations and assets with shape (n_observations, n_assets).
Subclasses define any additional metadata required by the array layout.
Parameters
----------
values : ndarray of shape (n_observations, n_assets)
Field values. The first two axes must be observations and assets.
"""
values: AnyArray
inactive_policy: InactivePolicy = field(
default=InactivePolicy.MISSING,
kw_only=True,
)
def __post_init__(self) -> None:
self.values = np.asarray(self.values)
if not isinstance(self.inactive_policy, InactivePolicy):
raise TypeError(
"`inactive_policy` must be an InactivePolicy, "
f"got {self.inactive_policy!r}."
)
if self.values.ndim < 2:
raise ValueError("Field values must have at least two dimensions.")
@property
def shape(self) -> tuple[int, ...]:
"""Shape of the underlying values."""
return self.values.shape
@property
def dtype(self) -> np.dtype:
"""Dtype of the underlying values."""
return self.values.dtype
@property
def ndim(self) -> int:
"""Number of dimensions of the underlying values."""
return self.values.ndim
@property
def n_observations(self) -> int:
"""Number of observations."""
return int(self.values.shape[0])
@property
def n_assets(self) -> int:
"""Number of assets."""
return int(self.values.shape[1])
@property
def is_categorical(self) -> bool:
"""Whether or not the field is a categorical field."""
return isinstance(self, FieldCategorical)
@property
def is_3d(self) -> bool:
"""Whether or not the field is a 3D field."""
return isinstance(self, Field3D)
@property
def missing_mask(self) -> BoolArray:
"""Return a boolean mask indicating missing entries.
Returns
-------
mask : ndarray of bool
Boolean array with the same shape as `values`. Entries are `True`
where `values` is NaN for floating-point fields and `False` elsewhere.
Non-floating fields without a missing-value convention return an all-`False`
mask; subclasses override this method when they define their own convention.
"""
if np.issubdtype(self.values.dtype, np.floating):
return np.isnan(self.values)
return np.zeros(self.values.shape, dtype=np.bool_)
[docs]
def copy(self, *, deep: bool = False) -> BaseField:
"""Return a copy of the field.
Parameters
----------
deep : bool, default=False
If `True`, copy NumPy arrays stored by the field. If `False`, reuse the same
array objects.
Returns
-------
field : BaseField
Field instance of the same concrete class.
"""
kwargs = {}
for f in dataclass_fields(self):
value = getattr(self, f.name)
if deep and isinstance(value, np.ndarray):
value = value.copy()
kwargs[f.name] = value
return type(self)(**kwargs)
[docs]
def with_values(self, values: ArrayLike) -> BaseField:
"""Return a field of the same type with replacement values.
Metadata arrays are reused. The replacement values are validated by the
concrete field class constructor.
Parameters
----------
values : array-like
Replacement field values.
Returns
-------
field : BaseField
Field instance of the same concrete class.
"""
kwargs = {
field.name: getattr(self, field.name)
for field in dataclass_fields(self)
if field.name != "values"
}
return type(self)(values=np.asarray(values), **kwargs)
[docs]
@dataclass(slots=True)
class Field2D(BaseField):
"""Numeric 2D field with axes (observations, assets).
Parameters
----------
values : ndarray of shape (n_observations, n_assets)
Numeric field values. Object dtype is rejected. Use `FieldCategorical` for
categorical data.
"""
def __post_init__(self) -> None:
BaseField.__post_init__(self)
if self.values.ndim != 2:
raise ValueError(f"Field2D values must be 2D, got ndim={self.values.ndim}.")
if self.values.dtype == object:
raise ValueError(
"Field2D values cannot have dtype=object. Store categoricals as "
"integer codes with FieldCategorical."
)
[docs]
@dataclass(slots=True)
class FieldCategorical(Field2D):
"""Integer-coded categorical 2D field.
Codes are stored as integers in a 2D array with axes (observations, assets).
Code -1 is reserved for missing values. Code 0 selects `levels[0]`, code 1 selects
`levels[1]` and so on.
Parameters
----------
values : integers ndarray of shape (n_observations, n_assets)
Integer category codes. Missing values must be encoded with
`MISSING_CATEGORY_CODE`.
levels : ndarray of shape (n_levels,)
Category labels selected by codes 0, 1 and so on.
Labels must be unique.
"""
levels: StrArray
def __post_init__(self) -> None:
Field2D.__post_init__(self)
if self.inactive_policy == InactivePolicy.ZERO:
raise ValueError(
"`inactive_policy=InactivePolicy.ZERO` is not supported for "
"FieldCategorical because 0 is a valid category code."
)
self.levels = _as_pickle_safe_array(self.levels)
if not np.issubdtype(self.values.dtype, np.integer):
raise ValueError(
"FieldCategorical values must have an integer dtype; "
f"got {self.values.dtype}."
)
if self.levels.ndim != 1:
raise ValueError("FieldCategorical levels must be a 1D array.")
_validate_unique_labels(self.levels, name="FieldCategorical levels")
if self.values.size != 0:
min_code = int(np.min(self.values))
max_code = int(np.max(self.values))
if min_code < MISSING_CATEGORY_CODE:
raise ValueError(
f"Categorical codes must be >= {MISSING_CATEGORY_CODE}; "
f"got minimum code {min_code}."
)
if max_code >= len(self.levels):
raise ValueError(
f"Categorical code {max_code} is out of bounds for "
f"{len(self.levels)} levels."
)
@property
def missing_mask(self) -> BoolArray:
"""Return a boolean mask indicating missing categorical codes.
Returns
-------
mask : ndarray of bool
Boolean array with the same shape as `values`. Entries are `True`
where `values` equals `MISSING_CATEGORY_CODE`.
"""
return self.values == MISSING_CATEGORY_CODE
[docs]
def decode(self, *, missing_label: str = "MISSING") -> StrArray:
"""Decode integer codes to level labels.
Parameters
----------
missing_label : str, default="MISSING"
Label assigned to missing or out-of-bound codes.
Returns
-------
decoded : ndarray
Decoded labels with the same shape as `values`.
"""
codes = self.values
levels = self.levels
if codes.size == 0:
return np.asarray([], dtype=levels.dtype).reshape(codes.shape)
dtype = np.result_type(levels.dtype, np.asarray(missing_label).dtype)
decoded = np.empty(codes.shape, dtype=dtype)
valid = (codes >= 0) & (codes < len(levels))
decoded[valid] = levels[codes[valid]]
decoded[~valid] = str(missing_label)
return decoded
[docs]
@dataclass(slots=True)
class Field3D(BaseField):
"""Numeric 3D field with axes (observations, assets, third_axis).
Use this field for homogeneous tensors (e.g. factor exposures). The third-axis
metadata is stored on the field. The array is stored physically in 3D so that
operations along any axis remain vectorized and avoid stacking many 2D arrays which
is expensive for large panels. When you need repeated tensor operations, prefer a
`Field3D` instead of multiple `Field2D`.
Parameters
----------
values : ndarray of shape `(n_observations, n_assets, n_third_axis)`
Numeric 3D values. Object dtype is rejected.
third_axis_name : str
Name describing what the third axis represents (e.g. `factor`).
third_axis_labels : ndarray of shape (n_third_axis,)
Labels for entries along the third axis such as factor names (`size`,
`momentum`). Labels must be unique.
third_axis_groups : ndarray of shape (n_third_axis,), optional
Optional group label for each third-axis entry such as factor families (e.g.
`style`, `industry`, `country`).
"""
third_axis_name: str
third_axis_labels: StrArray
third_axis_groups: StrArray | None = None
def __post_init__(self) -> None:
BaseField.__post_init__(self)
self.third_axis_labels = _as_pickle_safe_array(self.third_axis_labels)
if self.third_axis_groups is not None:
self.third_axis_groups = _as_pickle_safe_array(self.third_axis_groups)
if self.values.ndim != 3:
raise ValueError(f"Field3D values must be 3D, got ndim={self.values.ndim}.")
if self.values.dtype == object:
raise ValueError("Field3D values cannot have dtype=object.")
if not isinstance(self.third_axis_name, str) or not self.third_axis_name:
raise ValueError("third_axis_name must be a non-empty string.")
if self.third_axis_labels.ndim != 1:
raise ValueError("third_axis_labels must be a 1D array.")
_validate_unique_labels(self.third_axis_labels, name="third_axis_labels")
if len(self.third_axis_labels) != self.values.shape[2]:
raise ValueError(
f"Field3D has {self.values.shape[2]} entries on the third axis but "
f"third_axis_labels has length {len(self.third_axis_labels)}."
)
if self.third_axis_groups is not None:
if self.third_axis_groups.ndim != 1:
raise ValueError("third_axis_groups must be a 1D array.")
if len(self.third_axis_groups) != self.values.shape[2]:
raise ValueError(
"third_axis_groups length must match third_axis_labels length."
)
