# 多因子選股策略 — 營收動能 + 價格動能 + ROE 品質 + 低波動(rank-weighted 四因子)
# 截至 2026-06 回測。本檔產出文章中「四因子策略」的所有 CAGR / Sharpe / 回撤數字。
# 數據可追溯性(data_traceability):每個對外數字都來自此腳本的 sim() 實跑輸出,禁止編造。
# 環境:finlab(conda)。執行:
#   eval "$(/Users/cheng-yuhan/miniconda3/bin/conda shell.bash hook 2>/dev/null)" && conda activate finlab
#   python strategy.py
import warnings; warnings.filterwarnings("ignore")
import numpy as np, pandas as pd
from finlab import data
from finlab.backtest import sim

close = data.get('price:收盤價')
adj   = data.get('etl:adj_close')                                     # 含息,benchmark 鐵則用調整後收盤價
rev   = data.get('monthly_revenue:去年同月增減(%)')
roe   = data.get('fundamental_features:ROE稅後').index_str_to_date()  # 對齊財報「公布日」,避免前視偏差
START = '2018-01-01'


def monthly(df):
    """把任何頻率的資料對齊交易日後,取每月最後一筆。"""
    return df.reindex(close.index, method='ffill').resample('M').last()


revm = monthly(rev)                                  # 月營收年增 (%)
momm = monthly(close.pct_change(60))                 # 價格動能:近 60 交易日漲幅
roem = monthly(roe)                                  # ROE 品質(公布日對齊)
volm = monthly(close.pct_change().rolling(60).std()) # 低波動:近 60 日日報酬標準差(越低越好)

# 選股池:連續 3 個月營收正成長,且年增介於 10%~150%(下限濾雜訊,上限避開低基期失真)
pool = (revm > 10) & (revm < 150) & ((revm > 0).rolling(3).sum() == 3)


def rk(df, asc=True):
    """池內百分位排名 [0,1];asc=False 代表「越小越好」(用於低波動)。"""
    return df.where(pool).rank(axis=1, pct=True, ascending=asc).fillna(0)


# ---- 四因子 rank 合成:四個百分位分數相加 → 取 top 40 → 分數平方加權(集中持有強訊號)----
score  = rk(revm) + rk(momm) + rk(roem) + rk(volm, asc=False)
mask   = score.rank(axis=1, ascending=False) <= 40
masked = (score.where(mask, 0)) ** 2
w      = masked.div(masked.sum(axis=1).replace(0, np.nan), axis=0).fillna(0)
w      = w[w.index >= START]

# resample_offset='14D':每月營收約 10 號公布,延後 14 天換股,確保用「已公開」資訊(捕捉 PEAD 漂移)
report = sim(w, resample='M', resample_offset='14D', upload=False)
print("=== 四因子策略 ===")
print(report.get_stats())
# report.to_html('report_strategy.html')   # 互動式報告,上傳 R2 後以 {{embed:report_strategy.html}} 內嵌


# =====================================================================
# 對照組:各因子「單獨」回測(同一選股池、同樣 top 40、等權),量化「疊加」價值
# =====================================================================
def single(factor_rank):
    m  = factor_rank.rank(axis=1, ascending=False) <= 40
    ww = m.div(m.sum(axis=1).replace(0, np.nan), axis=0).fillna(0)
    ww = ww[ww.index >= START]
    return sim(ww, resample='M', resample_offset='14D', upload=False).get_stats()


for name, fr in [("營收動能", rk(revm)), ("價格動能", rk(momm)),
                 ("ROE 品質", rk(roem)), ("低波動", rk(volm, asc=False))]:
    s = single(fr)
    print(f"[單因子] {name}: CAGR={s['cagr']:.4f} Sharpe={s['daily_sharpe']:.2f} MDD={s['max_drawdown']:.4f}")

# =====================================================================
# 0050 benchmark(buy & hold,用 etl:adj_close 含息,鐵則)
# =====================================================================
bench = pd.DataFrame(index=adj.index)
bench['0050'] = 1.0
b = sim(bench[bench.index >= START], resample=None, upload=False,
        position_limit=1, fee_ratio=0)   # 直接持有 0050
print("=== 0050 含息 ===")
print(b.get_stats())

# =====================================================================
# 因子相關性矩陣(池內、月頻、橫斷面 rank 的時間序列相關)→ 證明因子互補
# =====================================================================
fr_dict = {"營收動能": rk(revm), "價格動能": rk(momm),
           "ROE品質": rk(roem), "低波動": rk(volm, asc=False)}
flat = {k: v.where(pool).stack() for k, v in fr_dict.items()}
corr = pd.DataFrame(flat).corr()
print("=== 因子相關性矩陣 ===")
print(corr.round(2))

# =====================================================================
# 交易成本敏感度:不同 fee_ratio(含滑價)對 CAGR 的拖累
# =====================================================================
print("=== 交易成本敏感度 ===")
for fee in [0.000, 0.001425, 0.003, 0.005]:
    r = sim(w, resample='M', resample_offset='14D', upload=False, fee_ratio=fee)
    print(f"fee={fee:.4%}  CAGR={r.get_stats()['cagr']:.4f}")

# =====================================================================
# 最近一期實際選出的個股(供文章列名單)
# =====================================================================
last = w.iloc[-1]
top = last[last > 0].sort_values(ascending=False).head(10)
print("=== 最近一期前 10 大持股(代號:權重)===")
print(top)