"""
Attribution Dashboard — Main orchestrator class.
Coordinates all dashboard components, layouts, callbacks, and visualizations.
Follows the same pattern as kaxanuk.backtest_engine.dashboard.backtest_dashboard.
"""
import base64
import logging
from pathlib import Path
from typing import TypedDict
import dash
import pyarrow as pa
from kaxanuk.attribution_analysis.dashboard.callbacks import register_callbacks
from kaxanuk.attribution_analysis.dashboard.components import create_header_with_logo
from kaxanuk.attribution_analysis.dashboard.constants import EXTERNAL_STYLESHEETS
from kaxanuk.attribution_analysis.dashboard.layouts import create_main_layout
from kaxanuk.attribution_analysis.dashboard.styles import INDEX_STRING
from kaxanuk.attribution_analysis.modules.attribution_visualizers_arrow import (
create_factor_attribution_summary_chart,
create_decomposition_bar_chart,
)
logger = logging.getLogger(__name__)
[docs]
class AttributionResultDict(TypedDict, total=False):
"""
Pre-calculated attribution results consumed by :class:`AttributionDashboard`.
Attributes
----------
portfolio_returns : pyarrow.Table
Columns: ``[date, portfolio_returns]``.
simulated_rets : pyarrow.Table
Columns: ``[date, factor1, factor2, ...]``.
portfolio_attribution_ts : pyarrow.Table
Columns: ``[date, factor1, factor2, ...]``.
pct_df_returns : pyarrow.Table
Columns: ``[date, factor1, factor2, ...]``.
total_pct_decomposition : dict
Mapping of ``{factor_name: contribution_fraction}``.
summary_stats : dict
Mapping of ``{metric_name: value}``.
brinson_fachler_indexes : pyarrow.Table or None
Brinson-Fachler attribution index table, if computed.
portfolio_name : str
Display name of the portfolio.
benchmark_name : str
Display name of the benchmark.
"""
portfolio_returns: pa.Table
simulated_rets: pa.Table
portfolio_attribution_ts: pa.Table
pct_df_returns: pa.Table
total_pct_decomposition: dict
summary_stats: dict
brinson_fachler_indexes: pa.Table | None
portfolio_name: str
benchmark_name: str
[docs]
class AttributionDashboard:
"""
Interactive Dash-based dashboard for attribution analysis.
Metrics are pre-calculated by the engine and passed in rather than
computed inside the dashboard.
Parameters
----------
attribution_results : AttributionResultDict
Pre-calculated attribution results to display.
portfolio_name : str or None, default None
Display name of the portfolio, shown in the dashboard header.
benchmark_name : str or None, default None
Display name of the benchmark.
logo_path : str or None, default None
Path to a logo image file to embed in the dashboard header.
days_of_smoothing : int, default 45
Number of days used for smoothing in relevant chart components.
date_column : str, default "date"
Name of the date column shared across the attribution result
tables.
Attributes
----------
results : AttributionResultDict
The raw attribution results passed to the constructor.
date_column : str
Name of the date column shared across the attribution result
tables.
portfolio_returns : pyarrow.Table
Columns: ``[date, portfolio_returns]``.
simulated_rets : pyarrow.Table
Columns: ``[date, factor1, factor2, ...]``.
portfolio_attribution_ts : pyarrow.Table
Columns: ``[date, factor1, factor2, ...]``.
pct_df_returns : pyarrow.Table
Columns: ``[date, factor1, factor2, ...]``.
total_pct_decomposition : dict
Mapping of ``{factor_name: contribution_fraction}``.
summary_stats : dict
Mapping of ``{metric_name: value}``.
brinson_fachler_indexes : pyarrow.Table or None
Brinson-Fachler attribution index table, if computed.
portfolio_name : str or None
Display name of the portfolio.
benchmark_name : str or None
Display name of the benchmark.
logo_path : str or None
Path to the logo image file, if any.
days_of_smoothing : int
Number of days used for smoothing in relevant chart components.
app : dash.Dash
The underlying Dash application instance.
"""
def __init__(
self,
attribution_results: AttributionResultDict,
portfolio_name: str | None = None,
benchmark_name: str | None = None,
logo_path: str | None = None,
days_of_smoothing: int = 45,
date_column: str = "date",
) -> None:
self.results = attribution_results
self.date_column = date_column
self.portfolio_returns = attribution_results['portfolio_returns']
self.simulated_rets = attribution_results['simulated_rets']
self.portfolio_attribution_ts = attribution_results['portfolio_attribution_ts']
self.pct_df_returns = attribution_results['pct_df_returns']
self.total_pct_decomposition = attribution_results['total_pct_decomposition']
self.summary_stats = attribution_results['summary_stats']
self.brinson_fachler_indexes = attribution_results.get('brinson_fachler_indexes')
self.portfolio_name = portfolio_name
self.benchmark_name = benchmark_name
self.logo_path = logo_path
self.days_of_smoothing = days_of_smoothing
self.app = dash.Dash(__name__, external_stylesheets=EXTERNAL_STYLESHEETS)
self._setup_layout()
self._setup_callbacks()
def _encode_logo(self) -> str | None:
"""
Encode the configured logo image as a base64 data URI.
Returns
-------
str or None
A ``"data:image/jpeg;base64,..."`` data URI if `logo_path` is
set, points to an existing file, and was read successfully.
Returns None if no logo is configured, the file doesn't
exist, or reading it fails.
"""
if self.logo_path:
logo_path = Path(self.logo_path)
if logo_path.exists():
try:
with logo_path.open("rb") as image_file:
encoded_string = base64.b64encode(image_file.read()).decode()
return f"data:image/jpeg;base64,{encoded_string}"
except Exception as e:
logger.warning("Could not encode logo: %s", e)
return None
return None
def _calculate_key_metrics(self) -> dict:
"""
Extract the headline metrics shown in the dashboard's metric cards.
Returns
-------
dict
Mapping with keys ``total_return``, ``annualized_return``,
``sharpe_ratio``, ``max_drawdown``, ``annualized_vol``, and
``skewness``, each a float pulled from `summary_stats`
(defaulting to ``0.0`` if the corresponding key is missing).
"""
stats = self.summary_stats
def _get_stat(key: str, default: float = 0.0) -> float:
return float(stats.get(key, default))
return {
'total_return': _get_stat('Annualized Return'),
'annualized_return': _get_stat('Annualized Return'),
'sharpe_ratio': _get_stat('Sharpe Ratio'),
'max_drawdown': _get_stat('Max Drawdown'),
'annualized_vol': _get_stat('Annualized Vol'),
'skewness': _get_stat('Skewness'),
}
def _setup_layout(self) -> None:
"""
Build and assign the Dash application's layout.
Computes the key metrics, encodes the logo, derives the
portfolio's date range, and assembles the header, attribution
summary chart, and decomposition bar chart into `self.app.layout`.
"""
metrics = self._calculate_key_metrics()
logo_src = self._encode_logo()
date_col = self.portfolio_returns.column(self.date_column)
start_date = date_col[0].as_py().strftime('%Y-%m-%d')
end_date = date_col[-1].as_py().strftime('%Y-%m-%d')
header_content = create_header_with_logo(
logo_src,
self.portfolio_name or "Portfolio",
start_date,
end_date,
)
attribution_summary_chart = create_factor_attribution_summary_chart(
self.simulated_rets,
date_column=self.date_column,
)
decomposition_bar = create_decomposition_bar_chart(
self.total_pct_decomposition,
)
self.app.layout = create_main_layout(
header_content=header_content,
metrics=metrics,
attribution_summary_chart=attribution_summary_chart,
decomposition_bar_chart=decomposition_bar,
)
def _setup_callbacks(self) -> None:
"""Register all interactive Dash callbacks on `self.app`."""
register_callbacks(self.app, self)
[docs]
def run_server(
self,
*,
debug: bool = True,
port: int = 8050,
) -> None:
"""
Start the Dash development server.
Parameters
----------
debug : bool, default True
Whether to run the Dash app in debug mode.
port : int, default 8050
Local port on which to serve the dashboard.
"""
self.app.index_string = INDEX_STRING
logger.info("Dashboard will be available at: http://localhost:%s", port)
self.app.run(debug=debug, port=port, host='127.0.0.1', use_reloader=False)
[docs]
def create_dashboard_from_attribution(
attribution_results: AttributionResultDict,
portfolio_name: str | None = None,
benchmark_name: str | None = None,
logo_path: str | None = None,
days_of_smoothing: int = 45,
date_column: str = "date",
) -> AttributionDashboard:
"""
Create an :class:`AttributionDashboard` from pre-calculated results.
Parameters
----------
attribution_results : AttributionResultDict
Pre-calculated attribution results to display.
portfolio_name : str or None, default None
Display name of the portfolio, shown in the dashboard header.
benchmark_name : str or None, default None
Display name of the benchmark.
logo_path : str or None, default None
Path to a logo image file to embed in the dashboard header.
days_of_smoothing : int, default 45
Number of days used for smoothing in relevant chart components.
date_column : str, default "date"
Name of the date column shared across the attribution result
tables.
Returns
-------
AttributionDashboard
A dashboard instance configured with the given results and
options, ready to be served with :meth:`AttributionDashboard.run_server`.
"""
return AttributionDashboard(
attribution_results=attribution_results,
portfolio_name=portfolio_name,
benchmark_name=benchmark_name,
logo_path=logo_path,
days_of_smoothing=days_of_smoothing,
date_column=date_column,
)
