# -*- coding: utf-8 -*-
"""AI Agent 量化交易完整指南:三段對話策略重現腳本。

對應文章:
  https://finlab.finance/blog/ai-agent-quant-trading-guide

執行:
  cd ~/Documents/finlab && UV_ENV_FILE=.env uv run --with finlab --with pandas \
    --with numpy --with matplotlib python <path>/agent_strategies.py

輸出:
  - /tmp/seo-bt/ai-agent-guide/metrics.json、equity.csv、selected_stocks.csv
  - /tmp/seo-bt/ai-agent-guide/report_basic_strategy.html、report_ranking_strategy.html
  - static/blog/ai-agent-quant-trading-guide/*.png(本機預覽,正式發佈前上 R2)
  - static/blog/ai-agent-quant-trading-guide/metrics.json(入庫供日後對賬)

共同設定:
  - 回測區間:資料起點(約 2007)到 DATA_END(預設 2026-06-09,全站 canonical 快照日)
  - 交易成本:策略使用 finlab sim() 台股預設值(手續費 0.1425%、賣出證交稅 0.3%);未另設滑價
  - 0050 基準:etl:adj_close 還原價 buy-and-hold 純指數算術(聖經 §D canonical 口徑,不經 sim)
  - 因子對齊:月營收/ROE 依公布時點 ffill 到日頻,避免前視

投資警語:本程式僅供量化研究與教學用途,過去績效不代表未來表現,
不構成任何投資建議;實際交易前請自行評估風險、滑價與交易容量。
"""
from __future__ import annotations

import json
import os
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")

END = os.environ.get("DATA_END", "2026-06-09")
TMP_OUT = Path(os.environ.get("OUT_DIR", "/tmp/seo-bt/ai-agent-guide"))
STATIC_OUT = Path(__file__).resolve().parent
TMP_OUT.mkdir(parents=True, exist_ok=True)


# ---------- 資料(公布時點對齊,避免前視) ----------
def cap(df):
    return df[df.index <= END] if hasattr(df, "index") else df


close = cap(data.get("price:收盤價"))
adj = cap(data.get("etl:adj_close"))
volume = cap(data.get("price:成交股數"))
pe = cap(data.get("price_earning_ratio:本益比"))
rev = cap(data.get("monthly_revenue:去年同月增減(%)"))
roe = cap(data.get("fundamental_features:ROE稅後").index_str_to_date())

rev_d = rev.reindex(close.index, method="ffill")
roe_d = roe.reindex(close.index, method="ffill")

# 流動性過濾:近 20 日平均成交金額 > 1,000 萬
amount20 = (close * volume).rolling(20).mean()
liquid = amount20 > 10_000_000


def summarize(name: str, report, position: pd.DataFrame) -> dict:
    s = report.get_stats()
    cr = report.creturn.dropna()
    # 月底持股數分佈(支撐「持股數穩定性」的敘事,全部真算)
    hold_m = (position.resample("ME").last().fillna(False) > 0).sum(axis=1)
    hold_m = hold_m[hold_m.index >= cr.index[0]]
    years = (cr.index[-1] - cr.index[0]).days / 365.25
    out = {
        "name": name,
        "start": str(cr.index[0].date()),
        "end": str(cr.index[-1].date()),
        "years": round(years, 1),
        "cagr": round(float(s["cagr"]) * 100, 1),
        "daily_sharpe": round(float(s["daily_sharpe"]), 2),
        "daily_sortino": round(float(s["daily_sortino"]), 2),
        "monthly_sortino": round(float(s["monthly_sortino"]), 2),
        "max_drawdown": round(float(s["max_drawdown"]) * 100, 1),
        "total_return": round(float(s["total_return"]) * 100, 0),
        "final_value_of_1m": round(100 * (1 + float(s["total_return"]))),
        "avg_holdings": round(float(hold_m.mean()), 0) if len(hold_m) else 0,
        "min_holdings": int(hold_m.min()) if len(hold_m) else 0,
        "max_holdings": int(hold_m.max()) if len(hold_m) else 0,
        "zero_holding_months": int((hold_m == 0).sum()),
    }
    # 近 3 / 5 年日夏普(品質閘 OR 指標之一)
    ret = cr.pct_change().dropna()
    for label, yrs in (("recent_3y_sharpe", 3), ("recent_5y_sharpe", 5)):
        seg = ret[ret.index >= cr.index[-1] - pd.DateOffset(years=yrs)]
        out[label] = round(float(seg.mean() / seg.std() * (252 ** 0.5)), 2)
    return out


