btc-5m-edge-study/ml/xgb_eval.py

#!/usr/bin/env python3
"""Offline XGBoost inspection — load the REAL trained models and report them.

    python ml/xgb_eval.py

Needs only `xgboost` + `numpy` + the committed model files under models/. No live
feeds, no keys, no 16 GB of ticks. Prints each model's feature-importance ranking,
and — if a held-out sample is present at ml/data/lock_sample.csv — the test AUC.

The point this makes, straight from the model: the strongest features are the
market maker's OWN quote (ask, ask_margin, ask_vel), not BTC's price dynamics.
The model taught itself that the best estimate of the outcome is what the maker is
already charging — which is exactly why the edge isn't capturable. And the AUC is
deliberately weak (~coin-flip-plus): that weakness IS the finding. See the README.
"""

from __future__ import annotations

import json
import os

import xgboost as xgb

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)


def _importances(name: str, model_file: str, feat_file: str, top: int = 8) -> None:
    # Native Booster API — no sklearn needed (XGBClassifier would require it).
    try:
        booster = xgb.Booster()
        booster.load_model(os.path.join(ROOT, "models", model_file))
        with open(os.path.join(ROOT, "models", feat_file)) as f:
            names = json.load(f)
    except Exception as e:
        print(f"  [{name}] could not load ({e})")
        return
    gain = booster.get_score(importance_type="gain")   # {feature: score}, unused feats absent
    # if the model stored generic f0/f1 names, remap them onto the real feature list
    if gain and all(k[1:].isdigit() for k in gain if k.startswith("f")):
        gain = {names[int(k[1:])]: v for k, v in gain.items()
                if k.startswith("f") and int(k[1:]) < len(names)} or gain
    total = sum(gain.values()) or 1.0
    ranked = sorted(((n, gain.get(n, 0.0) / total) for n in names), key=lambda x: -x[1])
    print(f"\n{name}  ({model_file}, {len(names)} features)")
    top1 = ranked[0][1] or 1.0
    for feat, w in ranked[:top]:
        bar = "█" * int(round(w / top1 * 28)) if w > 0 else "·"
        print(f"    {feat:<14} {w:6.3f}  {bar}")


def _economics_from_log() -> None:
    """The 'high win-rate trap', straight from real paper trades.

    A trade filled at ask A pays (1 - A) on a win and loses A on a loss. Paying up
    for a high win rate hands the payoff to the maker — high WR, negative PnL.
    """
    path = os.path.join(HERE, "data", "paper_trades_sample.jsonl")
    if not os.path.exists(path):
        return
    rows = []
    with open(path) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            d = json.loads(line)
            if d.get("ask") is not None and d.get("won") is not None and d.get("pnl_5sh") is not None:
                rows.append(d)
    if not rows:
        return
    wr_all = sum(r["won"] for r in rows) / len(rows)
    pnl_all = sum(r["pnl_5sh"] for r in rows) / len(rows)
    print("\n" + "=" * 68)
    print(f"Real paper trades — no price bucket is profitable ({len(rows)} trades, ml/data/)")
    print("=" * 68)
    print("  fill at ask A → a win pays (1-A), a loss costs A. Split by entry price:\n")
    print(f"    {'ask range':<11} {'n':>4} {'win rate':>9} {'mean PnL / 5sh':>15}")
    for lo, hi in [(0.0, 0.5), (0.5, 0.7), (0.7, 0.85), (0.85, 1.01)]:
        b = [r for r in rows if lo <= r["ask"] < hi]
        if not b:
            continue
        wr = sum(r["won"] for r in b) / len(b)
        pnl = sum(r["pnl_5sh"] for r in b) / len(b)
        print(f"    {lo:.2f}-{hi:<6.2f} {len(b):>4} {wr:>8.0%} {pnl:>+15.2f}")
    print(f"    {'ALL':<11} {len(rows):>4} {wr_all:>8.0%} {pnl_all:>+15.2f}")
    print("\n  Mean PnL is negative in every bucket — cheap or expensive, high win rate or")
    print("  low. Across this cross-section of 29 strategies there is no entry price that")
    print("  comes out ahead: the maker's spread plus execution costs take the rest.")


def main() -> None:
    print("=" * 68)
    print("XGBoost models — feature importances (what the model actually learned)")
    print("=" * 68)
    _importances("Lock model — will the current leader hold to settlement?",
                 "lock_xgb.json", "lock_xgb_features.json")
    _importances("Direction model (left=140s)", "direction_left140.json", "direction_features.json")
    _importances("Hold-7s model", "direction_hold7s.json", "hold7s_features.json")
    _economics_from_log()
    print("\n" + "-" * 68)
    print("Read the top features: `ask`, `ask_margin`, `ask_vel` are the maker's own")
    print("price. The best predictor of the outcome is what the maker already charges —")
    print("you can't beat a signal that IS the counterparty. See README, Pillar 2.")


if __name__ == "__main__":
    main()