780 lines
26 KiB
Python
780 lines
26 KiB
Python
# -*- 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()
|