Source code for kaxanuk.attribution_analysis.dashboard.callbacks

"""
Dashboard callbacks for interactivity.
"""

import logging
import typing

import dash
from dash import Input, Output, html, dcc

from kaxanuk.attribution_analysis.dashboard.components import (
    format_metric_value,
    create_statistics_table,
)
from kaxanuk.attribution_analysis.dashboard.constants import (
    PROFESSIONAL_COLORS,
    METRIC_ORDER,
)
from kaxanuk.attribution_analysis.modules.attribution_visualizers_arrow import (
    create_factor_attribution_chart,
    create_stacked_decomposition_chart,
    create_brinson_fachler_chart,
    create_returns_distribution,
    create_drawdown_chart,
)

if typing.TYPE_CHECKING:
    from kaxanuk.attribution_analysis.dashboard.attribution_dashboard import AttributionDashboard

logger = logging.getLogger(__name__)


[docs] def register_callbacks(app: dash.Dash, dashboard_instance: "AttributionDashboard") -> None: """ Register all interactive Dash callbacks on the application. Wires up two callbacks: one that (re)builds the performance statistics table from `dashboard_instance.summary_stats`, and one that renders the chart for whichever analysis tab is active, reading its data from `dashboard_instance`. Both callbacks catch any exception and render an error message in place of the chart or table rather than letting the callback fail. Parameters ---------- app : dash.Dash The Dash application to register callbacks on. dashboard_instance : AttributionDashboard The dashboard instance whose precomputed data (returns, attribution, summary stats, etc.) backs the callbacks. """ @app.callback( Output('performance-stats-table', 'children'), [Input('attribution-summary-chart', 'id')], ) def create_performance_stats_table(_): try: summary = dashboard_instance.summary_stats stats_data = [] for metric, value in summary.items(): formatted = format_metric_value(metric, value) stats_data.append({ 'Metric': metric, 'Value': formatted, }) def get_sort_key(item): try: return METRIC_ORDER.index(item['Metric']) except ValueError: return len(METRIC_ORDER) stats_data.sort(key=get_sort_key) return create_statistics_table( stats_data, dashboard_instance.portfolio_name or "Strategy", ) except Exception as e: logger.error("Error creating statistics table: %s", e) return html.P( f"Error creating statistics table: {e!s}", style={"color": PROFESSIONAL_COLORS["danger"]}, ) @app.callback( Output('tab-content', 'children'), [Input('analysis-tabs', 'value')], ) def render_tab_content(active_tab): try: date_column = dashboard_instance.date_column if active_tab == 'factor-attribution-tab': fig = create_factor_attribution_chart( dashboard_instance.simulated_rets, date_column=date_column, ) return dcc.Graph(figure=fig) elif active_tab == 'decomposition-tab': fig = create_stacked_decomposition_chart( dashboard_instance.pct_df_returns, date_column=date_column, days_of_smoothing=dashboard_instance.days_of_smoothing, ) return dcc.Graph(figure=fig) elif active_tab == 'brinson-fachler-tab': if dashboard_instance.brinson_fachler_indexes is not None: fig = create_brinson_fachler_chart( dashboard_instance.brinson_fachler_indexes, date_column=date_column, ) return dcc.Graph(figure=fig) return html.P( "Brinson-Fachler analysis not available. " "Enable brinson_fachler_method in configuration.", className="placeholder-text", ) elif active_tab == 'returns-tab': fig = create_returns_distribution( dashboard_instance.portfolio_returns, date_column=date_column, ) return dcc.Graph(figure=fig) elif active_tab == 'drawdown-tab': fig = create_drawdown_chart( dashboard_instance.portfolio_returns, date_column=date_column, ) return dcc.Graph(figure=fig) except Exception as e: logger.error("Error rendering tab '%s': %s", active_tab, e) return html.P( f"Error loading analysis: {e!s}", style={"color": PROFESSIONAL_COLORS["danger"]}, ) return html.P("Select a tab to view analysis.")