""" πŸ—„οΈ TRADING DATABASE MODULE - SQLite Integration Ersetzt JSON Logging mit strukturierter Datenbank FEATURES: - Trade History Tracking - Performance Analytics - Session-based Queries - Confidence-based Analysis - Easy Migration von JSON """ import sqlite3 import json from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple import os class TradingDatabase: """ SQLite Database fΓΌr Trading Bot Performance Tracking """ def __init__(self, db_path: str = "trading_bot.db"): """ Initialize Database Connection Args: db_path: Path to SQLite database file """ self.db_path = db_path self.conn = None self.cursor = None self._connect() self._create_tables() def _connect(self): """Establish database connection""" self.conn = sqlite3.connect(self.db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row # Enable column access by name self.cursor = self.conn.cursor() def _create_tables(self): """Create all necessary tables""" # Trades Table - Main trade history self.cursor.execute(""" CREATE TABLE IF NOT EXISTS trades ( id INTEGER PRIMARY KEY AUTOINCREMENT, ticket INTEGER UNIQUE, position_id INTEGER, symbol TEXT NOT NULL, strategy_name TEXT NOT NULL, -- Trade Details type TEXT NOT NULL, -- 'BUY' or 'SELL' volume REAL NOT NULL, entry_price REAL NOT NULL, exit_price REAL, -- Stops sl_price REAL, tp_price REAL, -- Timing entry_time DATETIME NOT NULL, exit_time DATETIME, duration_hours REAL, -- Session Info session TEXT NOT NULL, -- 'asian', 'london', 'overlap', 'ny' regime TEXT, -- 'trending', 'ranging' quality TEXT, -- 'excellent', 'good', 'medium', 'poor' -- Signal Quality confidence REAL, timeframe_alignment INTEGER, -- How many timeframes aligned -- Performance profit REAL, commission REAL DEFAULT 0, swap REAL DEFAULT 0, net_profit REAL, profit_pct REAL, -- Risk Management risk_amount REAL, risk_pct REAL, rr_ratio REAL, -- Risk/Reward ratio -- Status status TEXT DEFAULT 'open', -- 'open', 'closed', 'cancelled' exit_reason TEXT, -- 'tp', 'sl', 'manual', 'timeout' -- Metadata created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # Performance Summary Table - Daily/Weekly aggregates self.cursor.execute(""" CREATE TABLE IF NOT EXISTS performance_summary ( id INTEGER PRIMARY KEY AUTOINCREMENT, date DATE NOT NULL UNIQUE, -- Trade Counts total_trades INTEGER DEFAULT 0, wins INTEGER DEFAULT 0, losses INTEGER DEFAULT 0, win_rate REAL DEFAULT 0, -- Profit gross_profit REAL DEFAULT 0, gross_loss REAL DEFAULT 0, net_profit REAL DEFAULT 0, profit_factor REAL DEFAULT 0, -- Session Breakdown asian_trades INTEGER DEFAULT 0, london_trades INTEGER DEFAULT 0, overlap_trades INTEGER DEFAULT 0, ny_trades INTEGER DEFAULT 0, asian_profit REAL DEFAULT 0, london_profit REAL DEFAULT 0, overlap_profit REAL DEFAULT 0, ny_profit REAL DEFAULT 0, -- Risk Metrics max_drawdown REAL DEFAULT 0, avg_trade REAL DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # Bot Status Table - Health monitoring self.cursor.execute(""" CREATE TABLE IF NOT EXISTS bot_status ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, status TEXT NOT NULL, -- 'running', 'stopped', 'error' version TEXT, active_sessions TEXT, -- JSON array of enabled sessions confidence_threshold REAL, -- Health Metrics uptime_hours REAL, last_trade_time DATETIME, total_positions INTEGER DEFAULT 0, error_message TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # Create indexes for common queries self.cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_trades_entry_time ON trades(entry_time) """) self.cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_trades_session ON trades(session) """) self.cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_trades_status ON trades(status) """) self.cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_trades_confidence ON trades(confidence) """) self.conn.commit() # ========================================== # INSERT OPERATIONS # ========================================== def log_trade_entry(self, trade_data: Dict) -> int: """ Log a new trade entry Args: trade_data: Dictionary with trade information Returns: trade_id: Database ID of inserted trade """ query = """ INSERT INTO trades ( ticket, position_id, symbol, strategy_name, type, volume, entry_price, sl_price, tp_price, entry_time, session, regime, quality, confidence, timeframe_alignment, risk_amount, risk_pct, status ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ values = ( trade_data.get('ticket'), trade_data.get('position_id'), trade_data.get('symbol'), trade_data.get('strategy_name'), trade_data.get('type'), trade_data.get('volume'), trade_data.get('entry_price'), trade_data.get('sl_price'), trade_data.get('tp_price'), trade_data.get('entry_time'), trade_data.get('session'), trade_data.get('regime'), trade_data.get('quality'), trade_data.get('confidence'), trade_data.get('timeframe_alignment'), trade_data.get('risk_amount'), trade_data.get('risk_pct'), 'open' ) self.cursor.execute(query, values) self.conn.commit() return self.cursor.lastrowid def update_trade_exit(self, ticket: int, exit_data: Dict): """ Update trade with exit information Args: ticket: MT5 ticket number exit_data: Dictionary with exit information """ query = """ UPDATE trades SET exit_price = ?, exit_time = ?, duration_hours = ?, profit = ?, commission = ?, swap = ?, net_profit = ?, profit_pct = ?, rr_ratio = ?, status = 'closed', exit_reason = ?, updated_at = CURRENT_TIMESTAMP WHERE ticket = ? """ values = ( exit_data.get('exit_price'), exit_data.get('exit_time'), exit_data.get('duration_hours'), exit_data.get('profit'), exit_data.get('commission', 0), exit_data.get('swap', 0), exit_data.get('net_profit'), exit_data.get('profit_pct'), exit_data.get('rr_ratio'), exit_data.get('exit_reason'), ticket ) self.cursor.execute(query, values) self.conn.commit() def close_trade(self, ticket: int, exit_price: float, exit_time: str, profit: float, status: str = 'closed', exit_reason: str = None, commission: float = 0, swap: float = 0): """ Simplified wrapper for closing a trade (used by Position Monitor) Args: ticket: MT5 ticket number exit_price: Exit price exit_time: Exit datetime profit: Trade profit status: Trade status (default 'closed') exit_reason: Reason for exit ('tp', 'sl', 'manual', etc.) commission: Commission paid swap: Swap paid """ from datetime import datetime # Calculate duration if we have entry_time duration_hours = None try: query = "SELECT entry_time FROM trades WHERE ticket = ?" self.cursor.execute(query, (ticket,)) row = self.cursor.fetchone() if row: entry_time = datetime.fromisoformat(row[0]) if isinstance(exit_time, str): exit_dt = datetime.fromisoformat(exit_time) else: exit_dt = exit_time duration_hours = (exit_dt - entry_time).total_seconds() / 3600 except: pass net_profit = profit - commission - swap exit_data = { 'exit_price': exit_price, 'exit_time': exit_time if isinstance(exit_time, str) else exit_time.isoformat(), 'duration_hours': duration_hours, 'profit': profit, 'commission': commission, 'swap': swap, 'net_profit': net_profit, 'exit_reason': exit_reason, 'profit_pct': None, # Would need entry data to calculate 'rr_ratio': None # Would need entry data to calculate } self.update_trade_exit(ticket, exit_data) def get_open_trades(self) -> List[Dict]: """ Get all currently open trades from database Returns: List of open trades as dictionaries """ query = """ SELECT ticket, position_id, symbol, strategy_name, type, volume, entry_price, sl_price, tp_price, entry_time, session, regime, quality, confidence, timeframe_alignment, risk_amount, risk_pct, status FROM trades WHERE status = 'open' ORDER BY entry_time DESC """ self.cursor.execute(query) rows = self.cursor.fetchall() # Convert to list of dictionaries trades = [] for row in rows: trades.append(dict(row)) return trades def log_bot_status(self, status_data: Dict): """ Log bot status for health monitoring Args: status_data: Dictionary with bot status information """ query = """ INSERT INTO bot_status ( status, version, active_sessions, confidence_threshold, uptime_hours, last_trade_time, total_positions, error_message ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """ values = ( status_data.get('status', 'running'), status_data.get('version'), json.dumps(status_data.get('active_sessions', [])), status_data.get('confidence_threshold'), status_data.get('uptime_hours'), status_data.get('last_trade_time'), status_data.get('total_positions', 0), status_data.get('error_message') ) self.cursor.execute(query, values) self.conn.commit() # ========================================== # QUERY OPERATIONS # ========================================== def get_recent_trades(self, limit: int = 50) -> List[Dict]: """Get most recent trades""" query = """ SELECT * FROM trades ORDER BY entry_time DESC LIMIT ? """ self.cursor.execute(query, (limit,)) return [dict(row) for row in self.cursor.fetchall()] def get_open_positions(self) -> List[Dict]: """Get all currently open positions""" query = """ SELECT * FROM trades WHERE status = 'open' ORDER BY entry_time DESC """ self.cursor.execute(query) return [dict(row) for row in self.cursor.fetchall()] def get_session_performance(self, days: int = 30) -> Dict: """ Get performance breakdown by session Args: days: Number of days to analyze Returns: Dictionary with session statistics """ query = """ SELECT session, COUNT(*) as trade_count, SUM(CASE WHEN net_profit > 0 THEN 1 ELSE 0 END) as wins, SUM(CASE WHEN net_profit < 0 THEN 1 ELSE 0 END) as losses, ROUND(AVG(CASE WHEN net_profit > 0 THEN 1.0 ELSE 0.0 END) * 100, 2) as win_rate, ROUND(SUM(net_profit), 2) as total_profit, ROUND(AVG(net_profit), 2) as avg_profit FROM trades WHERE status = 'closed' AND entry_time >= datetime('now', '-' || ? || ' days') GROUP BY session ORDER BY total_profit DESC """ self.cursor.execute(query, (days,)) results = {} for row in self.cursor.fetchall(): results[row['session']] = { 'count': row['trade_count'], 'wins': row['wins'], 'losses': row['losses'], 'win_rate': row['win_rate'], 'total_profit': row['total_profit'], 'avg_profit': row['avg_profit'] } return results def get_confidence_analysis(self, days: int = 30) -> Dict: """ Analyze performance by confidence levels Args: days: Number of days to analyze Returns: Dictionary with confidence-based statistics """ query = """ SELECT CASE WHEN confidence >= 80 THEN '80-100' WHEN confidence >= 70 THEN '70-79' WHEN confidence >= 60 THEN '60-69' ELSE '<60' END as confidence_range, COUNT(*) as trade_count, SUM(CASE WHEN net_profit > 0 THEN 1 ELSE 0 END) as wins, ROUND(AVG(CASE WHEN net_profit > 0 THEN 1.0 ELSE 0.0 END) * 100, 2) as win_rate, ROUND(SUM(net_profit), 2) as total_profit, ROUND(AVG(net_profit), 2) as avg_profit FROM trades WHERE status = 'closed' AND confidence IS NOT NULL AND entry_time >= datetime('now', '-' || ? || ' days') GROUP BY confidence_range ORDER BY confidence_range DESC """ self.cursor.execute(query, (days,)) results = {} for row in self.cursor.fetchall(): results[row['confidence_range']] = { 'count': row['trade_count'], 'wins': row['wins'], 'win_rate': row['win_rate'], 'total_profit': row['total_profit'], 'avg_profit': row['avg_profit'] } return results def get_daily_summary(self, date: str = None) -> Dict: """ Get daily performance summary Args: date: Date string (YYYY-MM-DD), defaults to today Returns: Dictionary with daily statistics """ if date is None: date = datetime.now().strftime('%Y-%m-%d') query = """ SELECT COUNT(*) as total_trades, SUM(CASE WHEN net_profit > 0 THEN 1 ELSE 0 END) as wins, SUM(CASE WHEN net_profit < 0 THEN 1 ELSE 0 END) as losses, ROUND(AVG(CASE WHEN net_profit > 0 THEN 1.0 ELSE 0.0 END) * 100, 2) as win_rate, ROUND(SUM(CASE WHEN net_profit > 0 THEN net_profit ELSE 0 END), 2) as gross_profit, ROUND(SUM(CASE WHEN net_profit < 0 THEN net_profit ELSE 0 END), 2) as gross_loss, ROUND(SUM(net_profit), 2) as net_profit, ROUND(AVG(net_profit), 2) as avg_trade FROM trades WHERE status = 'closed' AND DATE(entry_time) = ? """ self.cursor.execute(query, (date,)) row = self.cursor.fetchone() if row and row['total_trades'] > 0: return dict(row) else: return { 'total_trades': 0, 'wins': 0, 'losses': 0, 'win_rate': 0, 'gross_profit': 0, 'gross_loss': 0, 'net_profit': 0, 'avg_trade': 0 } def get_overall_statistics(self, days: int = None) -> Dict: """ Get overall performance statistics Args: days: Number of days to analyze (None = all time) Returns: Dictionary with comprehensive statistics """ date_filter = "" params = [] if days: date_filter = "AND entry_time >= datetime('now', '-' || ? || ' days')" params.append(days) query = f""" SELECT COUNT(*) as total_trades, SUM(CASE WHEN net_profit > 0 THEN 1 ELSE 0 END) as wins, SUM(CASE WHEN net_profit < 0 THEN 1 ELSE 0 END) as losses, ROUND(AVG(CASE WHEN net_profit > 0 THEN 1.0 ELSE 0.0 END) * 100, 2) as win_rate, ROUND(SUM(CASE WHEN net_profit > 0 THEN net_profit ELSE 0 END), 2) as gross_profit, ROUND(SUM(CASE WHEN net_profit < 0 THEN net_profit ELSE 0 END), 2) as gross_loss, ROUND(SUM(net_profit), 2) as net_profit, ROUND(AVG(net_profit), 2) as avg_trade, ROUND(SUM(commission), 2) as total_commission, ROUND(SUM(swap), 2) as total_swap, ROUND(AVG(confidence), 2) as avg_confidence, ROUND(AVG(duration_hours), 2) as avg_duration FROM trades WHERE status = 'closed' {date_filter} """ self.cursor.execute(query, params) row = self.cursor.fetchone() stats = dict(row) if row else {} # Calculate profit factor gross_profit = stats.get('gross_profit') or 0 gross_loss = stats.get('gross_loss') or 0 if gross_loss != 0: stats['profit_factor'] = round( abs(gross_profit / gross_loss), 2 ) else: stats['profit_factor'] = 0 return stats # ========================================== # UTILITY FUNCTIONS # ========================================== def migrate_from_json(self, json_file: str): """ Migrate existing JSON trade data to SQLite Args: json_file: Path to JSON performance file """ if not os.path.exists(json_file): print(f"❌ JSON file not found: {json_file}") return with open(json_file, 'r') as f: data = json.load(f) # Extract trades from JSON structure trades = data.get('trades', []) migrated = 0 skipped = 0 for trade in trades: try: # Check if trade already exists self.cursor.execute( "SELECT id FROM trades WHERE ticket = ?", (trade.get('ticket'),) ) if self.cursor.fetchone(): skipped += 1 continue # Insert trade self.log_trade_entry(trade) # If trade is closed, update exit data if trade.get('exit_time'): self.update_trade_exit( trade['ticket'], trade ) migrated += 1 except Exception as e: print(f"⚠️ Error migrating trade {trade.get('ticket')}: {e}") print(f"βœ… Migration complete: {migrated} trades migrated, {skipped} skipped") def close(self): """Close database connection""" if self.conn: self.conn.close() def __enter__(self): """Context manager entry""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit""" self.close() # ========================================== # USAGE EXAMPLE # ========================================== if __name__ == "__main__": # Initialize database db = TradingDatabase("trading_bot.db") print("="*70) print("πŸ—„οΈ TRADING DATABASE - System Check") print("="*70) # Check tables db.cursor.execute(""" SELECT name FROM sqlite_master WHERE type='table' ORDER BY name """) tables = db.cursor.fetchall() print(f"\nπŸ“Š Tables created: {len(tables)}") for table in tables: print(f" βœ… {table['name']}") # Get overall stats stats = db.get_overall_statistics() print(f"\nπŸ“ˆ Overall Statistics:") print(f" Total Trades: {stats.get('total_trades', 0)}") print(f" Win Rate: {stats.get('win_rate', 0)}%") print(f" Net Profit: ${stats.get('net_profit', 0)}") db.close() print("\n" + "="*70) print("βœ… Database ready!") print("="*70)