Database.py

from pathlib import Path
from datetime import datetime
from User import Users, User
from Exchange import Exchange, _resolve_ccxt_symbol_from_mapping
from ccxt.base.errors import BadSymbol
from pbgui_purefunc import PBGDIR
from logging_helpers import human_log as _human_log

SERVICE = "Database"
import sqlite3
import json
import time
import threading
from functools import wraps
from uuid import uuid4
from database_lock import DatabaseBusyError, acquire_database_lock
from master_update_lock import acquire_master_runtime_lock
from secure_files import ensure_private_directory
from sqlite_backup import backup_sqlite_database, restore_sqlite_backup

DB_MAINTENANCE_PROTOCOL = 1


def _guard_database_writes(cls):
    """Lease complete write operations, never the lifetime of cached connections.

    Include leaf helpers because legacy callers also invoke them directly. Nested
    SH leases are supported; history/restore/backup already own their admission.
    """
    def guarded(method):
        """Wrap one method without changing its signature or return contract."""
        @wraps(method)
        def call(self, *args, **kwargs):
            """Refuse startup or mutation while recovery/maintenance owns the DB."""
            try:
                lease = acquire_database_lock(Path(PBGDIR))
            except DatabaseBusyError:
                _human_log(SERVICE, f"Database operation {method.__name__} blocked by maintenance/recovery", level="WARNING")
                raise
            with lease:
                return method(self, *args, **kwargs)
        return call

    for name in (
        "__init__", "create_tables", "create_trades_tables", "_repair_bitget_execution_sides",
        "update_executions", "update_positions", "update_orders", "update_prices", "update_balances",
        "add_history", "add_position", "add_order", "add_price", "remove_position", "remove_order",
        "remove_price", "update_position", "update_order", "update_price", "upsert_price",
        "batch_upsert_prices", "update_balance", "set_last_scan_ts", "fetch_history",
        "delete_income_by_ids", "delete_income_older_than_user", "delete_income_older_than",
        "import_from_save_income_other",
    ):
        setattr(cls, name, guarded(getattr(cls, name)))
    return cls


