#!/usr/bin/env python3
"""Backtest harian Patah Tren dengan aturan entry/SL/TP yang sama seperti dashboard.

Backtest memakai data OHLC Yahoo Finance, menunggu breakout/retest, dan memakai
asumsi konservatif: jika SL dan TP tersentuh pada candle harian yang sama, SL
dianggap terjadi lebih dulu.
"""

import json
import os
import time
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import timedelta

import pandas as pd

from neobdm_common import (
    PREMIUM_WATCHLIST_TICKERS,
    WIB,
    datetime,
    get_idx_tick_offset,
    round_idx_tick,
)
from tv_enrichment import atomic_write_json


OUTPUT_FILE = "trend_break_backtest.json"
HISTORY_RANGE = "1y"
ENTRY_WINDOW = 5
HOLD_BARS = 10


def fetch_yahoo_daily_history(ticker, history_range=HISTORY_RANGE):
    symbol = f"{ticker}.JK"
    url = (
        "https://query2.finance.yahoo.com/v8/finance/chart/"
        f"{urllib.parse.quote(symbol)}?range={history_range}&interval=1d&events=history"
    )
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
    with urllib.request.urlopen(req, timeout=20) as response:
        payload = json.loads(response.read().decode("utf-8"))
    result = ((payload.get("chart") or {}).get("result") or [None])[0]
    if not result:
        return None
    timestamps = result.get("timestamp") or []
    quote = ((((result.get("indicators") or {}).get("quote") or [{}])[0]) or {})
    rows = []
    for idx, stamp in enumerate(timestamps):
        try:
            values = {
                "Open": quote["open"][idx],
                "High": quote["high"][idx],
                "Low": quote["low"][idx],
                "Close": quote["close"][idx],
                "Volume": quote["volume"][idx],
            }
            if any(values[key] is None for key in ("Open", "High", "Low", "Close")):
                continue
            rows.append({
                "Date": datetime.fromtimestamp(stamp, tz=WIB).date().isoformat(),
                **{key: float(value or 0) for key, value in values.items()},
            })
        except (IndexError, KeyError, TypeError, ValueError):
            continue
    if len(rows) < 80:
        return None
    frame = pd.DataFrame(rows)
    frame.index = pd.to_datetime(frame.pop("Date"))
    return frame


def trade_plan(signal):
    entry = round_idx_tick(float(signal["swing_high"]))
    entry_low = get_idx_tick_offset(entry, -1)
    entry_high = get_idx_tick_offset(entry, 1)
    sl = min(get_idx_tick_offset(entry, -2), round_idx_tick(entry * 0.98))
    support_ref = min(float(signal.get("sma20") or entry), float(signal.get("sma50") or entry), entry)
    range_target = entry + max(entry - support_ref, 0)
    t1 = round_idx_tick(max(entry * 1.05, range_target))
    if t1 <= entry:
        t1 = round_idx_tick(entry * 1.03)
    t2 = round_idx_tick(max(entry * 1.10, t1 * 1.05))
    return {
        "entry_low": entry_low,
        "entry_price": entry,
        "entry_high": entry_high,
        "sl_price": sl,
        "t1_price": t1,
        "t2_price": t2,
    }


def find_entry(df, signal_idx, signal, plan):
    signal_close = float(df.iloc[signal_idx]["Close"])
    if signal["status"] == "CONFIRM" and plan["entry_low"] <= signal_close <= plan["entry_high"]:
        return signal_idx, "READY"

    end = min(len(df), signal_idx + ENTRY_WINDOW + 2)
    breakout_idx = signal_idx if signal["status"] == "CONFIRM" else None
    for idx in range(signal_idx + 1, end):
        bar = df.iloc[idx]
        if breakout_idx is None:
            if float(bar["High"]) >= plan["entry_high"]:
                breakout_idx = idx
            continue
        # Untuk sinyal NEAR, retest baru dihitung setelah candle breakout selesai.
        if signal["status"] == "NEAR" and idx <= breakout_idx:
            continue
        if float(bar["Low"]) <= plan["entry_high"] and float(bar["High"]) >= plan["entry_low"]:
            return idx, "WAIT_RETEST"
    return None, None


def bucket_for_score(score):
    if score >= 90:
        return "SCORE_9_PLUS"
    if score >= 80:
        return "SCORE_8_PLUS"
    return "SCORE_LT_8"


