chart_intraday_seaborn.py

"""
Elliott Wave Intraday Analysis Dashboard
Compares 60-minute vs 30-minute K-line backtests on TWII (Taiwan Stock Index)
using seaborn statistical visualization.
"""

import sys
import os
import numpy as np
import pandas as pd
import yfinance as yf
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import matplotlib.patches as mpatches
import seaborn as sns

# ── Import wave detection functions from backtest module ─────────────
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from elliott_wave_backtest import (
    find_swing_points,
    merge_swing_points,
    detect_impulse_waves,
    detect_corrective_waves,
    is_bearish_candle,
    is_bullish_candle,
    Wave,
    Trade,
)


# ── Data Fetching ────────────────────────────────────────────────────
def fetch_and_prepare(interval: str, period: str = "60d") -> pd.DataFrame:
    """Download TWII data and flatten MultiIndex columns if needed."""
    df = yf.download("^TWII", period=period, interval=interval, progress=False)
    if df.empty:
        raise ValueError(f"No data returned for ^TWII interval={interval}")
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = df.columns.get_level_values(0)
    df = df.dropna()
    return df


# ── Run wave detection + backtest (adapted from backtest module) ─────
def run_analysis(df: pd.DataFrame, swing_window: int) -> dict:
    """Detect waves and generate trades, returning full result dict."""
    swing_highs, swing_lows = find_swing_points(df, window=swing_window)
    zigzag = merge_swing_points(swing_highs, swing_lows)
    impulse_waves = detect_impulse_waves(zigzag)
    corrective_waves = detect_corrective_waves(zigzag)

    # ── Generate trades (same logic as run_backtest) ─────────────
    trades = []
    position = None

    signals = []
    for w in impulse_waves:
        end_idx = w.points[-1][0]
        end_price = w.points[-1][1]
        signals.append(("sell", end_idx, end_price, w))
    for w in corrective_waves:
        end_idx = w.points[-1][0]
        end_price = w.points[-1][1]
        signals.append(("buy", end_idx, end_price, w))
    signals.sort(key=lambda x: x[1])

    for sig_type, sig_idx, sig_price, wave in signals:
        entry_idx = sig_idx + 1
        if entry_idx >= len(df):
            continue
        if sig_type == "sell" and not is_bearish_candle(df, sig_idx):
            continue
        if sig_type == "buy" and not is_bullish_candle(df, sig_idx):
            continue

        entry_price = df["Open"].iloc[entry_idx]
        entry_date = df.index[entry_idx]

        if position is not None:
            exit_price = entry_price
            exit_date = entry_date
            if position["direction"] == "long":
                pnl = exit_price - position["entry_price"]
            else:
                pnl = position["entry_price"] - exit_price
            pnl_pct = pnl / position["entry_price"] * 100
            trades.append(Trade(
                entry_date=position["entry_date"], exit_date=exit_date,
                direction=position["direction"], entry_price=position["entry_price"],
                exit_price=exit_price, pnl=pnl, pnl_pct=pnl_pct,
            ))
            position = None

        if sig_type == "buy":
            stop_loss = wave.points[-1][1]
            direction = "long"
        else:
            stop_loss = wave.points[-1][1]
            direction = "short"

        position = {
            "direction": direction, "entry_price": entry_price,
            "entry_date": entry_date, "stop_loss": stop_loss,
        }

        risk = abs(entry_price - stop_loss)
        target = entry_price + (2 * risk if direction == "long" else -2 * risk)

        for j in range(entry_idx + 1, len(df)):
            hit_stop = hit_target = False
            if direction == "long":
                if df["Low"].iloc[j] <= stop_loss:
                    hit_stop = True
                if df["High"].iloc[j] >= target:
                    hit_target = True
            else:
                if df["High"].iloc[j] >= stop_loss:
                    hit_stop = True
                if df["Low"].iloc[j] <= target:
                    hit_target = True

            if hit_target:
                exit_price = target
                exit_date = df.index[j]
                pnl = (exit_price - entry_price) if direction == "long" else (entry_price - exit_price)
                pnl_pct = pnl / entry_price * 100
                trades.append(Trade(
                    entry_date=entry_date, exit_date=exit_date,
                    direction=direction, entry_price=entry_price,
                    exit_price=exit_price, pnl=pnl, pnl_pct=pnl_pct,
                ))
                position = None
                break

            if hit_stop:
                exit_price = stop_loss
                exit_date = df.index[j]
                pnl = (exit_price - entry_price) if direction == "long" else (entry_price - exit_price)
                pnl_pct = pnl / entry_price * 100
                trades.append(Trade(
                    entry_date=entry_date, exit_date=exit_date,
                    direction=direction, entry_price=entry_price,
                    exit_price=exit_price, pnl=pnl, pnl_pct=pnl_pct,
                ))
                position = None
                break

    if position is not None:
        exit_price = float(df["Close"].iloc[-1])
        exit_date = df.index[-1]
        pnl = (exit_price - position["entry_price"]) if position["direction"] == "long" else (position["entry_price"] - exit_price)
        pnl_pct = pnl / position["entry_price"] * 100
        trades.append(Trade(
            entry_date=position["entry_date"], exit_date=exit_date,
            direction=position["direction"], entry_price=position["entry_price"],
            exit_price=exit_price, pnl=pnl, pnl_pct=pnl_pct,
        ))

    return {
        "trades": trades,
        "impulse_waves": impulse_waves,
        "corrective_waves": corrective_waves,
        "zigzag": zigzag,
        "swing_highs": swing_highs,
        "swing_lows": swing_lows,
    }


