""" 台股新股與興櫃：7 個訊號、6 個月營收動能組合版的三項驗證，以及抽籤、溢價、解禁、興櫃轉上市的統計（2007 年起至最新資料） 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：每個設計 2 個主要參數各取 3 個值，9 組裡至少 7 組要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 組合版另外要比底好：原參數的日夏普高於同持股數的月營收動能，而且 9 組裡至少 7 組高於同持股數的底。 新股清單：ipo_subscription_lottery 的初上市、初上櫃、第一上市／上櫃公司初上市櫃、創新板初上市，四碼普通股、未取消抽籤； 上市日 D1＝撥券日期起第一個有收盤價的交易日。訊號日收盤後才算得出來，sim() 在隔一個交易日收盤成交。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入，約 5 分鐘。 輸出：metrics.json（全部結果）、data.csv（各設計原參數與 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 START = "2007-06-01" MRM_START = "2015-01-01" IPO_TYPES = ["初上市", "初上櫃", "第一上市公司初上市", "第一上櫃公司初上櫃", "創新板初上市"] 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 adj_close = plain("etl:adj_close").reindex(days) bench = adj_close["0050"].dropna() 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), "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.iloc[max(moved[0] - 1, 0):] if len(moved) else curve def compare(curve): curve = active(curve) s = stats(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 return [compare(curve.loc[lo:hi]) for lo, hi in ((curve.index[0], mid), (mid, curve.index[-1]))] def trade_stats(report): t = report.get_trades() if len(t) == 0: return {} years = (report.creturn.index[-1] - report.creturn.index[0]).days / 365.25 return {"trades": int(len(t)), "trades_per_year": round(len(t) / years, 1), "win_rate": round(float((t["return"] > 0).mean()), 4), "mean_return": round(float(t["return"].mean()), 6)} def avg_holdings(report): n = (report.position.fillna(0) != 0).sum(axis=1) n = n[n > 0] return round(float(n.mean()), 1) if len(n) else None def at(date): """日期在交易日序列裡的位置（當天或之後第一個交易日）。""" return int(days.searchsorted(pd.Timestamp(date))) # --------------------------------------------------------------------------- # 2. 新股清單：上市日、承銷價、中籤率、首五日溢價、興櫃動能、上市前營收年增 # --------------------------------------------------------------------------- lottery = plain("ipo_subscription_lottery") lottery["stock_id"] = lottery["stock_id"].astype(str) ipo = lottery[lottery["發行市場"].isin(IPO_TYPES) & (lottery["取消公開抽籤"].fillna("") == "") & lottery["stock_id"].str.fullmatch(r"\d{4}")] ipo = ipo.sort_values("撥券日期").drop_duplicates("stock_id", keep="last").copy() first_close = close.apply(lambda s: s.first_valid_index()) ipo["d1"] = ipo["stock_id"].map(first_close) lag = (ipo["d1"] - ipo["撥券日期"]).dt.days ipo = ipo[ipo["d1"].notna() & (ipo["d1"] > pd.Timestamp("2007-05-01")) & lag.between(0, 10)].copy() ipo["i1"] = [days.get_loc(d) for d in ipo["d1"]] ipo["price"] = ipo["實際承銷價"].where(ipo["實際承銷價"] > 0, ipo["承銷價"]) ipo["lottery_rate"] = ipo["中籤率"] / 100 ipo["c1"] = [close[s].iloc[i] for s, i in zip(ipo["stock_id"], ipo["i1"])] ipo["prem5"] = [close[s].iloc[i:i + 5].mean() / p - 1 if i + 5 <= len(days) else np.nan for s, i, p in zip(ipo["stock_id"], ipo["i1"], ipo["price"])] # 興櫃：上市前 10 天內還有興櫃成交才算興櫃轉上市櫃；最後 L 個興櫃交易日的報酬 rotc_close = plain("rotc_price:收盤價") rotc_last, rotc_ret = [], {L: [] for L in (20, 60, 120)} for s, d in zip(ipo["stock_id"], ipo["d1"]): hist = rotc_close[s].loc[:d - pd.Timedelta(days=1)] if s in rotc_close.columns else pd.Series(dtype=float) last = hist.last_valid_index() if last is None or (d - last).days > 10: rotc_last.append(np.nan) for L in rotc_ret: rotc_ret[L].append(np.nan) continue hist = hist.ffill() rotc_last.append(hist.iloc[-1]) for L in rotc_ret: ok = len(hist) > L and pd.notna(hist.iloc[-1 - L]) rotc_ret[L].append(hist.iloc[-1] / hist.iloc[-1 - L] - 1 if ok else np.nan) ipo["rotc_last_close"] = rotc_last for L, v in rotc_ret.items(): ipo[f"rotc_ret{L}"] = v def monthly_revenue(): """上市櫃月營收優先，缺的月份補興櫃月營收；index 換成營收月份，另回傳各月的公布期限日。""" tables, deadlines = [], [] for key in ("monthly_revenue:當月營收", "rotc_monthly_revenue:當月營收"): t = plain(key) month = (t.index - pd.DateOffset(months=1)).to_period("M") deadline = pd.Series(t.index, index=month) t.index = month tables.append(t[~t.index.duplicated(keep="last")]) deadlines.append(deadline[~deadline.index.duplicated(keep="last")]) rev = tables[0].combine_first(tables[1]).sort_index() return rev, deadlines[0].combine_first(deadlines[1]).reindex(rev.index) rev_m, rev_deadline = monthly_revenue() growth = [] for s, i in zip(ipo["stock_id"], ipo["i1"]): known = rev_deadline[rev_deadline <= days[i - 1]] # 上市前一天已過公布期限的月份 if s not in rev_m.columns or known.empty: growth.append(np.nan) continue m = known.index[-1] now = rev_m[s].reindex(pd.period_range(m - 2, m, freq="M")) year_ago = rev_m[s].reindex(pd.period_range(m - 14, m - 12, freq="M")) ok = now.notna().all() and year_ago.notna().all() and year_ago.sum() > 0 growth.append(now.sum() / year_ago.sum() - 1 if ok else np.nan) ipo["rev_g3"] = growth ipo = ipo.reset_index(drop=True) ipo_stocks = sorted(ipo["stock_id"]) # --------------------------------------------------------------------------- # 3. 單獨版 S1–S7：事件後持有固定天數，等權、不設 resample # --------------------------------------------------------------------------- def mask(spans): """spans：(股票, 開始位置, 結束位置) → 每檔股票在這段期間為 True 的表。""" table = np.zeros((len(days), len(close.columns)), dtype=bool) col = {s: k for k, s in enumerate(close.columns)} for s, lo, hi in spans: lo, hi = max(int(lo), 0), min(int(hi), len(days) - 1) if s in col and hi >= lo: table[lo:hi + 1, col[s]] = True return pd.DataFrame(table, index=days, columns=close.columns) def holding(rows): """rows：(股票, 訊號開始位置, 訊號結束位置)，訊號為 True 的隔天收盤持有、結束後隔天收盤賣出。""" pos = mask(rows).loc[START:, ipo_stocks] return pos.loc[:, pos.any()] def run_events(rows, cost): return sim(holding(rows), upload=False, **COSTS[cost]) def hold(pairs, E, H, cost): """pairs：(股票, 起算日位置)。起算日算第 1 天，第 E 天收盤後出訊號、隔天收盤買進，持有 H 日。""" return run_events([(s, a + E - 1, a + E + H - 2) for s, a in pairs], cost) def lockup_dates(months=(6, 12)): """上市滿 months 個月那天（之後第一個交易日）的位置，預設 6、12 個月。""" out = [] for s, d1 in zip(ipo["stock_id"], ipo["d1"]): for m in months: day = d1 + pd.DateOffset(months=m) if day <= days[-1]: out.append((s, at(day))) return out def s1(E, H, cost): """新股第 E 天收盤後買進（第 E+1 天收盤成交），持有 H 日。""" return hold(zip(ipo.stock_id, ipo.i1), E, H, cost) def s2(Y, S, cost): """新股組合：第 5 天收盤後延後 S 個月進場，持有到上市滿 Y 年。""" rows = [] for s, i, d1 in zip(ipo.stock_id, ipo.i1, ipo.d1): if i + 4 >= len(days): continue start = at(days[i + 4] + pd.DateOffset(months=S)) rows.append((s, start, at(d1 + pd.DateOffset(years=Y)) - 1)) return run_events(rows, cost) def s3(X, H, cost): """熱門新股：中籤率 ≤ X，上市前一天訊號、上市首日收盤買進，持有 H 日。""" e = ipo[ipo.lottery_rate <= X] return hold(zip(e.stock_id, e.i1), 0, H, cost) def s4(P, H, cost): """低溢價新股：首五日均價 ÷ 承銷價 − 1 ≤ P，第 6 天收盤買進，持有 H 日。""" e = ipo[ipo.prem5 <= P] return hold(zip(e.stock_id, e.i1), 5, H, cost) def s5(k, H, cost): """集保解禁：上市滿 6、12 個月那天往後 k 個交易日收盤買進，持有 H 日。""" return hold(lockup_dates(), k, H, cost) def s6(L, H, cost): """興櫃轉上市櫃、興櫃最後 L 日報酬 > 0：上市首日收盤買進，持有 H 日。""" e = ipo[ipo[f"rotc_ret{L}"] > 0] return hold(zip(e.stock_id, e.i1), 0, H, cost) def s7(G, H, cost): """上市前近 3 個月營收年增 ≥ G：上市首日收盤買進，持有 H 日。""" e = ipo[ipo.rev_g3 >= G] return hold(zip(e.stock_id, e.i1), 0, H, cost) # --------------------------------------------------------------------------- # 4. 組合版 C2–C7：月營收動能（/strategies/monthly-revenue-momentum）＋新股濾網 # --------------------------------------------------------------------------- fclose = data.get("price:收盤價") rev = data.get("monthly_revenue:當月營收") trend = (fclose > fclose.average(60)) & (fclose > fclose.average(20)) & (fclose > fclose.average(120)) rev_up = rev.average(3) > rev.average(12) momentum = (fclose / fclose.shift() - 1).rolling(5).mean() base_cond = trend & rev_up def young(e, years=3, start=0): """上市第 start+1 天起到滿 years 年前一天（條件要到第 start 天收盤才知道時，從隔天起排除）。""" return mask([(s, i + start, at(d1 + pd.DateOffset(years=years)) - 1) for s, i, d1 in zip(e.stock_id, e.i1, e.d1)]) def mrm(N, cost, exclude=None): cond = base_cond if exclude is not None: exclude = exclude.reindex(index=cond.index, columns=cond.columns, fill_value=False).astype(bool) cond = cond & ~exclude position = momentum[cond].loc[MRM_START:].is_largest(N) return sim(position, resample="M", resample_offset="11D", stop_loss=0.1, upload=False, **COSTS[cost]) def c2(Y, N, cost): return mrm(N, cost, young(ipo, Y)) def c3(X, N, cost): return mrm(N, cost, young(ipo[ipo.lottery_rate <= X])) def c4(Q, N, cost): return mrm(N, cost, young(ipo[ipo.prem5 >= Q], start=5)) def c5(W, N, cost): return mrm(N, cost, mask([(s, a - W, a - 1) for s, a in lockup_dates()])) def c6(L, N, cost): return mrm(N, cost, young(ipo[ipo[f"rotc_ret{L}"] <= 0])) def c7(G, N, cost): return mrm(N, cost, young(ipo[ipo.rev_g3 < G])) DESIGNS = { "S1": ("新股第 E 天收盤後買進、持有 H 日", s1, ("E", [5, 10, 20]), ("H", [60, 120, 250]), (5, 120)), "S2": ("新股組合持有到上市滿 Y 年", s2, ("Y", [1, 2, 3]), ("S（延後月數）", [0, 3, 6]), (3, 0)), "S3": ("熱門新股（中籤率 ≤ X）首日收盤買進", s3, ("X", [0.005, 0.01, 0.02]), ("H", [20, 60, 120]), (0.01, 60)), "S4": ("低溢價新股（首五日溢價 ≤ P）第 6 天買進", s4, ("P", [0.10, 0.25, 0.40]), ("H", [60, 120, 250]), (0.25, 120)), "S5": ("集保解禁（上市滿 6、12 個月）後買進", s5, ("k", [-20, 0, 20]), ("H", [20, 60, 120]), (0, 60)), "S6": ("興櫃轉上市櫃、興櫃最後 L 日上漲", s6, ("L", [20, 60, 120]), ("H", [20, 60, 120]), (60, 60)), "S7": ("上市前營收年增 ≥ G 的新股", s7, ("G", [0.10, 0.30, 0.50]), ("H", [20, 60, 120]), (0.30, 60)), "C2": ("月營收動能排除上市未滿 Y 年", c2, ("Y", [1, 2, 3]), ("N", [8, 10, 12]), (3, 10)), "C3": ("月營收動能排除 3 年內的熱門新股", c3, ("X", [0.005, 0.01, 0.02]), ("N", [8, 10, 12]), (0.01, 10)), "C4": ("月營收動能排除 3 年內的高溢價新股", c4, ("Q", [0.30, 0.50, 0.75]), ("N", [8, 10, 12]), (0.50, 10)), "C5": ("月營收動能避開解禁前 W 日", c5, ("W", [10, 20, 40]), ("N", [8, 10, 12]), (20, 10)), "C6": ("月營收動能排除 3 年內興櫃期間下跌的新股", c6, ("L", [20, 60, 120]), ("N", [8, 10, 12]), (60, 10)), "C7": ("月營收動能排除 3 年內上市前營收成長低的新股", c7, ("G", [0.10, 0.30, 0.50]), ("N", [8, 10, 12]), (0.30, 10)), } def validate(key, base_by_n): title, fn, (xn, xs), (yn, ys), base = DESIGNS[key] grid, base_cell, base_report = [], None, None for x in xs: for y in ys: report = fn(x, y, "default") cell = compare(report.creturn) cell.update({"x": x, "y": y, "avg_holdings": avg_holdings(report)}) grid.append(cell) if (x, y) == base: base_cell, base_report = cell, report result = { "title": title, "x": {"name": xn, "values": xs}, "y": {"name": yn, "values": ys}, "base": {"x": base[0], "y": base[1]}, "grid": grid, "grid_wins": sum(c["win"] for c in grid), "base_default_cost": base_cell, "halves": halves(base_report.creturn), "base_zero_cost": compare(fn(*base, "zero").creturn), "base_trades": trade_stats(base_report), } passed = {"neighbors": result["grid_wins"] >= 7, "halves": all(h["win"] for h in result["halves"]), "cost": base_cell["win"]} if key.startswith("C"): beats = [c["strategy"]["sharpe"] > base_by_n[c["y"]]["strategy"]["sharpe"] for c in grid] result["beats_base"] = int(sum(beats)) passed["beats_base"] = bool(base_cell["strategy"]["sharpe"] > base_by_n[base[1]]["strategy"]["sharpe"] and sum(beats) >= 7) passed["all"] = all(passed.values()) result["pass"] = passed print(f"{key} {title}：日夏普 {base_cell['strategy']['sharpe']:.2f}（0050 {base_cell['bench']['sharpe']:.2f}），" f"9 組贏 {result['grid_wins']} 組，通過 {passed['all']}", flush=True) return result, base_report # --------------------------------------------------------------------------- # 5. 統計：抽籤期望值、溢價與中籤率分組、解禁前後、興櫃到上市、三年買進持有 # --------------------------------------------------------------------------- def forward_excess(s, i0, i1): """i0 收盤到 i1 收盤：還原報酬與同期 0050 的差。""" if i0 < 0 or i1 >= len(days) or s not in adj_close.columns: return np.nan, np.nan a0, a1 = adj_close[s].iloc[i0], adj_close[s].iloc[i1] if not (a0 > 0 and a1 > 0): return np.nan, np.nan r = a1 / a0 - 1 b0, b1 = adj_close["0050"].iloc[i0], adj_close["0050"].iloc[i1] return r, r - (b1 / b0 - 1) def summary(x): x = pd.Series(x, dtype=float).dropna() return {"n": int(len(x)), "mean": round(float(x.mean()), 6), "median": round(float(x.median()), 6), "pos_share": round(float((x > 0).mean()), 4), "t": round(float(x.mean() / (x.std() / math.sqrt(len(x)))), 2) if len(x) > 2 else None} def histogram(x, lo, hi, step): """固定區間的次數分配，超出範圍的併到兩端。""" x = pd.Series(x, dtype=float).dropna().clip(lo, hi - 1e-9) edges = np.round(np.arange(lo, hi + step / 2, step), 6) counts, _ = np.histogram(x, bins=edges) return {"edges": edges.tolist(), "counts": counts.tolist()} def descriptive(): e = ipo.copy() e["gap1"] = e.c1 / e.price - 1 # 抽籤期望值：中籤率 ×（首日收盤賣出的價差），淨額再扣申購處理費 20 元、中籤通知郵寄費 50 元、賣出手續費與證交稅 e["ev_ntd"] = e.lottery_rate * e["申購股數"] * (e.c1 - e.price) e["ev_net"] = e.lottery_rate * (e["申購股數"] * (e.c1 - e.price) - e["申購股數"] * e.c1 * 0.004425 - 50) - 20 e["rate_q"] = pd.qcut(e.lottery_rate.rank(method="first"), 5, labels=[1, 2, 3, 4, 5]) e["prem_q"] = pd.qcut(e.prem5.rank(method="first"), 5, labels=[1, 2, 3, 4, 5]) e["ex60_d1"] = [forward_excess(s, i, i + 60)[1] for s, i in zip(e.stock_id, e.i1)] e["ex120_d6"] = [forward_excess(s, i + 5, i + 125)[1] for s, i in zip(e.stock_id, e.i1)] ret3, ex3 = zip(*[forward_excess(s, i + 5, i + 755) for s, i in zip(e.stock_id, e.i1)]) out = { "n_ipo": int(len(e)), "first_listing": str(e.d1.min().date()), "last_listing": str(e.d1.max().date()), "lottery": { "gap1": summary(e.gap1), "rate": summary(e.lottery_rate), "ev_ntd": summary(e.ev_ntd), "ev_net": summary(e.ev_net), "below_issue_d1": round(float((e.c1 < e.price).mean()), 4), "by_year": {int(y): {"n": int(len(g)), "rate_median": round(float(g.lottery_rate.median()), 5), "gap1_median": round(float(g.gap1.median()), 4), "ev_median": round(float(g.ev_ntd.median()), 1), "ev_net_sum": round(float(g.ev_net.sum()), 0)} for y, g in e.groupby(e.d1.dt.year)}, "by_rate_quintile": {int(q): {"rate_max": round(float(g.lottery_rate.max()), 5), "gap1_median": round(float(g.gap1.median()), 4), "ev_median": round(float(g.ev_ntd.median()), 1), "ex60_d1": summary(g.ex60_d1)} for q, g in e.groupby("rate_q", observed=True)}, }, "premium_quintiles": {int(q): {"prem_lo": round(float(g.prem5.min()), 4), "prem_hi": round(float(g.prem5.max()), 4), **summary(g.ex120_d6)} for q, g in e.groupby("prem_q", observed=True)}, "three_year": {"return": summary(ret3), "excess": summary(ex3), "excess_hist": histogram(ex3, -1.2, 2.0, 0.1)}, } lock = {} for months in (6, 12): dates = lockup_dates((months,)) lock[f"{months}m"] = {"pre20": summary([forward_excess(s, a - 20, a)[1] for s, a in dates]), "post20": summary([forward_excess(s, a, a + 20)[1] for s, a in dates]), "around3": summary([forward_excess(s, a - 2, a + 1)[1] for s, a in dates])} # 對照組：同一批新股上市第 100～300 天裡隨機的 20 天，避開滿 6、12 個月前後 20 天 anniversaries = {} for s, a in lockup_dates(): anniversaries.setdefault(s, []).append(a) rng = np.random.default_rng(0) control = [] for s, i in zip(e.stock_id, e.i1): for _ in range(50): j = i + int(rng.integers(100, 300)) if all(j + 20 < a - 20 or j > a + 20 for a in anniversaries.get(s, [])): control.append(forward_excess(s, j, j + 20)[1]) break lock["control_any20_day100_300"] = summary(control) out["lockup"] = lock r = e[e.rotc_last_close.notna()].copy() r["pop_d1"] = r.c1 / r.rotc_last_close - 1 r["pop_d5"] = [close[s].iloc[i + 4] / c - 1 if i + 4 < len(days) else np.nan for s, i, c in zip(r.stock_id, r.i1, r.rotc_last_close)] out["emerging"] = {"n": int(len(r)), "share_of_ipos": round(len(r) / len(e), 4), "issue_vs_last_rotc": summary(r.price / r.rotc_last_close - 1), "pop_d1": summary(r.pop_d1), "pop_d5": summary(r.pop_d5), "pop_d1_hist": histogram(r.pop_d1, -0.6, 0.6, 0.05), "ex60_d1_by_rotc_ret60_pos": {str(k): summary(g.ex60_d1) for k, g in r.groupby(r.rotc_ret60 > 0)}} out["ex60_d1_by_rev_g3_ge_30"] = {str(k): summary(g.ex60_d1) for k, g in e[e.rev_g3.notna()].groupby(e.rev_g3 >= 0.3)} return out def sid(trades): """get_trades() 的 stock_id 是「代號 名稱」，只取代號。""" return trades["stock_id"].astype(str).str.split(" ").str[0] def trade_diff(a, b): """兩個報告的交易差異（以股票＋進場日比對）。""" def keyed(report): t = report.get_trades().copy() t["key"] = sid(t) + "|" + pd.to_datetime(t["entry_date"]).dt.strftime("%Y-%m-%d") return t ta, tb = keyed(a), keyed(b) only_a, only_b = ta[~ta.key.isin(tb.key)], tb[~tb.key.isin(ta.key)] return {"trades": int(len(ta)), "replaced": int(len(only_a)), "removed_mean_return": round(float(only_a["return"].mean()), 6) if len(only_a) else None, "added_mean_return": round(float(only_b["return"].mean()), 6) if len(only_b) else None, "common_mean_return": round(float(ta[ta.key.isin(tb.key)]["return"].mean()), 6)} # --------------------------------------------------------------------------- # 6. 執行 # --------------------------------------------------------------------------- def main(): base_by_n, base_reports = {}, {} for n in (8, 10, 12): base_reports[n] = mrm(n, "default") base_by_n[n] = compare(base_reports[n].creturn) base_by_n[n]["trades"] = trade_stats(base_reports[n]) results, reports = {}, {} for key in DESIGNS: results[key], reports[key] = validate(key, base_by_n) trades = base_reports[10].get_trades() age = (pd.to_datetime(trades["entry_date"]) - sid(trades).map(ipo.set_index("stock_id")["d1"])).dt.days young_share = {f"{y}y": round(float(((age >= 0) & (age < 365.25 * y)).mean()), 4) for y in (1, 3)} metrics = { "run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "data_end": str(bench.index[-1].date()), "base_strategy": {"name": "月營收動能（monthly-revenue-momentum）", "start": MRM_START, "by_n": {str(k): v for k, v in base_by_n.items()}, "halves_n10": halves(base_reports[10].creturn), "trades_in_ipo_years_n10": young_share}, "designs": results, "c5_vs_base_trades": trade_diff(base_reports[10], reports["C5"]), "descriptive": descriptive(), } with open("metrics.json", "w", encoding="utf-8") as f: json.dump(metrics, f, ensure_ascii=False, indent=1) curves = {key: r.creturn for key, r in reports.items()} curves["月營收動能10檔"] = base_reports[10].creturn frame = {} for key, c in curves.items(): c = active(c) frame[key] = c / c.iloc[0] frame = pd.DataFrame(frame) frame["0050含息"] = bench.reindex(frame.index) frame.index.name = "date" frame.round(6).to_csv("data.csv") print("完成：metrics.json、data.csv") if __name__ == "__main__": main()