台灣十五小市值（均線＋低波動，每季換股）

"""台灣十五小市值（均線＋低波動，每季換股）：跑到最新資料的版本（/strategies/optimize-stock-selection-beating-0050-etf 的數字來源）。

/blog/optimize-stock-selection-beating-0050-etf「步驟三，減少檔數」的程式，只是不截斷結束日：
- 近 20 日平均成交量大於 200 張
- 股價同時站上 60、120、250 日均線
- 近 60 日日報酬標準差的全市場排名在前 70% 以內（去掉波動最大的 30%）
- 符合條件的股票取市值最小的 15 檔等權，每季換股
- 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 = "台灣十五小市值（均線＋低波動，每季換股）"
VOL_RANK = 0.7  # 波動排名門檻：0.7 代表去掉波動最大的 30%
TOP_N = 15


def build_report():
    cap = data.get("etl:market_value")
    vol = data.get("price:成交股數")
    close = data.get("price:收盤價")
    # pandas 2 的 pct_change() 會先 ffill，這樣寫 pandas 2、3 結果一樣
    std = close.ffill().pct_change(fill_method=None).rolling(60).std().rank(axis=1, pct=True)
    position = cap[
        (vol.average(20) > 200_000)
        & (close > close.average(60))
        & (close > close.average(120))
        & (close > close.average(250))
        & (std < VOL_RANK)
    ].is_smallest(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]]

    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/optimize-stock-selection-beating-0050-etf/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()