def run(name: str, position: pd.DataFrame):
    report = sim(position, resample="M", upload=False, name=name)
    return summarize(name, report, position), report


# ---------- 對話一:篩選法(營收年增 > 20% 且 ROE > 10%) ----------
pos1 = ((rev_d > 20) & (roe_d > 10) & close.notna() & liquid)
row1, report1 = run("篩選法:營收>20%+ROE>10%", pos1)

# ---------- 對話二:加嚴條件(營收 > 30%、ROE > 15%、本益比 < 20) ----------
pos2 = ((rev_d > 30) & (roe_d > 15) & (pe < 20) & close.notna() & liquid)
row2, report2 = run("加嚴篩選:營收>30%+ROE>15%+PE<20", pos2)

# ---------- 對話三:排名法多因子(營收 40% + ROE 30% + 動能 30%,前 20 檔) ----------
mom60 = close.pct_change(60)


def ranking_position(w_rev: float, w_roe: float, w_mom: float, top_n: int) -> pd.DataFrame:
    rev_rank = rev_d.rank(axis=1, pct=True)
    roe_rank = roe_d.rank(axis=1, pct=True)
    mom_rank = mom60.rank(axis=1, pct=True)
    composite = rev_rank * w_rev + roe_rank * w_roe + mom_rank * w_mom
    return composite.where(liquid & close.notna()).is_largest(top_n)


pos3 = ranking_position(0.4, 0.3, 0.3, 20)
row3, report3 = run("排名法多因子:營收40%+ROE30%+動能30%", pos3)

# 排名法月周轉率(每月汰換比例)
pos3_m = pos3.resample("ME").last().fillna(False)
prev = pos3_m.shift(1).fillna(False)
held = prev.sum(axis=1)
replaced = (prev & ~pos3_m).sum(axis=1)
turnover = (replaced / held.replace(0, np.nan)).dropna()
row3["monthly_turnover_pct"] = round(float(turnover.mean()) * 100, 0)
row3["monthly_turnover_max_pct"] = round(float(turnover.max()) * 100, 0)

# ---------- 0050 含息基準(canonical:純指數算術,不經 sim) ----------
# 與策略同窗口起算,讓四條序列可同口徑比較
common_start = max(report1.creturn.dropna().index[0],
                   report3.creturn.dropna().index[0])
bser = adj["0050"].dropna()
bser = bser[(bser.index >= common_start) & (bser.index <= END)]
bret = bser.pct_change().dropna()
_years = (bser.index[-1] - bser.index[0]).days / 365.25
_total = float(bser.iloc[-1] / bser.iloc[0] - 1)
_mret = bser.resample("ME").last().pct_change().dropna()
bench = {
    "name": "0050 含息買進持有",
    "start": str(bser.index[0].date()),
    "end": str(bser.index[-1].date()),
    "years": round(_years, 1),
    "cagr": round(((1 + _total) ** (1 / _years) - 1) * 100, 1),
    "daily_sharpe": round(float(bret.mean() / bret.std() * (252 ** 0.5)), 2),
    "daily_sortino": round(float(bret.mean() / bret[bret < 0].std() * (252 ** 0.5)), 2),
    "monthly_sortino": round(float(_mret.mean() / _mret[_mret < 0].std() * (12 ** 0.5)), 2),
    "max_drawdown": round(float((bser / bser.cummax() - 1).min()) * 100, 1),
    "total_return": round(_total * 100, 0),
    "final_value_of_1m": round(100 * (1 + _total)),
    "avg_holdings": 1.0,
}
bench_creturn = bser / bser.iloc[0]

