mt5_report_parser/mt5_report_parser.py

# -*- coding: utf-8 -*-
"""
MT5 策略测试报告解析器（中文版 xlsx）
=====================================
读取 MT5 Strategy Tester 导出的 .xlsx 报告，结构化为：
  - meta     : 元信息（EA、品种、期间、经纪商、初始资金、杠杆、输入参数）
  - summary  : “结果”区的汇总指标字典
  - orders   : 订单明细 DataFrame
  - deals    : 成交明细 DataFrame
  - trades   : 由 in/out 成交对重建的“逐笔交易”DataFrame

设计为可复用：后续新的回测报告可直接喂给本解析器。
"""
from __future__ import annotations

import re
import os as _os_std
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
from html.parser import HTMLParser

import numpy as np
import openpyxl
import pandas as pd

# 报告中区段标题
SEC_SETTINGS = "设置"
SEC_RESULTS = "结果"
SEC_ORDERS = "订单"
SEC_DEALS = "成交"

# 支持的报告格式
_Supported_Extensions = {".xlsx", ".htm", ".html"}


class _MT5HTMLParser(HTMLParser):
    """MT5 HTML 报告解析器。"""
    def __init__(self):
        super().__init__()
        self.in_td = False
        self.in_th = False
        self.in_b = False
        self.current_row: List[Optional[str]] = []
        self.current_cell: str = ""
        self.b_content: str = ""  # Content inside <b> tags
        self.rows: List[List[Optional[str]]] = []
        self.section_rows: Dict[str, int] = {}  # Track section headers

    def handle_starttag(self, tag, attrs):
        if tag == "tr":
            if self.current_row:
                self.rows.append(self.current_row)
                self.current_row = []
        elif tag == "td":
            self.current_cell = ""
            self.b_content = ""
            self.in_td = True
            self.in_b = False
        elif tag == "th":
            self.current_cell = ""
            self.b_content = ""
            self.in_th = True
            self.in_b = False
        elif tag == "b":
            self.in_b = True
            self.b_content = ""

    def handle_endtag(self, tag):
        if tag == "tr":
            if self.current_row:
                self.rows.append(self.current_row)
                self.current_row = []
            self.current_cell = ""
            self.in_td = False
            self.in_th = False
        elif tag == "td":
            # Build full cell: non-b text + <b> content
            full_cell = self.current_cell + self.b_content
            self.current_row.append(full_cell.strip() if full_cell.strip() else "")
            self.current_cell = ""
            self.b_content = ""
            self.in_td = False
        elif tag == "th":
            full_cell = self.current_cell + self.b_content
            # Check if this is a section header
            b_val = self.b_content.strip()
            if b_val in (SEC_SETTINGS, SEC_RESULTS, SEC_ORDERS, SEC_DEALS):
                row_idx = len(self.rows)
                self.section_rows[b_val] = row_idx
            self.current_row.append(full_cell.strip() if full_cell.strip() else "")
            self.current_cell = ""
            self.b_content = ""
            self.in_th = False
        elif tag == "b":
            self.in_b = False

    def handle_data(self, data):
        if self.in_td or self.in_th:
            if not self.in_b:
                self.current_cell += data
            else:
                self.b_content += data

    def finish(self):
        if self.current_row:
            self.rows.append(self.current_row)
        # Also search all cells for section markers (they may be in <td> not just <th>)
        for i, row in enumerate(self.rows):
            if row:
                for j, cell in enumerate(row):
                    if cell:
                        b_val = _get_b_text(cell)
                        if b_val and b_val.strip() in (SEC_SETTINGS, SEC_RESULTS, SEC_ORDERS, SEC_DEALS):
                            sec = b_val.strip()
                            if sec not in self.section_rows:
                                self.section_rows[sec] = i
        return self.rows, self.section_rows


@dataclass
class MT5Report:
    meta: Dict[str, Any] = field(default_factory=dict)
    summary: Dict[str, Any] = field(default_factory=dict)
    orders: Optional[pd.DataFrame] = None
    deals: Optional[pd.DataFrame] = None
    trades: Optional[pd.DataFrame] = None
    source_file: str = ""


# --------------------------------------------------------------------------- #
# 工具函数
# --------------------------------------------------------------------------- #
def _find_section_rows(ws) -> Dict[str, int]:
    """扫描 A 列，定位各段标题所在行。"""
    rows: Dict[str, int] = {}
    for r in range(1, ws.max_row + 1):
        v = ws.cell(r, 1).value
        if v in (SEC_SETTINGS, SEC_RESULTS, SEC_ORDERS, SEC_DEALS):
            rows[v] = r
    return rows