def backtest_ticker(ticker, df):
    # Import lokal mencegah circular import saat scanner memanggil ensure_backtest().
    from scan_trend_break import detect_trend_break

    trades = []
    idx = 69
    while idx < len(df) - 1:
        signal = detect_trend_break(df.iloc[:idx + 1])
        if not signal:
            idx += 1
            continue
        plan = trade_plan(signal)
        entry_idx, action = find_entry(df, idx, signal, plan)
        if entry_idx is None:
            idx += 1
            continue

        exit_idx = min(len(df) - 1, entry_idx + HOLD_BARS)
        status = "TIME_EXIT"
        exit_price = float(df.iloc[exit_idx]["Close"])
        resolved_idx = exit_idx
        for bar_idx in range(entry_idx + 1, exit_idx + 1):
            bar = df.iloc[bar_idx]
            # Konservatif untuk candle ambigu: SL diprioritaskan.
            if float(bar["Low"]) <= plan["sl_price"]:
                status = "CUT_LOSS"
                exit_price = plan["sl_price"]
                resolved_idx = bar_idx
                break
            if float(bar["High"]) >= plan["t1_price"]:
                status = "TAKE_PROFIT"
                exit_price = plan["t1_price"]
                resolved_idx = bar_idx
                break

        risk = plan["entry_price"] - plan["sl_price"]
        trade_r = (exit_price - plan["entry_price"]) / risk if risk > 0 else None
        rr = (plan["t1_price"] - plan["entry_price"]) / risk if risk > 0 else 0
        signal_class = signal["status"]
        source_suffix = "Ready" if action == "READY" else "Retest"
        trades.append({
            "ticker": ticker,
            "entry_date": df.index[entry_idx].date().isoformat(),
            "close_date": df.index[resolved_idx].date().isoformat(),
            "signal_date": df.index[idx].date().isoformat(),
            "source": f"Patah Tren {signal_class.title()} {source_suffix}",
            "signal_class": signal_class,
            "action_bucket": action,
            "score_bucket": bucket_for_score(float(signal.get("score") or 0)),
            "rr_bucket": "RR_2_PLUS" if rr >= 2 else ("RR_1_3_PLUS" if rr >= 1.3 else "RR_LOW"),
            "status": status,
            **plan,
            "exit_price": round(float(exit_price), 2),
            "return_pct": round((exit_price - plan["entry_price"]) / plan["entry_price"] * 100, 2),
            "label": 1 if exit_price > plan["entry_price"] else 0,
            "trade_r": round(trade_r, 3) if trade_r is not None else None,
            "total_score": round(float(signal.get("score") or 0) / 10, 2),
            "live_rr": round(rr, 2),
            "risk_pct": round(risk / plan["entry_price"] * 100, 2) if plan["entry_price"] else 0,
            "break_margin_pct": signal.get("break_margin_pct"),
            "swing_high": signal.get("swing_high"),
            "vol_ratio": signal.get("vol_ratio"),
            "downtrend_pct": signal.get("downtrend_below_pct"),
            "source_detail": (
                f"Backtest D1; entry {action}; hold maks {HOLD_BARS} bar; "
                "candle ambigu dihitung SL lebih dulu"
            ),
        })
        # Jangan menghitung sinyal tumpang-tindih untuk ticker yang sama.
        idx = resolved_idx + 1
    return trades


def run_backtest(tickers=None):
    tickers = sorted(set(tickers or PREMIUM_WATCHLIST_TICKERS))
    histories = {}
    errors = []
    with ThreadPoolExecutor(max_workers=6) as pool:
        futures = {pool.submit(fetch_yahoo_daily_history, ticker): ticker for ticker in tickers}
        for future in as_completed(futures):
            ticker = futures[future]
            try:
                frame = future.result()
                if frame is not None:
                    histories[ticker] = frame
                else:
                    errors.append(ticker)
            except Exception:
                errors.append(ticker)

    trades = []
    for ticker, frame in histories.items():
        trades.extend(backtest_ticker(ticker, frame))
    trades.sort(key=lambda row: (row["entry_date"], row["ticker"]))
    resolved = list(trades)
    wins = sum(row.get("label") == 1 for row in resolved)
    payload = {
        "updated_at": datetime.now().isoformat(timespec="seconds"),
        "method": "historical_backtest",
        "history_range": HISTORY_RANGE,
        "entry_window_bars": ENTRY_WINDOW,
        "hold_bars": HOLD_BARS,
        "assumption": "Daily OHLC; wait breakout/retest; same-candle ambiguity counts SL first.",
        "tickers_requested": len(tickers),
        "tickers_loaded": len(histories),
        "tickers_failed": errors,
        "closed_trades": len(resolved),
        "wins": wins,
        "losses": len(resolved) - wins,
        "win_rate": round(wins / len(resolved) * 100, 1) if resolved else None,
        "take_profit_trades": sum(
            row["status"] == "TAKE_PROFIT" for row in resolved
        ),
        "cut_loss_trades": sum(
            row["status"] == "CUT_LOSS" for row in resolved
        ),
        "time_exit_trades": sum(
            row["status"] == "TIME_EXIT" for row in resolved
        ),
        "trades": trades,
    }
    atomic_write_json(OUTPUT_FILE, payload, indent=2)
    return payload


def ensure_backtest(max_age_hours=18):
    if os.path.exists(OUTPUT_FILE):
        age = datetime.now() - datetime.fromtimestamp(os.path.getmtime(OUTPUT_FILE))
        try:
            with open(OUTPUT_FILE) as handle:
                existing = json.load(handle)
        except (OSError, ValueError):
            existing = {}
        if age < timedelta(hours=max_age_hours) and existing.get("method") == "historical_backtest" and existing.get("closed_trades", 0) > 0:
            return existing
    return run_backtest()


if __name__ == "__main__":
    result = run_backtest()
    print(
        f"Win rate Patah Tren: {result['win_rate']}% "
        f"({result['wins']}W-{result['losses']}L, N={result['closed_trades']})"
    )