""" 0050 正2（00631L）研究：均線擇時、槓桿 ETF 投資法、正2＋美債、定期定額與波動耗損 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個策略 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，原始參數在兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 日夏普＝日報酬平均 / 標準差 × √252，不扣無風險利率。 0050 含息＝etl:adj_close 還原價買進持有，不經 sim()、不含成本。 00631L 在 2026-03-25～03-30 因分割停止交易，還原價已處理分割。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入。 輸出：metrics.json、data.csv（月底淨值與定期定額帳戶）、report_strategy.html（50% 正2＋50% 現金）。 """ import datetime import json import math import numpy as np import pandas as pd import finlab from finlab import data from finlab.backtest import sim L2, TW50, BOND = "00631L", "0050", "00679B" START_L2 = "2014-12-31" # 00631L 2014-10-31 上市；年底起算，月／季／年再平衡同一天進場 START_BOND = "2017-12-29" # 00679B 2017-01-17 上市；年再平衡要從年底起算 TRADING_DAYS = 252 adj = pd.DataFrame(data.get("etl:adj_close")[[TW50, L2, BOND]]) adj.index = pd.to_datetime(adj.index) taiex_tr = pd.DataFrame(data.get("benchmark_return:發行量加權股價報酬指數")).iloc[:, 0] taiex_tr.index = pd.to_datetime(taiex_tr.index) days = adj.index # --------------------------------------------------------------------------- # 1. 統計 # --------------------------------------------------------------------------- def perf(curve): curve = curve.dropna() curve = curve / curve.iloc[0] daily = 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(daily.mean() / daily.std() * math.sqrt(TRADING_DAYS)), 6), "vol": round(float(daily.std() * math.sqrt(TRADING_DAYS)), 6), "mdd": round(float((curve / curve.cummax() - 1).min()), 6), "total_return": round(float(curve.iloc[-1] - 1), 6), } def versus_0050(curve, lo=None, hi=None): part = curve.loc[lo:hi] s = perf(part) b = perf(adj[TW50].loc[s["start"]:s["end"]]) return {"strategy": s, "bench": b, "win": bool(s["sharpe"] > b["sharpe"])} def compare_halves(curve): midpoint = curve.index[0] + (curve.index[-1] - curve.index[0]) / 2 mid = curve.index[curve.index.get_indexer([midpoint], method="nearest")[0]] return [versus_0050(curve, None, mid), versus_0050(curve, mid, None)] def next_day(date): """訊號日收盤產生訊號，sim() 在下一個交易日收盤成交。""" return days[days.get_indexer([pd.Timestamp(date)])[0] + 1] def curve_of(report, start): return pd.Series(report.creturn).dropna().loc[next_day(start):] COSTS = {"default": {}, "zero": {"fee_ratio": 0, "tax_ratio": 0}, "etf_tax": {"tax_ratio": 0.001}} # --------------------------------------------------------------------------- # 2. 三個策略 # --------------------------------------------------------------------------- def ma_state(price, n, band): """站上均線 ×(1+band) 才進場，跌破均線 ×(1-band) 才出場，中間維持原狀態。""" ma = price.rolling(n).mean() state = pd.Series(np.nan, index=price.index) state[price > ma * (1 + band)] = 1.0 state[price < ma * (1 - band)] = 0.0 return state.ffill().fillna(0) def ma_timing(n, band, cost="default", hold=L2): """看 0050 還原價與 n 日均線，在線上持有 hold（預設 00631L），否則空手。""" state = ma_state(adj[TW50], n, band).loc[START_L2:] return sim(pd.DataFrame({hold: state}), resample=None, upload=False, **COSTS[cost]) def fixed_mix(w, freq, other=None, cost="default", start=START_L2): """00631L 固定權重 w，其餘是現金（other=None）或另一檔 ETF，每期再平衡回 w。""" position = pd.DataFrame({L2: w}, index=days[days >= start]) if other: position[other] = 1 - w return sim(position, resample=freq, upload=False, **COSTS[cost]) STRATEGIES = { "ma_timing": { "name": "0050 站上均線才持有 00631L", "x": ("ma_days", [60, 120, 240]), "y": ("band", [0.0, 0.01, 0.02]), "center": (120, 0.01), "run": ma_timing, "start": START_L2, }, "l2_cash": { "name": "00631L＋現金固定比例再平衡", "x": ("weight_00631L", [0.4, 0.5, 0.6]), "y": ("rebalance", ["M", "Q", "YE"]), "center": (0.5, "Q"), "run": lambda x, y, cost: fixed_mix(x, y, None, cost), "start": START_L2, }, "l2_bond": { "name": "00631L＋00679B（美債 20 年）固定比例再平衡", "x": ("weight_00631L", [0.4, 0.5, 0.6]), "y": ("rebalance", ["M", "Q", "YE"]), "center": (0.5, "Q"), "run": lambda x, y, cost: fixed_mix(x, y, BOND, cost, START_BOND), "start": START_BOND, }, } results, curves, center_reports = {}, {}, {} for key, spec in STRATEGIES.items(): grid = [] for xv in spec["x"][1]: for yv in spec["y"][1]: report = spec["run"](xv, yv, "default") curve = curve_of(report, spec["start"]) row = versus_0050(curve) row.update({spec["x"][0]: xv, spec["y"][0]: yv}) grid.append(row) if (xv, yv) == spec["center"]: center_reports[key], curves[key] = report, curve print(key, xv, yv, row["strategy"]["sharpe"], row["bench"]["sharpe"], flush=True) curve = curves[key] center_result = versus_0050(curve) halves = compare_halves(curve) costs = {c: versus_0050(curve_of(spec["run"](*spec["center"], c), spec["start"])) for c in ("zero", "etf_tax")} wins = sum(row["win"] for row in grid) results[key] = { "name": spec["name"], "grid": grid, "center": dict(zip((spec["x"][0], spec["y"][0]), spec["center"])), "center_result": center_result, "grid_wins": wins, "halves": halves, "costs": costs, "pass": { "neighbors": wins / len(grid) >= 0.7, "halves": all(h["win"] for h in halves), "cost": center_result["win"], }, } # 00631L 買進持有（也是定期定額的時間加權報酬） for key, start in (("l2_buy_hold", START_L2), ("l2_buy_hold_since_2018", START_BOND)): report = sim(pd.DataFrame({L2: 1.0}, index=days[days >= start]), resample=None, upload=False) curve = curve_of(report, start) results[key] = {"center_result": versus_0050(curve)} if key == "l2_buy_hold": curves[key] = curve results[key]["halves"] = compare_halves(curve) # --------------------------------------------------------------------------- # 3. 拆解：為什麼正2 的夏普贏不了 0050 # --------------------------------------------------------------------------- extra = {} first = next_day(START_L2) r0050 = adj[TW50].ffill().pct_change(fill_method=None) # 0050 2025-06 分割停止交易，沿用前價 # 理想的 2 倍：每天 0050 報酬 ×2，零成本 ideal = (1 + 2 * r0050.loc[first:].iloc[1:]).cumprod() ideal = pd.concat([pd.Series([1.0], index=[first]), ideal]) curves["ideal_2x"] = ideal extra["ideal_2x"] = perf(ideal) # 同一個均線訊號套在 0050 本身：訊號有沒有價值 extra["ma_on_0050"] = versus_0050(curve_of(ma_timing(120, 0.01, hold=TW50), START_L2)) state = ma_state(adj[TW50], 120, 0.01).loc[first:] trades = center_reports["ma_timing"].get_trades() years = (days[-1] - first).days / 365.25 extra["ma_timing_stats"] = { "time_in_market": round(float(state.mean()), 6), "round_trips": int(len(trades)), "round_trips_per_year": round(len(trades) / years, 3), "trade_win_rate": round(float((trades["return"] > 0).mean()), 6), } ma120 = adj[TW50].rolling(120).mean() # 現金部位有 1% 年息時的槓桿 ETF 投資法（日報酬加上一半資金的利息） cash_daily = curves["l2_cash"].pct_change().fillna(0) + 0.5 * 0.01 / TRADING_DAYS extra["l2_cash_with_1pct_interest"] = versus_0050((1 + cash_daily).cumprod()) # 逐年：0050、00631L、0050 每日 2 倍、加權報酬指數每日 2 倍 r = pd.DataFrame({ "0050": r0050, "00631L": adj[L2].ffill().pct_change(fill_method=None), "taiex_tr": taiex_tr.reindex(days).pct_change(fill_method=None), "00679B": adj[BOND].pct_change(fill_method=None), }).loc["2014-11-01":].fillna(0) r["2x_0050"] = 2 * r["0050"] r["2x_taiex_tr"] = 2 * r["taiex_tr"] growth = (1 + r).cumprod().resample("YE").last() yearly = growth / growth.shift(1).fillna(1) - 1 extra["yearly"] = {str(d.year): row.round(6).to_dict() for d, row in yearly.iterrows()} extra["yearly"] = {y: {k: v for k, v in row.items() if k != "00679B" or int(y) >= 2018} for y, row in extra["yearly"].items()} daily = r.loc[adj[L2].first_valid_index():] extra["daily_tracking"] = { k: { "beta": round(float(np.polyfit(daily[k], daily["00631L"], 1)[0]), 6), "median_abs_gap_vs_2x": round(float((daily["00631L"] - 2 * daily[k]).abs().median()), 6), } for k in ("0050", "taiex_tr") } # 00631L 與 00679B 日報酬的逐年相關係數 rr = adj[[L2, BOND]].pct_change(fill_method=None).loc["2017-02-01":].dropna() extra["stock_bond_corr_by_year"] = {str(y): round(float(g[L2].corr(g[BOND])), 4) for y, g in rr.groupby(rr.index.year)} def drawdown_episode(curve): dd = curve / curve.cummax() - 1 trough = dd.idxmin() peak = curve.loc[:trough].idxmax() back = curve.loc[trough:][curve.loc[trough:] >= curve.loc[peak]] return {"mdd": round(float(dd.min()), 6), "peak": str(peak.date()), "trough": str(trough.date()), "recovered": str(back.index[0].date()) if len(back) else None} # 樣本外的提醒：00631L 上市前的 2008 年，用 0050 每日 2 倍零成本模擬 r_old = r0050.loc["2007-01-01":"2014-10-31"].dropna() extra["simulated_2008"] = { "0050": drawdown_episode((1 + r_old).cumprod()), "2x_0050_zero_cost": drawdown_episode((1 + 2 * r_old).cumprod()), "year_2008_0050": round(float((1 + r_old.loc["2008"]).prod() - 1), 6), "year_2008_2x": round(float((1 + 2 * r_old.loc["2008"]).prod() - 1), 6), } # 均線擇時最大回撤那一段，以及 0050 在高點與低點離半年線多遠 episode = drawdown_episode(curves["ma_timing"]) episode["0050_vs_ma120_at_peak"] = round(float(adj[TW50][episode["peak"]] / ma120[episode["peak"]] - 1), 6) episode["0050_vs_ma120_at_trough"] = round(float(adj[TW50][episode["trough"]] / ma120[episode["trough"]] - 1), 6) extra["ma_timing_stats"]["max_drawdown_episode"] = episode # 00631L 跌超過 35% 的每一段（從創新高到收復），和同一段 0050 的最大回撤 l2_price, tw_price = adj[L2].ffill().dropna(), adj[TW50].ffill() underwater = l2_price < l2_price.cummax() episodes = [] for _, part in l2_price.groupby((~underwater).cumsum()): dd = part / part.iloc[0] - 1 if dd.min() < -0.35: after = l2_price.index[l2_price.index > part.index[-1]] recovered = after[0] if len(after) else None span = tw_price.loc[part.index[0]:recovered] episodes.append({ "peak": str(part.index[0].date()), "trough": str(dd.idxmin().date()), "recovered": str(recovered.date()) if recovered is not None else None, "mdd_00631L": round(float(dd.min()), 6), "mdd_0050_same_span": round(float((span / span.cummax() - 1).min()), 6), }) extra["drawdowns_over_35pct"] = episodes # 分割：還原價 ÷ 收盤價 在分割前後的比值 raw = pd.DataFrame(data.get("price:收盤價")[[TW50, L2]]) raw.index = pd.to_datetime(raw.index) factor = (adj[[TW50, L2]] / raw).dropna() jumps = factor / factor.shift(1) extra["splits"] = {s: {str(d.date()): round(float(v), 4) for d, v in jumps[s][jumps[s] > 1.5].items()} for s in (TW50, L2)} # --------------------------------------------------------------------------- # 4. 波動耗損：持有不同長度，00631L 和「0050 報酬 ×2」差多少 # --------------------------------------------------------------------------- log_growth = np.log1p(r[["0050", "00631L", "2x_0050"]].loc[adj[L2].first_valid_index():]).cumsum() holding = {} for label, h in {"1m": 21, "3m": 63, "1y": 252, "3y": 756, "5y": 1260}.items(): R = np.expm1((log_growth.shift(-h) - log_growth).dropna()) holding[label] = { "windows": int(len(R)), "median_00631L": round(float(R["00631L"].median()), 6), "median_two_times_0050": round(float((2 * R["0050"]).median()), 6), "median_compounding_gap": round(float((R["2x_0050"] - 2 * R["0050"]).median()), 6), "median_tracking_gap": round(float((R["00631L"] - R["2x_0050"]).median()), 6), "share_below_two_times_0050": round(float((R["00631L"] < 2 * R["0050"]).mean()), 6), "share_below_0050": round(float((R["00631L"] < R["0050"]).mean()), 6), "share_loss_00631L": round(float((R["00631L"] < 0).mean()), 6), "share_loss_0050": round(float((R["0050"] < 0).mean()), 6), "p10_00631L": round(float(R["00631L"].quantile(0.1)), 6), "p10_0050": round(float(R["0050"].quantile(0.1)), 6), } if label == "1y": below = R["00631L"] < R["0050"] holding[label]["share_of_below_0050_when_0050_fell"] = round(float((R.loc[below, "0050"] < 0).mean()), 6) month_starts = R.groupby(R.index.to_period("M")).head(1) above = month_starts["00631L"] >= 2 * month_starts["0050"] buckets = pd.cut(month_starts["0050"], [-1, 0, 0.4, 0.8, 10], labels=["<0", "0-40%", "40-80%", ">80%"]) holding["1y_month_start_above_two_times_by_0050_return"] = { str(b): {"points": int(len(g)), "above": int(above[g.index].sum())} for b, g in month_starts.groupby(buckets, observed=True) } red = month_starts.index[~above] holding["1y_month_start_below_two_times"] = { "points": int(len(red)), "started_2023_03_to_2025_10": int(((red >= "2023-03-01") & (red <= "2025-10-31")).sum()), } holding["1y_by_month"] = [ {"start": str(d.date()), "0050": round(float(v["0050"]), 4), "00631L": round(float(v["00631L"]), 4)} for d, v in month_starts.iterrows() ] extra["holding_periods"] = holding # --------------------------------------------------------------------------- # 5. 定期定額：每月 6 日（遇假日順延）收盤各買 1 萬元，手續費 0.1425%，不賣出 # --------------------------------------------------------------------------- FEE = 0.001425 buy_days = pd.DatetimeIndex(sorted({ days[days >= pd.Timestamp(y, m, 6)][0] for y in range(2014, days[-1].year + 1) for m in range(1, 13) if (days >= pd.Timestamp(y, m, 6)).any() })) buy_days = buy_days[buy_days >= pd.Timestamp("2014-11-06")] price = pd.DataFrame({"0050": adj[TW50].ffill(), "00631L": adj[L2].ffill()}) def dca(symbol, start, months, end=None): """回傳帳戶市值、累計投入、內部報酬率（年化）。期末預設＝最後一筆扣款後一個月。""" dates = buy_days[buy_days >= start][:months] if len(dates) < months: return None end = end or days[days <= dates[-1] + pd.DateOffset(months=1)][-1] p = price[symbol].loc[dates[0]:end] units = pd.Series(0.0, index=p.index) units[dates] = 10000 * (1 - FEE) / p[dates] invested = pd.Series(0.0, index=p.index) invested[dates] = 10000.0 value, invested = units.cumsum() * p, invested.cumsum() t = np.array([(d - dates[0]).days / 365.25 for d in list(dates) + [end]]) cash = np.array([-10000.0] * len(dates) + [value.iloc[-1]]) lo, hi = -0.99, 5.0 for _ in range(200): # 二分法解 IRR mid = (lo + hi) / 2 lo, hi = (mid, hi) if (cash / (1 + mid) ** t).sum() > 0 else (lo, mid) return value, invested, mid dca_out = {} for years_held in (3, 5, 10): rows = [] for start in buy_days: a, b = dca("00631L", start, 12 * years_held), dca("0050", start, 12 * years_held) if a is None or b is None: break rows.append({ "start": start.strftime("%Y-%m"), "irr_00631L": a[2], "irr_0050": b[2], "multiple_00631L": a[0].iloc[-1] / a[1].iloc[-1], "multiple_0050": b[0].iloc[-1] / b[1].iloc[-1], "worst_vs_invested_00631L": (a[0] / a[1]).min() - 1, "worst_vs_invested_0050": (b[0] / b[1]).min() - 1, }) df = pd.DataFrame(rows) dca_out[f"{years_held}y"] = { "starts": len(df), "first_start": df["start"].iloc[0], "last_start": df["start"].iloc[-1], "share_00631L_ends_higher": round(float((df["multiple_00631L"] > df["multiple_0050"]).mean()), 6), "median_irr_00631L": round(float(df["irr_00631L"].median()), 6), "median_irr_0050": round(float(df["irr_0050"].median()), 6), "min_irr_00631L": round(float(df["irr_00631L"].min()), 6), "min_irr_0050": round(float(df["irr_0050"].min()), 6), "median_multiple_00631L": round(float(df["multiple_00631L"].median()), 6), "median_multiple_0050": round(float(df["multiple_0050"].median()), 6), "worst_vs_invested_00631L": round(float(df["worst_vs_invested_00631L"].min()), 6), "worst_vs_invested_0050": round(float(df["worst_vs_invested_0050"].min()), 6), "worst_vs_invested_00631L_start": df.loc[df["worst_vs_invested_00631L"].idxmin(), "start"], } if years_held == 3: dca_out["3y_by_start"] = df[["start", "irr_00631L", "irr_0050"]].round(4).to_dict("records") # 全期：第一筆到最近一次扣款，市值算到資料最後一天 months = len(buy_days) (v_l2, inv, irr_l2), (v_tw, _, irr_tw) = (dca(s, buy_days[0], months, days[-1]) for s in ("00631L", "0050")) dca_out["full"] = { "first_buy": str(buy_days[0].date()), "months": months, "invested": float(inv.iloc[-1]), "value_00631L": round(float(v_l2.iloc[-1])), "value_0050": round(float(v_tw.iloc[-1])), "irr_00631L": round(irr_l2, 6), "irr_0050": round(irr_tw, 6), "worst_vs_invested_00631L": round(float((v_l2 / inv).min() - 1), 6), "worst_vs_invested_0050": round(float((v_tw / inv).min() - 1), 6), } extra["dca"] = dca_out # --------------------------------------------------------------------------- # 6. 輸出 # --------------------------------------------------------------------------- center_reports["l2_cash"].to_html("report_strategy.html") meta = { "run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "data_end": str(days[-1].date()), "benchmark": "0050 含息買進持有（etl:adj_close，不含交易成本）", "sharpe": "日報酬平均 / 標準差 × √252，未扣無風險利率", "report_strategy": "l2_cash", } json.dump({"meta": meta, "strategies": results, "analysis": extra}, open("metrics.json", "w"), ensure_ascii=False, indent=1, default=str) # 每月最後一個交易日，加上兩個起算日（淨值從起算日的 1 開始） month_end = days[np.append(days.to_period("M")[1:] != days.to_period("M")[:-1], True)] rows = month_end[month_end >= "2014-11-01"].union(pd.DatetimeIndex([first, next_day(START_BOND)])) table = pd.DataFrame({k: curves[k] for k in ("ma_timing", "l2_cash", "l2_bond", "l2_buy_hold", "ideal_2x")}) table["0050"] = adj[TW50].loc[first:] table = table.reindex(rows) for col in table: table[col] = table[col] / table[col].dropna().iloc[0] table["0050_since_2018"] = adj[TW50].reindex(rows).loc[next_day(START_BOND):] / adj[TW50][next_day(START_BOND)] dca_table = pd.DataFrame({"dca_invested": inv, "dca_00631L": v_l2, "dca_0050": v_tw}).reindex(rows) pd.concat([table.round(4), dca_table.round(0)], axis=1).to_csv("data.csv", index_label="date") print("done", meta)