# ---------- 參數敏感度(真跑,取代任何示意數字) ----------
sensitivity_specs = [
    ("營收50/ROE25/動能25・前20", 0.50, 0.25, 0.25, 20),
    ("營收30/ROE30/動能40・前20", 0.30, 0.30, 0.40, 20),
    ("等權33/33/34・前20", 0.33, 0.33, 0.34, 20),
    ("營收20/ROE40/動能40・前20", 0.20, 0.40, 0.40, 20),
    ("營收40/ROE30/動能30・前10", 0.40, 0.30, 0.30, 10),
    ("營收40/ROE30/動能30・前30", 0.40, 0.30, 0.30, 30),
]
sens_rows = [dict(row3, name="營收40/ROE30/動能30・前20(本文)")]
for name, wr, wo, wm, n in sensitivity_specs:
    r, _ = run(name, ranking_position(wr, wo, wm, n))
    sens_rows.append(r)

# ---------- 三次股災區間回撤(真算,取代任何示意數字) ----------
crisis_windows = {
    "2015 下半年修正": ("2015-04-01", "2015-12-31"),
    "2020 疫情爆發": ("2020-01-01", "2020-05-31"),
    "2022 升息修正": ("2022-01-01", "2022-10-31"),
}
s3_cr = report3.creturn.dropna()
crisis_rows = []
for label, (w0, w1) in crisis_windows.items():
    seg_s = s3_cr[(s3_cr.index >= w0) & (s3_cr.index <= w1)]
    seg_b = bench_creturn[(bench_creturn.index >= w0) & (bench_creturn.index <= w1)]
    crisis_rows.append({
        "window": label,
        "strategy_dd": round(float((seg_s / seg_s.cummax() - 1).min()) * 100, 1),
        "bench_dd": round(float((seg_b / seg_b.cummax() - 1).min()) * 100, 1),
    })

# ---------- 目前選股清單 ----------
latest_row = pos3[pos3.any(axis=1)].iloc[-1]
picks = sorted(latest_row[latest_row].index.tolist())
picks_date = str(pos3[pos3.any(axis=1)].index[-1].date())

# ---------- 報告與數據輸出 ----------
report1.to_html(str(TMP_OUT / "report_basic_strategy.html"), title="篩選法策略:營收>20%+ROE>10%")
report3.to_html(str(TMP_OUT / "report_ranking_strategy.html"), title="排名法多因子策略:營收40%+ROE30%+動能30%")

eq = pd.DataFrame({
    "0050": bench_creturn,
    "screen_basic": report1.creturn,
    "screen_strict": report2.creturn,
    "ranking_top20": report3.creturn,
})
eq.to_csv(TMP_OUT / "equity.csv")
pd.DataFrame({"股票代號": picks}).to_csv(TMP_OUT / "selected_stocks.csv", index=False)

metrics = {
    "data_end": END,
    "window_start": str(common_start.date()),
    "benchmark": bench,
    "strategies": [row1, row2, row3],
    "sensitivity": sens_rows,
    "crisis_drawdowns": crisis_rows,
    "latest_picks": {"date": picks_date, "tickers": picks},
    "method": {
        "universe": "全上市櫃(finlab 資料含已下市股票,無生存者偏誤);未排除金融股/KY/ETF",
        "liquidity": "近 20 日平均成交金額 > 1,000 萬",
        "alignment": "月營收/ROE 依公布時點 ffill 至日頻;動能 = 60 個交易日報酬",
        "cost": "策略使用 finlab sim() 台股預設成本(手續費 0.1425%、賣出證交稅 0.3%);未另設滑價",
        "weighting": "等權,每月再平衡",
        "benchmark": "0050 etl:adj_close 還原價 buy-and-hold 純指數算術(不經 sim、不含交易成本)",
        "sample": "全段 in-sample,未做樣本外測試",
    },
}
(TMP_OUT / "metrics.json").write_text(json.dumps(metrics, ensure_ascii=False, indent=2), encoding="utf-8")
(STATIC_OUT / "metrics.json").write_text(json.dumps(metrics, ensure_ascii=False, indent=2), encoding="utf-8")


