Initial commit - MT5 Tools dashboard

This commit is contained in:
unknown
2026-04-12 18:25:46 +10:00
commit 6b677b8259
20 changed files with 3202 additions and 0 deletions
+48
View File
@@ -0,0 +1,48 @@
# MT5 Tools
Standalone Streamlit dashboard for MT5 trade analysis and comparison.
## Setup
```powershell
cd C:\Users\pc\MT5Tools
pip install -r requirements.txt
```
## Launch
```powershell
streamlit run app.py
```
## Files
```
MT5Tools\
├── app.py ← main Streamlit app
├── mt5_parser.py ← parsers for all 3 formats + stats
├── mt5_batch_backtest.py ← batch backtest runner (copy from ea\)
├── set_comparator.py ← EA set file comparator (copy from ea\)
├── requirements.txt
├── mt5_batch_config.json ← gitignored, created on first run
└── pages\
├── trade_analysis.py ← single report analysis
├── trade_compare.py ← side-by-side comparison
└── settings.py
```
## Supported Formats
| Format | Extension | Notes |
|---|---|---|
| MT5 Account History | `.htm` / `.html` | Export from MT5 → Account History → Save as Report |
| MT5 Backtest Report | `.htm` / `.html` | Generated by batch backtest runner or manual tester |
| Quant Analyzer CSV | `.csv` | Export listOfTrades from Quant Analyzer |
## .gitignore
Add the following to `.gitignore`:
```
mt5_batch_config.json
```
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+186
View File
@@ -0,0 +1,186 @@
"""
MT5 Tools Dashboard
===================
Streamlit app for MT5 trade analysis and comparison.
Launch: streamlit run app.py
"""
import streamlit as st
from streamlit_option_menu import option_menu
import importlib, sys, os
# ── Page config ───────────────────────────────────────────────────────────────
st.set_page_config(
page_title = "MT5 Tools",
page_icon = "📈",
layout = "wide",
initial_sidebar_state = "expanded"
)
# ── Theme ─────────────────────────────────────────────────────────────────────
st.markdown("""
<style>
/* Base */
@import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;500;700&family=Syne:wght@400;600;800&display=swap');
html, body, [class*="css"] {
font-family: 'Syne', sans-serif;
background-color: #0a0a0f;
color: #e0e0e8;
}
code, .mono { font-family: 'JetBrains Mono', monospace; }
/* Sidebar */
[data-testid="stSidebar"] {
background: linear-gradient(180deg, #0d0d1a 0%, #0a0a12 100%);
border-right: 1px solid rgba(255,255,255,0.06);
}
[data-testid="stSidebar"] .stMarkdown h3 {
color: #7c6af7;
font-size: 11px;
letter-spacing: 0.15em;
text-transform: uppercase;
font-weight: 800;
}
/* Cards */
.stat-card {
background: rgba(255,255,255,0.03);
border: 1px solid rgba(255,255,255,0.07);
border-radius: 8px;
padding: 14px 18px;
margin-bottom: 8px;
}
.stat-label {
font-size: 10px;
letter-spacing: 0.1em;
text-transform: uppercase;
color: #666;
margin-bottom: 4px;
}
.stat-value {
font-family: 'JetBrains Mono', monospace;
font-size: 22px;
font-weight: 700;
color: #e0e0e8;
}
.stat-pos { color: #2dc653; }
.stat-neg { color: #e63946; }
/* Info card */
.info-card {
background: rgba(124,106,247,0.08);
border: 1px solid rgba(124,106,247,0.2);
border-radius: 8px;
padding: 12px 16px;
font-size: 13px;
color: #aaa;
margin: 8px 0;
}
/* Comparison cells */
.diff-pos { color: #2dc653; font-weight: 600; }
.diff-neg { color: #e63946; font-weight: 600; }
.diff-neu { color: #888; }
/* Divider */
hr { border-color: rgba(255,255,255,0.06); }
/* Metric overrides */
[data-testid="metric-container"] {
background: rgba(255,255,255,0.02);
border: 1px solid rgba(255,255,255,0.06);
border-radius: 8px;
padding: 12px;
}
[data-testid="stMetricValue"] {
font-family: 'JetBrains Mono', monospace;
font-size: 18px !important;
}
/* Buttons */
.stButton > button {
background: rgba(124,106,247,0.15);
border: 1px solid rgba(124,106,247,0.3);
color: #c5beff;
font-family: 'Syne', sans-serif;
font-weight: 600;
letter-spacing: 0.05em;
border-radius: 6px;
transition: all 0.15s;
}
.stButton > button:hover {
background: rgba(124,106,247,0.3);
border-color: rgba(124,106,247,0.6);
}
/* File uploader */
[data-testid="stFileUploader"] {
background: rgba(255,255,255,0.02);
border: 1px dashed rgba(255,255,255,0.12);
border-radius: 8px;
}
/* Tab overrides */
.stTabs [data-baseweb="tab"] {
font-family: 'Syne', sans-serif;
font-weight: 600;
font-size: 13px;
}
/* Scrollbar */
::-webkit-scrollbar { width: 6px; height: 6px; }
::-webkit-scrollbar-track { background: transparent; }
::-webkit-scrollbar-thumb { background: rgba(255,255,255,0.1); border-radius: 3px; }
</style>
""", unsafe_allow_html=True)
# ── Sidebar nav ───────────────────────────────────────────────────────────────
with st.sidebar:
st.markdown("### 📈 MT5 Tools")
st.markdown("---")
page = option_menu(
menu_title = None,
options = ["Trade Analysis", "Trade Compare", "EA Comparator", "Settings"],
icons = ["bar-chart-line", "arrow-left-right", "sliders", "gear"],
default_index = 0,
styles = {
"container" : {"background-color": "transparent", "padding": "0"},
"icon" : {"color": "#7c6af7", "font-size": "14px"},
"nav-link" : {
"font-family" : "Syne, sans-serif",
"font-size" : "13px",
"font-weight" : "600",
"color" : "#aaa",
"border-radius": "6px",
"margin" : "2px 0",
},
"nav-link-selected": {
"background-color": "rgba(124,106,247,0.15)",
"color" : "#c5beff",
"border" : "1px solid rgba(124,106,247,0.25)",
},
}
)
# ── Route pages ───────────────────────────────────────────────────────────────
if page == "Trade Analysis":
import view_trade_analysis as p
importlib.reload(p)
p.render()
elif page == "Trade Compare":
import view_trade_compare as p
importlib.reload(p)
p.render()
elif page == "EA Comparator":
import view_set_comparator as p
importlib.reload(p)
p.render()
elif page == "Settings":
import view_settings as p
importlib.reload(p)
p.render()
+669
View File
@@ -0,0 +1,669 @@
"""
MT5 Batch Backtest Runner
=========================
Generates a MT5 tester .ini file for each .set file in a folder,
updates EA_Comment in each .set file, then launches MT5 terminal
for each backtest sequentially.
On first run: detects MT5 installations and saves config.
Subsequent runs: loads saved config, prompts to use same or modify.
Usage: python mt5_batch_backtest.py
"""
import os
import sys
import glob
import subprocess
import time
import shutil
import json
# ── MT5 Period constants ───────────────────────────────────────────────────────
PERIOD_MAP = {
'M1' : 'M1',
'M5' : 'M5',
'M15' : 'M15',
'M30' : 'M30',
'H1' : 'H1',
'H4' : 'H4',
'D' : 'Daily',
'D1' : 'Daily',
'DAILY': 'Daily',
'W1' : 'Weekly',
'MN' : 'Monthly',
}
# ── Model labels ─────────────────────────────────────────────────────────────
MODEL_LABELS = {
'1' : 'OHLC',
'2' : 'CTRLPTS',
'4' : 'EVERYTICK',
'5' : 'EVERYTICKREAL',
}
# ── Defaults (used if no config found) ────────────────────────────────────────
DEFAULTS = {
'terminal_path' : r"C:\Program Files\MetaTrader 5\terminal64.exe",
'tester_folder' : '',
'ea_name' : r"Market\Ultimate Breakout System.ex5",
'from_date' : '2018.01.01',
'to_date' : '2026.04.01',
'model' : '1',
'deposit' : '10000',
'currency' : 'USD',
'leverage' : '100',
'optimization' : '0',
'suffix' : '.a',
}
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'mt5_batch_config.json')
# ── Helpers ────────────────────────────────────────────────────────────────────
def prompt(text, default=None):
if default is not None and default != '':
val = input(f" {text} [{default}]: ").strip()
return val if val else default
else:
while True:
val = input(f" {text}: ").strip()
if val:
return val
print(" (required)")
def load_config():
if os.path.isfile(CONFIG_FILE):
try:
with open(CONFIG_FILE, 'r') as f:
return json.load(f)
except:
pass
return None
def save_config(cfg):
try:
with open(CONFIG_FILE, 'w') as f:
json.dump(cfg, f, indent=2)
except Exception as e:
print(f" WARNING: Could not save config: {e}")
def find_mt5_terminals():
"""Scan MetaQuotes Terminal folder for all MT5 installations."""
appdata = os.environ.get('APPDATA', '')
base = os.path.join(appdata, 'MetaQuotes', 'Terminal')
results = []
if not os.path.isdir(base):
return results
for entry in os.listdir(base):
entry_path = os.path.join(base, entry)
if not os.path.isdir(entry_path):
continue
# Check for origin.txt which contains the terminal exe path
origin = os.path.join(entry_path, 'origin.txt')
tester = os.path.join(entry_path, 'Tester')
label = entry
if os.path.isfile(origin):
try:
with open(origin, 'r', encoding='utf-8', errors='replace') as f:
label = f.read().strip() or entry
except:
pass
# Skip folders that don't look like real MT5 terminals
if not os.path.isdir(os.path.join(entry_path, 'MQL5')):
continue
results.append({
'id' : entry,
'label' : label,
'tester_folder' : tester,
'data_folder' : entry_path,
})
return results
def pick_terminal():
"""Let user pick from detected MT5 terminals."""
terminals = find_mt5_terminals()
if not terminals:
print(" No MT5 terminals found in AppData\\MetaQuotes\\Terminal\\")
print(" You will need to enter the tester folder path manually.")
return None, None
print()
print(" Detected MT5 terminal(s):")
for i, t in enumerate(terminals, 1):
print(f" {i}) {t['label']}")
print(f" Tester: {t['tester_folder']}")
if len(terminals) == 1:
choice = input(f"\n Select terminal [1]: ").strip()
idx = 0
else:
while True:
choice = input(f"\n Select terminal [1-{len(terminals)}]: ").strip()
try:
idx = int(choice) - 1
if 0 <= idx < len(terminals):
break
except:
pass
print(" Invalid selection.")
selected = terminals[idx]
return selected['tester_folder'], selected['label']
def find_ea_files(tester_folder):
"""
Scan MQL5/Experts folder for .ex5 files.
Returns list of dicts with label (display) and value (ini path).
"""
experts_dir = os.path.join(os.path.dirname(tester_folder), 'MQL5', 'Experts')
results = []
if not os.path.isdir(experts_dir):
return results
for root, dirs, files in os.walk(experts_dir):
dirs[:] = [d for d in dirs if not d.startswith('.')]
for fn in sorted(files):
if fn.lower().endswith('.ex5'):
full_path = os.path.join(root, fn)
rel = os.path.relpath(full_path, experts_dir)
# Only include EAs in the Market subfolder
if rel.startswith('Market' + os.sep) or rel.startswith('Market/'):
results.append({'label': rel, 'value': rel})
return results
def pick_ea_name(tester_folder, current=None):
"""List available EAs and let user pick, or enter manually."""
eas = find_ea_files(tester_folder)
if not eas:
print(" No .ex5 files found in MQL5/Experts -- enter EA name manually.")
return prompt("EA Name", current or DEFAULTS['ea_name'])
print()
print(" Available EAs:")
for i, ea in enumerate(eas, 1):
marker = ' <' if current and ea['value'] == current else ''
print(f" {i:>3}) {ea['label']}{marker}")
print(f" {len(eas)+1:>3}) Enter manually")
while True:
default_idx = None
if current:
for i, ea in enumerate(eas, 1):
if ea['value'] == current:
default_idx = i
break
hint = f"1-{len(eas)+1}" + (f", Enter={default_idx}" if default_idx else "")
choice = input(f" Select EA [{hint}]: ").strip()
if choice == '' and default_idx:
return eas[default_idx - 1]['value']
try:
idx = int(choice) - 1
if idx == len(eas):
return prompt("EA Name", current or DEFAULTS['ea_name'])
if 0 <= idx < len(eas):
return eas[idx]['value']
except:
pass
print(" Invalid selection.")
def setup_config():
"""First-run setup — detect terminals and build config."""
print()
print(" ── First Run Setup ──────────────────────────────────────")
tester_folder, terminal_label = pick_terminal()
if not tester_folder:
tester_folder = prompt("Tester folder path")
terminal_path = prompt("Path to terminal64.exe", DEFAULTS['terminal_path'])
print()
print(" ── Backtest Defaults ────────────────────────────────────")
cfg = {
'terminal_path' : terminal_path,
'tester_folder' : tester_folder,
'terminal_label': terminal_label or '',
'ea_name' : pick_ea_name(tester_folder, DEFAULTS['ea_name']),
'from_date' : prompt("From Date (YYYY.MM.DD)", DEFAULTS['from_date']),
'to_date' : prompt("To Date (YYYY.MM.DD)", DEFAULTS['to_date']),
'model' : prompt("Model (1=OHLC M1, 2=Control points, 4=Every tick)", DEFAULTS['model']),
'deposit' : prompt("Deposit", DEFAULTS['deposit']),
'currency' : prompt("Currency", DEFAULTS['currency']),
'leverage' : prompt("Leverage", DEFAULTS['leverage']),
'suffix' : prompt("Instrument suffix (e.g. .a)", DEFAULTS['suffix']),
}
save_config(cfg)
print()
print(f" Config saved to: {CONFIG_FILE}")
return cfg
def review_config(cfg):
"""Show saved config and ask to use same or modify."""
print()
print(" ── Saved Settings ───────────────────────────────────────")
print(f" Terminal : {cfg.get('terminal_label', cfg['tester_folder'])}")
print(f" Tester : {cfg['tester_folder']}")
print(f" EA : {cfg['ea_name']}")
print(f" Dates : {cfg['from_date']}{cfg['to_date']}")
print(f" Model : {cfg['model']} Deposit: {cfg['deposit']} {cfg['currency']} Leverage: {cfg['leverage']}")
print(f" Suffix : {cfg['suffix']}")
print()
choice = input(" Use these settings? [Y/n/reset]: ").strip().lower()
if choice == 'reset':
os.remove(CONFIG_FILE)
print(" Config reset — re-running setup.")
return setup_config()
if choice == 'n':
print()
print(" ── Modify Settings ──────────────────────────────────────")
redetect = input(" Re-detect MT5 terminals? [y/N]: ").strip().lower()
if redetect == 'y':
tester_folder, terminal_label = pick_terminal()
if tester_folder:
cfg['tester_folder'] = tester_folder
cfg['terminal_label'] = terminal_label or ''
cfg['terminal_path'] = prompt("terminal64.exe path", cfg['terminal_path'])
cfg['ea_name'] = pick_ea_name(cfg['tester_folder'], cfg['ea_name'])
cfg['from_date'] = prompt("From Date", cfg['from_date'])
cfg['to_date'] = prompt("To Date", cfg['to_date'])
cfg['model'] = prompt("Model (1=OHLC M1, 2=Control points, 4=Every tick)", cfg['model'])
cfg['deposit'] = prompt("Deposit", cfg['deposit'])
cfg['currency'] = prompt("Currency", cfg['currency'])
cfg['leverage'] = prompt("Leverage", cfg['leverage'])
cfg['suffix'] = prompt("Suffix", cfg['suffix'])
save_config(cfg)
print(" Config updated.")
return cfg
def detect_timeframe(filename):
name = os.path.splitext(filename)[0].upper()
for token in PERIOD_MAP:
if name.endswith('_' + token) or name.endswith('-' + token) or \
('_' + token + '_') in name or ('-' + token + '-') in name:
return PERIOD_MAP[token]
return None
def detect_instrument(filename, n_chars):
return os.path.splitext(filename)[0][:n_chars].upper()
def read_utf16(path):
with open(path, 'rb') as f:
raw = f.read()
if raw[:2] == b'\xff\xfe':
text = raw[2:].decode('utf-16-le')
elif raw[:2] == b'\xfe\xff':
text = raw[2:].decode('utf-16-be')
else:
text = raw.decode('utf-8', errors='replace')
return text.splitlines()
def write_utf16(path, lines):
text = '\r\n'.join(lines) + '\r\n'
with open(path, 'wb') as f:
f.write(b'\xff\xfe')
f.write(text.encode('utf-16-le'))
def update_set_file(set_path, ea_comment, lot_mode, lot_value):
lines = read_utf16(set_path)
def update_param(lines, key, new_val):
for i, line in enumerate(lines):
if line.strip().startswith(key + '='):
parts = line.strip().split('||')
parts[0] = f'{key}={new_val}'
lines[i] = '||'.join(parts)
return True
return False
# Always update EA_Comment
updated = False
for i, line in enumerate(lines):
if line.strip().startswith('EA_Comment='):
lines[i] = f'EA_Comment={ea_comment}'
updated = True
break
if not updated:
lines.append(f'EA_Comment={ea_comment}')
if lot_mode == 'manual':
update_param(lines, 'Risk', '0')
update_param(lines, 'StartLots', str(lot_value))
elif lot_mode == 'balance':
update_param(lines, 'Risk', '9999')
update_param(lines, 'LotPerBalance_step', str(lot_value))
# lot_mode None/'asis': only EA_Comment updated
write_utf16(set_path, lines)
def build_ini(symbol, period, set_file_path, ini_out_path, report_folder, cfg):
name_stem = os.path.splitext(os.path.basename(set_file_path))[0]
model_label = MODEL_LABELS.get(cfg['model'], f"M{cfg['model']}")
report_name = f"{name_stem}_{model_label}"
content = (
'[Tester]\r\n'
f'Expert={cfg["ea_name"]}\r\n'
f'Symbol={symbol}\r\n'
f'Period={period}\r\n'
f'Optimization={cfg.get("optimization","0")}\r\n'
f'Model={cfg["model"]}\r\n'
f'FromDate={cfg["from_date"]}\r\n'
f'ToDate={cfg["to_date"]}\r\n'
'ForwardMode=0\r\n'
f'Deposit={cfg["deposit"]}\r\n'
f'Currency={cfg["currency"]}\r\n'
'ProfitInPips=0\r\n'
f'Leverage={cfg["leverage"]}\r\n'
'ExecutionMode=0\r\n'
'OptimizationCriterion=0\r\n'
'Visual=0\r\n'
f'Report={report_name}\r\n'
'ReplaceReport=1\r\n'
f'Inputs={set_file_path}\r\n'
'ShutdownTerminal=1\r\n'
)
with open(ini_out_path, 'wb') as f:
f.write(content.encode('utf-8'))
# ── Main ───────────────────────────────────────────────────────────────────────
def main():
print("=" * 60)
print(" MT5 Batch Backtest Runner")
print("=" * 60)
# ── Load or create config ──────────────────────────────────────
cfg = load_config()
if cfg is None:
cfg = setup_config()
else:
cfg = review_config(cfg)
terminal_path = cfg['terminal_path']
tester_folder = cfg['tester_folder']
os.makedirs(tester_folder, exist_ok=True)
if not os.path.isfile(terminal_path):
print(f"\n WARNING: terminal64.exe not found at: {terminal_path}")
# ── Folder of set files ────────────────────────────────────────
print()
set_folder = prompt("Path to folder containing .set files")
set_folder = os.path.expandvars(set_folder.strip('"').strip("'"))
if not os.path.isdir(set_folder):
print(f" ERROR: Folder not found: {set_folder}")
sys.exit(1)
all_set_files = sorted(glob.glob(os.path.join(set_folder, '**', '*.set'), recursive=True))
set_files = [f for f in all_set_files
if not os.path.basename(f).lower().startswith('optimization')
and '_batch_modified' not in f.replace('\\', '/')]
skipped = len(all_set_files) - len(set_files)
if not set_files:
print(" ERROR: No .set files found (after exclusions).")
sys.exit(1)
print(f" Found {len(set_files)} .set file(s).", end='')
if skipped:
print(f" ({skipped} Optimization file(s) excluded)", end='')
print()
# ── Report output folder ───────────────────────────────────────
default_reports = os.path.join(set_folder, 'reports')
report_folder = prompt("Path to save reports", default_reports)
report_folder = report_folder.strip('"').strip("'")
os.makedirs(report_folder, exist_ok=True)
# ── Lot size mode ──────────────────────────────────────────────
print()
print(" Lot size mode:")
print(" 0 = Use set file as-is (no changes to Risk/Lots)")
print(" 1 = Manual lot size (same for all) — sets Risk=0, StartLots=X")
print(" 2 = Lots per balance (from each .set file) — sets Risk=9999")
print(" 3 = Lots per balance (enter per file) — sets Risk=9999")
lot_mode_choice = prompt("Choose [0/1/2/3]", "2")
lot_mode = {'0': 'asis', '1': 'manual'}.get(lot_mode_choice, 'balance')
balance_ask = lot_mode_choice == '3'
manual_lots = None
if lot_mode == 'asis':
print(" Set files used as-is — no Risk/Lots changes.")
elif lot_mode == 'manual':
manual_lots = prompt("StartLots for all files", "0.01")
elif balance_ask:
print(" Will prompt LotPerBalance_step per file. Risk=9999.")
else:
print(" Using LotPerBalance_step from each file. Risk=9999.")
# ── Instrument mode ────────────────────────────────────────────
print()
print(" Instrument detection:")
print(" 1 = Enter one instrument for all set files")
print(" 2 = Extract from filename (specify number of characters)")
print(" 3 = Ask per file")
instr_mode = prompt("Choose [1/2/3]", "1")
instr_global = None
instr_n_chars = None
if instr_mode == '1':
instr_global = prompt("Instrument (without suffix, e.g. GBPJPY)").upper()
elif instr_mode == '2':
instr_n_chars = int(prompt("Characters from start of filename", "6"))
# ── Timeframe mode ─────────────────────────────────────────────
print()
print(" Timeframe:")
print(" 1 = One timeframe for all set files")
print(" 2 = Detect from filename (D/Daily/H1/H4 etc.)")
print(" 3 = Ask per file")
tf_mode = prompt("Choose [1/2/3]", "2")
tf_global = None
if tf_mode == '1':
tf_raw = prompt("Timeframe (e.g. Daily, H1, H4, M15)").upper()
tf_global = PERIOD_MAP.get(tf_raw, tf_raw)
# ── Output folder for modified .set copies ─────────────────────
# (subfolders mirrored inside _batch_modified and reports)
out_set_base = os.path.join(set_folder, '_batch_modified')
os.makedirs(out_set_base, exist_ok=True)
# ── Process each set file ──────────────────────────────────────
print()
print("=" * 60)
print(f" Processing {len(set_files)} file(s)...")
print("=" * 60)
results = []
for set_path in set_files:
filename = os.path.basename(set_path)
name_stem = os.path.splitext(filename)[0]
# Mirror subfolder structure from set_folder root
rel_path = os.path.relpath(set_path, set_folder)
rel_subdir = os.path.dirname(rel_path)
out_set_folder = os.path.join(out_set_base, rel_subdir) if rel_subdir else out_set_base
file_report_dir = os.path.join(report_folder, rel_subdir) if rel_subdir else report_folder
os.makedirs(out_set_folder, exist_ok=True)
os.makedirs(file_report_dir, exist_ok=True)
subfolder_label = f" ({rel_subdir})" if rel_subdir else ""
print(f"\n [{filename}]{subfolder_label}")
# Instrument
if instr_mode == '1':
instrument = instr_global
elif instr_mode == '2':
instrument = detect_instrument(filename, instr_n_chars)
print(f" Instrument: {instrument}")
else:
instrument = prompt(f"Instrument for {filename} (without suffix)").upper()
symbol = instrument + cfg['suffix']
# Timeframe
if tf_mode == '1':
period = tf_global
elif tf_mode == '2':
period = detect_timeframe(filename)
if period:
print(f" Timeframe : {period}")
else:
tf_raw = prompt(f"Timeframe for {filename} (e.g. Daily, H1, H4)").upper()
period = PERIOD_MAP.get(tf_raw, tf_raw)
else:
tf_raw = prompt(f"Timeframe for {filename}").upper()
period = PERIOD_MAP.get(tf_raw, tf_raw)
# Lot value
if lot_mode == 'asis':
lot_value = None
print(f" Lots : as-is")
elif lot_mode == 'manual':
lot_value = manual_lots
print(f" Lots : Manual StartLots={lot_value}")
else:
if balance_ask:
file_lines = read_utf16(set_path)
file_lot = None
for line in file_lines:
if line.strip().startswith('LotPerBalance_step='):
parts = line.strip().split('||')
file_lot = parts[0].replace('LotPerBalance_step=', '').strip()
break
lot_value = prompt(f"LotPerBalance_step for {filename}", file_lot or "100")
else:
file_lines = read_utf16(set_path)
lot_value = None
for line in file_lines:
if line.strip().startswith('LotPerBalance_step='):
parts = line.strip().split('||')
lot_value = parts[0].replace('LotPerBalance_step=', '').strip()
break
if lot_value is None:
lot_value = prompt(f"LotPerBalance_step not found, enter value", "100")
print(f" Lots : Balance LotPerBalance_step={lot_value} Risk=9999")
# EA_Comment
model_label = MODEL_LABELS.get(cfg['model'], f"M{cfg['model']}")
ea_comment = f"{name_stem} {symbol} {period} {model_label}"
print(f" EA_Comment: {ea_comment}")
# Copy and modify set file
modified_set = os.path.join(out_set_folder, filename)
shutil.copy2(set_path, modified_set)
if lot_mode == 'asis':
update_set_file(modified_set, ea_comment, None, None)
else:
update_set_file(modified_set, ea_comment, lot_mode, lot_value)
# Write ini
model_label = MODEL_LABELS.get(cfg['model'], f"M{cfg['model']}")
report_name = f"{name_stem}_{model_label}"
ini_path = os.path.join(tester_folder, f"{name_stem}.ini")
build_ini(symbol, period, modified_set, ini_path, report_folder, cfg)
print(f" INI : {ini_path}")
print(f" Report as : {report_name}.htm")
# Launch MT5 minimised
cmd = [terminal_path, f'/config:{ini_path}']
print(f" Launching MT5", end='', flush=True)
si = subprocess.STARTUPINFO()
si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
si.wShowWindow = 6 # SW_MINIMIZE
proc = subprocess.Popen(cmd, startupinfo=si)
while proc.poll() is None:
time.sleep(10)
print('.', end='', flush=True)
print(f" done (exit {proc.returncode})")
# Copy report files from MT5 terminal folder
search_dirs = [
os.path.dirname(tester_folder),
os.path.join(os.path.dirname(tester_folder), 'MQL5', 'Profiles', 'Tester'),
tester_folder,
]
success = False
for search_dir in search_dirs:
htm_src = os.path.join(search_dir, report_name + '.htm')
if os.path.isfile(htm_src):
htm_dest = os.path.join(file_report_dir, report_name + '.htm')
shutil.copy2(htm_src, htm_dest)
copied = [report_name + '.htm']
for fn in os.listdir(search_dir):
if fn.startswith(report_name) and not fn.endswith('.htm'):
shutil.copy2(os.path.join(search_dir, fn),
os.path.join(file_report_dir, fn))
copied.append(fn)
# Remove from MT5 folder
os.remove(htm_src)
for fn in os.listdir(search_dir):
if fn.startswith(report_name) and not fn.endswith('.htm'):
try:
os.remove(os.path.join(search_dir, fn))
except:
pass
print(f" Report : {len(copied)} file(s) → {file_report_dir}")
success = True
break
if not success:
print(f" FAIL : Report not found. Checked:")
for d in search_dirs:
print(f" {d}")
results.append({
'file' : filename,
'subfolder': rel_subdir,
'symbol' : symbol,
'period' : period,
'success' : success,
})
# ── Summary ────────────────────────────────────────────────────
print()
print("=" * 60)
print(" SUMMARY")
print("=" * 60)
passed = [r for r in results if r['success']]
failed = [r for r in results if not r['success']]
print(f" Completed: {len(passed)}/{len(results)}")
if failed:
print(f"\n Failed (no report generated):")
for r in failed:
subfolder = f" [{r['subfolder']}]" if r.get('subfolder') else ""
print(f" - {r['file']}{subfolder} ({r['symbol']} {r['period']})")
failed_log = os.path.join(report_folder, 'failed_backtests.txt')
with open(failed_log, 'w') as f:
for r in failed:
f.write(f"{r['file']}\t{r['symbol']}\t{r['period']}\n")
print(f"\n Failed list saved to: {failed_log}")
print()
if __name__ == '__main__':
main()
+13
View File
@@ -0,0 +1,13 @@
{
"terminal_path": "C:\\Program Files\\MetaTrader 5\\terminal64.exe",
"tester_folder": "C:\\Users\\pc\\AppData\\Roaming\\MetaQuotes\\Terminal\\D0E8209F77C8CF37AD8BF550E51FF075\\Tester",
"terminal_label": "\ufffd\ufffdC\u0000:\u0000\\\u0000P\u0000r\u0000o\u0000g\u0000r\u0000a\u0000m\u0000 \u0000F\u0000i\u0000l\u0000e\u0000s\u0000\\\u0000M\u0000e\u0000t\u0000a\u0000T\u0000r\u0000a\u0000d\u0000e\u0000r\u0000 \u00005\u0000",
"ea_name": "Market\\Ultimate Breakout System.ex5",
"from_date": "2018.01.01",
"to_date": "2026.04.01",
"model": "1",
"deposit": "10000",
"currency": "USD",
"leverage": "100",
"suffix": ".a"
}
+389
View File
@@ -0,0 +1,389 @@
"""
mt5_parser.py
=============
Parsers for three MT5 trade report formats:
1. MT5 Real Account HTM export
2. MT5 Backtest HTM report
3. Quant Analyzer CSV export
All normalise to a common DataFrame schema.
"""
import pandas as pd
import re
# ── Common schema ─────────────────────────────────────────────────────────────
# open_time, close_time, symbol, type, volume, open_price, close_price,
# sl, tp, commission, swap, profit, net_profit, comment, strategy,
# duration_min, win, day_of_week, hour, source
def _decode(file_bytes):
for enc in ['utf-16', 'utf-8', 'latin-1', 'cp1252']:
try:
return file_bytes.decode(enc)
except:
continue
return ''
def _strip(s):
return re.sub(r'<[^>]+>', '', s).strip().replace('\xa0', '').replace('\u00a0', '')
def _to_float(s):
try:
return float(str(s).replace(' ', '').replace(',', ''))
except:
return None
def _to_dt(s, fmt='%Y.%m.%d %H:%M:%S'):
return pd.to_datetime(s, format=fmt, errors='coerce')
def _enrich(df):
"""Add derived columns common to all formats."""
df['open_time'] = pd.to_datetime(df['open_time'], errors='coerce')
df['close_time'] = pd.to_datetime(df['close_time'], errors='coerce')
df['open_date'] = df['open_time'].dt.date
df['close_date'] = df['close_time'].dt.date
df['day_of_week'] = df['open_time'].dt.day_name()
df['hour'] = df['open_time'].dt.hour
df['duration_min'] = ((df['close_time'] - df['open_time'])
.dt.total_seconds() / 60).round(1)
for col in ['volume', 'open_price', 'close_price', 'sl', 'tp',
'commission', 'swap', 'profit']:
if col in df.columns:
df[col] = pd.to_numeric(
df[col].astype(str).str.replace(' ', '').str.replace(',', ''),
errors='coerce'
)
if 'net_profit' not in df.columns:
df['net_profit'] = (
df.get('profit', 0).fillna(0) +
df.get('commission', 0).fillna(0) +
df.get('swap', 0).fillna(0)
)
df['win'] = df['net_profit'] > 0
df['type'] = df['type'].str.lower().str.strip()
df['strategy'] = df['comment'].apply(extract_strategy)
# Normalise symbol — strip .a suffix for display matching
df['symbol_base'] = df['symbol'].str.replace(r'\.[a-z]+$', '', regex=True).str.upper()
return df
# ── Format 1: Real Account HTM ────────────────────────────────────────────────
def parse_mt5_report(file_bytes):
"""Parse MT5 real account HTML trade history report."""
text = _decode(file_bytes)
rows = re.findall(r'<tr[^>]*>(.*?)</tr>', text, re.DOTALL)
trades = []
in_trades = False
COLS = ['open_time', 'position', 'symbol', 'type', 'comment', 'volume',
'open_price', 'sl', 'tp', 'close_time', 'close_price',
'commission', 'swap', 'profit']
for row in rows:
cells = re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row, re.DOTALL)
cells = [re.sub(r'\s+', ' ', _strip(c)).strip() for c in cells]
if cells and cells[0] == 'Time' and len(cells) >= 13:
in_trades = True
continue
if not in_trades:
continue
if cells and any(kw in cells[0] for kw in
['Total Net Profit', 'Results', 'Balance', 'Equity']):
break
if len(cells) >= 14 and re.match(r'\d{4}\.\d{2}\.\d{2}', cells[0]):
last = [c.lower() for c in cells if c]
if any(s in last for s in ['placed', 'cancelled', 'expired', 'partial']):
continue
if len(cells) < 10 or not re.match(r'\d{4}\.\d{2}\.\d{2}', cells[9]):
continue
if '/' in str(cells[5]):
continue
try:
trade = dict(zip(COLS, cells[:14]))
trades.append(trade)
except:
continue
if not trades:
return None
df = pd.DataFrame(trades)
df['source'] = 'real'
return _enrich(df)
# ── Format 2: Backtest HTM ────────────────────────────────────────────────────
def parse_backtest_report(file_bytes):
"""
Parse MT5 Strategy Tester HTML report.
Pairs in/out deals into complete trades.
"""
text = _decode(file_bytes)
tables = re.findall(r'<table[^>]*>(.*?)</table>', text, re.DOTALL)
if len(tables) < 2:
return None
rows = re.findall(r'<tr[^>]*>(.*?)</tr>', tables[1], re.DOTALL)
# Find deals section
in_deals = False
deal_rows = []
for row in rows:
cells = [re.sub(r'\s+', ' ', _strip(c)).strip()
for c in re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row, re.DOTALL)]
cells = [c for c in cells if c]
if not cells:
continue
if 'Deals' in cells:
in_deals = True
continue
if in_deals and cells[0] == 'Time' and 'Deal' in cells:
continue # header row
if in_deals and len(cells) >= 7 and re.match(r'\d{4}\.\d{2}\.\d{2}', cells[0]):
deal_rows.append(cells)
if not deal_rows:
return None
# Columns: Time, Deal, Symbol, Type, Direction, Volume, Price, Order,
# Commission, Swap, Profit, Balance, Comment
DEAL_COLS = ['time', 'deal', 'symbol', 'type', 'direction', 'volume',
'price', 'order', 'commission', 'swap', 'profit', 'balance', 'comment']
deals = []
for row in deal_rows:
d = dict(zip(DEAL_COLS, row[:len(DEAL_COLS)]))
deals.append(d)
df_deals = pd.DataFrame(deals)
df_deals = df_deals[df_deals['direction'].isin(['in', 'out'])]
# Match in/out pairs — pair consecutive in→out by symbol+type
open_stack = {} # key: (symbol, type_) -> list of open deals
trades = []
for _, deal in df_deals.iterrows():
sym = deal.get('symbol', '')
typ = deal.get('type', '')
dirn = deal.get('direction', '')
key = (sym, typ)
if dirn == 'in':
open_stack.setdefault(key, []).append(deal)
elif dirn == 'out':
stack = open_stack.get(key, [])
if stack:
entry = stack.pop(0)
trades.append({
'open_time' : entry['time'],
'close_time' : deal['time'],
'symbol' : sym,
'type' : typ,
'volume' : entry['volume'],
'open_price' : entry['price'],
'close_price': deal['price'],
'sl' : None,
'tp' : None,
'commission' : _to_float(entry.get('commission', 0)),
'swap' : _to_float(deal.get('swap', 0)),
'profit' : _to_float(deal.get('profit', 0)),
'comment' : deal.get('comment', ''),
'position' : entry.get('deal', ''),
})
if not trades:
return None
df = pd.DataFrame(trades)
df['source'] = 'backtest'
return _enrich(df)
# ── Format 3: Quant Analyzer CSV ─────────────────────────────────────────────
def parse_quant_csv(file_bytes):
"""Parse Quant Analyzer listOfTrades CSV export."""
try:
text = file_bytes.decode('utf-8-sig')
except:
text = file_bytes.decode('latin-1')
from io import StringIO
df_raw = pd.read_csv(StringIO(text))
# Normalise column names
df_raw.columns = [c.strip().lower().replace(' ', '_').replace('/', '_').replace('(', '').replace(')', '') for c in df_raw.columns]
col_map = {
'open_time' : ['open_time_$', 'open_time_$_', 'open_time', 'opentime'],
'close_time' : ['close_time_$', 'close_time_$_', 'close_time', 'closetime'],
'symbol' : ['symbol_$', 'symbol_$_', 'symbol'],
'type' : ['type_$', 'type_$_', 'type', 'direction'],
'volume' : ['size_$', 'size_$_', 'size', 'volume', 'lots'],
'open_price' : ['open_price_$', 'open_price_$_', 'open_price', 'openprice'],
'close_price' : ['close_price_$', 'close_price_$_', 'close_price', 'closeprice'],
'profit' : ['profit_loss_$', 'profit_loss_$_', 'profit_loss', 'profit', 'net_profit'],
'commission' : ['comm_swap_$', 'comm_swap_$_', 'commission', 'comm'],
'swap' : ['swap_$', 'swap'],
'sl' : ['stop_loss_$', 'stop_loss_$_', 'stop_loss', 'sl'],
'comment' : ['comment_$', 'comment_$_', 'comment'],
'strategy' : ['strategy_name_$', 'strategy_name_$_', 'strategy_name', 'strategy'],
'mae' : ['mae_$', 'mae_$_', 'mae'],
'mfe' : ['mfe_$', 'mfe_$_', 'mfe'],
'drawdown' : ['drawdown_$', 'drawdown_$_', 'drawdown'],
}
result = {}
for target, candidates in col_map.items():
for cand in candidates:
if cand in df_raw.columns:
result[target] = df_raw[cand]
break
df = pd.DataFrame(result)
# Parse datetimes — QA uses DD.MM.YYYY HH:MM:SS
for col in ['open_time', 'close_time']:
if col in df.columns:
df[col] = pd.to_datetime(df[col], format='%d.%m.%Y %H:%M:%S', errors='coerce')
if df[col].isna().all():
df[col] = pd.to_datetime(df[col], infer_datetime_format=True, errors='coerce')
# QA comm_swap is combined — split evenly as approximation if no separate swap
if 'commission' in df.columns and 'swap' not in df.columns:
df['swap'] = 0.0
if 'sl' not in df.columns:
df['sl'] = None
if 'tp' not in df.columns:
df['tp'] = None
if 'position' not in df.columns:
df['position'] = df.get('ticket', range(len(df)))
df['source'] = 'quant_csv'
# Add extra QA-specific columns if present
for extra in ['mae', 'mfe', 'drawdown']:
if extra in df_raw.columns:
df[extra] = pd.to_numeric(df_raw[extra], errors='coerce')
return _enrich(df)
# ── Auto-detect format ────────────────────────────────────────────────────────
def detect_and_parse(file_bytes, filename=''):
"""
Auto-detect file format and parse.
Returns (df, format_name) or (None, None).
"""
fname = filename.lower()
if fname.endswith('.csv'):
df = parse_quant_csv(file_bytes)
return df, 'Quant Analyzer CSV'
# HTML/HTM — detect backtest vs real account
try:
text = _decode(file_bytes)
except:
return None, None
if 'Strategy Tester Report' in text or 'strategy tester' in text.lower():
df = parse_backtest_report(file_bytes)
return df, 'MT5 Backtest Report'
df = parse_mt5_report(file_bytes)
return df, 'MT5 Account History'
# ── Stats ─────────────────────────────────────────────────────────────────────
def calc_stats(df):
if df is None or len(df) == 0:
return {}
total = len(df)
wins = df[df['win'] == True]
losses = df[df['win'] == False]
win_rate = round(len(wins) / total * 100, 1) if total > 0 else 0
gross_profit = round(wins['net_profit'].sum(), 2)
gross_loss = round(losses['net_profit'].sum(), 2)
net_profit = round(df['net_profit'].sum(), 2)
pf = round(abs(gross_profit / gross_loss), 2) if gross_loss != 0 else float('inf')
avg_win = round(wins['net_profit'].mean(), 2) if len(wins) > 0 else 0
avg_loss = round(losses['net_profit'].mean(), 2) if len(losses) > 0 else 0
rr = round(abs(avg_win / avg_loss), 2) if avg_loss != 0 else float('inf')
expectancy = round((win_rate/100 * avg_win) + ((1 - win_rate/100) * avg_loss), 2)
results = df.sort_values('close_time')['win'].tolist()
max_cw = _max_consec(results, True)
max_cl = _max_consec(results, False)
cumulative = df.sort_values('close_time')['net_profit'].cumsum()
rolling_max = cumulative.cummax()
max_dd = round((cumulative - rolling_max).min(), 2)
avg_dur = round(df['duration_min'].mean(), 1) if 'duration_min' in df.columns else 0
avg_win_dur = round(wins['duration_min'].mean(), 1) if len(wins) > 0 else 0
avg_los_dur = round(losses['duration_min'].mean(), 1) if len(losses) > 0 else 0
longs = df[df['type'] == 'buy']
shorts = df[df['type'] == 'sell']
return {
'total_trades' : total,
'win_rate' : win_rate,
'net_profit' : net_profit,
'gross_profit' : gross_profit,
'gross_loss' : gross_loss,
'profit_factor' : pf,
'avg_win' : avg_win,
'avg_loss' : avg_loss,
'rr_ratio' : rr,
'expectancy' : expectancy,
'max_consec_wins' : max_cw,
'max_consec_losses' : max_cl,
'max_drawdown' : max_dd,
'best_trade' : round(df['net_profit'].max(), 2),
'worst_trade' : round(df['net_profit'].min(), 2),
'avg_duration_min' : avg_dur,
'avg_win_duration' : avg_win_dur,
'avg_loss_duration' : avg_los_dur,
'long_trades' : len(longs),
'short_trades' : len(shorts),
'long_win_rate' : round(len(longs[longs['win']]) / len(longs) * 100, 1) if len(longs) > 0 else 0,
'short_win_rate' : round(len(shorts[shorts['win']]) / len(shorts) * 100, 1) if len(shorts) > 0 else 0,
}
def extract_strategy(comment):
if not comment or str(comment).strip() == '':
return 'Manual'
parts = str(comment).split('_')
while parts and re.match(r'^\d+$', parts[-1]):
parts.pop()
if parts and re.match(r'^[A-Z]{3,8}(\.a)?$', parts[-1]):
parts.pop()
return '_'.join(parts) if parts else str(comment)
def _max_consec(results, target):
max_c = cur_c = 0
for r in results:
if r == target:
cur_c += 1
max_c = max(max_c, cur_c)
else:
cur_c = 0
return max_c
+4
View File
@@ -0,0 +1,4 @@
streamlit
streamlit-option-menu
pandas
plotly
+94
View File
@@ -0,0 +1,94 @@
import io
import zipfile
from pathlib import Path
def parse_set_file(file_bytes, filename):
"""Parse a .set file and return dict of {param: value} and ordered param list"""
text = file_bytes.decode('utf-16')
params = {}
raw_lines = {} # preserve full line for export
order = []
for line in text.splitlines():
line = line.strip()
if not line or line.startswith(';'):
continue
if '=' not in line:
continue
key, _, rest = line.partition('=')
key = key.strip()
# Value is first field before ||
parts = rest.split('||')
value = parts[0].strip()
params[key] = value
raw_lines[key] = rest # preserve everything after =
order.append(key)
return params, raw_lines, order
def build_comparison_df(files_data):
"""
files_data: list of (filename, params, raw_lines, order)
Returns DataFrame with param names as index, filenames as columns
"""
import pandas as pd
# Build union of all param keys preserving order from first file
all_keys = []
seen = set()
for _, _, _, order in files_data:
for k in order:
if k not in seen:
all_keys.append(k)
seen.add(k)
# Build dataframe
rows = []
for key in all_keys:
row = {'Parameter': key}
for filename, params, _, _ in files_data:
row[filename] = params.get(key, '')
rows.append(row)
return pd.DataFrame(rows)
def export_set_file(filename, params_edited, raw_lines, order, original_bytes):
"""
Rebuild .set file with edited values, preserving || fields
Returns bytes (utf-16 encoded)
"""
# Get original header comments
text = original_bytes.decode('utf-16')
lines = text.splitlines()
header_lines = []
for line in lines:
if line.startswith(';'):
header_lines.append(line)
else:
break
output_lines = header_lines.copy()
for key in order:
if key not in params_edited:
continue
new_value = params_edited[key]
rest = raw_lines.get(key, new_value)
parts = rest.split('||')
parts[0] = str(new_value)
output_lines.append(f"{key}={'||'.join(parts)}")
content = '\r\n'.join(output_lines) + '\r\n'
return content.encode('utf-16')
def create_zip(files_export):
"""
files_export: list of (filename, bytes)
Returns zip bytes
"""
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
for fname, fbytes in files_export:
zf.writestr(fname, fbytes)
buf.seek(0)
return buf.read()
+37
View File
@@ -0,0 +1,37 @@
"""
pages/settings.py
=================
Settings page for MT5 Tools dashboard.
"""
import streamlit as st
def render():
st.title("⚙️ Settings")
st.markdown("""
<div class="info-card">
MT5 Tools — settings and information.
</div>
""", unsafe_allow_html=True)
st.subheader("About")
st.markdown("""
**MT5 Tools** is a standalone trade analysis and comparison dashboard.
**Supported file formats:**
- MT5 Account History HTML export (`.htm` / `.html`)
- MT5 Strategy Tester Backtest Report (`.htm` / `.html`)
- Quant Analyzer CSV export (`listOfTrades_*.csv`)
**Pages:**
- **Trade Analysis** — statistics, equity curves, day/hour breakdown for a single report
- **Trade Compare** — match and compare two reports to measure slippage and variance
""")
st.subheader("Requirements")
st.code("pip install streamlit streamlit-option-menu pandas plotly", language="bash")
st.subheader("Launch")
st.code("streamlit run app.py", language="bash")
+333
View File
@@ -0,0 +1,333 @@
"""
pages/trade_analysis.py
=======================
MT5 Trade Analysis page — migrated from main dashboard.
"""
import streamlit as st
import plotly.graph_objects as go
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from mt5_parser import detect_and_parse, calc_stats
def render():
st.title("📊 Trade Analysis")
# ── Session state ─────────────────────────────────────────────────────────
if 'ta_df' not in st.session_state:
st.session_state['ta_df'] = None
st.session_state['ta_format'] = None
# ── File upload ───────────────────────────────────────────────────────────
col1, col2 = st.columns([4, 1])
with col1:
uploaded = st.file_uploader(
"Upload MT5 Report (HTM/HTML) or Quant Analyzer CSV",
type=['html', 'htm', 'csv'],
key='ta_upload'
)
with col2:
st.markdown("<br>", unsafe_allow_html=True)
if st.button("🗑 Clear", key='ta_clear'):
st.session_state['ta_df'] = None
st.session_state['ta_format'] = None
st.rerun()
if uploaded:
df, fmt = detect_and_parse(uploaded.read(), uploaded.name)
if df is not None:
st.session_state['ta_df'] = df
st.session_state['ta_format'] = fmt
st.success(f"✓ Loaded {len(df)} trades — {fmt}")
else:
st.error("Could not parse report — check file format")
df_all = st.session_state['ta_df']
fmt = st.session_state['ta_format']
if df_all is None or len(df_all) == 0:
st.markdown("""
<div class="info-card">
Upload an MT5 account history report (.htm/.html), MT5 backtest report,
or a Quant Analyzer CSV export to begin analysis.
</div>
""", unsafe_allow_html=True)
return
if fmt:
st.caption(f"Format detected: **{fmt}** · {len(df_all)} total trades")
# ── Filters ───────────────────────────────────────────────────────────────
st.divider()
fc1, fc2, fc3, fc4 = st.columns(4)
with fc1:
date_min = df_all['open_time'].min().date()
date_max = df_all['open_time'].max().date()
date_from = st.date_input("From", value=date_min, min_value=date_min,
max_value=date_max, key='ta_from')
date_to = st.date_input("To", value=date_max, min_value=date_min,
max_value=date_max, key='ta_to')
with fc2:
symbols = sorted(df_all['symbol'].dropna().unique().tolist())
sel_symbol = st.multiselect("Symbol", symbols, key='ta_sym')
with fc3:
strategies = sorted(df_all['strategy'].dropna().unique().tolist())
sel_strategy = st.multiselect("Strategy / EA", strategies, key='ta_strat')
with fc4:
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
sel_days = st.multiselect("Day of week", days, key='ta_days')
sel_type = st.multiselect("Type", ['buy', 'sell'], key='ta_type')
# Apply filters
df = df_all.copy()
df = df[(df['open_time'].dt.date >= date_from) &
(df['open_time'].dt.date <= date_to)]
if sel_symbol:
df = df[df['symbol'].isin(sel_symbol)]
if sel_strategy:
df = df[df['strategy'].isin(sel_strategy)]
if sel_days:
df = df[df['day_of_week'].isin(sel_days)]
if sel_type:
df = df[df['type'].isin(sel_type)]
st.caption(f"Showing **{len(df)}** trades after filters")
# ── Analysis mode ─────────────────────────────────────────────────────────
mode = st.radio(
"Analysis mode",
["Overall", "By Strategy", "By Symbol", "By Day of Week"],
horizontal=True, key='ta_mode'
)
st.divider()
# ── Helpers ───────────────────────────────────────────────────────────────
def render_stats(stats, label=""):
if label:
st.markdown(f"**{label}**")
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("Net Profit", f"${stats['net_profit']:,.2f}")
c2.metric("Win Rate", f"{stats['win_rate']}%")
c3.metric("Profit Factor", f"{stats['profit_factor']}")
c4.metric("R:R Ratio", f"{stats['rr_ratio']}")
c5.metric("Expectancy", f"${stats['expectancy']:,.2f}")
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("Total Trades", stats['total_trades'])
c2.metric("Avg Win", f"${stats['avg_win']:,.2f}")
c3.metric("Avg Loss", f"${stats['avg_loss']:,.2f}")
c4.metric("Max DD", f"${stats['max_drawdown']:,.2f}")
c5.metric("Best Trade", f"${stats['best_trade']:,.2f}")
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("Max Consec Wins", stats['max_consec_wins'])
c2.metric("Max Consec Losses", stats['max_consec_losses'])
c3.metric("Avg Win Dur", f"{stats['avg_win_duration']}m")
c4.metric("Avg Loss Dur", f"{stats['avg_loss_duration']}m")
c5.metric("Worst Trade", f"${stats['worst_trade']:,.2f}")
c1, c2, c3, c4 = st.columns(4)
c1.metric("Long Trades", stats['long_trades'])
c2.metric("Long Win Rate", f"{stats['long_win_rate']}%")
c3.metric("Short Trades", stats['short_trades'])
c4.metric("Short Win Rate",f"{stats['short_win_rate']}%")
def render_equity_curve(df_plot, label="Equity Curve"):
df_s = df_plot.sort_values('close_time').copy()
df_s['cumulative'] = df_s['net_profit'].cumsum()
fig = go.Figure()
fig.add_trace(go.Scatter(
x=df_s['close_time'], y=df_s['cumulative'],
mode='lines',
line=dict(color='#7c6af7', width=2),
fill='tozeroy',
fillcolor='rgba(124,106,247,0.08)',
name='Equity'
))
fig.update_layout(
title=label, height=300,
plot_bgcolor='rgba(10,10,15,1)',
paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa', family='JetBrains Mono'),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
margin=dict(l=60, r=20, t=40, b=40)
)
st.plotly_chart(fig, use_container_width=True)
def render_dow_chart(df_plot):
dow_order = ['Monday','Tuesday','Wednesday','Thursday','Friday','Saturday','Sunday']
dow = df_plot.groupby('day_of_week').agg(
trades = ('net_profit', 'count'),
net_profit = ('net_profit', 'sum'),
win_rate = ('win', lambda x: round(x.mean()*100, 1))
).reindex([d for d in dow_order if d in df_plot['day_of_week'].unique()])
wins_dow = df_plot[df_plot['win']].groupby('day_of_week')['net_profit'].sum().reindex(dow.index, fill_value=0)
losses_dow = df_plot[~df_plot['win']].groupby('day_of_week')['net_profit'].sum().reindex(dow.index, fill_value=0)
fig = go.Figure()
fig.add_trace(go.Bar(x=dow.index, y=wins_dow, name='Profit', marker_color='rgba(45,198,83,0.8)'))
fig.add_trace(go.Bar(x=dow.index, y=losses_dow, name='Loss', marker_color='rgba(230,57,70,0.8)'))
fig.update_layout(
title='P&L by Day of Week', height=280, barmode='relative',
plot_bgcolor='rgba(10,10,15,1)', paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa'), margin=dict(l=60, r=20, t=40, b=40),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
legend=dict(bgcolor='rgba(0,0,0,0.3)')
)
st.plotly_chart(fig, use_container_width=True)
dt = dow.reset_index()
dt.columns = ['Day', 'Trades', 'Net Profit', 'Win Rate %']
dt['Net Profit'] = dt['Net Profit'].round(2)
st.dataframe(dt, use_container_width=True, hide_index=True)
def render_hour_chart(df_plot):
hourly = df_plot.groupby('hour').agg(
trades = ('net_profit', 'count'),
net_profit = ('net_profit', 'sum'),
)
wins_h = df_plot[df_plot['win']].groupby('hour')['net_profit'].sum().reindex(hourly.index, fill_value=0)
losses_h = df_plot[~df_plot['win']].groupby('hour')['net_profit'].sum().reindex(hourly.index, fill_value=0)
fig = go.Figure()
fig.add_trace(go.Bar(x=wins_h.index, y=wins_h, name='Profit', marker_color='rgba(45,198,83,0.8)'))
fig.add_trace(go.Bar(x=losses_h.index, y=losses_h, name='Loss', marker_color='rgba(230,57,70,0.8)'))
fig.update_layout(
title='P&L by Hour of Day', height=280, barmode='relative',
plot_bgcolor='rgba(10,10,15,1)', paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa'), margin=dict(l=60, r=20, t=40, b=40),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)', title='Hour (UTC)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
legend=dict(bgcolor='rgba(0,0,0,0.3)')
)
st.plotly_chart(fig, use_container_width=True)
def colour_profit(val):
try:
v = float(str(val).replace(',', ''))
if v > 0: return 'background-color: rgba(0,180,0,0.12)'
if v < 0: return 'background-color: rgba(180,0,0,0.12)'
except:
pass
return ''
# ── Render mode ───────────────────────────────────────────────────────────
if mode == "Overall":
stats = calc_stats(df)
render_stats(stats, "Overall Statistics")
render_equity_curve(df)
col1, col2 = st.columns(2)
with col1:
render_dow_chart(df)
with col2:
render_hour_chart(df)
elif mode == "By Strategy":
strats = sorted(df['strategy'].dropna().unique().tolist())
if not strats:
st.info("No strategies found")
else:
st.subheader("Strategy Comparison")
rows = []
for s in strats:
sdf = df[df['strategy'] == s]
stat = calc_stats(sdf)
rows.append({
'Strategy' : s,
'Trades' : stat['total_trades'],
'Net Profit' : stat['net_profit'],
'Win Rate %' : stat['win_rate'],
'Profit Factor' : stat['profit_factor'],
'R:R' : stat['rr_ratio'],
'Expectancy' : stat['expectancy'],
'Max DD' : stat['max_drawdown'],
'Max Consec W' : stat['max_consec_wins'],
'Max Consec L' : stat['max_consec_losses'],
})
sdf_sum = __import__('pandas').DataFrame(rows).sort_values('Net Profit', ascending=False)
st.dataframe(
sdf_sum.style.map(colour_profit, subset=['Net Profit', 'Expectancy', 'Max DD']),
use_container_width=True, hide_index=True
)
st.divider()
sel = st.selectbox("Select strategy for detail", strats)
if sel:
sdf = df[df['strategy'] == sel]
stat = calc_stats(sdf)
render_stats(stat, sel)
render_equity_curve(sdf, f"{sel} — Equity Curve")
col1, col2 = st.columns(2)
with col1: render_dow_chart(sdf)
with col2: render_hour_chart(sdf)
elif mode == "By Symbol":
syms = sorted(df['symbol'].dropna().unique().tolist())
rows = []
for s in syms:
sdf = df[df['symbol'] == s]
stat = calc_stats(sdf)
rows.append({
'Symbol' : s,
'Trades' : stat['total_trades'],
'Net Profit' : stat['net_profit'],
'Win Rate %' : stat['win_rate'],
'Profit Factor' : stat['profit_factor'],
'R:R' : stat['rr_ratio'],
'Expectancy' : stat['expectancy'],
'Max DD' : stat['max_drawdown'],
})
sdf_sum = __import__('pandas').DataFrame(rows).sort_values('Net Profit', ascending=False)
st.dataframe(
sdf_sum.style.map(colour_profit, subset=['Net Profit', 'Expectancy', 'Max DD']),
use_container_width=True, hide_index=True
)
sel = st.selectbox("Select symbol for detail", syms)
if sel:
sdf = df[df['symbol'] == sel]
stat = calc_stats(sdf)
render_stats(stat, sel)
render_equity_curve(sdf, f"{sel} — Equity Curve")
col1, col2 = st.columns(2)
with col1: render_dow_chart(sdf)
with col2: render_hour_chart(sdf)
elif mode == "By Day of Week":
render_dow_chart(df)
render_hour_chart(df)
# ── Raw trade log ─────────────────────────────────────────────────────────
st.divider()
with st.expander("Raw Trade Log"):
show_cols = ['open_time', 'close_time', 'symbol', 'type', 'strategy',
'volume', 'open_price', 'close_price', 'sl', 'tp',
'commission', 'swap', 'profit', 'net_profit', 'duration_min']
show_cols = [c for c in show_cols if c in df.columns]
def colour_net(val):
try:
v = float(val)
if v > 0: return 'background-color: rgba(0,180,0,0.12)'
if v < 0: return 'background-color: rgba(180,0,0,0.12)'
except:
pass
return ''
st.dataframe(
df[show_cols].style.map(colour_net, subset=['net_profit', 'profit']),
use_container_width=True, hide_index=True, height=400
)
st.download_button(
"⬇ Download filtered trades CSV",
data = df[show_cols].to_csv(index=False),
file_name = f"mt5_trades_{date_from}_{date_to}.csv",
mime = 'text/csv'
)
+422
View File
@@ -0,0 +1,422 @@
"""
pages/trade_compare.py
======================
Side-by-side comparison of two trade history files.
Matches trades by symbol + type + open time within a tolerance window.
Highlights slippage, profit variance, and timing differences.
"""
import streamlit as st
import pandas as pd
import plotly.graph_objects as go
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from mt5_parser import detect_and_parse, calc_stats
# ── Match trades ──────────────────────────────────────────────────────────────
def match_trades(df_a, df_b, tolerance_hours):
"""
Match trades between two DataFrames.
Match criteria: same symbol_base + same type + open_time within tolerance.
Returns DataFrame of matched pairs with diff columns.
"""
tol = pd.Timedelta(hours=tolerance_hours)
matched = []
used_b = set()
for i, a in df_a.iterrows():
best_match = None
best_delta = tol + pd.Timedelta(seconds=1)
for j, b in df_b.iterrows():
if j in used_b:
continue
if a['symbol_base'] != b['symbol_base']:
continue
if a['type'] != b['type']:
continue
delta = abs(a['open_time'] - b['open_time'])
if delta <= tol and delta < best_delta:
best_delta = delta
best_match = (j, b)
if best_match:
j, b = best_match
used_b.add(j)
open_slip = round(float(b['open_price']) - float(a['open_price']), 5) if pd.notna(a['open_price']) and pd.notna(b['open_price']) else None
close_slip = round(float(b['close_price']) - float(a['close_price']), 5) if pd.notna(a['close_price']) and pd.notna(b['close_price']) else None
profit_var = round(float(b['net_profit']) - float(a['net_profit']), 2) if pd.notna(a['net_profit']) and pd.notna(b['net_profit']) else None
time_diff = round((b['open_time'] - a['open_time']).total_seconds() / 60, 1)
dur_diff = round(float(b.get('duration_min', 0) or 0) - float(a.get('duration_min', 0) or 0), 1)
matched.append({
# File A
'A_open_time' : a['open_time'],
'A_close_time' : a['close_time'],
'A_symbol' : a['symbol'],
'A_type' : a['type'],
'A_volume' : a.get('volume'),
'A_open_price' : a.get('open_price'),
'A_close_price': a.get('close_price'),
'A_profit' : a.get('net_profit'),
'A_duration' : a.get('duration_min'),
# File B
'B_open_time' : b['open_time'],
'B_close_time' : b['close_time'],
'B_symbol' : b['symbol'],
'B_type' : b['type'],
'B_volume' : b.get('volume'),
'B_open_price' : b.get('open_price'),
'B_close_price': b.get('close_price'),
'B_profit' : b.get('net_profit'),
'B_duration' : b.get('duration_min'),
# Differences
'open_slippage' : open_slip,
'close_slippage': close_slip,
'profit_var' : profit_var,
'time_diff_min' : time_diff,
'duration_diff' : dur_diff,
})
return pd.DataFrame(matched)
# ── Render ────────────────────────────────────────────────────────────────────
def render():
st.title("🔄 Trade Compare")
st.markdown("""
<div class="info-card">
Compare two trade history files — backtest vs real account, or any two exports.
Trades are matched by symbol, direction, and open time within a configurable
tolerance window to account for gaps, slippage, and market open variations.
</div>
""", unsafe_allow_html=True)
# ── Session state ─────────────────────────────────────────────────────────
for k in ['tc_df_a', 'tc_df_b', 'tc_fmt_a', 'tc_fmt_b']:
if k not in st.session_state:
st.session_state[k] = None
# ── File upload ───────────────────────────────────────────────────────────
st.subheader("Load Files")
col_a, col_b = st.columns(2)
with col_a:
st.markdown("**File A** — Reference (e.g. Backtest)")
up_a = st.file_uploader("Upload File A", type=['html','htm','csv'], key='tc_up_a')
if up_a:
df_a, fmt_a = detect_and_parse(up_a.read(), up_a.name)
if df_a is not None:
st.session_state['tc_df_a'] = df_a
st.session_state['tc_fmt_a'] = fmt_a
st.success(f"{len(df_a)} trades — {fmt_a}")
else:
st.error("Could not parse File A")
if st.session_state['tc_df_a'] is not None:
st.caption(f"Loaded: **{st.session_state['tc_fmt_a']}** · {len(st.session_state['tc_df_a'])} trades")
with col_b:
st.markdown("**File B** — Comparison (e.g. Real Account)")
up_b = st.file_uploader("Upload File B", type=['html','htm','csv'], key='tc_up_b')
if up_b:
df_b, fmt_b = detect_and_parse(up_b.read(), up_b.name)
if df_b is not None:
st.session_state['tc_df_b'] = df_b
st.session_state['tc_fmt_b'] = fmt_b
st.success(f"{len(df_b)} trades — {fmt_b}")
else:
st.error("Could not parse File B")
if st.session_state['tc_df_b'] is not None:
st.caption(f"Loaded: **{st.session_state['tc_fmt_b']}** · {len(st.session_state['tc_df_b'])} trades")
df_a = st.session_state['tc_df_a']
df_b = st.session_state['tc_df_b']
if df_a is None or df_b is None:
return
# ── Filters ───────────────────────────────────────────────────────────────
st.divider()
st.subheader("Filters")
fa1, fa2, fa3 = st.columns(3)
fb1, fb2, fb3 = st.columns(3)
with fa1:
st.markdown("**File A filters**")
with fb1:
st.markdown("**File B filters**")
col1, col2, col3, col4, col5, col6 = st.columns(6)
with col1:
a_date_min = df_a['open_time'].min().date()
a_date_max = df_a['open_time'].max().date()
a_from = st.date_input("A — From", value=a_date_min, min_value=a_date_min,
max_value=a_date_max, key='tc_a_from')
a_to = st.date_input("A — To", value=a_date_max, min_value=a_date_min,
max_value=a_date_max, key='tc_a_to')
with col2:
a_syms = sorted(df_a['symbol'].dropna().unique().tolist())
a_sel_sym = st.multiselect("A — Symbol", a_syms, key='tc_a_sym')
with col3:
a_strats = sorted(df_a['strategy'].dropna().unique().tolist())
a_sel_strat = st.multiselect("A — Strategy", a_strats, key='tc_a_strat')
a_sel_type = st.multiselect("A — Type", ['buy', 'sell'], key='tc_a_type')
with col4:
b_date_min = df_b['open_time'].min().date()
b_date_max = df_b['open_time'].max().date()
b_from = st.date_input("B — From", value=b_date_min, min_value=b_date_min,
max_value=b_date_max, key='tc_b_from')
b_to = st.date_input("B — To", value=b_date_max, min_value=b_date_min,
max_value=b_date_max, key='tc_b_to')
with col5:
b_syms = sorted(df_b['symbol'].dropna().unique().tolist())
b_sel_sym = st.multiselect("B — Symbol", b_syms, key='tc_b_sym')
with col6:
b_strats = sorted(df_b['strategy'].dropna().unique().tolist())
b_sel_strat = st.multiselect("B — Strategy", b_strats, key='tc_b_strat')
b_sel_type = st.multiselect("B — Type", ['buy', 'sell'], key='tc_b_type')
# ── Matching tolerance ────────────────────────────────────────────────────
st.divider()
col_tol, col_run = st.columns([3, 1])
with col_tol:
tolerance = st.slider(
"Match tolerance (hours) — max time difference between A and B open times",
min_value=1, max_value=24, value=4, step=1,
help="Trades within this window are considered the same setup. "
"Increase for daily charts, decrease for intraday."
)
with col_run:
st.markdown("<br>", unsafe_allow_html=True)
run = st.button("🔍 Match Trades", type="primary", use_container_width=True)
if not run and 'tc_matched' not in st.session_state:
return
# Apply filters
fa = df_a.copy()
fa = fa[(fa['open_time'].dt.date >= a_from) & (fa['open_time'].dt.date <= a_to)]
if a_sel_sym: fa = fa[fa['symbol'].isin(a_sel_sym)]
if a_sel_strat: fa = fa[fa['strategy'].isin(a_sel_strat)]
if a_sel_type: fa = fa[fa['type'].isin(a_sel_type)]
fb = df_b.copy()
fb = fb[(fb['open_time'].dt.date >= b_from) & (fb['open_time'].dt.date <= b_to)]
if b_sel_sym: fb = fb[fb['symbol'].isin(b_sel_sym)]
if b_sel_strat: fb = fb[fb['strategy'].isin(b_sel_strat)]
if b_sel_type: fb = fb[fb['type'].isin(b_sel_type)]
if run:
with st.spinner("Matching trades..."):
matched = match_trades(fa, fb, tolerance)
st.session_state['tc_matched'] = matched
st.session_state['tc_fa_len'] = len(fa)
st.session_state['tc_fb_len'] = len(fb)
matched = st.session_state.get('tc_matched', pd.DataFrame())
fa_len = st.session_state.get('tc_fa_len', len(fa))
fb_len = st.session_state.get('tc_fb_len', len(fb))
if matched is None or len(matched) == 0:
st.warning("No matching trades found — try increasing the tolerance window or adjusting filters.")
return
# ── Summary stats ─────────────────────────────────────────────────────────
st.divider()
st.subheader("Match Summary")
m1, m2, m3, m4, m5 = st.columns(5)
m1.metric("File A Trades", fa_len)
m2.metric("File B Trades", fb_len)
m3.metric("Matched Pairs", len(matched))
m4.metric("Unmatched A", fa_len - len(matched))
m5.metric("Unmatched B", fb_len - len(matched))
st.divider()
# ── Aggregate comparison ───────────────────────────────────────────────────
st.subheader("Aggregate Comparison")
ac1, ac2 = st.columns(2)
with ac1:
st.markdown("**File A (Reference)**")
a_net = matched['A_profit'].sum()
a_wr = (matched['A_profit'] > 0).mean() * 100
a_avg = matched['A_profit'].mean()
a_dur = matched['A_duration'].mean() if 'A_duration' in matched else None
st.metric("Net Profit", f"${a_net:,.2f}")
st.metric("Win Rate", f"{a_wr:.1f}%")
st.metric("Avg Profit", f"${a_avg:,.2f}")
if a_dur:
st.metric("Avg Duration", f"{a_dur:.0f}m")
with ac2:
st.markdown("**File B (Comparison)**")
b_net = matched['B_profit'].sum()
b_wr = (matched['B_profit'] > 0).mean() * 100
b_avg = matched['B_profit'].mean()
b_dur = matched['B_duration'].mean() if 'B_duration' in matched else None
delta_net = b_net - a_net
st.metric("Net Profit", f"${b_net:,.2f}",
delta=f"{delta_net:+.2f}", delta_color="normal")
st.metric("Win Rate", f"{b_wr:.1f}%",
delta=f"{b_wr - a_wr:+.1f}%", delta_color="normal")
st.metric("Avg Profit", f"${b_avg:,.2f}",
delta=f"{b_avg - a_avg:+.2f}", delta_color="normal")
if b_dur and a_dur:
st.metric("Avg Duration", f"{b_dur:.0f}m",
delta=f"{b_dur - a_dur:+.0f}m", delta_color="off")
# ── Slippage summary ───────────────────────────────────────────────────────
st.divider()
st.subheader("Slippage & Variance Summary")
sc1, sc2, sc3, sc4 = st.columns(4)
avg_open_slip = matched['open_slippage'].mean()
avg_close_slip = matched['close_slip'].mean() if 'close_slip' in matched else matched['close_slippage'].mean()
avg_profit_var = matched['profit_var'].mean()
avg_time_diff = matched['time_diff_min'].mean()
sc1.metric("Avg Entry Slippage", f"{avg_open_slip:+.5f}" if pd.notna(avg_open_slip) else "N/A",
help="B open price minus A open price. Positive = B filled higher.")
sc2.metric("Avg Exit Slippage", f"{avg_close_slip:+.5f}" if pd.notna(avg_close_slip) else "N/A",
help="B close price minus A close price.")
sc3.metric("Avg Profit Variance", f"${avg_profit_var:+.2f}" if pd.notna(avg_profit_var) else "N/A",
help="B net profit minus A net profit per trade.")
sc4.metric("Avg Time Difference", f"{avg_time_diff:+.0f}m" if pd.notna(avg_time_diff) else "N/A",
help="B open time minus A open time in minutes.")
# ── Equity curve overlay ───────────────────────────────────────────────────
st.divider()
st.subheader("Equity Curve Overlay")
m_sorted = matched.sort_values('A_open_time')
fig = go.Figure()
fig.add_trace(go.Scatter(
x=m_sorted['A_open_time'],
y=m_sorted['A_profit'].cumsum(),
mode='lines', name='File A',
line=dict(color='#7c6af7', width=2),
fill='tozeroy', fillcolor='rgba(124,106,247,0.05)'
))
fig.add_trace(go.Scatter(
x=m_sorted['B_open_time'],
y=m_sorted['B_profit'].cumsum(),
mode='lines', name='File B',
line=dict(color='#2dc653', width=2),
fill='tozeroy', fillcolor='rgba(45,198,83,0.05)'
))
fig.update_layout(
height=320,
plot_bgcolor='rgba(10,10,15,1)',
paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa', family='JetBrains Mono'),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
legend=dict(bgcolor='rgba(0,0,0,0.3)'),
margin=dict(l=60, r=20, t=20, b=40)
)
st.plotly_chart(fig, use_container_width=True)
# ── Profit variance scatter ────────────────────────────────────────────────
st.subheader("Profit Variance per Trade")
fig2 = go.Figure()
colours = matched['profit_var'].apply(
lambda v: 'rgba(45,198,83,0.7)' if v >= 0 else 'rgba(230,57,70,0.7)'
)
fig2.add_trace(go.Bar(
x=list(range(len(matched))),
y=matched['profit_var'],
marker_color=colours,
name='Profit Variance (B - A)'
))
fig2.update_layout(
height=250,
plot_bgcolor='rgba(10,10,15,1)',
paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa'),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)', title='Trade #'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
margin=dict(l=60, r=20, t=20, b=40)
)
st.plotly_chart(fig2, use_container_width=True)
# ── Matched trade table ────────────────────────────────────────────────────
st.divider()
st.subheader("Matched Trade Detail")
def colour_diff(val):
try:
v = float(str(val).replace('+', ''))
if v > 0: return 'color: #2dc653; font-weight: 600'
if v < 0: return 'color: #e63946; font-weight: 600'
except:
pass
return 'color: #666'
def colour_profit_cell(val):
try:
v = float(str(val).replace(',', ''))
if v > 0: return 'background-color: rgba(0,180,0,0.10)'
if v < 0: return 'background-color: rgba(180,0,0,0.10)'
except:
pass
return ''
display = matched[[
'A_open_time', 'A_symbol', 'A_type',
'A_open_price', 'A_close_price', 'A_profit', 'A_duration',
'B_open_time',
'B_open_price', 'B_close_price', 'B_profit', 'B_duration',
'open_slippage', 'close_slippage', 'profit_var', 'time_diff_min'
]].copy()
display.columns = [
'A Open Time', 'Symbol', 'Type',
'A Entry', 'A Exit', 'A Profit', 'A Dur(m)',
'B Open Time',
'B Entry', 'B Exit', 'B Profit', 'B Dur(m)',
'Entry Slip', 'Exit Slip', 'Profit Var', 'Time Diff(m)'
]
# Format numeric columns
for col in ['A Entry', 'A Exit', 'B Entry', 'B Exit']:
if col in display.columns:
display[col] = display[col].apply(
lambda x: f"{x:.5f}" if pd.notna(x) else '')
for col in ['A Profit', 'B Profit', 'Profit Var']:
display[col] = display[col].apply(
lambda x: f"{x:+.2f}" if pd.notna(x) else '')
for col in ['Entry Slip', 'Exit Slip']:
display[col] = display[col].apply(
lambda x: f"{x:+.5f}" if pd.notna(x) else '')
st.dataframe(
display.style
.map(colour_diff, subset=['Entry Slip', 'Exit Slip', 'Profit Var', 'Time Diff(m)'])
.map(colour_profit_cell, subset=['A Profit', 'B Profit']),
use_container_width=True, hide_index=True, height=500
)
# ── Export ────────────────────────────────────────────────────────────────
st.download_button(
"⬇ Download matched trades CSV",
data = display.to_csv(index=False),
file_name = "trade_comparison.csv",
mime = 'text/csv'
)
+215
View File
@@ -0,0 +1,215 @@
"""
view_set_comparator.py
======================
EA Settings Comparator page — migrated from main dashboard.
"""
import streamlit as st
import pandas as pd
from datetime import datetime
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from set_comparator import parse_set_file, build_comparison_df, export_set_file, create_zip
def render():
st.markdown("""
<style>
[data-testid="stDataFrame"] {
margin-left: auto;
margin-right: auto;
}
</style>
""", unsafe_allow_html=True)
st.title("⚙ EA Settings Comparator")
# ── Session state init ────────────────────────────────────────────────────
if 'ea_files' not in st.session_state: st.session_state['ea_files'] = {}
if 'ea_raw' not in st.session_state: st.session_state['ea_raw'] = {}
if 'ea_order' not in st.session_state: st.session_state['ea_order'] = {}
if 'ea_bytes' not in st.session_state: st.session_state['ea_bytes'] = {}
if 'ea_edited' not in st.session_state: st.session_state['ea_edited'] = {}
# ── Controls ──────────────────────────────────────────────────────────────
col1, col2, col3 = st.columns([1, 1, 4])
with col1:
n_files = st.selectbox("Number of files", list(range(2, 11)), index=0)
with col2:
st.markdown("<br>", unsafe_allow_html=True)
if st.button("🗑 Clear All", type="secondary"):
for key in ['ea_files', 'ea_raw', 'ea_order', 'ea_bytes', 'ea_edited']:
st.session_state[key] = {}
st.rerun()
# ── File upload slots ─────────────────────────────────────────────────────
st.divider()
upload_cols = st.columns(min(n_files, 5))
for i in range(n_files):
with upload_cols[i % 5]:
uploaded = st.file_uploader(
f"File {i+1}", type=['set'], key=f"ea_upload_{i}"
)
if uploaded is not None:
file_bytes = uploaded.read()
fname = uploaded.name
params, raw_lines, order = parse_set_file(file_bytes, fname)
st.session_state['ea_files'][fname] = params
st.session_state['ea_raw'][fname] = raw_lines
st.session_state['ea_order'][fname] = order
st.session_state['ea_bytes'][fname] = file_bytes
if fname not in st.session_state['ea_edited']:
st.session_state['ea_edited'][fname] = params.copy()
st.success(f"{fname}{len(params)} params")
# ── Comparison table ──────────────────────────────────────────────────────
files_data = st.session_state['ea_files']
if len(files_data) >= 2:
st.divider()
filenames = list(files_data.keys())
col1, col2, col3 = st.columns(3)
with col1:
source_file = st.selectbox("Source file for comparison", filenames)
with col2:
pct_threshold = st.slider("Highlight % variation from source", 0, 100, 10)
with col3:
show_diff_only = st.toggle("Show different rows only", value=False)
files_list = [
(fn, st.session_state['ea_files'][fn],
st.session_state['ea_raw'][fn],
st.session_state['ea_order'][fn])
for fn in filenames
]
df = build_comparison_df(files_list)
value_cols = [c for c in df.columns if c != 'Parameter']
df['_diff'] = df[value_cols].nunique(axis=1) > 1
df_display = df[df['_diff']].copy() if show_diff_only else df.copy()
df_display = df_display.drop(columns=['_diff'])
source_vals = files_data.get(source_file, {})
def style_cells(row):
styles = [''] * len(row)
param = row['Parameter']
src_v = source_vals.get(param, '')
for j, col in enumerate(row.index):
if col == 'Parameter':
continue
cell_v = row[col]
if col == source_file:
styles[j] = 'background-color: rgba(100,100,255,0.15)'
continue
if cell_v == '' or src_v == '':
if cell_v != src_v:
styles[j] = 'background-color: rgba(255,180,0,0.2)'
continue
try:
sv = float(src_v)
cv = float(cell_v)
if sv == 0:
if cv != 0:
styles[j] = 'background-color: rgba(255,100,100,0.2)'
else:
pct_diff = abs((cv - sv) / sv) * 100
if pct_diff > pct_threshold:
styles[j] = 'background-color: rgba(255,100,100,0.2)'
elif pct_diff > 0:
styles[j] = 'background-color: rgba(255,180,0,0.15)'
except:
if cell_v != src_v:
styles[j] = 'background-color: rgba(255,180,0,0.2)'
return styles
st.markdown(
f"**{len(df_display)} parameters** — "
f"{int(df[df['_diff']].shape[0])} rows differ across files"
)
styled = df_display.style.apply(style_cells, axis=1)
row_height = 35
table_h = min(len(df_display) * row_height + 40, 2000)
col_config = {'Parameter': st.column_config.TextColumn('Parameter', width='medium')}
for fn in filenames:
col_config[fn] = st.column_config.TextColumn(fn, width='small')
st.dataframe(
styled, width='content', hide_index=True,
height=table_h, column_config=col_config
)
# ── Edit & Export ─────────────────────────────────────────────────────
st.divider()
st.subheader("Edit & Export")
edit_file = st.selectbox("Select file to edit", filenames, key='ea_edit_sel')
if edit_file:
edited_params = st.session_state['ea_edited'].get(edit_file, {})
order = st.session_state['ea_order'].get(edit_file, [])
raw_lines = st.session_state['ea_raw'].get(edit_file, {})
edit_df = pd.DataFrame([
{'Parameter': k, 'Value': edited_params.get(k, '')}
for k in order
])
edited = st.data_editor(
edit_df, width='stretch', hide_index=True, height=400,
column_config={
'Parameter': st.column_config.TextColumn('Parameter', disabled=True),
'Value' : st.column_config.TextColumn('Value'),
},
key=f"ea_editor_{edit_file}"
)
st.session_state['ea_edited'][edit_file] = dict(
zip(edited['Parameter'], edited['Value'].astype(str))
)
col1, col2 = st.columns(2)
with col1:
export_bytes = export_set_file(
edit_file,
st.session_state['ea_edited'][edit_file],
raw_lines, order,
st.session_state['ea_bytes'][edit_file]
)
st.download_button(
label = f"⬇ Export {edit_file}",
data = export_bytes,
file_name = edit_file,
mime = 'application/octet-stream',
key = 'ea_export_single'
)
with col2:
all_exports = []
for fn in filenames:
fb = export_set_file(
fn,
st.session_state['ea_edited'].get(fn, files_data[fn]),
st.session_state['ea_raw'][fn],
st.session_state['ea_order'][fn],
st.session_state['ea_bytes'][fn]
)
all_exports.append((fn, fb))
zip_bytes = create_zip(all_exports)
st.download_button(
label = "⬇ Export All as ZIP",
data = zip_bytes,
file_name = f"ea_settings_{datetime.today().strftime('%Y%m%d')}.zip",
mime = 'application/zip',
key = 'ea_export_all'
)
elif len(files_data) == 1:
st.info("Upload at least 2 files to compare")
else:
st.info("Upload .set files above to begin comparison")
+37
View File
@@ -0,0 +1,37 @@
"""
pages/settings.py
=================
Settings page for MT5 Tools dashboard.
"""
import streamlit as st
def render():
st.title("⚙️ Settings")
st.markdown("""
<div class="info-card">
MT5 Tools — settings and information.
</div>
""", unsafe_allow_html=True)
st.subheader("About")
st.markdown("""
**MT5 Tools** is a standalone trade analysis and comparison dashboard.
**Supported file formats:**
- MT5 Account History HTML export (`.htm` / `.html`)
- MT5 Strategy Tester Backtest Report (`.htm` / `.html`)
- Quant Analyzer CSV export (`listOfTrades_*.csv`)
**Pages:**
- **Trade Analysis** — statistics, equity curves, day/hour breakdown for a single report
- **Trade Compare** — match and compare two reports to measure slippage and variance
""")
st.subheader("Requirements")
st.code("pip install streamlit streamlit-option-menu pandas plotly", language="bash")
st.subheader("Launch")
st.code("streamlit run app.py", language="bash")
+333
View File
@@ -0,0 +1,333 @@
"""
pages/trade_analysis.py
=======================
MT5 Trade Analysis page — migrated from main dashboard.
"""
import streamlit as st
import plotly.graph_objects as go
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mt5_parser import detect_and_parse, calc_stats
def render():
st.title("📊 Trade Analysis")
# ── Session state ─────────────────────────────────────────────────────────
if 'ta_df' not in st.session_state:
st.session_state['ta_df'] = None
st.session_state['ta_format'] = None
# ── File upload ───────────────────────────────────────────────────────────
col1, col2 = st.columns([4, 1])
with col1:
uploaded = st.file_uploader(
"Upload MT5 Report (HTM/HTML) or Quant Analyzer CSV",
type=['html', 'htm', 'csv'],
key='ta_upload'
)
with col2:
st.markdown("<br>", unsafe_allow_html=True)
if st.button("🗑 Clear", key='ta_clear'):
st.session_state['ta_df'] = None
st.session_state['ta_format'] = None
st.rerun()
if uploaded:
df, fmt = detect_and_parse(uploaded.read(), uploaded.name)
if df is not None:
st.session_state['ta_df'] = df
st.session_state['ta_format'] = fmt
st.success(f"✓ Loaded {len(df)} trades — {fmt}")
else:
st.error("Could not parse report — check file format")
df_all = st.session_state['ta_df']
fmt = st.session_state['ta_format']
if df_all is None or len(df_all) == 0:
st.markdown("""
<div class="info-card">
Upload an MT5 account history report (.htm/.html), MT5 backtest report,
or a Quant Analyzer CSV export to begin analysis.
</div>
""", unsafe_allow_html=True)
return
if fmt:
st.caption(f"Format detected: **{fmt}** · {len(df_all)} total trades")
# ── Filters ───────────────────────────────────────────────────────────────
st.divider()
fc1, fc2, fc3, fc4 = st.columns(4)
with fc1:
date_min = df_all['open_time'].min().date()
date_max = df_all['open_time'].max().date()
date_from = st.date_input("From", value=date_min, min_value=date_min,
max_value=date_max, key='ta_from')
date_to = st.date_input("To", value=date_max, min_value=date_min,
max_value=date_max, key='ta_to')
with fc2:
symbols = sorted(df_all['symbol'].dropna().unique().tolist())
sel_symbol = st.multiselect("Symbol", symbols, key='ta_sym')
with fc3:
strategies = sorted(df_all['strategy'].dropna().unique().tolist())
sel_strategy = st.multiselect("Strategy / EA", strategies, key='ta_strat')
with fc4:
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday']
sel_days = st.multiselect("Day of week", days, key='ta_days')
sel_type = st.multiselect("Type", ['buy', 'sell'], key='ta_type')
# Apply filters
df = df_all.copy()
df = df[(df['open_time'].dt.date >= date_from) &
(df['open_time'].dt.date <= date_to)]
if sel_symbol:
df = df[df['symbol'].isin(sel_symbol)]
if sel_strategy:
df = df[df['strategy'].isin(sel_strategy)]
if sel_days:
df = df[df['day_of_week'].isin(sel_days)]
if sel_type:
df = df[df['type'].isin(sel_type)]
st.caption(f"Showing **{len(df)}** trades after filters")
# ── Analysis mode ─────────────────────────────────────────────────────────
mode = st.radio(
"Analysis mode",
["Overall", "By Strategy", "By Symbol", "By Day of Week"],
horizontal=True, key='ta_mode'
)
st.divider()
# ── Helpers ───────────────────────────────────────────────────────────────
def render_stats(stats, label=""):
if label:
st.markdown(f"**{label}**")
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("Net Profit", f"${stats['net_profit']:,.2f}")
c2.metric("Win Rate", f"{stats['win_rate']}%")
c3.metric("Profit Factor", f"{stats['profit_factor']}")
c4.metric("R:R Ratio", f"{stats['rr_ratio']}")
c5.metric("Expectancy", f"${stats['expectancy']:,.2f}")
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("Total Trades", stats['total_trades'])
c2.metric("Avg Win", f"${stats['avg_win']:,.2f}")
c3.metric("Avg Loss", f"${stats['avg_loss']:,.2f}")
c4.metric("Max DD", f"${stats['max_drawdown']:,.2f}")
c5.metric("Best Trade", f"${stats['best_trade']:,.2f}")
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("Max Consec Wins", stats['max_consec_wins'])
c2.metric("Max Consec Losses", stats['max_consec_losses'])
c3.metric("Avg Win Dur", f"{stats['avg_win_duration']}m")
c4.metric("Avg Loss Dur", f"{stats['avg_loss_duration']}m")
c5.metric("Worst Trade", f"${stats['worst_trade']:,.2f}")
c1, c2, c3, c4 = st.columns(4)
c1.metric("Long Trades", stats['long_trades'])
c2.metric("Long Win Rate", f"{stats['long_win_rate']}%")
c3.metric("Short Trades", stats['short_trades'])
c4.metric("Short Win Rate",f"{stats['short_win_rate']}%")
def render_equity_curve(df_plot, label="Equity Curve"):
df_s = df_plot.sort_values('close_time').copy()
df_s['cumulative'] = df_s['net_profit'].cumsum()
fig = go.Figure()
fig.add_trace(go.Scatter(
x=df_s['close_time'], y=df_s['cumulative'],
mode='lines',
line=dict(color='#7c6af7', width=2),
fill='tozeroy',
fillcolor='rgba(124,106,247,0.08)',
name='Equity'
))
fig.update_layout(
title=label, height=300,
plot_bgcolor='rgba(10,10,15,1)',
paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa', family='JetBrains Mono'),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
margin=dict(l=60, r=20, t=40, b=40)
)
st.plotly_chart(fig, use_container_width=True)
def render_dow_chart(df_plot):
dow_order = ['Monday','Tuesday','Wednesday','Thursday','Friday','Saturday','Sunday']
dow = df_plot.groupby('day_of_week').agg(
trades = ('net_profit', 'count'),
net_profit = ('net_profit', 'sum'),
win_rate = ('win', lambda x: round(x.mean()*100, 1))
).reindex([d for d in dow_order if d in df_plot['day_of_week'].unique()])
wins_dow = df_plot[df_plot['win']].groupby('day_of_week')['net_profit'].sum().reindex(dow.index, fill_value=0)
losses_dow = df_plot[~df_plot['win']].groupby('day_of_week')['net_profit'].sum().reindex(dow.index, fill_value=0)
fig = go.Figure()
fig.add_trace(go.Bar(x=dow.index, y=wins_dow, name='Profit', marker_color='rgba(45,198,83,0.8)'))
fig.add_trace(go.Bar(x=dow.index, y=losses_dow, name='Loss', marker_color='rgba(230,57,70,0.8)'))
fig.update_layout(
title='P&L by Day of Week', height=280, barmode='relative',
plot_bgcolor='rgba(10,10,15,1)', paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa'), margin=dict(l=60, r=20, t=40, b=40),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
legend=dict(bgcolor='rgba(0,0,0,0.3)')
)
st.plotly_chart(fig, use_container_width=True)
dt = dow.reset_index()
dt.columns = ['Day', 'Trades', 'Net Profit', 'Win Rate %']
dt['Net Profit'] = dt['Net Profit'].round(2)
st.dataframe(dt, use_container_width=True, hide_index=True)
def render_hour_chart(df_plot):
hourly = df_plot.groupby('hour').agg(
trades = ('net_profit', 'count'),
net_profit = ('net_profit', 'sum'),
)
wins_h = df_plot[df_plot['win']].groupby('hour')['net_profit'].sum().reindex(hourly.index, fill_value=0)
losses_h = df_plot[~df_plot['win']].groupby('hour')['net_profit'].sum().reindex(hourly.index, fill_value=0)
fig = go.Figure()
fig.add_trace(go.Bar(x=wins_h.index, y=wins_h, name='Profit', marker_color='rgba(45,198,83,0.8)'))
fig.add_trace(go.Bar(x=losses_h.index, y=losses_h, name='Loss', marker_color='rgba(230,57,70,0.8)'))
fig.update_layout(
title='P&L by Hour of Day', height=280, barmode='relative',
plot_bgcolor='rgba(10,10,15,1)', paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa'), margin=dict(l=60, r=20, t=40, b=40),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)', title='Hour (UTC)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
legend=dict(bgcolor='rgba(0,0,0,0.3)')
)
st.plotly_chart(fig, use_container_width=True)
def colour_profit(val):
try:
v = float(str(val).replace(',', ''))
if v > 0: return 'background-color: rgba(0,180,0,0.12)'
if v < 0: return 'background-color: rgba(180,0,0,0.12)'
except:
pass
return ''
# ── Render mode ───────────────────────────────────────────────────────────
if mode == "Overall":
stats = calc_stats(df)
render_stats(stats, "Overall Statistics")
render_equity_curve(df)
col1, col2 = st.columns(2)
with col1:
render_dow_chart(df)
with col2:
render_hour_chart(df)
elif mode == "By Strategy":
strats = sorted(df['strategy'].dropna().unique().tolist())
if not strats:
st.info("No strategies found")
else:
st.subheader("Strategy Comparison")
rows = []
for s in strats:
sdf = df[df['strategy'] == s]
stat = calc_stats(sdf)
rows.append({
'Strategy' : s,
'Trades' : stat['total_trades'],
'Net Profit' : stat['net_profit'],
'Win Rate %' : stat['win_rate'],
'Profit Factor' : stat['profit_factor'],
'R:R' : stat['rr_ratio'],
'Expectancy' : stat['expectancy'],
'Max DD' : stat['max_drawdown'],
'Max Consec W' : stat['max_consec_wins'],
'Max Consec L' : stat['max_consec_losses'],
})
sdf_sum = __import__('pandas').DataFrame(rows).sort_values('Net Profit', ascending=False)
st.dataframe(
sdf_sum.style.map(colour_profit, subset=['Net Profit', 'Expectancy', 'Max DD']),
use_container_width=True, hide_index=True
)
st.divider()
sel = st.selectbox("Select strategy for detail", strats)
if sel:
sdf = df[df['strategy'] == sel]
stat = calc_stats(sdf)
render_stats(stat, sel)
render_equity_curve(sdf, f"{sel} — Equity Curve")
col1, col2 = st.columns(2)
with col1: render_dow_chart(sdf)
with col2: render_hour_chart(sdf)
elif mode == "By Symbol":
syms = sorted(df['symbol'].dropna().unique().tolist())
rows = []
for s in syms:
sdf = df[df['symbol'] == s]
stat = calc_stats(sdf)
rows.append({
'Symbol' : s,
'Trades' : stat['total_trades'],
'Net Profit' : stat['net_profit'],
'Win Rate %' : stat['win_rate'],
'Profit Factor' : stat['profit_factor'],
'R:R' : stat['rr_ratio'],
'Expectancy' : stat['expectancy'],
'Max DD' : stat['max_drawdown'],
})
sdf_sum = __import__('pandas').DataFrame(rows).sort_values('Net Profit', ascending=False)
st.dataframe(
sdf_sum.style.map(colour_profit, subset=['Net Profit', 'Expectancy', 'Max DD']),
use_container_width=True, hide_index=True
)
sel = st.selectbox("Select symbol for detail", syms)
if sel:
sdf = df[df['symbol'] == sel]
stat = calc_stats(sdf)
render_stats(stat, sel)
render_equity_curve(sdf, f"{sel} — Equity Curve")
col1, col2 = st.columns(2)
with col1: render_dow_chart(sdf)
with col2: render_hour_chart(sdf)
elif mode == "By Day of Week":
render_dow_chart(df)
render_hour_chart(df)
# ── Raw trade log ─────────────────────────────────────────────────────────
st.divider()
with st.expander("Raw Trade Log"):
show_cols = ['open_time', 'close_time', 'symbol', 'type', 'strategy',
'volume', 'open_price', 'close_price', 'sl', 'tp',
'commission', 'swap', 'profit', 'net_profit', 'duration_min']
show_cols = [c for c in show_cols if c in df.columns]
def colour_net(val):
try:
v = float(val)
if v > 0: return 'background-color: rgba(0,180,0,0.12)'
if v < 0: return 'background-color: rgba(180,0,0,0.12)'
except:
pass
return ''
st.dataframe(
df[show_cols].style.map(colour_net, subset=['net_profit', 'profit']),
use_container_width=True, hide_index=True, height=400
)
st.download_button(
"⬇ Download filtered trades CSV",
data = df[show_cols].to_csv(index=False),
file_name = f"mt5_trades_{date_from}_{date_to}.csv",
mime = 'text/csv'
)
+422
View File
@@ -0,0 +1,422 @@
"""
pages/trade_compare.py
======================
Side-by-side comparison of two trade history files.
Matches trades by symbol + type + open time within a tolerance window.
Highlights slippage, profit variance, and timing differences.
"""
import streamlit as st
import pandas as pd
import plotly.graph_objects as go
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mt5_parser import detect_and_parse, calc_stats
# ── Match trades ──────────────────────────────────────────────────────────────
def match_trades(df_a, df_b, tolerance_hours):
"""
Match trades between two DataFrames.
Match criteria: same symbol_base + same type + open_time within tolerance.
Returns DataFrame of matched pairs with diff columns.
"""
tol = pd.Timedelta(hours=tolerance_hours)
matched = []
used_b = set()
for i, a in df_a.iterrows():
best_match = None
best_delta = tol + pd.Timedelta(seconds=1)
for j, b in df_b.iterrows():
if j in used_b:
continue
if a['symbol_base'] != b['symbol_base']:
continue
if a['type'] != b['type']:
continue
delta = abs(a['open_time'] - b['open_time'])
if delta <= tol and delta < best_delta:
best_delta = delta
best_match = (j, b)
if best_match:
j, b = best_match
used_b.add(j)
open_slip = round(float(b['open_price']) - float(a['open_price']), 5) if pd.notna(a['open_price']) and pd.notna(b['open_price']) else None
close_slip = round(float(b['close_price']) - float(a['close_price']), 5) if pd.notna(a['close_price']) and pd.notna(b['close_price']) else None
profit_var = round(float(b['net_profit']) - float(a['net_profit']), 2) if pd.notna(a['net_profit']) and pd.notna(b['net_profit']) else None
time_diff = round((b['open_time'] - a['open_time']).total_seconds() / 60, 1)
dur_diff = round(float(b.get('duration_min', 0) or 0) - float(a.get('duration_min', 0) or 0), 1)
matched.append({
# File A
'A_open_time' : a['open_time'],
'A_close_time' : a['close_time'],
'A_symbol' : a['symbol'],
'A_type' : a['type'],
'A_volume' : a.get('volume'),
'A_open_price' : a.get('open_price'),
'A_close_price': a.get('close_price'),
'A_profit' : a.get('net_profit'),
'A_duration' : a.get('duration_min'),
# File B
'B_open_time' : b['open_time'],
'B_close_time' : b['close_time'],
'B_symbol' : b['symbol'],
'B_type' : b['type'],
'B_volume' : b.get('volume'),
'B_open_price' : b.get('open_price'),
'B_close_price': b.get('close_price'),
'B_profit' : b.get('net_profit'),
'B_duration' : b.get('duration_min'),
# Differences
'open_slippage' : open_slip,
'close_slippage': close_slip,
'profit_var' : profit_var,
'time_diff_min' : time_diff,
'duration_diff' : dur_diff,
})
return pd.DataFrame(matched)
# ── Render ────────────────────────────────────────────────────────────────────
def render():
st.title("🔄 Trade Compare")
st.markdown("""
<div class="info-card">
Compare two trade history files — backtest vs real account, or any two exports.
Trades are matched by symbol, direction, and open time within a configurable
tolerance window to account for gaps, slippage, and market open variations.
</div>
""", unsafe_allow_html=True)
# ── Session state ─────────────────────────────────────────────────────────
for k in ['tc_df_a', 'tc_df_b', 'tc_fmt_a', 'tc_fmt_b']:
if k not in st.session_state:
st.session_state[k] = None
# ── File upload ───────────────────────────────────────────────────────────
st.subheader("Load Files")
col_a, col_b = st.columns(2)
with col_a:
st.markdown("**File A** — Reference (e.g. Backtest)")
up_a = st.file_uploader("Upload File A", type=['html','htm','csv'], key='tc_up_a')
if up_a:
df_a, fmt_a = detect_and_parse(up_a.read(), up_a.name)
if df_a is not None:
st.session_state['tc_df_a'] = df_a
st.session_state['tc_fmt_a'] = fmt_a
st.success(f"{len(df_a)} trades — {fmt_a}")
else:
st.error("Could not parse File A")
if st.session_state['tc_df_a'] is not None:
st.caption(f"Loaded: **{st.session_state['tc_fmt_a']}** · {len(st.session_state['tc_df_a'])} trades")
with col_b:
st.markdown("**File B** — Comparison (e.g. Real Account)")
up_b = st.file_uploader("Upload File B", type=['html','htm','csv'], key='tc_up_b')
if up_b:
df_b, fmt_b = detect_and_parse(up_b.read(), up_b.name)
if df_b is not None:
st.session_state['tc_df_b'] = df_b
st.session_state['tc_fmt_b'] = fmt_b
st.success(f"{len(df_b)} trades — {fmt_b}")
else:
st.error("Could not parse File B")
if st.session_state['tc_df_b'] is not None:
st.caption(f"Loaded: **{st.session_state['tc_fmt_b']}** · {len(st.session_state['tc_df_b'])} trades")
df_a = st.session_state['tc_df_a']
df_b = st.session_state['tc_df_b']
if df_a is None or df_b is None:
return
# ── Filters ───────────────────────────────────────────────────────────────
st.divider()
st.subheader("Filters")
fa1, fa2, fa3 = st.columns(3)
fb1, fb2, fb3 = st.columns(3)
with fa1:
st.markdown("**File A filters**")
with fb1:
st.markdown("**File B filters**")
col1, col2, col3, col4, col5, col6 = st.columns(6)
with col1:
a_date_min = df_a['open_time'].min().date()
a_date_max = df_a['open_time'].max().date()
a_from = st.date_input("A — From", value=a_date_min, min_value=a_date_min,
max_value=a_date_max, key='tc_a_from')
a_to = st.date_input("A — To", value=a_date_max, min_value=a_date_min,
max_value=a_date_max, key='tc_a_to')
with col2:
a_syms = sorted(df_a['symbol'].dropna().unique().tolist())
a_sel_sym = st.multiselect("A — Symbol", a_syms, key='tc_a_sym')
with col3:
a_strats = sorted(df_a['strategy'].dropna().unique().tolist())
a_sel_strat = st.multiselect("A — Strategy", a_strats, key='tc_a_strat')
a_sel_type = st.multiselect("A — Type", ['buy', 'sell'], key='tc_a_type')
with col4:
b_date_min = df_b['open_time'].min().date()
b_date_max = df_b['open_time'].max().date()
b_from = st.date_input("B — From", value=b_date_min, min_value=b_date_min,
max_value=b_date_max, key='tc_b_from')
b_to = st.date_input("B — To", value=b_date_max, min_value=b_date_min,
max_value=b_date_max, key='tc_b_to')
with col5:
b_syms = sorted(df_b['symbol'].dropna().unique().tolist())
b_sel_sym = st.multiselect("B — Symbol", b_syms, key='tc_b_sym')
with col6:
b_strats = sorted(df_b['strategy'].dropna().unique().tolist())
b_sel_strat = st.multiselect("B — Strategy", b_strats, key='tc_b_strat')
b_sel_type = st.multiselect("B — Type", ['buy', 'sell'], key='tc_b_type')
# ── Matching tolerance ────────────────────────────────────────────────────
st.divider()
col_tol, col_run = st.columns([3, 1])
with col_tol:
tolerance = st.slider(
"Match tolerance (hours) — max time difference between A and B open times",
min_value=1, max_value=24, value=4, step=1,
help="Trades within this window are considered the same setup. "
"Increase for daily charts, decrease for intraday."
)
with col_run:
st.markdown("<br>", unsafe_allow_html=True)
run = st.button("🔍 Match Trades", type="primary", use_container_width=True)
if not run and 'tc_matched' not in st.session_state:
return
# Apply filters
fa = df_a.copy()
fa = fa[(fa['open_time'].dt.date >= a_from) & (fa['open_time'].dt.date <= a_to)]
if a_sel_sym: fa = fa[fa['symbol'].isin(a_sel_sym)]
if a_sel_strat: fa = fa[fa['strategy'].isin(a_sel_strat)]
if a_sel_type: fa = fa[fa['type'].isin(a_sel_type)]
fb = df_b.copy()
fb = fb[(fb['open_time'].dt.date >= b_from) & (fb['open_time'].dt.date <= b_to)]
if b_sel_sym: fb = fb[fb['symbol'].isin(b_sel_sym)]
if b_sel_strat: fb = fb[fb['strategy'].isin(b_sel_strat)]
if b_sel_type: fb = fb[fb['type'].isin(b_sel_type)]
if run:
with st.spinner("Matching trades..."):
matched = match_trades(fa, fb, tolerance)
st.session_state['tc_matched'] = matched
st.session_state['tc_fa_len'] = len(fa)
st.session_state['tc_fb_len'] = len(fb)
matched = st.session_state.get('tc_matched', pd.DataFrame())
fa_len = st.session_state.get('tc_fa_len', len(fa))
fb_len = st.session_state.get('tc_fb_len', len(fb))
if matched is None or len(matched) == 0:
st.warning("No matching trades found — try increasing the tolerance window or adjusting filters.")
return
# ── Summary stats ─────────────────────────────────────────────────────────
st.divider()
st.subheader("Match Summary")
m1, m2, m3, m4, m5 = st.columns(5)
m1.metric("File A Trades", fa_len)
m2.metric("File B Trades", fb_len)
m3.metric("Matched Pairs", len(matched))
m4.metric("Unmatched A", fa_len - len(matched))
m5.metric("Unmatched B", fb_len - len(matched))
st.divider()
# ── Aggregate comparison ───────────────────────────────────────────────────
st.subheader("Aggregate Comparison")
ac1, ac2 = st.columns(2)
with ac1:
st.markdown("**File A (Reference)**")
a_net = matched['A_profit'].sum()
a_wr = (matched['A_profit'] > 0).mean() * 100
a_avg = matched['A_profit'].mean()
a_dur = matched['A_duration'].mean() if 'A_duration' in matched else None
st.metric("Net Profit", f"${a_net:,.2f}")
st.metric("Win Rate", f"{a_wr:.1f}%")
st.metric("Avg Profit", f"${a_avg:,.2f}")
if a_dur:
st.metric("Avg Duration", f"{a_dur:.0f}m")
with ac2:
st.markdown("**File B (Comparison)**")
b_net = matched['B_profit'].sum()
b_wr = (matched['B_profit'] > 0).mean() * 100
b_avg = matched['B_profit'].mean()
b_dur = matched['B_duration'].mean() if 'B_duration' in matched else None
delta_net = b_net - a_net
st.metric("Net Profit", f"${b_net:,.2f}",
delta=f"{delta_net:+.2f}", delta_color="normal")
st.metric("Win Rate", f"{b_wr:.1f}%",
delta=f"{b_wr - a_wr:+.1f}%", delta_color="normal")
st.metric("Avg Profit", f"${b_avg:,.2f}",
delta=f"{b_avg - a_avg:+.2f}", delta_color="normal")
if b_dur and a_dur:
st.metric("Avg Duration", f"{b_dur:.0f}m",
delta=f"{b_dur - a_dur:+.0f}m", delta_color="off")
# ── Slippage summary ───────────────────────────────────────────────────────
st.divider()
st.subheader("Slippage & Variance Summary")
sc1, sc2, sc3, sc4 = st.columns(4)
avg_open_slip = matched['open_slippage'].mean()
avg_close_slip = matched['close_slip'].mean() if 'close_slip' in matched else matched['close_slippage'].mean()
avg_profit_var = matched['profit_var'].mean()
avg_time_diff = matched['time_diff_min'].mean()
sc1.metric("Avg Entry Slippage", f"{avg_open_slip:+.5f}" if pd.notna(avg_open_slip) else "N/A",
help="B open price minus A open price. Positive = B filled higher.")
sc2.metric("Avg Exit Slippage", f"{avg_close_slip:+.5f}" if pd.notna(avg_close_slip) else "N/A",
help="B close price minus A close price.")
sc3.metric("Avg Profit Variance", f"${avg_profit_var:+.2f}" if pd.notna(avg_profit_var) else "N/A",
help="B net profit minus A net profit per trade.")
sc4.metric("Avg Time Difference", f"{avg_time_diff:+.0f}m" if pd.notna(avg_time_diff) else "N/A",
help="B open time minus A open time in minutes.")
# ── Equity curve overlay ───────────────────────────────────────────────────
st.divider()
st.subheader("Equity Curve Overlay")
m_sorted = matched.sort_values('A_open_time')
fig = go.Figure()
fig.add_trace(go.Scatter(
x=m_sorted['A_open_time'],
y=m_sorted['A_profit'].cumsum(),
mode='lines', name='File A',
line=dict(color='#7c6af7', width=2),
fill='tozeroy', fillcolor='rgba(124,106,247,0.05)'
))
fig.add_trace(go.Scatter(
x=m_sorted['B_open_time'],
y=m_sorted['B_profit'].cumsum(),
mode='lines', name='File B',
line=dict(color='#2dc653', width=2),
fill='tozeroy', fillcolor='rgba(45,198,83,0.05)'
))
fig.update_layout(
height=320,
plot_bgcolor='rgba(10,10,15,1)',
paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa', family='JetBrains Mono'),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
legend=dict(bgcolor='rgba(0,0,0,0.3)'),
margin=dict(l=60, r=20, t=20, b=40)
)
st.plotly_chart(fig, use_container_width=True)
# ── Profit variance scatter ────────────────────────────────────────────────
st.subheader("Profit Variance per Trade")
fig2 = go.Figure()
colours = matched['profit_var'].apply(
lambda v: 'rgba(45,198,83,0.7)' if v >= 0 else 'rgba(230,57,70,0.7)'
)
fig2.add_trace(go.Bar(
x=list(range(len(matched))),
y=matched['profit_var'],
marker_color=colours,
name='Profit Variance (B - A)'
))
fig2.update_layout(
height=250,
plot_bgcolor='rgba(10,10,15,1)',
paper_bgcolor='rgba(10,10,15,1)',
font=dict(color='#aaa'),
xaxis=dict(gridcolor='rgba(255,255,255,0.04)', title='Trade #'),
yaxis=dict(gridcolor='rgba(255,255,255,0.04)', tickprefix='$'),
margin=dict(l=60, r=20, t=20, b=40)
)
st.plotly_chart(fig2, use_container_width=True)
# ── Matched trade table ────────────────────────────────────────────────────
st.divider()
st.subheader("Matched Trade Detail")
def colour_diff(val):
try:
v = float(str(val).replace('+', ''))
if v > 0: return 'color: #2dc653; font-weight: 600'
if v < 0: return 'color: #e63946; font-weight: 600'
except:
pass
return 'color: #666'
def colour_profit_cell(val):
try:
v = float(str(val).replace(',', ''))
if v > 0: return 'background-color: rgba(0,180,0,0.10)'
if v < 0: return 'background-color: rgba(180,0,0,0.10)'
except:
pass
return ''
display = matched[[
'A_open_time', 'A_symbol', 'A_type',
'A_open_price', 'A_close_price', 'A_profit', 'A_duration',
'B_open_time',
'B_open_price', 'B_close_price', 'B_profit', 'B_duration',
'open_slippage', 'close_slippage', 'profit_var', 'time_diff_min'
]].copy()
display.columns = [
'A Open Time', 'Symbol', 'Type',
'A Entry', 'A Exit', 'A Profit', 'A Dur(m)',
'B Open Time',
'B Entry', 'B Exit', 'B Profit', 'B Dur(m)',
'Entry Slip', 'Exit Slip', 'Profit Var', 'Time Diff(m)'
]
# Format numeric columns
for col in ['A Entry', 'A Exit', 'B Entry', 'B Exit']:
if col in display.columns:
display[col] = display[col].apply(
lambda x: f"{x:.5f}" if pd.notna(x) else '')
for col in ['A Profit', 'B Profit', 'Profit Var']:
display[col] = display[col].apply(
lambda x: f"{x:+.2f}" if pd.notna(x) else '')
for col in ['Entry Slip', 'Exit Slip']:
display[col] = display[col].apply(
lambda x: f"{x:+.5f}" if pd.notna(x) else '')
st.dataframe(
display.style
.map(colour_diff, subset=['Entry Slip', 'Exit Slip', 'Profit Var', 'Time Diff(m)'])
.map(colour_profit_cell, subset=['A Profit', 'B Profit']),
use_container_width=True, hide_index=True, height=500
)
# ── Export ────────────────────────────────────────────────────────────────
st.download_button(
"⬇ Download matched trades CSV",
data = display.to_csv(index=False),
file_name = "trade_comparison.csv",
mime = 'text/csv'
)