ForexBot2/backtest/research/book_report.py

"""Full descriptive report of a backtest trade book (research only).

Usage:
    python backtest/research/book_report.py --book BOOK.csv --bar-minutes N \
        [--split YYYY-MM-DD|median] [--parquet-4h PATH] [--parquet-1d PATH] \
        [--out DIR]

Prints (and writes to DIR/report.txt) headline metrics for the whole book
and for each half, group tables (weekday / decision hour / session / quality
/ regime / vol_regime / direction / strategy / exit_reason / indicators_agree),
win-loss streaks, a rolling win-rate summary, and a quantile-bucket screen of
every numeric decision-time feature the book carries (`f_*` / `x_*` columns
plus score, confidence, rr_planned, indicators_agree/oppose and `ind_*_value`).

The screen is deliberately half-split: a feature only "passes" when the
top-bucket-minus-bottom-bucket mean-R gap has the SAME SIGN in both halves of
the book and both extreme buckets are populated in both halves. Bin edges are
fit on the WHOLE book (so the halves share edges and stay comparable) — that
leaks a little cross-half information into the edges themselves, which is why
the sign test, not the size of the gap, is the thing to trust.

Everything here is measurement, not simulation: it never re-runs the signal
engine, it only describes a book that gpu_backtest.py already produced.
Analysis tooling only — never imported by the live code.
"""
from __future__ import annotations

import argparse
import math
import os
from typing import Iterable

import numpy as np
import pandas as pd

SESSIONS = (
    ("asian", 3, 6),
    ("london", 7, 12),
    ("overlap", 13, 16),
    ("ny", 17, 22),
)

# Numeric non-`f_`/`x_` columns that are known at DECISION time and are
# therefore legal screen candidates. bars_held / r_multiple / mfe_r / mae_r
# are post-entry outcomes and must never appear here.
BASE_SCREEN_COLS = ("score", "confidence", "rr_planned",
                    "indicators_agree", "indicators_oppose")


def session_of(hour: int) -> str:
    for name, lo, hi in SESSIONS:
        if lo <= hour <= hi:
            return name
    return "off"


def feature_columns(df: pd.DataFrame) -> list[str]:
    """Decision-time feature columns carried by this book, generically."""
    extra = [c for c in df.columns if c.startswith(("f_", "x_"))]
    base = [c for c in BASE_SCREEN_COLS if c in df.columns]
    ind = [c for c in df.columns
           if c.startswith("ind_") and c.endswith("_value")]
    out, seen = [], set()
    for c in extra + base + ind:
        if c not in seen:
            seen.add(c)
            out.append(c)
    return out


# ───────────────────────────── loading ──────────────────────────────

def load_book(csv: str, bar_minutes: int = 5,
              split: str = "median") -> pd.DataFrame:
    """Read a gpu_backtest trade book and add the decision-time columns.

    `decision_time` = entry_time − one bar: entries fill at the NEXT bar's
    open, so the bar the signal actually decided on closed one bar earlier.
    That is the clock blocked_hours_utc and every session filter evaluate.
    `half` is 1 before the split instant and 2 at/after it.
    """
    df = pd.read_csv(csv)
    for col in ("entry_time", "exit_time"):
        if col in df.columns:
            df[col] = pd.to_datetime(df[col], utc=True, format="mixed")
    df["decision_time"] = df["entry_time"] - pd.Timedelta(minutes=bar_minutes)
    df["decision_hour"] = df["decision_time"].dt.hour
    df["weekday"] = df["decision_time"].dt.weekday
    df["session"] = df["decision_hour"].map(session_of)
    df = df.sort_values("entry_time", kind="mergesort").reset_index(drop=True)
    df["half"] = _halves(df["entry_time"], split)
    return df


def split_instant(entry_time: pd.Series, split: str) -> pd.Timestamp:
    if split == "median":
        return entry_time.median()
    return pd.Timestamp(split, tz="UTC")


def _halves(entry_time: pd.Series, split: str) -> pd.Series:
    cut = split_instant(entry_time, split)
    return np.where(entry_time < cut, 1, 2)


# ───────────────────────────── metrics ──────────────────────────────

