"""
Conversion utilities for reshaping PyArrow tables into the format expected
by the Brinson-Fachler Arrow attribution engine.
Provides functions equivalent to table_conversion.py but without converting
to pandas — all operations stay in PyArrow.
"""
import pyarrow as pa
import pyarrow.compute as pc
from kaxanuk.attribution_analysis.entities.portfolio_weights_arrow import PortfolioWeights
from kaxanuk.attribution_analysis.entities.returns_table import ReturnsTable
[docs]
def portfolio_table_to_arrow(table: pa.Table) -> PortfolioWeights:
"""
Transpose a wide portfolio/benchmark weights table into attribution format.
This is the PyArrow equivalent of ``portfolio_table_to_pandas``, which
does ``set_index("Ticker").T``.
Parameters
----------
table : pyarrow.Table
Wide-format weights table with columns ``[Ticker, date1, date2,
...]``, one row per ticker and one column per date. Date column
names are parsed as ``"%Y-%m-%d"``, falling back to
``"%d/%m/%Y"`` if that fails.
Returns
-------
PortfolioWeights
Wrapper around a table with columns ``[date, asset1, asset2,
...]``, i.e. one row per date and one column per ticker.
"""
tickers = table.column("Ticker").to_pylist()
date_columns = [name for name in table.column_names if name != "Ticker"]
date_strings = date_columns
try:
dates = pc.strptime(pa.array(date_strings, type=pa.string()), format="%Y-%m-%d", unit="ns")
except (pa.ArrowInvalid, pa.ArrowNotImplementedError, ValueError):
dates = pc.strptime(pa.array(date_strings, type=pa.string()), format="%d/%m/%Y", unit="ns")
asset_arrays = {}
for i, ticker in enumerate(tickers):
row_values = []
for date_col_name in date_columns:
col = table.column(date_col_name)
val = col[i].as_py()
row_values.append(float(val) if val is not None else None)
asset_arrays[ticker] = pa.array(row_values, type=pa.float64())
columns = [dates]
column_names = ["date"]
for ticker in tickers:
columns.append(asset_arrays[ticker])
column_names.append(ticker)
return PortfolioWeights(table=pa.table(columns, names=column_names))
[docs]
def compute_returns_from_prices_arrow(
prices_table: pa.Table,
date_column: str,
) -> ReturnsTable:
"""
Compute percentage returns from a prices table.
Equivalent to ``prices_df.pct_change(fill_method=None)`` — the first
row is kept with null values so the date range stays aligned with the
original prices table (and therefore with the weights tables).
Parameters
----------
prices_table : pyarrow.Table
Prices table containing a date column and one or more numeric
asset price columns.
date_column : str
Name of the column in `prices_table` containing the dates. It's
passed through unchanged and always renamed to ``"date"`` in the
output, regardless of its original name.
Returns
-------
ReturnsTable
Wrapper around a table with the same columns and row count as
`prices_table`, where each asset column holds period-over-period
percentage returns and the first row is null.
"""
date_col = prices_table.column(date_column)
result_columns = [date_col]
result_names = ["date"]
for name in prices_table.column_names:
if name == date_column:
continue
col = prices_table.column(name)
previous = col.slice(0, len(col) - 1)
current = col.slice(1)
returns = pc.divide(
pc.subtract(current, previous),
previous
)
if isinstance(returns, pa.ChunkedArray):
returns = returns.combine_chunks()
first_null = pa.array([None], type=pa.float64())
returns_with_null = pa.concat_arrays([first_null, returns])
result_columns.append(returns_with_null)
result_names.append(name)
return ReturnsTable(table=pa.table(result_columns, names=result_names))
[docs]
def returns_table_to_arrow(table: pa.Table) -> ReturnsTable:
"""
Transpose a wide returns table into attribution format.
Same transpose logic as :func:`portfolio_table_to_arrow`, but wraps
the result in a :class:`ReturnsTable` (no weight constraints).
Parameters
----------
table : pyarrow.Table
Wide-format returns table with columns ``[Ticker, date1, date2,
...]``, one row per ticker and one column per date. Date column
names are parsed as ``"%Y-%m-%d"``, falling back to
``"%d/%m/%Y"`` if that fails.
Returns
-------
ReturnsTable
Wrapper around a table with columns ``[date, asset1, asset2,
...]``, i.e. one row per date and one column per ticker.
"""
tickers = table.column("Ticker").to_pylist()
date_columns = [name for name in table.column_names if name != "Ticker"]
try:
dates = pc.strptime(pa.array(date_columns, type=pa.string()), format="%Y-%m-%d", unit="ns")
except (pa.ArrowInvalid, pa.ArrowNotImplementedError, ValueError):
dates = pc.strptime(pa.array(date_columns, type=pa.string()), format="%d/%m/%Y", unit="ns")
asset_arrays = {}
for i, ticker in enumerate(tickers):
row_values = []
for date_col_name in date_columns:
col = table.column(date_col_name)
val = col[i].as_py()
row_values.append(float(val) if val is not None else None)
asset_arrays[ticker] = pa.array(row_values, type=pa.float64())
columns = [dates]
column_names = ["date"]
for ticker in tickers:
columns.append(asset_arrays[ticker])
column_names.append(ticker)
return ReturnsTable(table=pa.table(columns, names=column_names))
def _filter_arrow_table_by_date_range(
table: pa.Table,
start: pa.Scalar,
end: pa.Scalar,
date_col: str = "date",
) -> pa.Table:
"""
Filter a table to rows whose date column falls within an inclusive range.
Parameters
----------
table : pyarrow.Table
Table to filter. Must contain a column named `date_col`.
start : pyarrow.Scalar
Earliest date to include (inclusive). Must be comparable to the
`date_col` column's type (e.g. a ``timestamp("ns")`` scalar).
end : pyarrow.Scalar
Latest date to include (inclusive). Must be comparable to the
`date_col` column's type.
date_col : str, default "date"
Name of the date column to filter on.
Returns
-------
pyarrow.Table
A new table containing only the rows where `date_col` is between
`start` and `end`, inclusive.
"""
col = table.column(date_col)
mask = pc.and_(pc.greater_equal(col, start), pc.less_equal(col, end))
return table.filter(mask)
