512 lines
19 KiB
Python
512 lines
19 KiB
Python
# -*- 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']}")
|