""" 台股公司行動事件：庫藏股、減資、現金增資、股票股利 7 個訊號與 8 個組合的三項驗證 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個設計 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）； 放空另扣融券手續費 0.08%（放空時收，sim 沒有這一項）。 單獨訊號：2013-05 起（每日市值資料的起點，用來算已發行股數）。上市櫃 4 位數普通股。 事件可取得的那天收盤後產生訊號，隔一個交易日開盤進場，持有固定交易日數； 同時持有的事件等權，沒有事件時空手；進場那天沒有開盤價（興櫃、停牌）的事件不算。 組合：疊在「月營收動能 10 檔」（/strategies/monthly-revenue-momentum）上，2015 年起， 加分項＝近期有事件的候選股排前面，避開＝近期有事件的股票不選。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入。 輸出：metrics.json（每個設計 9 組參數的結果與三項驗證）、data.csv（原參數的月底淨值與 0050）。 """ import datetime import itertools import json import numpy as np import pandas as pd import finlab from finlab import data from finlab.backtest import sim START = "2013-05-01" # etl:market_value 從 2013-04-19 起 COMBO_START = "2015-01-01" # 底策略「月營收動能 10 檔」策略頁的起點 SHORT_FEE = 0.0008 # 融券手續費 0.08% TRADING_DAYS = 252 # --------------------------------------------------------------------------- # 1. 股票池、價格、已發行股數與 0050 含息基準 # --------------------------------------------------------------------------- def plain(key): df = pd.DataFrame(data.get(key)) df.index = pd.to_datetime(df.index) return df def events_of(table, columns, key): """寬表（index＝可取得日、欄＝股票）攤成一列一個事件。""" out = pd.concat({c: plain(f"{table}:{c}").stack() for c in columns}, axis=1) out = out[out[key].notna()].reset_index() return out.rename(columns={out.columns[0]: "date", out.columns[1]: "stock"}) close = plain("price:收盤價") stocks = [s for s in close.columns if len(s) == 4 and s.isdigit() and not s.startswith("0")] close = close[stocks] days = close.index bench = plain("etl:adj_close")["0050"].dropna() shares = (plain("etl:market_value").reindex(index=days, columns=stocks) / close).ffill() adj_open = plain("etl:adj_open").reindex(index=days) START_POS = int(days.searchsorted(pd.Timestamp(START))) def last_day(d): """d 當天（或之前最近一個交易日）的位置：d 收盤後才知道的事，隔天開盤才能進場。""" return int(days.searchsorted(pd.Timestamp(d), side="right")) - 1 def tradable(ev): """訊號隔天有開盤價才買得到；興櫃或停牌中的事件不算。""" entry = np.minimum(ev.sig.to_numpy() + 1, len(days) - 1) return ev[[s in adj_open.columns and np.isfinite(adj_open[s].iloc[i]) for s, i in zip(ev.stock, entry)]] def value_on(df, d, stock): i = last_day(d) return df[stock].iloc[i] if i >= 0 else np.nan # --------------------------------------------------------------------------- # 2. 事件表：每列一個事件，sig＝產生訊號的交易日位置 # --------------------------------------------------------------------------- # 庫藏股：index 是公告日；執行結果（已買回比例）在買回期間結束後才回填，只能用已結束的上一次 treasury = events_of("treasury_stock", [ "預定買回股數", "買回價格區間-最高", "預定買回期間-迄", "本次已買回股數佔預定買回股數比例(%)"], "預定買回股數") treasury = treasury[treasury.stock.isin(stocks)].sort_values("date").reset_index(drop=True) treasury["sig"] = treasury.date.map(last_day) treasury["size"] = [n / value_on(shares, d, s) for n, d, s in zip(treasury["預定買回股數"], treasury.date, treasury.stock)] treasury["premium"] = [p / value_on(close, d, s) - 1 for p, d, s in zip(treasury["買回價格區間-最高"], treasury.date, treasury.stock)] done = treasury.assign(known=treasury["預定買回期間-迄"] + pd.Timedelta(days=7)).dropna(subset=["known"]).sort_values("known") treasury = pd.merge_asof( treasury, done[["stock", "known", "本次已買回股數佔預定買回股數比例(%)"]].rename( columns={"本次已買回股數佔預定買回股數比例(%)": "prev_rate"}), left_on="date", right_on="known", by="stock", suffixes=("", "_prev")) treasury = tradable(treasury) # 減資：index 是恢復買賣日；上櫃的減資原因 2019 年起才有。 # *_cap_divide_ratio＝停止買賣前收盤價 ÷ 恢復買賣參考價；size＝1 − 這個比例。 # 彌補虧損減資的 size 就是減資比例；現金減資有退款，size 會比減資比例小。 reductions = [] for market, ratio in (("tse", "twse_cap_divide_ratio"), ("otc", "otc_cap_divide_ratio")): e = events_of(f"capital_reduction_{market}", ["減資原因", ratio], "減資原因") reductions.append(e.assign(size=1 - e[ratio].astype(float), reason=e["減資原因"].astype(str))) reductions = pd.concat(reductions)[["date", "stock", "size", "reason"]] reductions = reductions[reductions.stock.isin(stocks)].sort_values("date").reset_index(drop=True) reductions["sig"] = reductions.date.map(last_day) reductions["cash"] = reductions.reason.isin(["退還股款", "現金減資"]) reductions["deficit"] = reductions.reason == "彌補虧損" reductions = tradable(reductions) # 現金增資與股票股利：除權交易日開盤進場，訊號在前一個交易日，公告日要在訊號日以前 dividends = pd.DataFrame(data.get("dividend_announcement")) dividends = dividends.assign(stock=dividends["stock_id"].astype(str), ex=dividends["除權交易日"]) dividends = dividends[dividends.stock.isin(stocks) & dividends.ex.notna()].sort_values("公告日期") dividends = dividends.drop_duplicates(["stock", "ex"]) dividends["sig"] = dividends.ex.map(lambda d: int(days.searchsorted(d)) - 1) dividends = dividends[(dividends.sig >= 0) & (dividends["公告日期"].values <= days[dividends.sig.clip(lower=0)].values)] issues = tradable(dividends[dividends["現金增資總股數(股)"] > 0]).copy() issues["size"] = [n / value_on(shares, d, s) for n, d, s in zip(issues["現金增資總股數(股)"], issues["公告日期"], issues.stock)] stock_dividends = tradable(dividends[dividends["盈餘轉增資配股(元/股)"] > 0]).copy() stock_dividends["size"] = stock_dividends["盈餘轉增資配股(元/股)"] def windows(ev, hold): """事件 → 布林持有表：訊號日起連續 hold 個交易日為 True（sim 隔天進場，持有 hold 天）。""" a = np.zeros((len(days), len(stocks)), dtype=bool) col = {s: i for i, s in enumerate(stocks)} for s, sig in zip(ev.stock, ev.sig): a[sig:sig + hold, col[s]] = True return pd.DataFrame(a, index=days, columns=stocks) # --------------------------------------------------------------------------- # 3. 七個單獨訊號：(門檻, 持有天數) → 持有表；short=True 代表放空 # --------------------------------------------------------------------------- def treasury_size(x): return treasury[treasury["size"] >= x] def treasury_premium(x): return treasury[treasury["premium"] >= x] def treasury_repeat(x): return treasury[treasury["prev_rate"] >= x] def cash_reduction(x): return reductions[reductions.cash & (reductions["size"] >= x)] def deficit_reduction(x): return reductions[reductions.deficit & (reductions["size"] >= x)] def cash_issue(x): return issues[issues["size"] >= x] def stock_dividend(x): return stock_dividends[stock_dividends["size"] >= x] HOLD = [20, 60, 120] SINGLES = { # 名稱: (事件篩選, 門檻 3 值, 持有天數 3 值, 原參數, 放空) "treasury_size": (treasury_size, [0.02, 0.03, 0.05], HOLD, (0.03, 60), False), "treasury_premium": (treasury_premium, [0.3, 0.5, 0.7], HOLD, (0.5, 60), False), "treasury_repeat": (treasury_repeat, [80, 90, 100], HOLD, (90, 60), False), "cash_reduction": (cash_reduction, [0.05, 0.1, 0.2], HOLD, (0.1, 60), False), "deficit_reduction": (deficit_reduction, [0.2, 0.4, 0.6], [5, 20, 60], (0.4, 20), True), "cash_issue": (cash_issue, [0.05, 0.1, 0.2], HOLD, (0.1, 60), True), "stock_dividend": (stock_dividend, [0.5, 1.0, 2.0], HOLD, (1.0, 60), False), } # --------------------------------------------------------------------------- # 4. 組合：月營收動能 10 檔（與 /strategies/monthly-revenue-momentum 同一套條件） # --------------------------------------------------------------------------- fclose = data.get("price:收盤價") revenue = data.get("monthly_revenue:當月營收") momentum = (fclose / fclose.shift() - 1).rolling(5).mean() trend = (fclose > fclose.average(60)) & (fclose > fclose.average(20)) & (fclose > fclose.average(120)) candidates = trend & (revenue.average(3) > revenue.average(12)) def recent(ev, window): """過去 window 個交易日內（含訊號日）有事件 → True，對齊到底策略的表格。""" flag = windows(ev, window).reindex(index=fclose.index, columns=fclose.columns) return flag.fillna(False).astype(bool) def momentum_top(n, bonus=None, avoid=None, **cost): score, cond = momentum, candidates if bonus is not None: score = momentum + bonus.astype(float) if avoid is not None: cond = candidates & ~avoid position = score[cond].is_largest(n).loc[COMBO_START:] return sim(position, resample="M", resample_offset="11D", stop_loss=0.1, upload=False, **cost) def bonus_combo(name): select, _, _, (x0, _), _ = SINGLES[name] return lambda window, n, **cost: momentum_top(n, bonus=recent(select(x0), window), **cost) def avoid_combo(*names): def run(window, n, **cost): flags = [recent(SINGLES[k][0](SINGLES[k][3][0]), window) for k in names] return momentum_top(n, avoid=sum(flags) > 0, **cost) return run N_STOCKS = [8, 10, 12] COMBOS = { # 名稱: (回測函式, 窗口 3 值, 持股檔數 3 值, 原參數) "treasury_size+momentum": (bonus_combo("treasury_size"), [20, 60, 120], N_STOCKS, (60, 10)), "treasury_premium+momentum": (bonus_combo("treasury_premium"), [20, 60, 120], N_STOCKS, (60, 10)), "treasury_repeat+momentum": (bonus_combo("treasury_repeat"), [20, 60, 120], N_STOCKS, (60, 10)), "cash_reduction+momentum": (bonus_combo("cash_reduction"), [20, 60, 120], N_STOCKS, (60, 10)), "stock_dividend+momentum": (bonus_combo("stock_dividend"), [20, 60, 120], N_STOCKS, (60, 10)), "avoid_deficit_reduction": (avoid_combo("deficit_reduction"), [60, 120, 250], N_STOCKS, (120, 10)), "avoid_cash_issue": (avoid_combo("cash_issue"), [60, 120, 250], N_STOCKS, (120, 10)), "avoid_all_three": (avoid_combo("deficit_reduction", "cash_issue", "stock_dividend"), [60, 120, 250], N_STOCKS, (120, 10)), } # --------------------------------------------------------------------------- # 5. 績效與三項驗證 # --------------------------------------------------------------------------- def perf(equity): equity = equity / equity.iloc[0] ret = equity.pct_change().dropna() years = (equity.index[-1] - equity.index[0]).days / 365.25 return { "cagr": round(float(equity.iloc[-1] ** (1 / years) - 1), 6), "sharpe": round(float(ret.mean() / ret.std() * np.sqrt(TRADING_DAYS)), 6), "mdd": round(float((equity / equity.cummax() - 1).min()), 6), "start": str(equity.index[0].date()), "end": str(equity.index[-1].date()), } def versus_0050(equity): """全期與前後兩段，策略與 0050 在完全相同的日期上比較。""" b = bench.reindex(equity.index).ffill() mid = equity.index[0] + (equity.index[-1] - equity.index[0]) / 2 out = {} for name, part in [("full", slice(None)), ("first_half", slice(None, mid)), ("second_half", slice(mid, None))]: s, k = perf(equity.loc[part]), perf(b.loc[part]) out[name] = {"strategy": s, "bench": k, "win": s["sharpe"] > k["sharpe"]} return out def equity_of(report, short=False): """從第一次有部位的前一天起算；放空另扣融券手續費。""" equity = report.creturn.dropna() first_move = (equity != equity.iloc[0]).argmax() equity = equity.iloc[max(first_move - 1, 0):] if short: trades = report.get_trades() fee = trades.groupby(pd.to_datetime(trades["entry_date"]))["position"].apply(lambda w: SHORT_FEE * w.abs().sum()) equity = equity * (1 - fee.reindex(equity.index).fillna(0)).cumprod() return equity def trade_stats(report): trades = report.get_trades() held = (report.position.fillna(0) != 0).sum(axis=1) return { "trades": int(len(trades)), "win_rate": round(float((trades["return"] > 0).mean()), 4), "avg_holdings": round(float(held[held > 0].mean()), 1), "invested_days": round(float((held > 0).mean()), 4), } def single_report(name, x, hold, **cost): select, *_, short = SINGLES[name] position = windows(select(x), hold).loc[START:] position = position.loc[:, position.any()] return sim(-position.astype(float) if short else position, trade_at_price="open", upload=False, **cost) def event_excess(ev, hold, short): """事件層級：隔天開盤進場、持有 hold 天（開盤到開盤）的報酬減同期 0050 含息；放空兩者都取負號。""" ev = ev[(ev.sig >= START_POS) & (ev.sig + 1 + hold < len(days))] rows = [] for s, sig in zip(ev.stock, ev.sig): r = adj_open[s].iloc[sig + 1 + hold] / adj_open[s].iloc[sig + 1] - 1 rb = adj_open["0050"].iloc[sig + 1 + hold] / adj_open["0050"].iloc[sig + 1] - 1 if np.isfinite(r) and np.isfinite(rb): rows.append((r - rb) * (-1 if short else 1)) x = pd.Series(rows) return {"events": int(len(x)), "mean_excess": round(float(x.mean()), 4), "median_excess": round(float(x.median()), 4), "hit_rate": round(float((x > 0).mean()), 4)} def three_checks(rows, center): wins = sum(r["full"]["win"] for r in rows) base = next(r for r in rows if (r["x"], r["y"]) == center) checks = {"neighbors": wins / len(rows) >= 0.7, "halves": base["first_half"]["win"] and base["second_half"]["win"], "cost": base["full"]["win"]} checks["all"] = all(checks.values()) return wins, checks, base def main(): results, curves = {}, {} for name, (select, xs, ys, center, short) in SINGLES.items(): rows = [] for x, hold in itertools.product(xs, ys): report = single_report(name, x, hold) equity = equity_of(report, short) row = {"x": x, "y": hold, "events": int((select(x).sig >= START_POS).sum()), **versus_0050(equity), **trade_stats(report)} if (x, hold) == center: curves[name] = equity row["sim_cost_only"] = perf(equity_of(report)) row["zero_cost"] = perf(equity_of(single_report(name, x, hold, fee_ratio=0, tax_ratio=0))) row["event_excess"] = event_excess(select(x), hold, short) rows.append(row) print(name, x, hold, f"sharpe {row['full']['strategy']['sharpe']:.2f} vs 0050 {row['full']['bench']['sharpe']:.2f}", flush=True) wins, checks, _ = three_checks(rows, center) results[name] = {"x": xs, "y": ys, "center": center, "short": short, "wins": wins, "checks": checks, "grid": rows} print(name, "wins", wins, "/", len(rows), checks, flush=True) base_alone = {} for n in N_STOCKS: report = momentum_top(n) equity = equity_of(report) base_alone[n] = {**versus_0050(equity), **trade_stats(report)} if n == 10: curves["momentum_base"] = equity print("base", n, f"sharpe {base_alone[n]['full']['strategy']['sharpe']:.2f}", flush=True) for name, (run, xs, ys, center) in COMBOS.items(): rows = [] for window, n in itertools.product(xs, ys): report = run(window, n) equity = equity_of(report) row = {"x": window, "y": n, **versus_0050(equity), **trade_stats(report)} row["beats_base"] = row["full"]["strategy"]["sharpe"] > base_alone[n]["full"]["strategy"]["sharpe"] if (window, n) == center: curves[name] = equity row["zero_cost"] = perf(equity_of(run(window, n, fee_ratio=0, tax_ratio=0))) rows.append(row) print(name, window, n, f"sharpe {row['full']['strategy']['sharpe']:.2f} base {base_alone[n]['full']['strategy']['sharpe']:.2f}", flush=True) wins, checks, base = three_checks(rows, center) checks["beats_base"] = base["beats_base"] checks["all"] = checks["all"] and checks["beats_base"] results[name] = {"x": xs, "y": ys, "center": center, "wins": wins, "checks": checks, "grid": rows} print(name, "wins", wins, "/", len(rows), checks, flush=True) # 6. 輸出 meta = { "run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "pandas_version": pd.__version__, "data_end": str(days[-1].date()), "start": START, "combo_start": COMBO_START, "short_fee": SHORT_FEE, "benchmark": "0050 含息買進持有（etl:adj_close，不含交易成本）", "sharpe": "日報酬平均 / 標準差 × √252，未扣無風險利率", } json.dump({"meta": meta, "base_alone": base_alone, "designs": results}, open("metrics.json", "w"), ensure_ascii=False, indent=1, default=str) table = pd.DataFrame(curves) table = table / table.bfill().iloc[0] table["0050"] = bench.reindex(table.index).ffill() table["0050"] = table["0050"] / table["0050"].dropna().iloc[0] month_id = table.index.to_period("M") table[np.append(month_id[1:] != month_id[:-1], True)].round(4).to_csv("data.csv", index_label="date") print("done", meta) if __name__ == "__main__": main()