意圖因子選股（前 50 檔、季換股）

"""意圖因子 ÷ 成交量前 50 檔（季換股）：跑到最新資料的版本（/strategies/linearfactor 的數字來源）。

/blog/linearfactor VIP 區的程式，條件逐行相同：
- 意圖因子 = 60 日報酬 ÷ 60 日每日報酬絕對值總和，選股指標 = 意圖因子 ÷ 當日成交股數
- 60 日報酬小於 20%、當日成交股數大於 VOLUME_MIN（5 萬股）、有月營收資料
- 取指標最大的 TOP_N 檔，等權持有，2010 年起每季換股；sim() 預設手續費 0.1425% 與證交稅 0.3%
- 每日報酬先把停牌日的收盤價往前補（等同原文在 pandas 2 執行時 pct_change() 的預設），
  pandas 2 與 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

STRATEGY_NAME = "意圖因子 ÷ 成交量前 50 檔（季換股）"
DAYS = 60
TOP_N = 50
VOLUME_MIN = 50_000


def build_report():
    close = data.get("price:收盤價")
    volume = data.get("price:成交股數")
    rev = data.get("monthly_revenue:當月營收")

    v = close.ffill().pct_change(fill_method=None).abs().rolling(DAYS).sum()
    s = close / close.shift(DAYS) - 1
    position = (s / v / volume)[(s < 0.2) & (volume > VOLUME_MIN) & rev.notna()].is_largest(TOP_N)
    return sim(position.loc["2010":], resample="Q", 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 main() -> None:
    finlab.login()
    bench_all = data.get("etl:adj_close")["0050"].dropna()
    report = build_report()

    # 從第一次有部位的那天起算，0050 含息用同一段期間
    creturn = report.creturn.dropna()
    first_move = (creturn != creturn.iloc[0]).argmax()
    creturn = creturn.iloc[max(first_move - 1, 0):]
    bench = bench_all.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/linearfactor/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()