"""
Performance Attribution Pipeline
=================================
Main orchestration module for running multi-method portfolio performance
attribution analysis.
This module provides the top-level ``main`` function that coordinates the
full attribution workflow: loading market and portfolio data, aligning date
ranges, executing Brinson-Fachler and/or factor-model attribution, and
optionally launching an interactive dashboard.
Notes
-----
The pipeline currently supports two attribution methodologies that can be
enabled independently via the ``Configuration`` object:
* **Brinson-Fachler** -- decomposes active return into allocation and
selection effects using sector/asset weights.
* **Factor Model** -- regresses portfolio returns against
user-supplied factor return series and decomposes performance into
factor exposures plus a residual (alpha).
All intermediate computations use PyArrow columnar tables for memory
efficiency and vectorized operations.
Examples
--------
>>> from kaxanuk.attribution_analysis.entities.configuration import Configuration
>>> from kaxanuk.attribution_analysis.performance_attribution import main
>>>
>>> config = Configuration(
... input_directory="data/input",
... weights_portfolio_directory="weights/portfolio",
... weights_benchmark_directory="weights/benchmark",
... benchmark_returns_directory="returns/benchmark",
... investable_assets_directory="data/prices",
... factor_returns_by_factor_directory="data/factors",
... portfolio_file_name="portfolio_weights.csv",
... benchmark_file_name="benchmark_weights.csv",
... benchmark_return_file_name="benchmark_returns.csv",
... user_column_date="date",
... user_column_price="close",
... market_data_input_format="csv",
... portfolio_input_format="csv",
... start_date="auto",
... end_date="auto",
... brinson_fachler_method=True,
... factor_model_method=True,
... )
>>> main(config, launch_dashboard=True, dashboard_port=8050)
"""
import logging
import pyarrow as pa
import pyarrow.compute as pc
from pathlib import Path
from kaxanuk.attribution_analysis.interfaces import (
FactorModelArrowAttribution,
BrinstonFachlerArrowAttribution
)
from kaxanuk.attribution_analysis.entities.configuration import Configuration
from kaxanuk.attribution_analysis.input_handlers.csv_input import CsvInput
from kaxanuk.attribution_analysis.input_handlers.parquet_input import ParquetInput
from kaxanuk.attribution_analysis.input_handlers.csv_portfolio_input_handler import CsvPortfolioInputHandler
from kaxanuk.attribution_analysis.input_handlers.excel_portfolio_input_handler import ExcelPortfolioInputHandler
from kaxanuk.attribution_analysis.market_data_pipeline.market_data_service import MarketDataService
from kaxanuk.attribution_analysis.market_data_pipeline.field_definitions import StandardField
from kaxanuk.attribution_analysis.portfolio_data_pipeline.portfolio_service import PortfolioService
from kaxanuk.attribution_analysis.entities.returns_table import ReturnsTable
from kaxanuk.attribution_analysis.entities.portfolio_weights_arrow import PortfolioWeights
from kaxanuk.attribution_analysis.entities.factor_returns import FactorReturns
from kaxanuk.attribution_analysis.modules.table_conversion_arrow import (
pyarrow_prices_to_attribution_format,
portfolio_table_to_arrow,
returns_table_to_arrow,
compute_returns_from_prices_arrow,
_filter_arrow_table_by_date_range,
)
from kaxanuk.attribution_analysis.modules.data_wrangling import (
load_by_factor_returns_arrow,
)
from kaxanuk.attribution_analysis.modules.performance_functions import (
summary_stats,
)
from kaxanuk.attribution_analysis.services.license_validator import validate_license
# Note: the Dash dashboard is imported lazily inside ``main`` (only when
# ``launch_dashboard=True``) so that importing this module — and therefore
# ``import kaxanuk.attribution_analysis`` — does not pull in Dash.
logger = logging.getLogger("kaxanuk.attribution_analysis")
[docs]
def main(
configuration: Configuration,
*,
launch_dashboard: bool = True,
dashboard_port: int,
) -> None:
"""
Run the full performance attribution pipeline.
Orchestrates data loading, date-range alignment, attribution
computation, and optional dashboard visualisation for a portfolio
defined by ``configuration``.
Parameters
----------
configuration : Configuration
Frozen dataclass holding all runtime settings including:
* **Directory paths** -- ``input_directory``,
``weights_portfolio_directory``, ``weights_benchmark_directory``,
``benchmark_returns_directory``, ``investable_assets_directory``,
``factor_returns_by_factor_directory``.
* **File names** -- ``portfolio_file_name``,
``benchmark_file_name``, ``benchmark_return_file_name``.
* **Column mappings** -- ``user_column_date``,
``user_column_price``.
* **Input formats** -- ``market_data_input_format``
(``"csv"`` | ``"parquet"``), ``portfolio_input_format``
(``"csv"`` | ``"excel"``).
* **Date range** -- ``start_date``, ``end_date``
(``datetime.date`` | ``"auto"``).
* **Method flags** -- ``brinson_fachler_method``,
``factor_model_method`` (both ``bool``).
launch_dashboard : bool, default True
If ``True`` and the factor-model method is enabled, launch a Dash
web application displaying interactive attribution charts and
summary statistics. When ``False``, static matplotlib plots are
generated instead.
dashboard_port : int
TCP port on which the Dash server listens when
``launch_dashboard=True``. Required (no default); typically read
from the configuration via ``ExcelConfigurator.get_dashboard_port()``.
Returns
-------
None
All outputs are side-effects: log messages, plots, and/or a
running Dash server.
Raises
------
LicenseError
Raised before anything is read when the license does not validate
(``services.license_validator.validate_license()``): one of
``LicenseNotFoundError``, ``LicenseValidationError``,
``LicenseServerError`` or ``LicenseConfigurationError``.
FileNotFoundError
If any configured data directory or file does not exist.
ConfigurationError
If ``configuration`` contains invalid field values (raised during
``Configuration.__post_init__``).
Notes
-----
The function aligns all input tables (asset returns, portfolio
weights, benchmark weights, benchmark returns, and factor returns) to
the intersection of their date ranges before running any attribution
calculation. This prevents shape mismatches and ensures consistent
time-series indexing across methods.
The pipeline proceeds in four stages:
1. **Data loading** -- market prices, portfolio/benchmark weights,
benchmark returns, and per-factor return series are read from disk
using the appropriate input handlers.
2. **Date alignment** -- the common start/end dates across all tables
are computed and each table is filtered accordingly.
3. **Attribution** -- enabled methods (Brinson-Fachler, Factor Model) are
executed sequentially.
4. **Visualisation** -- results are either plotted (matplotlib) or
served via an interactive Dash dashboard.
Examples
--------
Run attribution without the dashboard (useful in batch/CI contexts):
>>> main(config, launch_dashboard=False)
Run only the Brinson-Fachler method with the dashboard disabled:
>>> config = Configuration(..., brinson_fachler_method=True,
... factor_model_method=False)
>>> main(config, launch_dashboard=False)
"""
validate_license()
input_route = Path(configuration.input_directory)
# --- Build input handlers ---
if configuration.market_data_input_format == "parquet":
market_handlers = [
ParquetInput(input_dir=configuration.investable_assets_directory)
]
else:
market_handlers = [
CsvInput(input_dir=configuration.investable_assets_directory)
]
if configuration.portfolio_input_format == "excel":
portfolio_handlers = [
ExcelPortfolioInputHandler(base_dir=str(input_route / configuration.weights_portfolio_directory))
]
benchmark_weight_handlers = [
ExcelPortfolioInputHandler(base_dir=str(input_route / configuration.weights_benchmark_directory))
]
else:
portfolio_handlers = [
CsvPortfolioInputHandler(base_dir=str(input_route / configuration.weights_portfolio_directory))
]
benchmark_weight_handlers = [
CsvPortfolioInputHandler(base_dir=str(input_route / configuration.weights_benchmark_directory))
]
benchmark_return_handlers = [
CsvPortfolioInputHandler(base_dir=str(input_route / configuration.benchmark_returns_directory))
]
market_data_service = MarketDataService()
portfolio_service = PortfolioService()
# --- Load data ---
# The small weight/returns files (portfolio, benchmark weights, benchmark
# returns) are loaded and date-validated first, against the configured
# start/end dates. This fails fast on a misaligned date range before the
# (often much heavier) per-ticker market data and factor returns are read.
logger.info("Loading data")
portfolio_weights_table = portfolio_service.load_portfolio(
portfolio_handlers=portfolio_handlers,
file_name=configuration.portfolio_file_name,
start_date=configuration.start_date,
end_date=configuration.end_date,
)
benchmark_weights_table = portfolio_service.load_portfolio(
portfolio_handlers=benchmark_weight_handlers,
file_name=configuration.benchmark_file_name,
start_date=configuration.start_date,
end_date=configuration.end_date,
)
benchmark_returns_table = portfolio_service.load_portfolio(
portfolio_handlers=benchmark_return_handlers,
file_name=configuration.benchmark_return_file_name,
start_date=configuration.start_date,
end_date=configuration.end_date,
)
portfolio_weights_arrow = portfolio_table_to_arrow(portfolio_weights_table)
benchmark_weights_arrow = portfolio_table_to_arrow(benchmark_weights_table)
benchmark_returns_arrow = returns_table_to_arrow(benchmark_returns_table)
tickers = set(portfolio_weights_arrow.asset_names)
prices_table = market_data_service.load_and_concat_prices(
input_handlers=market_handlers,
tickers=tickers,
configuration=configuration,
)
prices_arrow = pyarrow_prices_to_attribution_format(prices_table, date_column=StandardField.DATE.value)
returns_investible_assets_arrow = compute_returns_from_prices_arrow(prices_arrow, date_column="date")
available_factors = [
entry.name
for entry in Path(configuration.factor_returns_by_factor_directory).iterdir()
]
factor_start_date = None if configuration.start_date == "auto" else configuration.start_date
factor_end_date = None if configuration.end_date == "auto" else configuration.end_date
by_factor_factor_returns_arrow = load_by_factor_returns_arrow(
path=configuration.factor_returns_by_factor_directory,
available_factors=available_factors,
portfolio_weights=portfolio_weights_arrow,
start_date=factor_start_date,
end_date=factor_end_date,
)
# --- Align all tables to the same date range ---
_date_col = "date"
all_start_dates = [
pc.min(returns_investible_assets_arrow.table.column(_date_col)).as_py(),
pc.min(portfolio_weights_arrow.table.column(_date_col)).as_py(),
pc.min(benchmark_weights_arrow.table.column(_date_col)).as_py(),
pc.min(benchmark_returns_arrow.table.column(_date_col)).as_py(),
*(pc.min(fr.table.column(_date_col)).as_py() for fr in by_factor_factor_returns_arrow.values()),
]
all_end_dates = [
pc.max(returns_investible_assets_arrow.table.column(_date_col)).as_py(),
pc.max(portfolio_weights_arrow.table.column(_date_col)).as_py(),
pc.max(benchmark_weights_arrow.table.column(_date_col)).as_py(),
pc.max(benchmark_returns_arrow.table.column(_date_col)).as_py(),
*(pc.max(fr.table.column(_date_col)).as_py() for fr in by_factor_factor_returns_arrow.values()),
]
common_start = max(all_start_dates)
common_end = min(all_end_dates)
common_start_arrow = pa.scalar(common_start, type=pa.timestamp("ns"))
common_end_arrow = pa.scalar(common_end, type=pa.timestamp("ns"))
returns_investible_assets_arrow = ReturnsTable(
table=_filter_arrow_table_by_date_range(
returns_investible_assets_arrow.table, common_start_arrow, common_end_arrow
)
)
portfolio_weights_arrow = PortfolioWeights(
table=_filter_arrow_table_by_date_range(
portfolio_weights_arrow.table, common_start_arrow, common_end_arrow
)
)
benchmark_weights_arrow = PortfolioWeights(
table=_filter_arrow_table_by_date_range(
benchmark_weights_arrow.table, common_start_arrow, common_end_arrow
)
)
benchmark_returns_arrow = ReturnsTable(
table=_filter_arrow_table_by_date_range(
benchmark_returns_arrow.table, common_start_arrow, common_end_arrow
)
)
by_factor_factor_returns_arrow = {
factor: FactorReturns(
table=_filter_arrow_table_by_date_range(fr.table, common_start_arrow, common_end_arrow),
factor_name=factor,
)
for factor, fr in by_factor_factor_returns_arrow.items()
}
logger.info(
"Aligned all tables to common date range: %s to %s",
common_start.strftime('%Y-%m-%d'),
common_end.strftime('%Y-%m-%d'),
)
# --- Brinson-Fachler attribution ---
brinson_fachler_indexes_arrow = None
if configuration.brinson_fachler_method:
logger.info("Running Brinson-Fachler attribution")
brinston_fachler_arrow_result = BrinstonFachlerArrowAttribution(
returns_investable_assets=returns_investible_assets_arrow.table,
complete_portfolio_weights=portfolio_weights_arrow.table,
complete_benchmark_weights=benchmark_weights_arrow.table,
date_column=_date_col,
)
brinston_fachler_arrow_result.time_series_calculation()
brinston_fachler_arrow_result.attribution_plots()
brinson_fachler_indexes_arrow = brinston_fachler_arrow_result.brinston_fach_indexes
# --- Factor model attribution ---
factor_model_arrow = None
if configuration.factor_model_method:
logger.info("Running factor model attribution")
factor_model_arrow = FactorModelArrowAttribution(
daily_portfolio_weights=portfolio_weights_arrow.table,
by_factor_factor_returns={
name: fr.table for name, fr in by_factor_factor_returns_arrow.items()
},
asset_returns=returns_investible_assets_arrow.table,
benchmark_returns=benchmark_returns_arrow.table,
date_column=_date_col,
)
factor_model_arrow.run()
if not launch_dashboard:
logger.info("Plotting factor model results")
factor_model_arrow.attribution_plots(title='Factor Model Attribution')
# --- Dashboard ---
if launch_dashboard and factor_model_arrow is not None:
from kaxanuk.attribution_analysis.dashboard.attribution_dashboard import ( # noqa: PLC0415
create_dashboard_from_attribution,
)
logger.info("Preparing dashboard data")
factor_model_arrow.calc_pct_area()
total_pct_decomposition = factor_model_arrow.cummulative_pct_decomp()
pr_series = factor_model_arrow.portfolio_returns.column("portfolio_returns").to_pandas()
stats_df = summary_stats(pr_series)
stats_dict = {col: float(stats_df[col].iloc[0]) for col in stats_df.columns}
attribution_results = {
'portfolio_returns': factor_model_arrow.portfolio_returns,
'simulated_rets': factor_model_arrow.simulated_rets,
'portfolio_attribution_ts': factor_model_arrow.portfolio_attribution_ts,
'pct_df_returns': factor_model_arrow.pct_df_returns,
'total_pct_decomposition': total_pct_decomposition,
'summary_stats': stats_dict,
'brinson_fachler_indexes': brinson_fachler_indexes_arrow,
}
dashboard = create_dashboard_from_attribution(
attribution_results=attribution_results,
portfolio_name=getattr(configuration, 'portfolio_file_name', 'Portfolio'),
benchmark_name=getattr(configuration, 'benchmark_file_name', None),
days_of_smoothing=45,
)
logger.info("Launching dashboard on port %s", dashboard_port)
dashboard.run_server(port=dashboard_port)
