"""
台股四萬點選股文章的整合入口。

本腳本會重跑三組實驗：
1. 六個單因子與含息 0050 對照。
2. 品質 + 動能 + 低波 + 營收的複合策略。
3. 60% 0050 + 40% 複合策略的四萬點核心衛星配置。
4. 動能天數、持股檔數、低波權重與加權方式敏感度。

執行方式：
    uv run --python 3.12 --with finlab --with "pandas<3" --with numpy python strategy.py

輸出：
    metrics.json
    data.csv
    report_strategy.html
"""

from __future__ import annotations

import csv
import json
import warnings
from pathlib import Path

import numpy as np
import pandas as pd
from finlab import data
from finlab.backtest import sim


warnings.filterwarnings("ignore")

START = "2018-01-01"
BACKTEST_WINDOW = "2018-03/2026-06"
OUT_DIR = Path(__file__).resolve().parent


def pct_rank(frame: pd.DataFrame) -> pd.DataFrame:
    return pd.DataFrame(frame).astype(float).rank(axis=1, pct=True)


def max_drawdown(series: pd.Series) -> float:
    return float((series / series.cummax() - 1).min())


def benchmark_metrics(adj: pd.DataFrame) -> dict[str, float]:
    equity = adj["0050"].dropna()
    equity = equity[equity.index >= START]
    years = (equity.index[-1] - equity.index[0]).days / 365.25
    returns = equity.pct_change().dropna()
    monthly_returns = equity.resample("M").last().pct_change().dropna()
    downside = monthly_returns[monthly_returns < 0].std()

    return {
        "cagr_pct": round(((equity.iloc[-1] / equity.iloc[0]) ** (1 / years) - 1) * 100, 2),
        "daily_sharpe": round(float(returns.mean() / returns.std() * np.sqrt(252)), 2),
        "monthly_sortino": round(float(monthly_returns.mean() / downside * np.sqrt(12)), 2),
        "max_drawdown_pct": round(max_drawdown(equity) * 100, 2),
    }


def top_n(
    frame: pd.DataFrame,
    n: int = 30,
    largest: bool = True,
    lower_bound: float | None = None,
    upper_bound: float | None = None,
    liquid: pd.DataFrame | None = None,
) -> pd.DataFrame:
    ranked = pd.DataFrame(frame).astype(float)
    if lower_bound is not None:
        ranked = ranked.where(ranked > lower_bound)
    if upper_bound is not None:
        ranked = ranked.where(ranked < upper_bound)
    if liquid is not None:
        ranked = ranked.where(liquid)
    return ranked.rank(axis=1, ascending=not largest) <= n


def stats_from_position(position: pd.DataFrame, **kwargs) -> tuple[object, dict[str, float]]:
    report = sim(
        position[position.index >= START],
        resample="M",
        upload=False,
        **kwargs,
    )
    stats = report.get_stats()
    return report, {
        "cagr_pct": round(float(stats["cagr"]) * 100, 2),
        "daily_sharpe": round(float(stats.get("daily_sharpe", np.nan)), 2),
        "monthly_sortino": round(float(stats.get("monthly_sortino", np.nan)), 2),
        "max_drawdown_pct": round(float(stats["max_drawdown"]) * 100, 2),
    }


def monthly_one_way_turnover(position: pd.DataFrame) -> dict[str, float]:
    monthly = position.resample("M").last()
    monthly = monthly[(monthly.T != 0).any()]
    turnover = (monthly.diff().abs().sum(axis=1) / 2).dropna()
    return {
        "avg_monthly_one_way_turnover_pct": round(float(turnover.mean()) * 100, 1),
        "max_monthly_one_way_turnover_pct": round(float(turnover.max()) * 100, 1),
    }


def composite_position(
    close: pd.DataFrame,
    roe: pd.DataFrame,
    revenue: pd.DataFrame,
    amount: pd.DataFrame,
    mom_days: int = 120,
    vol_days: int = 120,
    lowvol_weight: int = 2,
    topn: int = 25,
    weighting: str = "power3",
) -> pd.DataFrame:
    quality = pct_rank(roe)
    momentum = pct_rank(close / close.shift(mom_days) - 1)
    lowvol = pct_rank(-close.pct_change().rolling(vol_days).std())
    revenue_score = pct_rank(revenue)

    score = (quality + momentum + lowvol_weight * lowvol + revenue_score) / (3 + lowvol_weight)
    score = score.where(amount.rank(axis=1, pct=True) > 0.5)

    selected = score.rank(axis=1, ascending=False) <= topn
    if weighting == "equal":
        weights = selected.astype(float)
    elif weighting == "score":
        weights = score.where(selected)
    elif weighting == "power2":
        weights = (score**2).where(selected)
    else:
        weights = (score**3).where(selected)

    return weights.div(weights.sum(axis=1), axis=0).fillna(0)


def add_row(rows: list[dict[str, object]], group: str, name: str, metrics: dict[str, float], source: str) -> None:
    rows.append(
        {
            "group": group,
            "name": name,
            "cagr_pct": metrics.get("cagr_pct", ""),
            "daily_sharpe": metrics.get("daily_sharpe", ""),
            "monthly_sortino": metrics.get("monthly_sortino", ""),
            "max_drawdown_pct": metrics.get("max_drawdown_pct", ""),
            "source_script": source,
        }
    )