def base_metrics(r: pd.Series, df: pd.DataFrame | None = None) -> dict:
    """Headline stats for a series of R multiples.

    max_dd uses the gpu_backtest.compute_metrics convention: a 0.0 is
    prepended to the cumulative-R curve so an opening losing streak counts
    as drawdown from flat, not from an already-negative first trade.
    pf/expectancy split on r > 0 / r < 0 (exact zeros are neither).
    """
    r = pd.Series(r, dtype="float64").dropna()
    n = int(len(r))
    if n == 0:
        return {"n": 0, "total_r": 0.0, "mean_r": 0.0, "median_r": 0.0,
                "wr": 0.0, "pf": 0.0, "max_dd": 0.0, "sharpe": 0.0,
                "expectancy": 0.0, "expectancy_alt": 0.0, "recovery": 0.0,
                "avg_bars": float("nan"), "avg_mfe": float("nan"),
                "avg_mae": float("nan")}
    wins, losses = r[r > 0], r[r < 0]
    gross_win = float(wins.sum())
    gross_loss = float(-losses.sum())
    cum = np.concatenate(([0.0], np.cumsum(r.to_numpy())))
    max_dd = float((np.maximum.accumulate(cum) - cum).max())
    std = float(r.std(ddof=0))
    mean_r = float(r.mean())
    wr = len(wins) / n
    lr = len(losses) / n
    avg_win = float(wins.mean()) if len(wins) else 0.0
    avg_loss = float(-losses.mean()) if len(losses) else 0.0
    expectancy_alt = wr * avg_win - lr * avg_loss
    # Identity check: the textbook expectancy form must equal the plain mean
    # whenever no trade is exactly flat (and it does even then, since a 0 R
    # trade contributes 0 to both sides).
    assert math.isclose(expectancy_alt, mean_r, rel_tol=1e-9, abs_tol=1e-9), (
        f"expectancy mismatch: {expectancy_alt} vs mean {mean_r}")
    out = {
        "n": n,
        "total_r": float(r.sum()),
        "mean_r": mean_r,
        "median_r": float(r.median()),
        "wr": wr * 100.0,
        "pf": (gross_win / gross_loss) if gross_loss > 0 else float("inf"),
        "max_dd": max_dd,
        "sharpe": (mean_r / std) if std > 0 else 0.0,
        "expectancy": mean_r,
        "expectancy_alt": expectancy_alt,
        "recovery": (float(r.sum()) / max_dd) if max_dd > 0 else float("inf"),
        "avg_bars": float("nan"),
        "avg_mfe": float("nan"),
        "avg_mae": float("nan"),
    }
    if df is not None:
        for key, col in (("avg_bars", "bars_held"), ("avg_mfe", "mfe_r"),
                         ("avg_mae", "mae_r")):
            if col in df.columns:
                out[key] = float(pd.to_numeric(df[col],
                                               errors="coerce").mean())
    return out


def _pf(r: pd.Series) -> float:
    """Profit factor, hour_analysis-compatible (loss side is r <= 0)."""
    pos = r[r > 0].sum()
    neg = -r[r <= 0].sum()
    return float(pos / max(neg, 1e-9))


def group_table(df: pd.DataFrame, by, min_n: int = 10) -> pd.DataFrame:
    """Per-group P&L with a half-1 / half-2 sign-stability flag."""
    keys = [by] if isinstance(by, str) else list(by)
    rows = []
    for key, sub in df.groupby(keys, dropna=False, sort=True):
        r = sub["r_multiple"]
        h1, h2 = sub[sub["half"] == 1]["r_multiple"], sub[sub["half"] == 2]["r_multiple"]
        m1 = float(h1.mean()) if len(h1) else float("nan")
        m2 = float(h2.mean()) if len(h2) else float("nan")
        stable = bool(len(h1) >= min_n and len(h2) >= min_n
                      and np.sign(m1) != 0 and np.sign(m1) == np.sign(m2))
        rec = {}
        # pandas >= 2.2 hands back a 1-tuple even for a single group key.
        vals = key if isinstance(key, tuple) else (key,)
        for k, v in zip(keys, vals):
            rec[k] = v
        rec.update({
            "n": int(len(sub)),
            "total_r": round(float(r.sum()), 2),
            "mean_r": round(float(r.mean()), 3),
            "wr": round(float((r > 0).mean() * 100), 2),
            "pf": round(_pf(r), 2),
            "n_h1": int(len(h1)),
            "total_r_h1": round(float(h1.sum()), 2),
            "mean_r_h1": round(m1, 3) if len(h1) else float("nan"),
            "n_h2": int(len(h2)),
            "total_r_h2": round(float(h2.sum()), 2),
            "mean_r_h2": round(m2, 3) if len(h2) else float("nan"),
            "sign_stable": stable,
        })
        rows.append(rec)
    out = pd.DataFrame(rows)
    if not out.empty:
        out = out.set_index(keys)
    return out


