IB Analytics MCP Server

#!/usr/bin/env python3 """ Interactive Brokers Trading Performance Analysis Analyzes 2025 trading data from IB Flex Query """ import csv from datetime import datetime from collections import defaultdict from typing import Dict, List, Any def parse_csv_sections(filepath: str) -> Dict[str, List[Dict[str, str]]]: """Parse multi-section CSV file from IB Flex Query""" sections = {} current_section = None headers = [] with open(filepath, 'r', encoding='utf-8') as f: reader = csv.reader(f) for row in reader: if not row or not row[0]: continue # Detect new section by first column name if row[0] == 'ClientAccountID': headers = row # Identify section type from subsequent columns if 'Name' in headers: current_section = 'account_info' elif 'StartingCash' in headers: current_section = 'cash_summary' elif 'Quantity' in headers and 'MarkPrice' in headers: current_section = 'positions' elif 'TradeID' in headers: current_section = 'trades' if current_section: sections[current_section] = [] elif current_section and headers: # Parse data row row_dict = dict(zip(headers, row)) sections[current_section].append(row_dict) return sections def analyze_cash_flow(cash_summary: List[Dict]) -> Dict[str, float]: """Analyze cash flow metrics""" if not cash_summary: return {} data = cash_summary[0] return { 'starting_cash': float(data.get('StartingCash', 0)), 'ending_cash': float(data.get('EndingCash', 0)), 'deposits': float(data.get('Deposits', 0)), 'withdrawals': float(data.get('Withdrawals', 0)), 'internal_transfers': float(data.get('InternalTransfers', 0)), 'commissions': float(data.get('CommissionsYTD', 0)), 'dividends': float(data.get('Dividends', 0)), 'broker_interest': float(data.get('BrokerInterest', 0)), 'broker_fees': float(data.get('BrokerFees', 0)), 'net_trades_sales': float(data.get('NetTradesSales', 0)), 'net_trades_purchases': float(data.get('NetTradesPurchases', 0)), 'other_fees': float(data.get('OtherFees', 0)), 'transaction_tax': float(data.get('TransactionTax', 0)), } def analyze_positions(positions: List[Dict]) -> Dict[str, Any]: """Analyze current positions""" if not positions: return {} total_value = 0 total_cost = 0 total_unrealized_pnl = 0 asset_classes = defaultdict(float) position_details = [] for pos in positions: value = float(pos.get('PositionValue', 0)) cost = float(pos.get('CostBasisMoney', 0)) unrealized = float(pos.get('FifoPnlUnrealized', 0)) asset_class = pos.get('AssetClass', 'Unknown') total_value += value total_cost += cost total_unrealized_pnl += unrealized asset_classes[asset_class] += value position_details.append({ 'symbol': pos.get('Symbol', ''), 'description': pos.get('Description', ''), 'asset_class': asset_class, 'quantity': float(pos.get('Quantity', 0)), 'mark_price': float(pos.get('MarkPrice', 0)), 'position_value': value, 'cost_basis': cost, 'unrealized_pnl': unrealized, 'percent_of_nav': float(pos.get('PercentOfNAV', 0)), }) return { 'total_position_value': total_value, 'total_cost_basis': total_cost, 'total_unrealized_pnl': total_unrealized_pnl, 'asset_allocation': dict(asset_classes), 'positions': position_details, } def calculate_performance_metrics(cash_flow: Dict, positions: Dict) -> Dict[str, Any]: """Calculate key performance metrics""" # Net deposits/withdrawals net_deposits = cash_flow.get('deposits', 0) - abs(cash_flow.get('withdrawals', 0)) net_internal_transfers = cash_flow.get('internal_transfers', 0) # Total invested capital total_invested = net_deposits + net_internal_transfers # Current total value (cash + positions) current_cash = cash_flow.get('ending_cash', 0) current_positions_value = positions.get('total_position_value', 0) total_value = current_cash + current_positions_value # Profit/Loss calculations unrealized_pnl = positions.get('total_unrealized_pnl', 0) # Trading activity gross_sales = cash_flow.get('net_trades_sales', 0) gross_purchases = abs(cash_flow.get('net_trades_purchases', 0)) trading_volume = gross_sales + gross_purchases # Costs total_commissions = abs(cash_flow.get('commissions', 0)) total_fees = abs(cash_flow.get('broker_fees', 0)) + abs(cash_flow.get('other_fees', 0)) total_costs = total_commissions + total_fees # Income dividends = cash_flow.get('dividends', 0) interest = cash_flow.get('broker_interest', 0) total_income = dividends + interest # Overall P&L realized_pnl = total_value - total_invested - total_income + total_costs total_pnl = realized_pnl + unrealized_pnl # Return percentage roi_pct = (total_pnl / total_invested * 100) if total_invested != 0 else 0 return { 'total_invested': total_invested, 'current_value': total_value, 'current_cash': current_cash, 'current_positions_value': current_positions_value, 'realized_pnl': realized_pnl, 'unrealized_pnl': unrealized_pnl, 'total_pnl': total_pnl, 'roi_percentage': roi_pct, 'trading_volume': trading_volume, 'gross_sales': gross_sales, 'gross_purchases': gross_purchases, 'total_commissions': total_commissions, 'total_fees': total_fees, 'total_costs': total_costs, 'dividends': dividends, 'interest': interest, 'total_income': total_income, } def generate_report(sections: Dict, metrics: Dict, cash_flow: Dict, positions: Dict): """Generate comprehensive performance report""" print("=" * 80) print("INTERACTIVE BROKERS - 2025 TRADING PERFORMANCE ANALYSIS") print("=" * 80) print() # Account Info if 'account_info' in sections and sections['account_info']: acc = sections['account_info'][0] print(f"Account ID: {acc.get('ClientAccountID', 'N/A')}") print(f"Account Holder: {acc.get('Name', 'N/A')}") print(f"Account Type: {acc.get('AccountType', 'N/A')}") print(f"Base Currency: {acc.get('CurrencyPrimary', 'N/A')}") print() print("-" * 80) print("PORTFOLIO SUMMARY") print("-" * 80) print(f"Total Invested Capital: ${metrics['total_invested']:,.2f}") print(f"Current Cash: ${metrics['current_cash']:,.2f}") print(f"Current Positions Value: ${metrics['current_positions_value']:,.2f}") print(f"Total Portfolio Value: ${metrics['current_value']:,.2f}") print() print("-" * 80) print("PROFIT & LOSS") print("-" * 80) print(f"Realized P&L: ${metrics['realized_pnl']:,.2f}") print(f"Unrealized P&L: ${metrics['unrealized_pnl']:,.2f}") print(f"Total P&L: ${metrics['total_pnl']:,.2f}") print(f"Return on Investment: {metrics['roi_percentage']:.2f}%") print() print("-" * 80) print("TRADING ACTIVITY") print("-" * 80) print(f"Gross Sales: ${metrics['gross_sales']:,.2f}") print(f"Gross Purchases: ${metrics['gross_purchases']:,.2f}") print(f"Total Trading Volume: ${metrics['trading_volume']:,.2f}") print() print("-" * 80) print("COSTS & INCOME") print("-" * 80) print(f"Commissions: -${abs(metrics['total_commissions']):,.2f}") print(f"Fees: -${abs(metrics['total_fees']):,.2f}") print(f"Total Costs: -${abs(metrics['total_costs']):,.2f}") print(f"Dividends: ${metrics['dividends']:,.2f}") print(f"Interest: ${metrics['interest']:,.2f}") print(f"Total Income: ${metrics['total_income']:,.2f}") print() # Asset Allocation if positions.get('asset_allocation'): print("-" * 80) print("ASSET ALLOCATION") print("-" * 80) total_pos_value = positions['total_position_value'] for asset_class, value in positions['asset_allocation'].items(): pct = (value / total_pos_value * 100) if total_pos_value > 0 else 0 print(f"{asset_class:20s} ${value:>12,.2f} ({pct:>5.2f}%)") print() # Current Positions if positions.get('positions'): print("-" * 80) print("CURRENT POSITIONS") print("-" * 80) print(f"{'Symbol':<15} {'Asset':<8} {'Quantity':>12} {'Value':>15} {'Unrealized P&L':>15}") print("-" * 80) for pos in positions['positions']: print(f"{pos['symbol']:<15} {pos['asset_class']:<8} {pos['quantity']:>12,.0f} " f"${pos['position_value']:>13,.2f} ${pos['unrealized_pnl']:>13,.2f}") print() print("=" * 80) def main(): filepath = '/Users/ken/Developer/private/ib-sec/trades_2025.csv' # Parse CSV sections = parse_csv_sections(filepath) # Analyze data cash_flow = analyze_cash_flow(sections.get('cash_summary', [])) positions = analyze_positions(sections.get('positions', [])) metrics = calculate_performance_metrics(cash_flow, positions) # Generate report generate_report(sections, metrics, cash_flow, positions) if __name__ == '__main__': main()