""" 營收更正、訴訟重訊、鉅亨新聞熱度、品質分數：12 個設計與 2 個對照的三項驗證與事件統計 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個設計 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 時點：訊號日收盤後才知道，sim() 在隔一個交易日收盤成交。流動性＝訊號日 20 日均成交金額 ≥ 2,000 萬。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 每個訊號的定義與參數在看結果前就寫定，之後沒有改過。 時點查證： - 訴訟與新聞的 key_date 是回補時間，改用 date（訴訟是發言時間，新聞是刊登日）。 - monthly_revenue_revised 只有 2025 年以後看得到更正，營收更正改用新月報的「上月營收」「去年當月營收」跟當初公布值比。 - 品質分數的日期是各季財報法定期限。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入，約 1 分鐘。 輸出：metrics.json（全部結果）、data.csv（各設計原參數、0050 含息與等權流動股的每日淨值）。 """ import datetime import json import math import re import numpy as np import pandas as pd import finlab from finlab import data from finlab.backtest import sim COSTS = {"default": {}, "zero": {"fee_ratio": 0, "tax_ratio": 0}} 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) amount_20d = amount.rolling(20, min_periods=20).mean() adj_frame = adj_close.reindex(columns=STOCKS).ffill() adj = adj_frame.to_numpy() adj0050 = adj_close["0050"].ffill().to_numpy() def liquid_mask(floor=2e7): """20 日均成交金額 ≥ floor（預設 2,000 萬）。""" return (amount_20d >= floor).fillna(False).to_numpy() liquid = liquid_mask() def first_day_on_or_after(dates): """每個日期之後（含當天）的第一個交易日位置。""" return days.searchsorted(pd.DatetimeIndex(pd.to_datetime(dates)).normalize(), side="left") def month_ends(): s = pd.Series(np.arange(ND), index=days) return s.groupby([s.index.year, s.index.month]).max().tolist() 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) if years > 0 else None, "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 if len(moved) == 0 else curve.iloc[max(moved[0] - 1, 0):] def compare(curve): s = stats(active(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 out = [] for part in (curve.loc[:mid], curve.loc[mid:]): s = stats(part) b = stats(bench.loc[s["start"]:s["end"]]) out.append({"strategy": s, "bench": b, "win": bool(s["sharpe"] > b["sharpe"])}) return out def backtest(table, cost, events=None): frame = pd.DataFrame(table, index=days, columns=STOCKS) report = sim(frame.loc[:, frame.any()], upload=False, **COSTS[cost]) return report, frame, events def event_sim(rows, cost): """rows：(股票, b, e)＝b 收盤買、e 收盤賣。sim 在持有表為真的隔天收盤成交，所以持有表是 [b−1, e−2]。""" table = np.zeros((ND, len(STOCKS)), dtype=bool) count = 0 for s, b, e in rows: lo, hi = max(int(b) - 1, 0), min(int(e) - 2, ND - 1) if s in COL and hi >= lo and lo < ND: table[lo:hi + 1, COL[s]] = True count += 1 return backtest(table, cost, count) def period_sim(selections, cost): """selections：[(t, 股票清單)]，t 收盤決定，持有到下一個 t。""" table = np.zeros((ND, len(STOCKS)), dtype=bool) for k, (t, stocks) in enumerate(selections): end = selections[k + 1][0] if k + 1 < len(selections) else ND table[t:end, [COL[s] for s in stocks if s in COL]] = True return backtest(table, cost) def position_stats(report, frame, events): curve = active(report.creturn) held = frame.sum(axis=1).reindex(curve.index).fillna(0) buys = (frame & ~frame.shift(1, fill_value=False)).sum().sum() years = max((curve.index[-1] - curve.index[0]).days / 365.25, .01) return {"events": events, "buys_per_year": round(float(buys / years), 1), "invested": round(float((held > 0).mean()), 4), "avg_holdings": round(float(held[held > 0].mean()), 2) if (held > 0).any() else None} def equal_weight_curve(): """等權流動股：每月底買進全部流動股、等權、含預設成本，當第二個對照組。""" picks = [(m, [STOCKS[j] for j in np.flatnonzero(liquid[m])]) for m in month_ends() if liquid[m].any()] return period_sim(picks, "default")[0].creturn def validate(key, equal_weight): design = DESIGNS[key] xs, ys, base = design["x"], design["y"], design["base"] grid, base_run = [], None for x in xs: for y in ys: run = design["fn"](x, y, "default") row = compare(run[0].creturn) row.update({"x": x, "y": y, "pos": position_stats(*run)}) grid.append(row) if (x, y) == base: base_run = run base_row = next(g for g in grid if (g["x"], g["y"]) == base) s = base_row["strategy"] result = {"title": design["title"], "x_name": design["x_name"], "y_name": design["y_name"], "x": xs, "y": ys, "base": list(base), "grid": grid, "grid_wins": sum(g["win"] for g in grid), "base_default_cost": base_row, "halves": halves(base_run[0].creturn), "base_zero_cost": compare(design["fn"](*base, "zero")[0].creturn), "equal_weight_same_window": stats(equal_weight.loc[s["start"]:s["end"]])} h = result["halves"] result["pass"] = {"neighbors": result["grid_wins"] >= 7, "halves": h[0]["win"] and h[1]["win"], "cost": base_row["win"]} return result, base_run[0] # --------------------------------------------------------------------------- # 2. 事件研究：減等權流動股（每日等權、無成本） # --------------------------------------------------------------------------- def equal_weight_index(): r = adj_frame.pct_change(fill_method=None).to_numpy() held = np.vstack([np.zeros((1, liquid.shape[1]), dtype=bool), liquid[:-1]]) v = np.where(held & np.isfinite(r), r, np.nan) with np.errstate(invalid="ignore"): return np.cumprod(1 + np.nan_to_num(np.nanmean(v, axis=1))) EW_INDEX = equal_weight_index() def study(rows, horizons=(5, 10, 20, 40, 60)): """b 收盤買進、持有 h 日的平均與中位超額報酬；t 值先按進場月平均，避免同月事件灌大樣本。""" out = {} for h in horizons: rec = [] for s, b, _ in rows: if b + h >= ND: continue j = COL[s] r = adj[b + h, j] / adj[b, j] - 1 if np.isfinite(r): rec.append((b, r - (EW_INDEX[b + h] / EW_INDEX[b] - 1), r - (adj0050[b + h] / adj0050[b] - 1))) if not rec: continue f = pd.DataFrame(rec, columns=["b", "x", "x0050"]) by_month = f.groupby(pd.Series(days[f.b.values].to_period("M"), index=f.index)).x.mean() se = by_month.std(ddof=1) / np.sqrt(len(by_month)) if len(by_month) > 2 else float("nan") out[str(h)] = {"n": len(f), "mean_excess": round(float(f.x.mean()), 5), "median_excess": round(float(f.x.median()), 5), "hit_rate": round(float((f.x > 0).mean()), 4), "t_month": round(float(by_month.mean() / se), 2) if np.isfinite(se) and se > 0 else None, "mean_excess_0050": round(float(f.x0050.mean()), 5)} return out def event_curve(rows, lo=-20, hi=60): """以 b−1（決策日）為 0 的平均累積超額報酬。""" offsets = list(range(lo, hi + 1)) total, count = np.zeros(len(offsets)), np.zeros(len(offsets)) for s, b, _ in rows: t0 = b - 1 if t0 + lo < 0 or not np.isfinite(adj[t0, COL[s]]): continue for k, o in enumerate(offsets): t = t0 + o if t >= ND: break v = adj[t, COL[s]] / adj[t0, COL[s]] - EW_INDEX[t] / EW_INDEX[t0] if np.isfinite(v): total[k] += v count[k] += 1 return {"offset": offsets, "mean": [round(float(x), 5) for x in total / np.maximum(count, 1)], "n": [int(x) for x in count]} # --------------------------------------------------------------------------- # 3. 訊號一：營收更正 # --------------------------------------------------------------------------- def monthly(key): t = plain(key) return t[[c for c in t.columns if c in COL]] def revisions(): """新一期月報的「上月營收」「去年當月營收」跟當初公布值比，日期用新月報的索引日（法定期限）。 monthly_revenue_revised 只有 2025 年以後才看得到更正，所以不用它。單月 ≥60 筆的整批差異（2013 年會計轉換） 與幅度超過 ±100% 的不算；同月兩者都有時取「上月」那筆。 """ rev, last_month, last_year = monthly("monthly_revenue:當月營收"), monthly("monthly_revenue:上月營收"), monthly("monthly_revenue:去年當月營收") frames = [] for name, x in (("上月", (last_month / rev.shift(1) - 1).where(rev.shift(1) > 0)), ("去年同月", (last_year / rev.shift(12) - 1).where(rev.shift(12) > 0))): m = x.stack().dropna() m = m[m.abs() >= .01] size = m.groupby(level=0).size() m = m[~m.index.get_level_values(0).isin(size[size >= 60].index)] m = m[m.abs() <= 1].rename("x").reset_index() m.columns = ["t", "stock_id", "x"] m["source"] = name frames.append(m) out = pd.concat(frames).sort_values(["stock_id", "t", "source"], ascending=[True, True, True]) return out.drop_duplicates(["stock_id", "t"]).reset_index(drop=True) REVISIONS = revisions() REV_YOY = monthly("monthly_revenue:去年同月增減(%)") def revision_rows(theta, H, sign, growth): r = REVISIONS[REVISIONS.x * sign >= theta] rows = [] for s, tt, t0 in zip(r.stock_id, first_day_on_or_after(r.t), r.t): if tt >= ND or not liquid[tt, COL[s]]: continue if growth and not REV_YOY.at[t0, s] >= 20: continue rows.append((s, tt + 1, tt + 1 + H)) return rows def rev_up(theta, H, cost): return event_sim(revision_rows(theta / 100, H, 1, False), cost) def rev_down(theta, H, cost): return event_sim(revision_rows(theta / 100, H, -1, False), cost) def rev_up_growth(theta, H, cost): return event_sim(revision_rows(theta / 100, H, 1, True), cost) # --------------------------------------------------------------------------- # 4. 訊號二：訴訟重訊 # --------------------------------------------------------------------------- NEGATIVE = re.compile(r"(遭|被|檢察官|檢方).{0,12}起訴|起訴書|羈押|搜索|檢調|調查局|交保|背信|掏空|敗訴|假扣押|強制執行|支付命令") QUALITY_START = "2014-03-01" def lawsuit_events(): """lawsuit_info 的 date 是公開資訊觀測站發言時間（台北）；key_date 是回補時間，不能用。""" t = plain("lawsuit_info") KEY_DATES["lawsuit"] = pd.to_datetime(t["key_date"]).dt.date.astype(str).value_counts() t["ts"] = pd.to_datetime(t["date"]) t["stock_id"] = t["stock_id"].astype(str) t = t[t.stock_id.isin(COL) & (t.ts >= days[1])].copy() t["t"] = first_day_on_or_after(t.ts) t = t[t.t < ND] t["negative"] = t["title"].fillna("").str.replace(r"\s+", "", regex=True).map(lambda x: bool(NEGATIVE.search(x))) return t.groupby(["stock_id", "t"]).agg(negative=("negative", "max")).reset_index() KEY_DATES = {} LAWSUITS = lawsuit_events() def quality_table(): p, g, s = (plain(f"quality_factor_z_score:{k}") for k in ("profitability", "growth", "safety")) q = (p + g + s) / 3 return q[[c for c in q.columns if c in COL]].where(p.notna() & g.notna() & s.notna()) QUALITY = quality_table() QUALITY_DAILY = QUALITY.reindex(index=QUALITY.index.union(days)).ffill().reindex(days).reindex(columns=STOCKS).to_numpy() def lawsuit_drift(delay, H, cost): rows = [] for s, t in zip(LAWSUITS.stock_id, LAWSUITS.t): i = t + delay if i - 1 < ND and liquid[i - 1, COL[s]]: rows.append((s, i, i + H)) return event_sim(rows, cost) def lawsuit_drop_rows(theta, H, negative_only, quality, from_quality_start): start = days.searchsorted(pd.Timestamp(QUALITY_START)) rows = [] for s, t, neg in zip(LAWSUITS.stock_id, LAWSUITS.t, LAWSUITS.negative): if (negative_only and not neg) or t < 1 or t + 1 >= ND or (from_quality_start and t < start): continue j = COL[s] move = adj[t + 1, j] / adj[t - 1, j] - adj0050[t + 1] / adj0050[t - 1] if not np.isfinite(move) or move > -theta or not liquid[t + 1, j]: continue if quality and not QUALITY_DAILY[t, j] > 0: continue rows.append((s, t + 2, t + 2 + H)) return rows def lawsuit_negative(theta, H, cost): return event_sim(lawsuit_drop_rows(theta / 100, H, True, False, False), cost) def lawsuit_quality(theta, H, cost): return event_sim(lawsuit_drop_rows(theta / 100, H, False, True, True), cost) def lawsuit_drop_plain(theta, H, cost): return event_sim(lawsuit_drop_rows(theta / 100, H, False, False, True), cost) # --------------------------------------------------------------------------- # 5. 訊號三：鉅亨新聞 # --------------------------------------------------------------------------- def clean_name(name): return re.sub(r"(-KY|-創|\*|^F-|-DR)", "", str(name)).strip() def stock_names(): out = {} def add(stock, name): name = clean_name(name) if len(name) >= 2: out.setdefault(str(stock), set()).add(name) info = plain("company_basic_info") for s, n in zip(info["stock_id"], info["公司簡稱"]): add(s, n) cats = plain("security_categories") for s, n in zip(cats["stock_id"], cats["name"]): add(s, n) suits = plain("lawsuit_info") for s, n in zip(suits["stock_id"], suits["name"]): add(s, n) for key in ("trading_halt:證券名稱", "block_trade:證券名稱"): # 已下市公司的舊名只查得到這裡 t = plain(key) for s in t.columns: for n in t[s].dropna().unique(): add(s, n) return out def news_table(): """同一篇有兩筆（網址帶 ?exp=a），先去重。stock_ids 雜訊多，標籤含該股、標題也有其簡稱才算到這檔。""" raw = plain("tw_news_cnyes") KEY_DATES["news"] = pd.to_datetime(raw["key_date"]).dt.date.astype(str).value_counts() raw["date"] = pd.to_datetime(raw["date"]).dt.normalize() raw["article"] = raw["url"].str.replace(r"\?.*$", "", regex=True) articles = raw.groupby("article").agg(date=("date", "min"), title=("title", "first"), tags=("stock_ids", ",".join)) names = stock_names() counts = np.zeros((ND, len(STOCKS)), dtype=np.int32) seen, tagged = set(), np.zeros(len(articles), dtype=bool) for i, (t, title, tags) in enumerate(zip(first_day_on_or_after(articles.date), articles.title, articles.tags)): title = str(title) ids = set(str(tags).split(",")) tagged[i] = any(n in title for s in ids for n in names.get(s, ())) if t >= ND: continue for s in ids: if s not in COL or not any(n in title for n in names.get(s, ())): continue if (s, t, title) not in seen: seen.add((s, t, title)) counts[t, COL[s]] += 1 return pd.DataFrame(counts, index=days, columns=STOCKS), articles.assign(tagged=tagged), len(raw) NEWS, ARTICLES, NEWS_RAW_ROWS = news_table() NEWS_START = days.searchsorted(pd.Timestamp("2013-01-01")) new_high_60 = (adj_frame >= adj_frame.rolling(60, min_periods=60).max()).to_numpy() def news_rows(M, H, high): last_5 = NEWS.rolling(5, min_periods=5).sum() usual = (NEWS.rolling(120, min_periods=120).sum().shift(5) / 24).clip(lower=.25) # 前 120 日每 5 日平均 hit = (last_5 >= 3) & (last_5 >= M * usual) first = (hit & ~(hit.astype(float).shift(1).rolling(20, min_periods=1).max().fillna(0) > 0)).to_numpy().copy() first[:NEWS_START + 125] = False ts, js = np.nonzero(first & liquid & (new_high_60 if high else True)) return [(STOCKS[j], t + 1, t + 1 + H) for t, j in zip(ts, js)] def news_spike(M, H, cost): return event_sim(news_rows(M, H, False), cost) def news_breakout(M, H, cost): return event_sim(news_rows(M, H, True), cost) def cold_mask(D): silent = NEWS.rolling(D, min_periods=D).sum() == 0 traded = close.reindex(columns=STOCKS).notna().rolling(D, min_periods=D).sum() >= .9 * D return (silent & traded).to_numpy() def months_from(start): return [m for m in month_ends() if m >= start] def cold_stocks(D, F, cost): cold, floor = cold_mask(D), liquid_mask(F * 1e4) return period_sim([(m, [STOCKS[j] for j in np.flatnonzero(cold[m] & floor[m])]) for m in months_from(NEWS_START + D)], cost) def revenue_rank(): r = plain("monthly_revenue:當月營收") x = r.rolling(3, min_periods=3).mean() / r.rolling(12, min_periods=12).mean() return x.reindex(index=x.index.union(days)).ffill().reindex(days).reindex(columns=STOCKS).to_numpy() REV_RANK = revenue_rank() def revenue_picks(D, N, cold_only): cold = cold_mask(D) if cold_only else None picks = [] for m in months_from(NEWS_START + D): ok = liquid[m] & np.isfinite(REV_RANK[m]) if cold_only: ok = ok & cold[m] j = np.flatnonzero(ok) picks.append((m, [STOCKS[i] for i in j[np.argsort(-REV_RANK[m, j], kind="stable")[:N]]])) return picks def cold_revenue(D, N, cost): return period_sim(revenue_picks(D, N, True), cost) def revenue_only(D, N, cost): return period_sim(revenue_picks(D, N, False), cost) # --------------------------------------------------------------------------- # 6. 訊號四：品質分數 # --------------------------------------------------------------------------- def quality_top(N, F, cost): floor, table = liquid_mask(F * 1e4), QUALITY.reindex(columns=STOCKS) picks = [] for date, row in table.iterrows(): t = int(first_day_on_or_after([date])[0]) if t >= ND or row.notna().sum() == 0: continue v = row.to_numpy() j = np.flatnonzero(np.isfinite(v) & floor[t]) picks.append((t, [STOCKS[i] for i in j[np.argsort(-v[j], kind="stable")[:N]]])) return period_sim(picks, cost) def quality_momentum(L, N, cost): picks = [] for m in months_from(days.searchsorted(pd.Timestamp(QUALITY_START))): j = np.flatnonzero(np.isfinite(QUALITY_DAILY[m]) & liquid[m]) if len(j) == 0: continue good = j[np.argsort(-QUALITY_DAILY[m, j], kind="stable")[:int(np.ceil(len(j) * .3))]] r = adj[m] / adj[m - L] - 1 good = good[np.isfinite(r[good])] picks.append((m, [STOCKS[i] for i in good[np.argsort(-r[good], kind="stable")[:N]]])) return period_sim(picks, cost) def quintile_curves(): """品質分數五等分，每個資料日換股，不扣成本。""" table = QUALITY.reindex(columns=STOCKS) groups = {q: [] for q in range(5)} for date, row in table.iterrows(): t = int(first_day_on_or_after([date])[0]) if t >= ND or row.notna().sum() == 0: continue v = row.to_numpy() j = np.flatnonzero(np.isfinite(v) & liquid[t]) j = j[np.argsort(-v[j], kind="stable")] for q, part in enumerate(np.array_split(j, 5)): groups[q].append((t, [STOCKS[i] for i in part])) return {str(q + 1): stats(active(period_sim(p, "zero")[0].creturn)) for q, p in groups.items()} REV = dict(x_name="重述幅度 θ（%）", x=[1, 3, 5], y_name="持有 H（交易日）", y=[20, 40, 60], base=(3, 40)) DROP = dict(x_name="跌幅 θ（%）", x=[3, 5, 8], y_name="持有 H（交易日）", y=[10, 20, 40], base=(5, 20)) NEWSG = dict(x_name="倍數 M", x=[3, 5, 8], y_name="持有 H（交易日）", y=[5, 10, 20], base=(5, 10)) COLD = dict(x_name="無新聞天數 D", x=[60, 120, 250], y_name="持股 N", y=[10, 20, 30], base=(120, 20)) DESIGNS = { "S2u": dict(fn=rev_up, title="營收上修買進", **REV), "S2d": dict(fn=rev_down, title="營收下修買進", **REV), "C2": dict(fn=rev_up_growth, title="上修＋年增 20%", **REV), "S3a": dict(fn=lawsuit_drift, title="訴訟重訊後買進", x_name="進場延遲 d（交易日）", x=[1, 3, 5], y_name="持有 H（交易日）", y=[10, 20, 40], base=(1, 20)), "S3b": dict(fn=lawsuit_negative, title="負面訴訟跌深買進", **DROP), "C3": dict(fn=lawsuit_quality, title="品質為正的公司遇訴訟跌深", **DROP), "C3base": dict(fn=lawsuit_drop_plain, title="訴訟跌深（C3 的對照）", **DROP), "S5a": dict(fn=news_spike, title="新聞暴增買進", **NEWSG), "C5a": dict(fn=news_breakout, title="新聞暴增＋60 日新高", **NEWSG), "S5b": dict(fn=cold_stocks, title="冷門股", x_name="無新聞天數 D", x=[60, 120, 250], y_name="均額下限 F（萬）", y=[1000, 2000, 5000], base=(120, 2000)), "C5b": dict(fn=cold_revenue, title="冷門股裡營收動能最高", **COLD), "C5bbase": dict(fn=revenue_only, title="全市場營收動能最高（C5b 的對照）", **COLD), "S7": dict(fn=quality_top, title="品質分數最高", x_name="持股 N", x=[20, 30, 50], y_name="均額下限 F（萬）", y=[1000, 2000, 5000], base=(30, 2000)), "C7": dict(fn=quality_momentum, title="品質前 30%＋動能", x_name="動能窗 L（交易日）", x=[60, 120, 240], y_name="持股 N", y=[10, 20, 30], base=(120, 20)), } # --------------------------------------------------------------------------- # 7. 描述統計 # --------------------------------------------------------------------------- def descriptive(): locked = plain("monthly_revenue:當月營收") revised = plain("monthly_revenue_revised:當月營收") revised_at = plain("monthly_revenue_revised:revised_at") index, cols = locked.index.intersection(revised.index), locked.columns.intersection(revised.columns) a, b = locked.loc[index, cols], revised.loc[index, cols] different = (a != b) & a.notna() & b.notna() years = pd.to_datetime(REVISIONS.t).dt.year rows = {"rev_up": revision_rows(.03, 40, 1, False), "rev_down": revision_rows(.03, 40, -1, False), "lawsuit_all": [(s, t + 1, t + 21) for s, t in zip(LAWSUITS.stock_id, LAWSUITS.t) if t < ND and liquid[t, COL[s]]], "lawsuit_negative": lawsuit_drop_rows(.05, 20, True, False, False), "lawsuit_quality": lawsuit_drop_rows(.05, 20, False, True, True), "lawsuit_drop_plain": lawsuit_drop_rows(.05, 20, False, False, True), "news_spike": news_rows(5, 10, False), "news_breakout": news_rows(5, 10, True)} ay = ARTICLES.date.dt.year first_close = close.reindex(columns=STOCKS).apply(pd.Series.first_valid_index) early = ARTICLES[ARTICLES.date < "2016-01-01"] tags = early.tags.str.split(",").explode() tags = tags[tags.isin(COL)] tag_dates = early.date.reindex(tags.index) listed = pd.to_datetime(tags.map(first_close)) later_tags = int((listed > tag_dates + pd.Timedelta(days=1095)).sum()) cold = cold_mask(120) ends = months_from(NEWS_START + 120) return {"revision": {"table_cells_different": int(different.sum().sum()), "first_year_different": int(index[different.any(axis=1)].year.min()), "revised_at_rows": int(revised_at.notna().sum().sum()), "events": len(REVISIONS), "by_source": REVISIONS.source.value_counts().to_dict(), "up_by_year": years[REVISIONS.x > 0].value_counts().sort_index().to_dict(), "down_by_year": years[REVISIONS.x < 0].value_counts().sort_index().to_dict()}, "lawsuit": {"events": len(LAWSUITS), "negative": int(LAWSUITS.negative.sum()), "backfill_2023_12_01": int(KEY_DATES["lawsuit"]["2023-12-01"]), "by_year": pd.Series(days[LAWSUITS.t.values].year).value_counts().sort_index().to_dict()}, "news": {"raw_rows": NEWS_RAW_ROWS, "articles": len(ARTICLES), "backfill_2026_02_26": int(KEY_DATES["news"]["2026-02-26"]), "backfill_2023_12_12": int(KEY_DATES["news"]["2023-12-12"]), "tags_before_2016": len(tags), "tags_listed_3y_later": later_tags, "articles_by_year": ay.value_counts().sort_index().to_dict(), "tag_in_title_share_by_year": {int(y): round(float(v), 4) for y, v in ARTICLES.tagged.groupby(ay).mean().items()}, "matched_by_year": {int(y): int(v) for y, v in NEWS.sum(axis=1).groupby(days.year).sum().items() if v > 0}, "cold_120_median_by_year": {int(y): float(v) for y, v in pd.Series((cold[ends] & liquid[ends]).sum(axis=1), index=days[ends]).groupby(lambda x: x.year).median().items()}, "liquid_median_by_year": {int(y): float(v) for y, v in pd.Series(liquid[ends].sum(axis=1), index=days[ends]).groupby(lambda x: x.year).median().items()}}, "quality_quintiles": quintile_curves(), "study": {k: study(v) for k, v in rows.items()}, "curves": {k: event_curve(v) for k, v in rows.items()}} # --------------------------------------------------------------------------- # 9. 執行 # --------------------------------------------------------------------------- def main(): equal_weight = equal_weight_curve() results, reports = {}, {} for key in DESIGNS: results[key], reports[key] = validate(key, equal_weight) metrics = {"run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "data_end": str(bench.index[-1].date()), "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(): curve = active(report.creturn) frame[f"{key} {DESIGNS[key]['title']}"] = curve / curve.iloc[0] frame = pd.DataFrame(frame) frame["0050含息"] = bench.reindex(frame.index) frame["等權流動股"] = equal_weight.reindex(frame.index) frame.index.name = "date" frame.round(6).to_csv("data.csv") print("完成：metrics.json、data.csv") if __name__ == "__main__": main()