272 lines
10 KiB
Python
272 lines
10 KiB
Python
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
|