"""US price-to-book factor analysis (finlab.finance: us-stock-backtest-price-to-book-strategy). Reproduces every number in the article: B/P decile forward returns, monthly Rank IC, a research-window vs out-of-sample split, and a tradable check (cheapest-quintile top-50, monthly rebalance) benchmarked against QQQ and SPY. Run: uv run --no-project --python 3.12 --with "finlab>=2.0.13" \ --with plotly --with kaleido python strategy.py finlab.login() will guide you through authentication when data is requested. """ from __future__ import annotations import json from pathlib import Path import numpy as np import pandas as pd import plotly.graph_objects as go from plotly.subplots import make_subplots from finlab import data from finlab.backtest import sim START_DATE = "2016-01-01" def liquid_us_universe(close, volume, *, top_n, min_price=5.0, min_dollar_volume=5_000_000, max_abs_daily_return=0.50): """Rolling, point-in-time liquidity and data-quality universe.""" dollar_volume = (close * volume).rolling(60, min_periods=20).mean() top_liquid = dollar_volume.is_largest(top_n) liquid_enough = dollar_volume > min_dollar_volume price_ok = close > min_price returns = close.pct_change() extreme_today = returns.abs() >= max_abs_daily_return known_bad = extreme_today.rolling(252, min_periods=1).max().shift(1).fillna(False) > 0 clean = ~known_bad.astype(bool) return top_liquid & liquid_enough & price_ok & clean def restrict_start(position, start=START_DATE): position = position.index_str_to_date() if hasattr(position, "index_str_to_date") else position return position.loc[start:] def safe_columns(df, columns): existing = [c for c in columns if c in df.columns] if not existing: raise ValueError("None of the requested tickers are available.") return df[existing] def set_finlab_market(market): data.set_market(market) C_STRAT = "#2563EB" C_BENCH = "#9CA3AF" C_POS = "#10B981" C_NEG = "#EF4444" OUT = Path(__file__).resolve().parent / "us_pb_report" OUT.mkdir(parents=True, exist_ok=True) RESEARCH_END = "2021-12-31" N_DECILES = 10 def save(fig: go.Figure, name: str, title: str) -> None: fig.update_layout( template="plotly_white", width=1280, height=720, title={"text": title, "y": 0.96, "x": 0.02, "font": {"size": 20}}, margin={"t": 95, "b": 90, "l": 70, "r": 70}, legend={"orientation": "h", "y": -0.16, "x": 0}, ) fig.update_xaxes(title_standoff=14) fig.update_yaxes(title_standoff=14) fig.write_image(str(OUT / f"{name}.png"), scale=2) print(f"saved chart: {name}.png") def rebase(series: pd.Series, start: pd.Timestamp) -> pd.Series: s = series.loc[start:].dropna() return s / s.iloc[0] def main() -> None: close = data.get("us_price:adj_close") volume = data.get("us_price:volume") eps = data.get("us_income_statement:eps_diluted") net_income = data.get("us_income_statement:net_income") equity = data.get("us_balance_sheet:total_stockholders_equity") fund_close = safe_columns(data.get("us_fund_price:adj_close"), ["QQQ", "SPY"]) universe = liquid_us_universe( close, volume, top_n=500, min_price=10, min_dollar_volume=10_000_000 ) # Book-to-price. Shares outstanding are not a direct dataset, so derive the # weighted diluted share count from TTM net income / TTM diluted EPS and # guard the degenerate cases (tiny EPS, negative equity, sign mismatch). ttm_eps = eps.rolling(4).sum() ttm_ni = net_income.rolling(4).sum() shares = ttm_ni / ttm_eps book_per_share = equity / shares bp = book_per_share / close # higher = cheaper valid = ( (ttm_eps.abs() > 0.05) & (equity > 0) & (shares > 0) & universe ) bp = bp[valid] # ---- monthly cross-section: decile forward returns + Rank IC ----------- month_ends = close.loc[START_DATE:].resample("ME").last().index month_ends = [d for d in month_ends if d <= close.index.max()] px_m = close.reindex(close.index.union(month_ends)).ffill().loc[month_ends] fwd_ret = px_m.shift(-1) / px_m - 1 # next-month simple return per stock bp_daily = bp.reindex(close.index).ffill() bp_m = bp_daily.reindex(close.index.union(month_ends)).ffill().loc[month_ends] decile_rows: list[dict] = [] ic_rows: list[dict] = [] for dt in month_ends[:-1]: x = bp_m.loc[dt].dropna() y = fwd_ret.loc[dt].reindex(x.index).dropna() x = x.reindex(y.index) if len(x) < 100: continue ranks = x.rank(pct=True) dec = np.ceil(ranks * N_DECILES).clip(1, N_DECILES) grp = y.groupby(dec).mean() decile_rows.append({"date": dt, **{int(k): float(v) for k, v in grp.items()}}) ic_rows.append({"date": dt, "rank_ic": float(x.rank().corr(y.rank()))}) dec_df = pd.DataFrame(decile_rows).set_index("date").sort_index() ic_df = pd.DataFrame(ic_rows).set_index("date").sort_index() def annualize(m: pd.Series) -> float: return float((1 + m.mean()) ** 12 - 1) dec_ann_full = {c: annualize(dec_df[c].dropna()) for c in sorted(dec_df.columns)} dec_ann_research = { c: annualize(dec_df.loc[:RESEARCH_END, c].dropna()) for c in sorted(dec_df.columns) } dec_ann_oos = { c: annualize(dec_df.loc[RESEARCH_END:, c].dropna()) for c in sorted(dec_df.columns) } spread_full = dec_ann_full[N_DECILES] - dec_ann_full[1] # ---- tradable check: cheapest quintile, top 50 by B/P, monthly --------- quintile_cut = bp_daily[universe].rank(axis=1, pct=True) > 0.8 top50 = bp_daily[quintile_cut].rank(axis=1, ascending=False) <= 50 position = restrict_start(top50.resample("ME").last().reindex(close.index).ffill()) set_finlab_market("us") report = sim( position, resample="M", name="us_pb_cheapest50", upload=False, fee_ratio=0, tax_ratio=0, trade_at_price="close", position_limit=0.05, ) creturn = report.creturn.loc[START_DATE:] qqq = fund_close["QQQ"].loc[START_DATE:].dropna() spy = fund_close["SPY"].loc[START_DATE:].dropna() common_start = max(creturn.index[0], qqq.index[0], spy.index[0]) def stats(cum: pd.Series) -> dict: cum = rebase(cum, common_start) yrs = (cum.index[-1] - cum.index[0]).days / 365.25 cagr = float(cum.iloc[-1] ** (1 / yrs) - 1) daily = cum.pct_change().dropna() sharpe = float(daily.mean() / daily.std() * np.sqrt(252)) mdd = float((cum / cum.cummax() - 1).min()) vol = float(daily.std() * np.sqrt(252)) return {"cagr": cagr, "daily_sharpe": sharpe, "mdd": mdd, "ann_vol": vol, "total_return": float(cum.iloc[-1] - 1)} strat_stats = stats(creturn) qqq_stats = stats(qqq) spy_stats = stats(spy) # ---- charts -------------------------------------------------------- fig1 = go.Figure( go.Bar( x=[f"D{i}" for i in sorted(dec_ann_full)], y=[dec_ann_full[i] * 100 for i in sorted(dec_ann_full)], marker_color=[C_NEG if i < N_DECILES else C_POS for i in sorted(dec_ann_full)], name="年化報酬", ) ) fig1.update_yaxes(title_text="年化報酬(%)") fig1.update_xaxes(title_text="B/P 十分位(D10 = 最便宜)") save(fig1, "us_pb_decile_returns", "美股 B/P 十分位年化報酬(2016 至今,流動性前 500 檔)") fig2 = go.Figure() fig2.add_scatter(x=rebase(creturn, common_start).index, y=rebase(creturn, common_start), name="低 PB 前 50 檔(月再平衡)", line={"color": C_STRAT, "width": 2.4}) fig2.add_scatter(x=rebase(qqq, common_start).index, y=rebase(qqq, common_start), name="QQQ 含息", line={"color": C_BENCH, "width": 2}) fig2.add_scatter(x=rebase(spy, common_start).index, y=rebase(spy, common_start), name="SPY 含息", line={"color": "#D1D5DB", "width": 1.6, "dash": "dot"}) fig2.update_yaxes(title_text="累積淨值(共同起始日 = 1)") save(fig2, "us_pb_strategy_vs_qqq", "美股低 PB 組合 vs QQQ/SPY(同起始日重設基準)") fig3 = make_subplots(specs=[[{"secondary_y": True}]]) fig3.add_bar(x=ic_df.index, y=ic_df["rank_ic"], name="B/P Rank IC(月)", marker_color=[C_POS if v >= 0 else C_NEG for v in ic_df["rank_ic"]]) fig3.add_scatter(x=rebase(qqq, common_start).index, y=rebase(qqq, common_start), name="QQQ 累積(右軸)", line={"color": C_BENCH, "width": 1.8}, secondary_y=True) fig3.update_yaxes(title_text="Rank IC", secondary_y=False) fig3.update_yaxes(title_text="QQQ 累積淨值", secondary_y=True) save(fig3, "us_pb_rank_ic", "美股 B/P 月 Rank IC 與 QQQ 走勢對照") fig4 = go.Figure() for label, series, color in [ (f"2016–2021(研究段)", dec_ann_research, C_STRAT), (f"2022–(樣本外)", dec_ann_oos, C_NEG), ]: fig4.add_bar(x=[f"D{i}" for i in sorted(series)], y=[series[i] * 100 for i in sorted(series)], name=label, marker_color=color) fig4.update_yaxes(title_text="年化報酬(%)") fig4.update_xaxes(title_text="B/P 十分位(D10 = 最便宜)") save(fig4, "us_pb_decile_subperiod", "美股 B/P 十分位報酬:研究段 vs 樣本外") # ---- metrics ------------------------------------------------------------ ic_mean = float(ic_df["rank_ic"].mean()) ic_tstat = float(ic_df["rank_ic"].mean() / ic_df["rank_ic"].std() * np.sqrt(len(ic_df))) monthly = creturn.resample("ME").last().pct_change().dropna() metrics = { "generated": str(close.index.max().date()), "window": {"start": str(common_start.date()), "end": str(close.index.max().date()), "research_end": RESEARCH_END}, "universe": "us liquid top-500 by 60d dollar volume, price>=10, ADV>=10M", "bp_construction": "book_per_share = equity * ttm_eps / ttm_net_income; bp = book_per_share / adj_close; guards: |ttm_eps|>0.05, equity>0, shares>0", "decile_annualized_full": dec_ann_full, "decile_annualized_research": dec_ann_research, "decile_annualized_oos": dec_ann_oos, "d10_minus_d1_full": spread_full, "rank_ic_mean": ic_mean, "rank_ic_tstat": ic_tstat, "rank_ic_pct_positive": float((ic_df["rank_ic"] > 0).mean()), "strategy": {**strat_stats, "monthly_sharpe": float(monthly.mean() / monthly.std() * np.sqrt(12)), "avg_holdings": float(position.sum(axis=1).replace(0, np.nan).mean())}, "qqq": qqq_stats, "spy": spy_stats, "costs": "sim fee_ratio=0, tax_ratio=0 (US commissions ~0; slippage not modeled, disclosed in article)", } (OUT / "metrics.json").write_text(json.dumps(metrics, indent=2, ensure_ascii=False)) print(json.dumps({k: metrics[k] for k in ["d10_minus_d1_full", "rank_ic_mean", "rank_ic_tstat"]}, indent=2)) print("strategy:", {k: round(v, 4) for k, v in strat_stats.items()}) print("qqq:", {k: round(v, 4) for k, v in qqq_stats.items()}) report.to_html(str(OUT / "report_strategy.html")) print("wrote report_strategy.html + metrics.json") if __name__ == "__main__": main()