"""市值型 ETF 實測比較與自組市值型回測：11 檔 ETF 的事實對照、18 組自組參數。 對應文章：/blog/market-cap-etf-comparison-backtest 全部設計與參數在看績效以前寫定，沒有依結果挑選。 ① 事實對照：0050、006208、00692、00850、00922、00923、00905、00912、00690、00921、009816 - 共同期間 2023-03-28 到 DATA_END（00922 上市隔天起）的含息報酬、波動、最大回撤、日夏普，加前後半段。 - 追蹤差距：淨值含息報酬（tw_etf_nav_daily 淨值＋配息）減去標的報酬指數（stock_index_price）。 00922 的 MSCI 指數沒有資料；009816 只算上市後。 - 台積電權重：DATA_END 那天的申購買回清單（tw_etf_pcf_daily）。 ② 市值加權加單檔上限：上市普通股總市值前 N 大，權重照市值，超過 C 的部分照市值比例分給其他成分股， 每季換股。N 30／50／100 × C 10%／20%／30%，原參數 N=50、C=20%。 ③ 前 N 大等權：總市值前 N 大各 1/N。N 30／50／100 × 每月／每季／每半年換股，原參數 N=50、每季。 ②③ 的共同設定： - 期間 2014-01-02 到 DATA_END。第一天與每期最後一個交易日收盤選股，sim() 下一個交易日收盤成交，預設手續費與證交稅。 - 上市普通股：company_basic_info 與 delisted_companies 的上市、下市日期，當天在證交所上市（上櫃轉上市從上市日算）， 排除存託憑證，當天要有收盤價。總市值用 etl:market_value，不做流通量調整。 - 基準：0050 含息價直接連乘，不經 sim()、不扣成本。 - 「贏」：日夏普（日報酬平均 ÷ 標準差 × √252，不扣無風險利率）高於同期 0050。 - 過關要四項都做到（全部扣預設成本）：9 組至少 7 組贏、原參數前後半段各自贏、原參數全期贏、 拿掉領先 0050 最多的那一年仍贏。 執行（finlab 需要資料時會自動引導登入）： pip install finlab python research.py # 輸出到目前目錄的 out/ python research.py --html # 另存 ② 原參數的互動回測報告 out/report_strategy.html """ import argparse import json import re from datetime import date from itertools import product from pathlib import Path import finlab import numpy as np import pandas as pd from finlab import data from finlab.backtest import sim DATA_END = "2026-11-12" START = "2014-01-02" BENCH = "0050" MIN_WINS = 7 ETFS = ["0050", "006208", "00692", "00850", "00922", "00923", "00905", "00912", "00690", "00921", "009816"] INDEX = { "0050": ["上市臺灣50報酬指數"], "006208": ["上市臺灣50報酬指數"], "00692": ["上市臺灣公司治理100報酬指數"], "00850": ["上市臺灣永續報酬指數"], "00905": ["上市特選Smart多因子報酬指數"], "00912": ["上市特選臺灣智慧50報酬指數"], "00921": ["上市特選臺灣產業龍頭存股等權重報酬指數"], "00923": ["上市臺灣ESG低碳50報酬指數"], "00690": ["上市藍籌30報酬指數", "上市藍籌 30 報酬指數"], "009816": ["上市特選臺灣TOP 50報酬指數"], "00922": [], } COMMON_START = "2023-03-27" # 00922 上市日收盤當基準，報酬從隔天算 NEW_START = "2026-02-03" # 009816 上市日 DIV_COL = "收益分配金額(每1受益權單位)" # ---------- 績效指標 ---------- def sharpe(ret): ret = ret.dropna() return float(ret.mean() / ret.std() * np.sqrt(252)) def metrics(ret, start): """start 是淨值＝1 的起點日（第一筆報酬的前一個交易日），年化天數從這天算起。""" ret = ret.dropna() eq = (1 + ret).cumprod() days = (ret.index[-1] - pd.Timestamp(start)).days return { "cagr": float(eq.iloc[-1] ** (365.25 / days) - 1), "sharpe": sharpe(ret), "mdd": float((eq / eq.cummax() - 1).min()), "volatility": float(ret.std() * np.sqrt(252)), "total": float(eq.iloc[-1] - 1), } def halves(ret, bench, start, name, bench_name): """依交易日數切成前後兩半，各自算績效；後半段從前半段最後一天起算。""" half = len(ret) // 2 parts = ((ret.iloc[:half], start), (ret.iloc[half:], ret.index[half - 1])) return [{"from": str(p.index[0].date()), "to": str(p.index[-1].date()), name: metrics(p, s0), bench_name: metrics(bench.reindex(p.index), s0)} for p, s0 in parts] def weekly_corr(a, b): weeks = pd.concat([(1 + a).resample("W-FRI").prod() - 1, (1 + b).resample("W-FRI").prod() - 1], axis=1).dropna() return float(weeks.corr().iloc[0, 1]) # ---------- 資料 ---------- class Market: """②③ 需要的價格、市值與上市名單。""" def __init__(self): self.adj = data.get("etl:adj_close").loc[:DATA_END] self.close = data.get("price:收盤價").loc[START:DATA_END] self.mv = data.get("etl:market_value").loc[START:DATA_END] self.bench_ret = self.adj[BENCH].ffill().pct_change() self.members = self.listed_common_stocks() def listed_common_stocks(self): """每天在證交所上市的普通股（含之後下市的），用上市、下市日期判斷。""" listings = [] for table, end_col in (("company_basic_info", None), ("delisted_companies", "下市日期")): df = data.get(table) for _, row in df[df["市場別"] == "sii"].iterrows(): sid = str(row["stock_id"]) name = str(row.get("公司簡稱", row.get("公司名稱", ""))) is_tdr = "DR" in name or "存託憑證" in str(row.get("產業類別", "")) listed = pd.to_datetime(row["上市日期"], errors="coerce") if not re.fullmatch(r"[1-9]\d{3}", sid) or is_tdr or pd.isna(listed): continue ended = pd.to_datetime(row[end_col], errors="coerce") if end_col else pd.NaT listings.append((sid, listed, ended)) days = self.close.index stocks = sorted({s for s, _, _ in listings} & set(self.close.columns) & set(self.mv.columns)) members = pd.DataFrame(False, index=days, columns=stocks) for sid, listed, ended in listings: if sid in members.columns: on = days >= listed if pd.notna(ended): on &= days < ended members[sid] |= on return members def caps_on(self, day): """當天上市、有收盤價的普通股總市值，由大到小。""" listed = self.members.loc[day] caps = self.mv.loc[day, listed[listed].index] has_price = self.close.loc[day, caps.index].notna() return caps[(caps > 0) & has_price].sort_values(ascending=False) def rebalance_days(self, freq): """第一天加上每期最後一個交易日；freq 是 M（月）、Q（季）或 H（6、12 月底）。""" days = self.close.index months = {"M": 1, "Q": 3, "H": 6}[freq] period = (days.year * 12 + days.month - 1) // months last_days = days.to_series().groupby(period).last() return pd.DatetimeIndex(sorted({days[0], *last_days})) # ---------- ②③ 權重 ---------- def capped_weights(caps, cap): """照市值配權重，超過上限的部分照市值比例分給還沒到上限的成分股，重複到沒有超過。""" weights = caps / caps.sum() capped = pd.Series(False, index=caps.index) while True: free = ~capped weights[free] = caps[free] / caps[free].sum() * (1 - cap * capped.sum()) over = free & (weights > cap + 1e-12) if not over.any(): return weights capped |= over weights[capped] = cap def equal_weights(caps, n): return pd.Series(1 / n, index=caps.index[:n]) def build_position(market, days, weight_fn): targets = pd.DataFrame({day: weight_fn(market.caps_on(day)) for day in days}).T.fillna(0.0) position = targets.reindex(market.close.index).ffill().fillna(0.0) return position.loc[:, (position > 0).any()] def run(position, days, **kwargs): report = sim(position, resample=days, end_date=DATA_END, upload=False, **kwargs) equity = report.creturn.loc[position.index[0]:DATA_END] return report, equity.pct_change().iloc[1:] def cap_weighted(market, params): """② 前 n 大市值加權，單檔上限 cap，每季換股。""" n, cap = params days = market.rebalance_days("Q") return build_position(market, days, lambda caps: capped_weights(caps.iloc[:n], cap)), days def equal_weighted(market, params): """③ 前 n 大等權，freq 換股。""" n, freq = params days = market.rebalance_days(freq) return build_position(market, days, lambda caps: equal_weights(caps, n)), days DESIGNS = { "cap_weighted": dict(grid=list(product([30, 50, 100], [0.1, 0.2, 0.3])), original=(50, 0.2), build=cap_weighted), "equal_weighted": dict(grid=list(product([30, 50, 100], ["M", "Q", "H"])), original=(50, "Q"), build=equal_weighted), } def annual_turnover(market, targets, days): """每年單邊周轉率：換股日目標權重與持有期間漂移後權重的差，取一半加總後年化，不含第一次建倉。""" prices = market.adj.loc[START:DATA_END].ffill() total, prev, prev_day = 0.0, None, None for day in days: target = targets.loc[day] target = target[target > 0] if prev is not None: drifted = prev * (prices.loc[day, prev.index] / prices.loc[prev_day, prev.index]).fillna(1.0) drifted = drifted / drifted.sum() names = drifted.index.union(target.index) total += (target.reindex(names, fill_value=0) - drifted.reindex(names, fill_value=0)).abs().sum() / 2 prev, prev_day = target, day return float(total / ((days[-1] - days[0]).days / 365.25)) def evaluate(market, cfg, key): out = {"start": START, "end": DATA_END, "original": list(cfg["original"]), "grid": []} for params in cfg["grid"]: position, days = cfg["build"](market, params) report, ret = run(position, days) bench = market.bench_ret.reindex(ret.index) row = metrics(ret, position.index[0]) bench_sharpe = sharpe(bench) row.update(params=list(params), benchmark_sharpe=bench_sharpe, wins=row["sharpe"] > bench_sharpe) out["grid"].append(row) print(key, params, f"日夏普 {row['sharpe']:.3f}，0050 {bench_sharpe:.3f}") if params == cfg["original"]: original = (position, days, report, ret, bench) position, days, report, ret, bench = original start = position.index[0] out["strategy"] = metrics(ret, start) out["benchmark"] = metrics(bench, start) out["trades"] = int(len(report.get_trades())) held = position.loc[days] out["avg_tsmc_weight"] = float(held.get("2330", pd.Series(0.0)).mean()) out["avg_names"] = float((held > 0).sum(axis=1).mean()) out["turnover_one_way_annual"] = annual_turnover(market, held, days) out["halves"] = halves(ret, bench, start, "strategy", "benchmark") for h in out["halves"]: h["wins"] = h["strategy"]["sharpe"] > h["benchmark"]["sharpe"] out["neighbor_wins"] = int(sum(g["wins"] for g in out["grid"])) out["neighbor_total"] = len(out["grid"]) yearly = (1 + ret).groupby(ret.index.year).prod() - 1 bench_yearly = (1 + bench).groupby(bench.index.year).prod() - 1 out["yearly"] = {int(y): float(v) for y, v in yearly.items()} out["benchmark_yearly"] = {int(y): float(v) for y, v in bench_yearly.items()} lead_year = int((yearly - bench_yearly).idxmax()) each_year = {int(y): {"strategy_sharpe": sharpe(ret[ret.index.year != y]), "benchmark_sharpe": sharpe(bench[bench.index.year != y])} for y in yearly.index} lead = each_year[lead_year] out["without_lead_year"] = {"year": lead_year, **lead, "wins": lead["strategy_sharpe"] > lead["benchmark_sharpe"]} out["without_each_year"] = each_year _, no_cost = run(position, days, fee_ratio=0, tax_ratio=0) out["no_cost"] = metrics(no_cost, start) out["weekly_corr_0050"] = weekly_corr(ret, bench) out["checks"] = {"neighbors": out["neighbor_wins"] >= MIN_WINS, "halves": all(h["wins"] for h in out["halves"]), "full_with_costs": out["strategy"]["sharpe"] > out["benchmark"]["sharpe"], "without_lead_year": out["without_lead_year"]["wins"]} out["passed"] = all(out["checks"].values()) return out, ret, report def uncapped_reference(market): """描述用：不設上限（上限 100%）的前 50 大市值加權，看自組版跟 0050 有多像，不參與過關判定。""" position, days = cap_weighted(market, (50, 1.0)) _, ret = run(position, days) bench = market.bench_ret.reindex(ret.index) held = position.loc[days] start = position.index[0] return {"uncapped_top50_q": metrics(ret, start), "benchmark": metrics(bench, start), "weekly_corr_0050": weekly_corr(ret, bench), "avg_tsmc_weight": float(held["2330"].mean()), "tsmc_weight_last": float(held["2330"].iloc[-1])} # ---------- ① ETF 事實對照 ---------- def nav_total_return(n, d): """每日淨值含息報酬：除息日把配息加回；淨值單日跌到四成以下或漲到 1.6 倍以上當成分割處理。""" ratio = n / n.shift(1) split = (ratio < 0.6) | (ratio > 1.6) factor = pd.Series(1.0, index=n.index) factor[split] = (1 / ratio[split]).round() at = n.index.searchsorted(d["date"]) # 除息日沒有淨值時，記到下一個有淨值的日子 d = d[at < len(n)].assign(at=n.index[at[at < len(n)]]) paid = d.groupby("at")[DIV_COL].sum().reindex(n.index).fillna(0.0) ret = (n * factor + paid) / n.shift(1) - 1 return ret.iloc[1:], {str(k.date()): float(v) for k, v in factor[split].items()} def index_return(sid, index_prices, days): """標的報酬指數，照淨值的日期區間算報酬，淨值缺日時兩邊涵蓋同樣的天數。""" cols = INDEX[sid] if not cols: return None level = index_prices[cols[0]] for col in cols[1:]: level = level.combine_first(index_prices[col]) level = level.dropna() return level.reindex(level.index.union(days)).ffill().reindex(days).pct_change() def after(ret, start): return ret.loc[pd.Timestamp(start) + pd.Timedelta(days=1):DATA_END] def pcf_weights(sid, payload): """從各投信的申購買回清單取出股票權重（%），格式依投信不同。""" p = json.loads(payload) def pct(x): return float(str(x).replace("%", "").replace(",", "")) if sid in ("0050", "00850"): rows = [(r["code"], r["weights"]) for r in p["FundWeights"]["StockWeights"]] elif sid in ("006208", "00692"): rows = [(r[0], r[-1]) for r in p["assets_tables"][1][1:]] elif sid == "009816": rows = [(r[0], r[-1]) for r in p["assets_tables"][0][1:]] elif sid in ("00690", "00921"): rows = [(r[0], r[-1]) for r in p["pcf_tables"][0][1:]] elif sid == "00922": rows = [(r["stockCode"], r["weights"]) for r in p["holdings"]] elif sid == "00923": rows = [(r["stocNo"], r["weight"]) for r in p["pcf"]["stocks"]] elif sid == "00905": rows = [(r["SecuritiesCode"], r["WeightingPercentage"]) for r in p["holdings"]["Secs"] if r.get("SecuritiesType") == "S"] elif sid == "00912": rows = [(r["code_"], r["weights_"]) for r in p["pcf"]["Detail"][0]["Data"] if r.get("invtp_") == "STOCK"] else: raise KeyError(sid) weights = pd.Series({code: pct(w) for code, w in rows}) return weights[[c.isdigit() and len(c) == 4 for c in weights.index]] def holdings_facts(sid, payload, day, market_value): w = pcf_weights(sid, payload) caps = market_value.loc[day].reindex(w.index) ok = caps.notna() ww, cc = w[ok] / w[ok].sum(), caps[ok] / caps[ok].sum() others_w, others_c = ww.drop("2330", errors="ignore"), cc.drop("2330", errors="ignore") top = w.sort_values(ascending=False) return {"date": str(day.date()), "n": int(len(w)), "tsmc": float(w.get("2330", 0.0)), "top10": float(top.head(10).sum()), "top3": [[k, float(v)] for k, v in top.head(3).items()], "max_weight": float(top.iloc[0]), "min_weight": float(top.iloc[-1]), "corr_cap_ex_tsmc": float(np.corrcoef(others_w / others_w.sum(), others_c / others_c.sum())[0, 1])} def etf_facts(market): nav = data.get("tw_etf_nav_daily:淨值").loc[:DATA_END] premium = data.get("tw_etf_nav_daily:折溢價(%)").loc[:DATA_END] dividends = data.get("tw_etf_dividend_events") dividends = dividends[dividends["date"] <= DATA_END] index_prices = data.get("stock_index_price:收盤指數").loc[:DATA_END] pcf = data.get("tw_etf_pcf_daily:source_payload").loc[:DATA_END] info = data.get("tw_etf_basic_info").set_index("symbol") bench = market.bench_ret facts = {} for sid in ETFS: price = market.adj[sid] first = price.first_valid_index() ret = price.loc[first:].ffill().pct_change().iloc[1:] f = {"name": str(info.loc[sid, "證券簡稱"]), "listed": str(info.loc[sid, "上市日期"])[:10], "index": str(info.loc[sid, "標的指數"]), "price_start": str(first.date()), "since_listing": metrics(ret, first), "since_listing_0050": metrics(bench.reindex(ret.index), first)} for label, start in (("common", COMMON_START), ("since_009816", NEW_START)): if price.loc[:start].dropna().empty: continue part = after(ret, start) f[label] = metrics(part, start) f[label + "_halves"] = halves(part, bench, start, "etf", "0050") if sid != BENCH: f[label + "_weekly_corr_0050"] = weekly_corr(part, bench.reindex(part.index)) n = nav[sid].dropna() d = dividends[dividends["symbol"] == sid].dropna(subset=[DIV_COL]) nav_ret, f["nav_splits"] = nav_total_return(n, d) start = max(COMMON_START, f["listed"]) # 比共同期間晚上市的（009816）從上市日起算 nav_part = after(nav_ret, start) f["nav_window"] = [str(nav_part.index[0].date()), str(nav_part.index[-1].date())] nav_start = n.index[n.index.get_loc(nav_part.index[0]) - 1] f["nav_tr"] = metrics(nav_part, nav_start) idx_ret = index_return(sid, index_prices, n.index) if idx_ret is not None: idx_part = idx_ret.reindex(nav_part.index) f["index_tr"] = metrics(idx_part, nav_start) f["tracking_difference"] = f["nav_tr"]["cagr"] - f["index_tr"]["cagr"] for year in (2024, 2025): n_y, i_y = nav_part[nav_part.index.year == year], idx_part[idx_part.index.year == year] if len(n_y) > 200: f[f"td_{year}"] = float((1 + n_y).prod() - (1 + i_y).prod()) if start > COMMON_START: first_month = nav_part.index.to_period("M") == nav_part.index[0].to_period("M") f["listing_month_gap"] = float((1 + nav_part[first_month]).prod() - (1 + idx_part[first_month]).prod()) prem = after(premium[sid].dropna(), start) f["premium_mean"] = float(prem.mean()) d2025 = d[d["date"].dt.year == 2025] f["dividend_months_2025"] = sorted(set(d2025["date"].dt.month.tolist())) f["dividends_2025"] = float(d2025[DIV_COL].sum()) payloads = pcf[sid].dropna() f["holdings"] = holdings_facts(sid, payloads.iloc[-1], payloads.index[-1], market.mv) f["tsmc_weight"] = f["holdings"]["tsmc"] facts[sid] = f print(sid, f["name"], f"台積電 {f['tsmc_weight']:.2f}%") return facts def extras(market): """文中引用的其他數字：台積電逐年含息漲幅、台積電占上市前 50 大總市值、2024 年 8 月 0050 與 006208 的單日漲跌。""" tsmc = market.adj["2330"].ffill() year_end = tsmc.groupby(tsmc.index.year).last() yearly = year_end.pct_change().dropna() share = {} for day in (COMMON_START, DATA_END): top50 = market.caps_on(pd.Timestamp(day)).head(50) share[day] = float(top50.get("2330", 0.0) / top50.sum()) pair = market.adj[["0050", "006208"]].ffill().pct_change().loc["2024-08-02":"2024-08-08"] return {"tsmc_yearly_return": {int(y): float(v) for y, v in yearly.loc[2014:].items()}, "tsmc_share_top50_total_cap": share, "aug2024_daily_return": {str(d.date()): {c: float(v) for c, v in row.items()} for d, row in pair.iterrows()}} def main(): parser = argparse.ArgumentParser() parser.add_argument("--out", default="out") parser.add_argument("--html", action="store_true") args = parser.parse_args() out_dir = Path(args.out) out_dir.mkdir(parents=True, exist_ok=True) market = Market() results = {"data_end": DATA_END, "finlab_version": finlab.__version__, "run_date": str(date.today()), "etfs": etf_facts(market)} curves, reports = {}, {} for key, cfg in DESIGNS.items(): results[key], curves[key], reports[key] = evaluate(market, cfg, key) if args.html: reports["cap_weighted"].to_html(str(out_dir / "report_strategy.html")) results["reference"] = uncapped_reference(market) results["extras"] = extras(market) daily = pd.DataFrame(curves) daily[f"{BENCH}含息"] = market.bench_ret.reindex(daily.index) equity = (1 + daily).cumprod() first_day = market.close.index[0] equity = pd.concat([pd.DataFrame(1.0, index=[first_day], columns=equity.columns), equity]) equity.index.name = "date" equity.round(6).to_csv(out_dir / "data.csv") (out_dir / "metrics.json").write_text(json.dumps(results, ensure_ascii=False, indent=1)) for key in DESIGNS: r = results[key] print(key, f"{r['neighbor_wins']}/{r['neighbor_total']} 組贏", "過關" if r["passed"] else "未過關") if __name__ == "__main__": main()