cbazza/Place-Order-Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
596718e30b50f0ff55d552410fd0cdfc09e2d142
Place-Order-Trading-Bot/trading_database.py
T

631 lines
19 KiB
Python
Raw Normal View History

feat: Trading Bot V1.8 - Aggressive Mode + Infrastructure
2025-11-26 21:14:01 +01:00
"""
🗄️ 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 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)
Reference in New Issue Copy Permalink