"""Walk-forward 重放：用滑动窗口逐步重算缠论，检验买卖点在实时环境下是否稳定。

要回答三个问题：
  1. 幻影率——实时曾报出、但在完整历史上并不存在的信号占多少？
  2. 撤销率——报出后又被结构演化抹掉的信号占多少？
  3. 真实滞后——信号第一次可被观测到的时刻，比 sure_time 晚多少？
"""
from __future__ import annotations

import os
import pickle
import sys
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from chanlun.core.ChanEnum import Chan_BSP_DIR

# 子进程共享的只读数据，避免每次任务重复 pickle 整个 DataFrame
_G: dict = {}

BspKey = tuple[str, int, str]  # (类型, 方向, 分型时间)


def _init_worker(df: pd.DataFrame, tf: str, zs_source: str) -> None:
    _G["df"] = df
    _G["tf"] = tf
    _G["zs_source"] = zs_source


def _bsp_keys_for_window(args: tuple[int, int]) -> tuple[int, list[BspKey]]:
    """在 df[start:end] 这个窗口上重算缠论，返回该时点可观测到的买卖点集合。"""
    from chanlun import TF_DF  # 延迟导入，避免父进程重复加载

    start, end = args
    df = _G["df"].iloc[start:end].reset_index(drop=True)
    try:
        chan = TF_DF(df, 1, _G["tf"])
        if _G["zs_source"] == "pure":
            bi_zs_list = chan.cal_bi_zs_list_pure(chan.bi_list)
        else:
            bi_zs_list = chan.cal_bi_zs(chan.seg_list)
        bsp_list = chan.find_all_bsp(chan.bi_list, bi_zs_list) if bi_zs_list else []
    except Exception:
        return end - 1, []

    keys = [
        (
            str(b.type).replace("Chan_BSP_TYPE.", ""),
            1 if b.dir == Chan_BSP_DIR.BUY else -1,
            str(b.end_time),
        )
        for b in bsp_list
        if b.is_sure and b.sure_time is not None
    ]
    return end - 1, keys


@dataclass
class ReplayResult:
    observations: dict[int, set[BspKey]]  # 每个重算时点 -> 当时可见的买卖点集合
    checkpoints: list[int]
    window: int
    step: int


def replay(
    df: pd.DataFrame,
    tf: str,
    window: int = 3000,
    step: int = 6,
    workers: int | None = None,
    cache_key: str | None = None,
    zs_source: str = "seg",
) -> ReplayResult:
    """滑动窗口重放。窗口右端每 step 根前进一次，每次完整重算一遍缠论。

    重放很贵（分钟级），cache_key 非空时把结果落盘复用。
    """
    n = len(df)
    if n <= window:
        raise ValueError(f"数据长度 {n} 不足以支撑窗口 {window}")

    cache_path = None
    if cache_key:
        cache_dir = Path(__file__).resolve().parents[1] / ".cache"
        cache_dir.mkdir(parents=True, exist_ok=True)
        cache_path = cache_dir / f"replay__{cache_key}__{zs_source}__w{window}_s{step}_n{n}.pkl"
        if cache_path.exists():
            with cache_path.open("rb") as fh:
                obs = pickle.load(fh)
            print(f"  [cache] 命中重放缓存 {cache_path.name}")
            return ReplayResult(observations=obs, checkpoints=sorted(obs), window=window, step=step)

    tasks = [(end - window, end) for end in range(window, n + 1, step)]
    workers = workers or max(1, (os.cpu_count() or 4) - 1)

    observations: dict[int, set[BspKey]] = {}
    with ProcessPoolExecutor(
        max_workers=workers, initializer=_init_worker, initargs=(df, tf, zs_source)
    ) as pool:
        for i, (bar_idx, keys) in enumerate(pool.map(_bsp_keys_for_window, tasks, chunksize=8)):
            observations[bar_idx] = set(keys)
            if (i + 1) % 500 == 0:
                print(f"  ...{i + 1}/{len(tasks)} 窗口", flush=True)

    if cache_path is not None:
        with cache_path.open("wb") as fh:
            pickle.dump(observations, fh)

    return ReplayResult(observations=observations, checkpoints=sorted(observations), window=window, step=step)


def analyze_stability(
    result: ReplayResult,
    final_keys: set[BspKey],
    df: pd.DataFrame,
    window: int,
    maturity_bars: int = 300,
    edge_buffer: int = 500,
) -> pd.DataFrame:
    """把重放结果整理成每个信号的生命周期：首次出现、最后可见、是否被撤销。

    两处偏差必须剔除，否则统计会失真：
      - 左截断：分型早于第一个窗口起点的信号，其 first_seen 是假的，标记 truncated。
      - 右截断：临近重放末尾才出现的信号还没经历足够演化，谈不上"没被撤销"，标记 immature。
    """
    checkpoints = result.checkpoints
    times = df["date"].dt.strftime("%Y-%m-%d %H:%M:%S").to_numpy()
    idx_of = {t: i for i, t in enumerate(times)}

    first_seen: dict[BspKey, int] = {}
    last_seen: dict[BspKey, int] = {}
    seen_count: dict[BspKey, int] = {}
    for bar_idx in checkpoints:
        for key in result.observations[bar_idx]:
            first_seen.setdefault(key, bar_idx)
            last_seen[key] = bar_idx
            seen_count[key] = seen_count.get(key, 0) + 1

    last_cp = checkpoints[-1]
    rows = []
    for key, first in first_seen.items():
        bsp_type, direction, fx_time = key
        fx_idx = idx_of.get(fx_time)

        # 窗口是滑动的，信号的分型一旦滑出窗口左端就必然不可见。
        # 只在「信号仍被窗口覆盖」的检查点上评价持续性，否则会把正常的滑出误判成撤销。
        # 另需 edge_buffer：贴着窗口左端时缠论缺少前置K线构造包含关系，信号会因边界效应消失。
        if fx_idx is None:
            in_scope = [cp for cp in checkpoints if cp >= first]
        else:
            in_scope = [
                cp for cp in checkpoints
                if cp >= first and (cp - window) < (fx_idx - edge_buffer)
            ]
        scope_end = in_scope[-1] if in_scope else first
        visible_in_scope = sum(1 for cp in in_scope if key in result.observations[cp])

        rows.append(
            {
                "bsp_type": bsp_type,
                "direction": direction,
                "fx_time": fx_time,
                "fx_idx": fx_idx,
                "first_seen_idx": first,
                "first_seen_time": times[first],
                "last_seen_idx": last_seen[key],
                "observed_lag": (first - fx_idx) if fx_idx is not None else None,
                "scope_checkpoints": len(in_scope),
                "persist_ratio": visible_in_scope / len(in_scope) if in_scope else 0.0,
                "in_final": key in final_keys,
                # 在「仍被窗口覆盖」的最后一个检查点上是否还活着
                "alive_at_scope_end": key in result.observations.get(scope_end, set()),
                # 分型发生在首个窗口完全覆盖之后，first_seen 才是真实的
                "truncated": fx_idx is None or fx_idx < window,
                # 覆盖范围太短则谈不上"没被撤销"
                "immature": len(in_scope) < maturity_bars // max(result.step, 1),
            }
        )
    return pd.DataFrame(rows).sort_values("first_seen_idx").reset_index(drop=True)