杜邦週轉率選股（週轉率＋ROE＋PB）

"""杜邦週轉率選股（2020 年原文策略）：跑到最新資料的版本（/strategies/dupont-analysis-stock-picking-intro 的數字來源）。

條件與 /blog/dupont-analysis-stock-picking-intro 附的 strategy.py 相同，只是不截斷結束日：
- 股票池：上市櫃排除金融股
- 單季總資產週轉率 > 0.3、單季 ROE > 5%、財報公告時的股價淨值比 < 3
- 符合條件全部等權持有，財報公告截止日換股，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

STRATEGY_NAME = "杜邦週轉率選股（2020 年原文策略）"
TAT_MIN = 0.3   # 單季總資產週轉率門檻
ROE_MIN = 5     # 單季 ROE 門檻（%）
PB_MAX = 3      # 股價淨值比上限

def build_report():
    data.set_universe(market="TSE_OTC", exclude_category="金融")
    pb = data.get("price_earning_ratio:股價淨值比")
    roe = data.get("fundamental_features:ROE稅後").index_str_to_date()
    revenue = data.get("financial_statement:營業收入淨額")
    assets = data.get("financial_statement:資產總額")

    # index_str_to_date() 把季別換成財報公告截止日，避免用到還沒公布的財報
    turnover = (revenue / assets).index_str_to_date()
    pb_at_report = pb.reindex(roe.index, method="ffill")
    position = (turnover > TAT_MIN) & (roe > ROE_MIN) & (pb_at_report < PB_MAX)
    return sim(position, 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()
    # 0050 要在策略設定股票池之前取得
    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/dupont-analysis-stock-picking-intro/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()