PEG＋營收動能（前 10 檔、季換股）

"""PEG＋營收動能（PEG 0～1.5、前 10 檔、季換股）：跑到最新資料的版本（/strategies/pe-pb-peg-backtest 的數字來源）。

/blog/pe-pb-peg-backtest VIP 區「PEG + 營收動能」的程式，條件逐行相同：
- PEG（本益比 ÷ 營業利益成長率）介於 0 到 PEG_MAX、營業利益成長率大於 5%
- 營收年增大於 10%、近 3 個月平均營收創 12 個月新高
- RSI(14) 大於 50、本益比介於 5 到 30、近 60 個交易日股價漲幅為正、20 日均量大於 300 張
- 符合條件的股票裡取 PEG 最低的 TOP_N 檔，等權持有，2015 年起每季換股
- sim() 預設手續費 0.1425% 與證交稅 0.3%；RSI 用 data.indicator，需要 TA-Lib

統計口徑：日夏普 = 日報酬平均 / 標準差 × √252，不扣無風險利率；
0050 含息 = etl:adj_close 還原價買進持有，與策略同一段期間。
執行後會在目前目錄輸出 data.csv（淨值曲線）與 metrics.json（統計）。

    pip install finlab TA-Lib
    python strategy.py

finlab 需要資料時會自動引導登入，不需要在程式裡填 token。
"""

import json

import pandas as pd

import finlab
from finlab import data
from finlab.backtest import sim

STRATEGY_NAME = "PEG＋營收動能（前 10 檔、季換股）"
PEG_MAX = 1.5
TOP_N = 10


def build_report():
    pe = data.get("price_earning_ratio:本益比")
    growth = data.get("fundamental_features:營業利益成長率")
    rev = data.get("monthly_revenue:當月營收")
    rev_yoy = data.get("monthly_revenue:去年同月增減(%)")
    close = data.get("price:收盤價")
    rsi = data.indicator("RSI", timeperiod=14)
    volume = data.get("price:成交股數")

    peg = pe / growth
    rev_ma3 = rev.average(3)
    rev_new_high = rev_ma3 == rev_ma3.rolling(12).max()
    momentum_3m = close / close.shift(60) - 1

    cond = (
        (peg > 0) & (peg < PEG_MAX)
        & (growth > 5)
        & (rev_yoy > 10)
        & rev_new_high
        & (rsi > 50)
        & (pe > 5) & (pe < 30)
        & (momentum_3m > 0)
        & (volume.average(20) > 300 * 1000)
    )
    position = peg[cond]
    position = position[position > 0].is_smallest(TOP_N)
    position = position.reindex(rev.index_str_to_date().index, method="ffill")
    return sim(position.loc["2015":], 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/pe-pb-peg-backtest/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()