# -*- coding: utf-8 -*-
"""data.indicator() 技術指標策略:RSI 超賣均值回歸(全台股一次算 158 種 TA-Lib 指標)。

對應文章:
  https://finlab.finance/blog/python-talib-158-technical-indicators

核心示範:
  finlab 的 data.indicator() 一行就把一個 TA-Lib / pandas-ta 指標
  套到全台股 2000+ 檔、約 19 年的價格上,回傳 FinlabDataFrame,
  取代「逐檔股票跑 talib.RSI() 再塞進 dict」的舊寫法。

執行環境:
  cd ~/Documents/finlab && UV_ENV_FILE=.env uv run --with TA-Lib --with matplotlib python \
    static/blog/python-talib-158-technical-indicators/strategy.py

輸出:
  - static/blog/python-talib-158-technical-indicators/metrics.json   (版控)
  - static/blog/python-talib-158-technical-indicators/data.csv       (版控,equity 對照表)
  - static/blog/python-talib-158-technical-indicators/indicator_examples.json (版控,真實樣本值)
  - static/blog/python-talib-158-technical-indicators/*.png          (4 張圖,上 R2)
  - static/blog/python-talib-158-technical-indicators/report_strategy.html (互動回測,上 R2)

回測口徑:
  - 區間:2016-01-01 ~ 資料最新日(約 10 年)
  - 股票池:60 日均量 > 500 萬股(流動性過濾,避免回測買得到實單買不到)
  - 訊號:data.indicator("RSI", timeperiod=14) < 30 視為超賣;月再平衡,等權持有
  - 交易成本:finlab sim() 台股預設(手續費 0.1425% + 證交稅 0.3%,賣出含稅)
  - 基準:0050 含息(etl:adj_close 還原價 buy-and-hold 純算術,同一視窗 .loc 對齊)

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

import json
import os
import warnings
from pathlib import Path

import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import numpy as np
import pandas as pd

import finlab
from finlab import data
from finlab.backtest import sim

warnings.filterwarnings("ignore")
finlab.login(os.environ["FINLAB_API_TOKEN"])

START = "2016-01-01"
END = os.environ.get("DATA_END", "2026-06-26")
OUT = Path(__file__).resolve().parent
OUT.mkdir(parents=True, exist_ok=True)

# 品牌色(§N)
C_STRAT = "#2563EB"
C_BENCH = "#9CA3AF"
C_POS = "#10B981"
C_NEG = "#EF4444"
C_ACCENT = "#F59E0B"

# 中文字型(避免方框)
from matplotlib import font_manager
for _fp in [
    os.path.expanduser("~/Library/Fonts/NotoSansCJKtc-DemiLight.otf"),
    "/System/Library/Fonts/Hiragino Sans GB.ttc",
    "/System/Library/Fonts/STHeiti Medium.ttc",
]:
    if os.path.exists(_fp):
        try:
            font_manager.fontManager.addfont(_fp)
        except Exception:
            pass
plt.rcParams.update({
    "figure.dpi": 120,
    "font.size": 12,
    "font.family": ["Noto Sans CJK TC", "Hiragino Sans GB", "Heiti TC", "sans-serif"],
    "axes.unicode_minus": False,
    "axes.grid": True,
    "grid.alpha": 0.25,
    "axes.spines.top": False,
    "axes.spines.right": False,
})


def cap(df):
    return df[df.index <= END]


# ---------------------------------------------------------------------------
# (a) data.indicator() 範例:真實樣本值(寫進文章的數字一律來自這裡)
# ---------------------------------------------------------------------------
print(">>> 計算 data.indicator() 範例 ...")

# 一行算全市場 RSI(14)
rsi = data.indicator("RSI", timeperiod=14)
# 一行算全市場 KD(STOCH 多輸出 -> k, d)
k, d = data.indicator("STOCH")
# 一行算全市場 MACD(三輸出 -> macd, signal, hist)
macd, macd_signal, macd_hist = data.indicator("MACD")
# 一行算全市場 20 日均線(用還原股價)
sma20 = data.indicator("SMA", adjust_price=True, timeperiod=20)

print(f"    RSI  shape={rsi.shape}  cols={rsi.shape[1]}  range={rsi.index[0].date()}~{rsi.index[-1].date()}")
print(f"    K/D  shape={k.shape}")
print(f"    MACD shape={macd.shape}")

# 取 2330 台積電最後一列的真實值
def last_real(series: pd.Series):
    s = series.dropna()
    return None if s.empty else (s.index[-1].strftime("%Y-%m-%d"), round(float(s.iloc[-1]), 4))

examples = {
    "as_of": END,
    "universe_stocks": int(rsi.shape[1]),
    "history_days": int(rsi.shape[0]),
    "history_start": rsi.index[0].strftime("%Y-%m-%d"),
    "history_end": rsi.index[-1].strftime("%Y-%m-%d"),
    "tsmc_2330": {
        "RSI14": last_real(rsi["2330"]),
        "STOCH_K": last_real(k["2330"]),
        "STOCH_D": last_real(d["2330"]),
        "MACD": last_real(macd["2330"]),
        "MACD_signal": last_real(macd_signal["2330"]),
        "MACD_hist": last_real(macd_hist["2330"]),
        "SMA20_adj": last_real(sma20["2330"]),
    },
}
# 全市場當日 RSI 分布:限定有正常交易的流動股票池
# (RSI=0 的死水股/停牌股會虛增超賣檔數,排除後才是可交易的真實超賣家數)
_vol_for_dist = cap(data.get("price:成交股數"))
_close_for_dist = cap(data.get("price:收盤價"))
_liquid_latest = (_vol_for_dist.rolling(60).mean() > 5_000_000).iloc[-1] & _close_for_dist.iloc[-1].notna()
rsi_latest = rsi.iloc[-1].where(_liquid_latest).dropna()
rsi_latest = rsi_latest[rsi_latest > 0]  # 排除 RSI 釘在 0 的無量股
examples["liquid_universe_today"] = int(len(rsi_latest))
examples["market_oversold_count"] = int((rsi_latest < 30).sum())
examples["market_overbought_count"] = int((rsi_latest > 70).sum())
examples["market_median_rsi"] = round(float(rsi_latest.median()), 2)

(OUT / "indicator_examples.json").write_text(
    json.dumps(examples, ensure_ascii=False, indent=2), encoding="utf-8"
)
print("    範例真實值:", json.dumps(examples["tsmc_2330"], ensure_ascii=False))
print(f"    全市場當日 RSI<30 超賣 {examples['market_oversold_count']} 檔，RSI>70 超買 {examples['market_overbought_count']} 檔")


# ---------------------------------------------------------------------------
# (b) 用 data.indicator 當訊號源的真實回測:RSI 超賣均值回歸
# ---------------------------------------------------------------------------
print(">>> 建立 RSI 超賣均值回歸策略 ...")

close = cap(data.get("price:收盤價"))
vol = cap(data.get("price:成交股數"))
adj = cap(data.get("etl:adj_close"))

rsi_c = cap(rsi)

# 流動性過濾:60 日均量 > 500 萬股
liquid = (vol.rolling(60).mean() > 5_000_000) & close.notna()

# 訊號:RSI(14) < 30 超賣 -> 等權買進;月再平衡
oversold = (rsi_c < 30) & liquid
position = oversold.astype(float)
# 等權正規化(留給 sim 也可,但先正規化讓 avg_holdings 容易計)
position = position.div(position.sum(axis=1).replace(0, np.nan), axis=0).fillna(0)
position = position[position.index >= START]

report = sim(position, resample="M", upload=False, fee_ratio=0.001425 / 2)  # 手續費 5 折


# ---------------------------------------------------------------------------
# 統計(與全站 canonical 0050 同式:純算術 creturn)
# ---------------------------------------------------------------------------
def stats(cr: pd.Series, name: str, avg_hold=None) -> dict:
    cr = cr[(cr.index >= START) & (cr.index <= END)]
    ret = cr.pct_change().dropna()
    mret = cr.resample("ME").last().pct_change().dropna()
    total = float(cr.iloc[-1] / cr.iloc[0] - 1)
    years = (cr.index[-1] - cr.index[0]).days / 365.25
    return {
        "name": name,
        "cagr": round(((1 + total) ** (1 / years) - 1) * 100, 2),
        "total_return": round(total * 100, 1),
        "daily_sharpe": round(float(ret.mean() / ret.std() * 252 ** 0.5), 2),
        "monthly_sharpe": round(float(mret.mean() / mret.std() * 12 ** 0.5), 2),
        "daily_sortino": round(float(ret.mean() / ret[ret < 0].std() * 252 ** 0.5), 2),
        "max_drawdown": round(float((cr / cr.cummax() - 1).min()) * 100, 2),
        "avg_holdings": avg_hold,
    }


strat_cr = report.creturn
strat_cr = strat_cr[(strat_cr.index >= START) & (strat_cr.index <= END)]
hold = (position[position.index >= START] > 0).sum(axis=1).replace(0, np.nan)
strat_stats = stats(strat_cr, "RSI 超賣均值回歸", avg_hold=round(float(hold.mean()), 1))

# ---------- 0050 含息基準(canonical 口徑,同一視窗 .loc 對齊) ----------
bser = adj["0050"]
bser = bser.loc[START:END].dropna()
# 把基準對齊到策略起算日(共同起始日 rebase)
bench_stats = stats(bser, "0050 含息", avg_hold=1.0)
bench_cr = bser / bser.iloc[0]

# 共同起始日 rebase(§N#1):兩條線都從同一天的 1.0 出發
common_start = max(strat_cr.index[0], bench_cr.index[0])
strat_eq = (strat_cr / strat_cr.loc[strat_cr.index >= common_start].iloc[0])
strat_eq = strat_eq[strat_eq.index >= common_start]
bench_eq = (bench_cr / bench_cr.loc[bench_cr.index >= common_start].iloc[0])
bench_eq = bench_eq[bench_eq.index >= common_start]
# 對齊到策略交易日 index
bench_eq = bench_eq.reindex(strat_eq.index).ffill()
bench_eq = bench_eq / bench_eq.iloc[0]
strat_eq = strat_eq / strat_eq.iloc[0]

metrics = {
    "slug": "python-talib-158-technical-indicators",
    "as_of": END,
    "backtest_window": f"{START} ~ {END}",
    "rebalance": "monthly",
    "universe": "60 日均量 > 500 萬股的上市櫃股票",
    "signal_source": 'data.indicator("RSI", timeperiod=14) < 30 視為超賣，等權持有',
    "fees": "finlab sim() 台股預設(手續費 0.1425% 打 5 折 + 證交稅賣出 0.3%)",
    "sharpe_basis": "daily_sharpe 為日頻年化；monthly_sharpe 為月頻年化(兩者不可混比)",
    "strategy": strat_stats,
    "benchmark_0050": bench_stats,
    "beats_0050_cagr": strat_stats["cagr"] > bench_stats["cagr"],
    "verdict": (
        "單一 RSI 超賣訊號的均值回歸策略在本文設定下、2016-2026 這段期間"
        + ("勝過" if strat_stats["cagr"] > bench_stats["cagr"] else "並未勝過")
        + "含息 0050；技術指標的價值在於把全市場一次算出來做為訊號層，而非單獨當聖杯。"
    ),
}
(OUT / "metrics.json").write_text(json.dumps(metrics, ensure_ascii=False, indent=2), encoding="utf-8")
print(">>> metrics.json 寫出")
print("    策略:", strat_stats)
print("    0050:", bench_stats)

# equity 對照表(版控 data.csv)
eq_df = pd.DataFrame({
    "date": strat_eq.index,
    "rsi_mean_reversion": strat_eq.values,
    "tw0050_total_return": bench_eq.values,
})
eq_df.to_csv(OUT / "data.csv", index=False)
print(">>> data.csv 寫出", eq_df.shape)


# ---------------------------------------------------------------------------
# 圖表(4 張 16:9,§N)
# ---------------------------------------------------------------------------
def finish(fig, ax, title, path):
    ax.set_title(title, fontsize=16, pad=16)
    fig.tight_layout()
    fig.savefig(path, dpi=120, bbox_inches="tight")
    plt.close(fig)
    print("    圖:", path.name)

FIG = (12.8, 7.2)  # 16:9

# 圖 1(縮圖):2330 台積電 KD 指標疊收盤價(指標 on price 範例)
fig, ax = plt.subplots(figsize=FIG)
seg = slice("2024-06-01", END)
px = adj["2330"].loc[seg]
ax.plot(px.index, px.values, color="#111827", linewidth=1.6, label="台積電 2330 還原收盤")
ax.set_ylabel("還原股價(元)")
ax2 = ax.twinx()
ax2.plot(k["2330"].loc[seg].index, k["2330"].loc[seg].values, color=C_STRAT, linewidth=1.3, label="KD 的 K 值", alpha=0.9)
ax2.plot(d["2330"].loc[seg].index, d["2330"].loc[seg].values, color=C_ACCENT, linewidth=1.3, label="KD 的 D 值", alpha=0.9)
ax2.axhline(80, color=C_NEG, linestyle="--", linewidth=1, alpha=0.6)
ax2.axhline(20, color=C_POS, linestyle="--", linewidth=1, alpha=0.6)
ax2.set_ylabel("KD 指標值(0-100)")
ax2.set_ylim(0, 100)
ax2.grid(False)
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m"))
lines1, labels1 = ax.get_legend_handles_labels()
lines2, labels2 = ax2.get_legend_handles_labels()
ax.legend(lines1 + lines2, labels1 + labels2, loc="upper left", fontsize=10, framealpha=0.9)
finish(fig, ax, "data.indicator 一行算出的 KD 指標疊在台積電 2330 股價上", OUT / "chart_kd_on_2330.png")

# 圖 2:策略 vs 0050 含息 權益曲線(log,共同起始日 rebase)
fig, ax = plt.subplots(figsize=FIG)
ax.plot(strat_eq.index, strat_eq.values, color=C_STRAT, linewidth=2.0, label=f"RSI 超賣均值回歸(CAGR {strat_stats['cagr']}%)")
ax.plot(bench_eq.index, bench_eq.values, color=C_BENCH, linewidth=2.0, label=f"0050 含息(CAGR {bench_stats['cagr']}%)")
ax.set_yscale("log")
ax.set_ylabel("累積淨值(log，起始=1)")
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y"))
ax.legend(loc="upper left", fontsize=11, framealpha=0.9)
finish(fig, ax, "RSI 超賣均值回歸 vs 0050 含息累積淨值(同一回測視窗)", OUT / "chart_equity_vs_0050.png")

# 圖 3:回撤(underwater)對照
fig, ax = plt.subplots(figsize=FIG)
strat_dd = (strat_eq / strat_eq.cummax() - 1) * 100
bench_dd = (bench_eq / bench_eq.cummax() - 1) * 100
ax.fill_between(strat_dd.index, strat_dd.values, 0, color=C_STRAT, alpha=0.35, label=f"策略(最大回撤 {strat_stats['max_drawdown']}%)")
ax.plot(bench_dd.index, bench_dd.values, color=C_NEG, linewidth=1.6, label=f"0050 含息(最大回撤 {bench_stats['max_drawdown']}%)")
ax.set_ylabel("回撤(%)")
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y"))
ax.legend(loc="lower left", fontsize=11, framealpha=0.9)
finish(fig, ax, "回撤對照:單一技術指標訊號的下檔風險", OUT / "chart_drawdown.png")

# 圖 4:全市場 RSI 分布(訊號分布圖,展示一行算全市場的威力)
fig, ax = plt.subplots(figsize=FIG)
vals = rsi_latest.values
ax.hist(vals, bins=40, color=C_STRAT, alpha=0.75, edgecolor="white")
ax.axvline(30, color=C_POS, linestyle="--", linewidth=1.6, label=f"RSI 30 超賣線(全市場 {examples['market_oversold_count']} 檔)")
ax.axvline(70, color=C_NEG, linestyle="--", linewidth=1.6, label=f"RSI 70 超買線(全市場 {examples['market_overbought_count']} 檔)")
ax.axvline(examples["market_median_rsi"], color=C_ACCENT, linewidth=1.6, label=f"全市場中位數 RSI {examples['market_median_rsi']}")
ax.set_xlabel("RSI(14) 值")
ax.set_ylabel("股票檔數")
ax.set_xlim(0, 100)
ax.legend(loc="upper right", fontsize=11, framealpha=0.9)
finish(fig, ax, f"一行 data.indicator 算出流動性股票池 {examples['liquid_universe_today']} 檔的 RSI 分布({END})", OUT / "chart_market_rsi_distribution.png")


# ---------------------------------------------------------------------------
# 互動回測 HTML(上 R2;非版控)
# ---------------------------------------------------------------------------
try:
    report.to_html(str(OUT / "report_strategy.html"))
    print(">>> report_strategy.html 寫出")
except Exception as e:
    print("report html 失敗(可忽略):", e)

print("\n=== 完成 ===")
print("策略 CAGR", strat_stats["cagr"], "% | 0050 含息 CAGR", bench_stats["cagr"], "%")
print("策略月夏普", strat_stats["monthly_sharpe"], "| 0050 月夏普", bench_stats["monthly_sharpe"])
print("勝過 0050:", metrics["beats_0050_cagr"])