庫藏股執行率選股（上次買回 9 成以上再公告）

"""庫藏股執行率選股（/strategies/treasury-stock-repeat-buyback 的數字來源）。

與 https://finlab.finance/blog/tw-corporate-actions-event-backtest 第 3 個訊號同一份條件：
- 事件：庫藏股公告（treasury_stock，index 是公告日）
- 條件：同一家公司上一次買回的預定期間已結束滿 7 天，且已買回股數占預定買回股數 90% 以上
  （執行結果要等期間結束才知道，所以只能看上一次）
- 公告隔天開盤買進，持有 60 個交易日；同時持有的公司等權，沒有事件時空手
- 上市櫃 4 位數普通股；隔天沒有開盤價（停牌）的事件不算
- 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 numpy as np
import pandas as pd

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

START = "2013-05-01"
STRATEGY_NAME = "庫藏股執行率 9 成以上再公告"
MIN_PREV_RATE = 90   # 上一次已買回股數占預定買回股數（%）
HOLD_DAYS = 60
RATE = "本次已買回股數佔預定買回股數比例(%)"


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 buyback_events(stocks) -> pd.DataFrame:
    """一列一次庫藏股公告，帶上同公司上一次（已結束滿 7 天）的執行率。"""
    columns = ["預定買回股數", "預定買回期間-迄", RATE]
    events = pd.concat({c: pd.DataFrame(data.get(f"treasury_stock:{c}")).stack() for c in columns}, axis=1)
    events = events[events["預定買回股數"].notna()].reset_index()
    events.columns = ["date", "stock"] + columns
    events["date"] = pd.to_datetime(events["date"])
    events = events[events["stock"].isin(stocks)].sort_values("date")
    done = events.assign(known=events["預定買回期間-迄"] + pd.Timedelta(days=7))
    done = done.dropna(subset=["known"]).sort_values("known")
    return pd.merge_asof(
        events, done[["stock", "known", RATE]].rename(columns={RATE: "prev_rate"}),
        left_on="date", right_on="known", by="stock", suffixes=("", "_prev"),
    )


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

    close = pd.DataFrame(data.get("price:收盤價"))
    close.index = pd.to_datetime(close.index)
    days = close.index
    stocks = [s for s in close.columns if len(s) == 4 and s.isdigit() and not s.startswith("0")]
    adj_open = pd.DataFrame(data.get("etl:adj_open")).reindex(index=days)

    events = buyback_events(stocks)
    events = events[events["prev_rate"] >= MIN_PREV_RATE]

    # 公告當天收盤後才知道，訊號放在公告日（或之前最近的交易日），sim 隔天開盤進場
    held = np.zeros((len(days), len(stocks)), dtype=bool)
    col = {s: i for i, s in enumerate(stocks)}
    for stock, day in zip(events["stock"], events["date"]):
        row = days.searchsorted(day, side="right") - 1
        if np.isfinite(adj_open[stock].iloc[min(row + 1, len(days) - 1)]):
            held[row:row + HOLD_DAYS, col[stock]] = True
    position = pd.DataFrame(held, index=days, columns=stocks).loc[START:]
    position = position.loc[:, position.any()]

    report = sim(position, trade_at_price="open", 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/tw-corporate-actions-event-backtest/treasury-stock-repeat-buyback/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()