backtest_modified.py

"""
修改策略回測：與原始策略比較
差異：+30% 賣70%, +50% 再賣20%, 剩10%永久持有
"""
import warnings
warnings.filterwarnings("ignore")

import sys, os
import numpy as np
import pandas as pd

sys.stdout.reconfigure(encoding='utf-8')

# ── Config ──
INITIAL_CAPITAL = 1_000_000
TICKER_0050 = "0050.TW"
TICKER_TAIEX = "^TWII"
START_DATE = "2008-01-01"
END_DATE = "2026-02-18"

KD_PERIOD = 9
KD_SMOOTH = 3
ENTRY_MONTHS = range(9, 13)
KD_THRESHOLD = 50
DRAWDOWN_THRESHOLD = -0.10
TRANCHE_PCT = [0.30, 0.30, 0.40]

COMMISSION_RATE = 0.001425
TAX_RATE = 0.003


def download_data(ticker, start, end):
    import yfinance as yf
    print(f"  下載 {ticker} ...")
    df = yf.download(ticker, start=start, end=end, auto_adjust=True, progress=False)
    if df.empty:
        raise RuntimeError(f"無法取得 {ticker}")
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)
    df.index = pd.to_datetime(df.index).tz_localize(None)
    return df


def to_weekly(df):
    return df.resample("W-FRI").agg({
        "Open": "first", "High": "max", "Low": "min",
        "Close": "last", "Volume": "sum",
    }).dropna()


def compute_kd(df):
    low_min = df["Low"].rolling(window=KD_PERIOD).min()
    high_max = df["High"].rolling(window=KD_PERIOD).max()
    denom = (high_max - low_min).replace(0, np.nan)
    rsv = ((df["Close"] - low_min) / denom * 100).fillna(50)
    k = rsv.rolling(window=KD_SMOOTH).mean()
    d = k.rolling(window=KD_SMOOTH).mean()
    return k, d


def buy_fee(amount):
    return amount * COMMISSION_RATE

def sell_fee(amount):
    return amount * COMMISSION_RATE + amount * TAX_RATE


