""" 台股籌碼進階：千張大戶、股東人數、主力分點、地緣券商、借券賣出、外資持股、券資比 7 個策略的三項驗證（資料起點各自不同，皆跑到最新資料）。 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個策略 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入。 地緣券商要用證交所 OpenAPI 的分公司地址（公開資料，不需登入）。 輸出：metrics.json、data.csv（各策略中心參數的月底淨值）。 """ import csv import datetime import io import itertools import json import re import numpy as np import pandas as pd import requests import finlab from finlab import data from finlab.backtest import sim START = "2010-01-01" # 融資券、借券、外資持股 START_INVENTORY = "2016-11-01" # 集保戶股權分散表起點 START_BROKER = "2022-07-01" # 分點資料從這天起才含已下市股票 MIN_TURNOVER_VALUE = 2e7 # 20 日平均成交金額至少 2,000 萬元 MIN_MARGIN_LOTS = 500 # 券資比的分母：融資餘額至少 500 張 TRADING_DAYS = 252 TAIPEI_AREA = {"台北市", "新北市"} TWSE_BRANCHES = "https://openapi.twse.com.tw/v1/opendata/OpenData_BRK02" TWSE_HEAD_OFFICES = "https://openapi.twse.com.tw/v1/brokerService/brokerList" # --------------------------------------------------------------------------- # 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) volume = plain("price:成交股數").reindex(columns=stocks) turnover_value = plain("price:成交金額").reindex(columns=stocks) days = adj_close.loc[START:].index liquid = turnover_value.rolling(20).mean().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 flows_on_days(df): """每日買超張數對齊交易日；當天沒有紀錄就是 0，不沿用前一天。""" return df.reindex(index=days, columns=stocks).fillna(0) 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. 籌碼資料 # --------------------------------------------------------------------------- big_holder_pct = plain("etl:inventory:大於一千張佔比").reindex(columns=stocks) # 週資料，% holders = plain("etl:inventory:全部人數").reindex(columns=stocks) # 週資料，人 top15_buy = plain("etl:broker_transactions:top15_buy").reindex(columns=stocks) # 張 top15_sell = plain("etl:broker_transactions:top15_sell").reindex(columns=stocks) sbl_balance = plain("security_lending_sell:借券賣出餘額").reindex(columns=stocks) # 股 issued = plain("foreign_investors_shareholding:發行股數").reindex(columns=stocks) # 股 foreign_pct = plain("foreign_investors_shareholding:全體外資及陸資持股比率").reindex(columns=stocks) # % short_lots = plain("margin_transactions:融券今日餘額").reindex(columns=stocks) # 張 margin_lots = plain("margin_transactions:融資今日餘額").reindex(columns=stocks) # 張 market_cap = close.reindex(days) * on_days(issued) volume_lots = volume.reindex(days) / 1000 def county(address): """地址的縣市（臺→台）；抓不到回傳 None。""" text = re.sub(r"^[\d\s]+", "", str(address)).replace("臺", "台") return text[:3] if len(text) >= 3 and text[2] in "市縣" else None def broker_key(name): """分點名稱正規化：去空白、臺→台、總公司去掉「證券」（元大證券 → 元大）。""" return str(name).replace(" ", "").replace("臺", "台").removesuffix("證券") def twse_offices(url): """證交所券商名單 →（代號, 名稱, 地址）；OpenAPI 有時回 JSON、有時回 CSV，兩種都接受。""" text = requests.get(url, timeout=60).content.decode("utf-8-sig").strip() rows = json.loads(text) if text.startswith("[") else list(csv.DictReader(io.StringIO(text))) keys = ("證券商代號", "證券商名稱", "地址") if "地址" in rows[0] else ("Code", "Name", "Address") return [tuple(r[k] for k in keys) for r in rows] def local_broker_net_lots(): """每天每檔股票「總公司所在縣市的券商分點」買超張數（只看前 15 大買賣分點）。""" branches = twse_offices(TWSE_BRANCHES) head_offices = twse_offices(TWSE_HEAD_OFFICES) offices = branches + head_offices county_by_id = {code.strip(): county(address) for code, _, address in offices} county_by_name = {broker_key(name): county(address) for _, name, address in offices} info = pd.DataFrame(data.get("company_basic_info")) company_county = info.assign(stock_id=info["stock_id"].astype(str)).set_index("stock_id")["住址"].map(county) raw = pd.DataFrame(data.get("broker_transactions")) raw = raw[raw["date"] >= pd.Timestamp(START_BROKER) - pd.Timedelta(days=60)] raw = raw[raw["stock_id"].astype(str).isin(stocks)] sid = raw["stock_id"].astype(str) # 2025 年以前多數列沒有券商代號：先用代號對證交所名單，對不到再用分點名稱 by_id = raw["broker_id"].astype(str).str.strip().map(county_by_id) by_name = raw["broker"].map({c: county_by_name.get(broker_key(c)) for c in raw["broker"].cat.categories}) branch_county = by_id.fillna(by_name.astype(object)) local = branch_county.notna() & (branch_county.to_numpy() == sid.map(company_county).to_numpy()) net = (raw["buy"] - raw["sell"]).astype(float).where(local, 0.0) table = net.groupby([raw["date"], sid]).sum().unstack() outside = company_county[~company_county.isin(TAIPEI_AREA) & company_county.notna()].index traded = close.columns[close.loc[START_BROKER:].notna().any()] coverage = { "twse_branch_offices": len(branches), "twse_head_offices": len(head_offices), "traded_without_company_info": int((~traded.isin(info["stock_id"].astype(str))).sum()), "rows": int(len(raw)), "rows_with_known_branch_county": float(branch_county.notna().mean()), "local_rows": float(local.mean()), "companies_with_county": int(company_county.notna().sum()), "outside_taipei_area": len(outside), } return table, coverage, outside local_net, local_coverage, outside_taipei = local_broker_net_lots() # --------------------------------------------------------------------------- # 3. 七個策略：參數 → 每日持股（True＝持有，等權重） # --------------------------------------------------------------------------- def top_n(score, n, eligible): return score.where(eligible).rank(axis=1, ascending=False, method="first") <= n def s1_big_holder_rising(weeks, n): """千張大戶持股比例比 weeks 週前上升最多的 n 檔。""" change = on_days(big_holder_pct.diff(weeks)) return top_n(change, n, liquid & (change > 0)) def s2_holders_falling(weeks, n): """股東人數比 weeks 週前減少最多（百分比）的 n 檔。""" drop = on_days(-holders.pct_change(weeks)) return top_n(drop, n, liquid & (drop > 0)) def s3_main_force(lookback, n): """前 15 大分點買超（買進減賣出）占同期成交量比例最高的 n 檔。""" net = flows_on_days(top15_buy - top15_sell) concentration = net.rolling(lookback).sum() / volume_lots.rolling(lookback).sum() return top_n(concentration, n, liquid & (concentration > 0)) def s4_local_broker(lookback, n): """總公司不在雙北的公司，同縣市分點買超占同期成交量比例最高的 n 檔。""" net = flows_on_days(local_net) share = net.rolling(lookback).sum() / volume_lots.rolling(lookback).sum() in_scope = liquid.columns.isin(outside_taipei) return top_n(share, n, liquid & in_scope & (share > 0)) def s5_sbl_exclusion(lookback, exclude): """市值前 50 大，剔除借券賣出餘額占股本 lookback 日增加最多的 exclude 檔。""" base = top_n(market_cap, 50, liquid) rise = on_days((sbl_balance.fillna(0) / issued).diff(lookback)) worst = top_n(rise, exclude, base) return base & ~worst def s6_foreign_momentum(lookback, n): """外資持股比例比 lookback 個交易日前上升最多的 n 檔。""" change = on_days(foreign_pct.diff(lookback)) return top_n(change, n, liquid & (change > 0)) def s7_short_covering(lookback, n): """融券餘額比 lookback 日前減少（回補中）的股票裡，券資比最高的 n 檔。""" ratio = on_days(short_lots / margin_lots.where(margin_lots >= MIN_MARGIN_LOTS)) covering = on_days(short_lots.diff(lookback)) < 0 return top_n(ratio, n, liquid & covering & (ratio > 0)) STRATEGIES = { # 名稱 函式 參數網格 中心參數 起點 換股 "s1_big_holder_rising": (s1_big_holder_rising, {"weeks": [2, 4, 8], "n": [20, 30, 50]}, {"weeks": 4, "n": 30}, START_INVENTORY, "W"), "s2_holders_falling": (s2_holders_falling, {"weeks": [2, 4, 8], "n": [20, 30, 50]}, {"weeks": 4, "n": 30}, START_INVENTORY, "W"), "s3_main_force": (s3_main_force, {"lookback": [5, 10, 20], "n": [20, 30, 50]}, {"lookback": 10, "n": 30}, START_BROKER, "W"), "s4_local_broker": (s4_local_broker, {"lookback": [5, 10, 20], "n": [10, 20, 30]}, {"lookback": 10, "n": 20}, START_BROKER, "W"), "s5_sbl_exclusion": (s5_sbl_exclusion, {"lookback": [5, 20, 60], "exclude": [5, 10, 15]}, {"lookback": 20, "exclude": 10}, START, "M"), "s6_foreign_momentum": (s6_foreign_momentum, {"lookback": [20, 60, 120], "n": [20, 30, 50]}, {"lookback": 60, "n": 30}, START, "M"), "s7_short_covering": (s7_short_covering, {"lookback": [5, 10, 20], "n": [10, 20, 30]}, {"lookback": 10, "n": 20}, START, "W"), } def turnover_per_rebalance(position, freq): """每次換股平均換掉多少比例的部位（單邊）。""" held = position.resample("ME" if freq == "M" else "W-FRI").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, start, freq) in STRATEGIES.items(): rows = [] for values in itertools.product(*grid.values()): params = dict(zip(grid, values)) position = build(**params).loc[start:] report = sim(position, resample=freq, 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": turnover_per_rebalance(position, freq), **versus_0050(equity), } rows.append(row) if params == center: curves[name] = equity # 同一組持股不扣手續費與證交稅，量成本吃掉多少 gross = sim(position, resample=freq, upload=False, fee_ratio=0, tax_ratio=0).creturn.loc[first_day:] row["zero_cost"] = perf(gross) print(name, params, f"sharpe {row['full']['strategy']['sharpe']:.2f} vs 0050 {row['full']['bench']['sharpe']:.2f}", flush=True) center_row = next(r for r in rows if r["params"] == center) results[name] = { "grid": grid, "center": center, "start": start, "rebalance": freq, "wins": sum(r["full"]["win"] for r in rows), "center_halves_win": center_row["first_half"]["win"] and center_row["second_half"]["win"], "combos": rows, } # --------------------------------------------------------------------------- # 5. 訊號本身：月底分五組，看下個月報酬與 Rank IC # --------------------------------------------------------------------------- month_end = adj_close.resample("ME").last() monthly_return = month_end.pct_change() liquid_month = turnover_value.rolling(20).mean().resample("ME").last() > MIN_TURNOVER_VALUE cap_month = market_cap.resample("ME").last() 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, start, scope=None): """因子由低到高分五組，等權持有一個月；回傳各組年化、分組當月報酬、市值中位數、平均 Rank IC 與 IC 為正的月份比例。""" factor = at_month_end(factor).where(liquid_month) if scope is not None: factor = factor[[s for s in scope if s in factor.columns]] factor = factor.loc[start:] groups, ics, priors, caps = [], [], [], [] 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) if len(x) < 50: continue q = pd.qcut(x.rank(method="first"), 5, labels=False) groups.append(y.groupby(q).mean()) ics.append(x.corr(y, method="spearman")) priors.append(monthly_return.loc[d].reindex(x.index).groupby(q).mean().to_numpy()) caps.append((cap_month.loc[d].reindex(x.index) / 1e8).groupby(q).median().to_numpy()) table = pd.DataFrame(groups) ics = pd.Series(ics).dropna() return { "annual_by_quintile": [annualize(table[q]) for q in range(5)], "prior_month_return_by_quintile": [float(np.nanmean(p)) for p in zip(*priors)], "median_market_cap_100m_by_quintile": [float(np.nanmedian(c)) for c in zip(*caps)], "rank_ic_mean": float(ics.mean()), "rank_ic_t": float(ics.mean() / ics.std() * np.sqrt(len(ics))), "ic_positive_share": float((ics > 0).mean()), "months": len(table), "start": start, } def lots_share(net_lots, lookback): net = net_lots.reindex(columns=stocks).fillna(0) vol = (volume.reindex(net.index) / 1000) return net.rolling(lookback).sum() / vol.rolling(lookback).sum() signals = { "big_holder_change_4w": quintiles(big_holder_pct.diff(4), START_INVENTORY), "holders_drop_4w": quintiles(-holders.pct_change(4), START_INVENTORY), "main_force_10d": quintiles(lots_share(top15_buy - top15_sell, 10), START_BROKER), "local_broker_10d": quintiles(lots_share(local_net, 10), START_BROKER, scope=outside_taipei), "sbl_ratio_rise_20d": quintiles((sbl_balance.fillna(0) / issued).diff(20), START), "foreign_pct_change_60d": quintiles(foreign_pct.diff(60), START), "short_margin_ratio": quintiles(short_lots / margin_lots.where(margin_lots >= MIN_MARGIN_LOTS), START), "local_broker_coverage": local_coverage, } # 借券剔除法的歸因：市值前 50 大等權、不剔除，對 0050 的差距有多少來自等權本身 top50 = top_n(market_cap, 50, liquid).loc[START:] equity = sim(top50, resample="M", upload=False).creturn.loc[top50.index[top50.any(axis=1)][0]:] signals["top50_equal_weight_no_exclusion"] = versus_0050(equity) curves["top50_equal_weight"] = equity # 最新一期讀數：千張大戶與股東人數的分布，借券賣出餘額占股本、券資比的分布 latest = {} for label, df in [("big_holder_pct", big_holder_pct), ("holders", holders)]: last = df.iloc[-1].dropna() latest[label] = {"date": str(df.index[-1].date()), "companies": int(len(last)), "percentiles_25_50_75_90": last.quantile([0.25, 0.5, 0.75, 0.9]).round(2).tolist()} ratio_now = (short_lots / margin_lots.where(margin_lots >= MIN_MARGIN_LOTS)).iloc[-1].dropna() * 100 latest["short_margin_ratio_pct"] = {"date": str(short_lots.index[-1].date()), "companies": int(len(ratio_now)), "percentiles_50_75_90_99": ratio_now.quantile([0.5, 0.75, 0.9, 0.99]).round(2).tolist()} sbl_now = (sbl_balance / issued).iloc[-1].dropna() * 100 latest["sbl_pct_of_shares"] = {"date": str(sbl_balance.index[-1].date()), "companies": int(len(sbl_now)), "percentiles_50_75_90_99": sbl_now.quantile([0.5, 0.75, 0.9, 0.99]).round(3).tolist()} signals["latest"] = latest # --------------------------------------------------------------------------- # 6. 輸出 # --------------------------------------------------------------------------- meta = { "run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "data_end": str(days[-1].date()), "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.resample("ME").last().round(4).to_csv("data.csv", index_label="date") print("done", meta)