""" 台股散戶與警示訊號：融資餘額、個股融資、注意股、跳空缺口 5 個策略的三項驗證（2008 年起至最新資料） 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個策略 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 股票池：上市櫃普通股（四碼數字，含已下市，排除 ETF 與存託憑證），20 日平均成交金額至少 2,000 萬元。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入，約 3 分鐘。 輸出：metrics.json、data.csv（5 個策略原參數與 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 from finlab.dataframe import FinlabDataFrame START = "2008-01-01" MIN_TURNOVER_VALUE = 2e7 def plain(key): return pd.DataFrame(data.get(key)) # --------------------------------------------------------------------------- # 1. 共同資料：股票池、0050、績效計算 # --------------------------------------------------------------------------- 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", "91"))] days = adj_close.loc[START:].index col_of = {s: i for i, s in enumerate(stocks)} def on_grid(df): return df.reindex(index=days, columns=stocks) turnover_value = plain("price:成交金額").reindex(columns=stocks).rolling(20).mean() liquid = turnover_value.reindex(days) > MIN_TURNOVER_VALUE 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 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() / ret.std() * math.sqrt(252)), 6), "mdd": round(float((curve / curve.cummax() - 1).min()), 6), "vol": round(float(ret.std() * math.sqrt(252)), 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 versus_0050(curve): s = stats(active(curve)) b = stats(bench.loc[s["start"]:s["end"]]) return {"strategy": s, "bench": b, "win": s["sharpe"] > b["sharpe"]} def halves(curve): curve = active(curve) mid = curve.index[0] + (curve.index[-1] - curve.index[0]) / 2 return [versus_0050(curve.loc[lo:hi]) for lo, hi in ((curve.index[0], mid), (mid, curve.index[-1]))] def trade_stats(report): trades = report.get_trades() curve = active(report.creturn) years = (curve.index[-1] - curve.index[0]).days / 365.25 held_days = (pd.to_datetime(trades["exit_date"]).fillna(curve.index[-1]) - pd.to_datetime(trades["entry_date"])).dt.days holdings = report.signal.sum(axis=1) holdings = holdings.loc[holdings.gt(0).idxmax():] return { "trades": int(len(trades)), "trades_per_year": round(len(trades) / years, 1), "avg_hold_calendar_days": round(float(held_days.mean()), 1), "avg_holdings": round(float(holdings[holdings > 0].mean()), 1), "share_of_days_holding": round(float((holdings > 0).mean()), 4), "win_rate": round(float((trades["return"] > 0).mean()), 4), "mean_return": round(float(trades["return"].mean()), 6), } def backtest(position, cost, **kw): position = position.reindex(index=days).fillna(False).astype(bool) position = position.loc[:, position.any()] fees = {} if cost == "default" else {"fee_ratio": 0, "tax_ratio": 0} report = sim(position, upload=False, **fees, **kw) report.signal = position # 送進 sim 的每日持有表，用來算持股檔數與持有天數比例 return report def windows(rows): """rows：（股票, 開始位置, 結束位置〔含〕）→ 每天的持有表。""" held = np.zeros((len(days), len(stocks)), dtype=bool) for stock_id, lo, hi in rows: if stock_id in col_of and lo <= hi and lo < len(days): held[max(lo, 0):min(hi, len(days) - 1) + 1, col_of[stock_id]] = True return pd.DataFrame(held, index=days, columns=stocks) # --------------------------------------------------------------------------- # 2. 五個策略：每個函式接兩個參數與成本設定，回傳 sim 報告 # --------------------------------------------------------------------------- margin_total = plain("margin_balance:融資券總餘額") loan = (margin_total["上市融資交易金額"] + margin_total["上櫃融資交易金額"]).dropna() def margin_balance(lookback, threshold, cost): """大盤融資餘額 lookback 日增幅超過 threshold 就出場，其餘時間持有 0050。""" growth = loan / loan.shift(lookback) - 1 hold = (growth.dropna() < threshold).reindex(days).ffill().fillna(False).astype(bool) return backtest(pd.DataFrame({"0050": hold}), cost) margin_shares = on_grid(plain("margin_transactions:融資今日餘額")) margin_usage_raw = on_grid(plain("margin_transactions:融資使用率")) def margin_decrease(lookback, n, cost): """融資餘額 lookback 日減幅最大的 n 檔，每月底換股；lookback 日前融資至少 200 張。""" change = margin_shares / margin_shares.shift(lookback) - 1 eligible = liquid & (margin_shares.shift(lookback) >= 200) position = FinlabDataFrame(change.where(eligible)).is_smallest(n) return backtest(position, cost, resample="M") def margin_usage(window, n, cost): """融資使用率（window 日平均）最低的 n 檔，每月底換股；只看使用率大於 0 的股票。""" usage = margin_usage_raw.rolling(window).mean() position = FinlabDataFrame(usage.where(liquid & (usage > 0))).is_smallest(n) return backtest(position, cost, resample="M") attention = plain("trading_attention") attention["stock_id"] = attention["stock_id"].astype(str) attention = attention[attention["stock_id"].isin(stocks) & (attention["date"] >= START)] attention = attention.sort_values(["stock_id", "date"]).reset_index(drop=True) attention["pos"] = days.searchsorted(attention["date"].values) attention["rise"] = attention["注意交易資訊"].astype(str).str.contains("漲幅") attention["first"] = attention.groupby("stock_id")["pos"].diff().fillna(999) > 20 attention["liquid"] = [p < len(days) and bool(liquid.iat[p, col_of[s]]) for s, p in zip(attention["stock_id"], attention["pos"])] chase_events = attention[attention["rise"] & attention["first"] & attention["liquid"]] def attention_chase(hold_days, delay, cost): """漲幅觸發、前 20 個交易日沒被列過的注意股，公告後第 delay 天收盤確認、隔天收盤買進，持有 hold_days 天。""" rows = [(s, p + delay, p + delay + hold_days - 1) for s, p in zip(chase_events["stock_id"], chase_events["pos"])] return backtest(windows(rows), cost) adj_high = on_grid(plain("etl:adj_high")) adj_low = on_grid(plain("etl:adj_low")) gap_level = adj_high.shift(1) # 前一天最高價 gap_day = (adj_low > gap_level) & liquid # 當天最低價都比前一天最高價高：向上跳空 def gap_up(confirm_days, hold_days, cost): """向上跳空後 confirm_days 天最低價都沒回到前一天最高價，第 confirm_days 天收盤確認，持有 hold_days 天。""" future_low = adj_low[::-1].rolling(confirm_days + 1, min_periods=confirm_days + 1).min()[::-1] confirmed = (gap_day & (future_low > gap_level)).shift(confirm_days).fillna(False).astype(bool) position = confirmed.astype(float).rolling(hold_days, min_periods=1).max().fillna(0).astype(bool) & liquid return backtest(position, cost) STRATEGIES = { "margin-balance": dict(fn=margin_balance, title="大盤融資餘額增幅過大就出場（0050 擇時）", x=("融資餘額回看天數", [10, 20, 40]), y=("增幅門檻", [0.03, 0.05, 0.08]), base=(20, 0.05)), "margin-decrease": dict(fn=margin_decrease, title="個股融資減少最多", x=("融資變化回看天數", [10, 20, 40]), y=("持股檔數", [10, 20, 30]), base=(20, 20)), "margin-usage": dict(fn=margin_usage, title="個股融資使用率最低", x=("融資使用率平均天數", [1, 5, 20]), y=("持股檔數", [10, 20, 30]), base=(5, 20)), "attention-chase": dict(fn=attention_chase, title="注意股公布後追進", x=("持有天數", [5, 10, 20]), y=("延後進場天數", [0, 2, 5]), base=(10, 0)), "gap-up": dict(fn=gap_up, title="向上跳空不補", x=("確認不補天數", [3, 5, 10]), y=("持有天數", [10, 20, 40]), base=(5, 20)), } # --------------------------------------------------------------------------- # 3. 三項驗證 # --------------------------------------------------------------------------- results, curves = {}, {} for key, cfg in STRATEGIES.items(): grid = [] for xv in cfg["x"][1]: for yv in cfg["y"][1]: report = cfg["fn"](xv, yv, "default") trades = trade_stats(report) cell = versus_0050(report.creturn) cell.update({"x": xv, "y": yv, "share_of_days_holding": trades["share_of_days_holding"]}) grid.append(cell) if (xv, yv) == cfg["base"]: base_report, base_cell, base_trades = report, cell, trades zero_cost = versus_0050(cfg["fn"](*cfg["base"], "zero").creturn) two_halves = halves(base_report.creturn) wins = sum(c["win"] for c in grid) results[key] = { "title": cfg["title"], "x": {"name": cfg["x"][0], "values": cfg["x"][1]}, "y": {"name": cfg["y"][0], "values": cfg["y"][1]}, "base": {"x": cfg["base"][0], "y": cfg["base"][1]}, "grid": grid, "grid_wins": wins, "base_default_cost": base_cell, "base_zero_cost": zero_cost, "halves": two_halves, "trades": base_trades, "pass": {"neighbors": wins >= 7, "halves": all(h["win"] for h in two_halves), "cost": base_cell["win"]}, } curves[key] = active(base_report.creturn) print(key, "grid", wins, "/ 9", results[key]["pass"], flush=True) # --------------------------------------------------------------------------- # 4. 描述統計 # --------------------------------------------------------------------------- facts = {} ret = on_grid(adj_close).pct_change(fill_method=None) market = ret.where(liquid).mean(axis=1) # 股票池等權平均日報酬 excess = ret.sub(market, axis=0).fillna(0).values # 4a. 融資餘額：歷年年底金額、占上市櫃總市值的比例（市值資料從 2013 年起） market_value = plain("etl:market_value").sum(axis=1) ratio = (loan / market_value.reindex(loan.index)).dropna() loan_year = loan.resample("YE").last() ratio_year = ratio.resample("YE").last() facts["margin_by_year"] = {str(d.year): {"loan_100m": round(float(v) / 1e8, 1), "pct_of_market_value": None if pd.isna(ratio_year.get(d)) else round(float(ratio_year[d]), 6)} for d, v in loan_year.items()} facts["margin_latest"] = {"date": str(loan.index[-1].date()), "loan_100m": round(float(loan.iloc[-1]) / 1e8, 1), "pct_of_market_value": round(float(loan.iloc[-1] / market_value.reindex(loan.index).iloc[-1]), 6)} before = loan.loc[:str(loan.index[-1].year - 1)] this_year = loan.loc[str(loan.index[-1].year)] facts["margin_extremes"] = { "high_before_this_year": {"date": str(before.idxmax().date()), "loan_100m": round(float(before.max()) / 1e8, 1)}, "high_this_year": {"date": str(this_year.idxmax().date()), "loan_100m": round(float(this_year.max()) / 1e8, 1)}, "first_day_this_year": {"date": str(this_year.index[0].date()), "loan_100m": round(float(this_year.iloc[0]) / 1e8, 1)}, "pct_of_market_value": {"since": str(ratio.index[0].date()), "median": round(float(ratio.median()), 6), "max": round(float(ratio.max()), 6), "max_date": str(ratio.idxmax().date()), "min": round(float(ratio.min()), 6), "min_date": str(ratio.idxmin().date())}, } # 4b. 融資餘額 20 日增幅分五組，0050 之後 20、60 個交易日的報酬（訊號隔天收盤起算） growth = (loan / loan.shift(20) - 1).reindex(bench.index).dropna() fwd = {h: bench.shift(-(h + 1)) / bench.shift(-1) - 1 for h in (20, 60)} past20 = bench / bench.shift(20) - 1 # 同一段 20 個交易日 0050 的漲跌 table = pd.DataFrame({"growth": growth, "past20": past20.reindex(growth.index), "f20": fwd[20].reindex(growth.index), "f60": fwd[60].reindex(growth.index)}).dropna() table["q"] = pd.qcut(table["growth"], 5, labels=False) + 1 facts["margin_growth_quintiles"] = { str(q): {"growth_min": round(float(g["growth"].min()), 6), "growth_max": round(float(g["growth"].max()), 6), "past20": round(float(g["past20"].mean()), 6), "fwd20": round(float(g["f20"].mean()), 6), "fwd60": round(float(g["f60"].mean()), 6), "days": int(len(g))} for q, g in table.groupby("q")} # 4c. 融資使用率分五組，每月底換股、等權持有 usage = margin_usage_raw.rolling(5).mean() usage = usage.where(liquid & (usage > 0)) rank = usage.rank(axis=1, pct=True) facts["margin_usage_quintiles"] = {} for q in range(1, 6): report = backtest((rank > (q - 1) / 5) & (rank <= q / 5), "default", resample="M") facts["margin_usage_quintiles"][str(q)] = versus_0050(report.creturn)["strategy"] latest_usage = margin_usage_raw.iloc[-1] latest_usage = latest_usage[latest_usage.notna()] facts["margin_usage_latest"] = {"date": str(days[-1].date()), "stocks": int(len(latest_usage)), "median": round(float(latest_usage.median()), 4), "p75": round(float(latest_usage.quantile(0.75)), 4)} # 4d. 注意股公告後的累積超額報酬（相對股票池等權，公告日收盤後才知道，從隔天起算） def event_path(events, horizon=20): paths = [excess[p + 1:p + 1 + horizon, col_of[s]] for s, p in zip(events["stock_id"], events["pos"]) if p + horizon < len(days)] paths = np.cumsum(np.array(paths), axis=1) return {"n": int(len(paths)), "cum_excess": [round(float(v), 6) for v in paths.mean(axis=0)], "share_positive_10d": round(float((paths[:, 9] > 0).mean()), 4)} liquid_events = attention[attention["liquid"]] facts["attention_paths"] = { "rise_first": event_path(liquid_events[liquid_events["rise"] & liquid_events["first"]]), "rise_all": event_path(liquid_events[liquid_events["rise"]]), "other_triggers": event_path(liquid_events[~liquid_events["rise"]]), } # 4e. 向上跳空多久會回補：之後 k 天內最低價回到前一天最高價就算補 gaps = gap_day.values # 前一天最高價缺值時 gap_day 已是 False low = adj_low.values level = gap_level.values fill = {} for k in (1, 3, 5, 10, 20, 60): future_min = pd.DataFrame(low).shift(-1)[::-1].rolling(k, min_periods=k).min()[::-1].values valid = gaps & ~np.isnan(future_min) fill[str(k)] = round(float((future_min[valid] <= level[valid]).mean()), 4) facts["gap_fill_rate"] = {"gaps": int(gaps.sum()), "filled_within_days": fill} # 「三日判定強弱」：跳空後 3 天內有沒有補，接下來 20 個交易日的超額報酬 three_day_min = pd.DataFrame(low)[::-1].rolling(4, min_periods=4).min()[::-1].values # 跳空當天到第 3 天 cum = np.vstack([np.zeros(len(stocks)), np.cumsum(excess, axis=0)]) rows, cols = np.nonzero(gaps) keep = rows + 24 < len(days) rows, cols = rows[keep], cols[keep] after = cum[rows + 25, cols] - cum[rows + 5, cols] # 第 4 天收盤起算 20 個交易日（第 3 天收盤確認、隔天收盤進場） unfilled = three_day_min[rows, cols] > level[rows, cols] facts["gap_three_day"] = { "unfilled": {"n": int(unfilled.sum()), "excess_20d": round(float(after[unfilled].mean()), 6)}, "filled": {"n": int((~unfilled).sum()), "excess_20d": round(float(after[~unfilled].mean()), 6)}, } # --------------------------------------------------------------------------- # 5. 輸出 # --------------------------------------------------------------------------- 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, "facts": facts}, 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) table["0050含息"] = table["0050含息"] / table["0050含息"].dropna().iloc[0] table.round(6).to_csv("data.csv", index_label="date") print("done", meta)