"""
Attribution visualization module — Plotly charts using PyArrow tables.
Provides chart-creation functions that return Plotly Figure objects
for use in the Dash dashboard, without any pandas dependency.
All data extraction uses pa.Table / pyarrow.compute and plain Python lists.
"""
import math
import plotly.graph_objects as go
import pyarrow as pa
from plotly.subplots import make_subplots
from kaxanuk.attribution_analysis.dashboard.chart_colors import CHART_COLORS
from kaxanuk.attribution_analysis.dashboard.constants import PROFESSIONAL_COLORS
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _ffill(values: list) -> list:
"""Forward-fill None / NaN values in a Python list."""
result = []
last_valid = 0.0
for v in values:
if v is None or (isinstance(v, float) and math.isnan(v)):
result.append(last_valid)
else:
last_valid = v
result.append(v)
return result
def _rolling_mean(values: list, window: int) -> list:
"""Simple rolling mean. Returns None for the leading warm-up period."""
n = len(values)
result = [None] * n
if window < 1 or n == 0:
return result
running_sum = 0.0
count = 0
for i in range(n):
v = values[i] if values[i] is not None else 0.0
running_sum += v
count += 1
if count > window:
old = values[i - window] if values[i - window] is not None else 0.0
running_sum -= old
count -= 1
if count >= window:
result[i] = running_sum / window
return result
def _get_non_date_columns(table: pa.Table, date_column: str) -> list[str]:
return [c for c in table.column_names if c != date_column]
def _extract_clean_series(
table: pa.Table,
date_column: str,
value_column: str,
) -> tuple[list, list]:
"""Return (dates, values) with null rows removed."""
dates = table.column(date_column).to_pylist()
values = table.column(value_column).to_pylist()
clean_dates, clean_values = [], []
for d, v in zip(dates, values, strict=False):
if v is not None:
clean_dates.append(d)
clean_values.append(v)
return clean_dates, clean_values
# ---------------------------------------------------------------------------
# Chart functions
# ---------------------------------------------------------------------------
[docs]
def create_factor_attribution_chart(
simulated_rets: pa.Table,
date_column: str = "date",
) -> go.Figure:
"""Create the main 3-panel factor attribution line chart.
Replicates FactorModelAttribution.attribution_plots() in Plotly.
Panels:
1. Total / Factor / Idiosyncratic cumulative returns
2. Style Factors
3. Industry Returns (non-zero columns only)
"""
dates = simulated_rets.column(date_column).to_pylist()
all_cols = _get_non_date_columns(simulated_rets, date_column)
totals_cols = [c for c in ["f_total_excess_returns", "f_total_factor_returns", "f_idyo_returns"] if c in all_cols]
style_cols = [c for c in ["f_size", "f_momentum", "f_beta", "f_residual volatility", "f_value"] if c in all_cols]
industry_candidates = [
"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",
]
industry_cols = []
for c in industry_candidates:
if c in all_cols:
col_values = simulated_rets.column(c).to_pylist()
if any(v != 0 and v is not None for v in col_values):
industry_cols.append(c)
fig = make_subplots(
rows=3, cols=1,
subplot_titles=(
"Total / Factor / Idiosyncratic",
"Style Factors",
"Industry Returns",
),
shared_xaxes=True,
vertical_spacing=0.08,
)
for color_idx, col in enumerate(totals_cols):
fig.add_trace(
go.Scatter(
x=dates, y=simulated_rets.column(col).to_pylist(),
mode='lines', name=col,
line={'color': CHART_COLORS[color_idx % len(CHART_COLORS)], 'width': 2},
hovertemplate=f'<b>{col}</b><br>Date: %{{x}}<br>Return: %{{y:.4f}}<extra></extra>',
),
row=1, col=1,
)
for color_idx, col in enumerate(style_cols):
fig.add_trace(
go.Scatter(
x=dates, y=simulated_rets.column(col).to_pylist(),
mode='lines', name=col,
line={'color': CHART_COLORS[color_idx % len(CHART_COLORS)], 'width': 2},
hovertemplate=f'<b>{col}</b><br>Date: %{{x}}<br>Return: %{{y:.4f}}<extra></extra>',
),
row=2, col=1,
)
for color_idx, col in enumerate(industry_cols):
fig.add_trace(
go.Scatter(
x=dates, y=simulated_rets.column(col).to_pylist(),
mode='lines', name=col,
line={'color': CHART_COLORS[color_idx % len(CHART_COLORS)], 'width': 1.5},
hovertemplate=f'<b>{col}</b><br>Date: %{{x}}<br>Return: %{{y:.4f}}<extra></extra>',
),
row=3, col=1,
)
for row in range(1, 4):
fig.add_hline(y=0, line_dash="solid", line_color="black", line_width=1, row=row, col=1)
fig.update_yaxes(tickformat='.1%', row=row, col=1)
fig.update_layout(
height=900,
template='plotly_white',
hovermode='x unified',
legend={
'orientation': 'v',
'yanchor': 'top', 'y': 1,
'xanchor': 'left', 'x': 1.02,
'font': {'size': 10, 'family': 'Segoe UI'},
},
margin={'l': 60, 'r': 160, 't': 60, 'b': 40},
)
return fig
[docs]
def create_factor_attribution_summary_chart(
simulated_rets: pa.Table,
date_column: str = "date",
) -> go.Figure:
"""Create a compact attribution chart showing just the 3 main series.
Used in the main charts row (large chart).
"""
dates = simulated_rets.column(date_column).to_pylist()
all_cols = _get_non_date_columns(simulated_rets, date_column)
cols = [c for c in ["f_total_excess_returns", "f_total_factor_returns", "f_idyo_returns"] if c in all_cols]
fig = go.Figure()
for i, col in enumerate(cols):
fig.add_trace(go.Scatter(
x=dates, y=simulated_rets.column(col).to_pylist(),
mode='lines', name=col,
line={'color': CHART_COLORS[i], 'width': 2},
hovertemplate=f'<b>{col}</b><br>Date: %{{x}}<br>Return: %{{y:.4f}}<extra></extra>',
))
fig.add_hline(y=0, line_dash="solid", line_color="black", line_width=1)
fig.update_layout(
title="Cumulative Factor Attribution",
xaxis_title="Date",
yaxis_title="Cumulative Return",
yaxis={'tickformat': '.1%'},
template='plotly_white',
height=500,
autosize=True,
hovermode='x unified',
legend={'yanchor': "top", 'y': 0.99, 'xanchor': "left", 'x': 0.01},
)
return fig
[docs]
def create_stacked_decomposition_chart(
pct_df_returns: pa.Table,
date_column: str = "date",
days_of_smoothing: int = 45,
) -> go.Figure:
"""Create a normalized stacked area chart of factor return decomposition.
Each factor's share = abs(factor) / sum(abs(all factors)), always sums to 100%.
Implemented with pure pyarrow + Python lists (no pandas).
"""
factor_cols = _get_non_date_columns(pct_df_returns, date_column)
dates = pct_df_returns.column(date_column).to_pylist()
n_rows = len(dates)
# Extract and forward-fill each column
columns_data = {}
for col in factor_cols:
columns_data[col] = _ffill(pct_df_returns.column(col).to_pylist())
# Compute absolute values and row-wise sums
abs_total = [0.0] * n_rows
abs_columns = {}
for col in factor_cols:
abs_vals = [abs(v) if v is not None else 0.0 for v in columns_data[col]]
abs_columns[col] = abs_vals
for i in range(n_rows):
abs_total[i] += abs_vals[i]
# Normalize: each column = abs(col) / abs_total
pct_columns = {}
for col in factor_cols:
pct_columns[col] = [
abs_columns[col][i] / abs_total[i] if abs_total[i] != 0 else 0.0
for i in range(n_rows)
]
# Rolling mean
smoothed = {}
for col in factor_cols:
raw = _rolling_mean(pct_columns[col], days_of_smoothing)
smoothed[col] = [v if v is not None else 0.0 for v in raw]
# Build chart
num_cols = len(factor_cols)
colors = [CHART_COLORS[i % len(CHART_COLORS)] for i in range(num_cols)]
fig = go.Figure()
for i, col in enumerate(factor_cols):
hex_color = colors[i]
rgb = tuple(int(hex_color[j:j + 2], 16) for j in (1, 3, 5))
rgba_fill = f'rgba({rgb[0]}, {rgb[1]}, {rgb[2]}, 0.75)'
fig.add_trace(go.Scatter(
x=dates,
y=smoothed[col],
mode='lines',
name=col,
stackgroup='one',
groupnorm='percent',
line={'width': 0.5, 'color': hex_color},
fillcolor=rgba_fill,
hovertemplate=f'<b>{col}</b><br>Date: %{{x}}<br>Share: %{{y:.1f}}%<extra></extra>',
))
fig.update_layout(
title=f"Return Decomposition (Rolling {days_of_smoothing}d)",
xaxis_title="Date",
yaxis_title="Share of Total Return",
yaxis={'tickformat': '.0%', 'range': [0, 100]},
template='plotly_white',
height=500,
hovermode='x unified',
legend={
'orientation': 'v',
'yanchor': 'top', 'y': 1,
'xanchor': 'left', 'x': 1.02,
'font': {'size': 10, 'family': 'Segoe UI'},
},
margin={'l': 60, 'r': 160, 't': 60, 'b': 40},
)
return fig
[docs]
def create_decomposition_bar_chart(
total_pct: dict,
) -> go.Figure:
"""Create a horizontal bar chart of end-of-period factor contributions.
Parameters
----------
total_pct : dict
{factor_name: contribution_fraction} — e.g. from
FactorModelArrowAttribution.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 = ['#DC2626' if v < 0 else '#059669' for v in values]
fig = go.Figure(go.Bar(
x=values,
y=names,
orientation='h',
marker_color=colors,
text=[f"{v:.2%}" for v in values],
textposition='outside',
hovertemplate='<b>%{y}</b><br>Contribution: %{x:.2%}<extra></extra>',
))
fig.update_layout(
title="Factor Contributions to Strategy Returns",
xaxis_title="Contribution",
xaxis={'tickformat': '.0%'},
template='plotly_white',
height=max(350, len(values) * 45 + 100),
autosize=True,
margin={'l': 180, 'r': 80, 't': 60, 'b': 40},
)
return fig
[docs]
def create_brinson_fachler_chart(
brinson_fachler_indexes: pa.Table,
date_column: str = "date",
) -> go.Figure:
"""Create the 3-panel Brinson-Fachler attribution chart.
Panels:
1. Strategy vs Benchmark Returns
2. Cumulative Alpha
3. Allocation, Selection, Interaction Effects
"""
dates = brinson_fachler_indexes.column(date_column).to_pylist()
all_cols = _get_non_date_columns(brinson_fachler_indexes, date_column)
fig = make_subplots(
rows=3, cols=1,
subplot_titles=(
"Strategy vs Benchmark Returns",
"Cumulative Alpha",
"Allocation, Selection and Interaction Effects",
),
shared_xaxes=True,
vertical_spacing=0.08,
)
for i, col in enumerate(['portfolio_returns', 'benchmark_returns']):
if col in all_cols:
fig.add_trace(
go.Scatter(
x=dates, y=brinson_fachler_indexes.column(col).to_pylist(),
mode='lines', name=col,
line={'color': CHART_COLORS[i], 'width': 2},
),
row=1, col=1,
)
if 'alpha' in all_cols:
fig.add_trace(
go.Scatter(
x=dates, y=brinson_fachler_indexes.column('alpha').to_pylist(),
mode='lines', name='alpha',
line={'color': CHART_COLORS[2], 'width': 2},
),
row=2, col=1,
)
for i, col in enumerate(['allocation', 'selection', 'interaction']):
if col in all_cols:
fig.add_trace(
go.Scatter(
x=dates, y=brinson_fachler_indexes.column(col).to_pylist(),
mode='lines', name=col,
line={'color': CHART_COLORS[i + 3], 'width': 2},
),
row=3, col=1,
)
for row in range(1, 4):
fig.add_hline(y=0, line_dash="solid", line_color="black", line_width=1, row=row, col=1)
fig.update_yaxes(tickformat='.2%', row=row, col=1)
fig.update_layout(
height=900,
template='plotly_white',
hovermode='x unified',
legend={
'orientation': 'v',
'yanchor': 'top', 'y': 1,
'xanchor': 'left', 'x': 1.02,
'font': {'size': 10, 'family': 'Segoe UI'},
},
margin={'l': 60, 'r': 160, 't': 60, 'b': 40},
)
return fig
[docs]
def create_returns_distribution(
portfolio_returns: pa.Table,
date_column: str = "date",
value_column: str = "portfolio_returns",
) -> go.Figure:
"""Create returns distribution analysis (histogram + time series)."""
clean_dates, clean_returns = _extract_clean_series(
portfolio_returns, date_column, value_column,
)
fig = make_subplots(
rows=1, cols=2,
subplot_titles=('Returns Distribution', 'Returns Over Time'),
column_widths=[0.4, 0.6],
)
fig.add_trace(
go.Histogram(
x=clean_returns,
nbinsx=50,
name='Returns Distribution',
marker={'color': PROFESSIONAL_COLORS["kaxanuk_azul"], 'opacity': 0.7},
hovertemplate='Return: %{x:.2%}<br>Count: %{y}<extra></extra>',
),
row=1, col=1,
)
fig.add_trace(
go.Scatter(
x=clean_dates,
y=clean_returns,
mode='lines',
name='Daily Returns',
line={'color': PROFESSIONAL_COLORS["kaxanuk_azul"], 'width': 1},
hovertemplate='Date: %{x}<br>Return: %{y:.2%}<extra></extra>',
),
row=1, col=2,
)
fig.add_hline(y=0, line_dash="dash", line_color="red", row=1, col=2)
fig.update_layout(
height=400,
template='plotly_white',
showlegend=False,
)
fig.update_xaxes(title_text="Return", row=1, col=1)
fig.update_yaxes(title_text="Frequency", row=1, col=1)
fig.update_xaxes(title_text="Date", row=1, col=2)
fig.update_yaxes(title_text="Return", row=1, col=2, tickformat='.1%')
return fig
[docs]
def create_drawdown_chart(
portfolio_returns: pa.Table,
date_column: str = "date",
value_column: str = "portfolio_returns",
) -> go.Figure:
"""Create drawdown analysis chart.
Computes cumulative returns, running max, and drawdown
entirely with Python lists (no pandas).
"""
clean_dates, clean_returns = _extract_clean_series(
portfolio_returns, date_column, value_column,
)
# Cumulative product: (1 + r_0) * (1 + r_1) * ...
cumulative = []
cum_val = 1.0
for r in clean_returns:
cum_val *= (1.0 + r)
cumulative.append(cum_val)
# Running (expanding) max
running_max = []
max_val = float('-inf')
for c in cumulative:
max_val = max(max_val, c)
running_max.append(max_val)
# Drawdown = cumulative / running_max - 1
drawdown = [
(c / m - 1.0) if m != 0 else 0.0
for c, m in zip(cumulative, running_max, strict=False)
]
fig = go.Figure()
fig.add_trace(go.Scatter(
x=clean_dates,
y=drawdown,
fill='tozeroy',
fillcolor='rgba(255, 0, 0, 0.3)',
line={'color': 'red', 'width': 2},
name='Portfolio Drawdown',
hovertemplate='Date: %{x}<br>Drawdown: %{y:.2%}<extra></extra>',
))
fig.add_hline(y=0, line_dash="solid", line_color="black", line_width=1)
fig.update_layout(
title="Drawdown Analysis",
xaxis_title="Date",
yaxis_title="Drawdown",
yaxis={'tickformat': '.1%'},
template='plotly_white',
height=400,
legend={
'orientation': 'h',
'yanchor': 'bottom', 'y': 1.02,
'xanchor': 'right', 'x': 1,
},
)
return fig
