#!/usr/bin/env python3 """ Trade Log Report Generator Reads trade CSV files and generates detailed inspection reports. Usage: python3 trade_report.py # latest trade file python3 trade_report.py results/trades_XXX.csv # specific file python3 trade_report.py --all # all trade files in results/ """ import csv, sys, os, glob, json from collections import defaultdict from datetime import datetime os.chdir(os.path.dirname(os.path.abspath(__file__))) def load_trades(filepath): trades = [] with open(filepath) as f: reader = csv.DictReader(f) for row in reader: trades.append({ 'id': int(row['trade_id']), 'side': row['side'], 'lots': float(row['lots']), 'entry_price': float(row['entry_price']), 'entry_time': row['entry_time'], 'exit_price': float(row['exit_price']), 'exit_time': row['exit_time'], 'profit': float(row['profit']), 'thread': row.get('magic_thread', ''), 'comment': row.get('comment', ''), 'duration': float(row.get('duration_min', 0)), 'return_pct': float(row.get('return_pct', 0)), }) return trades def generate_report(trades, filepath, starting_balance=20000): report = [] report.append("=" * 80) report.append(f" TRADE LOG REPORT") report.append(f" File: {os.path.basename(filepath)}") report.append(f" Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") report.append("=" * 80) # === SUMMARY === total_profit = sum(t['profit'] for t in trades) winners = [t for t in trades if t['profit'] > 0] losers = [t for t in trades if t['profit'] <= 0] gross_profit = sum(t['profit'] for t in winners) gross_loss = abs(sum(t['profit'] for t in losers)) report.append(f"\n--- SUMMARY ---") report.append(f"Total Trades: {len(trades):,}") report.append(f"Winners: {len(winners):,} ({len(winners)/len(trades)*100:.1f}%)") report.append(f"Losers: {len(losers):,} ({len(losers)/len(trades)*100:.1f}%)") report.append(f"Net Profit: ${total_profit:,.2f}") report.append(f"Gross Profit: ${gross_profit:,.2f}") report.append(f"Gross Loss: ${gross_loss:,.2f}") report.append(f"Profit Factor: {gross_profit/gross_loss:.2f}" if gross_loss > 0 else "Profit Factor: inf") report.append(f"Date Range: {trades[0]['entry_time'][:10]} to {trades[-1]['exit_time'][:10]}") # === LOT SIZE DISTRIBUTION === lots = [t['lots'] for t in trades] lot_buckets = defaultdict(int) for l in lots: bucket = round(l, 3) lot_buckets[bucket] += 1 sorted_lots = sorted(lot_buckets.items(), key=lambda x: x[1], reverse=True) report.append(f"\n--- LOT SIZE DISTRIBUTION ---") report.append(f"Min Lot: {min(lots):.4f}") report.append(f"Max Lot: {max(lots):.4f}") report.append(f"Avg Lot: {sum(lots)/len(lots):.4f}") report.append(f"Median: {sorted(lots)[len(lots)//2]:.4f}") report.append(f"\nTop 15 lot sizes:") report.append(f" {'Lots':>10} {'Count':>8} {'% of Total':>10}") for lot, count in sorted_lots[:15]: report.append(f" {lot:>10.4f} {count:>8,} {count/len(trades)*100:>9.1f}%") # === TRADE TYPE BREAKDOWN === type_stats = defaultdict(lambda: {'count': 0, 'profit': 0, 'lots_sum': 0}) for t in trades: comment = t['comment'] # Classify if 'ENTRY' in comment: ttype = 'ENTRY' elif 'REC_L' in comment and 'S' not in comment.split('REC_L')[1][:3]: ttype = 'RECOVERY_MAIN' elif 'SUB_RPT' in comment: ttype = 'SUB_REPEAT' elif 'REC_L' in comment: ttype = 'RECOVERY_SUB' else: ttype = comment[:20] if comment else 'UNKNOWN' type_stats[ttype]['count'] += 1 type_stats[ttype]['profit'] += t['profit'] type_stats[ttype]['lots_sum'] += t['lots'] report.append(f"\n--- TRADE TYPE BREAKDOWN ---") report.append(f" {'Type':<20} {'Count':>8} {'Net Profit':>12} {'Avg Lot':>10} {'Avg P&L':>10}") for ttype, stats in sorted(type_stats.items(), key=lambda x: x[1]['profit'], reverse=True): avg_lot = stats['lots_sum'] / stats['count'] if stats['count'] > 0 else 0 avg_pnl = stats['profit'] / stats['count'] if stats['count'] > 0 else 0 report.append(f" {ttype:<20} {stats['count']:>8,} ${stats['profit']:>11,.2f} {avg_lot:>10.4f} ${avg_pnl:>9.2f}") # === RECOVERY LEVEL ANALYSIS === level_stats = defaultdict(lambda: {'count': 0, 'profit': 0, 'max_lot': 0}) for t in trades: comment = t['comment'] level = 0 if 'REC_L' in comment: try: level = int(comment.split('REC_L')[1].split('_')[0]) except: pass level_stats[level]['count'] += 1 level_stats[level]['profit'] += t['profit'] level_stats[level]['max_lot'] = max(level_stats[level]['max_lot'], t['lots']) report.append(f"\n--- RECOVERY LEVEL ANALYSIS ---") report.append(f" {'Level':>6} {'Count':>8} {'Net Profit':>12} {'Max Lot':>10}") for level in sorted(level_stats.keys()): stats = level_stats[level] lbl = 'Entry' if level == 0 else f'L{level}' report.append(f" {lbl:>6} {stats['count']:>8,} ${stats['profit']:>11,.2f} {stats['max_lot']:>10.4f}") # === EQUITY CURVE (daily) === equity = starting_balance peak = starting_balance daily_equity = {} daily_trades = defaultdict(int) for t in trades: equity += t['profit'] peak = max(peak, equity) day = t['exit_time'][:10] daily_equity[day] = equity daily_trades[day] += 1 report.append(f"\n--- DAILY EQUITY ---") report.append(f" {'Date':<12} {'Equity':>14} {'Day P&L':>12} {'Trades':>7} {'DD from Peak':>12}") prev_eq = starting_balance for day in sorted(daily_equity.keys()): eq = daily_equity[day] day_pnl = eq - prev_eq dd = (eq - peak) / peak * 100 if peak > 0 else 0 report.append(f" {day:<12} ${eq:>13,.2f} ${day_pnl:>11,.2f} {daily_trades[day]:>7} {dd:>11.2f}%") prev_eq = eq # === BIGGEST TRADES === sorted_by_profit = sorted(trades, key=lambda x: x['profit'], reverse=True) report.append(f"\n--- TOP 10 WINNING TRADES ---") report.append(f" {'ID':>6} {'Side':<5} {'Lots':>8} {'Entry$':>10} {'Exit$':>10} {'P&L':>12} {'Comment':<25} {'Time'}") for t in sorted_by_profit[:10]: report.append(f" {t['id']:>6} {t['side']:<5} {t['lots']:>8.4f} {t['entry_price']:>10,.1f} {t['exit_price']:>10,.1f} ${t['profit']:>11,.2f} {t['comment']:<25} {t['exit_time']}") report.append(f"\n--- TOP 10 LOSING TRADES ---") for t in sorted_by_profit[-10:]: report.append(f" {t['id']:>6} {t['side']:<5} {t['lots']:>8.4f} {t['entry_price']:>10,.1f} {t['exit_price']:>10,.1f} ${t['profit']:>11,.2f} {t['comment']:<25} {t['exit_time']}") # === THREAD ANALYSIS === thread_stats = defaultdict(lambda: {'trades': 0, 'profit': 0, 'max_level': 0, 'max_lot': 0}) for t in trades: th = t['thread'] thread_stats[th]['trades'] += 1 thread_stats[th]['profit'] += t['profit'] thread_stats[th]['max_lot'] = max(thread_stats[th]['max_lot'], t['lots']) if 'REC_L' in t['comment']: try: lvl = int(t['comment'].split('REC_L')[1].split('_')[0]) thread_stats[th]['max_level'] = max(thread_stats[th]['max_level'], lvl) except: pass sorted_threads = sorted(thread_stats.items(), key=lambda x: x[1]['profit'], reverse=True) report.append(f"\n--- TOP 10 MOST PROFITABLE THREADS ---") report.append(f" {'Thread':>10} {'Trades':>7} {'Profit':>12} {'Max Lvl':>8} {'Max Lot':>10}") for th, stats in sorted_threads[:10]: report.append(f" {th:>10} {stats['trades']:>7} ${stats['profit']:>11,.2f} {'L'+str(stats['max_level']) if stats['max_level'] > 0 else 'Entry':>8} {stats['max_lot']:>10.4f}") report.append(f"\n--- TOP 10 WORST THREADS ---") for th, stats in sorted_threads[-10:]: report.append(f" {th:>10} {stats['trades']:>7} ${stats['profit']:>11,.2f} {'L'+str(stats['max_level']) if stats['max_level'] > 0 else 'Entry':>8} {stats['max_lot']:>10.4f}") return "\n".join(report) def main(): if len(sys.argv) > 1 and sys.argv[1] == '--all': files = sorted(glob.glob("results/trades_*.csv")) elif len(sys.argv) > 1 and os.path.exists(sys.argv[1]): files = [sys.argv[1]] else: files = sorted(glob.glob("results/trades_*.csv"), reverse=True)[:1] if not files: print("No trade files found in results/") return balance = float(sys.argv[2]) if len(sys.argv) > 2 else 20000.0 for filepath in files: trades = load_trades(filepath) report = generate_report(trades, filepath, balance) # Save report report_path = filepath.replace('trades_', 'report_').replace('.csv', '.txt') with open(report_path, 'w') as f: f.write(report) print(report) print(f"\nReport saved: {report_path}") if __name__ == "__main__": main()