""" 台股 ETF 成分股調整與 ETF 散戶熱度：5 個訊號單獨版，加上 5 個組合版，三項驗證（至最新資料） 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個策略 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 組合版另外要原參數日夏普高於同期、同持股數的底策略，而且 9 組裡至少 7 組高於同持股數的底。 指數規則照 FTSE TWSE Taiwan Index Series Ground Rules v5.2 與 Dividend+ Index Ground Rules v4.2： 台灣 50、中型 100 每年 3、6、9、12 月審核，高股息 6、12 月；資料截止日是生效週一前 4 週的週一， 第三個星期五收盤後換股。上市股票依全市值排名：台灣 50 排名 ≤40 納入、≥61 剔除；中型 100 ≤130 納入、≥171 剔除。 公眾流通量、流動性、全額交割三項篩選沒有資料，沒有做；高股息用的 I/B/E/S 預測殖利率改用證交所公布的殖利率。 台灣 50 從 2013-06 審核前的實際成分股起算，中型 100 從同一次審核以規則起算，之後每季照規則遞推，不再讀實際名單； 實際名單（維基百科逐季變動表，從 2026-10-30 的 0050 申購買回清單倒推）只用來抽查與事件研究。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 資料時點：資料可取得那天收盤後才用，sim 在下一個交易日收盤成交。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入。 輸出：metrics.json、data.csv（各策略原參數與 0050 含息的每日淨值）。 """ import collections import datetime import functools import json import math import numpy as np import pandas as pd from finlab import data from finlab.backtest import sim from finlab.dataframe import FinlabDataFrame EVENT_START = "2013-05-01" # 市值資料 2013-04 起，第一次重建的審核是 2013-06 COMBO_START = "2015-01-01" # 組合版與底策略頁同一段 DIVIDEND_START = "2023-01-01" # 高股息 50 檔規則 2022-12 起，該次以規則起算，2023-06 起才有預測 def plain(key): return pd.DataFrame(data.get(key)) # --------------------------------------------------------------------------- # 1. 共同資料：股票池、0050、績效計算 # --------------------------------------------------------------------------- adj_close = plain("etl:adj_close") bench = adj_close["0050"].dropna() days = adj_close.index basic = plain("company_basic_info").set_index("stock_id") market = plain("security_categories").set_index("stock_id")["market"] cap_all = plain("etl:market_value").reindex(days) def ordinary_shares(*markets): return [s for s in cap_all.columns if len(s) == 4 and s.isdigit() and not s.startswith(("0", "91")) and market.get(s) in markets] stocks = ordinary_shares("sii") col_of = {s: i for i, s in enumerate(stocks)} foreign = frozenset(basic.index[basic["外國企業註冊地國"].ne("－")]) listed_on = pd.to_datetime(basic["上市日期"], errors="coerce").reindex(stocks).fillna(pd.Timestamp("1900-01-01")) cap = cap_all[stocks].where(days.values[:, None] >= listed_on.values[None, :]) # 上櫃轉上市前不算 def stats(curve): curve = curve.dropna() curve = curve / curve.iloc[0] ret = curve.pct_change().dropna() years = (curve.index[-1] - curve.index[0]).days / 365.25 return { "start": str(curve.index[0].date()), "end": str(curve.index[-1].date()), "cagr": round(float(curve.iloc[-1] ** (1 / years) - 1), 6), "sharpe": round(float(ret.mean() / ret.std() * math.sqrt(252)), 6), "mdd": round(float((curve / curve.cummax() - 1).min()), 6), "vol": round(float(ret.std() * math.sqrt(252)), 6), } def active(curve): """從淨值第一次變動的前一個交易日起算。""" curve = curve.dropna() moved = np.flatnonzero(curve.values != curve.values[0]) return curve.iloc[max(moved[0] - 1, 0):] if len(moved) else curve def versus_0050(curve): s = stats(active(curve)) b = stats(bench.loc[s["start"]:s["end"]]) return {"strategy": s, "bench": b, "win": s["sharpe"] > b["sharpe"]} def halves(curve): curve = active(curve) mid = curve.index[0] + (curve.index[-1] - curve.index[0]) / 2 return [versus_0050(curve.loc[lo:hi]) for lo, hi in ((curve.index[0], mid), (mid, curve.index[-1]))] def trade_stats(report): trades = report.get_trades() curve = active(report.creturn) years = (curve.index[-1] - curve.index[0]).days / 365.25 held_days = (pd.to_datetime(trades["exit_date"]).fillna(curve.index[-1]) - pd.to_datetime(trades["entry_date"])).dt.days holdings = report.signal.sum(axis=1) holdings = holdings.loc[holdings.gt(0).idxmax():] return { "trades": int(len(trades)), "trades_per_year": round(len(trades) / years, 1), "avg_hold_calendar_days": round(float(held_days.mean()), 1), "avg_holdings": round(float(holdings[holdings > 0].mean()), 1), "share_of_days_holding": round(float((holdings > 0).mean()), 4), "win_rate": round(float((trades["return"] > 0).mean()), 4), "mean_return": round(float(trades["return"].mean()), 6), } def fees(cost): return {} if cost == "default" else {"fee_ratio": 0, "tax_ratio": 0} def run(position, cost, **kw): report = sim(position, upload=False, **fees(cost), **kw) report.signal = position.astype(bool) # 送進 sim 的持有表，用來算持股檔數 return report def backtest(position, cost, start=EVENT_START): position = position.reindex(index=days).fillna(False).astype(bool).loc[start:] return run(position.loc[:, position.any()], cost) def windows(rows): """rows：（股票, 開始位置, 結束位置〔含〕）→ 每天的訊號表。sim 在訊號隔天收盤成交：lo+1 收盤買進、hi+2 收盤賣出。""" held = np.zeros((len(days), len(stocks)), dtype=bool) for stock_id, lo, hi in rows: if stock_id in col_of and lo <= hi and lo < len(days): held[max(lo, 0):min(hi, len(days) - 1) + 1, col_of[stock_id]] = True return pd.DataFrame(held, index=days, columns=stocks) # --------------------------------------------------------------------------- # 2. 審核日曆與規則重建 # --------------------------------------------------------------------------- def last_trading_day(ts): """ts 當天或之前最後一個交易日的位置。""" return int(days.searchsorted(ts, side="right")) - 1 def friday_on_or_after(ts): return ts + pd.Timedelta(days=(4 - ts.weekday()) % 7) REVIEWS = [] for year in range(2013, days[-1].year + 1): for month in (3, 6, 9, 12): friday = friday_on_or_after(pd.Timestamp(year, month, 15)) # 第三個星期五 if pd.Timestamp("2013-06-01") <= friday <= days[-1]: REVIEWS.append({ "month": f"{year}-{month:02d}", "D": last_trading_day(friday), # 換股日收盤 "C": last_trading_day(friday - pd.Timedelta(days=25)), # 資料截止日（生效週一前 4 週的週一） "A": last_trading_day(friday_on_or_after(pd.Timestamp(year, month, 1))), # 第一個星期五盤後公告 }) @functools.lru_cache(maxsize=None) def rank_at(p): """交易日位置 p 收盤的全市值排名（1＝最大），只含當天有市值的上市股票。""" return cap.iloc[p].dropna().rank(ascending=False, method="first") def resize(after, order, adds, n): """刪減或補足到 n 檔：order 是排名順序，adds 是這次納入、不刪的股票。""" for s in reversed(order): # 太多：刪排名最後的原成分股 if len(after) <= n: break if s in after and s not in adds: after.discard(s) for s in order: # 太少：補排名最前的非成分股 if len(after) >= n: break after.add(s) return frozenset(after) def buffer_review(before, rank, n, add_at, drop_at, exclude=frozenset()): """規則 6.3：排名 ≤add_at 的非成分股納入、≥drop_at（或已不在股票池）的成分股剔除，再刪減或補足到 n 檔。""" order = [s for s in rank.sort_values().index if s not in exclude] adds = {s for s in order if rank[s] <= add_at and s not in before} after = {s for s in before if s in rank.index and rank[s] < drop_at} | adds return resize(after, order, adds, n) def index_review(tw_before, mid_before, p, tw_add=40, tw_drop=61, mid_add=130, mid_drop=171): rank = rank_at(p) tw_after = buffer_review(tw_before, rank, 50, tw_add, tw_drop) mid_in = (mid_before | (tw_before - tw_after)) - tw_after # 規則 6.3.3：台灣 50 剔除股轉入、納入股移出 return tw_after, buffer_review(mid_in, rank, 100, mid_add, mid_drop, exclude=tw_after) # 2013-06 審核前的台灣 50（從今天的 0050 倒推；3711 日月光投控 2018 年才成立，當時是 2311 日月光） TW50_SEED = frozenset(""" 1101 1102 1216 1301 1303 1326 1402 1722 2002 2105 2201 2207 2301 2303 2308 2311 2317 2324 2325 2330 2347 2353 2354 2357 2382 2409 2412 2454 2474 2498 2801 2880 2881 2882 2883 2885 2886 2890 2891 2892 2912 3008 3045 3231 3481 3673 3697 4904 5880 6505""".split()) tw = TW50_SEED mid = frozenset([s for s in rank_at(REVIEWS[0]["C"]).sort_values().index if s not in tw][:100]) for r in REVIEWS: r["tw_before"], r["mid_before"] = tw, mid tw, mid = index_review(tw, mid, r["C"]) r["tw_after"], r["mid_after"] = tw, mid # 高股息：證交所殖利率、6 個月均成交金額、6 個月含息報酬、權益／資產、股本／資產 pe_yield_all = plain("price_earning_ratio:殖利率(%)").reindex(days).ffill(limit=5) pe_yield = pe_yield_all.reindex(columns=stocks) traded_value = plain("price:成交金額").reindex(days) avg_value_6m = traded_value.reindex(columns=stocks).rolling(126, min_periods=60).mean() total = adj_close.reindex(columns=stocks) ret_6m = total / total.shift(126) - 1 def statement_ratio(numerator, denominator): ratio = data.get(f"financial_statement:{numerator}") / data.get(f"financial_statement:{denominator}") return pd.DataFrame(ratio.index_str_to_date()).reindex(columns=stocks).reindex(days, method="ffill") equity_assets = statement_ratio("股東權益總額", "資產總額") capital_assets = statement_ratio("股本", "資產總額") def dividend_review(before, universe, p, add_at=35, drop_at=66): """Dividend+ 規則 4.2：殖利率排名 ≤add_at 的非成分股納入、≥drop_at 的成分股剔除，各最多 5 檔，再補足或刪減到 50 檔。""" pool = pd.Index(sorted(universe)) y = pe_yield.iloc[p].reindex(pool) v, r6 = avg_value_6m.iloc[p].reindex(pool), ret_6m.iloc[p].reindex(pool) ea, ca = equity_assets.iloc[p].reindex(pool), capital_assets.iloc[p].reindex(pool) screened = ((v < v.quantile(0.1)) | ((r6 < r6.quantile(0.1)) & (r6 < 0)) | r6.isna() | (ea < ea.quantile(0.1)) | (ca < ca.quantile(0.05))) # 4.2.5–4.2.7 只篩非成分股 eligible = y.index[(y > 0) & (y.index.isin(list(before)) | ~screened)] rank = y[eligible].rank(ascending=False, method="first") order = list(rank.sort_values().index) if not before: return frozenset(order[:50]) members = {s for s in before if s in rank.index} # 不在母體、零股利的直接剔除 drops = [s for s in reversed(order) if s in members and rank[s] >= drop_at][:5] adds = [s for s in order if s not in members and rank[s] <= add_at][:5] return resize((members - set(drops)) | set(adds), order, adds, 50) div = frozenset() for r in REVIEWS: if r["month"] < "2022-12": continue universe = (r["tw_after"] | r["mid_after"]) - foreign r["div_before"] = div & universe # 季審核被剔出台灣 50／中型 100 的同步剔除 if r["month"][5:] in ("06", "12"): div = dividend_review(div, universe, r["C"]) else: div = r["div_before"] r["div_after"] = div DIV_REVIEWS = [i for i, r in enumerate(REVIEWS) if r["month"] >= "2023-06" and r["month"][5:] in ("06", "12")] # --------------------------------------------------------------------------- # 3. 預測名單：資料只用到預測日 p（＝進場訊號日與截止日較早的那天） # --------------------------------------------------------------------------- @functools.lru_cache(maxsize=None) def tw50_adds(i, p, add_at): r = REVIEWS[i] return index_review(r["tw_before"], r["mid_before"], p, tw_add=add_at)[0] - r["tw_before"] @functools.lru_cache(maxsize=None) def tw50_drops(i, drop_at): r = REVIEWS[i] after = index_review(r["tw_before"], r["mid_before"], r["C"], tw_drop=drop_at)[0] return frozenset(s for s in r["tw_before"] - after if s in col_of and not np.isnan(cap.iat[r["D"], col_of[s]])) # 換股日還在交易 @functools.lru_cache(maxsize=None) def mid100_adds(i, p, add_at): r = REVIEWS[i] return index_review(r["tw_before"], r["mid_before"], p, mid_add=add_at)[1] - r["mid_before"] - r["tw_before"] @functools.lru_cache(maxsize=None) def dividend_adds(i, p, add_at): r = REVIEWS[i] tw_after, mid_after = index_review(r["tw_before"], r["mid_before"], p) return dividend_review(r["div_before"], (tw_after | mid_after) - foreign, p, add_at=add_at) - r["div_before"] def before_review(adds_fn, N, R, reviews, exit_after=0): """D 前 N 個交易日收盤買、D 後 exit_after 個交易日收盤賣。""" rows = [] for i in reviews: r = REVIEWS[i] lo = r["D"] - N - 1 rows += [(s, lo, r["D"] + exit_after - 2) for s in adds_fn(i, min(lo, r["C"]), R)] return windows(rows) ALL = range(len(REVIEWS)) def tw50_add(N, R, cost): """S1. 台灣 50 預期納入。""" return backtest(before_review(tw50_adds, N, R, ALL), cost) def tw50_drop(R, H, cost): """S2. 台灣 50 預期剔除：換股日收盤買、持有 H 日。""" rows = [(s, REVIEWS[i]["D"] - 1, REVIEWS[i]["D"] + H - 2) for i in ALL for s in tw50_drops(i, R)] return backtest(windows(rows), cost) def mid100_add(N, R, cost): """S3. 中型 100 預期納入（兩個指數都不在的股票）。""" return backtest(before_review(mid100_adds, N, R, ALL), cost) def dividend_add(N, R, cost): """S4. 高股息預期納入：賣在換股日後第 5 個交易日（規則 4.6.1 分 5 天換股）。""" return backtest(before_review(dividend_adds, N, R, DIV_REVIEWS, exit_after=5), cost, start=DIVIDEND_START) # --------------------------------------------------------------------------- # 4. ETF 受益人數擇時：投信投顧公會月資料，次月月底後一天起才算知道 # --------------------------------------------------------------------------- beneficiary = plain("tw_etf_beneficiary_stats") holders = beneficiary.set_index(pd.to_datetime(beneficiary["date"]))["全體ETF受益人數"].sort_index() def surge_by_month(K, q): """K 個月成長率落在自己近 60 個月（至少 24 個）前 q → 暴增（1）；資料不足為 NaN。index 是資料月份。""" growth = holders / holders.shift(K) - 1 threshold = growth.shift(1).rolling(60, min_periods=24).quantile(1 - q) return (growth >= threshold).astype(float).where(threshold.notna()) @functools.lru_cache(maxsize=None) def surge(K, q): """每天能看到的暴增狀態。""" state = surge_by_month(K, q) state.index = state.index + pd.offsets.MonthEnd(1) + pd.Timedelta(days=1) return state.reindex(days.union(state.index)).ffill().reindex(days) K_GRID, Q_GRID = ("K", [1, 3, 6], 3), ("q", [0.1, 0.2, 0.3], 0.2) TIMING_START = str(max(surge(K, q).first_valid_index() for K in K_GRID[1] for q in Q_GRID[1]).date()) def timing(on, off, K, q, cost, start): s = surge(K, q).loc[start:] position = pd.DataFrame({on: s.eq(0)}) if off: position[off] = s.eq(1) return run(position, cost) def etf_holders(K, q, cost): """S6. 不暴增持有 0050，暴增持有現金。""" return timing("0050", None, K, q, cost, TIMING_START) def etf_holders_leveraged(K, q, cost): """C6. 不暴增持有 00631L，暴增持有 0050。""" return timing("00631L", "0050", K, q, cost, "2014-10-31") # --------------------------------------------------------------------------- # 5. 組合版：底策略＝月營收動能選股（/strategies/monthly-revenue-momentum） # 趨勢（站上 20／60／120 日線）＋近 3 月營收均值 > 近 12 月，5 日平均漲跌幅前 N 檔，每月 11 日換股，10% 停損 # --------------------------------------------------------------------------- close = data.get("price:收盤價") rev = data.get("monthly_revenue:當月營收") rev_up = rev.average(3) > rev.average(12) trend = (close > close.average(20)) & (close > close.average(60)) & (close > close.average(120)) momentum = (close / close.shift() - 1).rolling(5).mean() candidates = trend & rev_up def momentum_base(cost, flag=None, N=10, start=COMBO_START): """flag 為真的候選股排前面（加分）。""" score = momentum if flag is not None: f = flag.reindex(index=close.index, columns=close.columns, fill_value=False).astype(bool) score = momentum + FinlabDataFrame(f).astype(float) * 1000 position = score[candidates].is_largest(N).loc[start:] return run(position, cost, resample="M", resample_offset="11D", stop_loss=0.1) def pre_review_flag(adds_fn, R, reviews): """換股日前 20 個交易日內、預期會被納入的股票。""" rows = [] for i in reviews: r = REVIEWS[i] rows += [(s, t, t) for t in range(r["D"] - 20, r["D"]) for s in adds_fn(i, min(t, r["C"]), R)] return windows(rows) flag_c1 = lambda R: pre_review_flag(tw50_adds, R, ALL) flag_c2 = lambda W: windows([(s, REVIEWS[i]["D"], REVIEWS[i]["D"] + W - 1) for i in ALL for s in tw50_drops(i, 61)]) flag_c3 = lambda R: pre_review_flag(mid100_adds, R, ALL) flag_c4 = lambda R: pre_review_flag(dividend_adds, R, DIV_REVIEWS) def combo(flag_fn, start): flag = functools.lru_cache(maxsize=None)(flag_fn) # 同一個參數的訊號表給 3 個持股數共用，這組驗證完就釋放 return lambda a, N, cost: momentum_base(cost, flag(a), N, start) # --------------------------------------------------------------------------- # 6. 三項驗證 # --------------------------------------------------------------------------- N_DAYS = ("N", [10, 15, 20], 15) SINGLES = [ ("tw50_add", "台灣 50 預期納入，換股日前買進", tw50_add, N_DAYS, ("R", [35, 40, 45], 40)), ("tw50_drop", "台灣 50 預期剔除，換股日收盤買進", tw50_drop, ("R", [56, 61, 66], 61), ("H", [20, 60, 120], 60)), ("mid100_add", "中型 100 預期納入，換股日前買進", mid100_add, N_DAYS, ("R", [120, 130, 140], 130)), ("dividend_add", "高股息（0056）預期納入，換股前買進", dividend_add, N_DAYS, ("R", [30, 35, 40], 35)), ("etf_holders", "全體 ETF 受益人數暴增時 0050 換現金", etf_holders, K_GRID, Q_GRID), ] N_GRID = ("N", [8, 10, 12], 10) COMBOS = [ # 第二個參數都是持股數 N_GRID ("tw50_add_combo", "月營收動能＋預期納入台灣 50 加分", flag_c1, ("R", [35, 40, 45], 40), COMBO_START), ("tw50_drop_combo", "月營收動能＋剛被剔出台灣 50 加分", flag_c2, ("W", [20, 40, 60], 40), COMBO_START), ("mid100_add_combo", "月營收動能＋預期納入中型 100 加分", flag_c3, ("R", [120, 130, 140], 130), COMBO_START), ("dividend_add_combo", "月營收動能＋預期納入高股息加分", flag_c4, ("R", [30, 35, 40], 35), DIVIDEND_START), ] def plateau(wins, n): """參數高原：7 成以上（9 組裡至少 7 組）。""" return wins >= math.ceil(0.7 * n) def validate(key, name, fn, pa, pb): (na, va, oa), (nb, vb, ob) = pa, pb grid = [] for a in va: for b in vb: r = fn(a, b, "default") v = versus_0050(r.creturn) grid.append({na: a, nb: b, "start": v["strategy"]["start"], "end": v["strategy"]["end"], "sharpe": v["strategy"]["sharpe"], "cagr": v["strategy"]["cagr"], "mdd": v["strategy"]["mdd"], "bench_sharpe": v["bench"]["sharpe"], "win": v["win"]}) if a == oa and b == ob: base_report, base_v = r, v zero = versus_0050(fn(oa, ob, "zero").creturn) wins = sum(g["win"] for g in grid) hv = halves(base_report.creturn) out = { "key": key, "name": name, "params": {na: va, nb: vb}, "original": {na: oa, nb: ob}, "grid": grid, "grid_wins": wins, "grid_n": len(grid), "halves": hv, "base_default": base_v, "base_zero_cost": zero, "trade_stats": trade_stats(base_report), "pass": {"grid": plateau(wins, len(grid)), "halves": all(h["win"] for h in hv), "cost": base_v["win"]}, } out["pass"]["all"] = all(out["pass"].values()) print(f"{key}: grid {wins}/{len(grid)}, halves {[h['win'] for h in hv]}, " f"sharpe {base_v['strategy']['sharpe']:.2f} vs 0050 {base_v['bench']['sharpe']:.2f}", flush=True) return out, base_report def trade_keys(report): t = report.get_trades() return set(zip(t["stock_id"], pd.to_datetime(t["entry_date"]))) def against_base(out, base_curves, start, report, base_report): """組合版跟同期、同持股數的底策略比。""" for g in out["grid"]: g["base_sharpe"] = stats(base_curves[start, g["N"]].loc[g["start"]:g["end"]])["sharpe"] g["beats_base"] = g["sharpe"] > g["base_sharpe"] out["grid_beats_base"] = sum(g["beats_base"] for g in out["grid"]) base_curve = base_curves[start, N_GRID[2]] for h in out["halves"]: h["base"] = stats(base_curve.loc[h["strategy"]["start"]:h["strategy"]["end"]]) s = out["base_default"]["strategy"] b = stats(base_curve.loc[s["start"]:s["end"]]) out["momentum_base_same_window"] = b combo_trades, base_trades = trade_keys(report), trade_keys(base_report) out["trades_not_in_base"], out["base_trades_dropped"] = len(combo_trades - base_trades), len(base_trades - combo_trades) out["pass"]["beats_base"] = s["sharpe"] > b["sharpe"] and plateau(out["grid_beats_base"], out["grid_n"]) out["pass"]["all"] &= out["pass"]["beats_base"] print(f" {out['key']}: combo {s['sharpe']:.2f} vs base {b['sharpe']:.2f}, " f"grid beats base {out['grid_beats_base']}/{out['grid_n']}", flush=True) # --------------------------------------------------------------------------- # 7. 抽查與事件研究：公開的台灣 50 逐季變動（維基百科「臺灣50指數」），以及 2026-10-30 的 ETF 申購買回清單 # --------------------------------------------------------------------------- TW50_CHANGES = """ 2013-06 +2884,4938 -2347,3231 2013-09 +2887 -2353 2013-12 +2227,9904 -1722,3673 2014-03 +3474 -3697 2014-06 +2395 -2201 2014-12 +2408 -2324 2015-09 +1476 -2498 2016-06 +2324 -2227 2016-12 +2823 -3474 2017-06 +2633 -1476 2018-03 +5871 -2207 2018-06 +2327,2492 -2324,2325 2018-12 +5876 -2492 2019-03 +2207 -2354 2019-06 +9910 -3481 2019-09 +2888 -2409 2020-06 +6669 -9904 2020-09 +2379,6415 -2823,2888 2020-12 +3034 -2301 2021-03 +1590,8046 -2883,2890 2021-06 +2409,2603,2609,2615 -2105,2474,2633,6669 2021-09 +8454 -1102 2021-12 +3037 -1402 2022-03 +2883 -4938 2022-06 +6770 -8454 2022-09 +2890 -6770 2022-12 +1402 -2409 2023-03 +1605 -8046 2023-06 +4938 -6415 2023-09 +2301,2345,3231,6669 -1402,1605,2609,2615 2024-03 +3661 -9910 2024-06 +3017 -2801 2024-09 +6446 -2408 2024-12 +2609 -1590 2025-03 +2615 -1326 2025-06 +2383 -3037 2025-09 +2059,6919 -1101,6446 2025-12 +2408,3665,3653,2360 -2609,4938,5871,5876 2026-03 +2368,2449,2344,3037,7769 -3665,3034,2379,2912,2615 2026-06 +3443,3665,4958,8046 -1301,2002,2207,6919 2026-09 +6446 -3661 """ TW50_NOW = frozenset(""" 1216 1303 2059 2301 2303 2308 2317 2327 2330 2344 2345 2357 2360 2368 2382 2383 2395 2408 2412 2449 2454 2603 2880 2881 2882 2883 2884 2885 2886 2887 2890 2891 2892 3008 3017 3037 3045 3231 3443 3653 3665 3711 4904 4958 5880 6446 6505 6669 7769 8046""".split()) MID100_NOW = frozenset(""" 1101 1102 1229 1301 1326 1402 1476 1503 1504 1513 1519 1560 1590 1605 1717 1802 2002 2027 2049 2105 2207 2313 2324 2337 2347 2353 2354 2356 2376 2377 2379 2385 2404 2409 2451 2455 2467 2474 2492 2542 2609 2610 2615 2618 2633 2645 2646 2801 2812 2834 2838 2912 3005 3006 3023 3026 3030 3034 3036 3044 3167 3189 3481 3532 3533 3661 3702 3706 4938 5269 5434 5871 5876 6139 6196 6213 6239 6257 6285 6409 6415 6442 6515 6526 6531 6691 6770 6781 6789 6805 6919 6949 7610 7750 8210 8464 8996 9904 9910 9945""".split()) DIVIDEND_NOW = frozenset(""" 1102 1216 1303 1513 2027 2105 2301 2303 2317 2324 2344 2347 2353 2356 2357 2376 2377 2379 2382 2385 2408 2449 2454 2474 2603 2618 2645 2880 2883 2884 2885 2886 2887 2890 2891 3005 3023 3034 3036 3044 3045 3231 3702 3711 4904 4938 5876 6239 6285 9904""".split()) def parse_changes(): out = collections.defaultdict(lambda: (frozenset(), frozenset())) # 沒有變動的季度：沒有納入、剔除 for line in TW50_CHANGES.split("\n"): if line: month, adds, drops = line.split() out[month] = (frozenset(adds[1:].split(",")), frozenset(drops[1:].split(","))) return out def as_of(month, members): """2018-06 以前的名單把 3711 日月光投控換回 2311 日月光。""" return members if month >= "2018-06" else (members - {"3711"}) | {"2311"} def spot_check(): changes = parse_changes() actual, rows = TW50_NOW, [] for r in reversed(REVIEWS): # 從今天倒推每次審核後的實際名單 adds, drops = changes[r["month"]] after = as_of(r["month"], actual) actual = (actual - adds) | drops before = as_of(r["month"], actual) rule_adds = buffer_review(before, rank_at(r["C"]), 50, 40, 61) - before # 用實際名單套規則 rows.append({ "month": r["month"], "actual_adds": sorted(adds), "actual_drops": sorted(drops), "rebuilt_adds": sorted(r["tw_after"] - r["tw_before"]), "rebuilt_drops": sorted(r["tw_before"] - r["tw_after"]), "rule_on_actual_adds": sorted(rule_adds), "overlap_after": len(after & r["tw_after"]), }) rows.reverse() last = REVIEWS[-1] return { "tw50": { "reviews": rows, "actual_adds": sum(len(x["actual_adds"]) for x in rows), "rebuilt_adds": sum(len(x["rebuilt_adds"]) for x in rows), "rebuilt_hit": sum(len(set(x["actual_adds"]) & set(x["rebuilt_adds"])) for x in rows), "rule_on_actual_adds": sum(len(x["rule_on_actual_adds"]) for x in rows), "rule_on_actual_hit": sum(len(set(x["actual_adds"]) & set(x["rule_on_actual_adds"])) for x in rows), "overlap_after_mean": round(float(np.mean([x["overlap_after"] for x in rows])), 2), "overlap_after_min": int(min(x["overlap_after"] for x in rows)), "reviews_exact": int(sum(x["overlap_after"] == 50 for x in rows)), }, "now": { # 最近一次審核後的重建名單 vs 2026-10-30 申購買回清單 "review": last["month"], "tw50_overlap": len(last["tw_after"] & TW50_NOW), "mid100_overlap": len(last["mid_after"] & MID100_NOW), "dividend_overlap": len(last["div_after"] & DIVIDEND_NOW), "mid100_missing": sorted(MID100_NOW - last["mid_after"]), "mid100_extra": sorted(last["mid_after"] - MID100_NOW), "dividend_missing": sorted(DIVIDEND_NOW - last["div_after"]), "dividend_extra": sorted(last["div_after"] - DIVIDEND_NOW), }, } EVENT_WINDOWS = {"C_A": ("C", "A"), "A_D": ("A", "D"), "C_D": ("C", "D"), "D_D20": ("D", "D20"), "D_D60": ("D", "D60")} def event_study(): """台灣 50 實際納入、剔除股相對 0050 的超額報酬：截止日→公告、公告→換股、截止日→換股、換股→+20／+60 日。""" changes = parse_changes() etf = adj_close["0050"] out = {} for side, pick in (("adds", 0), ("drops", 1)): rows = [] for r in REVIEWS: marks = {"C": r["C"], "A": r["A"], "D": r["D"], "D20": r["D"] + 20, "D60": r["D"] + 60} for s in changes[r["month"]][pick]: if s not in adj_close: continue price, row = adj_close[s], {"month": r["month"], "stock": s} for name, (a, b) in EVENT_WINDOWS.items(): lo, hi = marks[a], marks[b] if hi >= len(days): continue excess = price.iloc[hi] / price.iloc[lo] - etf.iloc[hi] / etf.iloc[lo] if not np.isnan(excess): row[name] = float(excess) rows.append(row) df = pd.DataFrame(rows) out[side] = { "events": rows, "summary": {w: {"n": int(df[w].notna().sum()), "mean": round(float(df[w].mean()), 6), "median": round(float(df[w].median()), 6), "positive": round(float((df[w].dropna() > 0).mean()), 4)} for w in EVENT_WINDOWS}, } return out # MSCI 台灣（標準）指數兩次季度調整（財經媒體報導的名單），排名用公告前一個月月底、上市櫃合併的全市值 MSCI_CHANGES = { "2026-05": ("2026-04-30", ["6223"], ["1102", "1402", "1504", "2324", "2474", "2610", "2633"]), "2026-08": ("2026-07-31", ["2344", "2408", "3189", "6274", "8046", "8299"], ["1101", "2609", "2834", "3293", "3529", "6919"]), } common = ordinary_shares("sii", "otc") def msci_ranks(): """兩次調整的納入、剔除股，在兩個排名日各排第幾。""" out = {} for review, (date, _, _) in MSCI_CHANGES.items(): rank = cap_all.iloc[last_trading_day(pd.Timestamp(date))][common].dropna().rank(ascending=False) out[review] = {"date": date} for when, (_, adds, drops) in MSCI_CHANGES.items(): out[review][f"{when}_adds"] = {s: int(rank[s]) for s in adds} out[review][f"{when}_drops"] = {s: int(rank[s]) for s in drops} return out # 00919（特選臺灣上市上櫃精選高息指數）2026-10-30 持股；5 月審核在 5/20 股利宣告期限後 FUND_00919_NOW = frozenset(""" 1210 1319 1402 1477 2105 2211 2347 2357 2379 2382 2385 2458 2474 2504 2542 2597 2603 2606 2618 2637 2855 2881 2882 2883 2887 2891 3034 3293 4763 4915 4938 5522 5871 6121 6670 6890 8112 8454 9910 9941""".split()) def etf_00919_overlap(): """上市櫃市值前 300、近 3 個月日均成交金額 > 8,000 萬、近 4 季 ROE 合計 > 0，殖利率前 40 檔。""" p = last_trading_day(pd.Timestamp("2026-05-20")) top300 = cap_all.iloc[p][common].dropna().nlargest(300).index roe = pd.DataFrame(data.get("fundamental_features:ROE稅後").index_str_to_date()).rolling(4).sum() roe = roe.reindex(days, method="ffill").iloc[p].reindex(top300) value = traded_value[top300].iloc[p - 62:p + 1].mean() y = pe_yield_all.iloc[p].reindex(top300) picks = y[(value > 8e7) & (roe > 0)].nlargest(40).index return {"date": str(days[p].date()), "picks": sorted(picks), "overlap": len(set(picks) & FUND_00919_NOW)} def main(): metrics = {"generated": datetime.date.today().isoformat(), "data_end": str(days[-1].date()), "timing_start": TIMING_START, "singles": [], "combos": [], "base": {}} curves = {"0050": bench} for key, name, fn, pa, pb in SINGLES: out, report = validate(key, name, fn, pa, pb) metrics["singles"].append(out) curves[key] = active(report.creturn) base_curves, base_reports = {}, {} for start in (COMBO_START, DIVIDEND_START): reports = {N: momentum_base("default", N=N, start=start) for N in N_GRID[1]} base_curves.update({(start, N): active(r.creturn) for N, r in reports.items()}) r = base_reports[start] = reports[N_GRID[2]] metrics["base"][start] = {"default": versus_0050(r.creturn), "halves": halves(r.creturn), "zero_cost": versus_0050(momentum_base("zero", N=N_GRID[2], start=start).creturn), "trade_stats": trade_stats(r)} curves[f"momentum_base_{start[:4]}"] = base_curves[start, N_GRID[2]] for key, name, flag_fn, pa, start in COMBOS: out, report = validate(key, name, combo(flag_fn, start), pa, N_GRID) against_base(out, base_curves, start, report, base_reports[start]) metrics["combos"].append(out) curves[key] = active(report.creturn) out, report = validate("etf_holders_combo", "受益人數不暴增持有 00631L、暴增持有 0050", etf_holders_leveraged, K_GRID, Q_GRID) metrics["combos"].append(out) curves["etf_holders_combo"] = active(report.creturn) lev = adj_close["00631L"].dropna() metrics["leveraged_buy_hold"] = versus_0050(lev.loc[out["base_default"]["strategy"]["start"]:]) curves["00631L"] = lev metrics["spot_check"] = spot_check() metrics["event_study"] = event_study() metrics["msci"] = msci_ranks() metrics["etf_00919"] = etf_00919_overlap() metrics["holders"] = {str(k.date()): int(v) for k, v in holders.items()} state = surge_by_month(K_GRID[2], Q_GRID[2]) metrics["surge_months"] = [str(k.date()) for k, v in state.items() if v == 1] with open("metrics.json", "w") as f: json.dump(metrics, f, ensure_ascii=False, indent=1) pd.DataFrame(curves).sort_index().to_csv("data.csv", float_format="%.6f") if __name__ == "__main__": main()