def _kv_in_row(ws, r: int, max_col: int = 14) -> Dict[str, Any]:
    """
    按行解析 “标签: 值” 对。
    MT5 报告里标签以全角冒号 ':' 结尾，值位于其右侧下一个非空单元格。
    一行可能含多组 标签/值。
    """
    out: Dict[str, Any] = {}
    cells = [ws.cell(r, c).value for c in range(1, max_col + 1)]
    i = 0
    while i < len(cells):
        v = cells[i]
        if isinstance(v, str) and v.endswith(":"):
            label = v[:-1].strip()
            # 向右找第一个非空值
            j = i + 1
            while j < len(cells) and cells[j] is None:
                j += 1
            if j < len(cells):
                out[label] = cells[j]
                i = j + 1
                continue
        i += 1
    return out


def _parse_table(
    ws, header_row: int, end_row: int, max_col: int = 14
) -> pd.DataFrame:
    """从 header_row 读取列名，读取其后到 end_row 的数据，返回 DataFrame。"""
    headers = [ws.cell(header_row, c).value for c in range(1, max_col + 1)]
    # 去掉末尾连续的 None 列
    while headers and headers[-1] is None:
        headers.pop()
    ncol = len(headers)

    records: List[List[Any]] = []
    for r in range(header_row + 1, end_row + 1):
        row = [ws.cell(r, c).value for c in range(1, ncol + 1)]
        # 跳过整行空
        if all(v is None for v in row):
            continue
        records.append(row)

    df = pd.DataFrame(records, columns=headers)
    return df


def _to_float(x: Any) -> Optional[float]:
    """从形如 '971.54 (96.32%)' 或 '0.69415' 的字符串中提取首个数值。"""
    if x is None:
        return None
    if isinstance(x, (int, float)):
        return float(x)
    if isinstance(x, str):
        m = re.search(r"-?\d+\.?\d*", x)
        return float(m.group()) if m else None
    return None


def _pct_to_float(x: Any) -> Optional[float]:
    """从 '96.32%' 提取 96.32。"""
    if x is None:
        return None
    if isinstance(x, (int, float)):
        return float(x)
    m = re.search(r"(-?\d+\.?\d*)\s*%", str(x))
    return float(m.group(1)) if m else None


# --------------------------------------------------------------------------- #
# HTML 报告解析
# --------------------------------------------------------------------------- #
def _parse_html_report(text: str) -> MT5Report:
    """解析 MT5 HTML 报告。"""
    parser = _MT5HTMLParser()
    parser.feed(text)
    rows, section_rows = parser.finish()

    r_settings = section_rows.get(SEC_SETTINGS, 0)
    r_results = section_rows.get(SEC_RESULTS, r_settings)
    r_orders = section_rows.get(SEC_ORDERS, r_results)
    r_deals = section_rows.get(SEC_DEALS, r_orders)

    rep = MT5Report()
    rep.meta, rep.summary = _parse_html_meta_and_summary(rows, r_settings, r_results, r_orders)
    rep.orders, rep.deals = _parse_html_tables(rows, r_orders, r_deals, len(rows))

    rep.trades = _reconstruct_trades(rep.deals)
    _normalize_summary_numbers(rep)

    return rep


def _read_html_file(path: str) -> str:
    """读取 HTML 文件，自动检测编码。"""
    with open(path, "rb") as f:
        raw = f.read()

    # 检测 BOM
    if raw[:2] == b"\xff\xfe":
        return raw.decode("utf-16-le")
    elif raw[:2] == b"\xfe\xff":
        return raw.decode("utf-16-be")
    elif raw[:3] == b"\xef\xbb\xbf":
        return raw[3:].decode("utf-8")
    elif raw[:4] == b"\xff\xfe\x00\x00":
        return raw[4:].decode("utf-32-le")

    # 尝试 UTF-8
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        pass

    # 回退到 latin-1
    return raw.decode("latin-1")


def _get_b_text(cell_text: str) -> Optional[str]:
    """从 HTML 单元格文本中提取 <b>...</b> 内的文本。"""
    if cell_text is None or not cell_text.strip():
        return None
    # 直接提取 <b>...</b> 内容
    m = re.search(r'<b>(.*?)</b>', cell_text, re.DOTALL)
    if m:
        text = m.group(1).strip()
        # 去除内嵌的 <br> 标签
        text = re.sub(r'<br\s*/?>', '', text, flags=re.IGNORECASE)
        return text if text else None
    # 如果没有 <b> 标签，返回去除 HTML 标签的纯文本
    clean = re.sub(r'<[^>]+>', '', cell_text).strip()
    return clean if clean else None


