月營收驚喜選股（SURGE 前 30 檔）

"""月營收驚喜選股：每月買標準化營收驚喜（SURGE）最高的 30 檔。

與 https://finlab.finance/blog/monthly-revenue-surprise-strategy 同一份程式，也是
/strategies/monthly-revenue-surprise-strategy 的數字來源。

- 營收驚喜：本月營收減去年同月的差額，對照過去 24 個月同一差額的平均與標準差，
  算出標準化分數（Jegadeesh & Livnat 2006 的 SURGE，改成月營收版）
- 股票池：上市櫃普通股（排除金融保險、存託憑證，保留已下市股票），20 日平均成交金額 2,000 萬元以上
- 每月 11 日（營收公告期限隔天）換股、等權，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

START = "2015-01-01"
STRATEGY_NAME = "月營收驚喜30檔"
LOOKBACK_MONTHS = 24
N_STOCKS = 30
MIN_TURNOVER = 2e7


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()

    adj_close = data.get("etl:adj_close")
    categories = data.get("security_categories").set_index("stock_id")["category"]

    # 1. 股票池：4 碼普通股，排除 0 開頭（ETF）、28 開頭（金融）、91 開頭（存託憑證）與金融保險類
    excluded = {"金融保險業", "金融業", "存託憑證"}
    stocks = [
        s for s in adj_close.columns
        if len(s) == 4 and s.isdigit() and not s.startswith(("0", "28", "91"))
        and categories.get(s) not in excluded
    ]
    days = adj_close.loc["2014-01-01":].index

    # 2. 營收驚喜在月表上算：每列是一個月，index 是營收公告期限日（可取得日）
    rev = pd.DataFrame(data.get("monthly_revenue:當月營收")).reindex(columns=stocks)
    rev.index = pd.to_datetime(rev.index)
    yoy_change = rev - rev.shift(12)
    past = yoy_change.shift(1).rolling(LOOKBACK_MONTHS, min_periods=LOOKBACK_MONTHS)
    std = past.std()
    surge = (yoy_change - past.mean()) / std.where(std > 0)

    # 3. 展開到每個交易日：每天沿用最近一次公告期限日的分數
    surge = surge.reindex(days, method="ffill")

    # 4. 流動性：20 日平均成交金額 2,000 萬元以上
    turnover = pd.DataFrame(data.get("price:成交金額")).reindex(columns=stocks)
    turnover = turnover.rolling(20).mean().reindex(days)
    score = surge.where(turnover > MIN_TURNOVER)

    # 5. 分數最高的 30 檔；同分時成交金額大的優先
    rank = score.rank(axis=1, ascending=False)
    tie_break = turnover.where(score.notna()).rank(axis=1, ascending=False, pct=True) * 0.5
    position = ((rank + tie_break).rank(axis=1, method="first") <= N_STOCKS).loc[START:]

    report = sim(position, resample="M", resample_offset="11D", 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 = 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/monthly-revenue-surprise-strategy/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()