@_guard_database_writes
class Database():
    def __init__(self):
        self.db = Path(f'{PBGDIR}/data/pbgui.db')
        # Separate DB for high-volume trade/execution storage.
        # Keeps the main pbgui.db lean for UI queries and existing history logic.
        self.trades_db = Path(f'{PBGDIR}/data/pbgui_trades.db')
        # Global write lock to serialize all write operations
        self._write_lock = threading.Lock()
        # Thread-local storage for per-thread SQLite connections to avoid
        # repeated open/close syscalls (reduces openat/read activity).
        self._local = threading.local()
        self.create_tables()
        self.create_trades_tables()

    
    def _connect(self):
        """Return a per-thread cached SQLite connection.

        New connections get PRAGMA tuning applied. Connections are cached on
        `self._local.conn` so each worker thread reuses its own connection and
        we avoid repeated open/close syscalls that were visible in strace.
        """
        local = getattr(self, '_local', None)
        if local is None:
            # Fallback: create thread-local container
            self._local = threading.local()
            local = self._local

        conn = getattr(local, 'conn', None)
        try:
            # Quick health-check: if connection exists and is open, reuse it
            if conn is not None:
                try:
                    # lightweight check
                    conn.execute('SELECT 1')
                    return conn
                except Exception:
                    try:
                        conn.close()
                    except Exception:
                        pass
                    conn = None
        except Exception:
            conn = None

        # Create a fresh connection for this thread
        conn = sqlite3.connect(self.db, timeout=30)
        try:
            conn.execute('PRAGMA busy_timeout=30000')
        except Exception:
            pass
        # Apply journaling PRAGMAs on new connections for consistency
        try:
            conn.execute('PRAGMA journal_mode=WAL')
            conn.execute('PRAGMA synchronous=NORMAL')
        except Exception:
            pass
        local.conn = conn
        return conn

    def _connect_trades(self):
        """Return a per-thread cached SQLite connection to the trades DB."""
        local = getattr(self, '_local', None)
        if local is None:
            self._local = threading.local()
            local = self._local

        conn = getattr(local, 'trades_conn', None)
        try:
            if conn is not None:
                try:
                    conn.execute('SELECT 1')
                    return conn
                except Exception:
                    try:
                        conn.close()
                    except Exception:
                        pass
                    conn = None
        except Exception:
            conn = None

        conn = sqlite3.connect(self.trades_db, timeout=30)
        try:
            conn.execute('PRAGMA busy_timeout=30000')
        except Exception:
            pass
        try:
            conn.execute('PRAGMA journal_mode=WAL')
            conn.execute('PRAGMA synchronous=NORMAL')
        except Exception:
            pass
        local.trades_conn = conn
        return conn
    
    def backup_full_db(self, keep_last: int = 10):
        """Publish a complete SQLite snapshot, including committed WAL records."""
        try:
            backups_dir = ensure_private_directory(Path(PBGDIR) / 'data' / 'backup' / 'db')
            ts = datetime.now().strftime('%Y%m%d-%H%M%S')
            backup_path = backups_dir / f'pbgui-{ts}-{uuid4().hex}.db'
            with acquire_database_lock(Path(PBGDIR)):
                backup_sqlite_database(self.db, backup_path)
            # Rotate: keep only the last N backups by modified time
            try:
                backups = sorted(
                    [p for p in backups_dir.glob('pbgui-*.db') if p.is_file()],
                    key=lambda p: p.stat().st_mtime,
                    reverse=True
                )
                if keep_last is not None and keep_last > 0 and len(backups) > keep_last:
                    for old in backups[keep_last:]:
                        try:
                            old.unlink()
                        except Exception:
                            pass
            except Exception:
                pass
            _human_log(
                SERVICE,
                f"Created full database backup {backup_path.name}",
                meta={"operation": "backup_full_db"},
            )
            return str(backup_path)
        except Exception as e:
            try:
                _human_log(SERVICE, f"backup_full_db error: {e}", level='ERROR', meta={"operation": "backup_full_db"})
            except Exception:
                pass
            return None

    def restore_db_from(self, backup_path: str):
        """Restore via SQLite without replacing files used by cached connections."""
        try:
            with acquire_master_runtime_lock(Path(PBGDIR)), acquire_database_lock(Path(PBGDIR), exclusive=True):
                restore_sqlite_backup(
                    Path(backup_path), self.db, Path(PBGDIR) / 'data' / 'backup' / 'db',
                )
            _human_log(
                SERVICE,
                "Restored database from a validated SQLite snapshot",
                meta={"operation": "restore_db_from"},
            )
            return True
        except Exception as e:
            try:
                _human_log(SERVICE, f"restore_db_from error: {e}", level='ERROR', meta={"operation": "restore_db_from"})
            except Exception:
                pass
            return False

    # --- Change-detection helpers removed: fragments auto-refresh panels ---

    def create_tables(self):
        sql_statements = [ 
            """CREATE TABLE IF NOT EXISTS history (
                    id INTEGER PRIMARY KEY,
                    symbol TEXT NOT NULL,
                    timestamp INTEGER NOT NULL,
                    income REAL NOT NULL,
                    uniqueid text NOT NULL UNIQUE,
                    user TEXT NOT NULL
            );""",
            """CREATE TABLE IF NOT EXISTS position (
                    id INTEGER PRIMARY KEY,
                    symbol TEXT NOT NULL,
                    timestamp INTEGER NOT NULL,
                    psize REAL NOT NULL,
                    upnl REAL NOT NULL,
                    entry REAL NOT NULL,
                    user TEXT NOT NULL
            );""",
            """CREATE TABLE IF NOT EXISTS orders (
                    id INTEGER PRIMARY KEY,
                    symbol TEXT NOT NULL,
                    timestamp INTEGER NOT NULL,
                    amount REAL NOT NULL,
                    price REAL NOT NULL,
                    side TEXT NOT NULL,
                    uniqueid text NOT NULL UNIQUE,
                    user TEXT NOT NULL
            );""",
            """CREATE TABLE IF NOT EXISTS prices (
                    id INTEGER PRIMARY KEY,
                    symbol TEXT NOT NULL,
                    timestamp INTEGER NOT NULL,
                    price REAL NOT NULL,
                    user TEXT NOT NULL
            );""",
            """CREATE TABLE IF NOT EXISTS balances (
                    id INTEGER PRIMARY KEY,
                    timestamp INTEGER NOT NULL,
                    balance REAL NOT NULL,
                    user TEXT NOT NULL UNIQUE
            );""",
            """CREATE TABLE IF NOT EXISTS history_scan_meta (
                    user TEXT NOT NULL,
                    exchange TEXT NOT NULL,
                    last_scan_ts INTEGER NOT NULL,
                    PRIMARY KEY (user, exchange)
            );"""
            ]
        # create a database connection
        try:
            with self._connect() as conn:
                cursor = conn.cursor()
                for statement in sql_statements:
                    cursor.execute(statement)
                conn.commit()
                # Bugfix old database
                # Check if the 'side' column exists in the 'position' table
                cursor.execute("PRAGMA table_info(position);")
                columns = [column[1] for column in cursor.fetchall()]
                if 'side' not in columns:
                    # Add the 'side' column if it does not exist
                    cursor.execute("ALTER TABLE position ADD COLUMN side TEXT;")
                    conn.commit()
                    # Update existing records in the 'position' table to set 'side' to 'long'
                    cursor.execute("UPDATE position SET side = 'long';")
                    conn.commit()
                # Improve concurrency across multiple writers
                try:
                    cursor.execute('PRAGMA journal_mode=WAL')
                    cursor.execute('PRAGMA synchronous=NORMAL')
                    conn.commit()
                except Exception:
                    pass
                # Create indexes to speed up frequent WHERE queries and avoid full-table scans
                try:
                    cursor.execute("CREATE INDEX IF NOT EXISTS idx_position_user ON position(user)")
                    cursor.execute("CREATE INDEX IF NOT EXISTS idx_orders_user ON orders(user)")
                    cursor.execute("CREATE INDEX IF NOT EXISTS idx_prices_user ON prices(user)")
                    cursor.execute("CREATE INDEX IF NOT EXISTS idx_prices_symbol_user ON prices(symbol, user)")
                    cursor.execute("CREATE INDEX IF NOT EXISTS idx_history_user_ts ON history(user, timestamp)")
                    conn.commit()
                except Exception:
                    pass
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB create_tables error: {e}", level='ERROR')

    def create_trades_tables(self):
        sql_statements = [
            """CREATE TABLE IF NOT EXISTS executions (
                    id INTEGER PRIMARY KEY,
                    exchange TEXT NOT NULL,
                    symbol TEXT NOT NULL,
                    timestamp INTEGER NOT NULL,
                    side TEXT,
                    price REAL,
                    qty REAL,
                    fee REAL,
                    realized_pnl REAL,
                    order_id TEXT,
                    trade_id TEXT NOT NULL,
                    user TEXT NOT NULL,
                    raw_json TEXT,
                    UNIQUE(user, exchange, trade_id)
            );""",
        ]
        try:
            with self._connect_trades() as conn:
                cur = conn.cursor()
                for statement in sql_statements:
                    cur.execute(statement)
                try:
                    cur.execute('PRAGMA journal_mode=WAL')
                    cur.execute('PRAGMA synchronous=NORMAL')
                except Exception:
                    pass
                try:
                    cur.execute('CREATE INDEX IF NOT EXISTS idx_exec_user_ts ON executions(user, timestamp)')
                    cur.execute('CREATE INDEX IF NOT EXISTS idx_exec_user_symbol_ts ON executions(user, symbol, timestamp)')
                    cur.execute('CREATE INDEX IF NOT EXISTS idx_exec_user_exchange_ts ON executions(user, exchange, timestamp)')
                except Exception:
                    pass
                conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB create_trades_tables error: {e}", level='ERROR')

    def close_thread_connections(self):
        """Close any cached per-thread SQLite connection(s).

        This is safe to call from shutdown code to release file descriptors
        and ensure rotated/deleted files are freed promptly.
        """
        try:
            local = getattr(self, '_local', None)
            if local is None:
                return
            conn = getattr(local, 'conn', None)
            if conn is not None:
                try:
                    conn.close()
                except Exception:
                    pass
                try:
                    delattr(local, 'conn')
                except Exception:
                    pass
            tconn = getattr(local, 'trades_conn', None)
            if tconn is not None:
                try:
                    tconn.close()
                except Exception:
                    pass
                try:
                    delattr(local, 'trades_conn')
                except Exception:
                    pass
        except Exception:
            pass

    def update_history(self, user: User):
        """Keep the complete incremental scan/write cycle on one side of a restore."""
        try:
            lease = acquire_database_lock(Path(PBGDIR))
        except DatabaseBusyError:
            _human_log(SERVICE, "History update deferred during database restore or installation", level='INFO', user=user.name)
            return
        with lease:
            _human_log(SERVICE, f"update_history called for user={getattr(user, 'name', user)}", level='INFO', user=getattr(user, 'name', user))
            try:
                scan_started_ms = int(time.time() * 1000)
                history = self.fetch_history(user)
            except Exception as e:
                # Do not write partial history after an exchange fetch failure.
                _human_log(SERVICE, f"update_history aborting for user={user.name}: {e}", level='WARNING', user=user.name)
                return
            # The shared maintenance lease spans the network fetch, but ordinary
            # price/position writes remain independent of the short write lock.
            try:
                if not isinstance(history, list):
                    raise ValueError("History scan did not return a complete list")
                rows = [
                    (line['symbol'], line['timestamp'], line['income'], line['uniqueid'], user.name)
                    for line in history
                ]
                with self._write_lock:
                    with self._connect() as conn:
                        # Only duplicate IDs are harmless. All other write failures
                        # roll back the entire batch together with its checkpoint.
                        conn.executemany(
                            '''INSERT INTO history(symbol,timestamp,income,uniqueid,user)
                               VALUES(?,?,?,?,?) ON CONFLICT(uniqueid) DO NOTHING''',
                            rows,
                        )
                        conn.execute(
                            '''INSERT INTO history_scan_meta(user,exchange,last_scan_ts)
                               VALUES(?,?,?) ON CONFLICT(user,exchange)
                               DO UPDATE SET last_scan_ts = excluded.last_scan_ts''',
                            (user.name, user.exchange, scan_started_ms),
                        )
            except (sqlite3.Error, KeyError, TypeError, ValueError) as e:
                _human_log(SERVICE, f"DB update_history error for {user.name}: {e}", level='ERROR', user=user.name)

    def find_last_execution_timestamp(self, user: User, exchange: str):
        """Return max(timestamp) from executions for this user/exchange."""
        sql = 'SELECT MAX(timestamp) FROM executions WHERE user = ? AND exchange = ?'
        try:
            with self._connect_trades() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user.name, exchange))
                row = cur.fetchone()
                ts = row[0] if row else None
                return int(ts) if ts is not None else None
        except Exception:
            return None

    def _repair_bitget_execution_sides(self, user: str):
        """Repair Bitget executions 'side' from raw_json.

        Classic swap trades often encode position direction in info.side and whether it
        opened/closed in info.tradeSide. For matching we want the *order* side:
        - open: same as info.side
        - close: opposite of info.side

        UTA executions already carry the order side and must not be inverted.
        Explicit provenance and native execId identify UTA rows, including older
        imports without provenance. Malformed JSON is left untouched.
        This is safe to run repeatedly.
        """

        try:
            with self._write_lock:
                with self._connect_trades() as conn:
                    cur = conn.cursor()
                    cur.execute(
                        """
                        UPDATE executions
                        SET side = CASE
                            WHEN NOT json_valid(raw_json) THEN side
                            WHEN lower(json_extract(raw_json, '$.pbgui_account_mode')) = 'uta' THEN side
                            WHEN json_type(raw_json, '$.info.execId') IN ('text', 'integer')
                                AND trim(CAST(json_extract(raw_json, '$.info.execId') AS TEXT)) <> '' THEN side
                            WHEN lower(json_extract(raw_json, '$.info.tradeSide')) = 'close' THEN
                                CASE lower(json_extract(raw_json, '$.info.side'))
                                    WHEN 'buy' THEN 'sell'
                                    WHEN 'sell' THEN 'buy'
                                    ELSE side
                                END
                            WHEN lower(json_extract(raw_json, '$.info.tradeSide')) = 'open' THEN
                                CASE lower(json_extract(raw_json, '$.info.side'))
                                    WHEN 'buy' THEN 'buy'
                                    WHEN 'sell' THEN 'sell'
                                    ELSE side
                                END
                            ELSE side
                        END
                        WHERE user = ? AND exchange = 'bitget' AND raw_json IS NOT NULL AND LENGTH(raw_json) > 0
                        """,
                        (user,),
                    )
                    conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB Bitget execution side repair error: {e}", level='ERROR', user=user)

    def update_executions(
        self,
        user: User,
        *,
        since_override: int | None = None,
        ignore_last_execution: bool = False,
        initial_lookback_days: int | None = None,
        symbol_lookback_days: int | None = None,
    ):
        """Fetch and store execution-level trades/fills into the trades DB.

        By default, this is an incremental updater: it starts from the last stored
        execution timestamp (with a small overlap), and applies exchange-specific
        caps for initial backfill.

        Backfill controls:
        - since_override: force fetching from this ms timestamp (UTC).
        - ignore_last_execution: ignore existing executions and treat as initial run.
        - initial_lookback_days: override initial backfill cap (days). Only used
          when there is no last execution timestamp or when ignore_last_execution.
        - symbol_lookback_days: override the income-symbol discovery window (days).
        """
        exchange = Exchange(user.exchange, user)

        # Only fetch executions for exchanges where we have explicit support.
        if exchange.id not in ('hyperliquid', 'binance', 'bitget', 'bybit', 'kucoinfutures', 'okx', 'gateio', 'bitunix', 'weex'):
            return None

        since = None
        if since_override is not None:
            try:
                since = int(since_override)
            except Exception:
                since = None
        elif not ignore_last_execution:
            try:
                since = self.find_last_execution_timestamp(user, exchange.id)
            except Exception:
                since = None
        # If we have previous executions, back up slightly to avoid boundary misses.
        try:
            if since is not None:
                since = max(0, int(since) - 5 * 60 * 1000)
        except Exception:
            since = None

        # Binance futures userTrades endpoints only allow querying recent history.
        # For initial backfill, cap to a recent window by default; subsequent runs use since=last_execution.
        now_ms = None
        day = 24 * 60 * 60 * 1000
        try:
            now_ms = int(datetime.now().timestamp() * 1000)
        except Exception:
            now_ms = None
        if exchange.id == 'binance' and now_ms is not None:
            try:
                default_days = 240
                cap_days = int(initial_lookback_days) if (since is None and initial_lookback_days is not None) else default_days
                cap_start = max(0, now_ms - cap_days * day)
                if since is None:
                    since = cap_start
                else:
                    # Keep incremental runs within the endpoint's history window.
                    since = max(int(since), max(0, now_ms - default_days * day))
            except Exception:
                pass

        # Bitget: request a 365-day initial lookback by default. Exchange applies
        # the account-mode-specific retention limits and pagination windows.
        if exchange.id == 'bitget' and now_ms is not None:
            try:
                if since is None:
                    if initial_lookback_days is not None:
                        since = max(0, now_ms - int(initial_lookback_days) * day)
                    else:
                        since = max(0, now_ms - 365 * day)
            except Exception:
                pass

        # KuCoin futures fills can be fetched without a symbol. For initial backfill,
        # keep the scan bounded to one year by default.
        if exchange.id == 'kucoinfutures' and now_ms is not None:
            try:
                if since is None:
                    if initial_lookback_days is not None:
                        since = max(0, now_ms - int(initial_lookback_days) * day)
                    else:
                        since = max(0, now_ms - 365 * day)
            except Exception:
                pass

        # Bybit: generally supports longer history, but for initial backfill keep it bounded.
        # Use 365 days by default; caller can override.
        if exchange.id == 'bybit' and now_ms is not None:
            try:
                if since is None:
                    if initial_lookback_days is not None:
                        since = max(0, now_ms - int(initial_lookback_days) * day)
                    else:
                        since = max(0, now_ms - 365 * day)
            except Exception:
                pass

        # OKX: CCXT fetchMyTrades uses fills-history (≈90-day lookback). For initial backfill,
        # default to 90 days; caller can override.
        if exchange.id == 'okx' and now_ms is not None:
            try:
                if since is None:
                    if initial_lookback_days is not None:
                        since = max(0, now_ms - int(initial_lookback_days) * day)
                    else:
                        since = max(0, now_ms - 90 * day)
                else:
                    # Clamp incremental runs to the supported lookback window.
                    since = max(int(since), max(0, now_ms - 90 * day))
            except Exception:
                pass

        # Gate.io: CCXT supports swap trades without symbol. For initial backfill, keep it bounded.
        # Note: Gate docs mention the non-time-range personal trades endpoint defaults to the past ~6 months
        # and recommends using the *_timerange endpoint for longer periods. PBGui defaults to 365 days here,
        # while allowing callers to override via initial_lookback_days.
        if exchange.id == 'gateio' and now_ms is not None:
            try:
                if since is None:
                    if initial_lookback_days is not None:
                        since = max(0, now_ms - int(initial_lookback_days) * day)
                    else:
                        since = max(0, now_ms - 365 * day)
            except Exception:
                pass

        if exchange.id in {'bitunix', 'weex'} and now_ms is not None and since is None:
            try:
                days = int(initial_lookback_days) if initial_lookback_days is not None else 30
                since = max(0, now_ms - days * day)
            except Exception:
                since = max(0, now_ms - 30 * day)

        _human_log(SERVICE, f"fetch_executions: user={user.name} exchange={exchange.id} since={since}", level='INFO', user=user.name)
        start_ts = time.time()

        # Discover income symbols for Binance and Classic Bitget. Exchange ignores
        # these hints for account-wide UTA fetching and handles normalization.
        symbols = None
        if exchange.id in ('binance', 'bitget') and now_ms is not None:
            # Use caller-provided symbol window if specified; otherwise, cover the fetch window.
            sym_start = 0
            try:
                if symbol_lookback_days is not None:
                    sym_start = max(0, now_ms - int(symbol_lookback_days) * day)
                else:
                    sym_start = int(since) if since is not None else 0
            except Exception:
                sym_start = 0
            symbols = self.list_income_symbols(user.name, int(sym_start), int(now_ms))

        try:
            executions = exchange.fetch_executions(since, symbols=symbols)
        except Exception as e:
            exchange.close()  # Clean up resources before returning
            _human_log(SERVICE,
                f"fetch_executions ERROR: user={user.name} exchange={exchange.id} since={since} err={e}",
                level='WARNING',
                user=user.name,
            )
            return {'fetched': 0, 'prepared': 0, 'inserted': 0, 'error': str(e)}
        finally:
            # Always close exchange to release aiohttp sessions
            exchange.close()
        
        dur = time.time() - start_ts
        try:
            n = len(executions) if executions is not None else 0
        except Exception:
            n = 0
        _human_log(SERVICE, f"fetch_executions DONE: user={user.name} exchange={exchange.id} duration_s={dur:.3f} items={n}", level='INFO', user=user.name)
        if not executions:
            return {'fetched': n, 'prepared': 0, 'inserted': 0}

        rows = []
        for ex in executions:
            try:
                trade_id = ex.get('trade_id')
                if not trade_id:
                    continue
                rows.append(
                    (
                        exchange.id,
                        ex.get('symbol') or '',
                        int(ex.get('timestamp') or 0),
                        ex.get('side'),
                        ex.get('price'),
                        ex.get('qty'),
                        ex.get('fee'),
                        ex.get('realized_pnl'),
                        ex.get('order_id'),
                        str(trade_id),
                        user.name,
                        ex.get('raw_json'),
                    )
                )
            except Exception:
                continue

        if not rows:
            return {'fetched': n, 'prepared': 0, 'inserted': 0}

        sql = (
            'INSERT OR IGNORE INTO executions '
            '(exchange, symbol, timestamp, side, price, qty, fee, realized_pnl, order_id, trade_id, user, raw_json) '
            'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
        )
        inserted = 0
        with self._write_lock:
            with self._connect_trades() as conn:
                before = conn.total_changes
                cur = conn.cursor()
                cur.executemany(sql, rows)
                conn.commit()
                inserted = max(0, conn.total_changes - before)

        try:
            ts_vals = [r[2] for r in rows if r and r[2]]
            ts_min = min(ts_vals) if ts_vals else None
            ts_max = max(ts_vals) if ts_vals else None
        except Exception:
            ts_min, ts_max = None, None
        _human_log(SERVICE,
            f"store_executions DONE: user={user.name} exchange={exchange.id} fetched={n} prepared={len(rows)} inserted={inserted} ts_min={ts_min} ts_max={ts_max}",
            level='INFO',
            user=user.name,
        )

        # Bitget: existing rows may have side semantics that need normalization.
        if exchange.id == 'bitget':
            self._repair_bitget_execution_sides(user.name)

        return {'fetched': n, 'prepared': len(rows), 'inserted': inserted}
    
    def update_positions(self, user: User, _exchange=None):
        positions_db = self.fetch_positions(user)
        _owns_exchange = _exchange is None
        exchange = _exchange or Exchange(user.exchange, user)
        
        try:
            positions = exchange.fetch_positions()
        finally:
            if _owns_exchange:
                exchange.close()  # Release aiohttp resources

        # Live positions as set of (symbol, side) for correct membership checks
        symbols = set()
        for position in positions:
            if position['contracts'] == 0:
                continue
            sym = position['symbol'][0:-5].replace("/", "").replace("-", "")
            side = position['side']
            symbols.add((sym, side))

        # DB positions: ensure at most one row per (symbol, side) by
        # detecting duplicates and marking them for removal. We also
        # build the symbols_db set from the deduplicated view.
        symbols_db = set()
        latest_by_key = {}
        duplicate_ids = []
        for row in positions_db:
            key = (row[1], row[7])  # (symbol, side)
            if key in latest_by_key:
                # Older duplicate; mark for deletion
                duplicate_ids.append(row[0])  # id column
            else:
                latest_by_key[key] = row
                symbols_db.add(key)

        # Build quick lookup of live sides by symbol for debugging
        live_sides_by_symbol = {}
        for sym, side in symbols:
            live_sides_by_symbol.setdefault(sym, set()).add(side)

        with self._write_lock:
            try:
                with self._connect() as conn:
                    # First remove any duplicate DB rows for the same
                    # (symbol, side); keep only the latest row per key.
                    for dup_id in duplicate_ids:
                        _human_log(SERVICE, f"Removing duplicate position id={dup_id} for {user.name}", level='INFO', user=user.name)
                        self.remove_position(conn, dup_id)

                    # Remove positions that are not in the exchange
                    for position in latest_by_key.values():
                        if (position[1], position[7]) not in symbols:
                            _human_log(SERVICE, f"[positions] Removing position for user={user.name} symbol={position[1]} side={position[7]}", level='INFO', user=user.name)
                            self.remove_position(conn, position[0])

                    # Update positions
                    for position in positions:
                        upnl = position['unrealizedPnl']
                        if upnl is None:
                            upnl = 0.0
                        pos = [
                            position['timestamp'],
                            position['contracts'] * position['contractSize'],
                            upnl,
                            position['entryPrice'],
                            position['symbol'][0:-5].replace("/", "").replace("-", ""),
                            user.name,
                            position['side']
                        ]
                        if pos[1] == 0:
                            continue
                        # Use current timestamp if timestamp is None
                        if not pos[0]:
                            pos[0] = int(datetime.now().timestamp() * 1000)
                        key = (pos[4], pos[6])
                        if key in symbols_db:
                            _human_log(SERVICE, f"[positions] Updating position for user={user.name} symbol={pos[4]} side={pos[6]}", level='INFO', user=user.name)
                            self.update_position(conn, pos)
                        else:
                            _human_log(SERVICE, f"[positions] Adding position for user={user.name} symbol={pos[4]} side={pos[6]}", level='INFO', user=user.name)
                            self.add_position(conn, pos)
                            # Ensure we do not create multiple rows for the
                            # same (symbol, side) within one update run.
                            symbols_db.add(key)
            except sqlite3.Error as e:
                _human_log(SERVICE, f"DB update_positions error for {user.name}: {e}", level='ERROR', user=user.name)
    
    def update_orders(self, user: User, _exchange=None):
        positions_db = self.fetch_positions(user)
        orders_db = self.fetch_orders(user)
        _owns_exchange = _exchange is None
        exchange = _exchange or Exchange(user.exchange, user)
        all_orders = []
        fetch_failed = False
        skipped_symbols = set()
        successful_symbols = set()
        try:
            for position in positions_db:
                try:
                    symbol = position[1]
                    ccxt_symbol = _resolve_ccxt_symbol_from_mapping(user.exchange, symbol)
                    if not ccxt_symbol:
                        skipped_symbols.add(symbol)
                        _human_log(SERVICE, f"DB update_orders missing symbol mapping for {user.name} pos={symbol}", level='WARNING', user=user.name)
                        continue
                    orders = exchange.fetch_all_open_orders(ccxt_symbol)
                    # If fetch returns None or an exception-like value, skip
                    if orders is None:
                        fetch_failed = True
                        continue
                    all_orders.extend(orders)
                    successful_symbols.add(symbol)
                except BadSymbol as e:
                    skipped_symbols.add(position[1])
                    _human_log(SERVICE, f"DB update_orders unsupported symbol for {user.name} pos={position[1]}: {e}", level='WARNING', user=user.name)
                except Exception as e:
                    # Fetch failed (possibly rate limit) — log and skip this position
                    _human_log(SERVICE, f"DB update_orders fetch_all_open_orders failed for {user.name} pos={position[1]}: {e}", level='WARNING', user=user.name)
                    fetch_failed = True
                    continue
        finally:
            if _owns_exchange:
                exchange.close()  # Release aiohttp resources
        if fetch_failed:
            _human_log(SERVICE, f"DB update_orders aborting without mutation for {user.name}: incomplete exchange snapshot", level='WARNING', user=user.name)
            return
        if skipped_symbols and not successful_symbols:
            return
        # Existing order IDs in DB (uniqueid column); use a set to avoid
        # inserting duplicates even if the DB uniqueness constraint is
        # missing or was added after the table was first created.
        ids_db = {order[6] for order in orders_db}
        ids = [order['id'] for order in all_orders]
        with self._write_lock:
            try:
                with self._connect() as conn:
                    # Remove orders that are not in the exchange
                    for order in orders_db:
                        if order[6] not in ids and order[7] not in skipped_symbols:
                            _human_log(SERVICE, f"Removing order {order[6]} for user {user.name}", level='INFO', user=user.name)
                            self.remove_order(conn, order[0])
                    # Update orders
                    for order in all_orders:
                        uniqueid = order['id']
                        ord = [
                            order['timestamp'],
                            order['amount'],
                            order['price'],
                            order['side'],
                            uniqueid,
                            order['symbol'].split(":", 1)[0].replace("/", "").replace("-", ""),
                            user.name
                        ]
                        if uniqueid in ids_db:
                            _human_log(SERVICE, f"Updating order {uniqueid} for user {user.name}", level='INFO', user=user.name)
                            self.update_order(conn, ord)
                        else:
                            _human_log(SERVICE, f"Adding order {uniqueid} for user {user.name}", level='INFO', user=user.name)
                            self.add_order(conn, ord)
                            # Ensure we don't insert the same order twice in
                            # a single update run if the exchange returns
                            # duplicates.
                            ids_db.add(uniqueid)
            except sqlite3.Error as e:
                _human_log(SERVICE, f"DB update_orders error for {user.name}: {e}", level='ERROR', user=user.name)

    def update_prices(self, user: User):
        positions_db = self.fetch_positions(user)
        prices_db = self.fetch_prices(user)
        symbols_db = []
        for price in prices_db:
            symbols_db.append(price[1])
        exchange = Exchange(user.exchange, user)
        symbols = []
        prices = {}
        try:
            for position in positions_db:
                symbol = position[1]
                if symbol[-4:] == "USDT":
                    symbol_ccxt = f'{symbol[0:-4]}/USDT:USDT'
                elif symbol[-4:] == "USDC":
                    symbol_ccxt = f'{symbol[0:-4]}/USDC:USDC'
                symbols.append(symbol_ccxt)
            if symbols:
                market_type = "futures"
                prices = exchange.fetch_prices(symbols, market_type)
        finally:
            exchange.close()  # Release aiohttp resources
        symbols = []
        for symbol_ccxt in prices:
            symbol = symbol_ccxt[0:-5].replace("/", "").replace("-", "")
            symbols.append(symbol)
        with self._write_lock:
            try:
                with self._connect() as conn:
                    # Remove symbols that are not in the exchange
                    for symbol in symbols_db:
                        if symbol not in symbols:
                            _human_log(SERVICE, f"Removing price {symbol} for {user.name}", level='INFO', user=user.name)
                            self.remove_price(conn, symbol, user.name)
                    # Update prices
                    for symbol in symbols:
                        if symbol[-4:] == "USDT":
                            symbol_ccxt = f'{symbol[0:-4]}/USDT:USDT'
                        elif symbol[-4:] == "USDC":
                            symbol_ccxt = f'{symbol[0:-4]}/USDC:USDC'
                        timestamp = prices[symbol_ccxt]['timestamp']
                        if not timestamp:
                            timestamp = exchange.fetch_timestamp()
                        price = [
                            timestamp,
                            prices[symbol_ccxt]['last'],
                            symbol,
                            user.name
                        ]
                        if symbol in symbols_db:
                            _human_log(SERVICE, f"Updating price {symbol} for {user.name}", level='INFO', user=user.name)
                            self.update_price(conn, price)
                        else:
                            _human_log(SERVICE, f"Adding price {symbol} for {user.name}", level='INFO', user=user.name)
                            self.add_price(conn, price)
            except sqlite3.Error as e:
                _human_log(SERVICE, f"DB update_prices error for {user.name}: {e}", level='ERROR', user=user.name)

    def update_balances(self, user: User, _exchange=None):
        _owns_exchange = _exchange is None
        exchange = _exchange or Exchange(user.exchange, user)
        market_type = "swap"
        try:
            balance = exchange.fetch_balance(market_type)
        except Exception as e:
            if _owns_exchange:
                exchange.close()
            _human_log(SERVICE, f"DB update_balances fetch_balance failed for {user.name}: {e}", level='WARNING', user=user.name)
            return
        finally:
            if _owns_exchange:
                exchange.close()  # Release aiohttp resources
        with self._write_lock:
            try:
                with self._connect() as conn:
                    balance_list = [
                        int(datetime.now().timestamp() * 1000),
                        balance,
                        user.name
                    ]
                    _human_log(SERVICE, f"Updating balance {user.name}", level='INFO', user=user.name)
                    self.update_balance(conn, balance_list)
            except sqlite3.Error as e:
                _human_log(SERVICE, f"DB update_balances error for {user.name}: {e}", level='ERROR', user=user.name)

    def add_history(self, conn: sqlite3.Connection, history: list):
        sql = '''INSERT INTO history(symbol,timestamp,income,uniqueid,user)
                VALUES(?,?,?,?,?) '''
        try:
            cur = conn.cursor()
            cur.execute(sql, history)
            conn.commit()
        except sqlite3.Error as e:
            # Ignore duplicate uniqueid (already stored), log others
            msg = str(e).lower()
            if 'unique constraint failed' in msg and 'history.uniqueid' in msg:
                return
            _human_log(SERVICE, f"DB add_history error {e} data={history}", level='ERROR')
    
    def add_position(self, conn: sqlite3.Connection, position: list):
        sql = '''INSERT INTO position(timestamp,psize,upnl,entry,symbol,user,side)
                VALUES(?,?,?,?,?,?,?) '''
        try:
            cur = conn.cursor()
            cur.execute(sql, position)
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB add_position error {e} data={position}", level='ERROR')
        return cur.lastrowid

    def add_order(self, conn: sqlite3.Connection, order: list):
        sql = '''INSERT INTO orders(timestamp,amount,price,side,uniqueid,symbol,user)
                VALUES(?,?,?,?,?,?,?) '''
        try:
            cur = conn.cursor()
            cur.execute(sql, order)
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB add_order error {e} data={order}", level='ERROR')

    def add_price(self, conn: sqlite3.Connection, price: list):
        sql = '''INSERT INTO prices(timestamp,price,symbol,user)
                VALUES(?,?,?,?) '''
        try:
            cur = conn.cursor()
            cur.execute(sql, price)
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB add_price error {e} data={price}", level='ERROR')

    def remove_position(self, conn: sqlite3.Connection, id: int):
        sql = '''DELETE FROM position WHERE id = ? '''
        try:
            cur = conn.cursor()
            cur.execute(sql, [id])
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB remove_position error {e} id={id}", level='ERROR')
    
    def remove_order(self, conn: sqlite3.Connection, id: int):
        sql = '''DELETE FROM orders WHERE id = ? '''
        try:
            cur = conn.cursor()
            cur.execute(sql, [id])
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB remove_order error {e} id={id}", level='ERROR')

    def remove_price(self, conn: sqlite3.Connection, symbol: str, user: str):
        sql = '''DELETE FROM prices WHERE symbol = ? AND user = ? '''
        try:
            cur = conn.cursor()
            cur.execute(sql, [symbol, user])
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB remove_price error {e} symbol={symbol} user={user}", level='ERROR', user=user)

    def update_position(self, conn: sqlite3.Connection, position: list):
        sql = '''UPDATE position
                SET timestamp = ?,
                    psize = ?,
                    upnl = ?,
                    entry = ?
                WHERE symbol = ? AND user = ? AND side = ? '''
        try:
            cur = conn.cursor()
            cur.execute(sql, position)
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB update_position error {e} data={position}", level='ERROR')

    def update_order(self, conn: sqlite3.Connection, order: list):
        sql = '''UPDATE orders
                SET timestamp = ?,
                    amount = ?,
                    price = ?,
                    side = ?
                WHERE uniqueid = ? AND symbol = ? AND user = ? '''
        try:
            cur = conn.cursor()
            cur.execute(sql, order)
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB update_order error {e} data={order}", level='ERROR')

    def update_price(self, conn: sqlite3.Connection, price: list):
        sql = '''UPDATE prices
                SET timestamp = ?,
                    price = ?
                WHERE symbol = ? AND user = ? '''
        try:
            cur = conn.cursor()
            cur.execute(sql, price)
            conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB update_price error {e} data={price}", level='ERROR')

    def upsert_price(self, user: User, symbol: str, timestamp: int, price_value: float):
        """Insert or update a single price tick for (user, symbol)."""
        start_ts = datetime.now().timestamp()
        with self._write_lock:
            attempts = 0
            while True:
                try:
                    with self._connect() as conn:
                        cur = conn.cursor()
                        cur.execute('SELECT id FROM prices WHERE symbol = ? AND user = ?', (symbol, user.name))
                        exists = cur.fetchone() is not None
                        price_row = [timestamp, price_value, symbol, user.name]
                        if exists:
                            self.update_price(conn, price_row)
                        else:
                            self.add_price(conn, price_row)
                    break
                except sqlite3.OperationalError as e:
                    if 'database is locked' in str(e).lower() and attempts < 5:
                        attempts += 1
                        time.sleep(0.05 * attempts)
                        continue
                    _human_log(SERVICE, f"DB upsert_price error {e} user={user.name} symbol={symbol} price={price_value} attempts={attempts}", level='ERROR', user=user.name)
                    break
                except sqlite3.Error as e:
                    _human_log(SERVICE, f"DB upsert_price error {e} user={user.name} symbol={symbol} price={price_value} attempts={attempts}", level='ERROR', user=user.name)
                    break

    def batch_upsert_prices(self, rows: list):
        """Batch upsert a list of price rows.

        rows: iterable of (user, symbol, timestamp, price)
        This uses a single connection and transaction to reduce locking and
        improve throughput compared to many per-row connections.
        """
        if not rows:
            return
        with self._write_lock:
            attempts = 0
            while True:
                try:
                    with self._connect() as conn:
                        cur = conn.cursor()
                        for user, symbol, timestamp, price in rows:
                            cur.execute('UPDATE prices SET timestamp = ?, price = ? WHERE symbol = ? AND user = ?', (timestamp, price, symbol, user))
                            if cur.rowcount == 0:
                                cur.execute('INSERT INTO prices(timestamp,price,symbol,user) VALUES(?,?,?,?)', (timestamp, price, symbol, user))
                        conn.commit()
                    break
                except sqlite3.OperationalError as e:
                    if 'database is locked' in str(e).lower() and attempts < 5:
                        attempts += 1
                        time.sleep(0.05 * attempts)
                        continue
                    _human_log(SERVICE, f"DB batch_upsert_prices error {e} attempts={attempts}", level='ERROR')
                    raise
                except sqlite3.Error as e:
                    _human_log(SERVICE, f"DB batch_upsert_prices error {e} attempts={attempts}", level='ERROR')
                    raise

    def update_balance(self, conn: sqlite3.Connection, balance: list):
        try:
            if not balance or len(balance) < 3:
                _human_log(SERVICE, f"DB update_balance called with invalid balance data: {balance}", level='WARNING')
                return
            # Defensive checks: avoid writing Exception objects or other
            # unserializable types into the DB which cause SQLite bind errors.
            # If the balance payload contains an Exception, skip and warn.
            try:
                payload = balance[1]
            except Exception:
                payload = None
            if isinstance(payload, Exception):
                _human_log(SERVICE, f"DB update_balance aborting: balance fetch returned exception object for user={balance[2]}: {payload}", level='WARNING', user=balance[2])
                return
            # Try to coerce numeric-like payloads to float; otherwise stringify
            try:
                if not isinstance(payload, (int, float)):
                    balance[1] = float(payload)
            except Exception:
                try:
                    balance[1] = str(payload)
                    _human_log(SERVICE, f"DB update_balance coerced non-numeric balance to string for user={balance[2]}", level='WARNING', user=balance[2])
                except Exception:
                    _human_log(SERVICE, f"DB update_balance aborting: cannot coerce balance for user={balance[2]}: {payload}", level='WARNING', user=balance[2])
                    return
            cur = conn.cursor()
            # Preserve the rowid on every refresh. REPLACE deletes the old row
            # and inserts another one, causing unnecessary B-tree churn.
            cur.execute(
                'UPDATE balances SET timestamp = ?, balance = ? WHERE user = ?',
                (balance[0], balance[1], balance[2]),
            )
            if cur.rowcount == 0:
                cur.execute(
                    'INSERT INTO balances(timestamp, balance, user) VALUES (?, ?, ?)',
                    balance,
                )
            elif cur.rowcount > 1:
                # Legacy tables may lack UNIQUE(user). Keep one updated row.
                user = str(balance[2])
                cur.execute(
                    'DELETE FROM balances WHERE user = ? AND id != '
                    '(SELECT MAX(id) FROM balances WHERE user = ?)',
                    (user, user),
                )
                _human_log(SERVICE, f"DB update_balance removed {cur.rowcount} older balances for user={user}", level='INFO', user=user)

            conn.commit()
        except sqlite3.Error as e:
            msg = str(e).lower()
            if 'binding parameter' in msg or 'unsupported type' in msg:
                _human_log(SERVICE, f"DB update_balance error {e} data={balance}", level='WARNING')
            else:
                _human_log(SERVICE, f"DB update_balance error {e} data={balance}", level='ERROR')

    # History scan lookback: when we already scanned up to `now` previously,
    # the next cycle only needs to go back this far instead of starting from
    # the oldest DB entry (which can be months ago for inactive bots).
    _HISTORY_SCAN_LOOKBACK_MS = 6 * 60 * 60 * 1000  # 6 hours

    def get_last_scan_ts(self, user_name: str, exchange: str):
        """Return epoch ms up to which we last scanned history, or None."""
        sql = 'SELECT last_scan_ts FROM history_scan_meta WHERE user = ? AND exchange = ?'
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user_name, exchange))
                row = cur.fetchone()
                return int(row[0]) if row else None
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB get_last_scan_ts error {e} user={user_name}", level='ERROR', user=user_name)
            return None

    def set_last_scan_ts(self, user_name: str, exchange: str, ts_ms: int):
        """Upsert the last scan timestamp for a user/exchange pair."""
        sql = '''INSERT INTO history_scan_meta (user, exchange, last_scan_ts)
                 VALUES (?, ?, ?)
                 ON CONFLICT(user, exchange) DO UPDATE SET last_scan_ts = excluded.last_scan_ts'''
        try:
            with self._connect() as conn:
                conn.execute(sql, (user_name, exchange, int(ts_ms)))
                conn.commit()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB set_last_scan_ts error {e} user={user_name}", level='ERROR', user=user_name)

    def fetch_history(self, user: User):
        """Fetch history without advancing the persisted scan checkpoint.

        Uses history_scan_meta to avoid re-scanning months of empty history
        for inactive bots.  On first run (no meta entry) falls back to the
        last DB entry so the initial import is complete.
        update_history commits the checkpoint only together with all fetched rows.
        """
        exchange = Exchange(user.exchange, user)
        try:
            last_scan = self.get_last_scan_ts(user.name, user.exchange)
            if last_scan is not None:
                # We already scanned up to last_scan previously — only look back 6h
                since = max(0, last_scan - self._HISTORY_SCAN_LOOKBACK_MS)
            else:
                # First scan for this user/exchange — start from last DB entry
                try:
                    since = self.find_last_timestamp(user) or 0
                except Exception:
                    since = 0
        except Exception:
            try:
                since = self.find_last_timestamp(user) or 0
            except Exception:
                since = 0
        _human_log(SERVICE, f"fetch_history: user={user.name} exchange={user.exchange} since={since} last_scan={'cached' if last_scan is not None else 'initial'}", level='INFO', user=user.name)
        # Time the exchange.fetch_history call to help diagnose slow history polls.
        try:
            start_ts = time.time()
            history = exchange.fetch_history(int(since))
            dur = time.time() - start_ts
            try:
                length = len(history) if history is not None else 0
            except Exception:
                length = 0
            _human_log(SERVICE, f"fetch_history DONE: user={user.name} exchange={user.exchange} duration_s={dur:.3f} items={length}", level='INFO', user=user.name)
            return history
        except Exception as e:
            # Do not swallow exceptions here — re-raise so callers (and tests)
            # can inspect the full traceback and we can debug exchange.fetch_history.
            # On error: do NOT update last_scan_ts so next cycle retries from same point.
            _human_log(SERVICE, f"fetch_history ERROR for user={user.name} exchange={user.exchange}: {e}", level='ERROR', user=user.name)
            raise
        finally:
            exchange.close()  # Release aiohttp resources

    def fetch_positions(self, user: User):
        sql = '''SELECT * FROM "position"
                WHERE "position"."user" = ? '''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, [user.name])
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB fetch_positions error {e} user={user.name}", level='ERROR', user=user.name)

    def fetch_orders(self, user: User):
        sql = '''SELECT * FROM "orders"
                WHERE "orders"."user" = ? '''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, [user.name])
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB fetch_orders error {e} user={user.name}", level='ERROR', user=user.name)
    
    def fetch_orders_by_symbol(self, user: str, symbol: str):
        sql = '''SELECT * FROM "orders"
                WHERE "orders"."user" = ?
                    AND "orders"."symbol" = ? '''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, [user, symbol])
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB fetch_orders_by_symbol error {e} user={user} symbol={symbol}", level='ERROR', user=user)

    def fetch_prices(self, user: User):
        sql = '''SELECT * FROM "prices"
                WHERE "prices"."user" = ? '''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, [user.name])
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB fetch_prices error {e} user={user.name}", level='ERROR', user=user.name)

    def fetch_balances(self, user: list):
        """Return matching rows, or an empty result when no users are selected."""
        if not user:
            return []
        sql = '''SELECT * FROM "balances"
                WHERE "balances"."user" IN ({}) '''.format(','.join('?'*len(user)))
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, user)
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB fetch_balances error {e} users={user}", level='ERROR')

    def select_top(self, user: list, start: str, end: str, top: int):
        """Return matching rows, or an empty result when no users are selected."""
        if not user:
            return []
        if 'ALL' in user:
            sql = '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date, "history"."symbol" AS symbol, SUM("history"."income") AS sum FROM "history"
                    WHERE "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    GROUP BY "history"."symbol"
                    ORDER BY "sum" DESC, "history"."symbol"
                    LIMIT ? '''
            sql_parameters = (start, end, top)
        else:
            sql = '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date, "history"."symbol" AS symbol, SUM("history"."income") AS sum FROM "history"
                    WHERE "history"."user" IN ({})
                        AND "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    GROUP BY "history"."symbol"
                    ORDER BY "sum" DESC, "history"."symbol"
                    LIMIT ? '''.format(','.join('?'*len(user)))
            sql_parameters = tuple(user) + (start, end, top)
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, sql_parameters)
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_top error {e}", level='ERROR')
        
    def select_pnl(self, user: list, start: str, end: str):
        if 'ALL' in user:
            sql = '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date, SUM("income") AS "sum" FROM "history"
                    WHERE "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    GROUP BY date '''
            sql_parameters = (start, end)
        else:
            sql = '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date, SUM("income") AS "sum" FROM "history"
                    WHERE "history"."user" IN ({})
                        AND "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    GROUP BY date'''.format(','.join('?'*len(user)))
            sql_parameters = tuple(user) + (start, end)
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, sql_parameters)
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_pnl error {e}", level='ERROR')

    def sum_income(self, user: str, start: int, end: int) -> float:
        sql = '''SELECT COALESCE(SUM("income"), 0) FROM "history"
                WHERE "history"."user" = ?
                    AND "history"."timestamp" >= ?
                    AND "history"."timestamp" <= ?'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user, start, end))
                row = cur.fetchone()
                return float(row[0] if row and row[0] is not None else 0.0)
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB sum_income error {e} user={user}", level='ERROR', user=user)
            return 0.0

    def min_income_timestamp(self, user: str):
        sql = '''SELECT MIN("timestamp") FROM "history" WHERE "history"."user" = ?'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user,))
                row = cur.fetchone()
                if not row or row[0] is None:
                    return None
                return int(row[0])
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB min_income_timestamp error {e} user={user}", level='ERROR', user=user)
            return None

    def max_income_timestamp(self, user: str):
        sql = '''SELECT MAX("timestamp") FROM "history" WHERE "history"."user" = ?'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user,))
                row = cur.fetchone()
                if not row or row[0] is None:
                    return None
                return int(row[0])
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB max_income_timestamp error {e} user={user}", level='ERROR', user=user)
            return None

    def list_income_users(self) -> list:
        """Return distinct users present in the income/history table."""
        sql = '''SELECT DISTINCT "user" FROM "history" ORDER BY "user"'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql)
                rows = cur.fetchall()
                return [r[0] for r in rows if r and r[0] is not None]
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB list_income_users error {e}", level='ERROR')
            return []

    def list_income_symbols(self, user: str, start: int, end: int) -> list:
        sql = '''SELECT DISTINCT "symbol" FROM "history"
                WHERE "history"."user" = ?
                    AND "history"."timestamp" >= ?
                    AND "history"."timestamp" <= ?
                ORDER BY "symbol"'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user, start, end))
                rows = cur.fetchall()
                return [r[0] for r in rows if r and r[0] is not None]
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB list_income_symbols error {e} user={user}", level='ERROR', user=user)
            return []

    def select_pnl_symbol(self, user: str, symbol: str, start: int, end: int):
        sql = '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date, SUM("income") AS "sum" FROM "history"
                WHERE "history"."user" = ?
                    AND "history"."symbol" = ?
                    AND "history"."timestamp" >= ?
                    AND "history"."timestamp" <= ?
                GROUP BY date'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user, symbol, start, end))
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_pnl_symbol error {e} user={user} symbol={symbol}", level='ERROR', user=user)
            return []

    def select_history_daily(self, user: str, start: int, end: int, symbol: str | None = None):
        """Return daily sums and counts from history.

        Returns rows: (date_str, sum_income, count_rows)
        """
        if symbol:
            sql = '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date,
                            COALESCE(SUM("income"), 0) AS sum,
                            COUNT(*) AS n
                     FROM "history"
                     WHERE "user" = ? AND "symbol" = ?
                       AND "timestamp" >= ? AND "timestamp" <= ?
                     GROUP BY date
                     ORDER BY date'''
            params = (user, symbol, start, end)
        else:
            sql = '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date,
                            COALESCE(SUM("income"), 0) AS sum,
                            COUNT(*) AS n
                     FROM "history"
                     WHERE "user" = ?
                       AND "timestamp" >= ? AND "timestamp" <= ?
                     GROUP BY date
                     ORDER BY date'''
            params = (user, start, end)
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, params)
                return cur.fetchall()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_history_daily error {e} user={user}", level='ERROR', user=user)
            return []

    def select_history_rows(
        self,
        user: str,
        start: int,
        end: int,
        symbol: str | None = None,
        limit: int = 2000,
        newest_first: bool = False,
    ):
        """Return raw history rows for debugging.

        Returns rows: (timestamp, symbol, income, uniqueid)
        """
        order = 'DESC' if newest_first else 'ASC'
        if symbol:
            sql = f'''SELECT "timestamp", "symbol", "income", "uniqueid"
                      FROM "history"
                      WHERE "user" = ? AND "symbol" = ?
                        AND "timestamp" >= ? AND "timestamp" <= ?
                      ORDER BY "timestamp" {order}
                      LIMIT ?'''
            params = (user, symbol, start, end, int(limit))
        else:
            sql = f'''SELECT "timestamp", "symbol", "income", "uniqueid"
                      FROM "history"
                      WHERE "user" = ?
                        AND "timestamp" >= ? AND "timestamp" <= ?
                      ORDER BY "timestamp" {order}
                      LIMIT ?'''
            params = (user, start, end, int(limit))
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, params)
                return cur.fetchall()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_history_rows error {e} user={user}", level='ERROR', user=user)
            return []

    def select_executions_daily(
        self,
        user: str,
        exchange: str,
        start: int,
        end: int,
        coin: str | None = None,
    ):
        """Return daily aggregates from executions.

        Returns rows: (date_str, n_trades, sum_fee, sum_realized_pnl)
        """
        where_coin = ''
        params: tuple = (user, exchange, start, end)
        if coin:
            c = str(coin).strip().upper()
            if c:
                # Symbols are typically CCXT-style (e.g. 'ETH/USDC:USDC') or compact (e.g. 'ETHUSDC')
                where_coin = ' AND ("symbol" LIKE ? OR "symbol" LIKE ?) '
                params = (user, exchange, start, end, f"{c}/%", f"{c}%")

        sql = (
            '''SELECT strftime('%Y-%m-%d',"timestamp" / 1000, 'unixepoch') as date,
                      COUNT(*) AS n,
                      COALESCE(SUM("fee"), 0) AS fee_sum,
                      COALESCE(SUM("realized_pnl"), 0) AS rpnl_sum
               FROM "executions"
               WHERE "user" = ? AND "exchange" = ?
                 AND "timestamp" >= ? AND "timestamp" <= ?'''
            + where_coin +
            ''' GROUP BY date
               ORDER BY date'''
        )
        try:
            with self._connect_trades() as conn:
                cur = conn.cursor()
                cur.execute(sql, params)
                return cur.fetchall()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_executions_daily error {e} user={user}", level='ERROR', user=user)
            return []

    def select_executions_rows(
        self,
        user: str,
        exchange: str,
        start: int,
        end: int,
        coin: str | None = None,
        limit: int = 2000,
        newest_first: bool = False,
    ):
        """Return raw executions rows for debugging.

        Returns rows: (timestamp, symbol, side, price, qty, fee, realized_pnl, order_id, trade_id)
        """
        order = 'DESC' if newest_first else 'ASC'
        where_coin = ''
        params: tuple = (user, exchange, start, end)
        if coin:
            c = str(coin).strip().upper()
            if c:
                where_coin = ' AND ("symbol" LIKE ? OR "symbol" LIKE ?) '
                params = (user, exchange, start, end, f"{c}/%", f"{c}%")

        sql = (
            f'''SELECT "timestamp", "symbol", "side", "price", "qty", "fee", "realized_pnl", "order_id", "trade_id"
                FROM "executions"
                WHERE "user" = ? AND "exchange" = ?
                  AND "timestamp" >= ? AND "timestamp" <= ?'''
            + where_coin +
            f''' ORDER BY "timestamp" {order}
                LIMIT ?'''
        )
        try:
            with self._connect_trades() as conn:
                cur = conn.cursor()
                cur.execute(sql, params + (int(limit),))
                return cur.fetchall()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_executions_rows error {e} user={user}", level='ERROR', user=user)
            return []

    def select_executions_rows_any_exchange(
        self,
        user: str,
        start: int,
        end: int,
        coin: str | None = None,
        limit: int = 2000,
        newest_first: bool = False,
    ):
        """Return raw executions rows for debugging, across all exchanges.

        Returns rows: (timestamp, exchange, symbol, side, price, qty, fee, realized_pnl, order_id, trade_id)
        """

        order = 'DESC' if newest_first else 'ASC'
        where_coin = ''
        params: tuple = (user, start, end)
        if coin:
            c = str(coin).strip().upper()
            if c:
                where_coin = ' AND ("symbol" LIKE ? OR "symbol" LIKE ?) '
                params = (user, start, end, f"{c}/%", f"{c}%")

        sql = (
            f'''SELECT "timestamp", "exchange", "symbol", "side", "price", "qty", "fee", "realized_pnl", "order_id", "trade_id"
                FROM "executions"
                WHERE "user" = ?
                  AND "timestamp" >= ? AND "timestamp" <= ?'''
            + where_coin +
            f''' ORDER BY "timestamp" {order}
                LIMIT ?'''
        )
        try:
            with self._connect_trades() as conn:
                cur = conn.cursor()
                cur.execute(sql, params + (int(limit),))
                return cur.fetchall()
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_executions_rows_any_exchange error {e} user={user}", level='ERROR', user=user)
            return []

    
    def select_ppl(self, user: list, start: str, end: str, sum_period: str):
    # Define date formats for different sum_period values
        date_formats = {
            'DAY': "'%Y-%m-%d'",
            'WEEK': "'%Y-%W'",
            'MONTH': "'%Y-%m'",
            'YEAR': "'%Y'",
        }

        if sum_period == 'ALL_TIME':
            select_period = "'ALL_TIME' AS period"
            group_by_clause = ''
        else:
            date_format = date_formats.get(sum_period, "'%Y-%m-%d'")
            select_period = f"strftime({date_format}, \"timestamp\" / 1000, 'unixepoch') AS period"
            group_by_clause = 'GROUP BY period'

        if 'ALL' in user:
            sql = f'''
            SELECT
                {select_period},
                SUM(CASE WHEN "income" >= 0 THEN "income" ELSE 0 END) AS "sum_positive",
                SUM(CASE WHEN "income" < 0 THEN "income" ELSE 0 END) AS "sum_negative"
            FROM "history"
            WHERE "history"."timestamp" >= ?
                AND "history"."timestamp" <= ?
            {group_by_clause}
            '''
            sql_parameters = (start, end)
        else:
            placeholders = ','.join('?' * len(user))
            sql = f'''
            SELECT
                {select_period},
                SUM(CASE WHEN "income" >= 0 THEN "income" ELSE 0 END) AS "sum_positive",
                SUM(CASE WHEN "income" < 0 THEN "income" ELSE 0 END) AS "sum_negative"
            FROM "history"
            WHERE "history"."user" IN ({placeholders})
                AND "history"."timestamp" >= ?
                AND "history"."timestamp" <= ?
            {group_by_clause}
            '''
            sql_parameters = tuple(user) + (start, end)
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, sql_parameters)
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_ppl error {e}", level='ERROR')

    def select_income(self, user: list, start: str, end: str):
        if 'ALL' in user:
            sql = '''SELECT "timestamp", "income" FROM "history"
                    WHERE "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    ORDER BY "timestamp" ASC '''
            sql_parameters = (start, end)
        else:
            sql = '''SELECT "timestamp", "income" FROM "history"
                    WHERE "history"."user" IN ({})
                        AND "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    ORDER BY "timestamp" ASC'''.format(','.join('?'*len(user)))
            sql_parameters = tuple(user) + (start, end)
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, sql_parameters)
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_income error {e}", level='ERROR')
    
    # select income grouped by symbol not sum
    def select_income_by_symbol(self, user: list, start: str, end: str):
        if 'ALL' in user:
            sql = '''SELECT "timestamp", "symbol", "income", "user" FROM "history"
                    WHERE "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    ORDER BY "timestamp" ASC '''
            sql_parameters = (start, end)
        else:
            sql = '''SELECT "timestamp", "symbol", "income", "user" FROM "history"
                    WHERE "history"."user" IN ({})
                        AND "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    ORDER BY "timestamp" ASC'''.format(','.join('?'*len(user)))
            sql_parameters = tuple(user) + (start, end)
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, sql_parameters)
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_income_by_symbol error {e}", level='ERROR')

    # Same as select_income_by_symbol but includes row id for deletion mapping
    def select_income_by_symbol_with_id(self, user: list, start: str, end: str):
        if 'ALL' in user:
            sql = '''SELECT "id", "timestamp", "symbol", "income", "user" FROM "history"
                    WHERE "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    ORDER BY "timestamp" ASC '''
            sql_parameters = (start, end)
        else:
            sql = '''SELECT "id", "timestamp", "symbol", "income", "user" FROM "history"
                    WHERE "history"."user" IN ({})
                        AND "history"."timestamp" >= ?
                        AND "history"."timestamp" <= ?
                    ORDER BY "timestamp" ASC'''.format(','.join('?'*len(user)))
            sql_parameters = tuple(user) + (start, end)
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, sql_parameters)
                rows = cur.fetchall()
                return rows
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB select_income_by_symbol_with_id error {e}", level='ERROR')

    # Delete specific income rows by primary key ids
    def delete_income_by_ids(self, ids: list):
        if not ids:
            return 0
        placeholders = ','.join('?' * len(ids))
        sql = f'''DELETE FROM history WHERE id IN ({placeholders})'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, ids)
                conn.commit()
                return cur.rowcount
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB delete_income_by_ids error {e}", level='ERROR')
            return 0

    # Delete all income rows for a user older than or equal to a timestamp (ms)
    def delete_income_older_than_user(self, user: str, timestamp_ms: int):
        sql = '''DELETE FROM history WHERE user = ? AND timestamp <= ?'''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, (user, int(timestamp_ms)))
                conn.commit()
                return cur.rowcount
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB delete_income_older_than_user error {e}", level='ERROR')
            return 0

    # Delete all income rows older than or equal to a timestamp for given users list.
    # If users contains 'ALL', deletes across all users.
    def delete_income_older_than(self, users: list, timestamp_ms: int):
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                if not users or 'ALL' in users:
                    sql = 'DELETE FROM history WHERE timestamp <= ?'
                    cur.execute(sql, (int(timestamp_ms),))
                else:
                    placeholders = ','.join('?' * len(users))
                    sql = f'DELETE FROM history WHERE user IN ({placeholders}) AND timestamp <= ?'
                    cur.execute(sql, tuple(users) + (int(timestamp_ms),))
                conn.commit()
                return cur.rowcount
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB delete_income_older_than error {e}", level='ERROR')
            return 0

    def find_last_timestamp(self, user: User):
        sql = '''SELECT MAX("history"."timestamp") FROM "history"
                WHERE "history"."user" = ? '''
        try:
            with self._connect() as conn:
                cur = conn.cursor()
                cur.execute(sql, [user.name])
                rows = cur.fetchall()
                if rows[0][0] is None:
                    return 0
                return rows[0][0]
        except sqlite3.Error as e:
            _human_log(SERVICE, f"DB find_last_timestamp error {e} user={user.name}", level='ERROR', user=user.name)

    def fetch_history2(self, user: User):
        exchange = Exchange(user.exchange, user)
        try:
            return exchange.fetch_transactions(1724390528161)
        finally:
            exchange.close()

    def fetch_futures(self, user: User):
        exchange = Exchange(user.exchange, user)
        try:
            return exchange.fetch_futures(1724390528161)
        finally:
            exchange.close()
    
    def import_from_save_income_other(self, user: User):
        """Import legacy income records in one transaction under the write lease."""
        name = user.name
        if not isinstance(name, str) or not name or name in {".", ".."} or any(
            char in name for char in ("/", "\\")
        ) or any(ord(char) < 32 or ord(char) == 127 for char in name):
            raise ValueError("Invalid user name")
        root = (Path(PBGDIR) / "data" / "logs").resolve()
        source = root / f"income_other_{name}.json"
        if source.is_symlink() or source.resolve().parent != root:
            raise ValueError("Invalid income import path")
        if not source.exists():
            _human_log(SERVICE, "Income import file not found", level='WARNING', user=name)
            return
        try:
            data = '[' + source.read_text().replace('}{', '},{') + ']'
            rows = [(item['symbol'], item['time'], item['income'], item['tranId'], name)
                    for item in json.loads(data)
                    if item.get('incomeType') in {'COMMISSION', 'FUNDING_FEE'}]
        except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
            _human_log(SERVICE, f"Invalid income import: {type(exc).__name__}", level='ERROR', user=name)
            return
        if not rows:
            return
        try:
            with self._write_lock:
                with self._connect() as conn:
                    conn.executemany(
                        "INSERT INTO history(symbol,timestamp,income,uniqueid,user) "
                        "VALUES(?,?,?,?,?) ON CONFLICT(uniqueid) DO NOTHING", rows)
        except sqlite3.Error as exc:
            _human_log(SERVICE, f"DB import_from_save_income_other error {exc}", level='ERROR', user=name)


def main():
    try:
        _human_log(SERVICE, "Don't Run this Class from CLI", level='INFO')
    except Exception:
        # Silently ignore logging failures in CLI helper
        pass
    # users = Users()
    # user = users.find_user("hl_mani_crash_hunter")
    # db = Database()
    # db.fetch_history(user)
    # db.import_from_save_income_other(user)
    # exchange = Exchange("gateio", user)
    # history = exchange.fetch_history()
    # print(history)
    # balance = exchange.fetch_balance("swap")
    # print(balance)
    # positions = exchange.fetch_positions()
    # print(positions)
    # db = Database()
    # db.update_history(user)
    # db.update_positions(user)
    # db.update_orders(user)
    # db.update_prices(user)
    # db.update_balances(user)
    # exchange.connect()
    # print(db.find_last_timestamp(user))
    # history = db.fetch_history2(user)
    # print(history)
    # history = db.fetch_history2(user2)
    # print(history)
    # balances = db.fetch_balances(['gateio_cpt'])
    # print(balances)
    # db.update_positions(user)
    # db.update_prices(user)
    # db.update_balances(user)

if __name__ == '__main__':
    main()