def run_original(merged, cap):
    """原始策略：+30% 賣70%, +50% 全數出清"""
    cash = float(cap)
    shares = 0
    cost_basis = 0.0
    tranches = 0
    cycle_year = None
    partial_sold = False
    last_buy = None
    trades = []
    equity = []

    for date, row in merged.iterrows():
        price = row["ETF_Close"]
        k_val = row["K"]
        taiex_dd = row["TAIEX_DD"]
        month = date.month
        year = date.year

        if month >= 9 and cycle_year != year:
            cycle_year = year
            tranches = 0

        # EXIT
        if shares > 0 and cost_basis > 0:
            pnl = (shares * price - cost_basis) / cost_basis

            if not partial_sold and pnl >= 0.30:
                n_sell = max((int(shares * 0.70) // 1000) * 1000, 1000)
                n_sell = min(n_sell, shares)
                gross = n_sell * price
                fee = sell_fee(gross)
                frac = n_sell / shares
                cost_basis -= cost_basis * frac
                shares -= n_sell
                cash += gross - fee
                trades.append(dict(date=date, action="SELL", price=round(price,2),
                    shares=n_sell, reason=f"+{pnl*100:.1f}% sell 70%"))
                partial_sold = True

            elif partial_sold and shares > 0:
                pnl2 = (shares * price - cost_basis) / cost_basis if cost_basis > 0 else 0
                if pnl2 >= 0.50:
                    n_sell = shares
                    gross = n_sell * price
                    fee = sell_fee(gross)
                    cost_basis = 0
                    shares = 0
                    cash += gross - fee
                    trades.append(dict(date=date, action="SELL", price=round(price,2),
                        shares=n_sell, reason=f"+{pnl2*100:.1f}% sell ALL"))
                    partial_sold = False
                    tranches = 0

        # ENTRY
        if month in ENTRY_MONTHS and tranches < 3 and last_buy != date:
            sig_kd = (not np.isnan(k_val)) and k_val < KD_THRESHOLD
            sig_dd = taiex_dd <= DRAWDOWN_THRESHOLD
            if sig_kd or sig_dd:
                frac = TRANCHE_PCT[tranches]
                budget = min(cap * frac, cash)
                cost_per = price * (1 + COMMISSION_RATE)
                lots = int(budget / (cost_per * 1000))
                if lots > 0:
                    n_buy = lots * 1000
                    gross = n_buy * price
                    fee = buy_fee(gross)
                    cash -= gross + fee
                    shares += n_buy
                    cost_basis += gross + fee
                    last_buy = date
                    tranches += 1
                    trades.append(dict(date=date, action="BUY", price=round(price,2),
                        shares=n_buy, reason=f"Tranche {tranches}"))

        equity.append(dict(date=date, equity=cash + shares * price,
                           cash=cash, shares=shares, price=price))

    return trades, equity


def run_modified(merged, cap):
    """修改策略：+30% 賣70%, +50% 再賣剩餘的2/3, 保留~10%永久"""
    cash = float(cap)
    active_shares = 0       # 可交易部位
    permanent_shares = 0    # 永久持有部位
    cost_basis = 0.0        # 僅追蹤 active_shares 的成本
    tranches = 0
    cycle_year = None
    partial_sold = False
    last_buy = None
    trades = []
    equity = []

    for date, row in merged.iterrows():
        price = row["ETF_Close"]
        k_val = row["K"]
        taiex_dd = row["TAIEX_DD"]
        month = date.month
        year = date.year

        if month >= 9 and cycle_year != year:
            cycle_year = year
            tranches = 0

        # EXIT (only active_shares participate)
        if active_shares > 0 and cost_basis > 0:
            pnl = (active_shares * price - cost_basis) / cost_basis

            if not partial_sold and pnl >= 0.30:
                # 賣70% of active
                n_sell = max((int(active_shares * 0.70) // 1000) * 1000, 1000)
                n_sell = min(n_sell, active_shares)
                gross = n_sell * price
                fee = sell_fee(gross)
                frac = n_sell / active_shares
                cost_basis -= cost_basis * frac
                active_shares -= n_sell
                cash += gross - fee
                trades.append(dict(date=date, action="SELL", price=round(price,2),
                    shares=n_sell, reason=f"+{pnl*100:.1f}% sell 70%"))
                partial_sold = True

            elif partial_sold and active_shares > 0:
                pnl2 = (active_shares * price - cost_basis) / cost_basis if cost_basis > 0 else 0
                if pnl2 >= 0.50:
                    # 再賣剩餘的 2/3，保留 1/3 轉永久
                    n_keep = max((int(active_shares * (1/3)) // 1000) * 1000, 0)
                    n_sell = active_shares - n_keep
                    if n_sell > 0:
                        gross = n_sell * price
                        fee = sell_fee(gross)
                        frac_sold = n_sell / active_shares
                        cost_basis -= cost_basis * frac_sold
                        active_shares -= n_sell
                        cash += gross - fee
                        trades.append(dict(date=date, action="SELL", price=round(price,2),
                            shares=n_sell, reason=f"+{pnl2*100:.1f}% sell 2/3 of remaining"))

                    # 剩餘 active → permanent
                    if active_shares > 0:
                        permanent_shares += active_shares
                        trades.append(dict(date=date, action="HOLD_PERMANENT",
                            price=round(price,2), shares=active_shares,
                            reason=f"轉永久持有 (累計永久: {permanent_shares})"))
                        active_shares = 0
                        cost_basis = 0

                    partial_sold = False
                    tranches = 0

        # ENTRY
        if month in ENTRY_MONTHS and tranches < 3 and last_buy != date:
            sig_kd = (not np.isnan(k_val)) and k_val < KD_THRESHOLD
            sig_dd = taiex_dd <= DRAWDOWN_THRESHOLD
            if sig_kd or sig_dd:
                frac = TRANCHE_PCT[tranches]
                budget = min(cap * frac, cash)
                cost_per = price * (1 + COMMISSION_RATE)
                lots = int(budget / (cost_per * 1000))
                if lots > 0:
                    n_buy = lots * 1000
                    gross = n_buy * price
                    fee = buy_fee(gross)
                    cash -= gross + fee
                    active_shares += n_buy
                    cost_basis += gross + fee
                    last_buy = date
                    tranches += 1
                    trades.append(dict(date=date, action="BUY", price=round(price,2),
                        shares=n_buy, reason=f"Tranche {tranches}"))

        total_shares = active_shares + permanent_shares
        equity.append(dict(date=date, equity=cash + total_shares * price,
                           cash=cash, active_shares=active_shares,
                           permanent_shares=permanent_shares, price=price))

    return trades, equity


def compute_metrics(equity_hist, label, cap):
    eq = pd.DataFrame(equity_hist)
    eq["date"] = pd.to_datetime(eq["date"])
    eq = eq.set_index("date").sort_index()
    e = eq["equity"]

    end_eq = e.iloc[-1]
    total_ret = (end_eq / cap) - 1
    days = (e.index[-1] - e.index[0]).days
    years = days / 365.25
    cagr = (end_eq / cap) ** (1/years) - 1 if years > 0 else 0

    peak = e.expanding().max()
    dd = (e - peak) / peak
    max_dd = dd.min()

    daily_ret = e.pct_change().dropna()
    vol = daily_ret.std() * np.sqrt(252)
    excess = daily_ret
    sharpe = (excess.mean() / excess.std() * np.sqrt(252)) if excess.std() > 0 else 0

    down = daily_ret[daily_ret < 0]
    down_std = down.std() * np.sqrt(252) if len(down) > 0 else 0
    sortino = (daily_ret.mean() * 252 / down_std) if down_std > 0 else 0

    return {
        "label": label,
        "最終權益": f"{end_eq:,.0f}",
        "總報酬": f"{total_ret:.2%}",
        "CAGR": f"{cagr:.2%}",
        "最大回撤": f"{max_dd:.2%}",
        "年化波動率": f"{vol:.2%}",
        "Sharpe": f"{sharpe:.3f}",
        "Sortino": f"{sortino:.3f}",
        "_end_eq": end_eq,
        "_total_ret": total_ret,
        "_cagr": cagr,
        "_max_dd": max_dd,
        "_sharpe": sharpe,
    }


def main():
    sep = "=" * 70
    print(f"\n{sep}")
    print("  0050 原始策略 vs 修改策略 回測比較")
    print(sep)

    # Download
    etf_daily = download_data(TICKER_0050, START_DATE, END_DATE)
    taiex_daily = download_data(TICKER_TAIEX, START_DATE, END_DATE)

    etf_w = to_weekly(etf_daily)
    taiex_w = to_weekly(taiex_daily)

    etf_w["K"], etf_w["D"] = compute_kd(etf_w)
    taiex_w["Peak"] = taiex_w["Close"].cummax()
    taiex_w["Drawdown"] = (taiex_w["Close"] - taiex_w["Peak"]) / taiex_w["Peak"]

    merged = etf_w[["Close", "K"]].join(taiex_w[["Drawdown"]], how="inner")
    merged = merged.dropna()
    merged.columns = ["ETF_Close", "K", "TAIEX_DD"]

    print(f"  週線資料: {len(merged)} 筆 ({merged.index[0].date()} ~ {merged.index[-1].date()})")

    cap = INITIAL_CAPITAL

    # ── 原始策略 ──
    print(f"\n{sep}")
    print("  原始策略回測")
    print(sep)
    orig_trades, orig_equity = run_original(merged, cap)
    orig_m = compute_metrics(orig_equity, "原始策略", cap)
    print(f"  交易: {len(orig_trades)} 筆")
    print(f"  最終權益: TWD {orig_m['_end_eq']:,.0f}")
    print(f"  總報酬: {orig_m['_total_ret']:.2%}")
    print(f"  CAGR: {orig_m['_cagr']:.2%}")
    print(f"  最大回撤: {orig_m['_max_dd']:.2%}")
    print(f"  Sharpe: {orig_m['_sharpe']:.3f}")

    # ── 修改策略 ──
    print(f"\n{sep}")
    print("  修改策略回測 (保留10%永久部位)")
    print(sep)
    mod_trades, mod_equity = run_modified(merged, cap)
    mod_m = compute_metrics(mod_equity, "修改策略", cap)

    # Count permanent shares
    final_eq = mod_equity[-1]
    perm = final_eq.get("permanent_shares", 0)
    active = final_eq.get("active_shares", 0)

    print(f"  交易: {len(mod_trades)} 筆")
    print(f"  最終權益: TWD {mod_m['_end_eq']:,.0f}")
    print(f"  總報酬: {mod_m['_total_ret']:.2%}")
    print(f"  CAGR: {mod_m['_cagr']:.2%}")
    print(f"  最大回撤: {mod_m['_max_dd']:.2%}")
    print(f"  Sharpe: {mod_m['_sharpe']:.3f}")
    print(f"  永久持有股數: {perm:,}")
    print(f"  活躍部位股數: {active:,}")

    # ── Buy & Hold ──
    print(f"\n{sep}")
    print("  Buy & Hold 基準")
    print(sep)
    first_price = etf_daily.iloc[0]["Close"]
    bh_lots = int(cap / (first_price * (1+COMMISSION_RATE) * 1000))
    bh_shares = bh_lots * 1000
    bh_fee = buy_fee(bh_shares * first_price)
    bh_cash = cap - bh_shares * first_price - bh_fee
    bh_equity = [dict(date=d, equity=bh_cash + bh_shares * row["Close"],
                      cash=bh_cash, shares=bh_shares, price=row["Close"])
                 for d, row in etf_daily.iterrows()]
    bh_m = compute_metrics(bh_equity, "Buy & Hold", cap)
    print(f"  最終權益: TWD {bh_m['_end_eq']:,.0f}")
    print(f"  總報酬: {bh_m['_total_ret']:.2%}")
    print(f"  CAGR: {bh_m['_cagr']:.2%}")

    # ── 比較表 ──
    print(f"\n{sep}")
    print("  三策略比較")
    print(sep)

    metrics = [orig_m, mod_m, bh_m]
    keys = ["最終權益", "總報酬", "CAGR", "最大回撤", "年化波動率", "Sharpe", "Sortino"]

    print(f"\n{'指標':<14}", end="")
    for m in metrics:
        print(f"  {m['label']:>16}", end="")
    print()
    print("-" * (14 + 18 * len(metrics)))
    for k in keys:
        print(f"{k:<14}", end="")
        for m in metrics:
            print(f"  {m[k]:>16}", end="")
        print()

    # ── Trade log ──
    print(f"\n{sep}")
    print("  修改策略交易明細")
    print(sep)
    for t in mod_trades:
        print(f"  {t['date'].strftime('%Y-%m-%d')} {t['action']:<16} {t.get('shares',0):>8,} @ {t['price']:>8.2f}  {t['reason']}")

    # ── Save CSVs ──
    out_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
    os.makedirs(out_dir, exist_ok=True)

    pd.DataFrame(mod_trades).to_csv(
        os.path.join(out_dir, "modified_strategy_trades.csv"),
        index=False, encoding="utf-8-sig")

    # Comparison summary
    comp = {}
    for m in metrics:
        comp[m["label"]] = {k: m[k] for k in keys}
    pd.DataFrame(comp).to_csv(
        os.path.join(out_dir, "comparison_3way.csv"), encoding="utf-8-sig")

    print(f"\n  已儲存: output/modified_strategy_trades.csv")
    print(f"  已儲存: output/comparison_3way.csv")
    print()


if __name__ == "__main__":
    main()