"""韓股崩盤會拖累台股嗎？KOSPI 與台股、美股的歷史連動（FinLab 量化研究，2026-10）

重現文章 /blog/kospi-crash-taiwan-stock-impact 的核心數字：
  1. KOSPI 跌破一年高點 20% 之後，美股與台股未來一年的表現 vs 平常日子
  2. 「韓股距高點 15% 以上、台股／美股卻在高點附近」的背離狀態，之後一年的結果
  3. 韓國融資高點與美股高點的時間距離
  4. 月資料：KOSPI 過去 6 個月報酬對台股未來 6 個月報酬的預測力

資料：同目錄 data.csv（1990-01 至 2026-09-30 日收盤）
  kospi                  韓國綜合股價指數（KOFIA／Naver 官方收盤，1996-12 起）
  taiex                  台股加權指數（TWSE 1990–1998、finlab taiex_total_index 1999 起）
  sp500                  S&P 500
  kr_margin_trillion_krw 韓國信用融資餘額，兆韓元（KOFIA，KOSPI＋KOSDAQ）

執行：pip install pandas numpy statsmodels，然後 python strategy.py
"""
from pathlib import Path

import numpy as np
import pandas as pd

HERE = Path(__file__).parent
TRADING_YEAR = 252
NEAR_HIGH = -0.05          # 距一年高點 5% 以內 = 在高點附近
KOSPI_EVENT = -0.20        # KOSPI 跌破一年高點 20%
REARM = -0.05              # 回到高點 5% 以內才重新計算下一次
DIVERGENCE = -0.15         # 背離：KOSPI 距一年高點 15% 以上
EPISODE_GAP_DAYS = 120     # 120 天內再次進入算同一段
N_PLACEBO = 10_000         # 隨機日期抽樣次數
SEED = 20260930
MARGIN_DECLINE = 0.20      # 融資高點：之後餘額掉 20% 以上
PRICE_DECLINE = 0.10       # 美股高點：之後指數跌 10% 以上
KR_MARGIN_START = "2007-01-01"
MAX_MATCH_DAYS = 365       # 美股高點距融資高點超過一年 = 這一輪美股還沒出現對應高點


# ---------- 1. 載入資料 ----------
def load_prices() -> pd.DataFrame:
    df = pd.read_csv(HERE / "data.csv", index_col="date", parse_dates=True)
    return df


