0056 進階版（高息＋營收成長＋均線，前 10 檔）

"""0056 進階版（高息＋營收成長＋均線，前 10 檔）：跑到最新資料的版本（/strategies/etf-0056-replication 的數字來源）。

/blog/etf-0056-replication 第六章「終極版」的程式（原文 Colab 範例檔的最後一版，拿掉市值前 150、流動性與毛利成長三個條件）：
- 條件：近四季 EPS 合計為正、近三年都有發放現金股利、近 3 個月營收平均高於近 12 個月、
  股價在 120 日與 240 日均線之上、60 日波動度排名在全市場後 50%、殖利率在全市場前 40%
- 排序：近兩年平均現金殖利率、最近一次現金殖利率、月營收年增率三項排名相加，取前 10 檔等權
- 每年 5/31 與 12/27 換股，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 = "0056 進階版（高息＋營收成長＋均線，前 10 檔）"
YIELD_QUANTILE = 0.6  # 殖利率至少在全市場 60 分位數（前 40%）
TOP_N = 10


def board_cash_dividend() -> FinlabDataFrame:
    """每股現金股利，日期取股東會日期（沒有就用董事會決議日）。"""
    df = data.get("board_dividend_announcement")
    dates = df["股東會日期"].fillna(df["董事會決議（擬議）股利分派日"])
    out = (df.drop_duplicates(["stock_id", "股東配發之現金(股利)總金額(元)"], keep="first")
             .groupby([df.stock_id, dates]).first()["盈餘分配之現金股利(元/股)"]
             .dropna().reset_index()
             .pivot(index="股東會日期", columns="stock_id", values="盈餘分配之現金股利(元/股)"))
    return FinlabDataFrame(out.rename_axis("date").rename_axis("symbol", axis=1))


def fill_missing_years(series: pd.Series) -> pd.Series:
    """沒有配息紀錄的年份補一筆 0，讓「連續三年」算得出中斷。"""
    if len(series) == 0:
        return series
    years = set(range(series.index[0].year, series.index[-1].year + 1))
    for year in years - set(series.index.year):
        series.loc[pd.Timestamp(f"{year}-01-01")] = 0
    return series.sort_index()


def paid_three_years() -> FinlabDataFrame:
    df = data.get("board_dividend_announcement").copy()
    df["股利年份"] = pd.to_datetime(df["股東會日期"]).fillna(pd.to_datetime(df["董事會決議（擬議）股利分派日"]))
    cash = (df.drop_duplicates(["股利年份", "stock_id"])
              .pivot(index="股利年份", columns="stock_id", values="股東配發之現金(股利)總金額(元)")
              .fillna(0).iloc[1:])
    paid = cash.apply(lambda s: (fill_missing_years(s[s != 0]) != 0).rolling(3).sum() == 3)
    return FinlabDataFrame(paid.ffill().fillna(False).astype(bool))


def build_report():
    close = data.get("price:收盤價")
    cash_dividend = board_cash_dividend().replace(0, np.nan).ffill().fillna(0)
    div_yield = cash_dividend / close
    div_annual = div_yield.rolling(520, min_periods=1).sum()

    eps_positive = data.get("financial_statement:每股盈餘").rolling(4).sum() > 0
    rev = data.get("monthly_revenue:當月營收")
    rev_yoy = data.get("monthly_revenue:去年同月增減(%)")
    yield_ratio = data.get("price_earning_ratio:殖利率(%)")

    rev_growth = rev.average(3) > rev.average(12)
    above_ma = (close > close.average(120)) & (close > close.average(240))
    # pandas 2 的 pct_change() 會先 ffill，這樣寫 pandas 2、3 結果一樣
    calm = close.ffill().pct_change(fill_method=None).rolling(60).std().rank(axis=1, pct=True) < 0.5
    high_yield = yield_ratio.ge(yield_ratio.quantile(YIELD_QUANTILE, axis=1), axis=0)

    conds = eps_positive & paid_three_years() & rev_growth & above_ma & calm & high_yield
    score = (div_annual.rank(axis=1, pct=True) + div_yield.rank(axis=1, pct=True)
             + rev_yoy.rank(axis=1, pct=True))
    position = score[conds].is_largest(TOP_N)

    years = range(position.index[0].year, position.index[-1].year + 1)
    dates = pd.to_datetime([f"{y}-{md}" for y in years for md in ("05-31", "12-27")])
    return sim(position.reindex(dates, method="ffill"), resample=dates, 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/etf-0056-replication/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()