def _parse_html_meta_and_summary(
    rows: List[List[Optional[str]]],
    r_settings: int, r_results: int, r_orders: int,
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
    """解析 HTML 报告中的设置和结果部分。"""
    meta: Dict[str, Any] = {}
    summary: Dict[str, Any] = {}

    # 解析设置段
    for i in range(r_settings, r_results):
        if i >= len(rows):
            break
        row = rows[i]
        if not row or len(row) < 2:
            continue
        label = _get_b_text(row[0] or "")
        if label and label.endswith(":"):
            label = label[:-1].strip()
        # 值在第二个非空单元格
        for j in range(1, len(row)):
            val = _get_b_text(row[j] or "")
            if val is not None:
                if label == "输入":
                    # 输入参数
                    continue
                meta[label] = val
                break

    # 解析输入参数（在设置段中查找 Key=Value 格式）
    inputs: Dict[str, str] = {}
    for i in range(r_settings, r_results):
        if i >= len(rows):
            break
        for j in range(len(rows[i])):
            cell = rows[i][j]
            if cell:
                b_val = _get_b_text(cell)
                if b_val and "=" in b_val and not b_val.strip().startswith("==="):
                    k, _, val = b_val.partition("=")
                    inputs[k.strip()] = val.strip()
    meta["inputs"] = inputs

    # 解析结果段
    for i in range(r_results, r_orders):
        if i >= len(rows):
            break
        row = rows[i]
        if not row or len(row) < 2:
            continue
        label = _get_b_text(row[0] or "")
        if label and label.endswith(":"):
            label = label[:-1].strip()
        for j in range(1, len(row)):
            val = _get_b_text(row[j] or "")
            if val is not None:
                summary[label] = val
                break

    return meta, summary


def _parse_html_tables(
    rows: List[List[Optional[str]]],
    r_orders: int, r_deals: int, max_row: int,
) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame]]:
    """解析 HTML 报告中的订单表和成交表。"""
    orders_df: Optional[pd.DataFrame] = None
    deals_df: Optional[pd.DataFrame] = None

    # HTML 报告列名到标准英文列名的映射
    DEALS_COL_MAP = {
        "时间": "time",
        "成交": "deal_id",
        "交易品种": "symbol",
        "类型": "type",
        "趋势": "entry",
        "交易量": "volume",
        "价位": "price",
        "订单": "order",
        "手续费": "commission",
        "库存费": "swap",
        "盈利": "profit",
        "结余": "balance",
        "注释": "comment",
    }
    ORDERS_COL_MAP = {
        "时间": "time",
        "交易": "ticket",
        "交易品种": "symbol",
        "类型": "type",
        "状态": "state",
        "交易量": "volume",
        "价位": "price",
        "止损": "sl",
        "止盈": "tp",
        "注释": "comment",
        "订单": "order",
    }

    def _parse_table_section(start_row: int, end_row: int, col_map: Dict[str, str]) -> Optional[pd.DataFrame]:
        """Parse a table section from rows."""
        if start_row >= max_row:
            return None

        # Find header row
        header_row = None
        for i in range(start_row + 1, min(start_row + 5, end_row)):
            if i >= len(rows):
                break
            row = rows[i]
            if row:
                b_count = sum(1 for cell in row if cell and _get_b_text(cell))
                if b_count >= 3:
                    header_row = i
                    break

        if header_row is None:
            return None

        # Extract headers
        headers = [_get_b_text(cell) for cell in rows[header_row] if cell and _get_b_text(cell)]
        if not headers:
            return None

        # Rename columns using col_map
        renamed_headers = [col_map.get(h, h) for h in headers]

        # Extract data rows
        data_rows = []
        for i in range(header_row + 1, end_row):
            if i >= len(rows):
                break
            row = rows[i]
            if row:
                values = [_get_b_text(cell) or "" for cell in row]
                if any(v for v in values):
                    while len(values) < len(renamed_headers):
                        values.append("")
                    data_rows.append(values[:len(renamed_headers)])

        if not data_rows:
            return None

        df = pd.DataFrame(data_rows, columns=renamed_headers)
        return df

    if r_orders < max_row:
        orders_df = _parse_table_section(r_orders, r_deals, ORDERS_COL_MAP)

    if r_deals < max_row:
        deals_df = _parse_table_section(r_deals, max_row, DEALS_COL_MAP)

    return orders_df, deals_df


