from dataclasses import replace
from pathlib import Path as path_cls
import warnings
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from IPython.display import display
from matplotlib.ticker import PercentFormatter as percent_formatter
from quantfinlab.backtest.portfolio import run_many_weights_backtests
from quantfinlab.dataio import (
load_par_yield_curve,
read_equity_history,
read_sec_facts,
read_sec_metadata,
risk_free_returns,
)
from quantfinlab.fundamentals import (
diagnostic_model_metrics,
financial_company_metrics,
classify_duration_facts,
issuer_universe,
monthly_statement_values,
periodic_forms,
profitability_growth_metrics,
reconstruct_quarters,
scoring as score_math,
select_filing_facts,
statement_concepts,
statement_metrics,
statement_reconstruction_checks,
strength_efficiency_metrics,
)
from quantfinlab.plotting import fundamentals as fundamental_plots, portfolio as portfolio_plots
from quantfinlab.portfolio import (
covariance, expected_returns, factors as factor_math, optimizers, selection,
)
from quantfinlab.portfolio.cvar import mean_cvar_weight_frame
from quantfinlab.portfolio.hrp import hrp_weight_frame, nco_mv_weight_frame
from quantfinlab.portfolio.risk_parity import risk_parity_weight_frame
from quantfinlab.portfolio.robust import wasserstein_weight_frame
from quantfinlab.portfolio.walkforward import (
run_equal_weight_walkforward, run_walkforward_grid,
)
from quantfinlab.reports import fundamental_report, risk_report
repo_root = path_cls.cwd()
if not (repo_root / "data" / "sp500_market_data.parquet").exists():
repo_root = repo_root.parent
market_path = repo_root / "data" / "sp500_market_data.parquet"
fundamentals_path = repo_root / "data" / "sp500_fundamentals.parquet"
treasury_path = repo_root / "data" / "us_treasury_yields.csv"
statement_cache = (
repo_root / "data" / "datasets" / "project21" / "monthly_filing_financials_v2.parquet"
)
warmup_start = pd.Timestamp("2012-01-01")
backtest_start = pd.Timestamp("2013-01-01")
holdout_start = pd.Timestamp("2020-01-01")
top_n_values = (15, 50, 100)
lookback = 189
minimum_observations = 177
maximum_weight = 0.15
mean_variance_lambda = 10.0
trading_cost_bps = 10.0
turnover_penalty_bps = 10.0
annualization = 252.0
equity_data = read_equity_history(
market_path,
start=warmup_start,
columns=[
"date", "ticker", "adj_close", "is_sp500_member",
"industry", "market_cap",
],
drop_partial_last_date=True,
validate=True,
)
market_data = equity_data["market"]
adj_close = equity_data["adj_close"]
adj_close_filled = equity_data["adj_close_filled"]
volume = equity_data["volume"]
returns = equity_data["returns"]
date_map = equity_data["date_map"]
sec_data = read_sec_metadata(
fundamentals_path,
include_concepts=False,
include_mappings=True,
validate=True,
)
monthly_issuers = issuer_universe(
market_data,
sec_data["mappings"],
date_map,
exclude_reits=True,
)
required_concepts = sorted(
{
concept
for statement_group in statement_concepts.values()
for candidates in statement_group.values()
for concept in candidates
}
)
sec_facts = read_sec_facts(
fundamentals_path,
concepts=required_concepts,
ciks=monthly_issuers["cik"].unique(),
forms=periodic_forms,
period_start=warmup_start,
validate=True,
)
classified_facts = classify_duration_facts(sec_facts)
latest_facts = select_filing_facts(classified_facts)
_, quarter_candidates = reconstruct_quarters(latest_facts)
monthly_statements = monthly_statement_values(
classified_facts,
monthly_issuers,
cache=statement_cache,
force=False,
)
statement_checks = statement_reconstruction_checks(
quarter_candidates,
monthly_statements,
)
fundamental_metrics = (
monthly_issuers.merge(
monthly_statements,
on=["decision_date", "cik"],
how="left",
validate="one_to_one",
)
.pipe(statement_metrics)
.pipe(profitability_growth_metrics)
.pipe(strength_efficiency_metrics)
.pipe(financial_company_metrics)
.pipe(diagnostic_model_metrics)
)
financial_rows = fundamental_metrics["score_family"].eq("financial")
corporate_rows = fundamental_metrics["score_family"].eq("corporate")
score_configs = {
"corporate": score_math.corporate_score,
"financial": score_math.financial_score,
}
score_metrics = sorted(
{
metric_name
for score_config in score_configs.values()
for block_metrics in score_config["blocks"].values()
for metric_name in block_metrics
}
)
lower_is_better = sorted(
{
metric_name
for score_config in score_configs.values()
for metric_name in score_config["lower_is_better"]
}
)
metric_scores = score_math.metric_percentile_scores(
fundamental_metrics,
score_metrics,
lower_is_better=lower_is_better,
peer_weight=0.70,
minimum_peers=10,
winsor_limits=(0.01, 0.99),
)
fundamental_scores = pd.concat(
[
fundamental_metrics,
metric_scores,
],
axis=1,
)
corporate_block_scores = score_math.block_scores(
fundamental_scores,
score_math.corporate_score["blocks"],
prefix="corporate",
).where(corporate_rows, axis=0)
financial_block_scores = score_math.block_scores(
fundamental_scores,
score_math.financial_score["blocks"],
prefix="financial",
).where(financial_rows, axis=0)
fundamental_scores = pd.concat(
[
fundamental_scores,
corporate_block_scores,
financial_block_scores,
],
axis=1,
)
corporate_base = score_math.family_composite_score(
fundamental_scores,
corporate_block_scores,
score_math.corporate_score,
)
financial_base = score_math.family_composite_score(
fundamental_scores,
financial_block_scores,
score_math.financial_score,
)
fundamental_scores = pd.concat(
[
fundamental_scores,
corporate_base,
financial_base,
],
axis=1,
)
fundamental_scores["base_score"] = fundamental_scores["corporate_base_score"].combine_first(
fundamental_scores["financial_base_score"]
)
fundamental_scores["piotroski_penalty"] = score_math.piotroski_penalty(
fundamental_scores,
)
fundamental_scores["red_flag_penalty"] = score_math.red_flag_penalty(
fundamental_scores,
)
total_penalty = fundamental_scores["piotroski_penalty"] + fundamental_scores["red_flag_penalty"]
fixed_uncapped = fundamental_scores["base_score"] - total_penalty
family_keys = [
fundamental_scores["decision_date"],
fundamental_scores["score_family"],
]
fundamental_scores["fixed_score"] = fixed_uncapped.groupby(family_keys).rank(pct=True) * 100.0
price_signals = score_math.price_signal_frame(
adj_close_filled,
decision_dates=date_map["decision_date"],
horizons=(1, 3, 6, 12),
momentum_horizons=(3, 6, 12),
skip_recent_months=1,
)
fundamental_scores = fundamental_scores.join(
price_signals,
on=["decision_date", "ticker"],
)
score_weights = {}
adaptive_parts = []
for company_type, score_config in score_configs.items():
block_columns = [
f"{company_type}_{block_name}_score" for block_name in score_config["block_weights"]
]
weight_history = score_math.walkforward_block_weights(
fundamental_scores,
block_columns,
score_config["block_weights"],
family=company_type,
horizons=(3, 6, 12),
horizon_weights=(0.25, 0.50, 0.25),
window=36,
minimum_periods=12,
adaptive_share=0.75,
weight_cap=0.35,
)
score_weights[company_type] = weight_history
adaptive_parts.append(
score_math.adaptive_family_score(
fundamental_scores,
fundamental_scores,
weight_history,
score_config,
penalties=total_penalty,
)
)
adaptive_values = pd.concat(adaptive_parts)
adaptive_values = adaptive_values.groupby(level=0).first().reindex(fundamental_scores.index)
fundamental_scores["adaptive_base_score"] = adaptive_values["adaptive_base_score"]
fundamental_scores["uncapped_score"] = adaptive_values["uncapped_score"]
fundamental_scores["final_score"] = adaptive_values["final_score"]
selection_values = score_math.definitive_selection_score(
fundamental_scores,
fundamental_column="final_score",
momentum_column="momentum_6_1",
momentum_weight=0.10,
)
fundamental_scores = pd.concat(
[
fundamental_scores,
selection_values,
],
axis=1,
)
fundamental_scores["score_rank"] = fundamental_scores.groupby("decision_date")[
"selection_score"
].rank(ascending=False, method="first")
fundamental_scores["profitability_score"] = fundamental_scores[
"corporate_profitability_score"
].combine_first(fundamental_scores["financial_profitability_score"])
fundamental_scores["efficiency_score"] = fundamental_scores[
"corporate_efficiency_score"
].combine_first(fundamental_scores["financial_efficiency_score"])
fundamental_scores["valuation_score"] = fundamental_scores[
"corporate_valuation_score"
].combine_first(fundamental_scores["financial_valuation_return_score"])
fundamental_scores["cash_or_stability_score"] = fundamental_scores[
"corporate_cash_quality_score"
].combine_first(fundamental_scores["financial_stability_score"])
prediction_columns = [
"selection_score",
"final_score",
"fixed_score",
"profitability_score",
"cash_or_stability_score",
"efficiency_score",
"valuation_score",
]
score_validation = score_math.rank_ic_table(
fundamental_scores,
score_columns=prediction_columns,
horizons=(1, 3, 6, 12),
top_fraction=0.20,
)
bucket_returns = score_math.bucket_return_table(
fundamental_scores,
score_column="selection_score",
horizons=(1, 3, 6, 12),
buckets=5,
)
stock_selections = score_math.select_stocks(
fundamental_scores,
top_n=top_n_values,
score_column="selection_score",
start=backtest_start,
)
universes = score_math.investment_universes(
monthly_issuers,
stock_selections,
returns,
date_map,
top_n=top_n_values,
lookback=lookback,
minimum_observations=minimum_observations,
minimum_assets=10,
prices=adj_close_filled,
)
treasury_curve = load_par_yield_curve(
treasury_path,
source="us_treasury",
)
rf_daily = risk_free_returns(
treasury_curve["3M"],
returns.index,
).ffill().ffill().ffill().ffill()
selection_dates = pd.DatetimeIndex(
sorted(set(stock_selections["decision_date"]) & set(date_map["decision_date"]))
)
selection_dates = selection_dates[selection_dates >= backtest_start]
execution_by_decision = date_map.set_index("decision_date")["execution_date"]
portfolio_dates = [
pd.Timestamp(execution_by_decision.loc[decision_date]) for decision_date in selection_dates
]
equal_weight_settings = {
"returns": returns,
"rebalance_dates": portfolio_dates,
"max_weight": maximum_weight,
"min_weight": 0.0,
"long_only": True,
"trading_cost_bps": trading_cost_bps,
"rf_daily": rf_daily,
}
latest_date = fundamental_scores["decision_date"].max()
latest_top15 = stock_selections[
stock_selections["decision_date"].eq(latest_date) & stock_selections["top_n"].eq(15)
].sort_values("selection_rank")
latest_top15 = latest_top15[
[
"selection_rank",
"ticker",
"entity_name",
"score_family",
"industry",
"market_cap",
"final_score",
"momentum_score",
"selection_score",
]
].set_index("selection_rank")
def holdout_result(result, start):
net_returns = pd.Series(result.net_returns, dtype=float).loc[start:].copy()
gross_returns = pd.Series(result.gross_returns, dtype=float).loc[start:].copy()
earlier_dates = result.net_values.index[result.net_values.index < start]
baseline_date = pd.Timestamp(earlier_dates.max())
net_values = pd.concat([
pd.Series([1.0], index=[baseline_date]),
(1.0 + net_returns).cumprod(),
])
gross_values = pd.concat([
pd.Series([1.0], index=[baseline_date]),
(1.0 + gross_returns).cumprod(),
])
return replace(
result,
net_returns=net_returns,
gross_returns=gross_returns,
net_values=net_values,
gross_values=gross_values,
weights=result.weights.loc[result.weights.index >= start],
turnover=result.turnover.loc[result.turnover.index >= start],
costs=result.costs.loc[result.costs.index >= start],
)
full_ew = run_equal_weight_walkforward(
universe_by_date=universes["full"],
**equal_weight_settings,
)
top15_ew = run_equal_weight_walkforward(
universe_by_date=universes["top15"],
**equal_weight_settings,
)
layer1_results = {
"Full EW": holdout_result(full_ew, holdout_start),
"Top15 EW": holdout_result(top15_ew, holdout_start),
}
layer1_performance = selection.build_strategy_summary(
layer1_results,
rf_daily=rf_daily,
annualization=annualization,
)
top50_ew = run_equal_weight_walkforward(
universe_by_date=universes["top50"],
**equal_weight_settings,
)
top100_ew = run_equal_weight_walkforward(
universe_by_date=universes["top100"],
**equal_weight_settings,
)
layer2_results = {
"Top15 EW": holdout_result(top15_ew, holdout_start),
"Top50 EW": holdout_result(top50_ew, holdout_start),
"Top100 EW": holdout_result(top100_ew, holdout_start),
}
layer2_performance = selection.build_strategy_summary(
layer2_results,
rf_daily=rf_daily,
annualization=annualization,
)
cov_models = {
"Sample": covariance.sample_covariance,
"LedoitWolf": covariance.ledoit_wolf_covariance,
"OAS": covariance.oas_covariance,
"EWMA": covariance.ewma_covariance,
}
mu_models = {
"Momentum": expected_returns.momentum_mu,
"BayesStein": expected_returns.bayes_stein_mu,
"BayesSteinMomentum": expected_returns.bayes_stein_momentum_mu,
}
top15_optimizers = {
"MinVar": optimizers.minimum_variance,
"MV": optimizers.mean_variance,
"MaxSharpe": optimizers.max_sharpe_slsqp,
}
fixed_core_specs = [
{"name": "MinVar", "optimizer": "MinVar", "cov_model": "LedoitWolf"},
{
"name": "MV", "optimizer": "MV", "cov_model": "LedoitWolf",
"mu_model": "BayesStein",
},
{
"name": "MaxSharpe", "optimizer": "MaxSharpe", "cov_model": "Sample",
"mu_model": "BayesStein",
},
]
top15_grid = run_walkforward_grid(
returns=returns,
close=adj_close_filled,
rebalance_dates=portfolio_dates,
universe_by_date=universes["top15"],
cov_lookback=lookback,
mu_lookback=lookback,
min_cov_observations=minimum_observations,
min_mu_observations=minimum_observations,
max_weight=maximum_weight,
min_weight=0.0,
long_only=True,
trading_cost_bps=trading_cost_bps,
turnover_penalty_bps=turnover_penalty_bps,
optimizer_params={"MV": {"mv_lambda": mean_variance_lambda}},
blend_by_optimizer={"MinVar": 0.0, "MV": 0.0, "MaxSharpe": 0.0},
fallback="equal",
rf_daily=rf_daily,
annualization=annualization,
momentum_mode="6-1",
cov_models=cov_models,
mu_models=mu_models,
optimizers=top15_optimizers,
strategy_specs=fixed_core_specs,
)
top15_core = {"EW": top15_ew, **top15_grid.backtests}
top15_dates = list(top15_grid.metadata["rebalance_dates"])
with warnings.catch_warnings(record=True) as library_cvar_warnings:
warnings.simplefilter("always")
mean_cvar_weights = mean_cvar_weight_frame(
cache=top15_grid.cache,
rebalance_dates=top15_dates,
cov_model="EWMA",
mu_model="BayesStein",
reference="equal",
alpha=0.95,
budget_scale=0.75,
w_min=0.0,
w_max=maximum_weight,
)
print({"captured Mean-CVaR solver warnings": len(library_cvar_warnings)})
wro_weights = wasserstein_weight_frame(
cache=top15_grid.cache,
rebalance_dates=top15_dates,
cov_model="LedoitWolf",
mu_model="BayesStein",
radius=0.50,
mv_lambda=6.0,
radius_scale="avg_vol",
worst_case_variance=True,
w_min=0.0,
w_max=maximum_weight,
)
risk_parity_weights = risk_parity_weight_frame(
cache=top15_grid.cache,
rebalance_dates=top15_dates,
cov_model="Sample",
w_min=0.0,
w_max=maximum_weight,
)
hrp_weights = hrp_weight_frame(
cache=top15_grid.cache,
rebalance_dates=top15_dates,
cov_model="Sample",
linkage_method="complete",
w_min=0.0,
w_max=maximum_weight,
)
nco_weights = nco_mv_weight_frame(
cache=top15_grid.cache,
rebalance_dates=top15_dates,
cov_model="LedoitWolf",
mu_model="BayesSteinMomentum",
n_clusters=4,
inner_lambda=6.0,
outer_lambda=6.0,
cluster_cap=0.50,
linkage_method="average",
w_min=0.0,
w_max=maximum_weight,
)
advanced_results = run_many_weights_backtests(
{
"Mean-CVaR": mean_cvar_weights,
"WRO": wro_weights,
"Risk Parity": risk_parity_weights,
"HRP": hrp_weights,
"NCO": nco_weights,
},
returns=returns,
cost_bps=trading_cost_bps,
rf_daily=rf_daily,
w_min=0.0,
w_max=maximum_weight,
long_only=True,
normalize=True,
weight_timing="same_day",
)
top15_grid.cache.clear()
top15_grid.metadata.pop("_returns_source", None)
layer3_results_full = {**top15_core, **advanced_results}
layer3_results = {
name: holdout_result(result, holdout_start)
for name, result in layer3_results_full.items()
}
layer3_performance = selection.build_strategy_summary(
layer3_results,
rf_daily=rf_daily,
annualization=annualization,
).sort_values(
["Sharpe", "Max Drawdown"],
ascending=[False, False],
)
layer3_nav = pd.DataFrame(
{strategy_name: result.net_values for strategy_name, result in layer3_results.items()}
).dropna(how="all")
layer3_plot = layer3_performance.assign(drawdown=layer3_performance["Max Drawdown"].abs())
fig, axes = plt.subplots(5, 2, figsize=(16, 20))
fundamental_plots.plot_statement_coverage(
statement_checks["coverage"],
ax=axes[0, 0],
)
fundamental_plots.plot_reconstruction_sources(
statement_checks["ttm_sources"],
ax=axes[0, 1],
)
fundamental_plots.plot_score_counts(
fundamental_scores,
ax=axes[1, 0],
)
fundamental_plots.plot_score_weights(
score_weights,
company_type="corporate",
ax=axes[1, 1],
)
fundamental_plots.plot_score_weights(
score_weights,
company_type="financial",
ax=axes[2, 0],
)
fundamental_plots.plot_rank_ic(
score_validation,
score="selection_score",
ax=axes[2, 1],
)
fundamental_plots.plot_bucket_returns(
bucket_returns,
horizon=12,
ax=axes[3, 0],
)
portfolio_plots.plot_strategy_nav(
layer3_nav,
ax=axes[3, 1],
title="Costed growth of $1",
summary=layer3_performance,
apply_style=False,
)
portfolio_plots.plot_strategy_drawdowns(
layer3_nav,
ax=axes[4, 0],
title="Portfolio drawdowns",
summary=layer3_performance,
apply_style=False,
)
portfolio_plots.plot_risk_return_scatter(
layer3_plot,
ax=axes[4, 1],
title="Return, drawdown, and turnover",
risk_col="drawdown",
return_col="CAGR",
color_col="Sharpe",
size_col="Turnover",
apply_style=False,
)
axes[4, 1].xaxis.set_major_formatter(percent_formatter(1.0))
axes[4, 1].yaxis.set_major_formatter(percent_formatter(1.0))
portfolio_plots.apply_portfolio_subplot_layout(
fig,
axes,
hspace=0.45,
wspace=0.28,
bottom=0.04,
top=0.98,
)
plt.close(fig)
report_include = {
"snapshot": True,
"statements": True,
"profitability": True,
"cash_quality": True,
"growth": True,
"financial_strength": True,
"efficiency": True,
"capital_allocation": True,
"valuation": True,
"dupont": True,
"traditional_models": True,
"warnings": True,
"peer_comparison": True,
"score": True,
"score_history": True,
"summary": True,
}
report_statement_settings = {
"scale": "billions",
"periods": 8,
"show_common_size": True,
"show_growth": True,
}
report_history_settings = {
"periods": 12,
"frequency": "quarterly",
"rolling_periods": 4,
"show_latest_value": True,
}
report_peer_settings = {
"group": "industry",
"minimum_peers": 10,
"percentiles": (0.25, 0.50, 0.75),
"market_cap_band": (0.25, 4.0),
}
report_score_settings = {
"score": "selection_score",
"fundamental_score": "final_score",
"momentum_score": "momentum_score",
"show_blocks": True,
"show_rank": True,
"show_weight_history": True,
}
report_warning_settings = {
"active_only": True,
"minimum_severity": 1,
"show_history": True,
"show_penalty": True,
}
report_layout = {
"ncols": 2,
"sharex": False,
"sharey": False,
"figure_width": 11.0,
"panel_height": 3.2,
"combine_figures": True,
}
report_output = {
"round_tables": 4,
"display_tables": True,
"display_table_keys": ["fundamental_summary"],
"show_figures": True,
"display_figure_keys": ["overview"],
"print_summary": True,
"short_labels": False,
}
nke_report = fundamental_report(
metrics=fundamental_metrics,
scores=fundamental_scores,
ticker="NKE",
asof=latest_date,
include=report_include,
statement_settings=report_statement_settings,
history_settings=report_history_settings,
peer_settings=report_peer_settings,
score_settings=report_score_settings,
warning_settings=report_warning_settings,
layout=report_layout,
output=report_output,
)
axp_report = fundamental_report(
metrics=fundamental_metrics,
scores=fundamental_scores,
ticker="AXP",
asof=latest_date,
include=report_include,
statement_settings=report_statement_settings,
history_settings=report_history_settings,
peer_settings=report_peer_settings,
score_settings=report_score_settings,
warning_settings=report_warning_settings,
layout=report_layout,
output=report_output,
)
strategy_returns = pd.DataFrame(
{strategy_name: result.net_returns for strategy_name, result in layer3_results.items()}
).dropna(how="all")
market_return = layer1_results["Full EW"].net_returns.reindex(strategy_returns.index)
stress_windows = {
"2018 Q4": ("2018-10-01", "2018-12-31"),
"COVID crash": ("2020-02-20", "2020-04-30"),
"2022 inflation shock": ("2022-01-03", "2022-10-31"),
"2023 rebound": ("2023-01-03", "2023-12-29"),
}
available_stress_windows = {
stress_name: stress_dates
for stress_name, stress_dates in stress_windows.items()
if strategy_returns.index.min() <= pd.Timestamp(stress_dates[1])
and strategy_returns.index.max() >= pd.Timestamp(stress_dates[0])
}
layer3_risk_report = risk_report(
objects={
strategy_name: strategy_returns[strategy_name].dropna() for strategy_name in layer3_results
},
market_ret=market_return,
rf_daily=rf_daily,
include={
"performance_tables": True,
"shape_tables": False,
"drawdowns": False,
"rolling_vol": True,
"drawdown_episodes": False,
"var_es": True,
"var_backtest": True,
"stress": True,
"capm": True,
"rolling_beta": True,
"correlation": True,
"attribution": False,
"exec_bullets": False,
},
var_settings={
"alpha": 0.05,
"methods": ["hist", "cf", "fhs"],
},
backtest_settings={
"alpha": 0.05,
"methods": ["hist", "cf", "fhs"],
"lookback": 252,
"plot_method": "best",
},
rolling_settings={
"vol_windows": [20, 60, 252],
"beta_windows": [126, 252],
},
stress_settings={
"windows": available_stress_windows,
"worst_only": False,
},
layout={
"ncols": 3,
"sharex": True,
"sharey": False,
},
output={
"display_tables": False,
"display_table_keys": [
"var_es",
"stress",
"capm",
"corr",
],
"show_figures": False,
"print_exec_bullets": False,
"short_labels": False,
"round_tables": 4,
},
)
plt.close("all")
factor_columns = ["Mkt-RF", "SMB", "HML", "RMW", "CMA", "MOM"]
characteristic_columns = ["size", "momentum", "value", "beta", "volatility"]
ff5 = (
pd.read_csv(
repo_root / "data" / "fama_french_us_5_factors.csv",
parse_dates=["date"],
)
.set_index("date")
.sort_index()
)
momentum = (
pd.read_csv(
repo_root / "data" / "fama_french_us_momentum.csv",
parse_dates=["date"],
)
.set_index("date")
.sort_index()
)
academic_factors = ff5.join(momentum, how="inner")
monthly_returns = (1.0 + strategy_returns).resample("ME").prod().sub(1.0)
factor_results = factor_math.factor_attribution(
monthly_returns,
academic_factors,
factor_columns=factor_columns,
hac_lags=3,
)
market_return = factor_math.point_in_time_market_return(
returns,
monthly_issuers,
date_map,
)
market_variance = market_return.rolling(
252,
min_periods=189,
).var()
beta_daily = returns.rolling(252, min_periods=189).cov(
market_return
).divide(market_variance, axis=0)
volatility_daily = returns.rolling(
126,
min_periods=84,
).std() * np.sqrt(252.0)
decision_dates = pd.DatetimeIndex(date_map["decision_date"])
execution_dates = pd.DatetimeIndex(date_map["execution_date"])
decision_prices = adj_close_filled.reindex(decision_dates)
momentum_12_1 = decision_prices.shift(1).divide(
decision_prices.shift(12)
).sub(1.0)
execution_prices = adj_close_filled.reindex(execution_dates)
execution_prices.index = decision_dates
forward_returns = execution_prices.shift(-1).divide(
execution_prices
).sub(1.0)
decision_beta = beta_daily.reindex(decision_dates)
decision_beta.index = decision_dates
decision_volatility = volatility_daily.reindex(decision_dates)
decision_volatility.index = decision_dates
stock_factor_panel = pd.concat({
"forward_return": forward_returns.stack(),
"momentum": momentum_12_1.stack(),
"beta": decision_beta.stack(),
"volatility": decision_volatility.stack(),
}, axis=1)
stock_factor_panel.index.names = ["decision_date", "ticker"]
fama_macbeth_data = fundamental_metrics[[
"decision_date", "ticker", "industry", "market_cap", "book_to_market",
]].join(stock_factor_panel, on=["decision_date", "ticker"])
fama_macbeth_data["size"] = np.log(
fama_macbeth_data["market_cap"].where(
fama_macbeth_data["market_cap"].gt(0.0)
)
)
fama_macbeth_data["value"] = np.log(
fama_macbeth_data["book_to_market"].where(
fama_macbeth_data["book_to_market"].gt(0.0)
)
)
fama_macbeth = factor_math.fama_macbeth(
fama_macbeth_data,
date_column="decision_date",
return_column="forward_return",
characteristics=characteristic_columns,
industry_column="industry",
start=holdout_start,
min_cross_section=120,
min_industry_size=5,
hac_lags=3,
)
display(
factor_results.exposures.style
.format({
"months": "{:,.0f}", "annual_alpha": "{:.1%}",
"alpha_hac_t": "{:.2f}", "r_squared": "{:.1%}",
**{factor: "{:.2f}" for factor in factor_columns},
})
.set_caption("Library FF5 + momentum exposures and HAC alpha inference")
)
display(
factor_results.inference.style
.format("{:.2f}")
.set_caption("Library Newey–West HAC coefficient t-statistics")
)
display(
factor_results.attribution.style
.format("{:.1%}")
.set_caption("Library annualized factor attribution")
)
display(
fama_macbeth.summary.style
.format({
"annualized_premium": "{:.1%}", "annualized_hac_se": "{:.1%}",
"hac_t": "{:.2f}", "ci_95_low": "{:.1%}", "ci_95_high": "{:.1%}",
"positive_months": "{:.1%}", "months": "{:,.0f}",
"average_cross_section": "{:,.0f}",
"average_industry_controls": "{:.1f}",
})
.set_caption("Library monthly Fama–MacBeth premia with industry controls")
)