import logging
import math
import matplotlib.pyplot as plt
import pandas as pd
import pyarrow as pa
import pyarrow.compute as pc
from matplotlib.dates import DateFormatter
from matplotlib.ticker import FuncFormatter, PercentFormatter
from kaxanuk.attribution_analysis.interfaces.factor_model_arrow_interface import FactorModelArrowAttributionInterface
logger = logging.getLogger(__name__)
[docs]
class FactorModelArrowAttribution(FactorModelArrowAttributionInterface):
"""
Multi-factor model attribution implementation using PyArrow.
Decomposes portfolio returns into contributions from style factors,
industry factors, and an idiosyncratic residual using portfolio weights
and per-asset factor return tables.
Parameters
----------
daily_portfolio_weights : pa.Table
Daily portfolio weights with a ``date_column`` and one float64
column per asset ticker.
by_factor_factor_returns : dict[str, pa.Table]
Mapping of factor name (e.g. ``"f_momentum"``) to a ``pa.Table``
of per-asset factor returns, each with a ``date_column`` and one
float64 column per asset.
asset_returns : pa.Table
Total asset returns with a ``date_column`` and one float64 column
per asset ticker.
portfolio_returns : pa.Table, optional
Pre-computed portfolio returns with columns ``date`` and
``portfolio_returns``. Computed from weights and asset returns
if not provided.
benchmark_returns : pa.Table, optional
Benchmark returns table.
date_column : str, default ``"date"``
Name of the date column common to all input tables.
Attributes
----------
portfolio_attribution_ts : pa.Table
Daily factor contribution time series after ``multifactor_attribution``.
simulated_rets : pa.Table
Cumulative factor contributions after ``time_series_calculation``.
pct_df_returns : pa.Table
Daily percentage decomposition after ``calc_pct_area``.
"""
def __init__(
self,
daily_portfolio_weights: pa.Table,
by_factor_factor_returns: dict[str, pa.Table],
asset_returns: pa.Table,
portfolio_returns: pa.Table | None = None,
benchmark_returns: pa.Table | None = None,
date_column: str = "date",
) -> None:
super().__init__(
daily_portfolio_weights=daily_portfolio_weights,
by_factor_factor_returns=by_factor_factor_returns,
asset_returns=asset_returns,
portfolio_returns=portfolio_returns,
benchmark_returns=benchmark_returns,
)
self.date_column = date_column
def _get_non_date_columns(self, table: pa.Table) -> list[str]:
"""
Return column names excluding the date column.
Parameters
----------
table : pa.Table
Any table with a date column plus data columns.
Returns
-------
list[str]
Non-date 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 column name to its value. Returns an empty dict
if the date is not found. Nulls are replaced with 0.0.
"""
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_non_date_columns(table):
val = filtered.column(name)[0].as_py()
result[name] = val if val is not None else 0.0
return result
def _compute_portfolio_returns(self) -> None:
"""
Compute portfolio-level returns from weights and asset returns.
Calculates the weighted sum of asset returns for each date and
stores the result in ``self.portfolio_returns``.
Sets Attributes
---------------
portfolio_returns : pa.Table
Table with columns ``date_column`` and ``portfolio_returns``.
"""
w_dates = self.daily_portfolio_weights.column(self.date_column).to_pylist()
r_dates = self.asset_returns.column(self.date_column).to_pylist()
r_date_to_idx = {d: i for i, d in enumerate(r_dates)}
asset_cols = [
c for c in self._get_non_date_columns(self.daily_portfolio_weights)
if c in self.asset_returns.column_names
]
out_dates = []
out_rets = []
for w_idx, date_val in enumerate(w_dates):
r_idx = r_date_to_idx.get(date_val)
if r_idx is None:
continue
total = 0.0
for col in asset_cols:
w = self.daily_portfolio_weights.column(col)[w_idx].as_py()
r = self.asset_returns.column(col)[r_idx].as_py()
if w is not None and r is not None:
total += w * r
out_dates.append(date_val)
out_rets.append(total)
self.portfolio_returns = pa.table({
self.date_column: pa.array(out_dates),
"portfolio_returns": pa.array(out_rets, type=pa.float64()),
})
[docs]
def multifactor_attribution(self) -> None:
"""
Calculate multi-factor attribution across all dates.
For each date in ``daily_portfolio_weights``, computes the
portfolio-weighted sum of each factor's per-asset returns.
Logs average factor coverage across the portfolio.
Sets Attributes
---------------
portfolio_attribution_ts : pa.Table
Daily factor contributions with a date column and one float64
column per factor.
"""
non_date_cols = self._get_non_date_columns(self.daily_portfolio_weights)
dates = self.daily_portfolio_weights.column(self.date_column).to_pylist()
factor_date_indices = {}
for factor_name, factor_table in self.by_factor_factor_returns.items():
factor_dates = factor_table.column(self.date_column).to_pylist()
factor_date_indices[factor_name] = {d: i for i, d in enumerate(factor_dates)}
total_daily_coverage = []
b_x_f_ts_dates = []
b_x_f_ts_values = []
for w_idx, date_val in enumerate(dates):
weights_dict = {}
for col in non_date_cols:
val = self.daily_portfolio_weights.column(col)[w_idx].as_py()
if val is not None and val != 0.0:
weights_dict[col] = val
if not weights_dict:
continue
factor_sums = {}
all_valid_assets = set()
for factor_name, factor_table in self.by_factor_factor_returns.items():
f_idx = factor_date_indices[factor_name].get(date_val)
if f_idx is None:
continue
factor_non_date = self._get_non_date_columns(factor_table)
factor_row = {}
for col in factor_non_date:
val = factor_table.column(col)[f_idx].as_py()
if val is not None:
factor_row[col] = val
common_assets = set(weights_dict.keys()) & set(factor_row.keys())
all_valid_assets.update(common_assets)
weighted_sum = sum(
weights_dict[asset] * factor_row[asset]
for asset in common_assets
)
factor_sums[factor_name] = weighted_sum
if not factor_sums:
continue
coverage = sum(weights_dict[a] for a in all_valid_assets if a in weights_dict)
total_daily_coverage.append(coverage)
b_x_f_ts_dates.append(date_val)
b_x_f_ts_values.append(factor_sums)
if total_daily_coverage:
avg_coverage = sum(total_daily_coverage) / len(total_daily_coverage)
logger.info(
'Average total factor coverage throughout the portfolio : %s',
f'{avg_coverage: .4%}'
)
if not b_x_f_ts_dates:
self.portfolio_attribution_ts = pa.table({
self.date_column: pa.array([], type=pa.timestamp("ns"))
})
return
all_factors = set()
for sums in b_x_f_ts_values:
all_factors.update(sums.keys())
all_factors = sorted(all_factors)
columns = {self.date_column: pa.array(b_x_f_ts_dates)}
for factor in all_factors:
columns[factor] = pa.array(
[sums.get(factor, 0.0) for sums in b_x_f_ts_values],
type=pa.float64(),
)
self.portfolio_attribution_ts = pa.table(columns)
[docs]
def time_series_calculation(self) -> None:
"""
Compute cumulative factor contributions over time.
Applies ``cumulative_sum`` to each factor column in
``portfolio_attribution_ts``.
Sets Attributes
---------------
simulated_rets : pa.Table
Cumulative factor contributions with the same schema as
``portfolio_attribution_ts``.
"""
columns = {self.date_column: self.portfolio_attribution_ts.column(self.date_column)}
for col_name in self._get_non_date_columns(self.portfolio_attribution_ts):
col = self.portfolio_attribution_ts.column(col_name)
columns[col_name] = pc.cumulative_sum(col, skip_nulls=True)
self.simulated_rets = pa.table(columns)
[docs]
def run(self) -> None:
"""
Execute the full attribution pipeline.
Runs ``multifactor_attribution``, ``time_series_calculation``, and
optionally computes portfolio returns if not provided.
Raises
------
ValueError
If ``asset_returns`` is ``None``.
"""
if self.asset_returns is None:
raise ValueError("asset_returns must be provided")
self.multifactor_attribution()
self.time_series_calculation()
if self.portfolio_returns is None:
self._compute_portfolio_returns()
[docs]
def calc_pct_area(self) -> None:
"""
Compute daily percentage decomposition of returns by factor.
Divides each factor's daily contribution by the total portfolio
return to get the fraction attributable to that factor. An
idiosyncratic residual column (``f_idio_returns``) captures the
portion not explained by named factors. NaN and Inf values are
forward-filled.
Sets Attributes
---------------
pct_df_returns : pa.Table
Daily percentage attribution with the same date column and
one float64 column per factor plus ``f_idio_returns``.
"""
cols_to_drop = {
self.date_column,
'f_market', 'f_total_factor_returns',
'f_total_excess_returns', 'f_idyo_returns',
}
factor_cols = [
c for c in self.portfolio_attribution_ts.column_names
if c not in cols_to_drop
]
attr_dates = self.portfolio_attribution_ts.column(self.date_column).to_pylist()
pr_dates = self.portfolio_returns.column(self.date_column).to_pylist()
pr_date_to_idx = {d: i for i, d in enumerate(pr_dates)}
aligned_pr = []
for d in attr_dates:
idx = pr_date_to_idx.get(d)
if idx is not None:
aligned_pr.append(
self.portfolio_returns.column("portfolio_returns")[idx].as_py()
)
else:
aligned_pr.append(None)
total_returns = pa.array(aligned_pr, type=pa.float64())
row_sums = None
for col_name in factor_cols:
col = self.portfolio_attribution_ts.column(col_name)
row_sums = col if row_sums is None else pc.add(row_sums, col)
if row_sums is None:
row_sums = pa.array([0.0] * len(attr_dates), type=pa.float64())
idio = pc.subtract(total_returns, row_sums)
pct_columns = {self.date_column: self.portfolio_attribution_ts.column(self.date_column)}
for col_name in factor_cols:
col = self.portfolio_attribution_ts.column(col_name)
pct_columns[col_name] = pc.divide(col, total_returns)
pct_columns['f_idio_returns'] = pc.divide(idio, total_returns)
result_columns = {self.date_column: pct_columns[self.date_column]}
for col_name in list(pct_columns.keys()):
if col_name == self.date_column:
continue
values = pct_columns[col_name].to_pylist()
filled = []
last_valid = 0.0
for v in values:
if v is None or (isinstance(v, float) and not math.isfinite(v)):
filled.append(last_valid)
else:
last_valid = v
filled.append(v)
result_columns[col_name] = pa.array(filled, type=pa.float64())
self.pct_df_returns = pa.table(result_columns)
[docs]
def cummulative_pct_decomp(self) -> dict[str, float]:
"""
Compute the cumulative percentage decomposition at the final date.
Takes the last row of ``simulated_rets`` and expresses each factor's
cumulative contribution as a fraction of total excess returns,
including an idiosyncratic residual.
Returns
-------
dict[str, float]
Mapping of factor name (plus ``"idio_returns"``) to its
percentage share of total cumulative excess return.
"""
last_row_idx = self.simulated_rets.num_rows - 1
cols_to_drop = {
self.date_column,
'f_market', 'f_total_factor_returns',
'f_total_excess_returns', 'f_idyo_returns',
}
if 'f_total_excess_returns' in self.simulated_rets.column_names:
total_final_ret = self.simulated_rets.column(
'f_total_excess_returns'
)[last_row_idx].as_py()
else:
factor_cols = [
c for c in self.simulated_rets.column_names
if c not in cols_to_drop
]
total_final_ret = sum(
(self.simulated_rets.column(c)[last_row_idx].as_py() or 0.0)
for c in factor_cols
)
factor_cols = [
c for c in self.simulated_rets.column_names
if c not in cols_to_drop
]
factor_values = {}
total_factor_sum = 0.0
for col in factor_cols:
val = self.simulated_rets.column(col)[last_row_idx].as_py() or 0.0
factor_values[col] = val
total_factor_sum += val
idio = total_final_ret - total_factor_sum
factor_values['idio_returns'] = idio
if total_final_ret == 0:
total_pct = dict.fromkeys(factor_values, 0.0)
else:
total_pct = {k: v / total_final_ret for k, v in factor_values.items()}
return total_pct
# --- Plotting methods (convert to pandas at the matplotlib boundary) ---
[docs]
def attribution_plots(self, title: str | None = None) -> None:
"""
Plot cumulative multi-factor attribution in a 3-panel figure.
Panels show: (1) total / factor / idiosyncratic cumulative returns,
(2) style factor cumulative returns, (3) industry factor cumulative
returns. Requires ``time_series_calculation`` to have been run.
Parameters
----------
title : str, optional
Overall figure title. Currently unused by the plot layout.
"""
df = self.simulated_rets.to_pandas()
if self.date_column in df.columns:
df = df.set_index(self.date_column)
totals_cols = ["f_total_excess_returns", "f_total_factor_returns", "f_idyo_returns"]
style_cols = ["f_size", "f_momentum", "f_beta", "f_residual volatility", "f_value"]
industry_cols = [
"f_Communication Services", "f_Consumer Discretionary",
"f_Consumer Staples", "f_Energy", "f_Financials",
"f_Health Care", "f_Industrials", "f_Information Technology",
"f_Materials", "f_Real Estate", "f_Utilities",
]
totals_cols = [c for c in totals_cols if c in df.columns]
style_cols = [c for c in style_cols if c in df.columns]
industry_cols = [c for c in industry_cols if c in df.columns]
industries_df = df[industry_cols]
industries_df = industries_df.loc[:, (industries_df != 0).any(axis=0)]
_fig, axes = plt.subplots(3, 1, figsize=(12, 15), sharex=True)
def style_axis(ax):
ax.grid(visible=True, which="both", axis="both", alpha=0.35)
ax.axhline(0, color="black", linewidth=1)
ax.yaxis.set_major_formatter(PercentFormatter(xmax=1.0))
leg = ax.get_legend()
if leg is not None:
ax.legend(loc="upper left")
if totals_cols:
df[totals_cols].plot(ax=axes[0], alpha=0.85)
axes[0].set_title("Total / Factor / Idiosyncratic")
style_axis(axes[0])
if style_cols:
df[style_cols].plot(ax=axes[1], alpha=0.85)
axes[1].set_title("Style Factors")
style_axis(axes[1])
if not industries_df.empty:
industries_df.plot(ax=axes[2], alpha=0.85)
axes[2].set_title("Industry Returns (non-zero columns)")
style_axis(axes[2])
plt.tight_layout(rect=[0, 0, 1, 0.96])
plt.show()
[docs]
def plot_stacked_weights_normalized(
self,
figsize: tuple[int, int] = (14, 7),
title: str = "Return Attribution by Factor Over Time",
y_label: str = "Share of Total Return",
date_fmt: str = "%Y-%m",
legend_loc: str = "upper left",
days_of_smoothing: int = 5,
) -> None:
"""
Plot a normalized stacked area chart of factor contributions.
Each factor's absolute contribution is divided by the sum of all
absolute contributions to show relative importance over time,
smoothed with a rolling window.
Parameters
----------
figsize : tuple[int, int], default ``(14, 7)``
Matplotlib figure size in inches.
title : str, default ``"Return Attribution by Factor Over Time"``
Chart title. A rolling-window suffix is appended automatically.
y_label : str, default ``"Share of Total Return"``
Y-axis label.
date_fmt : str, default ``"%Y-%m"``
Date format string for the x-axis.
legend_loc : str, default ``"upper left"``
Matplotlib legend location.
days_of_smoothing : int, default ``5``
Rolling window size in trading days for smoothing.
"""
self.calc_pct_area()
df = self.pct_df_returns.to_pandas()
if self.date_column in df.columns:
df = df.set_index(self.date_column)
df = df.ffill()
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
df = df.sort_index()
abs_components = df.abs()
abs_total = abs_components.sum(axis=1)
pct = abs_components.div(abs_total, axis=0).rolling(days_of_smoothing).mean()
pct = pct.fillna(0.0)
colors = [plt.cm.tab20(i % 20) for i in range(len(df.columns))]
fig, ax = plt.subplots(figsize=figsize)
ax.stackplot(
pct.index,
[pct[col] for col in pct.columns],
labels=list(pct.columns),
colors=colors,
alpha=0.75,
)
ax.yaxis.set_major_formatter(FuncFormatter(lambda y, _: f"{y:.0%}"))
ax.xaxis.set_major_formatter(DateFormatter(date_fmt))
fig.autofmt_xdate()
ax.set_title(
f"{title} (Rolling {days_of_smoothing}d)",
fontsize=14, fontweight="bold",
)
ax.set_ylabel(y_label)
ax.set_ylim(0, 1)
ax.legend(loc=legend_loc, fontsize=8, ncol=3)
ax.grid(visible=True, alpha=0.3)
plt.tight_layout()
plt.show()
[docs]
def last_day_decomposition(self, title: str) -> None:
"""
Plot a horizontal bar chart of cumulative factor decomposition.
Shows each factor's percentage contribution to total cumulative
return at the final date, colored green (positive) or red (negative).
Parameters
----------
title : str
Chart title. Currently overridden internally to
``"Factor Contributions to Strategy Returns"``.
"""
total_pct = self.cummulative_pct_decomp()
sorted_items = sorted(total_pct.items(), key=lambda x: x[1])
names = [item[0] for item in sorted_items]
values = [item[1] for item in sorted_items]
colors = ['red' if v < 0 else 'green' for v in values]
plt.figure(figsize=(12, 8))
bars = plt.barh(names, values, color=colors)
for bar, value in zip(bars, values, strict=False):
plt.text(
bar.get_width() + (0.001 if value >= 0 else -0.001),
bar.get_y() + bar.get_height() / 2,
f"{value:.2%}",
va='center',
ha='left' if value >= 0 else 'right',
)
plt.title("Factor Contributions to Strategy Returns")
plt.xlabel("Contribution (%)")
plt.tight_layout()
plt.show()
