Warrior_EA/research/read_book.py

"""
Read the Mind's trade book (Common\\Files\\Warrior_EA\\Mind\\book_<SYMBOL>_<TF>.csv) the way a
trader reads a journal after a run.

    python research/read_book.py [book.csv ...]          # default: every book in the Common folder

For each context field: Spearman rank correlation with R, and the mean R / win rate of the low, mid
and high thirds. With ~29 fields and a few hundred trades, ONE field clearing p < 0.05 is what
chance alone produces; the table prints the count expected by luck so a single "significant"
line is not mistaken for a finding. Fields are ranked by |rho|; nothing here selects a rule -
a rule found in this table is a hypothesis for the NEXT run, checked out of sample.
"""

from __future__ import annotations

import glob
import os
import sys

import numpy as np
import pandas as pd
from scipy.stats import spearmanr

COMMON = os.path.expandvars(r"%APPDATA%\MetaQuotes\Terminal\Common\Files\Warrior_EA\Mind")
NA = -999.0
BASE = {"position", "open_time", "close_time", "side", "entry", "exit", "sl", "lots", "risk_money",
        "net", "r", "mae_r", "mfe_r", "bars", "reason", "scale", "review_p", "review_n", "story"}


def load(paths):
    frames = []
    for p in paths:
        d = pd.read_csv(p)
        d["book"] = os.path.basename(p)
        frames.append(d)
    return pd.concat(frames, ignore_index=True)


def psr(r: np.ndarray) -> float:
    from math import erf, sqrt
    n = len(r)
    if n < 10 or r.std(ddof=1) == 0:
        return float("nan")
    sr = r.mean() / r.std(ddof=1)
    m = r - r.mean()
    skew = (m ** 3).mean() / (m ** 2).mean() ** 1.5
    kurt = (m ** 4).mean() / (m ** 2).mean() ** 2
    den = 1 - skew * sr + (kurt - 1) / 4 * sr ** 2
    z = sr * sqrt(n - 1) / sqrt(den)
    return 0.5 * (1 + erf(z / sqrt(2)))


def main():
    paths = sys.argv[1:] or sorted(glob.glob(os.path.join(COMMON, "book_*.csv")))
    if not paths:
        sys.exit(f"no books under {COMMON}")
    d = load(paths)
    r = d["r"].to_numpy()
    print(f"{len(d)} trades from {len(paths)} book(s): " + ", ".join(sorted(set(d.book))))
    print(f"mean R {r.mean():+.3f}  win {100*(r>0).mean():.1f}%  PSR(0) {psr(r):.3f}  "
          f"MAE mean {d.mae_r.mean():+.2f}R  MFE mean {d.mfe_r.mean():+.2f}R  "
          f"risk scale mean {d.scale.mean():.2f}")
    print(f"exit reasons: {d.reason.value_counts().to_dict()}\n")

    fields = [c for c in d.columns if c not in BASE and c != "book"]
    rows = []
    for f in fields:
        x = d[f].to_numpy(dtype=float)
        ok = x != NA
        if ok.sum() < 30 or np.nanstd(x[ok]) == 0:
            continue
        rho, p = spearmanr(x[ok], r[ok])
        lo, hi = np.percentile(x[ok], [33.3, 66.7])
        parts = []
        for name, m in (("lo", ok & (x <= lo)), ("mid", ok & (x > lo) & (x < hi)), ("hi", ok & (x >= hi))):
            parts.append((m.sum(), r[m].mean() if m.any() else np.nan, 100 * (r[m] > 0).mean() if m.any() else np.nan))
        rows.append((f, ok.sum(), rho, p, parts))
    rows.sort(key=lambda t: -abs(t[2]))
    print(f"{'field':16s} {'n':>5s} {'rho':>7s} {'p':>7s} | {'lo: n meanR win%':>22s} | {'mid':>22s} | {'hi':>22s}")
    for f, n, rho, p, parts in rows:
        cells = " | ".join(f"{a:4d} {b:+6.3f} {c:5.1f}" for a, b, c in parts)
        print(f"{f:16s} {n:5d} {rho:+7.3f} {p:7.3f} | {cells}")
    print(f"\n{len(rows)} fields tested: about {0.05*len(rows):.1f} would clear p<0.05 by luck alone. "
          f"Treat anything short of p < {0.05/max(len(rows),1):.4f} (Bonferroni) as a lead, not a result.")

    if "review_p" in d and d.review_p.notna().sum() >= 20:
        v = d.dropna(subset=["review_p"])
        rho, p = spearmanr(v.review_p, v.r)
        print(f"\nThe journal's own forecast (review_p) vs realised R on {len(v)} trades: rho {rho:+.3f}, p {p:.3f}"
              f"  - it must be positive out of sample before SIZE is worth switching on.")


if __name__ == "__main__":
    main()