# --------------------------------------------------------------------------- #
# 解析缓存（文件级 LRU 缓存）
# --------------------------------------------------------------------------- #
_parse_cache: Dict[str, MT5Report] = {}


def parse_report(path: str, use_cache: bool = True) -> MT5Report:
    """解析一份 MT5 报告（支持 xlsx / htm / html 格式）。"""
    real_path = _os_std.path.realpath(str(path))
    if use_cache and real_path in _parse_cache:
        return _parse_cache[real_path]

    ext = _os_std.path.splitext(real_path)[1].lower()
    if ext == ".xlsx":
        rep = _parse_xlsx_report(real_path)
    elif ext in (".htm", ".html"):
        text = _read_html_file(real_path)
        rep = _parse_html_report(text)
        rep.source_file = real_path
    else:
        raise ValueError(f"不支持的报告格式: {ext}（仅支持 .xlsx / .htm / .html）")

    _parse_cache[real_path] = rep
    return rep


def _parse_xlsx_report(path: str) -> MT5Report:
    """解析 xlsx 格式报告（原 parse_report 逻辑）。"""
    wb = openpyxl.load_workbook(path, data_only=True)
    ws = wb[wb.sheetnames[0]]
    rep = MT5Report(source_file=path)

    sec = _find_section_rows(ws)
    r_settings = sec.get(SEC_SETTINGS, 1)
    r_results = sec.get(SEC_RESULTS, r_settings)
    r_orders = sec.get(SEC_ORDERS, r_results)
    r_deals = sec.get(SEC_DEALS, r_orders)

    # ---- 元信息 / 设置 ----
    meta: Dict[str, Any] = {}
    for r in range(r_settings + 1, r_results):
        kv = _kv_in_row(ws, r)
        for k, v in kv.items():
            if k == "输入":  # "输入:" 是子段标题，其值为分节字符串，跳过
                continue
            meta[k] = v

    # 输入参数：在 "输入:" 段落里，列 D 形如 Key=Value
    inputs: Dict[str, str] = {}
    for r in range(r_settings + 1, r_results):
        for c in (4,):  # 经验上输入参数在 D 列
            v = ws.cell(r, c).value
            if isinstance(v, str) and "=" in v and not v.endswith(":"):
                # 跳过分节标题（如 "=== PAINEL ===="）
                if v.strip().startswith("==="):
                    continue
                k, _, val = v.partition("=")
                inputs[k.strip()] = val.strip()
    meta["inputs"] = inputs

    # ---- 结果汇总 ----
    summary: Dict[str, Any] = {}
    for r in range(r_results, r_orders):
        summary.update(_kv_in_row(ws, r))

    # ---- 订单表 ----
    rep.orders = _parse_table(ws, header_row=r_orders + 1, end_row=r_deals - 1)

    # ---- 成交表 ----
    rep.deals = _parse_table(ws, header_row=r_deals + 1, end_row=ws.max_row)

    rep.meta = meta
    rep.summary = summary
    rep.trades = _reconstruct_trades(rep.deals)
    _normalize_summary_numbers(rep)

    return rep


def clear_parse_cache() -> None:
    """清除解析缓存（调试 / 释放内存用）。"""
    _parse_cache.clear()


# --------------------------------------------------------------------------- #
# 逐笔交易重建
# --------------------------------------------------------------------------- #
DEALS_COLS = {
    "time": "时间",
    "deal_id": "成交",
    "symbol": "交易品种",
    "type": "类型",       # buy / sell / balance
    "entry": "趋势",      # in / out / None
    "volume": "交易量",
    "price": "价位",
    "order": "订单",
    "commission": "手续费",
    "swap": "库存费",
    "profit": "盈利",
    "balance": "结余",
    "comment": "注释",
}


