"""
Concrete Brinson-Fachler return attribution implementation using PyArrow.
Implements :class:`BrinstonFachlerArrowInterface` by decomposing active
portfolio return into allocation, selection, and interaction effects at
the asset level for each period, then aggregating across time.
"""
import pyarrow as pa
import pyarrow.compute as pc
import matplotlib.pyplot as plt
from kaxanuk.attribution_analysis.interfaces.brinston_fachler_arrow_interface import BrinstonFachlerArrowInterface
[docs]
class BrinstonFachlerArrowAttribution(BrinstonFachlerArrowInterface):
"""
Brinson-Fachler return attribution implementation using PyArrow.
Decomposes active portfolio return into allocation, selection, and
interaction effects at the asset level for each period, then
aggregates across time.
Parameters
----------
returns_investable_assets : pa.Table
Asset returns with a ``date_column`` and one float64 column per asset.
complete_portfolio_weights : pa.Table
Portfolio weights with a ``date_column`` and one float64 column per
asset. Rows should sum to 1.0 for each date.
complete_benchmark_weights : pa.Table
Benchmark weights with a ``date_column`` and one float64 column per
asset. Rows should sum to 1.0 for each date.
date_column : str, default ``"date"``
Name of the date column common to all input tables.
Attributes
----------
detail_table : pa.Table
Per-asset detail from the most recent ``brinson_fachler_model`` call.
totals : pa.Table
Single-row totals from the most recent ``brinson_fachler_model`` call.
df : pa.Table
Concatenated totals time series after ``time_series_calculation``.
output_dict : dict
Mapping of date to per-asset detail table after
``time_series_calculation``.
brinston_fach_indexes : pa.Table
Cumulative attribution indexes after ``attribution_plots``.
"""
def __init__(
self,
returns_investable_assets: pa.Table,
complete_portfolio_weights: pa.Table,
complete_benchmark_weights: pa.Table,
date_column: str = "date",
) -> None:
self.returns_investable_assets = returns_investable_assets
self.complete_portfolio_weights = complete_portfolio_weights
self.complete_benchmark_weights = complete_benchmark_weights
self.date_column = date_column
def _get_asset_columns(self, table: pa.Table) -> list[str]:
"""
Return column names excluding the date column.
Parameters
----------
table : pa.Table
Any input table with a date column plus asset columns.
Returns
-------
list[str]
Asset column names.
"""
return [name for name in table.column_names if name != self.date_column]
def _filter_row(self, table: pa.Table, date) -> dict[str, float]:
"""
Extract a single date's row as a dictionary.
Parameters
----------
table : pa.Table
Table to filter.
date : scalar
Date value to match against the date column.
Returns
-------
dict[str, float]
Mapping of asset name to its value for the given date.
Returns an empty dict if the date is not found.
"""
mask = pc.equal(table.column(self.date_column), date)
filtered = table.filter(mask)
if filtered.num_rows == 0:
return {}
result = {}
for name in self._get_asset_columns(table):
val = filtered.column(name)[0].as_py()
result[name] = val if val is not None else 0.0
return result
[docs]
def brinson_fachler_model(self, date: str) -> tuple[pa.Table, pa.Table]:
"""
Calculate the Brinson-Fachler attribution for a single date.
Decomposes the active return into allocation, selection, and
interaction effects for every asset, then sums to produce
portfolio-level totals.
Parameters
----------
date : scalar
Target date matching a value in the date column.
Returns
-------
detail_table : pa.Table
Per-asset breakdown with columns: ``asset``, ``portfolio_weights``,
``benchmark_weights``, ``returns_data``, ``portfolio_returns``,
``benchmark_returns``, ``alpha``, ``allocation``, ``selection``,
``interaction``.
totals_table : pa.Table
Single-row aggregate with columns: ``date``, ``portfolio_returns``,
``benchmark_returns``, ``alpha``, ``allocation``, ``selection``,
``interaction``.
"""
assets = self._get_asset_columns(self.returns_investable_assets)
r_dict = self._filter_row(self.returns_investable_assets, date)
w_p_dict = self._filter_row(self.complete_portfolio_weights, date)
w_b_dict = self._filter_row(self.complete_benchmark_weights, date)
r = pa.array([r_dict.get(a, 0.0) for a in assets], type=pa.float64())
w_p = pa.array([w_p_dict.get(a, 0.0) for a in assets], type=pa.float64())
w_b = pa.array([w_b_dict.get(a, 0.0) for a in assets], type=pa.float64())
r_p = pc.multiply(w_p, r)
r_b = pc.multiply(w_b, r)
alpha_index = pc.subtract(r_p, r_b)
allocation = pc.multiply(pc.subtract(w_p, w_b), r_b)
selection = pc.multiply(alpha_index, w_p)
interaction = pc.subtract(pc.subtract(alpha_index, allocation), selection)
detail_table = pa.table({
"asset": pa.array(assets, type=pa.string()),
"portfolio_weights": w_p,
"benchmark_weights": w_b,
"returns_data": r,
"portfolio_returns": r_p,
"benchmark_returns": r_b,
"alpha": alpha_index,
"allocation": allocation,
"selection": selection,
"interaction": interaction,
})
totals_table = pa.table({
"date": pa.array([date]),
"portfolio_returns": pa.array([pc.sum(r_p).as_py()], type=pa.float64()),
"benchmark_returns": pa.array([pc.sum(r_b).as_py()], type=pa.float64()),
"alpha": pa.array([pc.sum(alpha_index).as_py()], type=pa.float64()),
"allocation": pa.array([pc.sum(allocation).as_py()], type=pa.float64()),
"selection": pa.array([pc.sum(selection).as_py()], type=pa.float64()),
"interaction": pa.array([pc.sum(interaction).as_py()], type=pa.float64()),
})
self.detail_table = detail_table
self.totals = totals_table
return detail_table, totals_table
[docs]
def time_series_calculation(self) -> None:
"""
Run Brinson-Fachler attribution across all common dates.
Finds dates present in both portfolio and benchmark weights, runs
``brinson_fachler_model`` for each date that also has return data,
and concatenates results.
Sets Attributes
---------------
df : pa.Table
Concatenated totals time series with columns: ``date``,
``portfolio_returns``, ``benchmark_returns``, ``alpha``,
``allocation``, ``selection``, ``interaction``.
output_dict : dict
Mapping of each date to its per-asset ``pa.Table`` detail.
"""
dates_portfolio = self.complete_portfolio_weights.column(self.date_column)
dates_benchmark = self.complete_benchmark_weights.column(self.date_column)
dates_returns = self.returns_investable_assets.column(self.date_column)
portfolio_set = set(dates_portfolio.to_pylist())
benchmark_set = set(dates_benchmark.to_pylist())
returns_set = set(dates_returns.to_pylist())
common_dates = sorted(portfolio_set.intersection(benchmark_set))
output_dict = {}
totals_list = []
for date in common_dates:
if date in returns_set:
detail_table, totals_table = self.brinson_fachler_model(date)
output_dict[date] = detail_table
totals_list.append(totals_table)
if totals_list:
self.df = pa.concat_tables(totals_list)
else:
self.df = pa.table({
"date": pa.array([], type=pa.timestamp("ns")),
"portfolio_returns": pa.array([], type=pa.float64()),
"benchmark_returns": pa.array([], type=pa.float64()),
"alpha": pa.array([], type=pa.float64()),
"allocation": pa.array([], type=pa.float64()),
"selection": pa.array([], type=pa.float64()),
"interaction": pa.array([], type=pa.float64()),
})
self.output_dict = output_dict
[docs]
def attribution_plots(self) -> None:
"""
Plot cumulative Brinson-Fachler attribution effects.
Produces a 3-panel figure:
1. Cumulative portfolio vs benchmark returns.
2. Cumulative alpha.
3. Cumulative allocation, selection, and interaction effects.
Must be called after ``time_series_calculation``.
Sets Attributes
---------------
brinston_fach_indexes : pa.Table
Cumulative sums of all attribution columns, keyed by date.
"""
date_col = self.df.column("date").to_pylist()
numeric_cols = ["portfolio_returns", "benchmark_returns", "alpha",
"allocation", "selection", "interaction"]
cumulative = {}
for col_name in numeric_cols:
col = self.df.column(col_name)
cumulative[col_name] = pc.cumulative_sum(col, skip_nulls=True).to_pylist()
self.brinston_fach_indexes = pa.table(
{"date": pa.array(date_col), **{k: pa.array(v, type=pa.float64()) for k, v in cumulative.items()}}
)
_fig, axes = plt.subplots(3, 1, figsize=(10, 14))
axes[0].plot(date_col, cumulative["portfolio_returns"], label="portfolio_returns")
axes[0].plot(date_col, cumulative["benchmark_returns"], label="benchmark_returns")
axes[0].set_title("Strategy vs Benchmark Returns")
axes[0].legend()
axes[0].grid(visible=True)
axes[1].plot(date_col, cumulative["alpha"], label="alpha")
axes[1].set_title("Cumulative Alpha")
axes[1].legend()
axes[1].grid(visible=True)
axes[2].plot(date_col, cumulative["allocation"], label="allocation")
axes[2].plot(date_col, cumulative["selection"], label="selection")
axes[2].plot(date_col, cumulative["interaction"], label="interaction")
axes[2].set_title("Allocation, Selection and Interaction Effects")
axes[2].legend()
axes[2].grid(visible=True)
plt.tight_layout(rect=[0, 0, 1, 0.95])
plt.show()
