營收動能低波三因子（大盤 120 日線濾網）

"""營收動能低波三因子（大盤 120 日線濾網）：跑到最新資料的版本（/strategies/python-quant-trading-tutorial 的數字來源）。

條件與 /blog/python-quant-trading-tutorial 附的 strategy.py 相同，差別只有兩處：
不截斷結束日，以及手續費改用 sim() 預設（原檔手續費打三折）。
- 股票池：60 日平均成交金額大於 5,000 萬、收盤價高於 10 元、不是處置或全額交割股，
  排除金融保險、ETF、ETN、存託憑證、受益證券與創新板
- 月營收年增、120 日價格動能、120 日低波動三項百分位排名平均，取前 30 檔
- 依 60 日波動度倒數加權；加權報酬指數站上 120 日均線才持股，否則空手
- 每月換股，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 = "營收動能低波三因子（大盤 120 日線濾網）"
TOP_N = 30          # 持股檔數
MARKET_MA = 120     # 大盤濾網均線天數
EXCLUDED = ["金融保險", "ETF", "ETN", "存託憑證", "受益證券", "創新板"]


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

    amount = (close * volume).average(60)
    flagged = data.get("etl:is_flagged_stock").reindex(close.index, method="ffill")
    flagged = flagged.fillna(False).astype(bool)
    cats = data.get("security_categories").set_index("stock_id")["category"]
    bad = set(cats[cats.isin(EXCLUDED)].index)
    universe = (amount > 50_000_000) & (close > 10) & (~flagged)

    rev_yoy = data.get("monthly_revenue:去年同月增減(%)").index_str_to_date()
    rev_yoy = rev_yoy.reindex(close.index, method="ffill")
    ret = close.pct_change(fill_method=None)
    momentum = close / close.shift(120)
    low_vol = -ret.rolling(120).std()
    cols = sorted((set(close.columns) - bad) & set(rev_yoy.columns))

    def rank(df):
        return df[cols].rank(axis=1, pct=True)

    score = (rank(rev_yoy) + rank(momentum) + rank(low_vol)) / 3
    score = score.where(universe[cols])
    selected = score.rank(axis=1, ascending=False) <= TOP_N

    inv_vol = (1 / ret.rolling(60).std())[cols]
    weight = selected * inv_vol
    weight = weight.div(weight.sum(axis=1), axis=0).fillna(0)

    twii = data.get("benchmark_return:發行量加權股價報酬指數").iloc[:, 0]
    market_up = (twii > twii.rolling(MARKET_MA).mean()).reindex(close.index, method="ffill")
    weight = weight.mul(market_up.fillna(False).astype(int), axis=0)
    return sim(weight, resample="M", 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/python-quant-trading-tutorial/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()