"""
Loaders for per-factor returns files, keyed by factor name.
Provides a pandas-based loader (`load_by_factor_returns_csv`) and a
PyArrow-based loader (`load_by_factor_returns_arrow`) that read one CSV
file per factor from a directory, restrict columns to the portfolio's
assets, and optionally slice the result to a date range.
"""
import csv as csv_mod
import datetime
import logging
from pathlib import Path
import pandas as pd
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.csv as pa_csv
from kaxanuk.attribution_analysis.entities.factor_returns import FactorReturns
from kaxanuk.attribution_analysis.entities.portfolio_weights_arrow import PortfolioWeights
from kaxanuk.attribution_analysis.exceptions import DateRangeError
from kaxanuk.attribution_analysis.input_handlers.portfolio_weights_input_handler import read_csv_intersection
logger = logging.getLogger(__name__)
def _validate_and_slice_date_range(
df: pd.DataFrame,
start_date: datetime.date | None,
end_date: datetime.date | None,
source_name: str,
) -> pd.DataFrame:
if start_date is None and end_date is None:
return df
if start_date is not None:
start_ts = pd.Timestamp(start_date)
if start_ts < df.index.min():
raise DateRangeError(
f"start_date {start_date} is before the earliest date "
f"({df.index.min().date()}) in {source_name}"
)
if end_date is not None:
end_ts = pd.Timestamp(end_date)
if end_ts > df.index.max():
logger.warning(
"end_date %s is after the latest date (%s) in %s — "
"slicing up to last available date",
end_date, df.index.max().date(), source_name,
)
return df.loc[
(start_date or df.index.min()):(end_date or df.index.max())
]
def _validate_and_filter_arrow(
table: pa.Table,
start_date: datetime.date | None,
end_date: datetime.date | None,
source_name: str,
) -> pa.Table:
if start_date is None and end_date is None:
return table
date_col = table.column("date")
if start_date is not None:
min_date = pc.min(date_col).as_py()
if pd.Timestamp(start_date) < min_date:
raise DateRangeError(
f"start_date {start_date} is before the earliest date "
f"({min_date.date()}) in {source_name}"
)
if end_date is not None:
max_date = pc.max(date_col).as_py()
if pd.Timestamp(end_date) > max_date:
logger.warning(
"end_date %s is after the latest date (%s) in %s — "
"slicing up to last available date",
end_date, max_date.date(), source_name,
)
start_val = pd.Timestamp(start_date) if start_date else pc.min(date_col).as_py()
end_val = pd.Timestamp(end_date) if end_date else pc.max(date_col).as_py()
mask = pc.and_(
pc.greater_equal(date_col, pa.scalar(start_val, type=pa.timestamp("ns"))),
pc.less_equal(date_col, pa.scalar(end_val, type=pa.timestamp("ns"))),
)
return table.filter(mask)
[docs]
def load_by_factor_returns_csv(
path: str,
available_factors: list[str],
portfolio_weights: pd.DataFrame,
start_date: datetime.date | None = None,
end_date: datetime.date | None = None,
) -> dict[str, pd.DataFrame]:
"""
Load per-factor returns CSV files into a dict of DataFrames.
Each file in `available_factors` is read from `path`, restricted to
the columns present in `portfolio_weights`, and sliced from the
portfolio's first weighted day onward.
Parameters
----------
path : str
Directory containing one CSV file per factor.
available_factors : list[str]
File names (with extension) to load from `path`. Entries equal
to ``".gitkeep"`` are ignored.
portfolio_weights : pd.DataFrame
Portfolio weights, dates as index and assets as columns; the
columns determine which factor-return columns are read, and the
index's first date lower-bounds the returned data.
start_date : datetime.date or None
If given, the earliest date to keep. Raises if earlier than the
data's minimum date.
end_date : datetime.date or None
If given, the latest date to keep. Silently clamped to the
data's maximum date if later.
Returns
-------
dict[str, pd.DataFrame]
Mapping of factor name (derived from the file name) to its
returns DataFrame, indexed by date with one column per asset.
Raises
------
DateRangeError
If `start_date` is before the earliest available date in a
factor's file.
"""
available_factors = [f for f in available_factors if f != ".gitkeep"]
first_day_weights = portfolio_weights.index[0]
# last_day_weights = portfolio_weights.index[-1]
portfolio_assets = list(portfolio_weights.columns)
logger.info("Loading %d factor return files from: %s", len(available_factors), path)
by_factor_factor_returns = {}
for factor_file in available_factors:
factor_name = factor_file.split(".")[0]
factor_df = read_csv_intersection(
str(Path(path) / factor_file),
portfolio_assets,
index_col=0,
date_format="%Y-%m-%d"
).loc[first_day_weights:]
factor_df = _validate_and_slice_date_range(
factor_df, start_date, end_date,
f"factor returns file '{factor_file}'"
)
by_factor_factor_returns[factor_name] = factor_df
return by_factor_factor_returns
[docs]
def load_by_factor_returns_arrow(
path: str,
available_factors: list[str],
portfolio_weights: PortfolioWeights,
start_date: datetime.date | None = None,
end_date: datetime.date | None = None,
) -> dict[str, FactorReturns]:
"""
Load per-factor returns CSV files into a dict of `FactorReturns`.
Arrow-based equivalent of `load_by_factor_returns_csv`. Each file in
`available_factors` is read from `path` with only the date column and
the portfolio's asset columns retained, dates are parsed (trying
``%Y-%m-%d`` then ``%d/%m/%Y``), and the result is sliced from the
portfolio's first weighted day onward before being validated and
wrapped in a `FactorReturns` entity.
Parameters
----------
path : str
Directory containing one CSV file per factor.
available_factors : list[str]
File names (with extension) to load from `path`. Entries equal
to ``".gitkeep"`` are ignored.
portfolio_weights : PortfolioWeights
Portfolio weights entity; its first date lower-bounds the
returned data and its asset names determine which factor-return
columns are read and validated against.
start_date : datetime.date or None
If given, the earliest date to keep. Raises if earlier than the
data's minimum date.
end_date : datetime.date or None
If given, the latest date to keep. Silently clamped to the
data's maximum date if later.
Returns
-------
dict[str, FactorReturns]
Mapping of factor name (derived from the file name) to its
`FactorReturns` entity.
Raises
------
DateRangeError
If `start_date` is before the earliest available date in a
factor's file.
DataValidationError
If a factor's asset columns are not a subset of the portfolio's
tickers, or the resulting table is empty.
MissingColumnError
If a factor's file is missing its date column.
EntityTypeError
If a factor's date or asset columns have an unexpected type.
"""
available_factors = [f for f in available_factors if f != ".gitkeep"]
first_day_weights = portfolio_weights.table.column("date")[0].as_py()
portfolio_assets = portfolio_weights.asset_names
portfolio_assets_set = set(portfolio_assets)
logger.info("Loading %d factor return files (arrow) from: %s", len(available_factors), path)
by_factor_factor_returns = {}
for factor_file in available_factors:
factor_name = factor_file.split(".")[0]
file_path = Path(path) / factor_file
with file_path.open() as f:
reader = csv_mod.reader(f)
available_cols = next(reader)
date_col_name = available_cols[0]
available_cols_set = set(available_cols)
cols_to_read = [date_col_name] + [c for c in portfolio_assets if c in available_cols_set]
logger.info("Factor '%s': reading %d/%d asset columns", factor_name, len(cols_to_read) - 1, len(portfolio_assets))
convert_options = pa_csv.ConvertOptions(include_columns=cols_to_read)
factor_table = pa_csv.read_csv(file_path, convert_options=convert_options)
raw_date_col = factor_table.column(date_col_name)
try:
parsed_dates = pc.strptime(raw_date_col.cast(pa.string()), format="%Y-%m-%d", unit="ns")
except (pa.ArrowInvalid, pa.ArrowNotImplementedError, ValueError):
parsed_dates = pc.strptime(raw_date_col.cast(pa.string()), format="%d/%m/%Y", unit="ns")
new_columns = [parsed_dates]
new_names = ["date"]
for name in factor_table.column_names:
if name == date_col_name:
continue
new_columns.append(pc.cast(factor_table.column(name), pa.float64()))
new_names.append(name)
factor_table = pa.table(new_columns, names=new_names)
first_day_scalar = pa.scalar(first_day_weights, type=pa.timestamp("ns"))
mask = pc.greater_equal(factor_table.column("date"), first_day_scalar)
factor_table = factor_table.filter(mask)
factor_table = _validate_and_filter_arrow(
factor_table, start_date, end_date,
f"factor returns file '{factor_file}'"
)
factor_entity = FactorReturns(table=factor_table, factor_name=factor_name)
factor_entity.validate_asset_coverage(portfolio_assets_set)
by_factor_factor_returns[factor_name] = factor_entity
return by_factor_factor_returns
