"""Smart 多因子（上市、分數前 25%）：/blog/smart-etf-00905-multi-factor-backtest 三項驗證用的程式。 原文「程式實作」的 Smart 多因子策略（去掉流動性限制），只是不截斷結束日、成本改用 sim() 預設： - 價值因子照原文程式；品質、動能因子原文沒附程式，照 00905 公開說明書（特選 Smart 多因子指數編製規則）寫： - 價值：營業現金流量對股價比、EBITDA 對企業價值比、益本比、股東收益率 - 品質：股東權益報酬率、收益變動率、現金流量對資產比、資產報酬率、毛利對資產比、槓桿度 - 動能：股價動能、盈利動能（原文稱營利動能） - 金融股與非金融股用不同權重（原文程式把兩條價值公式的標註寫反了，這裡照文字與公開說明書） - 多因子分數＝0.4 品質＋0.3 價值＋0.3 動能，轉成多因子權重係數 - 只留上市股票，係數高於 75 分位數的依係數加權，從 2020-08-10 起算，權重每天變動就調整（resample=None） - 長期負債用非流動負債、稅後純益用歸屬母公司淨利、收益變動率用近 20 季的近四季 EPS 年增率標準差，無風險利率設 0 - 原文不計成本，這裡用 sim() 預設手續費 0.1425% 與證交稅 0.3% 統計口徑：日夏普 = 日報酬平均 / 標準差 × √252，不扣無風險利率； 0050 含息 = etl:adj_close 還原價買進持有，與策略同一段期間。 執行後會在目前目錄輸出 data.csv（淨值曲線）與 metrics.json（統計）。 pip install finlab python strategy.py finlab 需要資料時會自動引導登入，不需要在程式裡填 token。 """ import json import numpy as np import pandas as pd import finlab from finlab import data from finlab.backtest import sim from finlab.dataframe import FinlabDataFrame STRATEGY_NAME = "Smart 多因子（上市、分數前 25%）" QUANTILE = 0.75 LONG_MONTHS = 12 START = "2020-08-10" def Z(df: pd.DataFrame) -> pd.DataFrame: """常態標準化：每天橫向 (x - 平均) / 標準差，超過 ±3 的當離群值不採納（同原文的 Z 函式）。""" df = df.replace([np.inf, -np.inf], np.nan) z = df.sub(df.mean(axis=1), axis=0).div(df.std(axis=1, ddof=0), axis=0) return z.where(z.abs() <= 3) def to_weight(z: pd.DataFrame) -> pd.DataFrame: """多因子權重係數：Z ≥ 0 時 1 + Z，Z < 0 時 (1 - Z)^-1。""" return z.where(z < 0, 1 + z).where(z >= 0, 1 / (1 - z)) def price_momentum(adj: pd.DataFrame, index: pd.Index) -> FinlabDataFrame: """股價動能＝0.5 ×（近 6 個月報酬＋近 LONG_MONTHS 個月報酬）÷ 近三年週報酬年化標準差（無風險利率設 0）。""" weekly = adj.resample("W-FRI").last() sigma = (weekly.ffill().pct_change(fill_method=None).rolling(156, min_periods=104).std() * np.sqrt(52)).resample("ME").last() m = adj.resample("ME").last() mom = 0.5 * ((m / m.shift(6) - 1) + (m / m.shift(LONG_MONTHS) - 1)) / sigma return mom.reindex(index, method="ffill") def build_report(): with data.universe(market="TSE_OTC", category="金融"): fin_cols = data.get("price:收盤價").columns with data.universe(market="TSE"): tse_cols = data.get("price:收盤價").columns with data.universe(market="TSE_OTC"): close = data.get("price:收盤價") adj = data.get("etl:adj_close") fs = lambda name: data.get(f"financial_statement:{name}") # noqa: E731 ocf, capital, eps = fs("營業活動之淨現金流入_流出"), fs("股本"), fs("每股盈餘") revenue, debt, cash = fs("營業收入淨額"), fs("負債總額"), fs("期末現金及約當現金餘額") assets, equity, gross = fs("資產總額"), fs("股東權益總額"), fs("營業毛利") net_income, long_debt = fs("歸屬母公司淨利_損"), fs("非流動負債").fillna(0) preferred = fs("特別股股本").fillna(0) # 沒發特別股的公司是空值 ebitda_margin = data.get("fundamental_features:稅前息前折舊前淨利率") dividend_yield = data.get("price_earning_ratio:殖利率(%)") shares = data.get("internal_equity_changes:發行股數") def daily(frame): return frame.index_str_to_date().reindex(close.index, method="ffill") share_count = capital / 10 # 股本 ÷ 面額 10 元 market_value = share_count * close # 價值因子 ocf_price = (ocf / share_count) / close ebitda_ev = (ebitda_margin * revenue / 100) / (market_value + debt - cash) earnings_yield = eps / close shareholder_yield = dividend_yield + (shares - shares.shift(12)) * (close / market_value) z_ocf, z_ey, z_sy = Z(ocf_price), Z(earnings_yield), Z(shareholder_yield) value_fin = 0.4 * z_ocf + 0.4 * z_ey + 0.2 * z_sy value = 0.3 * z_ocf + 0.3 * Z(ebitda_ev) + 0.3 * z_ey + 0.1 * z_sy # 品質因子 roe = daily(eps / (equity / share_count)) eps_4q = eps.rolling(4).sum() growth = (eps_4q / eps_4q.shift(4) - 1).replace([np.inf, -np.inf], np.nan) earnings_var = daily(-growth.rolling(20, min_periods=8).std()) roa = daily(net_income / assets) leverage = -(((market_value + long_debt + preferred) / market_value) + daily((equity + long_debt + preferred) / equity) + daily(debt / assets)) / 3 z_roe, z_var = Z(roe), Z(earnings_var) quality_fin = (z_roe + z_var + Z(roa)) / 3 quality = 0.2 * (z_roe + z_var + Z(daily(ocf / assets)) + Z(daily(gross / assets)) + Z(leverage)) # 動能因子 earnings_mom = daily((eps - eps.shift(1)) / close) momentum = 0.5 * (Z(price_momentum(adj, close.index)) + Z(earnings_mom)) def by_sector(frame, frame_fin): frame, frame_fin = daily(frame), daily(frame_fin) cols = frame.columns.intersection(fin_cols) frame[cols] = frame_fin[cols] return frame score = 0.4 * by_sector(quality, quality_fin) + 0.3 * by_sector(value, value_fin) + 0.3 * momentum weight = to_weight(score) weight = weight[weight.columns.intersection(tse_cols)] weight = weight.where(weight.ge(weight.quantile_row(QUANTILE), axis=0), 0) return sim(weight.loc[START:], resample=None, upload=False, name=STRATEGY_NAME) def curve_stats(curve: pd.Series) -> dict: curve = curve / curve.iloc[0] daily = curve.pct_change().dropna() years = (curve.index[-1] - curve.index[0]).days / 365.25 return { "cagr": curve.iloc[-1] ** (1 / years) - 1, "daily_sharpe": daily.mean() / daily.std() * 252 ** 0.5, "max_drawdown": (curve / curve.cummax() - 1).min(), "total_return": curve.iloc[-1] - 1, } def main() -> None: finlab.login() bench_all = data.get("etl:adj_close")["0050"].dropna() report = build_report() # 從第一次有部位的那天起算，0050 含息用同一段期間 creturn = report.creturn.dropna() first_move = (creturn != creturn.iloc[0]).argmax() creturn = creturn.iloc[max(first_move - 1, 0):] bench = bench_all.loc[creturn.index[0]:creturn.index[-1]] strategy_stats = curve_stats(creturn) bench_stats = curve_stats(bench) holdings = (report.position.fillna(0) != 0).sum(axis=1) strategy_stats["avg_holdings"] = holdings[holdings > 0].mean() start, end = creturn.index[0].date(), creturn.index[-1].date() print(f"期間 {start} 至 {end}") for label, s in [("策略", strategy_stats), ("0050 含息", bench_stats)]: print(f"{label}：年化 {s['cagr']:.2%}，日夏普 {s['daily_sharpe']:.2f}，最大回撤 {s['max_drawdown']:.2%}") curves = pd.DataFrame({STRATEGY_NAME: creturn / creturn.iloc[0], "0050含息": bench / bench.iloc[0]}) curves.index.name = "date" curves.round(6).to_csv("data.csv") metrics = { "window": f"{start}/{end}", "star": STRATEGY_NAME, "variants": {STRATEGY_NAME: {k: round(float(v), 6) for k, v in strategy_stats.items()}}, "benchmark_0050": {k: round(float(v), 6) for k, v in bench_stats.items()}, "run_date": str(pd.Timestamp.today().date()), "finlab_version": finlab.__version__, "generated_by": "static/blog/smart-etf-00905-multi-factor-backtest/validation/strategy.py", } with open("metrics.json", "w", encoding="utf-8") as f: json.dump(metrics, f, ensure_ascii=False, indent=2) if __name__ == "__main__": main()