""" 法說會、股東會、季底作帳與台股季節性：9 個訊號、9 個組合版的三項驗證，以及事件前後的統計（2007 年起至最新資料） 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個設計 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 三項都過的，再看拿掉相對 0050 最好的那一年還贏不贏。 疊在月營收動能上的組合版（1c–4c）另外要比底好：原參數日夏普高於同持股數的底、9 組至少 7 組高於同持股數的底， 而且原參數持有的股票日至少 20% 是底沒有持有的。 時點： - 法說會 investors_conference 的 date 是召開日，不是公布日；2024 年以前沒有公布時間。 「法說會前 N 日買」用預測日：同公司去年召開日＋364 天（Frazzini & Lamont 2007、Barber et al. 2013 的做法）。 - 股東常會用公告日期（2015 年起）；之前用開會日前 30 天（公司法第 172 條，公開發行公司常會 30 日前通知）。 - 訊號日收盤後才知道，sim() 在隔一個交易日收盤成交。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入，約 1 分鐘。 輸出：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 MRM_START = "2015-01-01" LEV_START = "2014-11-03" # 00631L 在 2014-10-31 上市 SMALL_START = "2013-12-01" # etl:market_value 從 2013-04 開始 COSTS = {"default": {}, "zero": {"fee_ratio": 0, "tax_ratio": 0}, "etf_tax": {"tax_ratio": 0.001}} 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) liquid = (amount.rolling(20, min_periods=20).mean() >= 2e7).to_numpy() # 20 日均成交金額 ≥ 2,000 萬 market_value = plain("etl:market_value").reindex(index=days, columns=STOCKS).to_numpy() volume = plain("price:成交股數").reindex(index=days, columns=STOCKS).to_numpy() adj = adj_close.reindex(columns=STOCKS).to_numpy() adj0050 = adj_close["0050"].to_numpy() 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), "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.iloc[max(moved[0] - 1, 0):] if len(moved) else curve def compare(curve): curve = active(curve) s = stats(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 return [compare(curve.loc[lo:hi]) for lo, hi in ((curve.index[0], mid), (mid, curve.index[-1]))] def best_year_removed(curve): """拿掉策略相對 0050 超額報酬最好的那個日曆年，兩邊同時拿掉後再比日夏普。""" curve = active(curve) rs = curve.pct_change().dropna() rb = bench.reindex(curve.index).ffill().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 sid(trades): return trades["stock_id"].astype(str).str.split(" ").str[0] def held(report): """實際持有的 (股票, 交易日)：每筆交易從 entry_date 到 exit_date 前一天。""" trades = report.get_trades() idx = report.creturn.index out = set() for s, a, b in zip(sid(trades), trades["entry_date"], trades["exit_date"]): hi = len(idx) if pd.isna(b) else idx.searchsorted(pd.Timestamp(b)) out.update((s, i) for i in range(idx.searchsorted(pd.Timestamp(a)), hi)) return out def trade_stats(report): trades = report.get_trades() if len(trades) == 0: return {} exit_date = pd.to_datetime(trades["exit_date"]).fillna(report.creturn.index[-1]) years = (report.creturn.index[-1] - report.creturn.index[0]).days / 365.25 per_day = pd.Series(0, index=report.creturn.index) counts = pd.Series([i for _, i in held(report)]).value_counts() per_day.iloc[counts.index] = counts.values per_day = per_day.loc[active(report.creturn).index] return {"trades": int(len(trades)), "trades_per_year": round(len(trades) / years, 1), "avg_hold_days": round(float((exit_date - pd.to_datetime(trades["entry_date"])).dt.days.mean()), 1), "win_rate": round(float((trades["return"] > 0).mean()), 4), "mean_return": round(float(trades["return"].mean()), 6), "invested": round(float((per_day > 0).mean()), 4), "avg_holdings": round(float(per_day[per_day > 0].mean()), 1) if (per_day > 0).any() else None} def at(dates): """每個日期當天或之後的第一個交易日位置。""" return days.searchsorted(pd.to_datetime(dates)) # --------------------------------------------------------------------------- # 2. 事件與日曆 # --------------------------------------------------------------------------- raw_conf = plain("investors_conference") conf = raw_conf[raw_conf["stock_id"].astype(str).isin(set(STOCKS))].copy() conf["stock_id"] = conf["stock_id"].astype(str) conf = conf[["stock_id", "date"]].drop_duplicates().sort_values(["stock_id", "date"]).reset_index(drop=True) conf["c"] = at(conf["date"]) # 預測法說日：同公司去年召開日 + 364 天（同一個星期幾） pred = conf[["stock_id"]].assign(pdate=conf["date"] + pd.Timedelta(days=364)) pred = pred[pred["pdate"] <= days[-1]].drop_duplicates().reset_index(drop=True) pred["p"] = at(pred["pdate"]) def excess(s, lo, hi): """股票 s 從 lo 收盤到 hi 收盤的還原報酬，減 0050 同期。""" if lo < 0 or hi >= ND: return np.nan j = COL[s] return adj[hi, j] / adj[lo, j] - adj0050[hi] / adj0050[lo] # 法說兩日反應：召開日前一天收盤到隔天收盤（早上、下午召開的都涵蓋） conf["react"] = [excess(s, c - 1, c + 1) for s, c in zip(conf["stock_id"], conf["c"])] react = conf.dropna(subset=["react"]) raw_agm = plain("shareholders_meeting") agm = raw_agm[(raw_agm["股東會類別"] == "常會") & raw_agm["stock_id"].astype(str).isin(set(STOCKS))].copy() agm["stock_id"] = agm["stock_id"].astype(str) agm = agm[["stock_id", "date", "公告日期"]].drop_duplicates(["stock_id", "date"]).reset_index(drop=True) agm["m"] = at(agm["date"]) # 有公告日期的，公告隔天才算知道；2015 年以前用開會日前 30 天 agm["A"] = np.where(agm["公告日期"].notna(), at(agm["公告日期"] + pd.Timedelta(days=1)), at(agm["date"] - pd.Timedelta(days=30))) trust = plain("institutional_investors_trading_summary:投信買賣超股數").reindex(index=days, columns=STOCKS).fillna(0) shares = market_value / close.reindex(columns=STOCKS).to_numpy() trust60 = trust.rolling(60, min_periods=60).sum().to_numpy() / shares # 近 60 日投信買超 ÷ 流通股數 position_of = pd.Series(np.arange(ND), index=days) month_last = list(position_of.groupby([days.year, days.month]).max()) month_first = list(position_of.groupby([days.year, days.month]).min()) quarter_ends = [i for i in month_last if days[i].month in (3, 6, 9, 12)] def settlement_days(): """台指期每月第三個週三，遇休市順延到下一個交易日。""" out = [] for y in range(days[0].year, days[-1].year + 1): for m in range(1, 13): first = pd.Timestamp(y, m, 1) wed = first + pd.Timedelta(days=(2 - first.weekday()) % 7 + 14) if days[0] <= wed <= days[-1]: out.append(int(days.searchsorted(wed))) return out def lunar_new_year(): """（封關日, 開紅盤日）：每年 1–2 月最長的休市空檔，至少 7 天。""" gap = pd.Series((days[1:] - days[:-1]).days, index=np.arange(1, ND)) out = [] for y in range(days[0].year, days[-1].year + 1): g = gap[[days[i].year == y and days[i].month <= 2 for i in gap.index]] if len(g) and g.max() >= 7: o = int(g.idxmax()) out.append((o - 1, o)) return out SETTLE = settlement_days() LNY = lunar_new_year() # --------------------------------------------------------------------------- # 3. 持有表 # --------------------------------------------------------------------------- def flag(spans, cols=STOCKS): """spans：(股票, lo, hi) 交易日位置（含兩端）→ 布林表。""" a = np.zeros((ND, len(cols)), dtype=bool) col = {c: i for i, c in enumerate(cols)} for s, lo, hi in spans: lo, hi = max(int(lo), 0), min(int(hi), ND - 1) if s in col and hi >= lo: a[lo:hi + 1, col[s]] = True return pd.DataFrame(a, index=days, columns=cols) def holding(rows, cols=STOCKS): """rows：(股票, b, s)＝第 b 天收盤買、第 s 天收盤賣 → 持有表第 b−1 到 s−2 列為真（sim 隔天收盤成交）。""" return flag(((s, b - 1, e - 2) for s, b, e in rows), cols) def run(position, cost, start=None): if start is not None: position = position.loc[start:] return sim(position.loc[:, position.any()], upload=False, **COSTS[cost]) def smallest(t, n): """第 t 天市值最小、當天有成交的 n 檔。""" v = np.where(np.isfinite(market_value[t]) & (np.nan_to_num(volume[t]) > 0), market_value[t], np.nan) ok = np.flatnonzero(np.isfinite(v)) return [STOCKS[i] for i in ok[np.argsort(v[ok], kind="stable")[:n]]] def top_trust(t, k): """第 t 天近 60 日投信買超比率最高的 k 檔（20 日均成交金額過門檻、買超比率 > 0）。""" v = np.where(liquid[t] & np.isfinite(trust60[t]) & (trust60[t] > 0), trust60[t], np.nan) ok = np.flatnonzero(np.isfinite(v)) return [STOCKS[i] for i in ok[np.argsort(-v[ok], kind="stable")[:k]]] def timing(want, cost, levered): """want[x]＝要賺第 x 天的報酬（x−1 收盤到 x 收盤），持有表往前移兩天。 單獨版：開的日子持有 0050、其他日子現金；組合版：開的日子持有 00631L、其他日子 0050。""" on = np.zeros(ND, dtype=bool) on[:-2] = want[2:] if levered: return run(pd.DataFrame({"00631L": on, "0050": ~on}, index=days), cost, LEV_START) return run(pd.DataFrame({"0050": on}, index=days), cost) # --------------------------------------------------------------------------- # 4. 9 個單獨版 # --------------------------------------------------------------------------- def s1(N, e, cost): """預測法說日 p 前 N 個交易日收盤買、p+e 收盤賣。""" rows = [(s, p - N, p + e) for s, p in zip(pred["stock_id"], pred["p"]) if p - N >= 1 and liquid[p - N - 1, COL[s]]] return run(holding(rows), cost) def s2(theta, H, cost): """法說兩日反應 > θ：召開日後第 2 個交易日收盤買、持有 H 個交易日。""" ev = react[react["react"] > theta] rows = [(s, c + 2, c + 2 + H) for s, c in zip(ev["stock_id"], ev["c"]) if c + 1 < ND and liquid[c + 1, COL[s]]] return run(holding(rows), cost) def s3(k, j, cost): """股東常會 m 前 k 個交易日收盤買（不早於公告後第一個交易日）、m+j 收盤賣。""" rows = [] for s, m, A in zip(agm["stock_id"], agm["m"], agm["A"]): b = max(m - k, A) if 1 <= b < m + j and b - 1 < ND and liquid[b - 1, COL[s]]: rows.append((s, b, m + j)) return run(holding(rows), cost) def s4(N, K, cost): """季底作帳：季底前 N 個交易日收盤買近 60 日投信買超比率前 K 檔，季底最後一個交易日收盤賣。""" rows = [(s, q - N, q) for q in quarter_ends for s in top_trust(q - N - 1, K)] return run(holding(rows), cost) def year_end_windows(H): """12 月最後一個交易日收盤買、新年第 H 個交易日收盤賣。""" return [(i, i + H) for i in month_last if days[i].month == 12 and days[i] >= pd.Timestamp(SMALL_START) and i + H < ND] def lny_windows(k, j): """封關日前 k 個交易日收盤買、開紅盤後第 j 個交易日收盤賣（j=1 是開紅盤當天）。""" return [(F - k, O + j - 1) for F, O in LNY if days[F] >= pd.Timestamp(SMALL_START) and O + j - 1 < ND] def small_basket(windows, n): return [(s, b, e) for b, e in windows for s in smallest(b - 1, n)] def basket_or_0050(windows, n, cost): """窗口內持有市值最小 n 檔，窗口外持有 0050。""" position = holding(small_basket(windows, n), STOCKS + ["0050"]) inside = holding((("0050", b, e) for b, e in windows), ["0050"])["0050"] position["0050"] = ~inside return run(position, cost, days[windows[0][0] - 1] - pd.Timedelta(days=40)) def s5(H, n, cost): return run(holding(small_basket(year_end_windows(H), n)), cost) def s9(k, j, cost): return run(holding(small_basket(lny_windows(k, j), 50)), cost) def turn_of_month(k, j): """每月最後 k 個交易日＋次月前 j 個交易日。""" want = np.zeros(ND, dtype=bool) for L in month_last: want[max(L - k + 1, 0):L + 1] = True want[L + 1:min(L + 1 + j, ND)] = True return want def outside_settlement(k, e): """結算日前 k 個交易日到結算日＋e 以外的日子。""" want = np.ones(ND, dtype=bool) for st in SETTLE: want[max(st - k, 0):min(st + e, ND - 1) + 1] = False return want def outside_may_june(a, b): """5 月第一個交易日＋a 到 6 月最後一個交易日＋b 以外的日子。""" want = np.ones(ND, dtype=bool) may = {days[f].year: f for f in month_first if days[f].month == 5} june = {days[L].year: L for L in month_last if days[L].month == 6} for y, f in may.items(): # 資料還沒到 6 月底時，空手到最新一天 want[max(f + a, 0):min(june.get(y, ND - 1) + b, ND - 1) + 1] = False return want # --------------------------------------------------------------------------- # 5. 組合版：月營收動能（/strategies/monthly-revenue-momentum）＋加分 # --------------------------------------------------------------------------- fclose = data.get("price:收盤價") revenue = data.get("monthly_revenue:當月營收") trend = (fclose > fclose.average(60)) & (fclose > fclose.average(20)) & (fclose > fclose.average(120)) candidates = trend & (revenue.average(3) > revenue.average(12)) momentum = (fclose / fclose.shift() - 1).rolling(5).mean() def mrm(N, cost, bonus=None): """符合加分條件的候選股排在前面，同組內依 5 日動能取前 N 檔；每月 11 日換股、10% 停損。""" score = momentum if bonus is not None: bonus = bonus.reindex(index=score.index, columns=score.columns, fill_value=False).astype(bool) score = score + bonus.astype(float) * 10 position = score[candidates].is_largest(N).loc[MRM_START:] return sim(position, resample="M", resample_offset="11D", stop_loss=0.1, upload=False, **COSTS[cost]) def c1(W, N, cost): """預測法說日落在之後 W 個交易日內。""" return mrm(N, cost, flag((s, p - W, p - 1) for s, p in zip(pred["stock_id"], pred["p"]))) def c2(W, N, cost): """過去 W 個交易日內開過法說會、兩日反應 > 2%。""" ev = react[react["react"] > 0.02] return mrm(N, cost, flag((s, c + 1, c + W) for s, c in zip(ev["stock_id"], ev["c"]))) def c3(W, N, cost): """已公告的股東常會落在之後 W 個交易日內。""" return mrm(N, cost, flag((s, max(m - W, A), m - 1) for s, m, A in zip(agm["stock_id"], agm["m"], agm["A"]))) def trust_percentile(): """每天近 60 日投信買超比率在股票池（20 日均成交金額過門檻）的百分位，只留買超 > 0 的。""" t = pd.DataFrame(trust60, index=days, columns=STOCKS) return t.where(liquid & np.isfinite(trust60)).rank(axis=1, pct=True).where(t > 0) TRUST_PCT = trust_percentile() def c4(q, N, cost): """3、6、9、12 月換股時，近 60 日投信買超比率在股票池前 q。""" top = TRUST_PCT >= 1 - q top.loc[~days.month.isin([3, 6, 9, 12])] = False return mrm(N, cost, top) DESIGNS = { "1": dict(name="預測法說日前 N 日買進", fn=s1, x=("預測法說日前進場交易日 N", [5, 10, 20]), y=("出場（相對預測法說日）e", [-1, 0, 1]), base=(10, 0)), "2": dict(name="法說反應為正後持有", fn=s2, x=("兩日超額反應門檻 θ", [0.0, 0.02, 0.04]), y=("持有交易日 H", [40, 60, 80]), base=(0.02, 60)), "3": dict(name="股東常會前買進", fn=s3, x=("股東會前進場交易日 k", [5, 10, 20]), y=("出場（相對股東會）j", [-3, 0, 3]), base=(10, 0)), "4": dict(name="季底作帳：投信買超股季底前買進", fn=s4, x=("季底前進場交易日 N", [5, 10, 20]), y=("投信買超前 K 檔", [10, 20, 30]), base=(10, 20)), "5": dict(name="元月效應：年底買市值最小股", fn=s5, x=("新年第 H 個交易日賣", [5, 10, 20]), y=("市值最小 n 檔", [30, 50, 100]), base=(10, 50)), "6": dict(name="月底月初持有 0050", fn=lambda k, j, cost: timing(turn_of_month(k, j), cost, False), x=("月底最後 k 個交易日", [0, 1, 2]), y=("次月前 j 個交易日", [2, 3, 4]), base=(1, 3)), "7": dict(name="避開結算週持有 0050", fn=lambda k, e, cost: timing(outside_settlement(k, e), cost, False), x=("結算日前 k 日起空手", [1, 2, 3]), y=("空手到結算日＋e", [-1, 0, 1]), base=(2, 0)), "8": dict(name="五窮六絕：5、6 月空手", fn=lambda a, b, cost: timing(outside_may_june(a, b), cost, False), x=("5 月第一個交易日＋a 起空手", [-10, 0, 10]), y=("空手到 6 月最後一個交易日＋b", [-10, 0, 10]), base=(0, 0)), "9": dict(name="封關前買市值最小股、開紅盤後賣", fn=s9, x=("封關前進場交易日 k", [3, 5, 10]), y=("開紅盤後第 j 個交易日賣", [1, 3, 5]), base=(5, 3)), "1c": dict(name="月營收動能＋預測法說日加分", fn=c1, x=("預測法說日在之後 W 日內", [10, 20, 30]), y=("持股數 N", [8, 10, 12]), base=(20, 10)), "2c": dict(name="月營收動能＋法說反應為正加分", fn=c2, x=("過去 W 日內法說反應 > 2%", [20, 40, 60]), y=("持股數 N", [8, 10, 12]), base=(40, 10)), "3c": dict(name="月營收動能＋股東會將至加分", fn=c3, x=("股東常會在之後 W 日內", [10, 20, 30]), y=("持股數 N", [8, 10, 12]), base=(20, 10)), "4c": dict(name="月營收動能＋季底月投信買超加分", fn=c4, x=("投信買超比率前 q", [0.05, 0.10, 0.20]), y=("持股數 N", [8, 10, 12]), base=(0.10, 10)), "5c": dict(name="0050，元月窗口換成市值最小股", fn=lambda H, n, cost: basket_or_0050(year_end_windows(H), n, cost), x=("新年第 H 個交易日賣", [5, 10, 20]), y=("市值最小 n 檔", [30, 50, 100]), base=(10, 50)), "6c": dict(name="0050↔00631L：月底月初加槓桿", fn=lambda k, j, cost: timing(turn_of_month(k, j), cost, True), x=("月底最後 k 個交易日", [0, 1, 2]), y=("次月前 j 個交易日", [2, 3, 4]), base=(1, 3)), "7c": dict(name="0050↔00631L：結算週以外加槓桿", fn=lambda k, e, cost: timing(outside_settlement(k, e), cost, True), x=("結算日前 k 日起降槓桿", [1, 2, 3]), y=("降槓桿到結算日＋e", [-1, 0, 1]), base=(2, 0)), "8c": dict(name="0050↔00631L：5、6 月以外加槓桿", fn=lambda a, b, cost: timing(outside_may_june(a, b), cost, True), x=("5 月第一個交易日＋a 起降槓桿", [-10, 0, 10]), y=("降槓桿到 6 月最後一個交易日＋b", [-10, 0, 10]), base=(0, 0)), "9c": dict(name="0050，封關窗口換成市值最小股", fn=lambda k, j, cost: basket_or_0050(lny_windows(k, j), 50, cost), x=("封關前進場交易日 k", [3, 5, 10]), y=("開紅盤後第 j 個交易日賣", [1, 3, 5]), base=(5, 3)), } MRM_COMBOS = ("1c", "2c", "3c", "4c") ETF_TIMING = ("6", "7", "8", "6c", "7c", "8c") def validate(key, base_reports): cfg = DESIGNS[key] grid, base_report = [], None for xv in cfg["x"][1]: for yv in cfg["y"][1]: report = cfg["fn"](xv, yv, "default") cell = compare(report.creturn) cell.update({"x": xv, "y": yv, "trades": trade_stats(report)}) grid.append(cell) if (xv, yv) == cfg["base"]: base_report, base_cell = report, cell res = {"name": cfg["name"], "x": {"name": cfg["x"][0], "values": cfg["x"][1]}, "y": {"name": cfg["y"][0], "values": cfg["y"][1]}, "base": list(cfg["base"]), "grid": grid, "grid_wins": sum(c["win"] for c in grid), "base_default_cost": base_cell, "halves": halves(base_report.creturn), "base_zero_cost": compare(cfg["fn"](*cfg["base"], "zero").creturn), "best_year_removed": best_year_removed(base_report.creturn)} if key in ETF_TIMING: res["base_etf_tax"] = compare(cfg["fn"](*cfg["base"], "etf_tax").creturn) passed = {"neighbors": res["grid_wins"] >= 7, "halves": all(h["win"] for h in res["halves"]), "cost": base_cell["win"]} if key in MRM_COMBOS: by_n = {n: compare(r.creturn)["strategy"]["sharpe"] for n, r in base_reports.items()} beats = [c["strategy"]["sharpe"] > by_n[c["y"]] for c in grid] held_base, held_combo = held(base_reports[cfg["base"][1]]), held(base_report) res["beats_base"] = int(sum(beats)) res["new_holding_share"] = round(len(held_combo - held_base) / len(held_combo), 4) passed["beats_base"] = bool(base_cell["strategy"]["sharpe"] > by_n[cfg["base"][1]] and sum(beats) >= 7 and res["new_holding_share"] >= 0.2) three = passed["neighbors"] and passed["halves"] and passed["cost"] passed["best_year"] = bool(res["best_year_removed"]["win"]) if three else None passed["all"] = all(v for v in passed.values() if v is not None) res["pass"] = passed print(f"{key} {cfg['name']}：{res['grid_wins']}/9 組贏，原參數日夏普 {base_cell['strategy']['sharpe']:.2f}" f"（0050 {base_cell['bench']['sharpe']:.2f}），通過：{passed['all']}", flush=True) return res, base_report # --------------------------------------------------------------------------- # 6. 事件前後的統計（不算設計數） # --------------------------------------------------------------------------- def descriptive(): out = {} # 法說會公布提前天數：key_date 2024-02 起才是首次出現時間；排除單日超過 60 筆的整批補抓日 kd = pd.to_datetime(raw_conf["key_date"]).dt.normalize() recent = raw_conf.assign(kd=kd)[(kd >= "2024-02-01") & (raw_conf["date"] >= kd) & (raw_conf["date"] <= days[-1])] recent = recent[recent["kd"].map(recent["kd"].value_counts()) <= 60] lead = at(recent["date"]) - at(recent["kd"]) calendar_lead = (recent["date"] - recent["kd"]).dt.days out["lead"] = {"n": int(len(recent)), "median_calendar_days": float(calendar_lead.median()), "share_ge_5_trading_days": round(float((lead >= 5).mean()), 4), "share_ge_10_trading_days": round(float((lead >= 10).mean()), 4), "histogram": {str(k): int(v) for k, v in pd.Series(np.clip(lead, 0, 15)).value_counts().sort_index().items()}} # 預測日命中率：預測日前後 7 天內，同公司真的有開法說會 by_stock = conf.groupby("stock_id")["date"].apply(lambda s: s.to_numpy()) hits = [bool((np.abs((by_stock[s] - np.datetime64(p)) / np.timedelta64(1, "D")) <= 7).any()) for s, p in zip(pred["stock_id"], pred["pdate"])] out["prediction_hit_within_7_days"] = round(float(np.mean(hits)), 4) # 召開日前後每天的平均超額報酬：實際召開日（事後才知道）vs 預測日 def path(events): rows = [(s, c) for s, c in events if c - 21 >= 0 and c + 5 < ND and liquid[c - 21, COL[s]]] return [round(float(np.nanmean([excess(s, c + k - 1, c + k) for s, c in rows])), 6) for k in range(-20, 6)] out["conference_path"] = {"days": list(range(-20, 6)), "actual": path(zip(conf["stock_id"], conf["c"])), "predicted": path(zip(pred["stock_id"], pred["p"]))} # 法說反應五分組，之後 60 個交易日的超額報酬 ev = react[[c + 62 < ND and liquid[c + 1, COL[s]] for s, c in zip(react["stock_id"], react["c"])]].copy() ev["post60"] = [excess(s, c + 2, c + 62) for s, c in zip(ev["stock_id"], ev["c"])] ev["group"] = pd.qcut(ev["react"], 5, labels=False) + 1 g = ev.dropna(subset=["post60"]).groupby("group").agg(react=("react", "mean"), post60=("post60", "mean"), n=("post60", "size")) out["conference_drift"] = [{"group": int(k), "react": round(float(r.react), 6), "post60": round(float(r.post60), 6), "n": int(r.n)} for k, r in g.iterrows()] # 股東常會前 10 個交易日（不早於公告）的超額報酬 pre = [] for s, m, A in zip(agm["stock_id"], agm["m"], agm["A"]): b = max(m - 10, A) if m < ND and 1 <= b < m and liquid[b - 1, COL[s]]: pre.append(excess(s, b, m)) lead = (agm["date"] - agm["公告日期"]).dt.days out["agm_notice_lead_days_median_by_year"] = {str(y): float(v) for y, v in lead.groupby(agm["date"].dt.year).median().dropna().items()} out["agm_pre10"] = {"mean": round(float(np.nanmean(pre)), 6), "median": round(float(np.nanmedian(pre)), 6), "n": int(np.isfinite(pre).sum())} # 季底作帳：投信買超前 20 檔，季底前 10 日、季底後 10 日的超額報酬 before, after = [], [] for q in quarter_ends: top = top_trust(q - 11, 20) if not top: continue before.append(np.nanmean([excess(s, q - 10, q) for s in top])) if q + 10 < ND: after.append(np.nanmean([excess(s, q, q + 10) for s in top])) before = np.array(before)[np.isfinite(before)] out["window_dressing"] = {"quarters": len(before), "pre10_mean": round(float(before.mean()), 6), "pre10_positive_share": round(float((before > 0).mean()), 4), "post10_mean": round(float(np.nanmean(after)), 6)} # 市值最小 50 檔：元月（12 月最後一個交易日→新年第 10 日）、封關前 5 日→開紅盤後第 3 日，每年超額報酬 def by_year(windows): return {str(days[e].year): round(float(np.nanmean([excess(s, b, e) for s in smallest(b - 1, 50)])), 6) for b, e in windows} out["small_cap_january"] = by_year(year_end_windows(10)) out["small_cap_lunar_new_year"] = by_year(lny_windows(5, 3)) # 0050 日報酬：月底月初、結算窗口、5–6 月 vs 其他日子 r = pd.Series(adj0050, index=days).pct_change() def split(inside): def f(x): return {"days": int(len(x)), "mean_annualized": round(float(x.mean() * 252), 6), "sharpe": round(float(x.mean() / x.std() * math.sqrt(252)), 6)} a, b = r[inside].dropna(), r[~inside].dropna() t = (a.mean() - b.mean()) / math.sqrt(a.var() / len(a) + b.var() / len(b)) # Welch t 值 return {"inside": f(a), "outside": f(b), "t_stat": round(float(t), 4)} out["calendar_days"] = {"turn_of_month": split(turn_of_month(1, 3)), "settlement_window": split(~outside_settlement(2, 0)), "may_june": split(~outside_may_june(0, 0))} out["may_june_by_year"] = {str(y): round(float((1 + r[(days.year == y) & days.month.isin([5, 6])]).prod() - 1), 6) for y in range(days[0].year, days[-1].year + 1)} return out # --------------------------------------------------------------------------- # 7. 執行 # --------------------------------------------------------------------------- def main(): base_reports = {n: mrm(n, "default") for n in (8, 10, 12)} results, reports = {}, {} for key in DESIGNS: results[key], reports[key] = validate(key, base_reports) metrics = { "run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "data_end": str(bench.index[-1].date()), "base_strategy": {"name": "月營收動能（monthly-revenue-momentum）", "start": MRM_START, "by_n": {str(n): compare(r.creturn) for n, r in base_reports.items()}, "halves_n10": halves(base_reports[10].creturn)}, "leveraged_buy_hold": compare(adj_close["00631L"].loc[LEV_START:].dropna()), "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(), ("月營收動能10檔", base_reports[10])]: c = active(report.creturn) frame[key] = c / c.iloc[0] frame = pd.DataFrame(frame) frame["0050含息"] = bench.reindex(frame.index) frame.index.name = "date" frame.round(6).to_csv("data.csv") print("完成：metrics.json、data.csv") if __name__ == "__main__": main()