def streaks(df: pd.DataFrame) -> dict:
    """Longest win/loss runs and the longest stretch under water (in trades)."""
    d = df.sort_values("entry_time", kind="mergesort")
    r = d["r_multiple"].to_numpy(dtype="float64")
    best_w = best_l = cur_w = cur_l = 0
    for x in r:
        if x > 0:
            cur_w, cur_l = cur_w + 1, 0
        elif x < 0:
            cur_l, cur_w = cur_l + 1, 0
        else:
            cur_w = cur_l = 0
        best_w, best_l = max(best_w, cur_w), max(best_l, cur_l)
    cum = np.concatenate(([0.0], np.cumsum(r)))
    peak = np.maximum.accumulate(cum)
    flat, longest = 0, 0
    for i in range(1, len(cum)):
        flat = 0 if cum[i] >= peak[i] else flat + 1
        longest = max(longest, flat)
    return {"max_win_streak": best_w, "max_loss_streak": best_l,
            "longest_flat_trades": longest}


def rolling_wr(df: pd.DataFrame, window: int = 20) -> pd.Series:
    d = df.sort_values("entry_time", kind="mergesort")
    s = (d["r_multiple"] > 0).astype(float).rolling(window).mean() * 100
    s.index = d["entry_time"].to_numpy()
    return s


def equity_curve(df: pd.DataFrame) -> pd.DataFrame:
    d = df.sort_values("entry_time", kind="mergesort")
    cum = d["r_multiple"].cumsum()
    peak = pd.concat([pd.Series([0.0]), cum]).cummax().iloc[1:]
    peak.index = cum.index
    return pd.DataFrame({"entry_time": d["entry_time"].to_numpy(),
                         "cum_r": cum.to_numpy(),
                         "drawdown": (peak - cum).to_numpy()})


# ─────────────────────── quantile bucket screen ─────────────────────

def quantile_buckets(df: pd.DataFrame, col: str, q: int = 5,
                     min_n: int = 10) -> pd.DataFrame:
    """Equal-frequency buckets of `col` with per-half mean R.

    Bin edges are fit on the WHOLE book (both halves share edges).
    Result carries gap_h1 / gap_h2 / sign_stable in `.attrs` and as
    repeated columns so the CSV is self-describing.
    """
    v = pd.to_numeric(df.get(col), errors="coerce")
    d = df.loc[v.notna()].copy()
    d["_v"] = v.loc[v.notna()]
    empty = pd.DataFrame(columns=["bin", "lo", "hi", "n", "mean_r", "pf", "wr",
                                  "n_h1", "mean_r_h1", "n_h2", "mean_r_h2",
                                  "gap_h1", "gap_h2", "sign_stable"])
    empty.attrs = {"feature": col, "gap_h1": float("nan"),
                   "gap_h2": float("nan"), "sign_stable": False, "n": 0,
                   "mode": "empty"}
    if d.empty:
        return empty

    nuniq = int(d["_v"].nunique())
    if nuniq < 3:
        mode = "values"
        d["_bin"] = d["_v"]
        order = sorted(d["_v"].unique())
        edges = {b: (float(b), float(b)) for b in order}
    else:
        mode = "qcut"
        binned = pd.qcut(d["_v"], q, duplicates="drop")
        d["_bin"] = binned
        order = list(binned.cat.categories)
        edges = {b: (float(b.left), float(b.right)) for b in order}

    rows = []
    for b in order:
        sub = d[d["_bin"] == b]
        if sub.empty:
            continue
        r = sub["r_multiple"]
        h1 = sub[sub["half"] == 1]["r_multiple"]
        h2 = sub[sub["half"] == 2]["r_multiple"]
        lo, hi = edges[b]
        rows.append({
            "bin": str(b), "lo": round(lo, 4), "hi": round(hi, 4),
            "n": int(len(sub)), "mean_r": round(float(r.mean()), 3),
            "pf": round(_pf(r), 2),
            "wr": round(float((r > 0).mean() * 100), 2),
            "n_h1": int(len(h1)),
            "mean_r_h1": round(float(h1.mean()), 3) if len(h1) else float("nan"),
            "n_h2": int(len(h2)),
            "mean_r_h2": round(float(h2.mean()), 3) if len(h2) else float("nan"),
        })
    out = pd.DataFrame(rows)
    if out.empty:
        return empty

    bot, top = out.iloc[0], out.iloc[-1]
    gap_h1 = float(top["mean_r_h1"] - bot["mean_r_h1"])
    gap_h2 = float(top["mean_r_h2"] - bot["mean_r_h2"])
    populated = min(bot["n_h1"], bot["n_h2"], top["n_h1"], top["n_h2"]) >= min_n
    signs_ok = (not math.isnan(gap_h1) and not math.isnan(gap_h2)
                and np.sign(gap_h1) != 0
                and np.sign(gap_h1) == np.sign(gap_h2))
    sign_stable = bool(populated and signs_ok and len(out) >= 2)
    out["gap_h1"] = round(gap_h1, 3) if not math.isnan(gap_h1) else float("nan")
    out["gap_h2"] = round(gap_h2, 3) if not math.isnan(gap_h2) else float("nan")
    out["sign_stable"] = sign_stable
    out.attrs = {"feature": col, "gap_h1": gap_h1, "gap_h2": gap_h2,
                 "sign_stable": sign_stable, "n": int(len(d)), "mode": mode}
    return out


