""" 當沖券差借券費率飆升後買進：2 個設計 18 組參數的三項驗證與事件統計（2015 年起至最新資料） 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個設計 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 時點：訊號日收盤後才知道，sim() 在隔一個交易日收盤成交。流動性＝訊號日 20 日均成交金額 ≥ 2,000 萬。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 每個訊號的定義與參數在看結果前就寫定，之後沒有改過。 intraday_lending_fee 是證交所「應付現股當日沖銷券差借券費率」與櫃買中心同類資料：一筆券差借一天的費率、日計價、上限 7%， 跟借券市場的年化費率不同。當晚公布，訊號日當天可得。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入，約 30 秒。 輸出：metrics.json（全部結果）、data.csv（各設計原參數、0050 含息與等權流動股的每日淨值）。 """ 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 COSTS = {"default": {}, "zero": {"fee_ratio": 0, "tax_ratio": 0}} def plain(key): return pd.DataFrame(data.get(key)) # --------------------------------------------------------------------------- # 1. 共同資料與績效計算 # --------------------------------------------------------------------------- close = plain("price:收盤價") days = close.index ND = len(days) adj_close = plain("etl:adj_close").reindex(days) bench = adj_close["0050"].dropna() # 上市櫃 4 碼普通股（不含 0 開頭的 ETF、91 開頭的存託憑證），含已下市 STOCKS = [c for c in close.columns if len(c) == 4 and c.isdigit() and c[0] != "0" and not c.startswith("91")] COL = {c: i for i, c in enumerate(STOCKS)} amount = plain("price:成交金額").reindex(index=days, columns=STOCKS) amount_20d = amount.rolling(20, min_periods=20).mean() adj_frame = adj_close.reindex(columns=STOCKS).ffill() adj = adj_frame.to_numpy() adj0050 = adj_close["0050"].ffill().to_numpy() def liquid_mask(floor=2e7): """20 日均成交金額 ≥ floor（預設 2,000 萬）。""" return (amount_20d >= floor).fillna(False).to_numpy() liquid = liquid_mask() def month_ends(): s = pd.Series(np.arange(ND), index=days) return s.groupby([s.index.year, s.index.month]).max().tolist() 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 sd = ret.std() return {"start": str(curve.index[0].date()), "end": str(curve.index[-1].date()), "cagr": round(float(curve.iloc[-1] ** (1 / years) - 1), 6) if years > 0 else None, "sharpe": round(float(ret.mean() / sd * math.sqrt(252)), 6) if sd > 0 else None, "mdd": round(float((curve / curve.cummax() - 1).min()), 6), "total_return": round(float(curve.iloc[-1] - 1), 6)} def active(curve): """從第一次淨值變動的前一個交易日起算。""" curve = curve.dropna() moved = np.flatnonzero(curve.values != curve.values[0]) return curve if len(moved) == 0 else curve.iloc[max(moved[0] - 1, 0):] def compare(curve): s = stats(active(curve)) b = stats(bench.loc[s["start"]:s["end"]]) return {"strategy": s, "bench": b, "win": bool(s["sharpe"] > b["sharpe"])} def halves(curve): curve = active(curve) mid = curve.index[0] + (curve.index[-1] - curve.index[0]) / 2 out = [] for part in (curve.loc[:mid], curve.loc[mid:]): s = stats(part) b = stats(bench.loc[s["start"]:s["end"]]) out.append({"strategy": s, "bench": b, "win": bool(s["sharpe"] > b["sharpe"])}) return out def backtest(table, cost, events=None): frame = pd.DataFrame(table, index=days, columns=STOCKS) report = sim(frame.loc[:, frame.any()], upload=False, **COSTS[cost]) return report, frame, events def event_sim(rows, cost): """rows：(股票, b, e)＝b 收盤買、e 收盤賣。sim 在持有表為真的隔天收盤成交，所以持有表是 [b−1, e−2]。""" table = np.zeros((ND, len(STOCKS)), dtype=bool) count = 0 for s, b, e in rows: lo, hi = max(int(b) - 1, 0), min(int(e) - 2, ND - 1) if s in COL and hi >= lo and lo < ND: table[lo:hi + 1, COL[s]] = True count += 1 return backtest(table, cost, count) def period_sim(selections, cost): """selections：[(t, 股票清單)]，t 收盤決定，持有到下一個 t。""" table = np.zeros((ND, len(STOCKS)), dtype=bool) for k, (t, stocks) in enumerate(selections): end = selections[k + 1][0] if k + 1 < len(selections) else ND table[t:end, [COL[s] for s in stocks if s in COL]] = True return backtest(table, cost) def position_stats(report, frame, events): curve = active(report.creturn) held = frame.sum(axis=1).reindex(curve.index).fillna(0) buys = (frame & ~frame.shift(1, fill_value=False)).sum().sum() years = max((curve.index[-1] - curve.index[0]).days / 365.25, .01) return {"events": events, "buys_per_year": round(float(buys / years), 1), "invested": round(float((held > 0).mean()), 4), "avg_holdings": round(float(held[held > 0].mean()), 2) if (held > 0).any() else None} def equal_weight_curve(): """等權流動股：每月底買進全部流動股、等權、含預設成本，當第二個對照組。""" picks = [(m, [STOCKS[j] for j in np.flatnonzero(liquid[m])]) for m in month_ends() if liquid[m].any()] return period_sim(picks, "default")[0].creturn def validate(key, equal_weight): design = DESIGNS[key] xs, ys, base = design["x"], design["y"], design["base"] grid, base_run = [], None for x in xs: for y in ys: run = design["fn"](x, y, "default") row = compare(run[0].creturn) row.update({"x": x, "y": y, "pos": position_stats(*run)}) grid.append(row) if (x, y) == base: base_run = run base_row = next(g for g in grid if (g["x"], g["y"]) == base) s = base_row["strategy"] result = {"title": design["title"], "x_name": design["x_name"], "y_name": design["y_name"], "x": xs, "y": ys, "base": list(base), "grid": grid, "grid_wins": sum(g["win"] for g in grid), "base_default_cost": base_row, "halves": halves(base_run[0].creturn), "base_zero_cost": compare(design["fn"](*base, "zero")[0].creturn), "equal_weight_same_window": stats(equal_weight.loc[s["start"]:s["end"]])} h = result["halves"] result["pass"] = {"neighbors": result["grid_wins"] >= 7, "halves": h[0]["win"] and h[1]["win"], "cost": base_row["win"]} return result, base_run[0] # --------------------------------------------------------------------------- # 2. 事件研究：減等權流動股（每日等權、無成本） # --------------------------------------------------------------------------- def equal_weight_index(): r = adj_frame.pct_change(fill_method=None).to_numpy() held = np.vstack([np.zeros((1, liquid.shape[1]), dtype=bool), liquid[:-1]]) v = np.where(held & np.isfinite(r), r, np.nan) with np.errstate(invalid="ignore"): return np.cumprod(1 + np.nan_to_num(np.nanmean(v, axis=1))) EW_INDEX = equal_weight_index() def study(rows, horizons=(5, 10, 20, 40, 60)): """b 收盤買進、持有 h 日的平均與中位超額報酬；t 值先按進場月平均，避免同月事件灌大樣本。""" out = {} for h in horizons: rec = [] for s, b, _ in rows: if b + h >= ND: continue j = COL[s] r = adj[b + h, j] / adj[b, j] - 1 if np.isfinite(r): rec.append((b, r - (EW_INDEX[b + h] / EW_INDEX[b] - 1), r - (adj0050[b + h] / adj0050[b] - 1))) if not rec: continue f = pd.DataFrame(rec, columns=["b", "x", "x0050"]) by_month = f.groupby(pd.Series(days[f.b.values].to_period("M"), index=f.index)).x.mean() se = by_month.std(ddof=1) / np.sqrt(len(by_month)) if len(by_month) > 2 else float("nan") out[str(h)] = {"n": len(f), "mean_excess": round(float(f.x.mean()), 5), "median_excess": round(float(f.x.median()), 5), "hit_rate": round(float((f.x > 0).mean()), 4), "t_month": round(float(by_month.mean() / se), 2) if np.isfinite(se) and se > 0 else None, "mean_excess_0050": round(float(f.x0050.mean()), 5)} return out def event_curve(rows, lo=-20, hi=60): """以 b−1（決策日）為 0 的平均累積超額報酬。""" offsets = list(range(lo, hi + 1)) total, count = np.zeros(len(offsets)), np.zeros(len(offsets)) for s, b, _ in rows: t0 = b - 1 if t0 + lo < 0 or not np.isfinite(adj[t0, COL[s]]): continue for k, o in enumerate(offsets): t = t0 + o if t >= ND: break v = adj[t, COL[s]] / adj[t0, COL[s]] - EW_INDEX[t] / EW_INDEX[t0] if np.isfinite(v): total[k] += v count[k] += 1 return {"offset": offsets, "mean": [round(float(x), 5) for x in total / np.maximum(count, 1)], "n": [int(x) for x in count]} # --------------------------------------------------------------------------- # 3. 券差借券費率 # --------------------------------------------------------------------------- raw = plain("intraday_lending_fee") raw["date"] = pd.to_datetime(raw["date"]) raw["stock_id"] = raw["stock_id"].astype(str) trades = raw[raw["stock_id"].isin(COL) & (raw["借券費率"] <= 20)].copy() # 超過 20% 的是異常值 trades["fee"] = trades["借券費率"] * trades["借券成交股數"] daily = trades.groupby(["date", "stock_id"]).agg(shares=("借券成交股數", "sum"), fee=("fee", "sum")) shares = daily["shares"].unstack().reindex(index=days) fee = daily["fee"].unstack().reindex(index=days) rate = fee / shares baseline = fee.fillna(0).rolling(60, min_periods=1).sum().shift(1) / shares.fillna(0).rolling(60, min_periods=1).sum().shift(1) # 前 60 日加權費率 def spike_events(X, multiple=2.0, min_shares=10000): """當日加權費率 ≥ X、≥ 前 60 日 2 倍（前 60 日沒成交只看 X）、當日借券 ≥ 10 張；同檔 20 日內只算第一次。""" hit = (rate >= X) & (shares >= min_shares) & ((rate >= multiple * baseline) | baseline.isna()) first = hit & ~(hit.astype(float).shift(1).rolling(20, min_periods=1).max().fillna(0) > 0) return first.reindex(columns=STOCKS).fillna(False).astype(bool).to_numpy() new_high_60 = (adj_frame >= adj_frame.rolling(60, min_periods=60).max()).to_numpy() # --------------------------------------------------------------------------- # 4. 設計：費率飆升隔日收盤買進，持有 H 日 # --------------------------------------------------------------------------- def spike_rows(X, H, high): ts, js = np.nonzero(spike_events(X) & liquid & (new_high_60 if high else True)) return [(STOCKS[j], t + 1, t + 1 + H) for t, j in zip(ts, js)] def spike(X, H, cost): return event_sim(spike_rows(X, H, False), cost) def squeeze(X, H, cost): return event_sim(spike_rows(X, H, True), cost) GRID = dict(x_name="費率門檻 X（%）", x=[4, 5, 6], y_name="持有 H（交易日）", y=[10, 20, 40], base=(5, 20)) DESIGNS = { "S1": dict(fn=spike, title="券差費率飆升買進", **GRID), "C1": dict(fn=squeeze, title="費率飆升＋60 日新高", **GRID), } # --------------------------------------------------------------------------- # 5. 描述統計 # --------------------------------------------------------------------------- def descriptive(): px = close.reindex(columns=STOCKS).to_numpy() edges = np.arange(0, 7.51, .5) rows = {"spike": spike_rows(5, 20, False), "squeeze": spike_rows(5, 20, True)} t = np.array([b - 1 for _, b, _ in rows["spike"]]) j = np.array([COL[s] for s, _, _ in rows["spike"]]) day_return = px[t, j] / px[t - 1, j] - 1 run_up = adj[t, j] / adj[t - 20, j] - 1 shares_by_rate = np.histogram(trades["借券費率"], edges, weights=trades["借券成交股數"])[0] spikes = {str(x): spike_events(x) for x in (4, 5, 6)} return {"trades": len(trades), "stock_days": len(daily), "stocks": int(trades["stock_id"].nunique()), "first_date": str(trades["date"].min().date()), "by_year": {str(y): {"trades": len(g), "stocks": int(g["stock_id"].nunique())} for y, g in trades.groupby(trades["date"].dt.year)}, "rate_quantiles": {str(q): float(trades["借券費率"].quantile(q)) for q in (.1, .25, .5, .75, .9, .95)}, "zero_fee_trades": int((trades["借券費率"] == 0).sum()), "share_0_5_to_7": round(float(trades["借券費率"].between(.5, 7).mean()), 4), "share_at_0_1": round(float(trades.loc[trades["借券費率"] == .1, "借券成交股數"].sum() / trades["借券成交股數"].sum()), 4), "share_at_2": round(float(trades.loc[trades["借券費率"] == 2, "借券成交股數"].sum() / trades["借券成交股數"].sum()), 4), "share_at_1": round(float(trades.loc[trades["借券費率"] == 1, "借券成交股數"].sum() / trades["借券成交股數"].sum()), 4), "share_at_3": round(float(trades.loc[trades["借券費率"] == 3, "借券成交股數"].sum() / trades["借券成交股數"].sum()), 4), "share_at_3_5": round(float(trades.loc[trades["借券費率"] == 3.5, "借券成交股數"].sum() / trades["借券成交股數"].sum()), 4), "share_at_7": round(float(trades.loc[trades["借券費率"] >= 7, "借券成交股數"].sum() / trades["借券成交股數"].sum()), 4), "trades_above_7": int((trades["借券費率"] > 7).sum()), "excluded_above_20": int((raw["借券費率"] > 20).sum()), "hist_edges": edges.tolist(), "hist_shares": [int(x) for x in shares_by_rate], "lots_quantiles": {str(q): float(daily["shares"].quantile(q) / 1000) for q in (.25, .5, .75)}, "events": {x: int(v.sum()) for x, v in spikes.items()}, "liquid_events_by_year": pd.Series(days[t].year).value_counts().sort_index().to_dict(), "event_day": {"mean_return": round(float(np.nanmean(day_return)), 5), "limit_up_share": round(float(np.nanmean(day_return >= .094)), 4), "up_share": round(float(np.nanmean(day_return > 0)), 4), "run_up_20_mean": round(float(np.nanmean(run_up)), 5), "run_up_20_median": round(float(np.nanmedian(run_up)), 5)}, "study": {k: study(v) for k, v in rows.items()}, "curves": {k: event_curve(v) for k, v in rows.items()}} # --------------------------------------------------------------------------- # 9. 執行 # --------------------------------------------------------------------------- def main(): equal_weight = equal_weight_curve() results, reports = {}, {} for key in DESIGNS: results[key], reports[key] = validate(key, equal_weight) metrics = {"run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "data_end": str(bench.index[-1].date()), "designs": results, "descriptive": descriptive()} with open("metrics.json", "w", encoding="utf-8") as f: json.dump(metrics, f, ensure_ascii=False, indent=1) frame = {} for key, report in reports.items(): curve = active(report.creturn) frame[f"{key} {DESIGNS[key]['title']}"] = curve / curve.iloc[0] frame = pd.DataFrame(frame) frame["0050含息"] = bench.reindex(frame.index) frame["等權流動股"] = equal_weight.reindex(frame.index) frame.index.name = "date" frame.round(6).to_csv("data.csv") print("完成：metrics.json、data.csv") if __name__ == "__main__": main()