def drawdown_from_high(s: pd.Series, window: int = TRADING_YEAR) -> pd.Series:
    """收盤價距過去一年最高點的跌幅（負數）。"""
    return s / s.rolling(window, min_periods=window // 2).max() - 1


def forward_outcomes(s: pd.Series) -> pd.DataFrame:
    """每個交易日之後一年的報酬、途中最深跌幅，與當天距高點的位置。"""
    s = s.dropna()
    future_low = s[::-1].rolling(TRADING_YEAR, min_periods=TRADING_YEAR).min()[::-1].shift(-1)
    return pd.DataFrame({
        "r12": s.shift(-TRADING_YEAR) / s - 1,
        "mdd12": future_low / s - 1,
        "dd_now": drawdown_from_high(s),
    })


def value_on(df: pd.DataFrame, dates: pd.DatetimeIndex) -> pd.DataFrame:
    """取每個日期當天或之前最近一個交易日的數值（兩地休市日不同）。"""
    pos = df.index.searchsorted(dates, side="right") - 1
    out = df.iloc[pos[pos >= 0]].copy()
    out.index = dates[pos >= 0]
    return out


# ---------- 2. KOSPI 跌破 20% 事件 ----------
def kospi_events(kospi: pd.Series) -> pd.DatetimeIndex:
    armed, events = True, []
    for day, dd in drawdown_from_high(kospi.dropna()).dropna().items():
        if armed and dd <= KOSPI_EVENT:
            events.append(day)
            armed = False
        elif not armed and dd >= REARM:
            armed = True
    return pd.DatetimeIndex(events)


def placebo_p(pool: np.ndarray, n: int, observed: float, rng: np.random.Generator) -> float:
    """隨機抽 n 個日期、重複 N_PLACEBO 次，平均跌幅比實際更深的比例。"""
    means = pool[rng.integers(0, len(pool), size=(N_PLACEBO, n))].mean(axis=1)
    return float((means <= observed).mean())


def event_table(prices: pd.DataFrame) -> pd.DataFrame:
    kospi = prices["kospi"].dropna()
    start = drawdown_from_high(kospi).first_valid_index()
    events = kospi_events(kospi)
    rng = np.random.default_rng(SEED)
    rows = {}
    for name in ("sp500", "taiex"):
        out = forward_outcomes(prices[name])
        ev = value_on(out, events).dropna(subset=["mdd12"])
        base = out[out.index >= start].dropna(subset=["mdd12"])
        rows[name] = {
            "事件數": len(ev),
            "事件後一年最深跌幅": ev.mdd12.mean(), "平常": base.mdd12.mean(),
            "隨機日期 p 值": placebo_p(base.mdd12.values, len(ev), ev.mdd12.mean(), rng),
            "事件後跌超過一成機率": (ev.mdd12 <= -0.10).mean(), "平常機率": (base.mdd12 <= -0.10).mean(),
            "事件後一年報酬": ev.r12.mean(), "平常報酬": base.r12.mean(),
        }
    print("KOSPI 跌破 20% 的日期：", [d.date() for d in events])
    return pd.DataFrame(rows).T


# ---------- 3. 背離狀態 ----------
def days_to_low(s: pd.Series, start: pd.Timestamp) -> int:
    """進入日之後一年內，最低收盤出現在第幾個交易日。"""
    i = s.index.searchsorted(start)
    return int(np.argmin(s.iloc[i:i + TRADING_YEAR + 1].values))


def divergence_episodes(prices: pd.DataFrame, target: str,
                        threshold: float = DIVERGENCE) -> tuple[pd.DataFrame, dict]:
    out = forward_outcomes(prices[target]).dropna(subset=["mdd12"])
    kospi_dd = drawdown_from_high(prices["kospi"].dropna()).reindex(out.index, method="ffill")
    out = out[kospi_dd.notna()]
    kospi_dd = kospi_dd[kospi_dd.notna()]
    near = out.dd_now >= NEAR_HIGH
    state = near & (kospi_dd <= threshold)

    starts = out.index[state & ~state.shift(1, fill_value=False)]
    keep = np.r_[True, np.diff(starts.values).astype("timedelta64[D]").astype(int) > EPISODE_GAP_DAYS]
    episodes = out.loc[starts[keep]].assign(kospi_dd=kospi_dd.loc[starts[keep]])
    target_px = prices[target].dropna()
    episodes["days_to_low"] = [days_to_low(target_px, d) for d in episodes.index]

    base = out[near]
    summary = {
        "段數": len(episodes),
        "一年內跌超過一成": (episodes.mdd12 <= -0.10).mean(), "平常跌超過一成": (base.mdd12 <= -0.10).mean(),
        "一年內跌超過兩成": (episodes.mdd12 <= -0.20).mean(), "平常跌超過兩成": (base.mdd12 <= -0.20).mean(),
        "一年報酬中位數": episodes.r12.median(), "平常報酬中位數": base.r12.median(),
        "平常日數": len(base),
    }
    return episodes, summary


# ---------- 4. 融資高點 vs 美股高點 ----------
def cycle_peaks(s: pd.Series, decline: float, window: int = 2 * TRADING_YEAR) -> pd.DatetimeIndex:
    """兩年新高，而且在被超越之前跌超過 decline 的日子。"""
    s = s.dropna()
    values = s.values
    is_high = (s >= s.rolling(window, min_periods=window // 4).max()).values
    peaks = []
    for i in np.flatnonzero(is_high):
        later = values[i + 1:]
        higher = np.flatnonzero(later > values[i])
        segment = later[:higher[0]] if len(higher) else later
        if len(segment) and segment.min() <= values[i] * (1 - decline):
            peaks.append(s.index[i])
    return pd.DatetimeIndex(peaks)


def margin_vs_sp500(prices: pd.DataFrame) -> pd.Series:
    margin_peaks = cycle_peaks(prices["kr_margin_trillion_krw"], MARGIN_DECLINE)
    margin_peaks = margin_peaks[margin_peaks >= KR_MARGIN_START]
    spx_peaks = cycle_peaks(prices["sp500"], PRICE_DECLINE)
    gaps = {}
    for d in margin_peaks:
        diff = (spx_peaks - d).days
        gaps[d.date()] = int(diff[np.argmin(np.abs(diff))])
    return pd.Series(gaps, name="美股高點減融資高點（天，正＝美股較晚）")


# ---------- 5. 月資料預測力 ----------
def monthly_lead(prices: pd.DataFrame, months: int = 6) -> None:
    import statsmodels.api as sm

    month_end = np.log(prices[["kospi", "taiex"]].resample("ME").last().dropna())
    y = month_end.taiex.shift(-months) - month_end.taiex
    x = pd.DataFrame({"kospi_past": month_end.kospi - month_end.kospi.shift(months),
                      "taiex_past": month_end.taiex - month_end.taiex.shift(months)})
    df = pd.concat([y.rename("taiex_future"), x], axis=1).dropna()
    fit = sm.OLS(df.taiex_future, sm.add_constant(df[["kospi_past", "taiex_past"]])).fit(
        cov_type="HAC", cov_kwds={"maxlags": 12})
    print(f"\nKOSPI 過去 {months} 月 → 台股未來 {months} 月：係數 {fit.params.kospi_past:.3f}，"
          f"t 值 {fit.tvalues.kospi_past:.2f}（n={len(df)}）")


if __name__ == "__main__":
    pd.set_option("display.width", 200, "display.float_format", "{:.3f}".format)
    prices = load_prices()

    print("== KOSPI 跌破 20% 之後 ==")
    print(event_table(prices))

    for target, threshold in (("sp500", DIVERGENCE), ("taiex", DIVERGENCE), ("taiex", -0.10)):
        episodes, summary = divergence_episodes(prices, target, threshold)
        print(f"\n== 背離狀態：KOSPI 距高點 {-threshold:.0%} 以上，{target} 在高點附近 ==")
        print(episodes[["kospi_dd", "dd_now", "r12", "mdd12", "days_to_low"]])
        print(pd.Series(summary).round(3).to_string())

    print("\n== 韓國融資高點 vs 美股高點 ==")
    gaps = margin_vs_sp500(prices)
    print(gaps.to_string())
    matched = gaps[gaps.abs() <= MAX_MATCH_DAYS]
    print(f"有對應美股高點的 {len(matched)} 次：距離中位數 {matched.abs().median():.0f} 天")

    monthly_lead(prices)