""" 台指期結算日回測：結算週空手、只在結算週持有 0050，以及結算日前後的加權指數與 0050 報酬 結算日：futures_price 的 TX 近月合約（TX一般 的到期月份）最後一個交易日。到期月份在結算日當天就換成下個月， 所以結算日＝某個月份合約最後出現那天的下一個交易日。1998-09 起每個月都跟「第三個星期三，遇休市順延」相同， 程式會再核對一次（metrics.json 的 settlement.rule_mismatches）。 兩個設計（窗口用交易日算，s＝結算日）： T1 結算週空手：s−k 到 s＋e 空手，其餘持有 0050。 T2 只在結算週持有：只在 s−k 到 s＋e 持有 0050，其餘現金。 參數 k 1/2/3 × e −1/0/1，原設定 k=2、e=0（結算週的星期一到星期三）。 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 9 組參數至少 7 組贏。 2. 原設定前後兩段（日期中點切開）各自贏。 3. sim() 預設成本：手續費 0.1425%（買賣各一次）、證交稅 0.3%（賣出）。 T2 平均在場不到一半，另要贏同平均權重固定持有 0050；全部通過再看拿掉相對 0050 最好的那一年。 0050 的實際證交稅是 0.1%，和零成本一起只當對照，不當判定。 時點：要賺第 x 天的報酬（x−1 收盤到 x 收盤），要在 x−1 收盤買進；sim(resample=None, trade_at_price="close") 在訊號隔一個交易日收盤成交，所以訊號放在 x−2。結算日是事前就知道的日曆。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 執行：python research.py（資料需要時 finlab 會自動引導登入），輸出 metrics.json 與 data.csv。 """ import json import math from pathlib import Path import numpy as np import pandas as pd import finlab from finlab import data from finlab.backtest import sim DATA_END = "2026-11-06" # 改成 None 就跑到最新資料 OUT = Path(__file__).resolve().parent K_GRID, E_GRID, BASE = (1, 2, 3), (-1, 0, 1), (2, 0) COSTS = { "default": {}, # 手續費 0.1425%、證交稅 0.3% "etf_tax": {"tax_ratio": 0.001}, # 0050 實際證交稅 0.1% "zero": {"fee_ratio": 0, "tax_ratio": 0}, } OFFSETS = range(-5, 6) def plain(key): table = pd.DataFrame(data.get(key)) return table.loc[:DATA_END] if DATA_END else table # --------------------------------------------------------------------------- # 1. 資料與結算日 # --------------------------------------------------------------------------- finlab.login() data.set_market("tw") bench = plain("etl:adj_close")["0050"].dropna() days = bench.index # 0050 的交易日，2007-04-23 起 taiex = plain("taiex_total_index:收盤指數")["TAIEX"].dropna() # 加權指數（價格指數），1999 起 def third_wednesday(year, month): first = pd.Timestamp(year, month, 1) return first + pd.Timedelta(days=(2 - first.weekday()) % 7 + 14) def settlement_days(): """futures_price 的 TX 近月合約：每個月份合約最後出現那天的下一個交易日。""" month = plain("futures_price:到期月份(週別)")["TX一般"].dropna().astype(str) trading = month.index last_seen = month.groupby(month, sort=False).apply(lambda s: s.index[-1]) settle, mismatches = [], [] for contract, day in last_seen.items(): i = trading.searchsorted(day) + 1 if i >= len(trading): # 還沒到結算日的合約 continue settle.append(trading[i]) wednesday = third_wednesday(int(contract[:4]), int(contract[4:])) rule = trading[trading.searchsorted(wednesday)] if rule != trading[i]: mismatches.append((contract, str(trading[i].date()), str(rule.date()))) if trading[i] != wednesday: deferred.append({"contract": contract, "third_wednesday": str(wednesday.date()), "settlement": str(trading[i].date())}) return pd.DatetimeIndex(settle), mismatches deferred = [] # 遇休市順延的結算日 SETTLE, RULE_MISMATCHES = settlement_days() # 資料最後一天之後的交易日與結算日：只用來決定最後兩天的訊號（不知道未來的休市日，用平日代替） future = pd.bdate_range(days[-1] + pd.Timedelta(days=1), periods=45) calendar = days.append(future) upcoming = [third_wednesday(m.year, m.month) for m in pd.period_range(SETTLE[-1].to_period("M") + 1, periods=3, freq="M")] EVENTS = SETTLE.append(pd.DatetimeIndex(upcoming)) EVENT_POS = [calendar.get_loc(s) for s in EVENTS if s in calendar] # --------------------------------------------------------------------------- # 2. 設計與回測 # --------------------------------------------------------------------------- def window(k, e): """結算日前 k 個交易日到結算日＋e（含兩端）為 True。""" mask = np.zeros(len(calendar), dtype=bool) for p in EVENT_POS: mask[max(p - k, 0):min(p + e, len(calendar) - 1) + 1] = True return mask DESIGNS = { "T1": {"name": "結算週空手，其餘持有 0050", "want": lambda k, e: ~window(k, e)}, "T2": {"name": "只在結算週持有 0050", "want": window}, } def position(want): """want[x]＝要賺第 x 天的報酬；訊號往前移兩個交易日。""" return pd.DataFrame({"0050": want[2:len(days) + 2]}, index=days) def backtest(pos, cost="default"): report = sim(pos, resample=None, trade_at_price="close", upload=False, **COSTS[cost]) curve = report.creturn.reindex(days).ffill().fillna(1.0).loc[days[1]:] # 第一天收盤才能買進 return curve / curve.iloc[0], report def stats(curve): 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), "total_return": round(float(curve.iloc[-1] - 1), 6), } def versus(curve): s, b = stats(curve), stats(bench.loc[curve.index[0]:curve.index[-1]]) return {"strategy": s, "bench": b, "win": bool(s["sharpe"] > b["sharpe"])} def halves(curve): mid = curve.index[0] + (curve.index[-1] - curve.index[0]) / 2 return [versus(curve.loc[:mid]), versus(curve.loc[mid:])] def best_year_removed(curve): """拿掉策略相對 0050 超額報酬最好的那個日曆年，兩邊同時拿掉後再比日夏普。""" rs = curve.pct_change().dropna() rb = bench.reindex(curve.index).pct_change().dropna() excess = (rs + 1).groupby(rs.index.year).prod() - (rb + 1).groupby(rb.index.year).prod() best = int(excess.idxmax()) def sharpe(r): r = r[r.index.year != best] return float(r.mean() / r.std() * math.sqrt(252)) s, b = sharpe(rs), sharpe(rb) return {"best_year": best, "excess_that_year": round(float(excess.max()), 6), "strategy_sharpe": round(s, 6), "bench_sharpe": round(b, 6), "win": s > b} def fixed_weight(curve, in_market): """平均在場比例的 0050、其餘現金，每天再平衡。""" rb = bench.reindex(curve.index).pct_change().fillna(0) fixed = (1 + in_market * rb).cumprod() s, f = stats(curve), stats(fixed) return {"weight": round(in_market, 6), "fixed": f, "win": bool(s["sharpe"] > f["sharpe"])} def trades_per_year(report, curve): years = (curve.index[-1] - curve.index[0]).days / 365.25 return round(len(report.get_trades()) / years, 2) def validate(key): spec = DESIGNS[key] grid = [] for k in K_GRID: for e in E_GRID: curve, _ = backtest(position(spec["want"](k, e))) row = versus(curve) grid.append({"k": k, "e": e, "sharpe": row["strategy"]["sharpe"], "cagr": row["strategy"]["cagr"], "mdd": row["strategy"]["mdd"], "bench_sharpe": row["bench"]["sharpe"], "win": row["win"]}) pos = position(spec["want"](*BASE)) in_market = float(pos["0050"].mean()) curve, report = backtest(pos) full = versus(curve) costs = {name: full if name == "default" else versus(backtest(pos, name)[0]) for name in COSTS} out = { "name": spec["name"], "base": {"k": BASE[0], "e": BASE[1]}, "grid": grid, "grid_wins": sum(r["win"] for r in grid), "full": full, "halves": halves(curve), "costs": costs, "in_market": round(in_market, 6), "trades_per_year": trades_per_year(report, curve), "best_year_removed": best_year_removed(curve), } if in_market < 0.5: out["fixed_weight"] = fixed_weight(curve, in_market) out["passes"] = bool(out["grid_wins"] >= 7 and all(h["win"] for h in out["halves"]) and costs["default"]["win"] and out.get("fixed_weight", {"win": True})["win"]) print(f"{key} {spec['name']}: 9 組贏 {out['grid_wins']} 組，原設定日夏普 {out['full']['strategy']['sharpe']:.2f}" f"（0050 {out['full']['bench']['sharpe']:.2f}），{'通過' if out['passes'] else '不通過'}") return out, curve # --------------------------------------------------------------------------- # 3. 結算日前後的報酬（事件統計，不算過關） # --------------------------------------------------------------------------- def event_table(close, events): r = close.pct_change() idx = r.index pos = [int(idx.searchsorted(s)) for s in events if s in idx] rows = [] for j in OFFSETS: v = pd.Series([r.iloc[p + j] for p in pos if 1 <= p + j < len(idx)]).dropna() se = v.std() / math.sqrt(len(v)) rows.append({"offset": j, "n": len(v), "mean": round(float(v.mean()), 6), "ci_low": round(float(v.mean() - 1.96 * se), 6), "ci_high": round(float(v.mean() + 1.96 * se), 6), "up_ratio": round(float((v > 0).mean()), 4)}) return {"first_event": str(idx[pos[0]].date()), "events": len(pos), "all_days_mean": round(float(r.dropna().mean()), 6), "offsets": rows} def inside_outside(close, events, k, e): """結算日前 k 日到＋e 的日報酬 vs 其他交易日，Welch t 值。""" r = close.pct_change().dropna() idx = r.index inside = np.zeros(len(idx), dtype=bool) for s in events: if s in idx: p = idx.get_loc(s) inside[max(p - k, 0):p + e + 1] = True a, b = r[inside], r[~inside] t = (a.mean() - b.mean()) / math.sqrt(a.var() / len(a) + b.var() / len(b)) return {"days_inside": int(inside.sum()), "mean_inside": round(float(a.mean()), 6), "mean_outside": round(float(b.mean()), 6), "t": round(float(t), 3)} def by_period(close, events): r = close.pct_change().dropna() mid = r.index[0] + (r.index[-1] - r.index[0]) / 2 out = [] for part in (close.loc[:mid], close.loc[mid:]): inside = [s for s in events if part.index[0] <= s <= part.index[-1]] minus2 = next(row for row in event_table(part, inside)["offsets"] if row["offset"] == -2) out.append({"start": str(part.index[0].date()), "end": str(part.index[-1].date()), **inside_outside(part, inside, *BASE), "offset_minus_2": minus2}) return out def main(): results, curves = {}, {} for key in DESIGNS: results[key], curves[key] = validate(key) events = { "taiex": {**event_table(taiex, SETTLE), "window": inside_outside(taiex, SETTLE, *BASE), "periods": by_period(taiex, SETTLE)}, "0050": {**event_table(bench, SETTLE), "window": inside_outside(bench, SETTLE, *BASE), "periods": by_period(bench, SETTLE)}, } metrics = { "data_end": str(days[-1].date()), "settlement": {"source": "futures_price:到期月份(週別) TX一般", "first": str(SETTLE[0].date()), "last": str(SETTLE[-1].date()), "count": len(SETTLE), "count_in_0050_window": int(((SETTLE >= days[0]) & (SETTLE <= days[-1])).sum()), "rule_mismatches": RULE_MISMATCHES, "deferred": deferred}, "bench": stats(bench.loc[days[1]:]), "designs": results, "events": events, } (OUT / "metrics.json").write_text(json.dumps(metrics, ensure_ascii=False, indent=1)) table = pd.DataFrame({"T1_skip_settlement_week": curves["T1"], "T2_settlement_week_only": curves["T2"], "0050_buy_hold": bench.loc[days[1]:] / bench.loc[days[1]]}) table.round(6).to_csv(OUT / "data.csv", index_label="date") if __name__ == "__main__": main()