def feature_screen(df: pd.DataFrame, q: int = 5, min_n: int = 10,
                   columns: Iterable[str] | None = None) -> pd.DataFrame:
    """Bucket screen over every decision-time numeric feature in the book."""
    cols = list(columns) if columns is not None else feature_columns(df)
    rows = []
    for c in cols:
        bk = quantile_buckets(df, c, q=q, min_n=min_n)
        a = bk.attrs
        stable = bool(a.get("sign_stable"))
        g1, g2 = a.get("gap_h1", float("nan")), a.get("gap_h2", float("nan"))
        min_abs = (min(abs(g1), abs(g2)) if stable else 0.0)
        rows.append({"feature": c,
                     "gap_h1": round(g1, 3) if g1 == g1 else float("nan"),
                     "gap_h2": round(g2, 3) if g2 == g2 else float("nan"),
                     "min_abs_gap": round(float(min_abs), 3),
                     "sign_stable": stable,
                     "n": int(a.get("n", 0)),
                     "bins": int(len(bk))})
    out = pd.DataFrame(rows)
    if out.empty:
        return out
    return out.sort_values(["min_abs_gap", "n"],
                           ascending=[False, False]).reset_index(drop=True)


# ────────────────────── higher-timeframe extension ──────────────────

def wilder_atr(bars: pd.DataFrame, period: int = 14) -> pd.Series:
    h, l, c = bars["High"], bars["Low"], bars["Close"]
    tr = pd.concat([h - l, (h - c.shift()).abs(), (l - c.shift()).abs()],
                   axis=1).max(axis=1)
    return tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()


def _ext_series(parquet: str, minutes: int) -> pd.DataFrame:
    bars = pd.read_parquet(parquet).sort_index()
    bars = bars[~bars.index.duplicated(keep="last")]
    ext = ((bars["Close"] - bars["Close"].rolling(20).mean())
           / wilder_atr(bars, 14))
    out = pd.DataFrame({
        "close_time": bars.index + pd.Timedelta(minutes=minutes),
        "ext": ext.to_numpy(),
    }).dropna().reset_index(drop=True)
    # Books parse to datetime64[us, UTC] (format="mixed"), parquet indexes are
    # [ns, UTC]; merge_asof refuses mismatched resolutions.
    out["close_time"] = out["close_time"].astype("datetime64[ns, UTC]")
    return out


def add_htf_extension(df: pd.DataFrame, parquet_4h: str | None,
                      parquet_1d: str | None) -> pd.DataFrame:
    """Add direction-signed H4 / D1 stretch-from-mean in ATR units.

    ext = (close − 20-bar SMA of close) / Wilder ATR(14) on that timeframe,
    taken from the LAST higher-TF bar whose CLOSE time (bar label + one bar)
    is <= decision_time — i.e. only information the live scan could have had.
    Signed +1 for BUY, −1 for SELL, so positive always means "extended in the
    direction of the trade".
    """
    out = df.copy()
    sign = np.where(out["direction"].astype(str).str.upper().str.startswith("B"),
                    1.0, -1.0)
    for col, parquet, minutes in (("x_h4_ext_atr", parquet_4h, 240),
                                  ("x_d1_ext_atr", parquet_1d, 1440)):
        if not parquet:
            continue
        right = _ext_series(parquet, minutes)
        left = out[["decision_time"]].reset_index()
        left["decision_time"] = left["decision_time"].astype(
            "datetime64[ns, UTC]")
        left = left.sort_values("decision_time", kind="mergesort")
        merged = pd.merge_asof(left, right, left_on="decision_time",
                               right_on="close_time", direction="backward")
        merged["_stale_h"] = ((merged["decision_time"] - merged["close_time"])
                              .dt.total_seconds() / 3600.0)
        merged = merged.set_index("index").reindex(out.index)
        out[col] = merged["ext"].to_numpy() * sign
        out.attrs[f"{col}_max_stale_h"] = float(
            merged["_stale_h"].max()) if len(merged) else float("nan")
    return out


