高股息量化策略（股息率＋趨勢＋抗跌）

"""高股息量化策略：跑到最新資料的版本（/strategies/python-finlab-optimization 的數字來源）。

條件與 /blog/python-finlab-optimization 附的 strategy.py 相同，只把兩個參數拉到程式開頭：
- 滾動 520 日年化現金股息率（現金股利以股東會日期對齊，缺值退回董事會決議日）
- 收盤價站上 10、30…130 日均線的個數（趨勢分數）排名前 30%、500 日滾動最小回撤排名前 30%、
  當日成交量 > 50 張
- 取年化股息率最高的 20 檔，每季換股，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 pandas as pd

import finlab
from finlab import data
from finlab.backtest import sim
from finlab.dataframe import FinlabDataFrame

STRATEGY_NAME = "高股息量化策略（股息率＋趨勢＋抗跌）"
TOP_N = 20          # 持股檔數：熱圖 15、20、25
RANK_MIN = 0.7      # 趨勢分數與最小回撤的排名門檻：熱圖 0.6、0.7、0.8


def build_report():
    close = data.get("price:收盤價")
    volume = data.get("price:成交股數")

    announcement = data.get("board_dividend_announcement")
    dates = announcement["股東會日期"].fillna(announcement["董事會決議（擬議）股利分派日"])
    cash_dividend = (
        announcement
        .drop_duplicates(["stock_id", "股東配發之現金(股利)總金額(元)"], keep="first")
        .groupby([announcement.stock_id, dates])
        .first()["盈餘分配之現金股利(元/股)"]
        .dropna()
        .reset_index()
        .pivot(index="股東會日期", columns="stock_id", values="盈餘分配之現金股利(元/股)")
    )
    cash_dividend_annual = (cash_dividend / close).rolling(520, min_periods=1).sum()
    trend_score = sum([close > close.average(i) for i in range(10, 150, 20)])
    drawdown = close / close.cummax()
    min_drawdown = drawdown.rolling(500, min_periods=200).min()

    position = FinlabDataFrame(cash_dividend_annual)[
        (trend_score.rank(axis=1, pct=True) > RANK_MIN)
        & (min_drawdown.rank(axis=1, pct=True) > RANK_MIN)
        & (volume > 50_000)
    ].is_largest(TOP_N)
    return sim(position, resample="Q", 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]]

    mid = creturn.index[0] + (creturn.index[-1] - creturn.index[0]) / 2
    stats = {}
    for label, part in [("全期", creturn), ("前段", creturn.loc[:mid]), ("後段", creturn.loc[mid:])]:
        s = curve_stats(part)
        b = curve_stats(bench.loc[part.index[0]:part.index[-1]])
        stats[label] = (s, b)
        print(f"{label} {part.index[0].date()} 至 {part.index[-1].date()}："
              f"策略年化 {s['cagr']:.1%}、日夏普 {s['daily_sharpe']:.2f}、最大回撤 {s['max_drawdown']:.1%}；"
              f"0050 含息年化 {b['cagr']:.1%}、日夏普 {b['daily_sharpe']:.2f}、最大回撤 {b['max_drawdown']:.1%}")

    strategy_stats, bench_stats = stats["全期"]
    holdings = (report.position.fillna(0) != 0).sum(axis=1)
    strategy_stats["avg_holdings"] = holdings[holdings > 0].mean()

    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"{creturn.index[0].date()}/{creturn.index[-1].date()}",
        "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/python-finlab-optimization/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()