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")
)