def main() -> None:
    close = data.get("price:收盤價")
    adj = data.get("etl:adj_close")
    pe = data.get("price_earning_ratio:本益比")
    pb = data.get("price_earning_ratio:股價淨值比")
    roe = data.get("fundamental_features:ROE稅後").index_str_to_date().reindex(close.index, method="ffill")
    revenue = data.get("monthly_revenue:去年同月增減(%)").reindex(close.index, method="ffill")
    amount = (close * data.get("price:成交股數")).rolling(60).mean()
    liquid = amount.rank(axis=1, pct=True) > 0.5

    rows: list[dict[str, object]] = []

    benchmark = benchmark_metrics(adj)
    add_row(rows, "benchmark", "0050 含息對照", benchmark, "strategy.py")

    single_factor_specs = [
        ("低本益比（PE）", top_n(pe, 30, largest=False, lower_bound=0, upper_bound=100)),
        ("低股價淨值比（PB）", top_n(pb, 30, largest=False, lower_bound=0, upper_bound=10)),
        ("高 ROE", top_n(roe, 30, largest=True)),
        ("價格動能（近 60 日漲幅）", top_n(close / close.shift(60) - 1, 30, largest=True)),
        ("營收動能（年增最高）", top_n(revenue, 30, largest=True, lower_bound=0)),
        (
            "低波動（近 60 日，流動性過濾後）",
            top_n(-close.pct_change().rolling(60).std(), 30, largest=True, liquid=liquid),
        ),
    ]

    single_factors: list[dict[str, object]] = []
    for name, position in single_factor_specs:
        _, metrics = stats_from_position(position)
        single_factors.append({"name": name, **metrics})
        add_row(rows, "single_factor", name, metrics, "strategy.py")

    composite = composite_position(close, roe, revenue, amount)
    _, composite_metrics = stats_from_position(composite, resample_offset="14D")
    composite_turnover = monthly_one_way_turnover(composite[composite.index >= START])
    add_row(rows, "composite", "品質+動能+低波+營收複合", composite_metrics, "strategy.py")
    rows[-1].update(composite_turnover)

    core_satellite = composite * 0.4
    if "0050" not in core_satellite.columns:
        core_satellite["0050"] = 0.0
    core_satellite["0050"] = core_satellite["0050"].fillna(0) + 0.6
    core_satellite = core_satellite.fillna(0)
    report, core_satellite_metrics = stats_from_position(core_satellite, resample_offset="14D")
    report.to_html(OUT_DIR / "report_strategy.html")
    core_satellite_turnover = monthly_one_way_turnover(core_satellite[core_satellite.index >= START])
    add_row(rows, "core_satellite", "60% 0050 + 40% 複合策略", core_satellite_metrics, "strategy.py")
    rows[-1].update(core_satellite_turnover)

    sensitivity: list[dict[str, object]] = []
    for days in (60, 120, 240):
        _, metrics = stats_from_position(
            composite_position(close, roe, revenue, amount, mom_days=days),
            resample_offset="14D",
        )
        name = f"動能 {days} 日"
        sensitivity.append({"name": name, **metrics})
        add_row(rows, "sensitivity", name, metrics, "strategy.py")

    for holdings in (20, 25, 30, 50):
        _, metrics = stats_from_position(
            composite_position(close, roe, revenue, amount, topn=holdings),
            resample_offset="14D",
        )
        name = f"持股 {holdings} 檔"
        sensitivity.append({"name": name, **metrics})
        add_row(rows, "sensitivity", name, metrics, "strategy.py")

    for weight in (1, 2, 3):
        _, metrics = stats_from_position(
            composite_position(close, roe, revenue, amount, lowvol_weight=weight),
            resample_offset="14D",
        )
        name = f"低波 {weight} 倍權重"
        sensitivity.append({"name": name, **metrics})
        add_row(rows, "sensitivity", name, metrics, "strategy.py")

    for weighting, label in (("equal", "等權"), ("score", "分數"), ("power2", "分數平方"), ("power3", "分數三次方")):
        _, metrics = stats_from_position(
            composite_position(close, roe, revenue, amount, weighting=weighting),
            resample_offset="14D",
        )
        sensitivity.append({"name": label, **metrics})
        add_row(rows, "sensitivity", label, metrics, "strategy.py")

    metrics_json = {
        "slug": "30k-stock-selection",
        "as_of": "2026-06",
        "market": "TW",
        "backtest_window": BACKTEST_WINDOW,
        "benchmark": {"name": "0050 含息對照", "data_key": "etl:adj_close", **benchmark},
        "core_satellite_strategy": {
            "name": "60% 0050 + 40% 品質+動能+低波+營收複合",
            **core_satellite_metrics,
            **core_satellite_turnover,
        },
        "satellite_strategy": {
            "name": "品質+動能+低波+營收複合",
            **composite_metrics,
            **composite_turnover,
        },
        "single_factors": single_factors,
        "sensitivity": sensitivity,
        "source_script": "strategy.py",
    }

    with (OUT_DIR / "metrics.json").open("w", encoding="utf-8") as fh:
        json.dump(metrics_json, fh, ensure_ascii=False, indent=2)

    with (OUT_DIR / "data.csv").open("w", encoding="utf-8", newline="") as fh:
        writer = csv.DictWriter(
            fh,
            fieldnames=[
                "group",
                "name",
                "cagr_pct",
                "daily_sharpe",
                "monthly_sortino",
                "max_drawdown_pct",
                "avg_monthly_one_way_turnover_pct",
                "max_monthly_one_way_turnover_pct",
                "source_script",
            ],
            lineterminator="\n",
        )
        writer.writeheader()
        writer.writerows(rows)

    print(json.dumps(metrics_json, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()