月營收動能加上現金減資

"""月營收動能＋現金減資加分（/strategies/cash-capital-reduction-revenue-momentum 的數字來源）。

與 https://finlab.finance/blog/cash-capital-reduction-revenue-momentum 同一份程式：
- 候選股：收盤價站上 20／60／120 日均線，近 3 個月平均營收 > 近 12 個月平均營收
- 近 60 個交易日內辦完現金減資、恢復買賣的候選股排在最前面（停止買賣前收盤價 ÷
  恢復買賣參考價 ≤ 0.9），其餘依 5 日平均漲跌幅（動能）排序，取前 10 檔
- 每月 11 日換股、10% 停損，sim() 預設手續費 0.1425% 與證交稅 0.3%
- 上櫃公司的減資原因從 2019 年起才有資料，之前的上櫃減資不算

統計口徑：日夏普 = 日報酬平均 / 標準差 × √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 = "月營收動能＋現金減資加分"
N_STOCKS = 10
WINDOW = 60           # 恢復買賣日起幾個交易日內算「剛辦完現金減資」
MIN_REDUCTION = 0.1   # 1 − 停止買賣前收盤價 ÷ 恢復買賣參考價，至少 0.1


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 cash_reduction_flag(close) -> pd.DataFrame:
    """現金減資恢復買賣日起 WINDOW 個交易日為 True（恢復買賣當天收盤後才算知道，隔天換股才用得到）。"""
    flag = pd.DataFrame(False, index=close.index, columns=close.columns)
    for market, ratio_key in (("tse", "twse_cap_divide_ratio"), ("otc", "otc_cap_divide_ratio")):
        reason = data.get(f"capital_reduction_{market}:減資原因")
        ratio = data.get(f"capital_reduction_{market}:{ratio_key}")  # 停止買賣前收盤價 ÷ 恢復買賣參考價
        is_cash = reason.isin(["退還股款", "現金減資"]) & (1 - ratio >= MIN_REDUCTION)
        events = is_cash.stack()
        for day, stock in events[events].index:
            if stock in flag.columns:
                row = close.index.searchsorted(day, side="right") - 1
                flag.iloc[row:row + WINDOW, flag.columns.get_loc(stock)] = True
    return flag


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

    close = data.get("price:收盤價")
    rev = data.get("monthly_revenue:當月營收")

    # 月營收動能：五日平均漲跌幅（動能）＋站上 20／60／120 日均線＋近三月均營收 > 近十二月均營收
    pct_change = (close / close.shift() - 1).rolling(5).mean()
    trend = (close > close.average(60)) & (close > close.average(20)) & (close > close.average(120))
    rev_up = rev.average(3) > rev.average(12)

    # 剛辦完現金減資的候選股加 1 分，一定排在其他候選股前面
    score = pct_change + cash_reduction_flag(close).astype(float)
    position = score[trend & rev_up].is_largest(N_STOCKS).loc[START:]

    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: {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/cash-capital-reduction-revenue-momentum/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()