#!/usr/bin/env python3
"""quant-trading-disadvantages-risks 的真數據來源(2026-09-18 補跑,原文四個單因子夏普與成本壓測無工件)。

產物(static/blog/quant-trading-disadvantages-risks/):
  metrics.json          單因子歸因(2013-01~2026-08,月頻、每月等權選分數最高 30 檔,內扣手續費 0.1425% 與證交稅 0.3%)
                        + 三因子複合策略的手續費壓力測試(2020-01~2026-08,fee 0% → 0.5%)
  strategy.py           可重現腳本(本檔複本)
  dr-single-factor.png  單因子夏普長條圖
  dr-cost-erosion.png   成本侵蝕折線圖
口徑:夏普 = 日報酬 mean/std 年化、不扣無風險利率(與 scripts/backtest_factory.py canonical 一致)。
三因子複合策略規格與 static/blog/python-quant-trading-tutorial/strategy.py 相同
(營收動能 + 120 日價格動能 + 低波動等權排名、反波動加權、加權報酬指數站上 120 日均線濾網)。
執行:cd ~/Documents/finlab && UV_ENV_FILE=.env uv run python <repo>/scripts/gen-quant-risk-single-factor-assets.py
"""
from __future__ import annotations

import json
import sys
from datetime import date
from pathlib import Path

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
import pandas as pd

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
import backtest_factory as bf  # noqa: E402
from finlab import data  # noqa: E402
from finlab.backtest import sim  # noqa: E402

SLUG = "quant-trading-disadvantages-risks"
OUT = ROOT / "static" / "blog" / SLUG
SF_START, SF_END = "2013-01-01", "2026-08-31"
COST_START, COST_END = "2020-01-01", "2026-08-31"
TOP_N = 30
TUTORIAL_FEE = 1.425 / 1000 / 3
COST_FEES = (0.0, TUTORIAL_FEE, bf.FEE, 0.003, 0.005)
EXCLUDED_CATEGORIES = ["金融保險", "ETF", "ETN", "存託憑證", "受益證券", "創新板"]
MIN_TURNOVER_NTD = 50_000_000
MIN_PRICE = 10


def load_universe(bundle: bf.DataBundle):
    close = bundle["price:收盤價"]
    volume = bundle["price:成交股數"]
    amount = (close * volume).average(60)
    flagged = bundle["etl:is_flagged_stock"].reindex(close.index, method="ffill").fillna(False)
    cats = bundle["security_categories"].set_index("stock_id")["category"]
    bad = set(cats[cats.isin(EXCLUDED_CATEGORIES)].index)
    keep = [c for c in close.columns if c not in bad]
    universe = (amount > MIN_TURNOVER_NTD) & (close > MIN_PRICE) & (~flagged)
    return close, keep, universe


def align_monthly(df: pd.DataFrame, close: pd.DataFrame) -> pd.DataFrame:
    """月營收類資料用公告期限對齊到日頻(避免前視偏差)。"""
    return df.index_str_to_date().reindex(close.index, method="ffill")


def top_n_equal_weight(score: pd.DataFrame, universe: pd.DataFrame, keep: list[str]) -> pd.DataFrame:
    cols = sorted(set(keep) & set(score.columns))
    score = score[cols].where(universe[cols])
    return score.rank(axis=1, ascending=False) <= TOP_N


def single_factor_scores(bundle: bf.DataBundle, close, keep, universe) -> dict[str, pd.DataFrame]:
    rev = align_monthly(bundle["monthly_revenue:當月營收"], close)
    rev_yoy = align_monthly(bundle["monthly_revenue:去年同月增減(%)"], close)
    pe = bundle["price_earning_ratio:本益比"].reindex(close.index, method="ffill")
    ret = close.pct_change()
    scores = {
        # 營收創新高:當月營收相對過去 12 個月最高的比值,>1 即創新高;比值越大分數越高
        "rev_new_high": rev / rev.rolling(12).max().shift(1),
        "low_vol": -ret.rolling(120).std(),
        "rev_yoy": rev_yoy,
        # 低本益比:只取正本益比,越低分數越高
        "low_pe": -pe.where(pe > 0),
    }
    return {k: top_n_equal_weight(v, universe, keep) for k, v in scores.items()}


def three_factor_weight(bundle: bf.DataBundle, close, keep, universe) -> pd.DataFrame:
    rev_yoy = align_monthly(bundle["monthly_revenue:去年同月增減(%)"], close)
    momentum = close / close.shift(120)
    ret = close.pct_change()
    low_vol = -ret.rolling(120).std()
    cols = sorted(set(keep) & set(rev_yoy.columns))

    def rank(df):
        return df[cols].rank(axis=1, pct=True)

    score = ((rank(rev_yoy) + rank(momentum) + rank(low_vol)) / 3).where(universe[cols])
    selected = score.rank(axis=1, ascending=False) <= TOP_N
    inv_vol = (1 / ret.rolling(60).std())[cols]
    weight = selected * inv_vol
    weight = weight.div(weight.sum(axis=1), axis=0).fillna(0)
    twii = bundle["benchmark_return:發行量加權股價報酬指數"].iloc[:, 0]
    market_up = (twii > twii.rolling(120).mean()).reindex(close.index, method="ffill").fillna(False)
    return weight.mul(market_up.astype(int), axis=0)


