ForexBot2/scripts/build_panel_base_rates.py

#!/usr/bin/env python3
"""Build agent_panel_base_rates.json from the reference backtest books.

The agent panel (agent_panel.py, prompt version v2) anchors every
persona's probability to how trades of the same class fared in the
validated books: (timeframe, strategy, grade, direction) with fallbacks
to (timeframe, direction) and (timeframe). This script turns the exported
63-column book CSVs into that table. Research-side (pandas only); the
live module reads the JSON and never imports this.

Usage (the section-9 harness books, regenerated from the kept cache):

    venv/bin/python scripts/build_panel_base_rates.py \\
        --book 5min=/path/m5_w2.csv --book 15min=/path/m15_w2.csv \\
        --book 1h=/path/h1_w3.csv --book 4h=/path/h4_w3.csv \\
        --out agent_panel_base_rates.json

Statistics per class: n, win_rate (percent of trades with r_multiple >
0), mean_r, tp1_reach_rate (percent whose best excursion reached the
planned first target: mfe_r >= rr_planned), median_bars_held. Every
class with n >= 1 is written; agent_panel._base_rate applies the
minimum-n rule (30) and the fallback order at read time.

Re-run whenever a reference book is re-pinned (CLAUDE.md section 9); the
JSON records each book's file name, MD5 and trade count so the table's
provenance is auditable.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import sys
from datetime import datetime, timezone
from typing import Dict, List

import pandas as pd

STATS_COLS = ("r_multiple", "mfe_r", "rr_planned", "bars_held",
              "strategy", "quality", "direction")


def _md5(path: str) -> str:
    h = hashlib.md5()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def _stats(df: pd.DataFrame, book: str) -> Dict[str, object]:
    r = df["r_multiple"].astype(float)
    reached = (df["mfe_r"].astype(float) >= df["rr_planned"].astype(float))
    return {
        "n": int(len(df)),
        "win_rate": round(float((r > 0).mean() * 100.0), 2),
        "mean_r": round(float(r.mean()), 4),
        "tp1_reach_rate": round(float(reached.mean() * 100.0), 2),
        "median_bars_held": int(round(float(df["bars_held"].median()))),
        "book": book,
    }


def build(books: Dict[str, str]) -> Dict[str, object]:
    classes: Dict[str, Dict[str, object]] = {}
    meta: Dict[str, Dict[str, object]] = {}
    for tf, path in books.items():
        df = pd.read_csv(path)
        missing = [c for c in STATS_COLS if c not in df.columns]
        if missing:
            raise SystemExit(f"{path}: missing columns {missing}")
        book = path.rsplit("/", 1)[-1]
        meta[tf] = {"file": book, "md5": _md5(path), "trades": int(len(df)),
                    "first_entry": str(df["entry_time"].min()),
                    "last_entry": str(df["entry_time"].max())}
        if df.empty:
            continue
        classes[f"{tf}|*|*|*"] = _stats(df, book)
        for direction, g in df.groupby("direction"):
            classes[f"{tf}|*|*|{direction}"] = _stats(g, book)
        for (strategy, grade, direction), g in df.groupby(
                ["strategy", "quality", "direction"]):
            classes[f"{tf}|{strategy}|{grade}|{direction}"] = _stats(g, book)
    return {
        "schema": "agent-panel-base-rates-v1",
        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "key": "timeframe|strategy|grade|direction ('*' = any)",
        "min_n_for_exact_class": 30,
        "books": meta,
        "classes": dict(sorted(classes.items())),
    }


def main(argv: List[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--book", action="append", default=[], metavar="TF=CSV",
                    help="timeframe (5min/15min/1h/4h) = exported book CSV")
    ap.add_argument("--out", default="agent_panel_base_rates.json")
    args = ap.parse_args(argv)
    books: Dict[str, str] = {}
    for item in args.book:
        if "=" not in item:
            ap.error(f"--book expects TF=CSV, got {item!r}")
        tf, path = item.split("=", 1)
        books[tf.strip()] = path.strip()
    if not books:
        ap.error("at least one --book is required")
    table = build(books)
    with open(args.out, "w", encoding="utf-8") as fh:
        json.dump(table, fh, indent=1, sort_keys=False)
        fh.write("\n")
    n_exact = sum(1 for k in table["classes"] if "|*|" not in k)
    print(f"wrote {args.out}: {len(table['classes'])} classes "
          f"({n_exact} exact) from {len(books)} book(s)")
    for tf, m in table["books"].items():
        print(f"  {tf:6s} {m['file']}  md5={m['md5']}  trades={m['trades']}")
    return 0


if __name__ == "__main__":
    sys.exit(main())