Source code for kaxanuk.attribution_analysis.modules.synthetic_exposures_creator
import datetime
import logging
from collections.abc import Collection
from pathlib import Path
import numpy as np
import pandas as pd
logger = logging.getLogger(__name__)
# --- synthetic exposure generator ---
[docs]
def generate_synthetic_exposure_dict(
portfolio_weights: pd.DataFrame,
universe_returns: pd.DataFrame,
date_list: list[str] | None = None,
factor_index: Collection[str] = (
"ci_intercept",
"nc_beta",
"nc_log_market_cap",
"nc_momentum",
"nc_resid_vol",
"nc_value",
),
seed: int | None = 42,
dtype: str = "float32",
na_percentage: float = 0.0,
) -> dict[datetime.date, pd.DataFrame]:
"""
Create an in-memory synthetic exposure dictionary that mirrors the real data shape,
with an optional percentage of random NaN values injected.
Parameters
----------
portfolio_weights : pd.DataFrame
DataFrame with dates as index and assets as columns. Columns are used for synthetic exposure columns.
universe_returns : pd.DataFrame
DataFrame containing universe assets (used for column reference).
date_list : list[str] | None
Dates to generate (format '%Y-%m-%d'). If None, use all dates from portfolio_weights.index.
factor_index : iterable[str]
Row index (factors) for each synthetic exposure matrix.
seed : int | None
RNG seed for reproducibility. Use None for non-deterministic draws.
dtype : str
Numpy/pandas dtype for the generated values.
na_percentage : float
Fraction (0-1) of entries to replace with NaN randomly. Default = 0 (no missing data).
Returns
-------
dict[datetime.date, pd.DataFrame]
Keys are datetime.date objects, values are DataFrames with index=factor_index and columns=portfolio_weights.columns.
"""
rng = np.random.default_rng(seed)
# Choose dates
if date_list is None:
date_strs = list(pd.to_datetime(portfolio_weights.index).strftime("%Y-%m-%d"))
else:
date_strs = list(date_list)
assets = list(universe_returns.columns)
n_factors = len(factor_index)
n_assets = len(assets)
exposure_dict = {}
for ds in date_strs:
# Draw standard normals (z-score-like)
data = rng.standard_normal(size=(n_factors, n_assets)).astype(dtype, copy=False)
# Optionally inject NaNs
if na_percentage > 0:
n_total = data.size
n_missing = int(n_total * na_percentage)
if n_missing > 0:
missing_idx = rng.choice(n_total, size=n_missing, replace=False)
data.flat[missing_idx] = np.nan
# Build DataFrame
df = pd.DataFrame(data, index=list(factor_index), columns=assets)
exposure_dict[ds] = df
return exposure_dict
[docs]
def stock_factor_returns_loader(
f_rets_by_stock_path: str,
files_to_load: list[str],
first_day_weights: pd.Timestamp,
) -> dict[str, pd.DataFrame]:
"""
Load per-stock factor-returns CSV files into a dict of DataFrames.
Parameters
----------
f_rets_by_stock_path : str
Directory containing one CSV file per stock.
files_to_load : list[str]
File names (with extension) to load from `f_rets_by_stock_path`.
The stock name is derived from each file name (text before the
first ``.``).
first_day_weights : pd.Timestamp
Earliest date to keep; each file is sliced from this date
onward.
Returns
-------
dict[str, pd.DataFrame]
Mapping of stock name to its factor-returns DataFrame, indexed
by date. Files listed in `files_to_load` that are not found are
logged as a warning and omitted from the result.
"""
by_stock_factor_returns = {}
missing_files = []
for file in files_to_load:
try:
full_path = Path(f_rets_by_stock_path) / file
df = pd.read_csv(
full_path,
index_col=0,
parse_dates=[0],
date_format='%Y-%m-%d'
)
stock = file.split('.')[0]
by_stock_factor_returns[stock] = df.loc[first_day_weights:]
except FileNotFoundError:
missing_files.append(file)
if missing_files:
logger.warning("%d files not found: %s", len(missing_files), missing_files)
# break
return by_stock_factor_returns