def _reconstruct_trades(deals: pd.DataFrame) -> pd.DataFrame:
    """
    由成交明细重建逐笔交易。
    规则：每笔 'in' 成交后紧跟其 'out' 平仓成交，两两配对。
    返回字段：open_time, close_time, direction, volume, open_price,
             close_price, profit, swap, commission, net_profit, duration_min
    """
    if deals is None or deals.empty:
        return pd.DataFrame()

    d = deals.copy()
    # 统一列名
    d = d.rename(columns={v: k for k, v in DEALS_COLS.items() if v in d.columns})

    # 类型转换
    d["time"] = pd.to_datetime(d["time"], errors="coerce")
    for col in ("price", "commission", "swap", "profit", "balance"):
        if col in d.columns:
            d[col] = pd.to_numeric(d[col], errors="coerce")
    # 成交量形如 '0.01 / 0.01'，取前半
    if "volume" in d.columns:
        d["volume"] = d["volume"].astype(str).str.split("/").str[0].str.strip()
        d["volume"] = pd.to_numeric(d["volume"], errors="coerce")

    # 仅保留 in/out 行（balance 行无 entry）
    d = d[d["entry"].isin(["in", "out"])].reset_index(drop=True)

    trades: List[Dict[str, Any]] = []
    i = 0
    while i < len(d) - 1:
        in_row = d.iloc[i]
        out_row = d.iloc[i + 1]
        if in_row["entry"] != "in" or out_row["entry"] != "out":
            i += 1
            continue
        direction = "short" if in_row["type"] == "sell" else "long"
        profit = float(out_row.get("profit", 0.0) or 0.0)
        swap = float(out_row.get("swap", 0.0) or 0.0)
        commission = float(out_row.get("commission", 0.0) or 0.0)
        dur = (out_row["time"] - in_row["time"]).total_seconds() / 60.0
        trades.append(
            {
                "open_time": in_row["time"],
                "close_time": out_row["time"],
                "direction": direction,
                "volume": in_row.get("volume"),
                "open_price": in_row.get("price"),
                "close_price": out_row.get("price"),
                "profit": profit,
                "swap": swap,
                "commission": commission,
                "net_profit": profit + swap + commission,
                "duration_min": dur,
            }
        )
        i += 2

    return pd.DataFrame(trades)


# --------------------------------------------------------------------------- #
# 单段通用指标（供 walk_forward / run_analysis 复用）
# --------------------------------------------------------------------------- #
def compute_segment_metrics(net: pd.Series) -> Dict[str, float]:
    """
    由净盈亏序列计算核心指标。
    无状态、纯函数，可直接用于单段或单份报告。
    """
    n = len(net)
    if n == 0:
        return {"n": 0, "net": 0.0, "pf": 0.0, "win": 0.0, "dd": 0.0, "exp": 0.0}
    wins = net[net > 0]
    losses = net[net <= 0]
    gp = wins.sum()
    gl = -losses.sum()
    pf = gp / gl if gl > 0 else np.inf
    equity = net.cumsum()
    dd = float((equity - equity.cummax()).min())
    return {
        "n": int(n),
        "net": float(net.sum()),
        "pf": float(pf),
        "win": float(len(wins) / n * 100),
        "dd": dd,
        "exp": float(net.mean()),
    }


# --------------------------------------------------------------------------- #
# 汇总指标数值化（便于程序对比）
# --------------------------------------------------------------------------- #
def _normalize_summary_numbers(rep: MT5Report) -> None:
    s = rep.summary
    norm: Dict[str, Optional[float]] = {}

    # 直接数值字段
    direct = {
        "总净盈利": "total_net_profit",
        "毛利": "gross_profit",
        "毛损": "gross_loss",
        "盈利因子": "profit_factor",
        "预期收益": "expected_payoff",
        "夏普比率": "sharpe",
        "采收率": "recovery",
        "LR 相关性": "lr_correlation",
        "LR 标准误差": "lr_std_error",
        "总成交": "total_deals",
        "交易总计": "total_trades",
    }
    for cn, en in direct.items():
        norm[en] = _to_float(s.get(cn))

    # 回撤类形如 "971.54 (96.32%)" —— 同时拆金额与百分比
    dd_pairs = {
        "最大结余亏损": "max_balance_dd",
        "最大净值亏损": "max_equity_dd",
        "相对结余亏损": "rel_balance_dd",
        "相对净值亏损": "rel_equity_dd",
    }
    for cn, en in dd_pairs.items():
        val = _to_float(s.get(cn))           # 取金额
        # 单独找百分比
        pct = _pct_to_float(s.get(cn))
        norm[en] = val
        norm[en + "_pct"] = pct

    # 胜率类形如 "732 (38.28%)" —— 取百分比
    wr_pairs = {
        "盈利交易 (% 全部)": "win_rate_pct",
        "亏损交易 (% 全部)": "loss_rate_pct",
        "卖出交易 (赢得 %)": "sell_win_rate_pct",
        "买入交易 (赢得 %)": "buy_win_rate_pct",
    }
    for cn, en in wr_pairs.items():
        norm[en] = _pct_to_float(s.get(cn))

    # 平均盈亏
    norm["avg_win"] = _to_float(s.get("平均 获利交易"))
    norm["avg_loss"] = _to_float(s.get("平均 亏损交易"))
    norm["max_win"] = _to_float(s.get("最大 获利交易"))
    norm["max_loss"] = _to_float(s.get("最大 亏损交易"))

    rep.summary_norm = norm  # type: ignore[attr-defined]


