mymt5opp/scripts/diag_mt5_trades.py

"""Parse the MT5 IS HTML report and dump the first N deal rows so we can
compare per-trade lots/entry/exit against Python's diag_size_after_warmup.py
output.

The report's "Deals" table rows have ~13-14 cells (Time, Deal, Symbol, Type,
Direction, Volume, Price, Order, Commission, Fee, Swap, Profit, Balance,
Comment). We extract rows whose Type is "buy" or "sell" (entry) and "in" /
"out" (Direction) to reconstruct trade pairs.

Usage:
    python scripts/diag_mt5_trades.py 10
"""
from __future__ import annotations
import sys
from pathlib import Path
PROJECT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT))

from shared.data.mt5_report import parse_mt5_report  # noqa: E402

PATH = PROJECT / "reports" / "IS-ReportTester-52845377.html"


def main() -> int:
    n = int(sys.argv[1]) if len(sys.argv) > 1 else 10
    raw = PATH.read_bytes()
    if raw[:2] in (b"\xff\xfe", b"\xfe\xff"):
        text = raw.decode("utf-16")
    else:
        text = raw.decode("utf-8", errors="replace")

    try:
        from lxml import html
        tree = html.fromstring(text)
    except Exception:
        import html5lib
        tree = html5lib.parse(text)

    # MT5 reports have multiple tables; the deals table is the last big one.
    # Each <tr> is a deal. Header row has "Time / Deal / Symbol / Type / ...
    rows = tree.iter("tr")
    deals = []
    headers_seen = False
    for row in rows:
        cells = row.findall("td") or row.findall("th")
        if not cells:
            continue
        texts = [c.text_content().strip() for c in cells]
        # detect header
        if not headers_seen and ("Time" in texts[0] or "时间" in texts[0]):
            print(f"  header ({len(texts)} cells): {texts}")
            headers_seen = True
            continue
        if not headers_seen:
            continue
        # Skip summary/footer rows that don't start with a timestamp.
        first = texts[0]
        if not first or not any(c.isdigit() for c in first[:4]):
            continue
        if len(texts) < 8:
            continue
        deals.append(texts)

    print(f"\n  parsed {len(deals)} deal rows from {PATH.name}")
    print(f"\n  first {n} deals:")
    print(f"  {'#':>3} {'time':<20} {'deal':>8} {'type':<6} {'dir':<4} {'volume':>8} {'price':>10} {'profit':>10} {'balance':>10}")
    for i, d in enumerate(deals[:n], 1):
        # Layout (typical): [Time, Deal, Symbol, Type, Direction, Volume, Price,
        #                    Order, Commission, Fee, Swap, Profit, Balance, Comment]
        time_s  = d[0]
        deal_s  = d[1] if len(d) > 1 else ""
        sym_s   = d[2] if len(d) > 2 else ""
        type_s  = d[3] if len(d) > 3 else ""
        dir_s   = d[4] if len(d) > 4 else ""
        vol_s   = d[5] if len(d) > 5 else ""
        price_s = d[6] if len(d) > 6 else ""
        # profit/balance positions vary; print last few cells
        profit_s = d[-3] if len(d) >= 3 else ""
        balance_s = d[-2] if len(d) >= 2 else ""
        print(f"  {i:>3} {time_s:<20} {deal_s:>8} {type_s:<6} {dir_s:<4} "
              f"{vol_s:>8} {price_s:>10} {profit_s:>10} {balance_s:>10}")

    # Also dump the full cell layout of the first deal for verification.
    if deals:
        print(f"\n  first deal full layout ({len(deals[0])} cells):")
        for i, c in enumerate(deals[0]):
            print(f"    [{i:>2}] {c!r}")

    # Try to pair entry/exit deals to reconstruct trades.
    # An "in" deal (Direction="in") opens a position; an "out" deal closes it.
    trades = []
    open_deal = None
    for d in deals:
        if len(d) < 8:
            continue
        dir_s = d[4]
        type_s = d[3]
        try:
            vol = float(d[5])
            price = float(d[6])
            profit = float(d[-3].split()[0]) if d[-3] else 0.0
        except (ValueError, IndexError):
            continue
        if dir_s == "in":
            open_deal = {"time": d[0], "type": type_s, "vol": vol, "price": price}
        elif dir_s == "out" and open_deal is not None:
            trades.append({
                "entry_time": open_deal["time"],
                "dir": open_deal["type"],
                "entry": open_deal["price"],
                "exit": price,
                "lots": open_deal["vol"],
                "pnl": profit,
            })
            open_deal = None

    if trades:
        print(f"\n  reconstructed {len(trades)} trade pairs (in→out)")
        print(f"\n  first {min(n, len(trades))} trades:")
        print(f"  {'#':>3} {'entry_time':<22} {'dir':<5} {'entry':>10} {'exit':>10} {'lots':>8} {'pnl':>10}")
        for i, t in enumerate(trades[:n], 1):
            print(f"  {i:>3} {t['entry_time']:<22} {t['dir']:<5} "
                  f"{t['entry']:>10.2f} {t['exit']:>10.2f} {t['lots']:>8.4f} {t['pnl']:>10.2f}")

        import numpy as np
        pnls = np.array([t["pnl"] for t in trades])
        wins = (pnls > 0).sum()
        print(f"\n  PnL stats  : trades={len(trades)}  wins={wins} ({wins/len(trades):.1%})  "
              f"sum=${pnls.sum():.2f}  avg=${pnls.mean():.2f}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())