#!/usr/bin/env python3 """ broker_normalizer.py Multi-broker CSV normalization and cross-platform reconciliation for MetaTrader 5 strategy export files. Requires: Python 3.6+, pandas >= 1.5, numpy >= 1.23, matplotlib >= 3.7, seaborn >= 0.12. Usage: python broker_normalizer.py """ import os import logging import warnings import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns from datetime import timedelta from typing import Optional warnings.filterwarnings("ignore", category=FutureWarning) # ── Broker Profile Registry ──────────────────────────────────────────────────── BROKER_PROFILES = { "BrokerA": { "display_name" : "Broker A (ECN)", "server_utc_offset" : 2, "digit_convention" : 5, "commission_model" : "per_trade", "commission_per_lot": 3.50, "account_currency" : "USD", "symbol_map" : { "EURUSD" : "EURUSD", "GBPUSD" : "GBPUSD", "XAUUSD" : "XAUUSD", "US30" : "US30", } }, "BrokerB": { "display_name" : "Broker B (Market Maker)", "server_utc_offset" : 3, "digit_convention" : 5, "commission_model" : "spread_only", "commission_per_lot": 0.0, "account_currency" : "EUR", "symbol_map" : { "EURUSDm" : "EURUSD", "GBPUSDm" : "GBPUSD", "GOLDm" : "XAUUSD", "DJ30" : "US30", } }, "BrokerC": { "display_name" : "Broker C (STP)", "server_utc_offset" : 0, "digit_convention" : 4, "commission_model" : "per_trade", "commission_per_lot": 5.00, "account_currency" : "GBP", "symbol_map" : { "EURUSD" : "EURUSD", "GBPUSD" : "GBPUSD", "XAUUSD" : "XAUUSD", "Dow Jones" : "US30", } } } STATIC_FX_RATES_TO_USD = { "USD": 1.0000, "EUR": 1.0820, "GBP": 1.2640, "JPY": 0.0066, "CHF": 1.1150, "CAD": 0.7380, "AUD": 0.6530, } # ── Light Theme for Publication ──────────────────────────────────────────────── plt.rcParams.update({ "figure.facecolor" : "white", "axes.facecolor" : "#f9f9f9", "axes.edgecolor" : "#888888", "axes.labelcolor" : "#333333", "xtick.color" : "#555555", "ytick.color" : "#555555", "text.color" : "#333333", "grid.color" : "#d0d0d0", "grid.linestyle" : "--", "grid.linewidth" : 0.5, "font.family" : "sans-serif", "font.size" : 9, }) logging.basicConfig( filename="normalization_audit.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) def load_broker_export(file_path: str, broker_id: str) -> Optional[pd.DataFrame]: if not os.path.isfile(file_path): logging.error(f"File not found: {file_path}") print(f"[Normalizer] Error: {file_path} not found.") return None try: df = pd.read_csv(file_path, encoding="ansi", low_memory=False) except UnicodeDecodeError: try: df = pd.read_csv(file_path, encoding="utf-8", low_memory=False) except Exception as exc: logging.error(f"Failed to parse {file_path}: {exc}") return None df["Broker_ID"] = broker_id logging.info(f"Loaded {len(df):,} rows from " f"{os.path.basename(file_path)} [{broker_id}]") return df def normalize_pip_convention(df: pd.DataFrame, broker_id: str) -> pd.DataFrame: profile = BROKER_PROFILES.get(broker_id) if profile is None: return df if profile.get("digit_convention", 5) == 4: point_cols = [c for c in df.columns if "Points" in c or "points" in c] for col in point_cols: if pd.api.types.is_numeric_dtype(df[col]): df[col] = df[col] * 10.0 logging.info(f"[{broker_id}] Scaled {col} x10 " f"(4-digit -> 5-digit).") return df def normalize_symbols(df: pd.DataFrame, broker_id: str) -> pd.DataFrame: profile = BROKER_PROFILES.get(broker_id, {}) sym_map = profile.get("symbol_map", {}) if "Symbol" not in df.columns: df["Symbol_Canonical"] = "UNKNOWN" return df df["Symbol_Canonical"] = df["Symbol"].map(sym_map) unmapped_mask = df["Symbol_Canonical"].isna() unmapped_symbols = df.loc[unmapped_mask, "Symbol"].unique() for sym in unmapped_symbols: logging.warning(f"[{broker_id}] Unmapped symbol: '{sym}'.") df.loc[unmapped_mask, "Symbol_Canonical"] = ( df.loc[unmapped_mask, "Symbol"] ) return df def normalize_timestamps(df: pd.DataFrame, broker_id: str) -> pd.DataFrame: profile = BROKER_PROFILES.get(broker_id, {}) utc_offset = profile.get("server_utc_offset", 0) delta = timedelta(hours=utc_offset) # Broader pattern: match any column containing "time" or "date" ts_patterns = ["time", "date", "timestamp"] ts_cols = [c for c in df.columns if any(p.lower() in c.lower() for p in ts_patterns)] for col in ts_cols: # Remove the deprecated infer_datetime_format parameter parsed = pd.to_datetime(df[col], errors="coerce") if parsed.notna().sum() > 0: df[col + "_UTC"] = parsed - delta logging.info(f"[{broker_id}] Normalized {col} " f"(offset: -{utc_offset}h).") return df def normalize_commission(df: pd.DataFrame, broker_id: str) -> pd.DataFrame: profile = BROKER_PROFILES.get(broker_id, {}) commission_model = profile.get("commission_model", "spread_only") commission_per_lot = profile.get("commission_per_lot", 0.0) if "Total_Commission_Paid" not in df.columns: vol_col = next((c for c in df.columns if "Volume" in c or "Lots" in c), None) if commission_model == "per_trade" and vol_col: if pd.api.types.is_numeric_dtype(df[vol_col]): df["Total_Commission_Paid"] = ( df[vol_col].fillna(0) * commission_per_lot * 2.0 ) else: df["Total_Commission_Paid"] = 0.0 else: df["Total_Commission_Paid"] = 0.0 df["Total_Commission_Paid"] = pd.to_numeric( df["Total_Commission_Paid"], errors="coerce" ).fillna(0.0) vol_col = next((c for c in df.columns if "Volume" in c or "Lots" in c), None) if vol_col and pd.api.types.is_numeric_dtype(df[vol_col]): notional = df[vol_col].fillna(0.0) * 100_000.0 df["Commission_BPS"] = np.where( notional > 0, (df["Total_Commission_Paid"] / notional) * 10_000.0, 0.0 ) else: df["Commission_BPS"] = 0.0 return df def normalize_currency(df: pd.DataFrame, broker_id: str) -> pd.DataFrame: profile = BROKER_PROFILES.get(broker_id, {}) account_currency = profile.get("account_currency", "USD") monetary_cols = ["Net_Profit", "Total_Commission_Paid", "Gross_Profit"] if account_currency == "USD": for col in monetary_cols: if col in df.columns: df[col + "_USD"] = df[col] return df fx_rate = STATIC_FX_RATES_TO_USD.get(account_currency) if fx_rate is None: logging.warning(f"[{broker_id}] No FX rate for '{account_currency}'.") return df for col in monetary_cols: if col in df.columns: df[col + "_USD"] = pd.to_numeric( df[col], errors="coerce" ).fillna(0.0) * fx_rate logging.info(f"[{broker_id}] {col} -> USD @ {fx_rate:.4f}.") return df def build_unified_dataset(broker_file_map: dict) -> pd.DataFrame: frames = [] for broker_id, file_path in broker_file_map.items(): print(f"\n[Normalizer] Processing: {broker_id} <- {file_path}") df = load_broker_export(file_path, broker_id) if df is None: continue df = normalize_pip_convention(df, broker_id) df = normalize_symbols(df, broker_id) df = normalize_timestamps(df, broker_id) df = normalize_commission(df, broker_id) df = normalize_currency(df, broker_id) profile = BROKER_PROFILES.get(broker_id, {}) df["Broker_Display"] = profile.get("display_name", broker_id) frames.append(df) print(f" [OK] {len(df):,} rows normalized.") if not frames: raise ValueError("No broker exports loaded.") unified = pd.concat(frames, ignore_index=True, sort=False) col_aliases = { "Net_Profit_USD" : "Net_Profit_USD", "Net_Profit" : "Net_Profit_USD", "Total_Commission_Paid_USD": "Commission_USD", "Total_Commission_Paid" : "Commission_USD", "Gross_Profit_USD" : "Gross_Profit_USD", "Gross_Profit" : "Gross_Profit_USD", } for src, tgt in col_aliases.items(): if src in unified.columns and tgt not in unified.columns: unified[tgt] = unified[src] numeric_cols = [ "Net_Profit_USD", "Commission_USD", "Gross_Profit_USD", "Sortino_Ratio", "Max_Drawdown_PCT", "False_Flips_Whipsaws", "Avg_Lag_On_Turn_Bars", "Commission_BPS", "Avg_Slippage_Points" ] for col in numeric_cols: if col in unified.columns: unified[col] = pd.to_numeric(unified[col], errors="coerce") logging.info(f"Unified: {len(unified):,} rows from {len(frames)} brokers.") print(f"\n[Normalizer] Unified: {len(unified):,} rows from " f"{len(frames)} broker(s).") return unified def summarize_execution_quality(unified: pd.DataFrame) -> pd.DataFrame: group_col = "Broker_Display" if "Broker_Display" in unified.columns \ else "Broker_ID" agg_dict = {} if "Commission_BPS" in unified.columns: agg_dict["Commission_BPS"] = "mean" if "Avg_Slippage_Points" in unified.columns: agg_dict["Avg_Slippage_Points"] = "mean" if "Total_Trades" in unified.columns: agg_dict["Total_Trades"] = "sum" if not agg_dict: return pd.DataFrame() summary = unified.groupby(group_col).agg(agg_dict).round(4) print("\n── Execution Quality Summary ──────────────────────────────") print(summary.to_string()) return summary def plot_slippage_distributions(unified: pd.DataFrame, save_path: str = None): """ Plots KDE of Avg_Slippage_Points per broker. Minimum 3 data points required per broker; otherwise skipped. """ if "Avg_Slippage_Points" not in unified.columns: return if "Broker_Display" not in unified.columns: unified["Broker_Display"] = unified.get("Broker_ID", "Unknown") fig, ax = plt.subplots(figsize=(9.8, 5.0)) fig.suptitle( "Slippage Distribution by Broker | " "Avg_Slippage_Points Density Comparison", fontsize=11, fontweight="bold", color="#222222" ) brokers = unified["Broker_Display"].unique() palette = sns.color_palette("husl", len(brokers)) for broker, color in zip(brokers, palette): values = unified.loc[ unified["Broker_Display"] == broker, "Avg_Slippage_Points" ].dropna() if len(values) < 3: continue sns.kdeplot(values, ax=ax, label=broker, color=color, linewidth=2.0, fill=True, alpha=0.12) ax.axvline(values.median(), color=color, linewidth=0.9, linestyle=":", alpha=0.9) ax.text(values.median() + 0.05, ax.get_ylim()[1] * 0.85, f"Med:{values.median():.1f}", fontsize=6.5, color=color, ha="left") ax.set_xlabel("Average Slippage (Normalized Points)", fontsize=9, labelpad=7) ax.set_ylabel("Density", fontsize=9, labelpad=7) ax.legend(title="Broker", fontsize=8, title_fontsize=8, loc="upper right", framealpha=0.9, edgecolor="#aaaaaa", facecolor="white") ax.grid(True, alpha=0.4) ax.spines[["top", "right"]].set_visible(False) plt.tight_layout() if save_path: plt.savefig(save_path, dpi=120, bbox_inches="tight", facecolor=fig.get_facecolor()) plt.show() def plot_net_of_cost_ranking(unified: pd.DataFrame, symbol_filter: str = None, save_path: str = None): """ Grouped bar chart: Gross vs Net profit by broker and indicator. """ subset = unified.copy() if symbol_filter and "Symbol_Canonical" in subset.columns: subset = subset[subset["Symbol_Canonical"] == symbol_filter] required = ["Broker_Display", "Indicator_Name", "Gross_Profit_USD", "Net_Profit_USD"] missing = [c for c in required if c not in subset.columns] if missing: print(f"[Net-of-Cost Plot] Missing: {missing}") return group_col = "Indicator_Name" agg = subset.groupby(["Broker_Display", group_col])[ ["Gross_Profit_USD", "Net_Profit_USD"] ].mean().reset_index() indicators = agg[group_col].unique() brokers = agg["Broker_Display"].unique() n_ind = len(indicators) n_brokers = len(brokers) palette = sns.color_palette("Set2", n_brokers) x = np.arange(n_ind) width = 0.35 / max(n_brokers, 1) fig, ax = plt.subplots(figsize=(9.8, 5.5)) title_sym = f" | {symbol_filter}" if symbol_filter else "" fig.suptitle( f"Net-of-Cost Performance Ranking{title_sym} | " "Gross vs Net Profit by Broker and Indicator", fontsize=10, fontweight="bold", color="#222222" ) for b_idx, (broker, color) in enumerate(zip(brokers, palette)): bm = agg["Broker_Display"] == broker gv, nv = [], [] for ind in indicators: im = agg[group_col] == ind rows = agg[bm & im] gv.append(float(rows["Gross_Profit_USD"].values[0]) if not rows.empty else 0.0) nv.append(float(rows["Net_Profit_USD"].values[0]) if not rows.empty else 0.0) offset = (b_idx - n_brokers / 2.0 + 0.5) * width * 2.2 ax.bar(x + offset - width / 2, gv, width=width, label=f"{broker} (Gross)", color=color, alpha=0.55, edgecolor="none") ax.bar(x + offset + width / 2, nv, width=width, label=f"{broker} (Net)", color=color, alpha=0.95, edgecolor="none") ax.axhline(0, color="#999999", linewidth=0.8) ax.set_xticks(x) ax.set_xticklabels(indicators, fontsize=8) ax.set_ylabel("Mean Net Profit (USD)", fontsize=9, labelpad=7) ax.set_xlabel("Indicator", fontsize=9, labelpad=7) ax.legend(fontsize=7, ncol=n_brokers, loc="upper right", framealpha=0.9, edgecolor="#aaaaaa", facecolor="white") ax.grid(axis="y", alpha=0.4) ax.spines[["top", "right"]].set_visible(False) plt.tight_layout() if save_path: plt.savefig(save_path, dpi=120, bbox_inches="tight", facecolor=fig.get_facecolor()) plt.show() # ── Entry Point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": OUTPUT_DIR = "normalized_outputs" os.makedirs(OUTPUT_DIR, exist_ok=True) broker_file_map = { "BrokerA" : "exports/brokerA_results.csv", "BrokerB" : "exports/brokerB_results.csv", "BrokerC" : "exports/brokerC_results.csv", } unified = build_unified_dataset(broker_file_map) canonical_path = os.path.join(OUTPUT_DIR, "unified_canonical.csv") unified.to_csv(canonical_path, index=False, encoding="utf-8") print(f"\n[Normalizer] Canonical dataset saved: {canonical_path}") summarize_execution_quality(unified) plot_slippage_distributions( unified, save_path=os.path.join(OUTPUT_DIR, "Slippage_Distributions.png") ) plot_net_of_cost_ranking( unified, symbol_filter="EURUSD", save_path=os.path.join(OUTPUT_DIR, "Net_of_Cost_Ranking_EURUSD.png") ) print(f"\n[Normalizer] All outputs written to: {OUTPUT_DIR}") print(f"[Normalizer] Audit log: normalization_audit.log")