# ---------- 圖表(16:9,本機預覽,正式發佈前上 R2) ----------
def save_charts() -> None:
    import matplotlib

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

    plt.rcParams["font.sans-serif"] = [
        "Noto Sans CJK TC", "Heiti TC", "PingFang TC", "Arial Unicode MS", "DejaVu Sans",
    ]
    plt.rcParams["axes.unicode_minus"] = False

    C = {"bench": "#9CA3AF", "s1": "#F59E0B", "s2": "#EF4444", "s3": "#2563EB", "green": "#10B981"}

    def style(ax):
        ax.grid(True, alpha=0.22)
        ax.spines["top"].set_visible(False)
        ax.spines["right"].set_visible(False)

    # 1) 縮圖:四組 CAGR 對照
    rows = [row1, row2, row3, bench]
    labels = ["篩選法\n(對話一)", "加嚴篩選\n(對話二)", "排名法\n(對話三)", "0050\n含息"]
    colors = [C["s1"], C["s2"], C["s3"], C["bench"]]
    fig, ax = plt.subplots(figsize=(12, 6.75), dpi=140)
    bars = ax.bar(labels, [r["cagr"] for r in rows], color=colors)
    for b, r in zip(bars, rows):
        ax.text(b.get_x() + b.get_width() / 2, b.get_height() + 0.3, f"{r['cagr']:.1f}%",
                ha="center", fontsize=13, fontweight="bold")
    ax.set_title(f"AI Agent 三段對話的策略年化報酬({metrics['window_start'][:4]}–{END[:7]},含交易成本)",
                 fontsize=16, fontweight="bold")
    ax.set_ylabel("年化報酬 CAGR(%)")
    style(ax)
    fig.tight_layout()
    fig.savefig(STATIC_OUT / "chart_strategy_cagr_comparison.png")
    plt.close(fig)

    # 2) 四條淨值曲線(同起點歸一,對數刻度)
    eq_plot = eq.dropna(how="all").ffill()
    eq_plot = eq_plot[eq_plot.index >= common_start]
    eq_plot = eq_plot / eq_plot.iloc[0]
    fig, ax = plt.subplots(figsize=(12, 6.75), dpi=140)
    ax.plot(eq_plot.index, eq_plot["0050"], label="0050 含息", color=C["bench"], linewidth=2.2)
    ax.plot(eq_plot.index, eq_plot["screen_basic"], label="篩選法(對話一)", color=C["s1"], linewidth=1.8)
    ax.plot(eq_plot.index, eq_plot["screen_strict"], label="加嚴篩選(對話二)", color=C["s2"], linewidth=1.8)
    ax.plot(eq_plot.index, eq_plot["ranking_top20"], label="排名法(對話三)", color=C["s3"], linewidth=2.4)
    ax.set_yscale("log")
    ax.set_title("三個策略與 0050 淨值曲線(同起點歸一,對數刻度)", fontsize=16, fontweight="bold")
    ax.set_ylabel("淨值(起點 = 1)")
    ax.legend(loc="upper left")
    style(ax)
    fig.tight_layout()
    fig.savefig(STATIC_OUT / "chart_all_strategies_equity.png")
    plt.close(fig)

    # 3) 風險報酬散佈(CAGR vs MDD)
    fig, ax = plt.subplots(figsize=(12, 6.75), dpi=140)
    for r, label, color in zip(rows, ["篩選法", "加嚴篩選", "排名法", "0050 含息"], colors):
        ax.scatter(abs(r["max_drawdown"]), r["cagr"], s=260, color=color, zorder=3)
        ax.annotate(f"{label}\nCAGR {r['cagr']:.1f}%/MDD {r['max_drawdown']:.1f}%",
                    (abs(r["max_drawdown"]), r["cagr"]),
                    textcoords="offset points", xytext=(12, -4), fontsize=11)
    ax.set_xlabel("最大回撤(%,絕對值,越左越好)")
    ax.set_ylabel("年化報酬 CAGR(%,越上越好)")
    ax.set_title("風險與報酬:三個策略與 0050 的位置(CAGR vs 最大回撤)", fontsize=16, fontweight="bold")
    style(ax)
    fig.tight_layout()
    fig.savefig(STATIC_OUT / "chart_risk_return_scatter.png")
    plt.close(fig)

    # 4) 參數敏感度(真跑七組)
    fig, ax = plt.subplots(figsize=(12, 6.75), dpi=140)
    labels_s = [r["name"].replace("・", "\n") for r in sens_rows]
    vals = [r["cagr"] for r in sens_rows]
    bar_colors = [C["s3"]] + ["#60A5FA"] * (len(sens_rows) - 1)
    bars = ax.bar(labels_s, vals, color=bar_colors)
    for b, r in zip(bars, sens_rows):
        ax.text(b.get_x() + b.get_width() / 2, b.get_height() + 0.2, f"{r['cagr']:.1f}%",
                ha="center", fontsize=10.5, fontweight="bold")
    ax.axhline(bench["cagr"], color=C["bench"], linestyle="--", linewidth=1.6,
               label=f"0050 含息 {bench['cagr']:.1f}%")
    ax.set_title("參數敏感度:權重與持股數變動下的年化報酬(全部真跑)", fontsize=15, fontweight="bold")
    ax.set_ylabel("年化報酬 CAGR(%)")
    ax.tick_params(axis="x", labelsize=9.5)
    ax.legend(loc="lower right")
    style(ax)
    fig.tight_layout()
    fig.savefig(STATIC_OUT / "chart_sensitivity_weights.png")
    plt.close(fig)

    # 5) 三次股災回撤對照
    x = np.arange(len(crisis_rows))
    width = 0.38
    fig, ax = plt.subplots(figsize=(12, 6.75), dpi=140)
    b1 = ax.bar(x - width / 2, [abs(r["strategy_dd"]) for r in crisis_rows], width,
                color=C["s3"], label="排名法策略")
    b2 = ax.bar(x + width / 2, [abs(r["bench_dd"]) for r in crisis_rows], width,
                color=C["bench"], label="0050 含息")
    for bars_, key in ((b1, "strategy_dd"), (b2, "bench_dd")):
        for b, r in zip(bars_, crisis_rows):
            ax.text(b.get_x() + b.get_width() / 2, b.get_height() + 0.5, f"{abs(r[key]):.1f}%",
                    ha="center", fontsize=11, fontweight="bold")
    ax.set_xticks(x)
    ax.set_xticklabels([r["window"] for r in crisis_rows], fontsize=12)
    ax.set_ylabel("區間最大回撤(%,絕對值)")
    ax.set_title("三次市場大跌的區間最大回撤:排名法 vs 0050 含息", fontsize=16, fontweight="bold")
    ax.legend()
    style(ax)
    fig.tight_layout()
    fig.savefig(STATIC_OUT / "chart_crisis_drawdown.png")
    plt.close(fig)


try:
    save_charts()
except Exception as exc:
    print("chart generation skipped:", repr(exc))

print("\n=== 0050 基準 ===")
print(bench)
print("\n=== 三策略 ===")
for row in (row1, row2, row3):
    print(row)
print("\n=== 敏感度 ===")
for row in sens_rows:
    print(row)
print("\n=== 股災回撤 ===")
for row in crisis_rows:
    print(row)
print(f"\n=== 最新選股({picks_date},{len(picks)} 檔)===")
print(picks)
print(f"\nSaved outputs to {TMP_OUT} and {STATIC_OUT}")