# --------------------------------------------------------------------------- #
# 便捷：从 trades 直接计算扩展指标
# --------------------------------------------------------------------------- #
def compute_trade_metrics(trades: pd.DataFrame) -> Dict[str, Any]:
    """由逐笔交易 DataFrame 计算扩展指标。"""
    if trades is None or trades.empty:
        return {}

    t = trades
    n = len(t)
    net = t["net_profit"].astype(float)
    wins = net[net > 0]
    losses = net[net <= 0]
    gross_profit = wins.sum()
    gross_loss = -losses.sum()
    pf = gross_profit / gross_loss if gross_loss > 0 else np.inf

    # 连续盈亏
    streak_win = streak_loss = cur_w = cur_l = 0
    max_streak_win = max_streak_loss = 0
    for v in net:
        if v > 0:
            cur_w += 1
            cur_l = 0
            max_streak_win = max(max_streak_win, cur_w)
        else:
            cur_l += 1
            cur_w = 0
            max_streak_loss = max(max_streak_loss, cur_l)

    # 资金曲线与回撤
    equity = net.cumsum()
    running_max = equity.cummax()
    dd = equity - running_max
    max_dd = dd.min()

    # 按方向
    by_dir = {}
    for direction, g in t.groupby("direction"):
        gn = g["net_profit"].astype(float)
        w = gn[gn > 0]
        l = gn[gn <= 0]
        gp = w.sum()
        gl = -l.sum()
        by_dir[direction] = {
            "n": len(g),
            "win_rate": len(w) / len(g) * 100 if len(g) else 0,
            "net_profit": gn.sum(),
            "profit_factor": gp / gl if gl > 0 else np.inf,
            "avg_win": w.mean() if len(w) else 0,
            "avg_loss": l.mean() if len(l) else 0,
            "expectancy": gn.mean(),
        }

    # 按小时 / 星期 / 月份
    t2 = t.copy()
    t2["hour"] = t2["open_time"].dt.hour
    t2["weekday"] = t2["open_time"].dt.dayofweek  # 0=Mon
    t2["month"] = t2["open_time"].dt.to_period("M").astype(str)
    by_hour = t2.groupby("hour")["net_profit"].agg(["count", "sum", "mean"]).to_dict("index")
    by_weekday = t2.groupby("weekday")["net_profit"].agg(["count", "sum", "mean"]).to_dict("index")
    by_month = t2.groupby("month")["net_profit"].agg(["count", "sum", "mean"]).to_dict("index")

    return {
        "n_trades": n,
        "net_profit": float(net.sum()),
        "win_rate": float(len(wins) / n * 100),
        "profit_factor": float(pf),
        "avg_win": float(wins.mean()) if len(wins) else 0.0,
        "avg_loss": float(losses.mean()) if len(losses) else 0.0,
        "expectancy": float(net.mean()),
        "max_streak_win": int(max_streak_win),
        "max_streak_loss": int(max_streak_loss),
        "max_dd": float(max_dd),
        "avg_duration_min": float(t["duration_min"].mean()),
        "median_duration_min": float(t["duration_min"].median()),
        "by_direction": by_dir,
        "by_hour": by_hour,
        "by_weekday": by_weekday,
        "by_month": by_month,
        "_equity": equity,
        "_dd": dd,
    }


if __name__ == "__main__":
    import sys

    for f in sys.argv[1:]:
        rep = parse_report(f)
        print(f"== {f} ==")
        print("EA:", rep.meta.get("专家"))
        print("symbol:", rep.meta.get("交易品种"))
        print("period:", rep.meta.get("期间"))
        print("inputs:", rep.meta.get("inputs"))
        print("trades rows:", 0 if rep.trades is None else len(rep.trades))
        print("deals rows:", 0 if rep.deals is None else len(rep.deals))
        print("summary_norm sample:", rep.summary_norm)  # type: ignore[attr-defined]
        print()