# -*- coding: utf-8 -*- """ AI Agent MT5 报告分析 API ========================= 专为 AI Agent 设计的结构化接口。解析 MT5 报告，输出 JSON 可序列化结果， 提供设参优化建议。 用法: from mt5_agent import parse_report, analyze_report, compare_reports, get_param_suggestions # 解析单份报告 report = parse_report("path/to/report.htm") print(report.summary) # 汇总指标 print(report.metrics) # 扩展指标（PF、胜率、回撤等） print(report.trades) # 逐笔交易列表 # 单份报告分析 analysis = analyze_report(report) print(analysis.suggestions) # 优化建议列表 # 对比两份报告 comparison = compare_reports(report_a, report_b) print(comparison.diffs) # 关键指标差异 # 批量分析 from mt5_agent import batch_analyze result = batch_analyze(["dir1/", "dir2/"], "*.htm") print(result.rankings) # 按 PF 排名 print(result.anomalies) # 异常报告 # 设参优化建议 suggestions = get_param_suggestions(report, analysis) print(suggestions) # 可写入 .set 的参数修改建议 """ from __future__ import annotations import json import os from typing import Any, Dict, List, Optional, Tuple import numpy as np import pandas as pd # 导入内部模块 import mt5_report_parser as mp from mt5_report_parser import parse_report as _parse_report, MT5Report import run_analysis as ra import walk_forward as wf from param_scan import detect_plateaus, build_pivot, load_scan_results, overfit_score import mae_mfe as mm # ============================================================================= # # 数据模型（JSON 可序列化） # ============================================================================= # class ReportData: """解析后的 MT5 报告数据（AI 友好结构）。""" def __init__(self, rep: MT5Report): self.source_file = rep.source_file self.meta = _to_dict(rep.meta) self.summary = _to_dict(rep.summary_norm) # 数值化后的汇总 self._trades_df = rep.trades # 保持 DataFrame 原类型 self.trades = _trades_to_dict(rep.trades) if rep.trades is not None else [] self.metrics = _compute_metrics(rep.trades) if rep.trades is not None else {} # What-If 需要 open_time 列，部分报告可能缺失 if rep.trades is not None and "open_time" in rep.trades.columns: self.what_if = _what_if_to_dict(ra.whatif_scenarios(rep)) else: self.what_if = [] def to_json(self, indent: int = 2) -> str: """序列化为 JSON 字符串。""" return json.dumps({ "source_file": self.source_file, "meta": self.meta, "summary": self.summary, "metrics": self.metrics, "trades_count": len(self.trades), "what_if_scenarios": self.what_if, }, ensure_ascii=False, indent=indent) class AnalysisResult: """单份报告分析结果。""" def __init__(self, report: ReportData): self.report = report self.walk_forward = _wf_analysis(report) self.anomalies = _detect_anomalies(report) self.suggestions = _generate_suggestions(report, self.walk_forward, self.anomalies) def to_json(self, indent: int = 2) -> str: return json.dumps({ "source_file": self.report.source_file, "summary": self.report.summary, "metrics": self.report.metrics, "walk_forward": self.walk_forward, "anomalies": self.anomalies, "suggestions": self.suggestions, }, ensure_ascii=False, indent=indent) class ComparisonResult: """两份报告对比结果。""" def __init__(self, report_a: ReportData, report_b: ReportData): self.report_a = report_a self.report_b = report_b self.diffs = _compute_diffs(report_a, report_b) self.suggestions = _compare_suggestions(report_a, report_b) def to_json(self, indent: int = 2) -> str: return json.dumps({ "report_a": {"file": self.report_a.source_file, "summary": self.report_a.summary}, "report_b": {"file": self.report_b.source_file, "summary": self.report_b.summary}, "diffs": self.diffs, "suggestions": self.suggestions, }, ensure_ascii=False, indent=indent) class BatchResult: """批量分析结果。""" def __init__(self, reports: List[ReportData]): self.reports = reports self.rankings = _batch_rankings(reports) self.anomalies = [r for r in reports if r.metrics.get("n_trades", 0) < 30] self.best = self.rankings[0] if self.rankings else None def to_json(self, indent: int = 2) -> str: return json.dumps({ "count": len(self.reports), "rankings": self.rankings, "anomalies_count": len(self.anomalies), "best": self.best, }, ensure_ascii=False, indent=indent) # ============================================================================= # # 核心函数 # ============================================================================= # def parse_report(path: str) -> ReportData: """解析 MT5 报告（支持 .xlsx / .htm / .html）。""" rep = _parse_report(path) return ReportData(rep) def analyze_report(report: ReportData) -> AnalysisResult: """对单份报告做深度分析。""" return AnalysisResult(report) def compare_reports(report_a: ReportData, report_b: ReportData) -> ComparisonResult: """对比两份报告。""" return ComparisonResult(report_a, report_b) def batch_analyze(paths: List[str], pattern: str = "*.htm") -> BatchResult: """批量分析多个报告。""" from batch_report import scan_reports, parse_batch, build_summary_df files = scan_reports(paths, pattern) raw_reports = parse_batch(files) report_datas = [] for name, rep in raw_reports: try: rd = ReportData(rep) rd._filename = name report_datas.append(rd) except Exception as e: print("警告: 跳过 %s - %s" % (name, e)) return BatchResult(report_datas) def get_param_suggestions(analysis: AnalysisResult) -> List[Dict[str, Any]]: """ 生成 .set 文件参数修改建议。 返回格式：[{"param_name": "InpStopLossPoints", "suggested_value": "150", "reason": "当前 200，亏损截断场景显示 150 更优"}] """ suggestions = [] report = analysis.report metrics = report.metrics summary = report.summary what_if = report.what_if # 建议 1：止损优化 if metrics.get("avg_loss", 0) != 0: # 找亏损截断场景 cut_scenarios = [s for s in what_if if "亏损截断" in s.get("name", "")] if cut_scenarios: base_pf = summary.get("profit_factor", 0) for s in cut_scenarios: if s.get("pf", 0) > base_pf and s.get("dd", 0) < metrics.get("max_dd", 0): cap = s.get("desc", "") suggestions.append({ "param_name": "InpStopLossPoints", "suggested_value": None, # 需要从场景推算 "reason": f"亏损截断场景 PF={s['pf']:.2f} > 基线 PF={base_pf:.2f}，回撤从 {metrics.get('max_dd',0):.0f} 降至 {s['dd']:.0f}", "priority": "high", }) # 建议 2：止盈优化 tp_scenarios = [s for s in what_if if "盈利单放大" in s.get("name", "")] if tp_scenarios: base_pf = summary.get("profit_factor", 0) for s in tp_scenarios: if s.get("pf", 0) > base_pf: suggestions.append({ "param_name": "InpTakeProfitPoints", "suggested_value": None, "reason": f"盈利单放大场景 PF={s['pf']:.2f} > 基线 PF={base_pf:.2f}，当前止盈可能太紧", "priority": "medium", }) # 建议 3：仓位优化 halve_scenarios = [s for s in what_if if "仓位减半" in s.get("name", "")] if halve_scenarios: base_dd = metrics.get("max_dd", 0) for s in halve_scenarios: if abs(s.get("dd", 0)) < abs(base_dd * 0.5): suggestions.append({ "param_name": "InpFixedLots", "suggested_value": None, "reason": f"仓位减半后回撤从 {base_dd:.0f} 降至 {s['dd']:.0f}，可考虑降低仓位", "priority": "medium", }) # 建议 4：方向性建议 if metrics.get("by_direction"): for direction, stats in metrics["by_direction"].items(): if stats.get("win_rate", 0) < 35 and stats.get("profit_factor", 0) < 0.8: suggestions.append({ "param_name": None, "suggested_value": None, "reason": f"方向 {direction} 胜率仅 {stats['win_rate']:.1f}%，PF={stats['profit_factor']:.2f}，考虑单独过滤该方向信号", "priority": "high", }) # 建议 5：时段过滤 if metrics.get("by_hour"): bad_hours = [h for h, stats in metrics["by_hour"].items() if stats.get("sum", 0) < 0 and stats.get("count", 0) > 5] if bad_hours: suggestions.append({ "param_name": "InpFreezeBarCount", "suggested_value": None, "reason": f"时段 {bad_hours[:3]} 净盈利持续为负，建议添加时段过滤", "priority": "low", }) return suggestions # ============================================================================= # # 内部工具函数 # ============================================================================= # def _to_dict(obj: Any) -> Any: """递归将 pandas/numpy 对象转为原生 Python 类型。""" if obj is None: return None if isinstance(obj, dict): return {k: _to_dict(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_to_dict(v) for v in obj] if isinstance(obj, (np.integer,)): return int(obj) if isinstance(obj, (np.floating,)): if np.isnan(obj) or np.isinf(obj): return None return float(obj) if isinstance(obj, pd.Timestamp): return str(obj) if isinstance(obj, pd.DataFrame): return _to_dict(obj.to_dict(orient="index")) if isinstance(obj, pd.Series): return _to_dict(obj.to_dict()) if isinstance(obj, pd.Index): return _to_dict(list(obj)) if isinstance(obj, (float, int)) and (np.isnan(obj) or np.isinf(obj)): return None if isinstance(obj, (int, float, str, bool)): return obj return str(obj) def _trades_to_dict(trades: pd.DataFrame) -> List[Dict[str, Any]]: """将逐笔交易 DataFrame 转为字典列表。""" if trades is None or trades.empty: return [] result = [] for _, row in trades.iterrows(): item = {} for col in ["open_time", "close_time", "direction", "volume", "open_price", "close_price", "profit", "swap", "commission", "net_profit", "duration_min"]: val = row.get(col) if val is None or (isinstance(val, float) and np.isnan(val)): item[col] = None elif isinstance(val, pd.Timestamp): item[col] = str(val) elif isinstance(val, (np.integer,)): item[col] = int(val) elif isinstance(val, (np.floating,)): item[col] = float(val) else: item[col] = val result.append(item) return result def _compute_metrics(trades: pd.DataFrame) -> Dict[str, Any]: """计算扩展指标（JSON 可序列化）。""" m = ra.extended_metrics(trades) if trades is not None and not trades.empty else {} result = _to_dict(m) # 移除内部字段 for key in ["_equity", "_dd", "_net"]: result.pop(key, None) return result def _what_if_to_dict(scenarios: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """将 What-If 场景转为字典列表。""" return [_to_dict(s) for s in scenarios] def _wf_analysis(report: ReportData) -> Dict[str, Any]: """Walk-Forward 分析。""" trades_df = getattr(report, "_trades_df", None) if trades_df is None or trades_df.empty: return {} # 确保 open_time 全为 Timestamp 类型 trades_df = trades_df.copy() trades_df["open_time"] = pd.to_datetime(trades_df["open_time"], errors="coerce") trades_df = trades_df.dropna(subset=["open_time"]) trades_df["net_profit"] = pd.to_numeric(trades_df["net_profit"], errors="coerce") trades_df = trades_df.dropna(subset=["net_profit"]) if trades_df.empty: return {} windows = wf.walk_forward(trades_df) summary = wf.wf_summary(windows) return _to_dict(summary) def _detect_anomalies(report: ReportData) -> List[Dict[str, Any]]: """检测异常指标。""" anomalies = [] metrics = report.metrics summary = report.summary if metrics.get("n_trades", 0) < 30: anomalies.append({ "type": "样本不足", "detail": f"仅 {metrics['n_trades']} 笔交易", }) if summary.get("profit_factor", 0) > 10: anomalies.append({ "type": "PF异常高", "detail": f"PF={summary['profit_factor']:.2f}，可能过拟合", }) if metrics.get("max_dd", 0) < -1000: anomalies.append({ "type": "回撤过大", "detail": f"最大回撤 {metrics['max_dd']:.0f}", }) return anomalies def _generate_suggestions( report: ReportData, wf_result: Dict[str, Any], anomalies: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: """生成优化建议。""" suggestions = [] metrics = report.metrics summary = report.summary # Walk-Forward 建议 if wf_result: wfe = wf_result.get("wfe", 0) if wfe is not None and wfe < 0.3: suggestions.append({ "type": "Walk-Forward", "detail": f"WFE={wfe:.2f}，OOS 表现弱，参数可能过拟合 IS 段", "priority": "high", }) # 什么-If 建议 for s in report.what_if: name = s.get("name", "") if "信号反向" in name and s.get("net", 0) > 0 and s.get("pf", 0) > 1: suggestions.append({ "type": "信号方向", "detail": f"反向场景净盈利={s['net']:+.2f}, PF={s['pf']:.2f}，可能方向逻辑写反", "priority": "critical", }) return suggestions def _compute_diffs(report_a: ReportData, report_b: ReportData) -> Dict[str, Any]: """计算两份报告的关键指标差异。""" diffs = {} for key in ["profit_factor", "win_rate", "net_profit", "max_dd", "sharpe", "sortino"]: va = report_a.summary.get(key) vb = report_b.summary.get(key) if va is not None and vb is not None: diff = vb - va if (not np.isinf(va) or not np.isinf(vb)) else None pct = ((vb - va) / va * 100) if va != 0 and not np.isinf(va) else None diffs[key] = { "a": va, "b": vb, "diff": diff, "pct_change": pct, } return diffs def _compare_suggestions(report_a: ReportData, report_b: ReportData) -> List[Dict[str, Any]]: """对比分析建议。""" suggestions = [] pf_a = report_a.summary.get("profit_factor", 0) pf_b = report_b.summary.get("profit_factor", 0) if pf_b > pf_a and pf_a > 0: suggestions.append({ "type": "参数优化", "detail": f"报告B PF={pf_b:.2f} 优于报告A PF={pf_a:.2f}，差 {pf_b-pf_a:.2f}", "priority": "medium", }) elif pf_b < pf_a and pf_b > 0: suggestions.append({ "type": "参数回退", "detail": f"报告B PF={pf_b:.2f} 差于报告A PF={pf_a:.2f}，参数可能过拟合", "priority": "medium", }) return suggestions def _batch_rankings(reports: List[ReportData]) -> List[Dict[str, Any]]: """按 PF 排名。""" valid = [r for r in reports if r.summary.get("profit_factor") is not None and r.metrics.get("n_trades", 0) > 0] sorted_reports = sorted(valid, key=lambda r: r.summary.get("profit_factor", 0), reverse=True) rankings = [] for i, r in enumerate(sorted_reports[:20]): # 只取前 20 rankings.append({ "rank": i + 1, "file": r.source_file, "filename": getattr(r, "_filename", os.path.basename(r.source_file)), "profit_factor": r.summary.get("profit_factor"), "net_profit": r.summary.get("total_net_profit"), "win_rate": r.metrics.get("win_rate"), "n_trades": r.metrics.get("n_trades"), "max_dd": r.metrics.get("max_dd"), }) return rankings # ============================================================================= # # CLI # ============================================================================= # if __name__ == "__main__": import argparse ap = argparse.ArgumentParser(description="AI Agent MT5 报告分析 API") ap.add_argument("path", help="报告文件路径或目录") ap.add_argument("--compare", help="对比的另一份报告路径") ap.add_argument("--batch", action="store_true", help="批量分析目录") ap.add_argument("--suggestions", action="store_true", help="输出设参建议") ap.add_argument("--format", choices=["json", "text"], default="json", help="输出格式") args = ap.parse_args() if args.batch: result = batch_analyze([args.path]) print(result.to_json()) elif args.compare: a = parse_report(args.path) b = parse_report(args.compare) comparison = compare_reports(a, b) if args.format == "json": print(comparison.to_json()) else: print(f"\n报告A: {args.path}") print(f" PF={a.summary.get('profit_factor')} 净盈利={a.summary.get('total_net_profit')}") print(f"报告B: {args.compare}") print(f" PF={b.summary.get('profit_factor')} 净盈利={b.summary.get('total_net_profit')}") print(f"\n关键差异:") for key, diff in comparison.diffs.items(): print(f" {key}: A={diff['a']} B={diff['b']} 差={diff['diff']:.2f}") else: report = parse_report(args.path) analysis = analyze_report(report) if args.suggestions: suggestions = get_param_suggestions(analysis) print(json.dumps(suggestions, ensure_ascii=False, indent=2)) elif args.format == "json": print(analysis.to_json()) else: print(f"\n报告: {args.path}") print(f" PF={report.summary.get('profit_factor')} 净盈利={report.summary.get('total_net_profit')}") print(f" 胜率={report.metrics.get('win_rate')} 笔数={report.metrics.get('n_trades')}") print(f" 回撤={report.metrics.get('max_dd')}") print(f"\n建议 ({len(analysis.suggestions)} 条):") for s in analysis.suggestions: print(f" [{s.get('priority','?')}] {s['type']}: {s['detail']}")