#!/usr/bin/env python3
"""Измерительный стенд для 5м-методов: загрузка вселенной, издержки, метрики честности.

Здесь только «измерение», без правил: правила живут в tools/methods.py и возвращают сырые сделки
(time, symbol, side, gross_bp). Издержки, плацебо, рандом-контроль, OOS, block-bootstrap, разбивка
по годам и просадка считаются ЦЕНТРАЛЬНО — одинаково для всех методов, чтобы их нельзя было
«подкрутить» внутри метода.

Издержки (от нотионала, как требует цель):
  тейкер 0,055% на ногу + проскальзывание 1 б.п. (топ-30 по обороту) или 3 б.п. (остальные),
  на круг (вход+выход) = 2 × (5,5 + слиппедж) б.п. Мейкер 0,02% на ногу — отдельным прогоном.
"""
from __future__ import annotations

import glob
import json
import os

import numpy as np
import pandas as pd

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
FEE_BP_TAKER, FEE_BP_MAKER = 5.5, 2.0     # 0,055% и 0,02% на ногу
SLIP_BP_LIQUID, SLIP_BP_OTHER = 1.0, 3.0
TOP_N_LIQUID = 30
BAR_MS = 300_000
BARS_PER_DAY = 288


# ------------------------------------------------------------- загрузка ----
def list_pairs(data_dir: str | None = None) -> list[str]:
    d = data_dir or os.path.join(ROOT, "data", "5m")
    return sorted(os.path.basename(p).replace("_5m.csv", "") for p in glob.glob(os.path.join(d, "*_5m.csv")))


def load_long(pairs: list[str] | None = None, data_dir: str | None = None) -> pd.DataFrame:
    """Все пары в один длинный кадр: ts, symbol, open, high, low, close, volume."""
    frames = []
    for sym in (pairs or list_pairs(data_dir)):
        path = os.path.join(data_dir or os.path.join(ROOT, "data", "5m"), f"{sym}_5m.csv")
        if not os.path.isfile(path):
            continue
        df = pd.read_csv(path, header=None, names=["ts", "open", "high", "low", "close", "volume"],
                         usecols=[0, 1, 2, 3, 4, 5])
        df = df.drop_duplicates("ts")
        df["symbol"] = sym
        frames.append(df)
    if not frames:
        return pd.DataFrame(columns=["ts", "symbol", "open", "high", "low", "close", "volume"])
    return pd.concat(frames, ignore_index=True).sort_values(["symbol", "ts"]).reset_index(drop=True)


def pivot_close(long_df: pd.DataFrame) -> pd.DataFrame:
    """Матрица close: индекс — 5м-сетка времени, колонки — пары (с продолжением вперёд)."""
    m = long_df.pivot_table(index="ts", columns="symbol", values="close", aggfunc="last")
    m = m.sort_index()
    m = m.ffill(limit=12)  # дырки больше часа не тянем
    return m


def turnover_top(long_df: pd.DataFrame, n: int = TOP_N_LIQUID) -> set[str]:
    """Топ-N пар по обороту (средний объём × цена) — для проскальзывания и теста ликвидности."""
    g = long_df.groupby("symbol").apply(
        lambda d: float((d["volume"] * d["close"]).tail(20_000).mean()) if len(d) else 0.0,
        include_groups=False)
    return set(g.sort_values(ascending=False).head(n).index)


def cost_bp_per_turn(symbol: str, liquid: set[str], maker: bool = False) -> float:
    """Издержки на круг (вход+выход) в б.п. от нотионала позиции."""
    fee = FEE_BP_MAKER if maker else FEE_BP_TAKER
    slip = SLIP_BP_LIQUID if symbol in liquid else SLIP_BP_OTHER
    return 2.0 * (fee + slip)


def apply_costs(trades: pd.DataFrame, liquid: set[str], maker: bool = False) -> pd.DataFrame:
    """Издержки начисляются НА МЕСТЕ (без копии) — на полной вселенной копия не влезает в RAM."""
    if trades.empty:
        return trades.assign(ret_bp=[])
    trades["cost_bp"] = trades["symbol"].map(lambda s: cost_bp_per_turn(s, liquid, maker))
    trades["ret_bp"] = trades["gross_bp"] - trades["cost_bp"]
    return trades


# -------------------------------------------------------------- метрики ----
def block_bootstrap_t(x: np.ndarray, block: int = BARS_PER_DAY, iters: int = 2000, seed: int = 11) -> float:
    """t-статистика среднего по блочному бутстрапу (блоки по дню) — вместо наивного t."""
    x = np.asarray(x, dtype=float)
    x = x[np.isfinite(x)]
    if len(x) < 10:
        return 0.0
    rng = np.random.default_rng(seed)
    n = len(x)
    nb = int(np.ceil(n / block))
    means = np.empty(iters)
    for k in range(iters):
        starts = rng.integers(0, max(1, n - block), size=nb)
        s = np.concatenate([x[i:i + block] for i in starts])[:n]
        means[k] = s.mean()
    se = means.std(ddof=1)
    return float(means.mean() / se) if se > 0 else 0.0


def daily_series(trades: pd.DataFrame) -> pd.Series:
    if trades.empty:
        return pd.Series(dtype=float)
    d = pd.to_datetime(trades["time"], unit="ms", utc=True).dt.floor("D")
    return trades.groupby(d)["ret_bp"].sum()


def sharpe_and_dd(trades: pd.DataFrame) -> tuple[float, float]:
    s = daily_series(trades)
    if len(s) < 5:
        return 0.0, 0.0
    sharpe = float(s.mean() / s.std(ddof=1) * np.sqrt(365)) if s.std(ddof=1) > 0 else 0.0
    eq = s.cumsum()
    dd = float((eq - eq.cummax()).min())  # в б.п.
    return sharpe, dd


