pairs_trading/lib/utils/db_inspector.py

#!/usr/bin/env python3
"""
Database inspector utility for pairs trading results database.
Provides functionality to view all tables and their contents.
"""

import sqlite3
import sys
import json
import os
from typing import List, Dict, Any

def list_tables(db_path: str) -> List[str]:
    """List all tables in the database."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT name FROM sqlite_master 
        WHERE type='table' 
        ORDER BY name
    """)
    
    tables = [row[0] for row in cursor.fetchall()]
    conn.close()
    return tables

def view_table_schema(db_path: str, table_name: str) -> None:
    """View the schema of a specific table."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute(f"PRAGMA table_info({table_name})")
    columns = cursor.fetchall()
    
    print(f"\nTable: {table_name}")
    print("-" * 50)
    print("Column Name".ljust(20) + "Type".ljust(15) + "Not Null".ljust(10) + "Default")
    print("-" * 50)
    
    for col in columns:
        cid, name, type_, not_null, default_value, pk = col
        print(f"{name}".ljust(20) + f"{type_}".ljust(15) + f"{bool(not_null)}".ljust(10) + f"{default_value or ''}")
    
    conn.close()

def view_config_table(db_path: str, limit: int = 10) -> None:
    """View entries from the config table."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute(f"""
        SELECT id, run_timestamp, config_file_path, fit_method_class, 
               datafiles, instruments, config_json
        FROM config 
        ORDER BY run_timestamp DESC
        LIMIT {limit}
    """)
    
    rows = cursor.fetchall()
    
    if not rows:
        print("No configuration entries found.")
        return
    
    print(f"\nMost recent {len(rows)} configuration entries:")
    print("=" * 80)
    
    for row in rows:
        id, run_timestamp, config_file_path, fit_method_class, datafiles, instruments, config_json = row
        
        print(f"ID: {id} | {run_timestamp}")
        print(f"Config: {config_file_path} | Strategy: {fit_method_class}")
        print(f"Files: {datafiles}")
        print(f"Instruments: {instruments}")
        print("-" * 80)
    
    conn.close()

def view_results_summary(db_path: str) -> None:
    """View summary of trading results."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # Get results summary
    cursor.execute("""
        SELECT date, COUNT(*) as trade_count, 
               ROUND(SUM(symbol_return), 2) as total_return
        FROM pt_bt_results 
        GROUP BY date 
        ORDER BY date DESC
    """)
    
    results = cursor.fetchall()
    
    if not results:
        print("No trading results found.")
        return
    
    print(f"\nTrading Results Summary:")
    print("-" * 50)
    print("Date".ljust(15) + "Trades".ljust(10) + "Total Return %")
    print("-" * 50)
    
    for date, trade_count, total_return in results:
        print(f"{date}".ljust(15) + f"{trade_count}".ljust(10) + f"{total_return}")
    
    # Get outstanding positions summary
    cursor.execute("""
        SELECT COUNT(*) as position_count,
               ROUND(SUM(unrealized_return), 2) as total_unrealized
        FROM outstanding_positions
    """)
    
    outstanding = cursor.fetchone()
    if outstanding and outstanding[0] > 0:
        print(f"\nOutstanding Positions: {outstanding[0]} positions")
        print(f"Total Unrealized Return: {outstanding[1]}%")
    
    conn.close()

def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: python db_inspector.py <database_path> [command]")
        print("Commands:")
        print("  tables     - List all tables")
        print("  schema     - Show schema for all tables")
        print("  config     - View configuration entries")
        print("  results    - View trading results summary")
        print("  all        - Show everything (default)")
        print("\nExample: python db_inspector.py results/equity.db config")
        sys.exit(1)
    
    db_path = sys.argv[1]
    command = sys.argv[2] if len(sys.argv) > 2 else "all"
    
    if not os.path.exists(db_path):
        print(f"Database file not found: {db_path}")
        sys.exit(1)
    
    try:
        if command in ["tables", "all"]:
            tables = list_tables(db_path)
            print(f"Tables in database: {', '.join(tables)}")
        
        if command in ["schema", "all"]:
            tables = list_tables(db_path)
            for table in tables:
                view_table_schema(db_path, table)
        
        if command in ["config", "all"]:
            if "config" in list_tables(db_path):
                view_config_table(db_path)
            else:
                print("Config table not found.")
        
        if command in ["results", "all"]:
            if "pt_bt_results" in list_tables(db_path):
                view_results_summary(db_path)
            else:
                print("Results table not found.")
                
    except Exception as e:
        print(f"Error inspecting database: {str(e)}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    main()