"""0050 搭配台灣掛牌的美債、公司債、黃金與反向 ETF：6 個設計、63 組參數。 四篇研究文共用這份程式： /blog/0050-us-treasury-etf-allocation-backtest ① 股債固定比例、② 股債輪動 /blog/00720b-corporate-bond-etf-backtest ③ 0050＋00720B 投資等級公司債 /blog/gold-etf-00635u-0050-backtest ④ 0050＋00635U 黃金、⑤ 股債金三分 /blog/00632r-inverse-etf-hedge-backtest ⑥ 跌破均線用 00632R 對沖 全部參數在看績效以前寫定，沒有依結果挑選。 - 價格：etl:adj_close 還原收盤價，先 ffill（0050 2025-06 分割停牌、00632R 2024-12 停牌）再算報酬與均線。 - 成交：訊號用當天收盤可得的資料，sim() 下一個交易日收盤成交，預設手續費與證交稅。 - 期間：從該設計所用 ETF 最晚上市的那天開始，所有參數組同一天建倉，到 DATA_END。 - 基準：0050 含息價直接連乘，不經 sim()、不扣成本。 - 「贏」：日夏普（日報酬平均 ÷ 標準差 × √252，不扣無風險利率）高於同期基準。 執行（finlab 需要資料時會自動引導登入）： pip install finlab python research.py # 全部設計，輸出到目前目錄的 out/ python research.py --html # 另存原參數的互動回測報告 """ import argparse import json 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-11" BENCH = "0050" ETFS = [BENCH, "00679B", "00687B", "00720B", "00635U", "00632R"] # ---------- 資料 ---------- def load_prices(): adj = data.get("etl:adj_close").loc[:DATA_END] return adj[ETFS].ffill() # ---------- 統計 ---------- def sharpe(returns): returns = returns.dropna() return float(returns.mean() / returns.std() * np.sqrt(252)) def metrics(returns): returns = returns.dropna() equity = (1 + returns).cumprod() days = (returns.index[-1] - returns.index[0]).days return { "cagr": float(equity.iloc[-1] ** (365.25 / days) - 1), "sharpe": sharpe(returns), "mdd": float((equity / equity.cummax() - 1).min()), "volatility": float(returns.std() * np.sqrt(252)), } def equity(returns, start_day): """淨值從建倉日（第一筆報酬的前一個交易日）的 1 開始。""" curve = (1 + returns).cumprod() return pd.concat([pd.Series([1.0], index=[pd.Timestamp(start_day)]), curve]) def yearly(returns): by_year = (1 + returns).groupby(returns.index.year).prod() - 1 return {int(y): float(v) for y, v in by_year.items()} # ---------- 部位 ---------- def rebalance_dates(index, freq): """第一天加上每期最後一個交易日；freq=None 表示部位一變就調整。""" if freq is None: return None days = pd.Series(index, index=index) periods = index.to_period({"W": "W-FRI", "M": "M", "Q": "Q", "YE": "Y"}[freq]) period_ends = days.groupby(periods).last() return pd.DatetimeIndex(sorted({index[0], *period_ends})) def fixed_weights(prices, start, weights): index = prices.loc[start:].index return pd.DataFrame({ticker: float(w) for ticker, w in weights.items()}, index=index) def below_average(prices, start, days): """0050 含息價收盤低於 N 日均線。""" close = prices[BENCH] return (close < close.rolling(days).mean()).loc[start:] def bond_rotation(prices, start, days): below = below_average(prices, start, days) return pd.DataFrame({BENCH: (~below).astype(float), "00679B": below.astype(float)}) def inverse_hedge(prices, start, days, hedge): """跌破均線時 00632R 金額是 0050 的 hedge 倍。""" below = below_average(prices, start, days) return pd.DataFrame({BENCH: np.where(below, 1 / (1 + hedge), 1.0), "00632R": np.where(below, hedge / (1 + hedge), 0.0)}, index=below.index) # 每個設計：起始日、參數格、原參數、鄰近參數至少要贏幾組、怎麼由參數組出部位與調整頻率 DESIGNS = { "d1": dict(name="0050＋美債固定比例", start="2017-04-13", min_wins=13, grid=list(product([0.5, 0.6, 0.7], ["M", "Q", "YE"], ["00679B", "00687B"])), original=(0.6, "Q", "00679B"), build=lambda p, s, x: (fixed_weights(p, s, {BENCH: x[0], x[2]: 1 - x[0]}), x[1])), "d2": dict(name="股債輪動", start="2017-01-17", min_wins=7, grid=list(product([20, 60, 120], [None, "W", "M"])), original=(60, None), average_control=True, build=lambda p, s, x: (bond_rotation(p, s, x[0]), x[1])), "d3": dict(name="0050＋00720B 公司債", start="2018-02-01", min_wins=7, grid=list(product([0.5, 0.6, 0.7], ["M", "Q", "YE"])), original=(0.6, "Q"), build=lambda p, s, x: (fixed_weights(p, s, {BENCH: x[0], "00720B": 1 - x[0]}), x[1])), "d4": dict(name="0050＋00635U 黃金", start="2015-04-15", min_wins=7, grid=list(product([0.1, 0.2, 0.3], ["M", "Q", "YE"])), original=(0.2, "Q"), build=lambda p, s, x: (fixed_weights(p, s, {BENCH: 1 - x[0], "00635U": x[0]}), x[1])), "d5": dict(name="股債金三分", start="2017-01-17", min_wins=7, grid=list(product([0.25, 1 / 3, 0.4], ["M", "Q", "YE"])), original=(1 / 3, "Q"), build=lambda p, s, x: (fixed_weights(p, s, {BENCH: x[0], "00679B": (1 - x[0]) / 2, "00635U": (1 - x[0]) / 2}), x[1])), "d6": dict(name="00632R 跌破均線對沖", start="2014-10-31", min_wins=7, grid=list(product([20, 60, 120], [0.25, 0.5, 1.0])), original=(60, 0.5), average_control=True, build=lambda p, s, x: (inverse_hedge(p, s, x[0], x[1]), None)), } # ---------- 回測 ---------- def backtest(position, freq, **costs): report = sim(position, resample=rebalance_dates(position.index, freq), upload=False, **costs) returns = report.creturn.loc[position.index[0]:DATA_END].pct_change().iloc[1:] return report, returns def drift_turnover(prices, position, freq): """每次調整時，持有比重偏離目標的單邊比例；回傳每年平均。""" if freq: dates = rebalance_dates(position.index, freq) else: # sim(resample=None) 只在目標比重改變的那天調整 dates = position.index[position.ne(position.shift()).any(axis=1)] returns = prices[position.columns].pct_change().loc[position.index].fillna(0) held = position.iloc[0].copy() traded = 0.0 for day in position.index[1:]: held = held * (1 + returns.loc[day]) held = held / held.sum() if day in dates: target = position.loc[day] traded += float((target - held).abs().sum() / 2) held = target.copy() years = (position.index[-1] - position.index[0]).days / 365.25 return traded / years def evaluate(prices, key): """回傳 metrics.json 的設計段落、原參數（與對照組）的淨值、原參數的回測報告。""" cfg = DESIGNS[key] start = cfg["start"] bench_returns = prices[BENCH].pct_change() result = {"name": cfg["name"], "start": start, "end": DATA_END, "original": list(cfg["original"]), "grid": []} for params in cfg["grid"]: position, freq = cfg["build"](prices, start, params) report, returns = backtest(position, freq) row = metrics(returns) row["params"] = list(params) row["benchmark_sharpe"] = sharpe(bench_returns.reindex(returns.index)) row["wins"] = row["sharpe"] > row["benchmark_sharpe"] if params == cfg["original"]: row["trades"] = len(report.get_trades()) row["one_way_turnover_per_year"] = drift_turnover(prices, position, freq) row["average_weights"] = {c: float(w) for c, w in position.mean().items()} original, original_position, original_freq, original_report = returns, position, freq, report result["grid"].append(row) bench = bench_returns.reindex(original.index) result["strategy"] = metrics(original) result["benchmark"] = metrics(bench) half = len(original) // 2 result["halves"] = [] for part in (original.iloc[:half], original.iloc[half:]): part_strategy, part_bench = metrics(part), metrics(bench.reindex(part.index)) result["halves"].append({"from": str(part.index[0].date()), "to": str(part.index[-1].date()), "strategy": part_strategy, "benchmark": part_bench, "wins": part_strategy["sharpe"] > part_bench["sharpe"]}) result["neighbor_wins"] = sum(row["wins"] for row in result["grid"]) result["neighbor_total"] = len(result["grid"]) result["yearly"] = yearly(original) result["benchmark_yearly"] = yearly(bench) # 逐一拿掉每個曆年後的日夏普是描述用；過關只看拿掉策略最好的那一年 without_each_year = { int(y): {"strategy_sharpe": sharpe(original[original.index.year != y]), "benchmark_sharpe": sharpe(bench[bench.index.year != y])} for y in sorted(set(original.index.year))} best = max(result["yearly"], key=result["yearly"].get) best_removed = without_each_year[best] result["without_best_year"] = {"year": best, **best_removed, "wins": best_removed["strategy_sharpe"] > best_removed["benchmark_sharpe"]} result["without_each_year"] = without_each_year checks = {"neighbors": result["neighbor_wins"] >= cfg["min_wins"], "halves": all(h["wins"] for h in result["halves"]), "full_with_costs": result["strategy"]["sharpe"] > result["benchmark"]["sharpe"]} start_day = original_position.index[0] curves = {cfg["name"]: equity(original, start_day)} if cfg.get("average_control"): # 同平均權重固定持有：每月調回原參數的平均比重，同樣扣預設成本 average = original_position.mean() control = fixed_weights(prices, start, {c: w for c, w in average.items() if w > 0}) _, control_returns = backtest(control, "M") control_metrics = metrics(control_returns) result["average_weight_control"] = {"weights": {c: float(w) for c, w in average.items()}, **control_metrics, "strategy_wins": result["strategy"]["sharpe"] > control_metrics["sharpe"]} checks["average_weight_control"] = result["average_weight_control"]["strategy_wins"] curves[cfg["name"] + "｜同平均權重固定持有"] = equity(control_returns, start_day) below = original_position[BENCH] < 1 result["signal"] = {"share_below_average": float(below.mean()), "switches": int((below != below.shift()).iloc[1:].sum())} # 成本敏感度（描述，不是過關條件）：不計成本、證交稅改 0.1% result["cost_sensitivity"] = { label: metrics(backtest(original_position, original_freq, **kw)[1]) for label, kw in [("no_costs", dict(fee_ratio=0, tax_ratio=0)), ("tax_0.1%", dict(tax_ratio=0.001))]} result["checks"] = checks result["passed"] = all(checks.values()) return result, curves, original_report # ---------- 描述性對照（不是過關條件） ---------- def asset_stats(prices): stats = {} for ticker in ETFS: returns = prices[ticker].dropna().pct_change().iloc[1:] stats[ticker] = {"from": str(returns.index[0].date()), **metrics(returns), "yearly": yearly(returns)} return stats def weekly_vs_bench(daily): """各檔和 0050 的週報酬關係；daily 第一欄是 0050，其餘欄照輸出的順序排。""" weekly = daily.resample("W-FRI").last() weekly = weekly[weekly.index <= pd.Timestamp(DATA_END)] # 只留完整的週 weekly = weekly.pct_change().iloc[1:].dropna() worst = weekly[BENCH] <= weekly[BENCH].quantile(0.05) others = [c for c in weekly if c != BENCH] return { "from": str(weekly.index[0].date()), "to": str(weekly.index[-1].date()), "weeks": len(weekly), "worst_weeks": int(worst.sum()), "correlation_with_0050": {c: float(weekly[c].corr(weekly[BENCH])) for c in others}, "average_in_worst_weeks": {c: float(weekly.loc[worst, c].mean()) for c in weekly}, "share_up_in_worst_weeks": {c: float((weekly.loc[worst, c] > 0).mean()) for c in others}, } def currency_comparison(prices): """台幣計價（00679B、00635U）與美元計價（TLT、GLD）各自和 0050 的週報酬關係。""" us = data.get("us_fund_price:adj_close")[["TLT", "GLD"]] daily = pd.concat([prices, us], axis=1, sort=True).ffill() return weekly_vs_bench(daily.loc["2017-01-13":DATA_END, [BENCH, "00679B", "TLT", "00635U", "GLD"]]) def corporate_side_by_side(prices): """③ 的期間同樣 60／40、每季調整，換成美國公債 ETF 並列。""" start = DESIGNS["d3"]["start"] rows, curves = {}, {} for bond in ["00720B", "00679B", "00687B"]: position = fixed_weights(prices, start, {BENCH: 0.6, bond: 0.4}) _, returns = backtest(position, "Q") rows[bond] = metrics(returns) rows[bond]["yearly"] = yearly(returns) if bond != "00720B": # 00720B 這組就是 ③ 的原參數，淨值已經在 ③ 裡 curves[f"0050六成＋{bond}四成"] = equity(returns, position.index[0]) # 單獨持有債券 ETF 時，跟 0050 的週報酬相關，以及 0050 最差 5% 週的平均週報酬 rows["weekly"] = weekly_vs_bench(prices.loc[start:, [BENCH, "00720B", "00679B", "00687B"]]) return rows, curves def held(prices, ticker, index): """單純持有的含息淨值：index 第一天記為 1，還沒上市的日子留空。""" returns = prices[ticker].pct_change().loc[index] returns.iloc[0] = 0 return (1 + returns).cumprod().where(prices[ticker].loc[index].notna()) # 每篇文章：用到的設計、data.csv 另外放哪些 ETF 單純持有的淨值、metrics.json 附上哪些描述性對照 ARTICLES = { "0050-us-treasury-etf-allocation-backtest": dict(designs=["d1", "d2"], assets=["00679B", "00687B"], currency=True), "00720b-corporate-bond-etf-backtest": dict(designs=["d3"], assets=["00720B", "00679B"], side_by_side=True), "gold-etf-00635u-0050-backtest": dict(designs=["d4", "d5"], assets=["00635U", "00679B"], currency=True), "00632r-inverse-etf-hedge-backtest": dict(designs=["d6"], assets=["00632R"]), } def main(out_dir, save_html): finlab.login() prices = load_prices() assets = asset_stats(prices) currency = currency_comparison(prices) side_by_side, side_by_side_curves = corporate_side_by_side(prices) common = {"run_date": str(date.today()), "finlab_version": finlab.__version__, "data_end": DATA_END, "benchmark": "0050 含息（etl:adj_close 直接連乘，不經 sim、不扣成本）", "sharpe": "日報酬平均 / 標準差 × √252，不扣無風險利率"} for slug, article in ARTICLES.items(): folder = Path(out_dir) / slug folder.mkdir(parents=True, exist_ok=True) output = dict(common, designs={}, assets={t: assets[t] for t in [BENCH, *article["assets"]]}) curves = {} for key in article["designs"]: result, design_curves, report = evaluate(prices, key) output["designs"][key] = result curves.update(design_curves) if save_html and key == article["designs"][0]: report.to_html(str(folder / "report_strategy.html")) print(slug, key, result["name"], "鄰近", result["neighbor_wins"], "/", result["neighbor_total"], "通過" if result["passed"] else "未通過", flush=True) if article.get("currency"): output["currency_comparison"] = currency if article.get("side_by_side"): output["side_by_side_2018"] = side_by_side curves.update(side_by_side_curves) frame = pd.DataFrame(curves) frame.insert(0, "0050含息", held(prices, BENCH, frame.index)) for ticker in article["assets"]: frame[ticker + "含息"] = held(prices, ticker, frame.index) frame.index.name = "date" frame.round(6).to_csv(folder / "data.csv") with open(folder / "metrics.json", "w", encoding="utf-8") as f: json.dump(output, f, ensure_ascii=False, indent=1) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--out", default="out") parser.add_argument("--html", action="store_true") args = parser.parse_args() main(args.out, args.html)