""" 台股董監與內部人訊號：7 個策略的三項驗證（2008-01 起至最新資料） finlab.finance/blog/director-shareholding-insider-signals-backtest 的數字全部來自這支程式。 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個策略 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入。 輸出：metrics.json、data.csv（各策略代表參數的淨值）、report_strategy.html。 """ 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 = "2008-01-01" MIN_TURNOVER_VALUE = 2e7 # 20 日平均成交金額至少 2,000 萬元 TRADING_DAYS = 252 MARKET_SALE = "一般交易|鉅額|盤後定價" # 申報轉讓方式中屬於「在市場賣出」的類別 # --------------------------------------------------------------------------- # 1. 股票池、價格與 0050 含息基準 # --------------------------------------------------------------------------- def plain(key): """取資料並轉成一般 DataFrame；月資料的 index 已是可取得日（月底）。""" df = pd.DataFrame(data.get(key)) df.index = pd.to_datetime(df.index) return df adj_close = plain("etl:adj_close") bench = adj_close["0050"].dropna() stocks = [s for s in adj_close.columns if len(s) == 4 and s.isdigit() and not s.startswith("0")] adj_close = adj_close[stocks] close = plain("price:收盤價").reindex(columns=stocks) turnover_value = plain("price:成交金額").reindex(columns=stocks) days = adj_close.loc[START:].index avg_turnover_value = turnover_value.rolling(20).mean() liquid = avg_turnover_value.reindex(days) > MIN_TURNOVER_VALUE def on_days(df): """把月資料展開到每個交易日：每天只看得到當天以前已公布的最新一期。""" df = df.reindex(columns=stocks) return df.reindex(df.index.union(days)).ffill().reindex(days) def perf(equity): equity = equity.dropna() ret = equity.pct_change().dropna() years = (equity.index[-1] - equity.index[0]).days / 365.25 return { "cagr": float((equity.iloc[-1] / equity.iloc[0]) ** (1 / years) - 1), "sharpe": float(ret.mean() / ret.std() * np.sqrt(TRADING_DAYS)), "mdd": float((equity / equity.cummax() - 1).min()), "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 # --------------------------------------------------------------------------- # 2. 董監與內部人資料（月資料，index 為可取得日） # --------------------------------------------------------------------------- board_pct = plain("internal_equity_changes:董監持有股數占比").reindex(columns=stocks) board_inc = plain("internal_equity_changes:董監增加股數").reindex(columns=stocks) board_dec = plain("internal_equity_changes:董監減少股數").reindex(columns=stocks) issued = plain("internal_equity_changes:發行股數").reindex(columns=stocks) major_holders = plain("internal_equity_changes:百分之十以上大股東持有股數").reindex(columns=stocks).fillna(0) pledge_pct = plain("internal_equity_pledge:董監設質股數占比").reindex(columns=stocks) shortfall = plain("internal_equity_insufficient:全體董事(不包含獨立董事)不足股數").reindex(columns=stocks) market_cap = close.reindex(days) * on_days(issued) # 內部人申報轉讓：只留在市場賣出的申報，換算成「申報後第一個交易日」的事件表 transfers = pd.DataFrame(data.get("insider_shareholding_transfer_declaration")) transfers = transfers[transfers["預定轉讓方式及股數_轉讓方式"].astype(str).str.contains(MARKET_SALE)] transfers["stock_id"] = transfers["stock_id"].astype(str) transfers = transfers[transfers["stock_id"].isin(stocks) & (transfers["date"] >= days[0]) & (transfers["date"] < days[-1])] transfers["event_day"] = days[np.searchsorted(days, transfers["date"], side="right").clip(0, len(days) - 1)] sale_events = pd.crosstab(transfers["event_day"], transfers["stock_id"]).reindex(index=days, columns=stocks, fill_value=0) # 股東常會的董監改選紀錄：預估下次改選年＝上次改選年＋3（董監任期 3 年） meetings = pd.DataFrame(data.get("shareholders_meeting")) meetings = meetings[meetings["股東會類別"].astype(str).str.contains("常會")] meetings["stock_id"] = meetings["stock_id"].astype(str) elections = meetings[meetings["是否改選董監"].astype(str) == "是"] election_years = elections.groupby("stock_id")["date"].apply(lambda d: sorted(set(d.dt.year))) def expected_election(year): """只用 year 以前的改選紀錄，預估 year 會改選董監的公司。""" names = [] for sid, years in election_years.items(): past = [y for y in years if y < year] if past and past[-1] + 3 == year and sid in stocks: names.append(sid) return names # --------------------------------------------------------------------------- # 3. 七個策略：參數 → 每日持股（True＝持有，等權重） # --------------------------------------------------------------------------- def top_n(score, n, eligible): return score.where(eligible).rank(axis=1, ascending=False, method="first") <= n def largest_caps(n): return top_n(market_cap, n, liquid) def s1_board_high(n, cap_rank): """董監持股比例最高的 n 檔（限市值前 cap_rank 大）。""" return top_n(on_days(board_pct), n, largest_caps(cap_rank)) def s2_board_rising(months, n): """董監持股比例比 months 個月前上升最多的 n 檔。""" change = on_days(board_pct.diff(months)) return top_n(change, n, liquid & (change > 0)) def s3_low_pledge(n, max_pledge): """市值前 n 大，排除董監設質比例高於 max_pledge% 的公司。""" return largest_caps(n) & (on_days(pledge_pct).fillna(0) <= max_pledge) def s4_no_insider_sale(n, lookback): """市值前 n 大，排除近 lookback 個交易日有內部人申報在市場賣出的公司。""" recent_sale = sale_events.rolling(lookback, min_periods=1).sum() > 0 return largest_caps(n) & ~recent_sale def s5_insider_buying(months, n): """近 months 個月董監淨增加股數＋10% 大股東增加股數（占發行股數）最高的 n 檔。""" board_net = (board_inc - board_dec).rolling(months).sum() / issued major_net = major_holders.diff(months) / issued score = on_days(board_net.fillna(0) + major_net.fillna(0)) return top_n(score, n, liquid & (score > 0)) def s6_election_year(entry_month, max_board_pct): """預估今年改選董監的公司：前一年 entry_month 月底換股時買進，當年 3 月底換股時賣出 （月換股在下一個交易日成交，所以 4 月初出場，在 4 月中停止過戶之前），其餘時間空手。""" position = pd.DataFrame(False, index=days, columns=stocks) board = on_days(board_pct) for year in range(days[0].year, days[-1].year + 1): names = expected_election(year) entry = pd.Timestamp(year - 1, entry_month, 1) + pd.offsets.MonthEnd(0) last_hold = pd.Timestamp(year, 2, 1) + pd.offsets.MonthEnd(0) # 2 月底仍持有，3 月底換股時賣出 window = (days > entry - pd.Timedelta(days=7)) & (days <= last_hold) if names and window.any(): eligible = liquid.loc[window] & (board.loc[window] <= max_board_pct) position.loc[window] = eligible & eligible.columns.isin(names) return position def s7_board_shortfall(hold_months, min_value): """全體董事持股成數不足的公司，被列入後持有 hold_months 個月。""" flagged = (shortfall > 0).astype(float).rolling(hold_months, min_periods=1).max() > 0 tradable = avg_turnover_value.reindex(days) > min_value return on_days(flagged.astype(float)).fillna(0).astype(bool) & tradable STRATEGIES = { "s1_board_high": (s1_board_high, {"n": [20, 30, 50], "cap_rank": [300, 500, 1000]}, {"n": 30, "cap_rank": 500}), "s2_board_rising": (s2_board_rising, {"months": [1, 3, 6], "n": [20, 30, 50]}, {"months": 3, "n": 30}), "s3_low_pledge": (s3_low_pledge, {"n": [50, 100, 200], "max_pledge": [10, 20, 30]}, {"n": 100, "max_pledge": 20}), "s4_no_insider_sale": (s4_no_insider_sale, {"n": [50, 100, 200], "lookback": [60, 120, 250]}, {"n": 100, "lookback": 120}), "s5_insider_buying": (s5_insider_buying, {"months": [1, 3, 6], "n": [20, 30, 50]}, {"months": 3, "n": 30}), "s6_election_year": (s6_election_year, {"entry_month": [10, 11, 12], "max_board_pct": [15, 25, 100]}, {"entry_month": 11, "max_board_pct": 25}), "s7_board_shortfall": (s7_board_shortfall, {"hold_months": [1, 3, 6], "min_value": [5e6, 1e7, 2e7]}, {"hold_months": 3, "min_value": 1e7}), } def monthly_turnover(position): """每月換股時，平均有多少比例的部位被換掉（單邊）。""" held = position.resample("ME").last() weights = held.div(held.sum(axis=1).replace(0, np.nan), axis=0).fillna(0) return float(weights.diff().abs().sum(axis=1).iloc[1:].mean() / 2) # --------------------------------------------------------------------------- # 4. 跑參數網格 # --------------------------------------------------------------------------- results, curves = {}, {} for name, (build, grid, center) in STRATEGIES.items(): rows = [] for values in itertools.product(*grid.values()): params = dict(zip(grid, values)) position = build(**params) report = sim(position, resample="M", upload=False) first_day = position.index[position.any(axis=1)][0] equity = report.creturn.loc[first_day:] row = { "params": params, "avg_holdings": float(position[position.any(axis=1)].sum(axis=1).mean()), "turnover": monthly_turnover(position), **versus_0050(equity), } rows.append(row) if params == center: curves[name] = equity if name == "s1_board_high": report.to_html("report_strategy.html") print(name, params, f"sharpe {row['full']['strategy']['sharpe']:.2f} vs 0050 {row['full']['bench']['sharpe']:.2f}", flush=True) results[name] = {"grid": grid, "center": center, "wins": sum(r["full"]["win"] for r in rows), "combos": rows} # --------------------------------------------------------------------------- # 5. 訊號本身：分組報酬與事件研究 # --------------------------------------------------------------------------- month_end = adj_close.resample("ME").last() monthly_return = month_end.pct_change() liquid_month = avg_turnover_value.resample("ME").last() > MIN_TURNOVER_VALUE def at_month_end(df): return df.reindex(df.index.union(month_end.index)).ffill().reindex(index=month_end.index, columns=stocks) def annualize(monthly_returns): r = pd.Series(monthly_returns).fillna(0) return float((1 + r).prod() ** (12 / len(r)) - 1) def quintiles(factor): factor = at_month_end(factor).where(liquid_month).loc[START:] groups, ics, cuts, zeros = [], [], [], [] for d, nxt in zip(factor.index[:-1], factor.index[1:]): x = factor.loc[d].dropna() y = monthly_return.loc[nxt].reindex(x.index).dropna() x = x.reindex(y.index) q = pd.qcut(x.rank(method="first"), 5, labels=False) groups.append(y.groupby(q).mean()) ics.append(x.corr(y, method="spearman")) cuts.append(x.quantile([0.2, 0.4, 0.6, 0.8]).to_numpy()) zeros.append(float((x == 0).mean())) table = pd.DataFrame(groups) return { "annual_by_quintile": [annualize(table[q]) for q in range(5)], "rank_ic_mean": float(np.mean(ics)), "months": len(table), "median_cutoffs_20_40_60_80": np.median(cuts, axis=0).round(2).tolist(), "avg_share_of_zero": float(np.mean(zeros)), } def sign_groups(factor): factor = at_month_end(factor).where(liquid_month).loc[START:] out = {"down": [], "flat": [], "up": [], "n_down": [], "n_up": []} for d, nxt in zip(factor.index[:-1], factor.index[1:]): x = factor.loc[d].dropna() y = monthly_return.loc[nxt].reindex(x.index) out["down"].append(y[x < 0].mean()); out["flat"].append(y[x == 0].mean()); out["up"].append(y[x > 0].mean()) out["n_down"].append(int((x < 0).sum())); out["n_up"].append(int((x > 0).sum())) return { "annual_down": annualize(out["down"]), "annual_flat": annualize(out["flat"]), "annual_up": annualize(out["up"]), "avg_count_down": float(np.mean(out["n_down"])), "avg_count_up": float(np.mean(out["n_up"])), } signals = { "board_pct_quintiles": quintiles(board_pct), "pledge_quintiles": quintiles(pledge_pct), "major_holder_change_3m": sign_groups(major_holders.diff(3) / issued), "board_holding_change_1m": sign_groups(board_pct.diff(1)), "board_pct_latest": { "date": str(board_pct.index[-1].date()), "companies": int(board_pct.iloc[-1].notna().sum()), "percentiles_25_50_75_90": board_pct.iloc[-1].dropna().quantile([0.25, 0.5, 0.75, 0.9]).round(2).tolist(), "pledge_over_20_share": float((pledge_pct.iloc[-1].dropna() > 20).mean()), }, "all_liquid_annual": annualize([monthly_return.loc[nxt][liquid_month.loc[d]].mean() for d, nxt in zip(liquid_month.loc[START:].index[:-1], liquid_month.loc[START:].index[1:])]), } # 增持排名前段的公司：淨增加幅度多大、多少主要來自 10% 大股東、多少同期股本也增加 center = STRATEGIES["s5_insider_buying"][2] picks = s5_insider_buying(**center).resample("ME").last() def picked(df): return on_days(df.fillna(0)).resample("ME").last().where(picks).stack().dropna() board_part = picked((board_inc - board_dec).rolling(center["months"]).sum() / issued) major_part = picked(major_holders.diff(center["months"]) / issued) signals["s5_picks"] = { "median_increase_of_shares": float((board_part + major_part).median()), "share_mainly_major_holders": float((major_part > board_part).mean()), "share_issued_up_over_5pct": float((picked(issued.pct_change(center["months"])) > 0.05).mean()), } # 排除條件的增量：市值前 100 大等權，不排除 vs 同時排除四種內部人警訊 base = largest_caps(100) warning = ( (on_days(pledge_pct).fillna(0) > 20) | (sale_events.rolling(120, min_periods=1).sum() > 0) | (on_days(major_holders.diff(3)) < 0) | on_days((shortfall > 0).astype(float)).fillna(0).astype(bool) ) signals["large_cap_filters"] = {} for label, position in [("top100_no_filter", base), ("top100_exclude_all_warnings", base & ~warning)]: equity = sim(position, resample="M", upload=False).creturn.loc[position.index[position.any(axis=1)][0]:] signals["large_cap_filters"][label] = {"avg_holdings": float(position.sum(axis=1).mean()), **versus_0050(equity)} # 申報轉讓事件研究：申報後第一個交易日收盤進場，對同日等權市場的超額報酬 daily_ret = adj_close.reindex(days).pct_change() market = daily_ret.where(liquid).mean(axis=1) growth = (1 + daily_ret.fillna(0)).cumprod() market_growth = (1 + market.fillna(0)).cumprod() HORIZON = 120 events = transfers.groupby(["stock_id", "event_day"])["預定轉讓總股數_自有持股"].sum().reset_index() paths, sizes = [], [] for sid, day, shares in events.itertuples(index=False): i = days.get_loc(day) if i + HORIZON >= len(days) or not liquid.iat[i, liquid.columns.get_loc(sid)]: continue window = days[i:i + HORIZON + 1] stock_path = growth.loc[window, sid] / growth.at[day, sid] market_path = market_growth.loc[window] / market_growth.at[day] paths.append((stock_path - market_path).to_numpy()) total = issued[sid].asof(day) sizes.append(shares / total if total and total > 0 else np.nan) paths = np.array(paths) sizes = pd.Series(sizes) size_q = pd.qcut(sizes.rank(method="first"), 4, labels=False) signals["insider_sale_event_study"] = { "events": int(len(paths)), "mean_car": {h: float(np.nanmean(paths[:, h])) for h in (5, 20, 60, 120)}, "median_car": {h: float(np.nanmedian(paths[:, h])) for h in (5, 20, 60, 120)}, "share_below_market_120": float(np.nanmean(paths[:, 120] < 0)), "car_path_by_size_quartile": {int(q): np.nanmean(paths[size_q.to_numpy() == q], axis=0).round(5).tolist() for q in range(4)}, "size_quartile_median_pct": {int(q): float(sizes[size_q == q].median()) for q in range(4)}, } # 改選年：預估改選公司 vs 其他公司，前一年 12 月底到當年 4 月 15 日 rows = [] for year in range(2009, days[-1].year + 1): s0 = days[days <= pd.Timestamp(year - 1, 12, 31)][-1] s1 = days[days <= pd.Timestamp(year, 4, 15)][-1] if s1 <= s0: continue eligible = liquid.loc[s0] names = set(expected_election(year)) & set(eligible[eligible].index) others = set(eligible[eligible].index) - names move = adj_close.loc[s1] / adj_close.loc[s0] - 1 rows.append({"year": year, "n": len(names), "election": float(move[list(names)].mean()), "others": float(move[list(others)].mean())}) actual = sum(len(set(expected_election(y)) & set(elections[elections["date"].dt.year == y]["stock_id"])) for y in range(2009, days[-1].year + 1)) predicted = sum(len(expected_election(y)) for y in range(2009, days[-1].year + 1)) signals["election_year"] = {"by_year": rows, "prediction_hit_rate": actual / predicted} # --------------------------------------------------------------------------- # 6. 輸出 # --------------------------------------------------------------------------- meta = { "run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "data_end": str(days[-1].date()), "start": START, "liquidity_filter_twd": MIN_TURNOVER_VALUE, "benchmark": "0050 含息買進持有（etl:adj_close，不含交易成本）", "sharpe": "日報酬平均 / 標準差 × √252，未扣無風險利率", } json.dump({"meta": meta, "strategies": results, "signals": signals}, open("metrics.json", "w"), ensure_ascii=False, indent=1, default=str) table = pd.DataFrame(curves) table["0050"] = bench.reindex(table.index).ffill() table["0050"] = table["0050"] / table["0050"].dropna().iloc[0] # 和策略一樣從 1 起算 table.resample("ME").last().round(4).to_csv("data.csv", index_label="date") print("done", meta)