營收創新高優先的月營收動能（10 檔）

"""營收創新高選股：月營收動能選股，優先買單月營收創 12 個月新高的股票。

與 https://finlab.finance/blog/revenue-new-high-stock-selection 同一份程式，也是
/strategies/revenue-new-high-stock-selection 的數字來源。

- 候選股：股價站上 20／60／120 日均線，近 3 個月平均營收 > 近 12 個月平均營收
  （和 /strategies/monthly-revenue-momentum 相同）
- 排序：當月營收創 12 個月新高的上市櫃普通股排前面（排除金融保險、存託憑證），
  同一組內依 5 日平均漲跌幅（動能）由高到低，取前 10 檔
- 每月 11 日換股、10% 停損，sim() 預設手續費 0.1425% 與證交稅 0.3%

統計口徑：日夏普 = 日報酬平均 / 標準差 × √252，不扣無風險利率；
0050 含息 = etl:adj_close 還原價買進持有，與策略同一段期間。
原版（不加營收創新高）的對照數字由 /blog/monthly-revenue-signals-backtest/research.py 產生。
執行後會在目前目錄輸出 data.csv（本文策略與 0050 含息的淨值曲線）與 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

START = "2015-01-01"
STRATEGY_NAME = "營收創新高優先的月營收動能10檔"
NEW_HIGH_MONTHS = 12
N_STOCKS = 10


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,
        "annual_vol": daily.std() * 252 ** 0.5,
        "max_drawdown": (curve / curve.cummax() - 1).min(),
        "total_return": curve.iloc[-1] - 1,
    }


def yearly_returns(curve: pd.Series) -> dict:
    year_end = curve.groupby(curve.index.year).last()
    prev = year_end.shift(1)
    prev.iloc[0] = curve.iloc[0]
    return {str(y): round(float(v), 6) for y, v in (year_end / prev - 1).items()}


def summary(stats: dict, curve: pd.Series) -> dict:
    """統計四捨五入到小數 6 位，加上逐年報酬。"""
    return {**{k: round(float(v), 6) for k, v in stats.items()}, "yearly": yearly_returns(curve)}


def main() -> None:
    finlab.login()

    close = data.get("price:收盤價")
    rev = data.get("monthly_revenue:當月營收")
    categories = data.get("security_categories").set_index("stock_id")["category"]

    # 1. 原版月營收動能的候選股與排序分數
    rev_up = rev.average(3) > rev.average(12)
    trend = (close > close.average(20)) & (close > close.average(60)) & (close > close.average(120))
    momentum = (close / close.shift() - 1).rolling(5).mean()
    score = momentum[trend & rev_up]

    # 2. 單月營收創 12 個月新高（在月表上算，index 是營收公告期限日），只看上市櫃普通股
    excluded = {"金融保險業", "金融業", "存託憑證"}
    stocks = [
        s for s in rev.columns
        if len(s) == 4 and s.isdigit() and not s.startswith(("0", "28", "91"))
        and categories.get(s) not in excluded
    ]
    monthly = pd.DataFrame(rev).reindex(columns=stocks)
    monthly.index = pd.to_datetime(monthly.index)
    new_high = (monthly > 0) & (monthly >= monthly.rolling(NEW_HIGH_MONTHS, min_periods=NEW_HIGH_MONTHS).max())

    # 3. 展開到每個交易日，創新高的加 1 分：5 日動能都遠小於 1，所以創新高的一定排在前面
    bonus = new_high.astype(float).reindex(close.index, method="ffill")
    bonus = bonus.reindex(columns=close.columns).fillna(0)

    position = (score + bonus).is_largest(N_STOCKS).loc[START:]

    # 每月 11 日（營收公告期限隔天）換股、10% 停損，手續費與證交稅用 sim 預設
    report = sim(position, resample="M", resample_offset="11D", stop_loss=0.1, upload=False,
                 name=STRATEGY_NAME)

    # 從第一次有部位的那天起算，0050 含息用同一段期間
    creturn = report.creturn.dropna()
    first_move = (creturn != creturn.iloc[0]).argmax()
    creturn = creturn.iloc[max(first_move - 1, 0):]
    bench = data.get("etl:adj_close")["0050"].dropna().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: summary(strategy_stats, creturn)},
        "benchmark_0050": summary(bench_stats, bench),
        "run_date": str(pd.Timestamp.today().date()),
        "finlab_version": finlab.__version__,
        "generated_by": "static/blog/revenue-new-high-stock-selection/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()