import logging
import os
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from fredapi import Fred
from matplotlib.dates import DateFormatter
from matplotlib.ticker import FuncFormatter, PercentFormatter
# from seaborn import sns
from kaxanuk.attribution_analysis.interfaces import FactorModelAttributionInterface
logger = logging.getLogger(__name__)
[docs]
class FactorModelAttribution(FactorModelAttributionInterface):
[docs]
def proxy_weights(self):
"""
Create a proxy weights DataFrame if daily portfolio weights are not provided.
This is used to ensure that the CAPM model (w_i@b_i = beta) can be run without explicit weights.
"""
weights = np.ones(self.portfolio_returns.shape[0])
self.daily_portfolio_weights = pd.DataFrame(weights, index=self.portfolio_returns.index, columns=['proxy_weights'])
[docs]
def rfr_substraction(self):
returns_df = self.benchmark_returns.copy()
if returns_df.index.dtype == 'O':
returns_df.index = pd.to_datetime(returns_df.index, format = '%d/%m/%Y')
returns_df_start = returns_df.index[0]
returns_df_end = returns_df.index[-1]
fred_key = os.environ.get('FRED_API_KEY')
if not fred_key:
raise OSError("FRED_API_KEY is not set. Add it to your .env file.")
fred = Fred(api_key=fred_key)
rfr = fred.get_series('GS10') / 100
# TODO: fix los NAs que vienen de no tener la verdadera fecha de inicio de rfr
rfr_index = list(rfr.index)
temp = rfr.loc[returns_df_start:]
a = rfr_index.index(temp.index[0])
rfr = rfr.iloc[a-1:]
# rfr_until = rfr_index[:returns_df_start]
# true_rfr_start = rfr_until[-1]
# rfr = rfr.loc[returns_df_start:]
rfr.name = 'rfr'
# rfr_daily = rfr.reindex(features['nc_book_to_price_data'].index)
rfr_daily = rfr.resample('D').asfreq().apply(resample_rate).ffill()
rfr_daily = rfr_daily.reindex(returns_df.index).ffill()
universe_rets = returns_df.sub(
rfr_daily.loc[returns_df_start:returns_df_end],
axis=0
)
# y_returns = self.regression_variables.copy()
self.asset_returns_mins_rfr = self.regression_variables.sub(
rfr_daily.loc[returns_df_start:returns_df_end],
axis=0
)
self.rfr_daily = rfr_daily
self.benchmark_minus_rfr = universe_rets
# return self.benchmark_minus_rfr
[docs]
def multifactor_attribution(self):
total_daily_coverage = []
# TODO: Rework this shit
b_x_f_ts = {}
for date, ws_i in self.daily_portfolio_weights.iterrows():
non_zero_ws = ws_i[ws_i != 0]
pseudo_matrix_dict = {}
for factor, df in self.by_factor_factor_returns.items():
try:
pseudo_matrix_dict[factor] = df.loc[date]
except KeyError:
# print(f"Date {date_str} not found in factor '{factor}' — skipping.")
continue
# If nothing available for this date, skip entirely
if not pseudo_matrix_dict:
# print(f"No factor data available for {date_str}, skipping date.")
continue
pseudo_matrix = pd.DataFrame(pseudo_matrix_dict)
valid_index = pseudo_matrix.index.intersection(non_zero_ws.index)
available_non_zw = non_zero_ws.loc[valid_index]
total_weight_coverage = available_non_zw.sum()
total_daily_coverage.append(total_weight_coverage)
# print(f'total coverage in {date}: {total_weight_coverage: .4%}')
pseudo_matrix = pseudo_matrix.loc[valid_index, :]
weighted_pseudo_matrix = pseudo_matrix.mul(available_non_zw, axis=0)
portfolio_composition_t = weighted_pseudo_matrix.sum(axis=0)
b_x_f_ts[date] = portfolio_composition_t
total_daily_coverage = np.array(total_daily_coverage)
logger.info('Average total factor coverage throughout the portfolio : %s', f'{total_daily_coverage.mean(): .4%}')
portfolio_attribution_ts = pd.DataFrame(b_x_f_ts).T
self.portfolio_attribution_ts = portfolio_attribution_ts
[docs]
def time_series_calculation(self):
simulated_rets = self.portfolio_attribution_ts.cumsum()
# TODO: ADD MENNCHERO DIFERENCES
self.simulated_rets = simulated_rets
[docs]
def rename_dfs(self):
self.benchmark_returns.columns = ['benchmark_returns']
self.portfolio_returns.columns = ['portfolio_returns']
[docs]
def run(self):
if self.daily_portfolio_weights is None:
self.proxy_weights()
if self.asset_returns is None:
self.asset_returns = self.portfolio_returns.copy()
self.regression_variables = self.asset_returns.copy()
# self.rfr_substraction()
# self.rename_dfs()
# self.time_series_calculation()
self.multifactor_attribution()
self.time_series_calculation()
[docs]
def calc_pct_area(self):
self.portfolio_returns.name = 'total_returns'
pct_df_returns = self.portfolio_attribution_ts.copy()
pct_df_returns = pct_df_returns.drop(
columns = ['f_market', 'f_total_factor_returns', 'f_total_excess_returns', 'f_idyo_returns']
)
total_returns = self.portfolio_returns.copy()
total_factor_ret = pct_df_returns.sum(axis = 1)
pct_df_returns['f_idio_returns'] = total_returns.sub(total_factor_ret)
# df_returns = pct_df_returns.copy()
pct_df_returns = pct_df_returns.div(total_returns, axis = 0)
pct_df_returns = pct_df_returns.ffill() # TODO hay nulls pq en self.portfolio_attribution_ts no hay las mismas fechas que en el dict by_factor_factor_returns
self.pct_df_returns = pct_df_returns
[docs]
def plot_stacked_weights(
self,
figsize=(14, 7),
title="Return Attribution by Factor Over Time",
y_label="Share of Total Return",
date_fmt="%Y-%m",
legend_loc="upper left"
):
"""
Plot a stacked area chart of return attribution by factor over time,
using absolute values normalized to total absolute return.
Each column's contribution is expressed as abs(col) / sum(abs(all cols)),
preserving the sign information via color-coded positive/negative stacking.
Parameters
----------
figsize : tuple
Size of the figure.
title : str
Figure title.
y_label : str
Label for y-axis.
date_fmt : str
Date format string for x-axis.
legend_loc : str
Location of the legend.
"""
self.calc_pct_area()
df = self.pct_df_returns.copy()
df = df.rolling(90).mean()
# df = df.resample('M').mean()
# df = df.resample('M').mean()
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
df = df.sort_index().fillna(0.0)
# --- Core change: normalize by abs total ---
abs_total = df.abs().sum(axis=1)
# Avoid division by zero
abs_total = abs_total.replace(0, np.nan)
df_pct = df.abs().div(abs_total, axis=0).fillna(0.0)
# Split into positive-origin and negative-origin shares
# Sign is taken from the original df to know which side each col goes
pos = df_pct.where(df >= 0, 0.0)
neg = df_pct.where(df < 0, 0.0) * -1 # keep positive magnitudes for stacking below 0
pos_cum = pos.cumsum(axis=1)
pos_base = pos_cum - pos
neg_cum = neg.cumsum(axis=1)
neg_base = neg_cum - neg
fig, ax = plt.subplots(figsize=figsize)
colors = plt.cm.tab20.colors
handles = []
labels = []
for i, col in enumerate(df.columns):
color = colors[i % len(colors)]
# --- Positive stack (above 0) ---
if (pos[col] != 0).any():
h = ax.fill_between(
df.index,
pos_base[col],
pos_cum[col],
alpha=0.7,
color=color
)
handles.append(h)
labels.append(col)
# --- Negative stack (below 0) ---
if (neg[col] != 0).any():
ax.fill_between(
df.index,
-neg_cum[col],
-neg_base[col],
alpha=0.7,
color=color
)
ax.axhline(0, linewidth=1, color="black")
ax.set_title(title, fontsize=14)
ax.set_ylabel(y_label)
ax.xaxis.set_major_formatter(DateFormatter(date_fmt))
fig.autofmt_xdate()
if handles:
ax.legend(handles, labels, loc=legend_loc, ncol=2, fontsize=9)
ax.yaxis.set_major_formatter(FuncFormatter(lambda y, _: f"{y:.0%}"))
ax.grid(visible=True, alpha=0.5)
plt.tight_layout()
plt.show()
[docs]
def plot_stacked_weights_abs_norm(
self,
figsize=(14, 7),
title="Return Attribution by Factor Over Time",
y_label="Share of Total Return",
date_fmt="%Y-%m",
legend_loc="upper left",
days_of_smoothing=5,
):
"""
Plot a stacked area chart of return attribution by factor over time.
Each factor's contribution = signed_value / sum(abs(all cols)).
Positive factors stack above zero, negative factors stack below zero.
The net area always sums to +100% or -100% depending on overall return direction.
Parameters
----------
figsize : tuple
Size of the figure.
title : str
Figure title.
y_label : str
Label for y-axis.
date_fmt : str
Date format string for x-axis.
legend_loc : str
Location of the legend.
days_of_smoothing : int
Rolling window (in days) for smoothing the percentage shares.
"""
self.calc_pct_area()
df = self.pct_df_returns.copy()
df = df.ffill()
df = df.rolling(90).mean()
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
df = df.sort_index()
# Signed normalization: keep sign, divide by abs total
abs_total = df.abs().sum(axis=1).replace(0, np.nan)
pct = df.div(abs_total, axis=0).rolling(days_of_smoothing).mean().fillna(0.0)
# Split into positive and negative stacks
pos = pct.clip(lower=0)
neg = pct.clip(upper=0)
colors = [plt.cm.tab20(i % 20) for i in range(len(df.columns))]
fig, ax = plt.subplots(figsize=figsize)
# Stack positives above 0
ax.stackplot(
pct.index,
[pos[col] for col in pct.columns],
labels=list(pct.columns),
colors=colors,
alpha=0.75,
)
# Stack negatives below 0 (same colors, no duplicate labels)
ax.stackplot(
pct.index,
[neg[col] for col in pct.columns],
colors=colors,
alpha=0.75,
)
ax.axhline(0, linewidth=1, color="black")
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.legend(loc=legend_loc, fontsize=8, ncol=3)
ax.grid(visible=True, alpha=0.3)
plt.tight_layout()
plt.show()
[docs]
def plot_stacked_weights_normalized(
self,
figsize=(14, 7),
title="Return Attribution by Factor Over Time",
y_label="Share of Total Return",
date_fmt="%Y-%m",
legend_loc="upper left",
days_of_smoothing=5,
):
"""
Plot a stacked area chart of return attribution by factor over time,
using absolute values normalized to total absolute return (always sums to 100%).
Parameters
----------
figsize : tuple
Size of the figure.
title : str
Figure title.
y_label : str
Label for y-axis.
date_fmt : str
Date format string for x-axis.
legend_loc : str
Location of the legend.
days_of_smoothing : int
Rolling window (in days) for smoothing the percentage shares.
"""
self.calc_pct_area()
df = self.pct_df_returns.copy()
df = df.ffill() # <---- Check this !!
# duplicate_df = df.duplicated()
# df.loc[duplicate_df == True]
# df = df.rolling(90).mean()
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
df = df.sort_index()
# Normalize: each column's share of total absolute return
abs_components = df.abs()
abs_total = abs_components.sum(axis=1)#.replace(0, np.nan)
pct = abs_components.div(abs_total, axis=0).rolling(days_of_smoothing).mean()
pct = pct.fillna(0.0)
# Dynamic colors for any number of columns
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 cummulative_pct_decomp(self):
last_day_values= self.simulated_rets.copy().iloc[-1]
total_final_ret = last_day_values.loc['f_total_excess_returns']
total_fact = last_day_values.drop(
index=['f_market', 'f_total_factor_returns', 'f_total_excess_returns', 'f_idyo_returns']
)
idio = total_final_ret - total_fact.sum()
total_fact.loc['idio_returns'] = idio
return total_fact.div(total_final_ret)
# def attribution_plots(self, title=None):
# fig, axes = plt.subplots(
# 1, 1, figsize=(10, 7),
# )
# self.simulated_rets.plot(ax=axes, title="Factor Model Return Attribution", alpha=0.75)
# plt.tight_layout()
# plt.show()
[docs]
def attribution_plots(self, title=None):
df = self.simulated_rets # DataFrame indexed by date/time
# --- column groups ---
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",
]
# Keep only cols that exist
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]
# Drop industry columns where ALL values are 0
industries_df = df[industry_cols]
industries_df = industries_df.loc[:, (industries_df != 0).any(axis=0)]
# --- figure: taller so panels are more "square" ---
_fig, axes = plt.subplots(
3, 1, figsize=(12, 15), sharex=True # <- taller figure
)
# fig.suptitle(title or "Factor Model Return Attribution")
def style_axis(ax):
ax.grid(visible=True, which="both", axis="both", alpha=0.35)
ax.axhline(0, color="black", linewidth=1) # black zero line
ax.yaxis.set_major_formatter(PercentFormatter(xmax=1.0)) # assumes returns are decimals (0.01=1%)
# legend top-left
leg = ax.get_legend()
if leg is not None:
ax.legend(loc="upper left")
# 1) totals
if totals_cols:
df[totals_cols].plot(ax=axes[0], alpha=0.85)
axes[0].set_title("Total / Factor / Idiosyncratic")
style_axis(axes[0])
# 2) style factors
if style_cols:
df[style_cols].plot(ax=axes[1], alpha=0.85)
axes[1].set_title("Style Factors")
style_axis(axes[1])
# 3) industries
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 attrib_area_plot(self, title=None):
self.calc_pct_area()
_fig, axes = plt.subplots(
1, 1, figsize=(10, 7),
)
self.simulated_rets.plot(ax=axes, title="Factor Model Return Attribution", alpha=0.75)
plt.tight_layout()
plt.show()
[docs]
def attribution_plots_log(self, title=None):
_fig, axes = plt.subplots(
1, 1, figsize=(10, 7),
)
log_index = np.log(self.simulated_rets)
log_index.plot(ax=axes, title="Factor Model Return Attribution", alpha=0.75)
plt.tight_layout()
plt.show()
[docs]
def attribution_plots_sax(self, title=None):
fig, ax1 = plt.subplots(figsize=(10, 7))
# Primary axis — portfolio_returns
self.simulated_rets['portfolio_returns'].plot(
ax=ax1, color='tab:blue', label='portfolio_returns'
)
ax1.set_ylabel('Portfolio Returns', color='tab:blue')
ax1.tick_params(axis='y', labelcolor='tab:blue')
# Secondary axis — all other columns
ax2 = ax1.twinx()
other_cols = [c for c in self.simulated_rets.columns if c != 'portfolio_returns']
self.simulated_rets[other_cols].plot(
ax=ax2, alpha=0.5, legend=True
)
ax2.set_ylabel('Attribution Factors', color='tab:gray')
ax2.tick_params(axis='y', labelcolor='tab:gray')
# # Force both y-axes to start at 0
# ax1.set_ylim(bottom=0)
# ax2.set_ylim(bottom=0)
# Title & layout
if title is None:
title = "Factor Model Return Attribution"
ax1.set_title(title)
fig.tight_layout()
plt.show()
[docs]
def capm_plots(self, title=None):
fig, axes = plt.subplots(
3, 1, figsize=(10, 14),
gridspec_kw={'height_ratios': [2, 1, 1]} # First plot gets twice the height
)
fig.suptitle(title, fontsize=16)
# First subplot
self.capm_attribution[
[
'portfolio_returns',
'factor_returns',
'residual_returns',
]
].plot(ax=axes[0], title="Return Attribution", alpha=0.75)
axes[0].grid(visible=True)
xlims = axes[0].get_xlim()
# Second subplot
self.capm_attribution['hbeta'].plot(ax=axes[1], title="Beta")
axes[1].grid(visible=True)
axes[1].set_xlim(xlims)
# Third subplot
self.capm_attribution['halpha'].plot(ax=axes[2], title="Alpha")
axes[2].grid(visible=True)
axes[2].set_xlim(xlims)
plt.tight_layout(rect=[0, 0, 1, 0.95])
plt.show()
[docs]
def plot_stacked_weights_neg(
self,
# pct_df_returns: pd.DataFrame,
figsize=(14, 7),
title="Stacked Portfolio Weights Over Time",
y_label="Weight",
date_fmt="%Y-%m",
legend_loc="upper left"
):
"""
Plot a stacked area chart of portfolio weights (long & short) over time.
Parameters
----------
pct_df_returns : pd.DataFrame
Index: dates (DatetimeIndex or convertible to datetime)
Columns: tickers
Values: portfolio weights (can be positive or negative).
figsize : tuple
Size of the figure.
title : str
Figure title.
y_label : str
Label for y-axis.
date_fmt : str
Date format string for x-axis.
legend_loc : str
Location of the legend.
"""
self.calc_pct_area()
df = self.pct_df_returns.copy()
df = df.rolling(90).mean()
# df = df.clip(lower=-3, upper=3)
# df = df.resample('M').mean()
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
df = df.sort_index().fillna(0.0)
pos = df.clip(lower=0)
neg = df.clip(upper=0)
pos_cum = pos.cumsum(axis=1)
pos_base = pos_cum - pos
neg_cum = neg.cumsum(axis=1)
neg_base = neg_cum - neg
fig, ax = plt.subplots(figsize=figsize)
colors = plt.cm.tab20.colors
handles = []
labels = []
for i, col in enumerate(df.columns):
color = colors[i % len(colors)]
# long
y_top_pos = pos_cum[col]
y_bottom_pos = pos_base[col]
if (pos[col] != 0).any():
h_pos = ax.fill_between(
df.index, y_bottom_pos, y_top_pos, alpha=0.7, color=color
)
handles.append(h_pos)
labels.append(col)
# short
y_top_neg = neg_cum[col]
y_bottom_neg = neg_base[col]
if (neg[col] != 0).any():
ax.fill_between(
df.index, y_bottom_neg, y_top_neg, alpha=0.7, color=color
)
ax.axhline(0, linewidth=1, color="black")
ax.set_title(title, fontsize=14)
ax.set_ylabel(y_label)
ax.xaxis.set_major_formatter(DateFormatter(date_fmt))
fig.autofmt_xdate()
if handles:
ax.legend(handles, labels, loc=legend_loc, ncol=2, fontsize=9)
ax.yaxis.set_major_formatter(FuncFormatter(lambda y, _: f"{y:.0%}"))
ax.grid(visible=True, alpha=0.5)
# === NEW: seaborn-style despine ===
# sns.despine(ax=ax)
plt.tight_layout()
plt.show()
# return fig, ax
[docs]
def plot_portfolio_weights_area(
self,
tickers=None,
figsize=(14, 7),
title="Portfolio Weights Over Time",
y_label="Weight",
date_fmt="%Y-%m"
):
"""
Area plot of portfolio weights (long and short) over time.
Parameters
----------
pct_df_returns : pd.DataFrame
Index: dates (daily, or any time index)
Columns: tickers
Values: portfolio weights (can be positive or negative)
tickers : list, optional
Subset of tickers to plot. If None, all columns are used.
figsize : tuple
Figure size (width, height).
title : str
Plot title.
y_label : str
Label for the y-axis.
date_fmt : str
Date format string for x-axis tick labels.
Returns
-------
fig, ax : matplotlib Figure and Axes
"""
self.calc_pct_area()
# Work on a copy so we don't mutate the original
df = self.pct_df_returns.copy()
# df = df.drop(columns = 'portfolio_returns')
df = df.clip(lower = -3, upper = 3)
df = df.resample('M').mean()
fig, ax = plt.subplots()
# weights_df = weights_df.to_dict()
# ax = df.plot.area(figsize=figsize)
# # ax.stackplot(weights_df.index,
# # weights_df.values,
# # labels = weights_df.columns,
# # alpha = 0.8)
# ax.set_title(title, fontweight="bold")
#
# # plt.legend(loc="center right" , fancybox=True, shadow=True, ncol=2,title = 'Asset Class')
# box = ax.get_position()
# ax.set_position([box.x0, box.y0, box.width * 0.8, box.height])
# # Put a legend to the right of the current axis
# ax.legend(loc='center left', bbox_to_anchor=(1, 0.5), title='Asset Class')
# ax.yaxis.set_major_formatter(FuncFormatter(lambda y, _: '{:.2%}'.format(y)))
# plt.xticks(rotation=0)
#
# plt.tight_layout()
# plt.show()
# fig = ax.get_figure()
# Restrict to selected tickers if provided
if tickers is not None:
df = df[tickers]
# Ensure index is datetime and sorted
if not isinstance(df.index, pd.DatetimeIndex):
df.index = pd.to_datetime(df.index)
df = df.sort_index()
# Optionally drop all-NaN columns
df = df.dropna(axis=1, how="all")
fig, ax = plt.subplots(figsize=figsize)
# Use a categorical colormap
colors = plt.cm.tab20.colors
for i, col in enumerate(df.columns):
series = df[col]
# Positive part
ax.fill_between(
series.index,
0,
series,
where=(series >= 0),
interpolate=True,
alpha=0.5,
color=colors[i % len(colors)],
step="mid",
label=col if i == 0 else None # temp, we'll handle legend better below
)
# Negative part (same color, different alpha/edge to differentiate if you want)
ax.fill_between(
series.index,
0,
series,
where=(series < 0),
interpolate=True,
alpha=0.5,
color=colors[i % len(colors)],
step="mid"
)
# Because we used the same label multiple times, rebuild legend from unique handles/labels
handles, labels = ax.get_legend_handles_labels()
by_label = dict(zip(labels, handles, strict=False))
ax.legend(by_label.values(), by_label.keys(), loc="upper left", ncol=2, fontsize=9)
# Axis labels and title
ax.set_title(title, fontsize=14)
ax.set_ylabel(y_label)
ax.axhline(0, color="black", linewidth=1) # zero line for long/short separation
# Date formatting
ax.xaxis.set_major_formatter(DateFormatter(date_fmt))
fig.autofmt_xdate()
# Tight layout
plt.tight_layout()
plt.show()
return fig, ax
[docs]
def last_day_decomposition(self, title):
"""
Plots positive and negative factor contributions as a diverging bar chart.
Parameters
----------
series : pd.Series
Index = factor names
Values = contribution percentages (can be negative)
"""
total_pct = self.cummulative_pct_decomp()
# Sort by value for nicer layout
s = total_pct.sort_values()
colors = s.apply(lambda x: 'red' if x < 0 else 'green')
plt.figure(figsize=(12, 8))
bars = plt.barh(s.index, s.values, color=colors)
# Label each bar with percentage
for bar, value in zip(bars, s.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.axvline(0, color='black', linewidth=1)
plt.tight_layout()
plt.show()
[docs]
def resample_rate(value):
return (1 + value) ** (1/252)-1
[docs]
def weighted_std(
values,
weights = None,
hl = None
):
"""
Return the weighted average and standard deviation.
values, weights -- Numpy ndarrays with the same shape.
IF weights is None, then it will be calculated using WEXP the half_life
"""
if weights is None:
if hl is not None:
weights = wexp(len(values), hl)
else:
logger.warning("Please provide weights or half_life")
# average = np.average(values, weights=weights)
average = np.average(values, weights=weights, axis = 0)
# variance = np.average((values-average)**2, weights=weights)
variance = np.average((values-average)**2, weights=weights, axis = 0)
return variance ** 0.5
[docs]
def wexp(num_periods, half_life):
"""
Correr funcion con num_periods fija y varias half lifes
parece que esta en dias
"""
c = np.log(0.5)/half_life
n = np.array(range(num_periods))
w = np.exp(c*n)
return w/np.sum(w)
