import csv import json import re from datetime import datetime from typing import Dict, List, Any, Optional from collections import defaultdict import statistics class TradeParser: \"\"\"\"Parse MT5 trade history CSV and convert to structured data\"\"\" COLUMN_MAPPING = { 'Ticket': 'ticket', 'Symbol': 'symbol', 'Type': 'type', 'Open time': 'open_time', 'Open price': 'open_price', 'Size': 'size', 'Close time': 'close_time', 'Close price': 'close_price', 'Time in trade': 'duration', 'Profit/Loss': 'profit_loss', 'Cummulative P/L': 'cumulative_pl', 'Comm/Swap': 'comm_swap', 'P/L in money': 'pl_money', 'Cummulative money P/L': 'cumulative_money', 'P/L in pips': 'pl_pips', 'Cummulative pips P/L': 'cumulative_pips', 'P/L in %': 'pl_percent', 'Cummulative % P/L': 'cumulative_percent', 'Comment': 'comment', 'Sample type': 'sample_type' } def __init__(self, csv_path: str): self.csv_path = csv_path self.trades = [] self.stats = {} def parse_duration(self, duration_str: str) -> int: \"\"\"\"Convert duration string to seconds\"\"\" if not duration_str: return 0 total_seconds = 0 # Parse patterns like \"2d 3h 18m\", \"5h 59m\", \"1m 0s\", etc. days_match = re.search(r'(\d+)d', duration_str) hours_match = re.search(r'(\d+)h', duration_str) mins_match = re.search(r'(\d+)m', duration_str) secs_match = re.search(r'(\d+)s', duration_str) if days_match: total_seconds += int(days_match.group(1)) * 86400 if hours_match: total_seconds += int(hours_match.group(1)) * 3600 if mins_match: total_seconds += int(mins_match.group(1)) * 60 if secs_match: total_seconds += int(secs_match.group(1)) return total_seconds def parse_date(self, date_str: str) -> datetime: \"\"\"\"Parse date string to datetime object\"\"\" formats = ['%d.%m.%Y %H:%M:%S', '%Y.%m.%d %H:%M:%S', '%m/%d/%Y %H:%M:%S'] for fmt in formats: try: return datetime.strptime(date_str.strip(), fmt) except ValueError: continue return datetime.now() def parse_csv(self) -> List[Dict]: \"\"\"Parse the CSV file and return list of trade dictionaries\"\"\" trades = [] with open(self.csv_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: trade = {} for col, key in self.COLUMN_MAPPING.items(): value = row.get(col, '') if col in ['Open time', 'Close time']: trade[key] = self.parse_date(value).isoformat() elif col in ['Open price', 'Close price', 'Size', 'Profit/Loss', 'Cummulative P/L', 'Comm/Swap', 'P/L in money', 'Cummulative money P/L', 'P/L in pips', 'Cummulative pips P/L', 'P/L in %', 'Cummulative % P/L']: try: trade[key] = float(value) if value else 0.0 except ValueError: trade[key] = 0.0 elif col == 'Time in trade': trade['duration_seconds'] = self.parse_duration(value) trade['duration_str'] = value else: trade[key] = value trades.append(trade) self.trades = trades return trades def get_statistics(self) -> Dict[str, Any]: \"\"\"Calculate comprehensive trading statistics\"\"\" if not self.trades: self.parse_csv() trades = self.trades # Basic counts total_trades = len(trades) buy_trades = [t for t in trades if t['type'] == 'Buy'] sell_trades = [t for t in trades if t['type'] == 'Sell'] # Profit/Loss analysis profits = [t['profit_loss'] for t in trades if t['profit_loss'] > 0] losses = [t['profit_loss'] for t in trades if t['profit_loss'] < 0] winning_trades = [t for t in trades if t['profit_loss'] > 0] losing_trades = [t for t in trades if t['profit_loss'] < 0] gross_profit = sum(profits) if profits else 0 gross_loss = abs(sum(losses)) if losses else 0 net_profit = sum(t['profit_loss'] for t in trades) # Win rate win_rate = len(winning_trades) / total_trades * 100 if total_trades > 0 else 0 # Profit Factor profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf') # Duration analysis durations = [t['duration_seconds'] for t in trades] avg_duration = statistics.mean(durations) if durations else 0 # Pips analysis pl_pips = [t['pl_pips'] for t in trades] total_pips = sum(pl_pips) # Time-based analysis open_times = [self.parse_date(t['open_time']) for t in trades] # Group by hour of day hourly_stats = defaultdict(lambda: {'trades': 0, 'profit': 0, 'wins': 0, 'losses': 0}) for t in trades: dt = self.parse_date(t['open_time']) hour = dt.hour hourly_stats[hour]['trades'] += 1 hourly_stats[hour]['profit'] += t['profit_loss'] if t['profit_loss'] > 0: hourly_stats[hour]['wins'] += 1 else: hourly_stats[hour]['losses'] += 1 # Group by day of week daily_stats = defaultdict(lambda: {'trades': 0, 'profit': 0, 'wins': 0, 'losses': 0}) day_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] for t in trades: dt = self.parse_date(t['open_time']) day = day_names[dt.weekday()] daily_stats[day]['trades'] += 1 daily_stats[day]['profit'] += t['profit_loss'] if t['profit_loss'] > 0: daily_stats[day]['wins'] += 1 else: daily_stats[day]['losses'] += 1 # Consecutive win/loss streaks sorted_trades = sorted(trades, key=lambda x: x['open_time']) current_streak = 0 current_streak_type = None # 'win' or 'loss' max_win_streak = 0 max_loss_streak = 0 win_streaks = [] loss_streaks = [] for t in sorted_trades: if t['profit_loss'] > 0: if current_streak_type == 'win': current_streak += 1 else: if current_streak > 0 and current_streak_type == 'loss': loss_streaks.append(current_streak) current_streak = 1 current_streak_type = 'win' else: if current_streak_type == 'loss': current_streak += 1 else: if current_streak > 0 and current_streak_type == 'win': win_streaks.append(current_streak) current_streak = 1 current_streak_type = 'loss' if current_streak > 0: if current_streak_type == 'win': win_streaks.append(current_streak) else: loss_streaks.append(current_streak) max_win_streak = max(win_streaks) if win_streaks else 0 max_loss_streak = max(loss_streaks) if loss_streaks else 0 # Drawdown analysis cumulative = [t['cumulative_money'] for t in trades] peak = cumulative[0] if cumulative else 0 max_drawdown = 0 max_drawdown_percent = 0 for val in cumulative: if val > peak: peak = val drawdown = peak - val if drawdown > max_drawdown: max_drawdown = drawdown if peak > 0: dd_percent = (drawdown / peak) * 100 if dd_percent > max_drawdown_percent: max_drawdown_percent = dd_percent self.stats = { 'total_trades': total_trades, 'buy_trades': len(buy_trades), 'sell_trades': len(sell_trades), 'winning_trades': len(winning_trades), 'losing_trades': len(losing_trades), 'win_rate': round(win_rate, 2), 'gross_profit': round(gross_profit, 2), 'gross_loss': round(gross_loss, 2), 'net_profit': round(net_profit, 2), 'profit_factor': round(profit_factor, 2) if profit_factor != float('inf') else 'INF', 'avg_profit': round(statistics.mean(profits), 2) if profits else 0, 'avg_loss': round(statistics.mean(losses), 2) if losses else 0, 'best_trade': round(max(profits), 2) if profits else 0, 'worst_trade': round(min(losses), 2) if losses else 0, 'avg_duration_seconds': round(avg_duration, 0), 'total_pips': round(total_pips, 2), 'avg_pips': round(statistics.mean(pl_pips), 2) if pl_pips else 0, 'max_win_streak': max_win_streak, 'max_loss_streak': max_loss_streak, 'max_drawdown': round(max_drawdown, 2), 'max_drawdown_percent': round(max_drawdown_percent, 2), 'hourly_stats': dict(hourly_stats), 'daily_stats': dict(daily_stats), } return self.stats def to_json(self, output_path: Optional[str] = None) -> str: \"\"\"\"Export to JSON format for AI consumption\"\"\" if not self.trades: self.parse_csv() if not self.stats: self.get_statistics() output = { 'metadata': { 'source_file': self.csv_path, 'generated_at': datetime.now().isoformat(), 'total_trades': len(self.trades) }, 'statistics': self.stats, 'trades': self.trades } json_str = json.dumps(output, indent=2, ensure_ascii=False) if output_path: with open(output_path, 'w', encoding='utf-8') as f: f.write(json_str) return json_str