FACTOR_NAMES = {
    "rev_new_high": "營收創新高",
    "low_vol": "低波動",
    "rev_yoy": "月營收年增率",
    "low_pe": "低本益比",
}


def chart_single_factor(rows: list[dict]) -> None:
    bf._setup_style()
    fig, ax = plt.subplots(figsize=(8, 4.2))
    names = [r["name"] for r in rows]
    vals = [r["daily_sharpe"] for r in rows]
    colors = [bf.BLUE if v >= 0.5 else bf.GRAY for v in vals]
    bars = ax.bar(names, vals, color=colors, width=0.55)
    for b, v in zip(bars, vals):
        ax.text(b.get_x() + b.get_width() / 2, v + 0.02, f"{v:.2f}", ha="center", va="bottom", fontsize=11, color=bf.INK)
    ax.axhline(0, color=bf.INK, linewidth=0.8)
    ax.set_ylabel("夏普比率(日報酬年化)")
    ax.set_title(f"台股單因子歸因:各因子單獨選前 {TOP_N} 檔的夏普比率({SF_START[:4]}–{SF_END[:4]},月頻,已扣成本)", fontsize=12)
    ax.set_ylim(min(0, min(vals)) - 0.1, max(vals) + 0.3)
    fig.tight_layout()
    bf._save(fig, OUT, "dr-single-factor")


def chart_cost(rows: list[dict]) -> None:
    bf._setup_style()
    fig, ax = plt.subplots(figsize=(8, 4.2))
    fees = [r["fee"] * 100 for r in rows]
    cagr = [r["cagr"] for r in rows]
    ax.plot(fees, cagr, marker="o", color=bf.RED, linewidth=2)
    for f, c in zip(fees, cagr):
        ax.text(f, c + 0.3, f"{c:.1f}%", ha="center", va="bottom", fontsize=10, color=bf.INK)
    ax.set_xlabel("單邊手續費率(%,含滑價代理;證交稅 0.3% 另計)")
    ax.set_ylabel("年化報酬(%)")
    ax.set_title(f"交易成本壓力測試:三因子月頻策略 {COST_START[:4]}–{COST_END[:4]} 年化報酬隨手續費率的變化", fontsize=12)
    ax.set_ylim(min(cagr) - 3, max(cagr) + 3)
    fig.tight_layout()
    bf._save(fig, OUT, "dr-cost-erosion")


def main() -> None:
    OUT.mkdir(parents=True, exist_ok=True)
    bundle = bf.DataBundle()
    close, keep, universe = load_universe(bundle)

    single_rows = []
    for key, position in single_factor_scores(bundle, close, keep, universe).items():
        report = sim(position[position.index >= SF_START], resample="M", fee_ratio=bf.FEE, tax_ratio=bf.TAX,
                     upload=False, name=FACTOR_NAMES[key])
        _, stats = bf.evaluate(report, SF_START, SF_END, position)
        single_rows.append({"key": key, "name": FACTOR_NAMES[key],
                            **{k: round(v, 4) for k, v in stats.items()}})
        print(f"[single] {FACTOR_NAMES[key]}: sharpe={stats['daily_sharpe']:.2f} cagr={stats['cagr']*100:.1f}% mdd={stats['max_drawdown']*100:.1f}%")

    weight = three_factor_weight(bundle, close, keep, universe)
    cost_rows = bf.cost_sensitivity(weight, COST_START, COST_END, "M", fees=COST_FEES)
    bench_curve, bench = bf.canonical_benchmark(bundle["etl:adj_close"], SF_START, SF_END)

    metrics = {
        "slug": SLUG,
        "generated": date.today().isoformat(),
        "single_factor": {
            "window": {"start": SF_START, "end": SF_END},
            "method": f"月頻換股、各因子單獨選分數最高 {TOP_N} 檔等權;股票池=60 日均成交金額 > 5,000 萬、收盤價 > 10、非全額交割、排除金融/ETF;手續費 {bf.FEE*100:.4f}%、證交稅 {bf.TAX*100:.1f}%;夏普=日報酬 mean/std 年化,不扣無風險利率",
            "rows": single_rows,
            "benchmark_0050": {k: (round(v, 4) if isinstance(v, float) else v) for k, v in bench.items()},
        },
        "cost_sensitivity": {
            "window": {"start": COST_START, "end": COST_END},
            "strategy": "營收動能 + 120 日價格動能 + 低波動三因子等權排名、反波動加權、大盤 120 日均線濾網(同 python-quant-trading-tutorial/strategy.py)",
            "tax_ratio": bf.TAX,
            "rows": cost_rows,
        },
        "disclaimer": "本資料為歷史回測統計與教育用途,不構成投資建議;過去績效不代表未來表現。",
    }
    (OUT / "metrics.json").write_text(json.dumps(metrics, ensure_ascii=False, indent=2), encoding="utf-8")
    (OUT / "strategy.py").write_text(Path(__file__).read_text(encoding="utf-8"), encoding="utf-8")
    chart_single_factor(single_rows)
    chart_cost(cost_rows)
    print("wrote", OUT)


if __name__ == "__main__":
    main()