# ── Helper: build wave phase array for coloring ─────────────────────
def build_wave_phases(df, impulse_waves, corrective_waves):
    """Return an array of phase labels ('impulse', 'corrective', 'neutral')
    for each bar in df, based on detected wave spans."""
    phases = np.full(len(df), "neutral", dtype=object)
    for w in impulse_waves:
        start_idx = w.points[0][0]
        end_idx = w.points[-1][0]
        phases[start_idx:end_idx + 1] = "impulse"
    for w in corrective_waves:
        start_idx = w.points[0][0]
        end_idx = w.points[-1][0]
        phases[start_idx:end_idx + 1] = "corrective"
    return phases


# ── Helper: compute cumulative return series ────────────────────────
def compute_cumulative_returns(trades, df):
    """Return DatetimeIndex-aligned cumulative return series for the wave strategy."""
    cum_ret = pd.Series(0.0, index=df.index, dtype=float)
    for t in trades:
        if t.exit_date is None:
            continue
        # distribute the return at exit date
        if t.exit_date in cum_ret.index:
            cum_ret.loc[t.exit_date] += t.pnl_pct / 100.0
    # convert individual returns to cumulative wealth
    wealth = (1 + cum_ret).cumprod()
    return wealth


# =====================================================================
#  MAIN DASHBOARD
# =====================================================================
def main():
    print("Fetching 60m data...")
    df_60m = fetch_and_prepare("60m", "60d")
    print(f"  -> {len(df_60m)} bars")

    print("Fetching 30m data...")
    df_30m = fetch_and_prepare("30m", "60d")
    print(f"  -> {len(df_30m)} bars")

    print("Running wave analysis (60m, swing_window=4)...")
    res_60m = run_analysis(df_60m, swing_window=4)
    print(f"  -> {len(res_60m['impulse_waves'])} impulse, "
          f"{len(res_60m['corrective_waves'])} corrective, "
          f"{len(res_60m['trades'])} trades")

    print("Running wave analysis (30m, swing_window=3)...")
    res_30m = run_analysis(df_30m, swing_window=3)
    print(f"  -> {len(res_30m['impulse_waves'])} impulse, "
          f"{len(res_30m['corrective_waves'])} corrective, "
          f"{len(res_30m['trades'])} trades")

    # ── Seaborn theme ────────────────────────────────────────────
    sns.set_theme(style="darkgrid", palette="muted", font_scale=0.95)
    muted = sns.color_palette("muted")

    fig, axes = plt.subplots(3, 2, figsize=(20, 14))
    fig.suptitle(
        "Elliott Wave Intraday Analysis | TWII | 60m vs 30m",
        fontsize=16, fontweight="bold", y=0.98,
    )

    datasets = [
        ("60-Minute", df_60m, res_60m, 4),
        ("30-Minute", df_30m, res_30m, 3),
    ]

    for col_idx, (label, df, res, sw) in enumerate(datasets):
        trades = res["trades"]
        impulse_waves = res["impulse_waves"]
        corrective_waves = res["corrective_waves"]
        phases = build_wave_phases(df, impulse_waves, corrective_waves)

        # ── ROW 0: Price + MA + wave phase regions + markers ─────
        ax0 = axes[0, col_idx]
        close = df["Close"].values.flatten()
        dates = df.index

        # Moving averages
        ma10 = pd.Series(close, index=dates).rolling(10).mean()
        ma20 = pd.Series(close, index=dates).rolling(20).mean()

        # Wave phase background shading
        for i in range(len(df) - 1):
            if phases[i] == "impulse":
                ax0.axvspan(dates[i], dates[i + 1], alpha=0.15, color="green", linewidth=0)
            elif phases[i] == "corrective":
                ax0.axvspan(dates[i], dates[i + 1], alpha=0.15, color="red", linewidth=0)

        ax0.plot(dates, close, color=muted[0], linewidth=0.9, alpha=0.85, label="Close")
        ax0.plot(dates, ma10, color=muted[1], linewidth=0.7, alpha=0.8, label="MA10")
        ax0.plot(dates, ma20, color=muted[2], linewidth=0.7, alpha=0.8, label="MA20")

        # Entry / Exit markers
        for t in trades:
            if t.direction == "long":
                ax0.scatter(t.entry_date, t.entry_price, marker="^", color="#2ecc71",
                            s=60, zorder=5, edgecolors="black", linewidths=0.4)
            else:
                ax0.scatter(t.entry_date, t.entry_price, marker="v", color="#e74c3c",
                            s=60, zorder=5, edgecolors="black", linewidths=0.4)
            if t.exit_date:
                exit_color = "#27ae60" if t.pnl > 0 else "#c0392b"
                ax0.scatter(t.exit_date, t.exit_price, marker="x",
                            color=exit_color, s=50, zorder=5, linewidths=1.2)

        ax0.set_title(f"{label} K-line  (swing_window={sw})", fontsize=11, fontweight="bold")
        ax0.set_ylabel("Price (TWD)")

        # Custom legend
        imp_patch = mpatches.Patch(color="green", alpha=0.25, label="Impulse Phase")
        cor_patch = mpatches.Patch(color="red", alpha=0.25, label="Corrective Phase")
        handles, labels_leg = ax0.get_legend_handles_labels()
        ax0.legend(handles=handles + [imp_patch, cor_patch], loc="upper left", fontsize=7,
                   framealpha=0.8, ncol=2)

        ax0.tick_params(axis="x", rotation=25, labelsize=7)
        ax0.tick_params(axis="y", labelsize=8)

        # ── ROW 1: Return distribution by wave phase ─────────────
        ax1 = axes[1, col_idx]

        if trades:
            # Classify each trade by the dominant phase at entry
            trade_returns = []
            for t in trades:
                entry_loc = df.index.get_indexer([t.entry_date], method="nearest")[0]
                phase = phases[entry_loc] if 0 <= entry_loc < len(phases) else "neutral"
                phase_label = phase.capitalize()
                trade_returns.append({"Return (%)": t.pnl_pct, "Phase": phase_label})

            tr_df = pd.DataFrame(trade_returns)

            # Histogram + KDE split by phase
            phase_colors = {"Impulse": muted[2], "Corrective": muted[3], "Neutral": muted[7]}
            for phase_name in ["Impulse", "Corrective", "Neutral"]:
                subset = tr_df[tr_df["Phase"] == phase_name]["Return (%)"]
                if len(subset) > 0:
                    sns.histplot(subset, kde=True if len(subset) > 2 else False,
                                 ax=ax1, color=phase_colors.get(phase_name, muted[0]),
                                 label=f"{phase_name} (n={len(subset)})",
                                 alpha=0.45, stat="density", bins=15, edgecolor="white", linewidth=0.3)

            # Statistics box
            all_rets = tr_df["Return (%)"]
            mean_r = all_rets.mean()
            std_r = all_rets.std()
            skew_r = float(all_rets.skew()) if len(all_rets) > 2 else 0.0
            try:
                kurt_r = float(all_rets.kurtosis()) if len(all_rets) > 3 else 0.0
            except Exception:
                kurt_r = 0.0

            stats_text = (
                f"n = {len(all_rets)}\n"
                f"mean = {mean_r:.3f}%\n"
                f"std  = {std_r:.3f}%\n"
                f"skew = {skew_r:.3f}\n"
                f"kurt = {kurt_r:.3f}"
            )
            ax1.text(0.97, 0.95, stats_text, transform=ax1.transAxes,
                     fontsize=8, verticalalignment="top", horizontalalignment="right",
                     bbox=dict(boxstyle="round,pad=0.4", facecolor="white", alpha=0.85, edgecolor="gray"))
            ax1.legend(fontsize=7, framealpha=0.8)
        else:
            ax1.text(0.5, 0.5, "No trades detected", transform=ax1.transAxes,
                     ha="center", va="center", fontsize=12, color="gray")

        ax1.set_title(f"{label} Return Distribution by Wave Phase", fontsize=11, fontweight="bold")
        ax1.set_xlabel("Trade Return (%)")
        ax1.set_ylabel("Density")

        # ── ROW 2: Cumulative return comparison ──────────────────
        ax2 = axes[2, col_idx]

        # Buy-and-hold
        bh_wealth = close / close[0]
        ax2.plot(dates, bh_wealth, color="#8e44ad", linewidth=1.2, alpha=0.85, label="Buy & Hold")

        # Wave strategy
        wave_wealth = compute_cumulative_returns(trades, df)
        ax2.plot(dates, wave_wealth.values, color=muted[0], linewidth=1.4, label="Wave Strategy")

        # Annotate final values
        bh_final = bh_wealth[-1]
        ws_final = wave_wealth.iloc[-1]
        ax2.annotate(f"{(bh_final - 1) * 100:+.2f}%",
                     xy=(dates[-1], bh_final), fontsize=8, color="#8e44ad",
                     fontweight="bold", ha="left", va="bottom",
                     xytext=(5, 3), textcoords="offset points")
        ax2.annotate(f"{(ws_final - 1) * 100:+.2f}%",
                     xy=(dates[-1], ws_final), fontsize=8, color=muted[0],
                     fontweight="bold", ha="left", va="top",
                     xytext=(5, -3), textcoords="offset points")

        ax2.axhline(1.0, color="gray", linestyle="--", linewidth=0.5, alpha=0.6)
        ax2.set_title(f"{label} Cumulative Return: Wave Strategy vs Buy & Hold", fontsize=11, fontweight="bold")
        ax2.set_ylabel("Growth of $1")
        ax2.legend(loc="upper left", fontsize=8, framealpha=0.8)
        ax2.tick_params(axis="x", rotation=25, labelsize=7)
        ax2.tick_params(axis="y", labelsize=8)

    # ── Footer / source text ─────────────────────────────────────
    fig.text(
        0.5, 0.005,
        "Source: Yahoo Finance (^TWII) | Elliott Wave detection with swing-point algorithm | "
        "Generated with seaborn + matplotlib",
        ha="center", fontsize=7.5, color="gray", style="italic",
    )

    plt.tight_layout(rect=[0, 0.02, 1, 0.96])
    output_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "chart_intraday_seaborn.png")
    plt.savefig(output_path, dpi=200, bbox_inches="tight", facecolor="white")
    print(f"\nDashboard saved to: {output_path}")
    plt.close()


if __name__ == "__main__":
    main()