""" 台股類股輪動：用上市類股成交金額占比選類股，2 個設計 18 組參數的三項驗證（2009 年起至最新資料） 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個設計 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 時點：訊號日收盤後才知道，sim() 在隔一個交易日收盤成交。流動性＝訊號日 20 日均成交金額 ≥ 2,000 萬。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 每個訊號的定義與參數在看結果前就寫定，之後沒有改過。 類股用 stock_index_vol 的上市類指數成交金額。排除電子、化學生技兩個合計類、2023 年新增 4 類與其他類，共 27 類。 類股成分用目前的產業分類，已下市公司不在內，有輕微生存者偏誤。 執行：安裝 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() 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 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] # --------------------------------------------------------------------------- # 3. 類股成交金額占比 # --------------------------------------------------------------------------- SECTORS = { # stock_index_vol 的上市類指數 → company_basic_info 產業類別 "水泥": "水泥工業", "食品": "食品工業", "塑膠": "塑膠工業", "紡織纖維": "紡織纖維", "電機機械": "電機機械", "電器電纜": "電器電纜", "化學": "化學工業", "生技醫療": "生技醫療業", "玻璃陶瓷": "玻璃陶瓷", "造紙": "造紙工業", "鋼鐵": "鋼鐵工業", "橡膠": "橡膠工業", "汽車": "汽車工業", "半導體": "半導體業", "電腦及週邊設備": "電腦及週邊設備業", "光電": "光電業", "通信網路": "通信網路業", "電子零組件": "電子零組件業", "電子通路": "電子通路業", "資訊服務": "資訊服務業", "其他電子": "其他電子業", "建材營造": "建材營造", "航運業": "航運業", "金融保險": "金融保險業", "貿易百貨": "貿易百貨", "油電燃氣": "油電燃氣業", "觀光": "觀光餐旅", } def sector_amount(): v = plain("stock_index_vol:成交金額") v.columns = [str(c).strip() for c in v.columns] out = {k: v[f"上市{k}類指數"] for k in SECTORS if k != "觀光"} out["觀光"] = v["上市觀光事業類指數"].combine_first(v["上市觀光餐旅類指數"]) # 2023 年改名 return pd.DataFrame(out)[list(SECTORS)].reindex(index=days) def sector_members(): """目前的產業分類，只有現存公司，有輕微生存者偏誤。""" info = plain("company_basic_info") info = info[info["市場別"].astype(str).str.lower().eq("sii")] reverse = {v: k for k, v in SECTORS.items()} out = {k: [] for k in SECTORS} for s, c in zip(info["stock_id"].astype(str), info["產業類別"].astype(str)): if s in COL and c in reverse: out[reverse[c]].append(s) return out AMOUNT = sector_amount() MEMBERS = sector_members() SHARE = AMOUNT.div(AMOUNT.sum(axis=1, min_count=20), axis=0) def inflow(S): """占比近 S 日平均 ÷ 近 120 日平均。""" return SHARE.rolling(S, min_periods=max(S - 2, 1)).mean() / SHARE.rolling(120, min_periods=100).mean() def top_sectors(S, K): signal = inflow(S) out = [] for m in month_ends(): row = signal.iloc[m].dropna() if len(row) >= 20: names = row.sort_values(ascending=False, kind="stable").index[:K] out.append((m, list(names))) return out # --------------------------------------------------------------------------- # 4. 設計：月底換股 # --------------------------------------------------------------------------- def rotation(S, K, cost): return period_sim([(m, [s for k in names for s in MEMBERS[k] if liquid[m, COL[s]]]) for m, names in top_sectors(S, K)], cost) def rotation_strength(S, N, cost): picks = [] for m, names in top_sectors(S, 3): r = adj[m] / adj[m - 120] - 1 if m >= 120 else adj[m] * np.nan j = np.array([COL[s] for k in names for s in MEMBERS[k] if liquid[m, COL[s]] and np.isfinite(r[COL[s]])], dtype=int) picks.append((m, [STOCKS[i] for i in j[np.argsort(-r[j], kind="stable")[:N]]])) return period_sim(picks, cost) DESIGNS = { "S6": dict(fn=rotation, title="類股資金流入", x_name="短窗 S（交易日）", x=[10, 20, 40], y_name="類股數 K", y=[1, 2, 3], base=(20, 2)), "C6": dict(fn=rotation_strength, title="資金流入類股內強勢股", x_name="短窗 S（交易日）", x=[10, 20, 40], y_name="持股 N", y=[10, 20, 30], base=(20, 20)), } # --------------------------------------------------------------------------- # 5. 描述統計 # --------------------------------------------------------------------------- def rank_study(S=20): """月底依資金流入排名，下個月（隔日收盤到下月底隔日收盤）類內流動股等權報酬，減 27 類平均。""" signal = inflow(S) ends = month_ends() rows = [] for m, n in zip(ends[:-1], ends[1:]): row = signal.iloc[m].dropna() if len(row) < 20 or n + 1 >= ND: continue ret = {} for k in row.index: j = [COL[s] for s in MEMBERS[k] if liquid[m, COL[s]]] if j: r = adj[n + 1, j] / adj[m + 1, j] - 1 if np.isfinite(r).any(): ret[k] = float(np.nanmean(r)) ranks = row.rank(ascending=False, method="first") mean = np.mean(list(ret.values())) rows += [(days[m], k, int(ranks[k]), v - mean) for k, v in ret.items()] f = pd.DataFrame(rows, columns=["date", "sector", "rank", "excess"]) buckets = {"1–2": (1, 2), "3–7": (3, 7), "8–13": (8, 13), "14–20": (14, 20), "21–27": (21, 27)} out = {} for name, (lo, hi) in buckets.items(): g = f[f["rank"].between(lo, hi)].groupby("date").excess.mean() out[name] = {"months": len(g), "mean_monthly_excess": round(float(g.mean()), 5), "t": round(float(g.mean() / (g.std(ddof=1) / np.sqrt(len(g)))), 2)} return out def descriptive(): selections = top_sectors(20, 2) picks = pd.Series([k for _, names in selections for k in names]).value_counts() yearly = SHARE.groupby(SHARE.index.year).mean().dropna(how="all") top = yearly.iloc[-1].sort_values(ascending=False).index[:6].tolist() monthly = SHARE.groupby([SHARE.index.year, SHARE.index.month]).mean().dropna(how="all") leaders = monthly.idxmax(axis=1).value_counts() slow = SHARE.rolling(120, min_periods=100).mean() picked = [slow.iloc[m][k] for m, names in selections for k in names] return {"months": len(selections), "first_month": str(days[selections[0][0]].date()), "members": {k: len(v) for k, v in MEMBERS.items()}, "share_months": len(monthly), "share_leaders": {k: int(v) for k, v in leaders.items()}, "picked_share_mean": round(float(np.mean(picked)), 4), "pick_counts": {k: int(v) for k, v in picks.items()}, "share_by_year": {k: {str(y): round(float(v), 4) for y, v in yearly[k].items()} for k in top}, "top_sectors_latest_year": top, "rank_study": rank_study(), "semiconductor_share_first_last": [round(float(yearly["半導體"].iloc[0]), 4), round(float(yearly["半導體"].iloc[-1]), 4)]} # --------------------------------------------------------------------------- # 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()