庫藏股家數爆量後持有正2

"""庫藏股家數爆量後一年改抱正 2（/blog/treasury-stock-buyback-bottom-signal 驗證段與同名策略頁的數字來源）。

規則：
- 每月數一次宣布買回庫藏股的公司家數（treasury_stock 的董事會決議日，同一家公司當月只算一次）
- 當月家數 > 截至當月的歷史平均 + 1.5 個標準差（至少 24 個月資料，不用之後的資料），就是訊號月
- 平常持有元大台灣50（0050）；訊號月月底起 250 個交易日，改持有元大台灣50正2（00631L）
- 訊號日的下一個交易日收盤換股，sim() 預設手續費 0.1425%、證交稅 0.3%
- 從 00631L 上市的 2014-10-31 起算；2014 年以後的訊號月是 2015-08、2018-11、2020-03、2025-04

原文的「實施家數超過 1.5 個標準差就是左側買點」若只用 0050 與現金做（訊號後持有 0050、其餘空手），
9 組參數全輸 0050，見 https://finlab.finance/blog/taiwan-market-timing-signals-backtest/research.py。

統計口徑：日夏普 = 日報酬平均 / 標準差 × √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 = "庫藏股家數爆量後持有正2"
START = "2014-10-31"
K = 1.5          # 家數超過歷史平均幾個標準差
HOLD_DAYS = 250  # 訊號後持有正 2 的交易日數


def signal_month_ends() -> list:
    buyback = data.get("treasury_stock:預定買回股數")
    month = buyback.index.to_period("M")
    count = buyback.notna().groupby(month).any().sum(axis=1)
    count = count.reindex(pd.period_range(count.index[0], count.index[-1], freq="M"), fill_value=0)
    line = count.expanding(24).mean() + K * count.expanding(24).std(ddof=0)
    return [p.to_timestamp(how="end").normalize() for p in count[count > line].index]


def build_report():
    dates = data.get("price:收盤價").loc[START:].index

    # 訊號月月底（遇假日取下一個交易日）起 HOLD_DAYS 個交易日為 True
    leveraged = pd.Series(False, index=dates)
    for month_end in signal_month_ends():
        if month_end >= dates[0]:
            i = dates.searchsorted(month_end)
            leveraged.iloc[i:i + HOLD_DAYS] = True

    position = pd.DataFrame({
        "00631L": leveraged.astype(float),
        "0050": (~leveraged).astype(float),
    })
    return sim(position, 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 rounded(stats: dict) -> dict:
    return {k: round(float(v), 6) for k, v in stats.items()}


def main() -> None:
    finlab.login()
    adj = data.get("etl:adj_close")
    report = build_report()

    creturn = report.creturn.loc[START:].dropna()
    bench = adj["0050"].ffill().loc[creturn.index[0]:creturn.index[-1]]
    lev = adj["00631L"].ffill().loc[creturn.index[0]:creturn.index[-1]]

    strategy_stats = curve_stats(creturn)
    strategy_stats["avg_holdings"] = 1.0
    bench_stats = curve_stats(bench)
    lev_stats = curve_stats(lev)

    start, end = creturn.index[0].date(), creturn.index[-1].date()
    print(f"期間 {start} 至 {end}")
    for label, s in [("策略", strategy_stats), ("0050 含息", bench_stats), ("00631L 買進持有", lev_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],
        "00631L買進持有": lev / lev.iloc[0],
    })
    curves.index.name = "date"
    curves.round(6).to_csv("data.csv")
    metrics = {
        "window": f"{start}/{end}",
        "star": STRATEGY_NAME,
        "variants": {STRATEGY_NAME: rounded(strategy_stats)},
        "benchmark_0050": rounded(bench_stats),
        "buy_hold_00631L": rounded(lev_stats),
        "run_date": str(pd.Timestamp.today().date()),
        "finlab_version": finlab.__version__,
        "generated_by": "static/blog/treasury-stock-buyback-bottom-signal/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()