# ──────────────────────────── reporting ─────────────────────────────

def _fmt(v, nd=3):
    if isinstance(v, float):
        if math.isinf(v):
            return "inf"
        if math.isnan(v):
            return "n/a"
        return f"{v:.{nd}f}"
    return str(v)


def _metrics_block(name: str, m: dict) -> list[str]:
    return [
        f"  {name:<10} n={m['n']:<5} totalR={_fmt(m['total_r'], 2):>8}  "
        f"meanR={_fmt(m['mean_r'])}  medR={_fmt(m['median_r'])}  "
        f"WR={_fmt(m['wr'], 2)}%  PF={_fmt(m['pf'], 2)}",
        f"  {'':<10} maxDD={_fmt(m['max_dd'], 2)}  recovery={_fmt(m['recovery'], 2)}"
        f"  sharpe={_fmt(m['sharpe'])}  expectancy={_fmt(m['expectancy'])}"
        f" (wr·avgW − lr·avgL = {_fmt(m['expectancy_alt'])})",
        f"  {'':<10} avg bars={_fmt(m['avg_bars'], 1)}  "
        f"avg MFE={_fmt(m['avg_mfe'])}R  avg MAE={_fmt(m['avg_mae'])}R",
    ]


def build_report(df: pd.DataFrame, book: str, bar_minutes: int, split: str,
                 out_dir: str, q: int = 5, min_n: int = 10) -> str:
    lines: list[str] = []
    add = lines.append
    cut = split_instant(df["entry_time"], split)
    add("=" * 78)
    add(f"BOOK REPORT  {os.path.basename(book)}")
    add("=" * 78)
    add(f"trades           : {len(df)}")
    add(f"entry window     : {df['entry_time'].min()}  ->  {df['entry_time'].max()}")
    add(f"bar minutes      : {bar_minutes}  (decision_time = entry_time − 1 bar)")
    add(f"split rule       : {split}  ->  cut at {cut}  "
        f"(half 1 = before, half 2 = at/after)")
    add(f"half sizes       : h1={int((df['half'] == 1).sum())}  "
        f"h2={int((df['half'] == 2).sum())}")
    add("session map      : asian 3-6, london 7-12, overlap 13-16, ny 17-22, "
        "else off (on DECISION hour, UTC)")
    add("bucket edges     : quantile bins are fit on the WHOLE book; the two "
        "halves share the same edges.")
    feats = feature_columns(df)
    add(f"screen features  : {len(feats)}  "
        f"({sum(1 for c in feats if c.startswith(('f_', 'x_')))} f_/x_ columns)")
    for k, v in df.attrs.items():
        add(f"htf staleness    : {k} = {_fmt(v, 1)} h (max age of the matched bar)")

    add("")
    add("── HEADLINE METRICS ──")
    lines.extend(_metrics_block("full", base_metrics(df["r_multiple"], df)))
    for h in (1, 2):
        sub = df[df["half"] == h]
        lines.extend(_metrics_block(f"half {h}",
                                    base_metrics(sub["r_multiple"], sub)))

    group_specs = [
        ("weekday", "weekday"), ("decision_hour", "decision_hour"),
        ("session", "session"), ("quality", "quality"), ("regime", "regime"),
        ("vol_regime", "vol_regime"), ("direction", "direction"),
        ("strategy", "strategy"), ("exit_reason", "exit_reason"),
        ("indicators_agree", "indicators_agree"),
    ]
    tables = {}
    for name, col in group_specs:
        if col not in df.columns:
            continue
        t = group_table(df, col, min_n=min_n)
        tables[name] = t
        add("")
        add(f"── BY {name.upper()} ──")
        add(t.to_string() if not t.empty else "  (empty)")

    # (weekday, decision_hour) pivots
    add("")
    add("── weekday x decision_hour : TOTAL R ──")
    wh = group_table(df, ["weekday", "decision_hour"], min_n=min_n)
    if not wh.empty:
        piv_r = wh["total_r"].unstack("decision_hour")
        piv_n = wh["n"].unstack("decision_hour")
        add(piv_r.fillna(0).round(2).to_string())
        add("")
        add("── weekday x decision_hour : N ──")
        add(piv_n.fillna(0).astype(int).to_string())
        tables["weekday_hour"] = wh
        piv_r.to_csv(os.path.join(out_dir, "groups_weekday_hour_total_r.csv"))
        piv_n.to_csv(os.path.join(out_dir, "groups_weekday_hour_n.csv"))

    st = streaks(df)
    add("")
    add("── STREAKS ──")
    add(f"  max consecutive wins   : {st['max_win_streak']}")
    add(f"  max consecutive losses : {st['max_loss_streak']}")
    add(f"  longest flat stretch   : {st['longest_flat_trades']} trades "
        f"between equity highs")

    rw = rolling_wr(df, 20).dropna()
    add("")
    add("── ROLLING WIN RATE (20 trades) ──")
    if len(rw):
        add(f"  min={_fmt(rw.min(), 2)}%  median={_fmt(rw.median(), 2)}%  "
            f"max={_fmt(rw.max(), 2)}%  n={len(rw)}")
        add("  last 5: " + ", ".join(
            f"{ts:%Y-%m-%d %H:%M}={v:.2f}%" for ts, v in rw.tail(5).items()))
    else:
        add("  (fewer than 20 trades)")

    scr = feature_screen(df, q=q, min_n=min_n)
    add("")
    add("── FEATURE SCREEN (top 25 by min |gap| across halves) ──")
    add("  gap = mean R(top bucket) − mean R(bottom bucket), per half; "
        "min_abs_gap is 0 unless the sign is stable.")
    add(scr.head(25).to_string(index=False) if not scr.empty else "  (none)")

    stable = scr[scr["sign_stable"]].head(10) if not scr.empty else scr
    add("")
    add(f"── BUCKET TABLES FOR THE TOP {len(stable)} SIGN-STABLE FEATURES ──")
    if stable.empty:
        add("  (no feature has a sign-stable top-vs-bottom gap)")
    for feat in stable["feature"]:
        bk = quantile_buckets(df, feat, q=q, min_n=min_n)
        add("")
        add(f"  · {feat}  (mode={bk.attrs['mode']}, n={bk.attrs['n']}, "
            f"gap_h1={_fmt(bk.attrs['gap_h1'])}, "
            f"gap_h2={_fmt(bk.attrs['gap_h2'])})")
        add(bk.to_string(index=False))
        bk.to_csv(os.path.join(out_dir, f"buckets_{feat}.csv"), index=False)

    # CSV artefacts
    for name, t in tables.items():
        t.to_csv(os.path.join(out_dir, f"groups_{name}.csv"))
    scr.to_csv(os.path.join(out_dir, "feature_screen.csv"), index=False)
    equity_curve(df).to_csv(os.path.join(out_dir, "equity.csv"), index=False)
    all_buckets = []
    for c in feats:
        bk = quantile_buckets(df, c, q=q, min_n=min_n)
        if bk.empty:
            continue
        bk = bk.copy()
        bk.insert(0, "feature", c)
        all_buckets.append(bk)
    if all_buckets:
        pd.concat(all_buckets, ignore_index=True).to_csv(
            os.path.join(out_dir, "buckets.csv"), index=False)
    return "\n".join(lines)


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--book", required=True)
    ap.add_argument("--bar-minutes", type=int, required=True)
    ap.add_argument("--split", default="median",
                    help="'median' (default) or YYYY-MM-DD")
    ap.add_argument("--parquet-4h", default=None)
    ap.add_argument("--parquet-1d", default=None)
    ap.add_argument("--q", type=int, default=5)
    ap.add_argument("--min-n", type=int, default=10)
    ap.add_argument("--out", default="backtest/research/book_report_out")
    args = ap.parse_args()

    os.makedirs(args.out, exist_ok=True)
    df = load_book(args.book, args.bar_minutes, args.split)
    if args.parquet_4h or args.parquet_1d:
        df = add_htf_extension(df, args.parquet_4h, args.parquet_1d)
    text = build_report(df, args.book, args.bar_minutes, args.split,
                        args.out, q=args.q, min_n=args.min_n)
    print(text)
    with open(os.path.join(args.out, "report.txt"), "w") as fh:
        fh.write(text + "\n")
    print(f"\n[written] {os.path.join(args.out, 'report.txt')}")


if __name__ == "__main__":
    main()