def summarize(method: str, trades: pd.DataFrame, placebo: pd.DataFrame, rand: pd.DataFrame,
              oos_frac: float = 0.5) -> dict:
    """Строка сводки по методу (все метрики считаются здесь, а не в методе)."""
    if trades.empty:
        return {"method": method, "n_trades": 0, "ev_bp": 0.0, "ev_bp_oos": 0.0, "t_boot": 0.0,
                "placebo_ev_bp": 0.0, "rand_ev_bp": 0.0, "sharpe": 0.0, "max_dd_pct": 0.0, "by_year_ok": "нет"}
    tr = trades.sort_values("time").reset_index(drop=True)
    cut = tr["time"].quantile(1.0 - oos_frac)
    oos = tr[tr["time"] >= cut]
    sharpe, dd_bp = sharpe_and_dd(tr)
    daily = daily_series(tr)
    years = pd.to_datetime(tr["time"], unit="ms", utc=True).dt.year
    by_year = tr.groupby(years)["ret_bp"].mean()
    pos_years = int((by_year > 0).sum())
    return {
        "method": method,
        "n_trades": int(len(tr)),
        "ev_bp": round(float(tr["ret_bp"].mean()), 2),
        "ev_bp_oos": round(float(oos["ret_bp"].mean()), 2) if len(oos) else 0.0,
        "t_boot": round(block_bootstrap_t(daily.to_numpy(), block=5), 2),
        "placebo_ev_bp": round(float(placebo["ret_bp"].mean()), 2) if not placebo.empty else 0.0,
        "rand_ev_bp": round(float(rand["ret_bp"].mean()), 2) if not rand.empty else 0.0,
        "sharpe": round(sharpe, 2),
        "max_dd_pct": round(dd_bp / 100.0, 2),  # б.п. суммы → % «в единицах выигрыша»
        "by_year_ok": "да" if pos_years >= 2 else "нет",
        "by_year": {int(y): round(float(v), 2) for y, v in by_year.items()},
    }


def save_trades(method: str, trades: pd.DataFrame) -> None:
    d = os.path.join(ROOT, "results", "methods")
    os.makedirs(d, exist_ok=True)
    cols = [c for c in ("time", "symbol", "side", "gross_bp", "cost_bp", "ret_bp") if c in trades.columns]
    trades.sort_values("time")[cols].to_csv(os.path.join(d, f"{method}.csv"), index=False)


def save_by_year(rows: list[dict], tag: str = "") -> None:
    recs = []
    for r in rows:
        for y, v in (r.get("by_year") or {}).items():
            recs.append({"method": r["method"], "year": y, "ev_bp": v, "n": r["n_trades"]})
    d = os.path.join(ROOT, "reports")
    os.makedirs(d, exist_ok=True)
    pd.DataFrame(recs).to_csv(os.path.join(d, f"by_year{tag}.csv"), index=False)


def save_counts(n_tests: int, tag: str = "") -> None:
    """Порог Бонферрони для числа проверенных вариантов (двусторонний, alpha=0,05)."""
    from scipy import stats
    thr = float(stats.norm.ppf(1 - 0.05 / (2 * max(n_tests, 1))))
    d = os.path.join(ROOT, "reports")
    os.makedirs(d, exist_ok=True)
    with open(os.path.join(d, f"count_tests{tag}.json"), "w") as fh:
        json.dump({"n_tests": int(n_tests), "bonferroni_t": round(thr, 2),
                   "alpha": 0.05, "note": "двусторонний порог |t| для числа проверенных вариантов"}, fh,
                  ensure_ascii=False, indent=1)


# ------------------------------------------- потоковая работа (экономно по памяти) ----
def iter_pairs(data_dir: str | None = None):
    """По одной паре за раз: 250 пар × 210 тыс. баров не влезают в RAM (на машине 7 ГБ)."""
    d = data_dir or os.path.join(ROOT, "data", "5m")
    for sym in list_pairs(d):
        path = os.path.join(d, f"{sym}_5m.csv")
        if not os.path.isfile(path):
            continue
        df = pd.read_csv(path, header=None, names=["ts", "open", "high", "low", "close", "volume"])
        df = df.drop_duplicates("ts").sort_values("ts").reset_index(drop=True)
        yield sym, df


def build_matrix(data_dir: str | None = None):
    """Матрица close (float32) на общей 5м-сетке + набор ликвидных пар. Память ≈ 200 МБ."""
    d = data_dir or os.path.join(ROOT, "data", "5m")
    syms, tss, closes, turns = [], {}, {}, {}
    for sym, df in iter_pairs(d):
        syms.append(sym)
        tss[sym] = df["ts"].to_numpy()
        closes[sym] = df["close"].to_numpy(np.float32)
        tail = df.tail(20_000)
        turns[sym] = float((tail["volume"] * tail["close"]).mean())
    grid = np.unique(np.concatenate(list(tss.values())))
    C = np.full((len(grid), len(syms)), np.nan, dtype=np.float32)
    for j, sym in enumerate(syms):
        C[np.searchsorted(grid, tss[sym]), j] = closes[sym]
    liquid = set(sorted(turns, key=lambda s: -turns[s])[:TOP_N_LIQUID])
    hours = pd.to_datetime(grid, unit="ms", utc=True).hour.to_numpy()
    return {"grid": grid, "C": C, "syms": syms, "liquid": liquid, "hours": hours}