""" 台股 10 個經典選股法與論文異常的三項驗證： 短期反轉、長期反轉、盈餘驚喜 SUE、價值綜合、品質減垃圾 QMJ、低換手率、現金流殖利率、 神奇公式、葛拉漢防禦型、CANSLIM 台股版 每個訊號做兩種版本： 單獨版：依訊號取前 n 檔（n = 20／30／50）等權，每月最後一個交易日換股； 神奇公式每季換股，葛拉漢與 CANSLIM 持有所有符合條件的股票（CANSLIM 每週換股）。 組合版：疊在月營收動能策略（/strategies/monthly-revenue-momentum）上，持股 N = 8／10／12。 1–8 號：當天在訊號最好的 30% 的候選股排前面；9、10 號：符合單獨版全部條件的排前面。 三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）： 1. 參數高原：兩個主要參數各 3 個值共 9 組（CANSLIM 三個參數 27 組），七成以上要贏。 2. 前後兩段：回測期間對半切，兩段各自要贏。 3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。 組合版另要比底好才算新策略：原參數日夏普高於同持股數的月營收動能單獨、9 組至少 7 組高於同持股數的底、 前後兩段都高於底的同一段，而且持股（持有日加權）至少 20% 跟底不同。 股票池（單獨版）：上市櫃普通股（四碼數字，含已下市），排除金融保險與存託憑證，20 日平均成交金額至少 2,000 萬元。 財報：損益與現金流都是單季值，各公司實際公布財報那天才用得到（index_str_to_date）。 期間：2016-04-01 起；SUE 2017-06-01、葛拉漢 2017-04-01、QMJ 2018-04-01 起（最長的鄰近參數要等財報累積）。 0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。 執行：安裝 finlab 後直接跑，資料需要時會自動引導登入，約 15 分鐘。 輸出：metrics.json（每組參數的結果）、data.csv（各版本原參數與 0050 含息的每日淨值）。 """ import datetime import itertools import json import math from functools import cache import numpy as np import pandas as pd import finlab from finlab import data from finlab.backtest import sim START = "2016-04-01" START_SUE = "2017-06-01" # 前 12 季的淨利年增標準差：2014-Q1 起的年增，到 2017-Q1 財報（5/15）才湊滿 START_GRAHAM = "2017-04-01" # 近 4 個四季淨利：2013-Q1 起，到 2016 年報（3/31）才湊滿 START_QMJ = "2018-04-01" # 獲利指標的 4 年變化：到 2017 年報才算得出 MIN_TURNOVER_VALUE = 2e7 HOLDINGS = [20, 30, 50] BASE_HOLDINGS = [8, 10, 12] BASE_CENTER = 10 # 組合版與月營收動能單獨的原參數持股數 BEST = 0.3 # 組合版加分：訊號最好的 30% THOUSAND = 1000 # 財報單位是千元，市值是元 def plain(key): df = pd.DataFrame(data.get(key)) df.index = pd.to_datetime(df.index) return df # --------------------------------------------------------------------------- # 1. 股票池、交易日、市值與 0050 含息 # --------------------------------------------------------------------------- adj_all = plain("etl:adj_close") bench = adj_all["0050"].dropna() categories = pd.DataFrame(data.get("security_categories")).set_index("stock_id")["category"] EXCLUDED_CATEGORIES = {"金融保險業", "金融業", "存託憑證"} def is_common_stock(sid): """4 位數普通股；已下市的公司沒有產業別，用代號排除金融（28xx）與存託憑證（91xx）。""" return ( len(sid) == 4 and sid.isdigit() and not sid.startswith(("0", "28", "91")) and categories.get(sid) not in EXCLUDED_CATEGORIES ) stocks = [s for s in adj_all.columns if is_common_stock(s)] adj = adj_all[stocks] days = adj.loc["2015-01-01":].index month_ends = days[~days.to_period("M").duplicated(keep="last")] quarter_ends = days[~days.to_period("Q").duplicated(keep="last")] def daily(frame): return frame.reindex(columns=stocks).reindex(days) def fill_to_days(frame): """展開到每個交易日，沿用最近一筆資料。""" return frame.reindex(frame.index.union(days)).ffill().reindex(days) avg_turnover_value = plain("price:成交金額").reindex(columns=stocks).rolling(20).mean().reindex(days) liquid = avg_turnover_value > MIN_TURNOVER_VALUE market_value = daily(plain("etl:market_value")) market_index = plain("benchmark_return:發行量加權股價報酬指數").iloc[:, 0] SENTINEL = -9e15 def on_days(quarterly): """季資料換成各公司實際公布日，展開到每個交易日。 不符合條件（NaN）的季先填成哨兵值：index_str_to_date 遇到 NaN 會沿用上一季的值， 被遮掉的季就會被補回來。 """ dated = pd.DataFrame(quarterly.astype(float).fillna(SENTINEL).index_str_to_date()) dated.index = pd.to_datetime(dated.index) out = fill_to_days(dated.reindex(columns=stocks)) return out.where(out > SENTINEL / 2) def statement(key): return data.get(f"financial_statement:{key}") def ttm(frame): """近四季合計。""" return frame.rolling(4).sum() net_income = statement("歸屬母公司淨利_損") operating_cash_flow = statement("營業活動之淨現金流入_流出") gross_profit, revenue_q, operating_income = statement("營業毛利"), statement("營業收入淨額"), statement("營業利益") total_assets, equity, total_liabilities = statement("資產總額"), statement("母公司股東權益合計"), statement("負債總額") current_assets, current_liabilities = statement("流動資產"), statement("流動負債") eps = statement("每股盈餘") positive_equity = equity.where(equity > 0) net_income_12m, eps_12m = ttm(net_income), ttm(eps) # --------------------------------------------------------------------------- # 2. 十個訊號：參數 → (每日分數, 由低到高?)；9、10 號是每日布林 # --------------------------------------------------------------------------- def short_term_reversal(lookback_days): """Jegadeesh (1990)：近 L 個交易日跌最多的股票。""" return daily(adj / adj.shift(lookback_days) - 1), True def long_term_reversal(lookback_months): """De Bondt & Thaler (1985)：近 L 個月（每月 21 個交易日）跌最多的股票。""" return daily(adj / adj.shift(21 * lookback_months) - 1), True def earnings_surprise(window): """Bernard & Thomas (1989) 的 SUE：本季淨利年增 ÷ 前 W 季同一差額的標準差。 用歸屬母公司淨利代替 EPS：台股配股會讓每股數字跳動，標準化後的比值不受股數影響。 """ change = net_income - net_income.shift(4) spread = change.shift(1).rolling(window).std() return on_days(change / spread.where(spread > 0)), False def percentile(frame, ascending=True): return frame.rank(axis=1, pct=True, ascending=ascending) def annualized_yield(flow, quarters): """近 q 季合計年化後 ÷ 市值。""" return on_days(flow.rolling(quarters).sum() * 4 / quarters) * THOUSAND / market_value def value_composite(quarters): """Lakonishok, Shleifer & Vishny (1994)：淨值／市值、盈餘／市值、營業現金流／市值的百分位平均。""" book_to_market = on_days(equity) * THOUSAND / market_value earnings_yield = annualized_yield(net_income, quarters) cash_yield = annualized_yield(operating_cash_flow, quarters) return (percentile(book_to_market) + percentile(earnings_yield) + percentile(cash_yield)) / 3, False def zscore(frame, higher_is_better=True): """每天把變數換成橫斷面排名，再標準化成 z 分數（Asness 等人的做法）。""" rank = frame.rank(axis=1, ascending=higher_is_better) return rank.sub(rank.mean(axis=1), axis=0).div(rank.std(axis=1), axis=0) @cache def quality_parts(): """QMJ 不隨參數變動的部分：獲利、安全、配發。""" gp, ni, cfo, rev = ttm(gross_profit), net_income_12m, ttm(operating_cash_flow), ttm(revenue_q) profitability = { "gpoa": gp / total_assets, "roe": ni / positive_equity, "roa": ni / total_assets, "cfoa": cfo / total_assets, "gmar": gp / rev.where(rev > 0), "accruals": (cfo - ni) / total_assets, # 應計越低越好：營業現金流高於淨利 } stock_return = adj.pct_change(fill_method=None) market_return = market_index.pct_change().reindex(stock_return.index) def mean(x): return x.rolling(252, min_periods=200).mean() beta = (mean(stock_return.mul(market_return, axis=0)).sub(mean(stock_return).mul(mean(market_return), axis=0)) .div(market_return.rolling(252, min_periods=200).var(ddof=0), axis=0)) roe_volatility = (net_income / positive_equity).rolling(12).std() dividends_paid = -ttm(statement("發放現金股利").fillna(0)).where(ni.notna()) payout = (dividends_paid / ni.where(ni > 0)).where(ni > 0, 0.0) # 虧損的公司記 0 return dict( profitability=profitability, profitability_z=sum(zscore(on_days(v)) for v in profitability.values()) / len(profitability), safety_z=(zscore(daily(beta), False) + zscore(on_days(total_liabilities / total_assets), False) + zscore(on_days(roe_volatility), False)) / 3, payout_z=zscore(on_days(payout)), ) def quality_minus_junk(growth_years): """Asness, Frazzini & Pedersen (2019) 的簡化版：獲利、成長、安全、配發四個面向的 z 分數加總。""" parts = quality_parts() growth = sum(zscore(on_days(v - v.shift(4 * growth_years))) for k, v in parts["profitability"].items() if k != "accruals") / 5 return (zscore(parts["profitability_z"]) + zscore(growth) + zscore(parts["safety_z"]) + zscore(parts["payout_z"])), False shares_outstanding = plain("foreign_investors_shareholding:發行股數").reindex(columns=stocks) volume = plain("price:成交股數").reindex(columns=stocks) daily_turnover = (volume / shares_outstanding.reindex(volume.index).ffill()).replace([np.inf, -np.inf], np.nan) def low_turnover(lookback_days): """Datar, Naik & Radcliffe (1998)：近 L 日平均換手率（成交股數 ÷ 發行股數）最低。""" return daily(daily_turnover.rolling(lookback_days, min_periods=int(lookback_days * 0.8)).mean()), True def cash_flow_yield(quarters): """近 q 季營業現金流年化 ÷ 市值。""" return annualized_yield(operating_cash_flow, quarters), False def largest(top, universe=liquid): """股票池裡市值前 top 大。""" return market_value.where(universe).rank(axis=1, ascending=False) <= top @cache def magic_parts(): """盈餘殖利率 = EBIT ÷ 企業價值；資本報酬率 = EBIT ÷（淨營運資金 + 不動產廠房設備）。""" def item(key): """沒列的科目當 0。""" return statement(key).fillna(0) short_debt = item("短期借款") + item("應付商業本票∕承兌匯票") + item("一年內到期長期負債") debt = short_debt + item("應付公司債_非流動") + item("長期借款") # 長期借款與銀行借款_非流動是同一筆，只取一次 cash = item("現金及約當現金") reported = total_assets.notna() ebit = ttm(operating_income) net_claims = (debt + item("非控制權益") - cash).where(reported) enterprise_value = market_value + on_days(net_claims) * THOUSAND working_capital = (current_assets - cash) - (current_liabilities - short_debt) capital = (working_capital.clip(lower=0) + item("不動產廠房及設備")).where(reported) return dict( earnings_yield=on_days(ebit) * THOUSAND / enterprise_value.where(enterprise_value > 0), return_on_capital=on_days(ebit / capital.where(capital > 0)), ) utility = pd.Series([categories.get(s) == "油電燃氣業" for s in stocks], index=stocks) liquid_ex_utility = liquid & ~utility def magic_formula(top): """Greenblatt (2005)：流動股票裡市值前 top 大、排除油電燃氣（書中排除金融與公用事業）， 盈餘殖利率排名＋資本報酬率排名。""" parts, universe = magic_parts(), largest(top, liquid_ex_utility) return (percentile(parts["earnings_yield"].where(universe)) + percentile(parts["return_on_capital"].where(universe))), False @cache def dividend_in_past_year(): """近 252 個交易日內有現金股利除息。""" events = [] for key in ("dividend_tse:權息", "dividend_otc:權息"): kind = plain(key).reindex(columns=stocks) cash = kind.apply(lambda col: col.astype(str).str.contains("息")).astype(float) events.append(cash.reindex(adj.index).fillna(0)) return (sum(events) > 0).astype(float).rolling(252, min_periods=1).sum() > 0 def paid_dividends_every_year(years): """近 years 段（每段 252 個交易日）每段都有現金股利除息。""" paid = dividend_in_past_year() ok = paid for k in range(1, years): ok = ok & paid.shift(252 * k, fill_value=False) return daily(ok) @cache def graham_valuation(): """不隨參數變動的部分：PE×PB，以及流動股票市值前 500 大裡 PE ≤ 15（PE 用近 3 個四季淨利的平均）。""" ni = net_income_12m average_3y = (ni + ni.shift(4) + ni.shift(8)) / 3 pe = market_value / (on_days(average_3y.where(average_3y > 0)) * THOUSAND) pb = market_value / (on_days(positive_equity) * THOUSAND) return pe * pb, (pe <= 15) & largest(500) & liquid @cache def graham_record(years): """財務保守，而且近 Y 年獲利、EPS 與現金股利都有紀錄。""" ok = ((current_assets >= 2 * current_liabilities) & (statement("非流動負債").fillna(0) <= current_assets - current_liabilities)) for k in range(years): ok = ok & (net_income_12m.shift(4 * k) > 0) # 近 Y 個四季淨利都 > 0 ok = ok & (eps_12m >= eps_12m.shift(4 * years)) # 近四季 EPS 不低於 Y 年前 return (on_days(ok) > 0.5) & paid_dividends_every_year(years) def graham_defensive(max_pe_pb, years): """Graham (1949) 防禦型投資人條件中，台股資料檢查得到的部分。""" pe_pb, cheap_and_large = graham_valuation() return graham_record(years) & cheap_and_large & (pe_pb <= max_pe_pb) @cache def canslim_parts(): year_ago = eps.shift(4) quarter_growth = (eps / year_ago.where(year_ago > 0) - 1).where(eps > 0) # C annual_growth = eps_12m / eps_12m.shift(4).where(eps_12m.shift(4) > 0) - 1 # A revenue_3m = plain("monthly_revenue:當月營收").reindex(columns=stocks).rolling(3).sum() revenue_growth = revenue_3m / revenue_3m.shift(12).where(revenue_3m.shift(12) > 0) - 1 # C（台股月營收） to_high = daily(adj / adj.rolling(252, min_periods=200).max()) # N strength = daily(sum(w * (adj / adj.shift(d) - 1) # L for w, d in ((0.4, 63), (0.2, 126), (0.2, 189), (0.2, 252)))) foreign = plain("institutional_investors_trading_summary:外陸資買賣超股數(不含外資自營商)").reindex(columns=stocks) trust = plain("institutional_investors_trading_summary:投信買賣超股數").reindex(columns=stocks) return dict( quarter_growth=on_days(quarter_growth), annual_growth=on_days(annual_growth), revenue_growth=fill_to_days(revenue_growth), to_high=to_high, rs_rank=percentile(strength.where(liquid)), institutions=daily((foreign.fillna(0) + trust.fillna(0)).rolling(20).sum() > 0), # I market_up=(market_index > market_index.rolling(200).mean()).reindex(days, fill_value=False), # M ) def canslim(growth, rs_min, near_high): """O'Neil (1988) CANSLIM 的台股版：C、A、N、L、I、M，加上月營收。""" c = canslim_parts() picks = ((c["quarter_growth"] >= growth) & (c["annual_growth"] >= growth) & (c["revenue_growth"] >= growth) & (c["to_high"] >= near_high) & (c["rs_rank"] >= rs_min) & c["institutions"] & liquid) return picks.mul(c["market_up"], axis=0).astype(bool) # --------------------------------------------------------------------------- # 3. 單獨版與組合版 # --------------------------------------------------------------------------- def top_n(score, n, ascending, dates): """換股日依分數取前 n 檔；同分時 20 日成交金額大的優先。""" score = score.reindex(dates).where(liquid.reindex(dates)) rank = score.rank(axis=1, ascending=ascending, method="average") tie = percentile(avg_turnover_value.reindex(dates).where(score.notna()), ascending=False) * 0.5 return (rank + tie).rank(axis=1, method="first") <= n @cache def score_of(signal, value): """同一組訊號與參數只算一次，單獨版各持股數、不扣成本對照與組合版共用；換下一個訊號前清掉。""" return signal(value) def ranked(title, signal, start, param, center, dates=month_ends, resample="M"): """依訊號排名：單獨版換股日取前 n 檔；組合版加分＝當天在訊號最好的 30%（流動股票池內的百分位）。""" def position(value, n): score, ascending = score_of(signal, value) return top_n(score, n, ascending, dates) def good(value): score, ascending = score_of(signal, value) return percentile(score.where(liquid), ascending) <= BEST return dict(title=title, start=start, position=position, resample=resample, good=good, params=[param, ("持股 n", HOLDINGS)], center=(center, 30)) def screened(title, rule, start, params, center, dates, resample): """持有所有符合條件的股票；組合版加分＝符合條件（第一個參數以外用原參數）。""" return dict(title=title, start=start, position=lambda *values: rule(*values).reindex(dates), resample=resample, good=lambda value: rule(value, *center[1:]), params=params, center=center) # 月營收動能（/strategies/monthly-revenue-momentum）：候選股不限普通股、沒有成交金額門檻 close = data.get("price:收盤價") monthly_revenue = data.get("monthly_revenue:當月營收") trend = (close > close.average(20)) & (close > close.average(60)) & (close > close.average(120)) revenue_up = monthly_revenue.average(3) > monthly_revenue.average(12) momentum = (close / close.shift() - 1).rolling(5).mean() base_score = momentum[trend & revenue_up] def revenue_momentum(score, n, start, **cost): position = score.is_largest(n).loc[start:] return sim(position, resample="M", resample_offset="11D", stop_loss=0.1, upload=False, **cost) def boosted(good): """組合版的分數：符合加分條件的候選股排前面。""" preferred = good.astype(float).reindex(index=close.index, columns=close.columns).ffill().fillna(0) return base_score + preferred SIGNALS = { "short-term-reversal": ranked("短期反轉", short_term_reversal, START, ("回看 L 日", [10, 20, 30]), 20), "long-term-reversal": ranked("長期反轉", long_term_reversal, START, ("回看 L 月", [24, 36, 48]), 36), "earnings-surprise": ranked("盈餘驚喜 SUE", earnings_surprise, START_SUE, ("標準差 W 季", [4, 8, 12]), 8), "value-composite": ranked("價值綜合", value_composite, START, ("盈餘與現金流 q 季", [2, 4, 8]), 4), "quality-minus-junk": ranked("品質減垃圾 QMJ", quality_minus_junk, START_QMJ, ("成長 G 年", [2, 3, 4]), 3), "low-turnover": ranked("低換手率", low_turnover, START, ("回看 L 日", [20, 60, 120]), 60), "cash-flow-yield": ranked("現金流殖利率", cash_flow_yield, START, ("現金流 q 季", [2, 4, 8]), 4), "magic-formula": ranked("神奇公式", magic_formula, START, ("市值前 M 大", [300, 500, 800]), 500, dates=quarter_ends, resample="Q"), "graham-defensive": screened("葛拉漢防禦型", graham_defensive, START_GRAHAM, [("PE×PB 上限", [15, 22.5, 30]), ("紀錄 Y 年", [2, 3, 4])], (22.5, 3), dates=month_ends, resample="M"), "canslim": screened("CANSLIM 台股版", canslim, START, [("成長門檻 g", [0.20, 0.25, 0.30]), ("相對強度 r", [0.7, 0.8, 0.9]), ("距高點 h", [0.80, 0.85, 0.90])], (0.25, 0.8, 0.85), dates=days, resample="W"), } # --------------------------------------------------------------------------- # 4. 統計與三項驗證 # --------------------------------------------------------------------------- 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 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": curve.index[0].strftime("%Y-%m-%d"), "end": curve.index[-1].strftime("%Y-%m-%d"), "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), "total_return": round(float(curve.iloc[-1] - 1), 6), } def versus_0050(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 return [versus_0050(curve.loc[:mid]), versus_0050(curve.loc[mid:])] def avg_holdings(report): held = (report.position.fillna(0) != 0).sum(axis=1) return round(float(held[held > 0].mean()), 1) def annual_turnover(report): """單邊年周轉率：每次換股權重變動絕對值總和的一半，加總後除以年數（不含停損賣出）。""" position = report.position.fillna(0) weight = position.div(position.abs().sum(axis=1).replace(0, np.nan), axis=0).fillna(0) curve = active(report.creturn) years = (curve.index[-1] - curve.index[0]).days / 365.25 return round(float(weight.diff().abs().sum(axis=1).sum() / 2 / years), 2) def differs_from(report, other): """持有日加權：report 的持股中，同一天 other 沒有持有的比例。""" a = report.position.fillna(0) != 0 b = (other.position.fillna(0) != 0).reindex(index=a.index, columns=a.columns, fill_value=False) return round(float(1 - (a.values & b.values).sum() / a.values.sum()), 4) ZERO_COST = {"fee_ratio": 0, "tax_ratio": 0} def validate(run, params, center): """所有參數組合各跑一次；原參數另跑一次不扣成本。回傳 (結果, 原參數回測)。""" grid = [] for combination in itertools.product(*[values for _, values in params]): report = run(*combination) row = {"params": list(combination), "avg_holdings": avg_holdings(report), **versus_0050(report.creturn)} grid.append(row) if combination == tuple(center): base, center_report = row, report print(combination, f"sharpe {row['strategy']['sharpe']:.3f} vs 0050 {row['bench']['sharpe']:.3f}", flush=True) result = { "param_names": [name for name, _ in params], "center": list(center), "grid": grid, "grid_wins": sum(r["win"] for r in grid), "grid_n": len(grid), "base": base, "halves": halves(center_report.creturn), "zero_cost": versus_0050(run(*center, **ZERO_COST).creturn), "annual_turnover": annual_turnover(center_report), } result["pass"] = { "neighbors": result["grid_wins"] / result["grid_n"] >= 0.7, "halves": all(h["win"] for h in result["halves"]), "cost": base["win"], } result["pass"]["all"] = all(result["pass"].values()) return result, center_report results = {"single": {}, "combo": {}, "base_alone": {}} curves = {} base_reports = {} # 底策略單獨（組合版要比它好） for start in sorted({spec["start"] for spec in SIGNALS.values()}): rows = {} for n in BASE_HOLDINGS: report = revenue_momentum(base_score, n, start) rows[n] = {**versus_0050(report.creturn), "halves": halves(report.creturn)} if n == BASE_CENTER: base_reports[start] = report curves[f"revenue-momentum-{start[:7]}"] = active(report.creturn) rows[n]["zero_cost"] = versus_0050(revenue_momentum(base_score, n, start, **ZERO_COST).creturn) results["base_alone"][start] = rows for key, spec in SIGNALS.items(): start, params, center = spec["start"], spec["params"], spec["center"] print(f"== {spec['title']}", flush=True) def run_single(*values, **cost): return sim(spec["position"](*values).loc[start:], resample=spec["resample"], upload=False, **cost) res, report = validate(run_single, params, center) res.update(title=spec["title"], start=start) results["single"][key] = res curves[key] = active(report.creturn) combo_score = cache(lambda value: boosted(spec["good"](value))) res, report = validate(lambda value, n, **cost: revenue_momentum(combo_score(value), n, start, **cost), [params[0], ("持股 N", BASE_HOLDINGS)], (center[0], BASE_CENTER)) score_of.cache_clear() base = results["base_alone"][start] base_center = base[BASE_CENTER] above_base = sum(r["strategy"]["sharpe"] > base[r["params"][1]]["strategy"]["sharpe"] for r in res["grid"]) halves_above = [res["halves"][i]["strategy"]["sharpe"] > base_center["halves"][i]["strategy"]["sharpe"] for i in (0, 1)] differs = differs_from(report, base_reports[start]) res.update(title=f"月營收動能＋{spec['title']}", start=start, above_base=above_base, halves_above_base=halves_above, differs_from_base=differs) res["pass"]["beats_base"] = bool(res["base"]["strategy"]["sharpe"] > base_center["strategy"]["sharpe"] and above_base >= 7 and all(halves_above) and differs >= 0.2) res["pass"]["new_strategy"] = res["pass"]["all"] and res["pass"]["beats_base"] results["combo"][key] = res curves[f"{key}+revenue-momentum"] = active(report.creturn) meta = { "run_date": str(datetime.date.today()), "finlab_version": finlab.__version__, "pandas_version": pd.__version__, "data_end": str(bench.index[-1].date()), "liquidity_filter_twd": MIN_TURNOVER_VALUE, "benchmark": "0050 含息買進持有（etl:adj_close，不含交易成本）", "sharpe": "日報酬平均 / 標準差 × √252，未扣無風險利率", } with open("metrics.json", "w", encoding="utf-8") as f: json.dump({"meta": meta, **results}, f, ensure_ascii=False, indent=1) table = pd.DataFrame({k: v / v.iloc[0] for k, v in curves.items()}) bench_part = bench.loc[table.index[0]:] table["0050含息"] = bench_part / bench_part.iloc[0] table.round(6).to_csv("data.csv", index_label="date") for group in ("single", "combo"): for res in results[group].values(): b = res["base"] line = (f"{group:6s} {res['title']}: 日夏普 {b['strategy']['sharpe']:.2f}（0050 {b['bench']['sharpe']:.2f}），" f"{res['grid_n']} 組贏 {res['grid_wins']} 組，三項 {'通過' if res['pass']['all'] else '沒過'}") if group == "combo": line += f"，新策略 {'是' if res['pass']['new_strategy'] else '否'}" print(line)