PBData.py

import ast
import psutil
import subprocess
import sys
import os
from contextlib import asynccontextmanager
from pathlib import Path, PurePath
from time import sleep
from datetime import datetime
import platform
import traceback
from pbgui_func import PBGDIR
import json
import re
from pathlib import Path as _Path
import tempfile
from Database import Database
from User import Users
import configparser
from collections import defaultdict
import asyncio
import random
from logging_helpers import human_log as _human_log, set_service_min_level, is_debug_enabled
from Exchange import set_ws_limits, Exchange as _Exchange
from market_data import load_market_data_config, get_daily_hour_coverage_for_dataset, set_enabled_coins
from rate_limit_budget import RateLimitBudget, EXCHANGE_RATE_LIMITS, get_weight
from hyperliquid_best_1m import update_latest_hyperliquid_1m_api_for_coin
from binance_best_1m import update_latest_binance_1m_for_coin
from inventory_cache import refresh_coin as _refresh_inventory_coin


async def _wait_for_flag(flag_path: _Path, timeout: float) -> bool:
    """Wait until flag_path is created (or timeout expires).

    Uses Linux inotify so the coroutine sleeps with zero CPU usage and wakes
    up immediately when the file appears.  Falls back to 5-second polling if
    inotify is unavailable (non-Linux or permission error).

    Returns True if the flag was detected, False on timeout.
    """
    # If flag already exists (e.g. clicked during a run), consume and return immediately.
    if flag_path.exists():
        try:
            flag_path.unlink(missing_ok=True)
        except Exception:
            pass
        return True

    import struct as _struct
    import ctypes as _ct

    try:
        _libc = _ct.CDLL(None, use_errno=True)
        # inotify_init1(IN_NONBLOCK)
        _ifd = _libc.inotify_init1(0o4000)
        if _ifd < 0:
            raise OSError("inotify_init1 failed")
        # Watch the parent directory for file creation / atomic rename
        _wd = _libc.inotify_add_watch(
            _ifd,
            str(flag_path.parent).encode(),
            0x100 | 0x80 | 0x08,  # IN_CREATE | IN_MOVED_TO | IN_CLOSE_WRITE
        )
        if _wd < 0:
            os.close(_ifd)
            raise OSError("inotify_add_watch failed")

        # Check again after setting up the watch (race-free)
        if flag_path.exists():
            try:
                _libc.inotify_rm_watch(_ifd, _wd)
                os.close(_ifd)
            except Exception:
                pass
            try:
                flag_path.unlink(missing_ok=True)
            except Exception:
                pass
            return True

        loop = asyncio.get_running_loop()
        fut: asyncio.Future = loop.create_future()
        _fname = flag_path.name.encode()

        def _readable():
            try:
                data = os.read(_ifd, 4096)
                offset = 0
                while offset + 16 <= len(data):
                    _wd2, _mask, _cookie, nlen = _struct.unpack_from('iIII', data, offset)
                    name = data[offset + 16: offset + 16 + nlen].rstrip(b'\x00')
                    offset += 16 + nlen
                    if name == _fname and not fut.done():
                        fut.set_result(True)
                        return
            except Exception:
                pass

        loop.add_reader(_ifd, _readable)
        try:
            return await asyncio.wait_for(asyncio.shield(fut), timeout=timeout)
        except asyncio.TimeoutError:
            return False
        finally:
            loop.remove_reader(_ifd)
            try:
                _libc.inotify_rm_watch(_ifd, _wd)
                os.close(_ifd)
            except Exception:
                pass
    except Exception:
        # Fallback: 5-second polling (non-Linux or inotify unavailable)
        elapsed = 0.0
        while elapsed < timeout:
            if flag_path.exists():
                try:
                    flag_path.unlink(missing_ok=True)
                except Exception:
                    pass
                return True
            await asyncio.sleep(5.0)
            elapsed += 5.0
        return False


async def _notify_api_balance():
    """Fire-and-forget: POST to FastAPI to fan-out balance_updated to all /ws/dashboard clients.

    Called via asyncio.create_task() after each successful update_balances() write.
    Errors are silently swallowed — a missed notification is not critical.
    """
    import urllib.request
    from pbgui_purefunc import load_ini
    try:
        port_val = load_ini("api_server", "port")
        port = int(port_val) if port_val and str(port_val).isdigit() else 8000
        req = urllib.request.Request(
            f"http://127.0.0.1:{port}/api/internal/notify/balance",
            data=b"",
            method="POST",
        )
        await asyncio.to_thread(urllib.request.urlopen, req, None, 2)
    except Exception as e:
        try:
            _human_log('PBData', f"[notify] balance notification to API failed: {e}", level='DEBUG')
        except Exception:
            pass


async def _notify_api_positions(user_name: str = ""):
    """Fire-and-forget: POST to FastAPI so chart subscribers get refreshed entry lines.

    Called via asyncio.create_task() after each successful update_positions() write.
    The API server reads the updated DB row and pushes it to all chart WS clients
    watching that user — this ensures the Orders widget clears stale Entry lines even
    when the WS missed the original 'position closed' event.
    Errors are silently swallowed — a missed notification is not critical.
    """
    import urllib.request
    from pbgui_purefunc import load_ini
    try:
        port_val = load_ini("api_server", "port")
        port = int(port_val) if port_val and str(port_val).isdigit() else 8000
        body = json.dumps({"user": user_name}).encode()
        req = urllib.request.Request(
            f"http://127.0.0.1:{port}/api/internal/notify/positions",
            data=body,
            method="POST",
            headers={"Content-Type": "application/json"},
        )
        await asyncio.to_thread(urllib.request.urlopen, req, None, 2)
    except Exception as e:
        try:
            _human_log('PBData', f"[notify] positions notification to API failed: {e}", level='DEBUG')
        except Exception:
            pass


async def _notify_api_income(user_name: str = ""):
    """Fire-and-forget: POST to FastAPI to fan-out income_updated to all /ws/dashboard clients.

    Called via asyncio.create_task() after each successful update_history() write.
    user_name is forwarded so clients can filter and only reload for their configured user(s).
    Errors are silently swallowed — a missed notification is not critical.
    """
    import urllib.request
    from pbgui_purefunc import load_ini
    try:
        port_val = load_ini("api_server", "port")
        port = int(port_val) if port_val and str(port_val).isdigit() else 8000
        body = json.dumps({"user": user_name}).encode()
        req = urllib.request.Request(
            f"http://127.0.0.1:{port}/api/internal/notify/income",
            data=body,
            method="POST",
            headers={"Content-Type": "application/json"},
        )
        await asyncio.to_thread(urllib.request.urlopen, req, None, 2)
    except Exception as e:
        try:
            _human_log('PBData', f"[notify] income notification to API failed: {e}", level='DEBUG')
        except Exception:
            pass


class PBData():
    def __init__(self):
        self.piddir = Path(f'{PBGDIR}/data/pid')
        if not self.piddir.exists():
            self.piddir.mkdir(parents=True)
        self.pidfile = Path(f'{self.piddir}/pbdata.pid')
        self.my_pid = None
        self.db = Database()
        self.users = Users()
        self._fetch_users = []
        self._trades_users = []
        self.load_fetch_users()
        self.load_trades_users()
        self._price_exchange_tasks = {}
        self._price_exchange_config = {}
        # Track which symbols we have already subscribed to per exchange
        self._price_subscribed_symbols = {}

        # In-memory buffer for latest prices per (user, symbol).
        # Key: (user_name, symbol) -> (timestamp, price)
        self._price_buffer = {}
        # Async lock protecting _price_buffer
        self._price_buffer_lock = asyncio.Lock()
        # Flush interval in seconds for buffered price writes
        self._price_flush_interval = 10.0
        # Background writer task handle
        self._price_writer_task = None

        # Enable/disable price buffering via env var `PB_PRICE_BUFFER` (default: enabled)
        self._price_buffer_enabled = True

        # (IO tracking removed) -- process/db IO debugging variables removed

        self._history_rest_last = {}
        # Log level used when a REST slot is busy and an update is skipped.
        # Default to 'DEBUG' to avoid spamming the logs; set to 'INFO' when
        # you want to surface these events without treating them as warnings.
        self._rest_slot_busy_log_level = 'DEBUG'
        # Log level used when the debounce flusher gives up after retries.
        # These events are informational in normal operation; set to 'DEBUG'
        # so they don't fill logs during normal runs.
        self._debounce_giveup_log_level = 'DEBUG'
        self._last_fetch_users_snapshot = set()
        self._last_exchange_queue_counts = {}
        self._last_queue_log_ts = 0.0
        self._queue_log_every_secs = 60.0
        self._last_loop_log_ts = 0.0
        self._loop_log_every_secs = 60.0
        self._last_mapping_log_ts_by_exchange = {}
        self._price_ticks_count = {}
        # network error log throttle map: (exchange,user) -> ts
        self._last_network_error_log_ts = {}
        self._network_error_log_throttle = 30.0
        # Max number of symbols to subscribe in one watch_tickers call
        self._price_subscribe_chunk_size = 20
        # Per-exchange overrides for subscribe chunk sizes (symbols per watch_tickers call)
        self._price_subscribe_chunk_size_by_exchange = {
            'hyperliquid': 5,
            'bitget': 5,
            'binance': 5,
            # bybit can be sensitive to large batch subscribes — use smaller chunks
            'bybit': 10,
        }
        # Stagger (ms) between starting private ws watchers to avoid bursts
        self._private_ws_stagger_ms = 200
        # Pause (s) between per-user REST calls in shared pollers to avoid bursts
        # Default small pause to reduce rate-limit triggers; can be overridden per-exchange
        # Tuned defaults to reduce observed 429s; adjust via pbgui.ini later if needed
        self._shared_rest_user_pause = 0.75
        self._shared_rest_pause_by_exchange = {
            # Increase hyperliquid pause to reduce 429s observed under load
            'hyperliquid': 3.0,
            # Increased to reduce REST fallback bursts that triggered 429s
            # (mitigation A): raised from 1.0 -> 3.0 seconds
            'bybit': 3.0,
        }
        # Per-exchange semaphore limits for REST slot gating are defined
        # near _rest_semaphores (see _rest_semaphore_limits_by_exchange).
        # Per-exchange limit for how many distinct users the price watcher may track
        # Some exchanges (hyperliquid) enforce a hard cap on tracked users for websocket topics.
        self._price_subscribe_user_limit_by_exchange = {
            'hyperliquid': 10,
        }
        self._mapping_rebuild_min_interval = 300.0  # seconds per exchange
        self._pollers_delay_seconds = 60.0
        self._pollers_enabled_after_ts = datetime.now().timestamp() + self._pollers_delay_seconds
        # Shared poller intervals (seconds)
        self._shared_combined_interval_seconds = 90
        self._shared_history_interval_seconds = 300
        self._shared_executions_interval_seconds = 1800
        # Balance has its own (longer) cadence — purely informational, no rush.
        self._poll_interval_balance_seconds = 300
        # Positions cadence — only changes after fills/liquidations.
        self._poll_interval_positions_seconds = 300
        # Orders cadence — position-per-symbol calls are expensive on HL; 60 s is still fresh enough.
        self._poll_interval_orders_seconds = 60
        # Users whose balance/positions/orders should be refreshed on the next combined cycle
        # (set by history/executions pollers when new entries are detected).
        self._balance_stale_users: set = set()
        self._positions_stale_users: set = set()
        self._orders_stale_users: set = set()
        # Flag to restart shared poller tasks when intervals change
        self._poll_intervals_changed = False
        self._pbgui_ini_mtime = None
        # Last loaded ws_max value from pbgui.ini (so we only reapply when changed)
        self._ws_max_loaded = None
        # Last loaded log_level for PBData (string like 'DEBUG'/'INFO')
        self._log_level_loaded = None
        # Snapshot of last trimmed allowed users per exchange to avoid repeated logs
        self._price_subscribe_trim_snapshot = {}
        # Track recent network-demoted users per exchange to avoid mass demotion
        self._exchange_network_error_users = defaultdict(dict)  # exchange -> {user_name: timestamp}
        self._network_error_locks = {}
        # Time window (seconds) during which only one demotion is allowed per exchange
        self._network_demotion_window = 60
        # Per-exchange backoff state and error tracking
        self._exchange_backoff_until = {}  # exchange -> timestamp until which we should backoff
        self._exchange_history_backoff_until = {}  # history-only backoff (long_history_poll) – does NOT block combined poller
        self._exchange_error_timestamps = defaultdict(list)  # exchange -> [ts1, ts2, ...]
        self._error_window_seconds = 30
        self._error_threshold = 6
        self._backoff_duration_seconds = 60
        # Timeout (seconds) used for asyncio.wait_for around ccxt.pro watch_* calls
        # Increase on slow VPS if you see ping-pong RequestTimeouts.
        self._price_watch_timeout = 120
        # Websocket restart-once state: track one restart per (exchange,user)
        # and consecutive successful watch messages to clear the restart marker.
        self._ws_restarted_once = set()  # set of (exchange, user_name)
        self._ws_success_counts = defaultdict(int)  # (exchange, user_name) -> consecutive successes
        self._ws_success_required = 3  # successes required to clear restart marker
        # Base sleep (s) before re-creating client. Increased to reduce
        # synchronized reconnect storms when many watchers detect keepalive
        # timeouts simultaneously.
        self._ws_restart_sleep = 1.5  # base sleep (s) before re-creating client
        # Per-exchange semaphore limiting concurrent reconnect attempts to 2.
        # When a Bybit/Hyperliquid server event drops all N connections at once,
        # without throttling all N tasks race to reconnect simultaneously, overwhelming
        # the asyncio event loop and delaying ping-pong for other exchanges.
        # Semaphore(2) lets at most 2 reconnects proceed at the same time; the rest
        # queue up naturally, spreading the reconnect storm over several seconds.
        self._exchange_reconnect_sem: dict = {}  # exchange -> asyncio.Semaphore(2)
        # Per-user last fetch timestamps (key: (user_name, kind) -> epoch seconds)
        # kind in {'balances','positions','orders','history'}
        self._last_fetch_ts = defaultdict(dict)
        # If a shared history poll takes longer than this (s) consider exchange overloaded
        # Increase threshold to avoid overly-aggressive backoffs for slower history endpoints
        self._long_poll_threshold_seconds = 60
        # Metrics task handle
        self._metrics_task = None
        # Metrics sampling interval (seconds), configurable via env PB_METRICS_INTERVAL
        self._metrics_interval = 10
        # ── Poller metrics (observable via API) ──────────────────
        # Per-exchange counters / timestamps that the GUI can display.
        # Written to data/logs/poller_metrics.json every _metrics_interval.
        self._poller_metrics = defaultdict(lambda: {
            'combined_last_ts': 0,        # epoch of last combined-poller cycle finish
            'combined_cycle_ms': 0,       # duration of last combined cycle (ms)
            'combined_users': 0,          # users polled in last combined cycle
            'history_last_ts': 0,         # epoch of last history-poller cycle finish
            'history_cycle_ms': 0,        # duration of last history cycle (ms)
            'history_users': 0,           # users polled in last history cycle
            'market_data_last_ts': 0,     # epoch of last market-data cycle finish
            'market_data_coins': 0,       # coins processed in last market-data cycle
            'rate_limit_429': 0,          # cumulative 429 count since startup
            'errors': 0,                  # cumulative error count since startup
            'rest_slot_timeouts': 0,      # cumulative REST slot timeout count
        })
        # ── Rate-limit budgets (token bucket per exchange) ───────
        self._rate_budgets: dict[str, RateLimitBudget] = {}
        # Per-exchange lock: ensures only ONE poller (combined/history/executions)
        # draws from the budget at a time, preventing concurrent token starvation.
        self._budget_poller_locks: dict[str, asyncio.Lock] = {}
        for _exch_name, _cfg in EXCHANGE_RATE_LIMITS.items():
            self._rate_budgets[_exch_name] = RateLimitBudget(**_cfg)
            self._budget_poller_locks[_exch_name] = asyncio.Lock()
        # (IO debugging disabled) -- per-metrics-cycle DB/process IO logging removed
        # Latest 1m candles (API) auto-refresh settings
        self._latest_1m_enabled = True
        self._latest_1m_interval_seconds = 1800
        self._latest_1m_coin_pause_seconds = 2.0
        self._latest_1m_api_timeout_seconds = 30.0
        self._latest_1m_min_lookback_days = 2
        self._latest_1m_max_lookback_days = 4
        self._latest_1m_gap_stale_minutes = 15
        self._latest_1m_hist_interval_seconds = 1800
        self._latest_1m_last_hist_scan_ts = 0.0
        self._latest_1m_task = None
        # Binance latest 1m auto-refresh settings
        self._binance_latest_1m_enabled = True
        self._binance_latest_1m_interval_seconds = 3600
        self._binance_latest_1m_coin_pause_seconds = 0.5
        self._binance_latest_1m_api_timeout_seconds = 30.0
        self._binance_latest_1m_min_lookback_days = 2
        self._binance_latest_1m_max_lookback_days = 7
        self._binance_latest_1m_task = None
        # Bybit latest 1m auto-refresh settings
        self._bybit_latest_1m_enabled = True
        self._bybit_latest_1m_interval_seconds = 3600
        self._bybit_latest_1m_coin_pause_seconds = 0.5
        self._bybit_latest_1m_api_timeout_seconds = 30.0
        self._bybit_latest_1m_min_lookback_days = 2
        self._bybit_latest_1m_max_lookback_days = 7
        self._bybit_latest_1m_task = None
        self._market_data_status_path = _Path(f"{PBGDIR}/data/logs/market_data_status.json")
        self._market_data_status_lock = asyncio.Lock()
        # Load initial settings (ws_max, log_level, ...)
        try:
            self._load_settings()
        except Exception:
            pass

        # Caller-side private-client manager (Queue + background manager task)
        # This manager serializes private-client creation requests from PBData
        # so that check+reserve logic can be performed atomically on the caller
        # side and duplicate cap-warning log spam is avoided.
        self._private_client_queue = None
        self._private_client_manager_task = None
        # manager_state contains transient reservation set and warned flags
        self._private_client_manager_state = {
            'inflight': set(),                # set of keys currently reserved by manager
            'warned_global': False,           # whether a global-cap warning was emitted
            'warned_per_exch': {},            # exchange -> bool
        }

        # REST-fallback throttling: per-exchange semaphores to limit concurrent
        # REST calls issued as fallbacks when WS handlers cannot persist data.
        self._rest_semaphores = {}
        # Tunable per-exchange concurrent REST slots (lower for sensitive exchanges)
        self._rest_semaphore_limits_by_exchange = {
            'bybit': 1,
            'hyperliquid': 2,
        }
        self._default_rest_semaphore_limit = 3
        # How long (s) to wait to acquire a REST slot before skipping the update
        self._rest_semaphore_acquire_timeout = 5.0

        # Debounce settings for WS-driven DB writes (seconds)
        self._debounce_interval_by_kind = {
            'balances': 2.0,
            'positions': 2.0,
            'orders': 5.0,
        }
        # Maximum time we'll retry gated REST writes before giving up (seconds)
        self._debounce_max_retry_seconds = 30.0

        # In-memory debounce buffers: { kind: { user_name: { 'user': User, 'first_ts': float, 'task': asyncio.Task } } }
        self._debounce_buffers = {
            'balances': {},
            'positions': {},
            'orders': {},
        }

        # Register with Exchange to be notified when private clients are closed
        try:
            from Exchange import register_private_client_close_listener
            # register a small synchronous callback that schedules the async clear
            def _on_closed_cb(exchange_id, user_name):
                try:
                    loop = asyncio.get_running_loop()
                    loop.create_task(self._private_manager_maybe_clear_flags_for_exchange(exchange_id))
                except Exception:
                    pass
            register_private_client_close_listener(_on_closed_cb)
        except Exception:
            pass





    # Logging is centralized via the module-level `_log_central` function.

    def _load_settings(self):
        """Read `pbgui.ini` and update runtime settings when file changes.

                Currently loads:
                    - pbdata.ws_max (int) -> passed to Exchange.set_ws_limits(global_max=...)
                    - pbdata.log_level (str) -> sets minimum log level for PBData

        The function uses the file mtime to avoid re-reading the INI too often.
        """
        try:
            p = Path('pbgui.ini')
            if not p.exists():
                return
            mtime = p.stat().st_mtime
            if self._pbgui_ini_mtime is not None and mtime == self._pbgui_ini_mtime:
                return
            self._pbgui_ini_mtime = mtime
            cfg = configparser.ConfigParser()
            cfg.read('pbgui.ini')

            # Note: payload debug flag removed — use log_level to control DEBUG logging.

            # ws_max (integer) - global cap for private websocket clients
            ws_max = None
            if cfg.has_option('pbdata', 'ws_max'):
                try:
                    raw = cfg.get('pbdata', 'ws_max')
                    sval = str(raw).strip() if raw is not None else ''
                    if sval != '':
                        try:
                            ws_max = int(sval)
                        except Exception:
                            ws_max = None
                except Exception:
                    ws_max = None

            if ws_max is not None and ws_max != getattr(self, '_ws_max_loaded', None):
                try:
                    set_ws_limits(global_max=ws_max)
                    self._ws_max_loaded = ws_max
                    _human_log('PBData', f"Set Exchange.ws global cap via pbgui.ini [pbdata] ws_max={ws_max}", level='INFO')
                except Exception:
                    try:
                        _human_log('PBData', f"Failed to call Exchange.set_ws_limits with ws_max={ws_max}", level='WARNING')
                    except Exception:
                        pass
            # log_level (string) - minimum log level for PBData
            log_level = None
            if cfg.has_option('pbdata', 'log_level'):
                try:
                    raw = cfg.get('pbdata', 'log_level')
                    s = str(raw).strip() if raw is not None else ''
                    if s != '':
                        log_level = s.upper()
                except Exception:
                    log_level = None

            if log_level is not None and log_level != getattr(self, '_log_level_loaded', None):
                try:
                    set_service_min_level('PBData', log_level)
                    self._log_level_loaded = log_level
                    _human_log('PBData', f"PBData log level set via pbgui.ini [pbdata] log_level={log_level}", level='INFO')
                except Exception:
                    try:
                        _human_log('PBData', f"Failed to set PBData log level to {log_level}", level='WARNING')
                    except Exception:
                        pass

            # -----------------------------
            # Timers / intervals
            # -----------------------------
            def _get_int_opt(section: str, key: str):
                try:
                    if not cfg.has_option(section, key):
                        return None
                    raw = cfg.get(section, key)
                    sval = str(raw).strip() if raw is not None else ''
                    if sval == '':
                        return None
                    return int(float(sval))
                except Exception:
                    return None

            def _get_float_opt(section: str, key: str):
                try:
                    if not cfg.has_option(section, key):
                        return None
                    raw = cfg.get(section, key)
                    sval = str(raw).strip() if raw is not None else ''
                    if sval == '':
                        return None
                    return float(sval)
                except Exception:
                    return None

            # Startup/grace period for starting shared pollers
            new_pollers_delay = _get_int_opt('pbdata', 'pollers_delay_seconds')
            if new_pollers_delay is not None and new_pollers_delay >= 0:
                if new_pollers_delay != getattr(self, '_pollers_delay_seconds', 60.0):
                    old_enable_after = getattr(self, '_pollers_enabled_after_ts', None)
                    self._pollers_delay_seconds = float(new_pollers_delay)
                    # Only re-apply enable-after if pollers have not started yet
                    try:
                        now_ts = datetime.now().timestamp()
                        if old_enable_after is not None and now_ts < old_enable_after:
                            self._pollers_enabled_after_ts = now_ts + float(new_pollers_delay)
                    except Exception:
                        pass

            # Shared poller intervals
            new_combined = _get_int_opt('pbdata', 'poll_interval_combined_seconds')
            if new_combined is not None and new_combined > 0:
                if new_combined != getattr(self, '_shared_combined_interval_seconds', 90):
                    self._shared_combined_interval_seconds = int(new_combined)
                    self._poll_intervals_changed = True

            new_history = _get_int_opt('pbdata', 'poll_interval_history_seconds')
            if new_history is not None and new_history > 0:
                if new_history != getattr(self, '_shared_history_interval_seconds', 90):
                    self._shared_history_interval_seconds = int(new_history)
                    self._poll_intervals_changed = True

            new_exec = _get_int_opt('pbdata', 'poll_interval_executions_seconds')
            if new_exec is not None and new_exec > 0:
                if new_exec != getattr(self, '_shared_executions_interval_seconds', 1800):
                    self._shared_executions_interval_seconds = int(new_exec)
                    self._poll_intervals_changed = True

            new_balance = _get_int_opt('pbdata', 'poll_interval_balance_seconds')
            if new_balance is not None and new_balance > 0:
                self._poll_interval_balance_seconds = int(new_balance)

            new_positions = _get_int_opt('pbdata', 'poll_interval_positions_seconds')
            if new_positions is not None and new_positions > 0:
                self._poll_interval_positions_seconds = int(new_positions)

            new_orders = _get_int_opt('pbdata', 'poll_interval_orders_seconds')
            if new_orders is not None and new_orders > 0:
                self._poll_interval_orders_seconds = int(new_orders)

            # Latest 1m API fetch interval
            new_latest_1m_interval = _get_int_opt('pbdata', 'latest_1m_interval_seconds')
            if new_latest_1m_interval is not None and new_latest_1m_interval > 0:
                if new_latest_1m_interval != getattr(self, '_latest_1m_interval_seconds', 120):
                    self._latest_1m_interval_seconds = int(new_latest_1m_interval)

            # Latest 1m pause between coins
            new_latest_1m_coin_pause = _get_float_opt('pbdata', 'latest_1m_coin_pause_seconds')
            if new_latest_1m_coin_pause is not None and new_latest_1m_coin_pause >= 0:
                self._latest_1m_coin_pause_seconds = float(new_latest_1m_coin_pause)

            # Latest 1m API timeout
            new_latest_1m_timeout = _get_float_opt('pbdata', 'latest_1m_api_timeout_seconds')
            if new_latest_1m_timeout is not None and new_latest_1m_timeout > 0:
                self._latest_1m_api_timeout_seconds = float(new_latest_1m_timeout)

            # Latest 1m lookback days (min/max)
            new_latest_1m_min_lb = _get_int_opt('pbdata', 'latest_1m_min_lookback_days')
            if new_latest_1m_min_lb is not None and new_latest_1m_min_lb > 0:
                self._latest_1m_min_lookback_days = int(new_latest_1m_min_lb)

            new_latest_1m_max_lb = _get_int_opt('pbdata', 'latest_1m_max_lookback_days')
            if new_latest_1m_max_lb is not None and new_latest_1m_max_lb > 0:
                self._latest_1m_max_lookback_days = int(new_latest_1m_max_lb)

            # Binance latest 1m settings
            bnc_interval = _get_int_opt('binance_data', 'latest_1m_interval_seconds')
            if bnc_interval is not None and bnc_interval > 0:
                self._binance_latest_1m_interval_seconds = int(bnc_interval)
            bnc_pause = _get_float_opt('binance_data', 'latest_1m_coin_pause_seconds')
            if bnc_pause is not None and bnc_pause >= 0:
                self._binance_latest_1m_coin_pause_seconds = float(bnc_pause)
            bnc_timeout = _get_float_opt('binance_data', 'latest_1m_api_timeout_seconds')
            if bnc_timeout is not None and bnc_timeout > 0:
                self._binance_latest_1m_api_timeout_seconds = float(bnc_timeout)
            bnc_min_lb = _get_int_opt('binance_data', 'latest_1m_min_lookback_days')
            if bnc_min_lb is not None and bnc_min_lb > 0:
                self._binance_latest_1m_min_lookback_days = int(bnc_min_lb)
            bnc_max_lb = _get_int_opt('binance_data', 'latest_1m_max_lookback_days')
            if bnc_max_lb is not None and bnc_max_lb > 0:
                self._binance_latest_1m_max_lookback_days = int(bnc_max_lb)

            # Bybit latest 1m settings
            bbt_interval = _get_int_opt('bybit_data', 'latest_1m_interval_seconds')
            if bbt_interval is not None and bbt_interval > 0:
                self._bybit_latest_1m_interval_seconds = int(bbt_interval)
            bbt_pause = _get_float_opt('bybit_data', 'latest_1m_coin_pause_seconds')
            if bbt_pause is not None and bbt_pause >= 0:
                self._bybit_latest_1m_coin_pause_seconds = float(bbt_pause)
            bbt_timeout = _get_float_opt('bybit_data', 'latest_1m_api_timeout_seconds')
            if bbt_timeout is not None and bbt_timeout > 0:
                self._bybit_latest_1m_api_timeout_seconds = float(bbt_timeout)
            bbt_min_lb = _get_int_opt('bybit_data', 'latest_1m_min_lookback_days')
            if bbt_min_lb is not None and bbt_min_lb > 0:
                self._bybit_latest_1m_min_lookback_days = int(bbt_min_lb)
            bbt_max_lb = _get_int_opt('bybit_data', 'latest_1m_max_lookback_days')
            if bbt_max_lb is not None and bbt_max_lb > 0:
                self._bybit_latest_1m_max_lookback_days = int(bbt_max_lb)

            # Pause between per-user REST calls in shared pollers
            new_rest_pause = _get_float_opt('pbdata', 'shared_rest_user_pause_seconds')
            if new_rest_pause is not None and new_rest_pause >= 0:
                self._shared_rest_user_pause = float(new_rest_pause)

            # Per-exchange overrides as JSON: {"exchange": seconds, ...}
            if cfg.has_option('pbdata', 'shared_rest_pause_by_exchange_json'):
                try:
                    raw = cfg.get('pbdata', 'shared_rest_pause_by_exchange_json')
                    sval = str(raw).strip() if raw is not None else ''
                    if sval != '':
                        obj = json.loads(sval)
                        if isinstance(obj, dict):
                            cleaned = {}
                            for k, v in obj.items():
                                try:
                                    if k is None:
                                        continue
                                    exid = str(k).strip()
                                    if exid == '':
                                        continue
                                    sec = float(v)
                                    if sec < 0:
                                        continue
                                    cleaned[exid] = sec
                                except Exception:
                                    continue
                            # merge into existing defaults
                            if cleaned:
                                self._shared_rest_pause_by_exchange.update(cleaned)
                except Exception:
                    pass

            # price_watch_timeout (float, seconds) — timeout for watch_* calls
            new_pw_timeout = _get_float_opt('pbdata', 'price_watch_timeout')
            if new_pw_timeout is not None and new_pw_timeout > 0:
                self._price_watch_timeout = float(new_pw_timeout)

            # rest_semaphore_acquire_timeout (float, seconds) — REST slot acquire timeout
            new_rest_timeout = _get_float_opt('pbdata', 'rest_semaphore_acquire_timeout')
            if new_rest_timeout is not None and new_rest_timeout > 0:
                self._rest_semaphore_acquire_timeout = float(new_rest_timeout)
        except Exception:
            return

    # -----------------------------
    # Private client manager (caller-side)
    # -----------------------------
    def _start_private_client_manager(self):
        """Lazily start the private client manager background task."""
        try:
            if self._private_client_queue is None:
                self._private_client_queue = asyncio.Queue()
            if self._private_client_manager_task is None or self._private_client_manager_task.done():
                self._private_client_manager_task = asyncio.create_task(self._private_client_manager_loop())
        except Exception:
            pass

    def _write_market_data_status(self, payload: dict) -> None:
        try:
            logs_dir = _Path(f"{PBGDIR}/data/logs")
            logs_dir.mkdir(parents=True, exist_ok=True)
            tmp = self._market_data_status_path.with_suffix(".tmp")
            with open(tmp, "w", encoding="utf-8") as f:
                json.dump(payload, f, indent=2, sort_keys=True)
            os.replace(tmp, self._market_data_status_path)
        except Exception:
            pass

    async def _update_market_data_status(self, key: str, value) -> None:
        """Atomically update a single top-level key in market_data_status.json.

        Uses an asyncio.Lock so concurrent loops (HL, Binance, …) never
        clobber each other's sections, even if they await between read and write.
        """
        async with self._market_data_status_lock:
            existing: dict = {}
            try:
                if self._market_data_status_path.exists():
                    existing = json.loads(
                        self._market_data_status_path.read_text(encoding="utf-8")
                    )
            except Exception:
                pass
            existing[key] = value
            existing["timestamp"] = datetime.now().isoformat(sep=" ", timespec="seconds")
            self._write_market_data_status(existing)

    async def _latest_1m_loop(self):
        await asyncio.sleep(5)
        _first_iter = True
        _resume_after_coin: str = ""
        while True:
            try:
                # Reload settings from pbgui.ini each loop so GUI changes
                # to latest_1m_* settings take effect without restart.
                try:
                    self._load_settings()
                except Exception:
                    pass
                
                if not self._latest_1m_enabled:
                    await asyncio.sleep(5)
                    continue

                # On first start after restart: respect remaining interval or resume crashed mid-cycle
                if _first_iter:
                    _first_iter = False
                    try:
                        _saved = json.loads(self._market_data_status_path.read_text(encoding="utf-8")).get("latest_1m", {})
                        _last_ts = float(_saved.get("last_run_ts") or 0.0)
                        _interval = float(self._latest_1m_interval_seconds)
                        _remaining = _interval - (datetime.now().timestamp() - _last_ts) - 5.0
                        if _saved.get("running"):
                            _resume_after_coin = str(_saved.get("current_coin") or "")
                            _human_log("PBData", f"[market-data] HL latest_1m: resuming after restart, skipping up to {_resume_after_coin!r}", level="INFO")
                        elif _remaining > 2.0:
                            _human_log("PBData", f"[market-data] HL latest_1m: {_remaining:.0f}s remaining in cycle — waiting on restart", level="INFO")
                            _hl_flag_r = _Path(f"{PBGDIR}/data/logs/hyperliquid_latest_1m_run_now.flag")
                            if await _wait_for_flag(_hl_flag_r, _remaining):
                                try:
                                    _hl_flag_r.unlink(missing_ok=True)
                                except Exception:
                                    pass
                    except Exception:
                        pass

                cfg = load_market_data_config()
                coins = list(cfg.enabled_coins.get("hyperliquid", []) or [])
                coins = [str(c).strip().upper() for c in coins if str(c).strip()]
                invalid_live_meta_coins: set[str] = set()

                # Determine how many coins to skip when resuming a crashed mid-cycle
                _coins_done_offset = 0
                if _resume_after_coin:
                    if _resume_after_coin in coins:
                        _coins_done_offset = coins.index(_resume_after_coin) + 1
                    else:
                        _resume_after_coin = ""  # coin no longer enabled, start fresh

                now = datetime.now()
                now_ts = now.timestamp()
                # Seed with previous coin results so table stays filled during the run
                _prev_hl: dict = {}
                try:
                    _prev_hl = json.loads(self._market_data_status_path.read_text(encoding="utf-8")).get("latest_1m", {}).get("coins", {})
                except Exception:
                    pass
                status = {
                    "exchange": "hyperliquid",
                    "interval_seconds": int(self._latest_1m_interval_seconds),
                    "last_run_ts": int(now_ts),
                    "running": True,
                    "current_coin": None,
                    "coins_done": _coins_done_offset,
                    "coins_total": len(coins),
                    "coins": dict(_prev_hl),
                }
                try:
                    await self._update_market_data_status("latest_1m", status)
                except Exception:
                    pass

                _skipping = bool(_resume_after_coin)
                for coin in coins:
                    if _skipping:
                        if coin == _resume_after_coin:
                            _skipping = False
                            _resume_after_coin = ""
                        continue
                    coin_status = {
                        "last_fetch": None,
                        "result": "skipped",
                    }
                    max_lb = int(self._latest_1m_max_lookback_days)
                    lookback_days = int(self._latest_1m_min_lookback_days)
                    newest_day = ""
                    try:
                        cov = get_daily_hour_coverage_for_dataset("hyperliquid", "1m_api", coin)
                        newest_day = str(cov.get("newest_day") or "")
                        if newest_day:
                            d_new = datetime.strptime(newest_day, "%Y%m%d").date()
                            days_since = (datetime.utcnow().date() - d_new).days
                            if days_since < 0:
                                days_since = 0
                            lookback_days = max(lookback_days, days_since + 1)
                        else:
                            # No local API data yet: pull the full allowed lookback window.
                            lookback_days = max_lb
                            coin_status["note"] = "no_local_api_data"
                    except Exception as e:
                        coin_status["error"] = f"coverage:{type(e).__name__}"

                    if lookback_days > max_lb:
                        coin_status["note"] = "api_window_limited"
                        lookback_days = max_lb

                    try:
                        # Acquire rate budget for HL candleSnapshot.
                        # Weight = 44/day (20 base + ceil(1440 candles/60) = 44 per HL docs).
                        # update_latest_hyperliquid_1m_api_for_coin makes one API call per
                        # missing day → acquire the full upper bound up-front.
                        _cs_weight = get_weight('hyperliquid', 'candle_snapshot') * max(1, int(lookback_days))
                        if not await self._acquire_rate_budget('hyperliquid', 'candle_snapshot', timeout=120.0, weight_override=_cs_weight):
                            coin_status["last_fetch"] = datetime.now().isoformat(sep=" ", timespec="seconds")
                            coin_status["result"] = "budget_timeout"
                        else:
                            res = await asyncio.to_thread(
                                update_latest_hyperliquid_1m_api_for_coin,
                                coin=coin,
                                lookback_days=int(lookback_days),
                                overwrite=False,
                                dry_run=False,
                                timeout_s=float(self._latest_1m_api_timeout_seconds),
                            )
                            coin_status["last_fetch"] = datetime.now().isoformat(sep=" ", timespec="seconds")
                            coin_status["result"] = "ok"
                            coin_status["lookback_days"] = int(lookback_days)
                            coin_status["newest_day"] = newest_day
                            coin_status["api_result"] = res
                            try:
                                if isinstance(res, dict) and bool(res.get("skipped")) and str(res.get("skip_reason") or "") == "not_in_live_meta":
                                    invalid_live_meta_coins.add(str(coin).strip().upper())
                            except Exception:
                                pass
                            try:
                                _refresh_inventory_coin("hyperliquid", "1m_api", coin)
                            except Exception:
                                pass
                    except Exception as e:
                        coin_status["last_fetch"] = datetime.now().isoformat(sep=" ", timespec="seconds")
                        coin_status["result"] = "error"
                        coin_status["error"] = str(e)

                    status["coins"][coin] = coin_status
                    status["coins_done"] = status.get("coins_done", 0) + 1
                    status["current_coin"] = coin
                    try:
                        await self._update_market_data_status("latest_1m", status)
                    except Exception:
                        pass

                    # Check stop flag
                    _hl_stop = _Path(f"{PBGDIR}/data/logs/hyperliquid_latest_1m_stop.flag")
                    if _hl_stop.exists():
                        try:
                            _hl_stop.unlink(missing_ok=True)
                        except Exception:
                            pass
                        status["running"] = False
                        status["current_coin"] = None
                        status["stopped"] = True
                        status["last_run_duration_s"] = int(datetime.now().timestamp() - now_ts)
                        try:
                            await self._update_market_data_status("latest_1m", status)
                        except Exception:
                            pass
                        break

                    # Pause between coins to avoid rate limits
                    if self._latest_1m_coin_pause_seconds > 0:
                        await asyncio.sleep(float(self._latest_1m_coin_pause_seconds))

                _resume_after_coin = ""  # consumed — next iteration starts fresh
                if invalid_live_meta_coins:
                    try:
                        updated_enabled = [c for c in coins if str(c).strip().upper() not in invalid_live_meta_coins]
                        set_enabled_coins("hyperliquid", updated_enabled)
                        _human_log(
                            "PBData",
                            f"[market-data] removed invalid live-meta coins from enabled list: {', '.join(sorted(invalid_live_meta_coins))}",
                            level="WARNING",
                        )
                    except Exception as e:
                        _human_log("PBData", f"[market-data] failed to auto-prune invalid live-meta coins: {e}", level="WARNING")

                if now_ts - float(self._latest_1m_last_hist_scan_ts or 0.0) >= float(self._latest_1m_hist_interval_seconds):
                    self._latest_1m_last_hist_scan_ts = now_ts
                    hist = {
                        "exchange": "hyperliquid",
                        "last_scan_ts": int(now_ts),
                        "coins": {},
                    }
                    for coin in coins:
                        cov = get_daily_hour_coverage_for_dataset("hyperliquid", "l2Book", coin)
                        hist["coins"][coin] = {
                            "oldest_day": str(cov.get("oldest_day") or ""),
                            "newest_day": str(cov.get("newest_day") or ""),
                            "missing_days_count": int(cov.get("missing_days_count") or 0),
                            "coverage_pct": float(cov.get("coverage_pct") or 0.0),
                        }
                    status["historical"] = hist

                status["running"] = False
                status["current_coin"] = None
                status["last_run_duration_s"] = int(datetime.now().timestamp() - now_ts)
                await self._update_market_data_status("latest_1m", status)
                if "historical" in status:
                    await self._update_market_data_status("historical", status["historical"])
            except Exception as e:
                try:
                    _human_log('PBData', f"[market-data] latest_1m loop error: {e}", level='WARNING')
                except Exception:
                    pass

            _hl_flag = _Path(f"{PBGDIR}/data/logs/hyperliquid_latest_1m_run_now.flag")
            try:
                _elapsed = datetime.now().timestamp() - now_ts
                _remaining_wait = max(0.0, float(self._latest_1m_interval_seconds) - _elapsed)
            except Exception:
                _remaining_wait = float(self._latest_1m_interval_seconds)
            if await _wait_for_flag(_hl_flag, _remaining_wait):
                try:
                    _hl_flag.unlink(missing_ok=True)
                except Exception:
                    pass

    async def _binance_latest_1m_loop(self):
        """Background loop: refresh Binance USDM 1m candles for enabled coins."""
        await asyncio.sleep(8)  # Slight offset from HL loop
        _first_iter = True
        _resume_after_coin: str = ""
        while True:
            try:
                try:
                    self._load_settings()
                except Exception:
                    pass

                if not self._binance_latest_1m_enabled:
                    await asyncio.sleep(5)
                    continue

                # On first start after restart: respect remaining interval or resume crashed mid-cycle
                if _first_iter:
                    _first_iter = False
                    try:
                        _saved = json.loads(self._market_data_status_path.read_text(encoding="utf-8")).get("binance_latest_1m", {})
                        _last_ts = float(_saved.get("last_run_ts") or 0.0)
                        _interval = float(self._binance_latest_1m_interval_seconds)
                        _remaining = _interval - (datetime.now().timestamp() - _last_ts) - 8.0
                        if _saved.get("running"):
                            _resume_after_coin = str(_saved.get("current_coin") or "")
                            _human_log("PBData", f"[market-data] Binance latest_1m: resuming after restart, skipping up to {_resume_after_coin!r}", level="INFO")
                        elif _remaining > 2.0:
                            _human_log("PBData", f"[market-data] Binance latest_1m: {_remaining:.0f}s remaining in cycle — waiting on restart", level="INFO")
                            _bnc_flag_r = _Path(f"{PBGDIR}/data/logs/binance_latest_1m_run_now.flag")
                            if await _wait_for_flag(_bnc_flag_r, _remaining):
                                try:
                                    _bnc_flag_r.unlink(missing_ok=True)
                                except Exception:
                                    pass
                    except Exception:
                        pass

                cfg = load_market_data_config()
                coins = list(cfg.enabled_coins.get("binance", []) or [])
                coins = [str(c).strip().upper() for c in coins if str(c).strip()]

                if not coins:
                    await asyncio.sleep(float(self._binance_latest_1m_interval_seconds))
                    continue

                # Determine how many coins to skip when resuming a crashed mid-cycle
                _coins_done_offset = 0
                if _resume_after_coin:
                    if _resume_after_coin in coins:
                        _coins_done_offset = coins.index(_resume_after_coin) + 1
                    else:
                        _resume_after_coin = ""  # coin no longer enabled, start fresh

                now = datetime.now()
                now_ts = now.timestamp()
                # Seed with previous coin results so table stays filled during the run
                _prev_bnc: dict = {}
                try:
                    _prev_bnc = json.loads(self._market_data_status_path.read_text(encoding="utf-8")).get("binance_latest_1m", {}).get("coins", {})
                except Exception:
                    pass
                status_bnc = {
                    "exchange": "binanceusdm",
                    "interval_seconds": int(self._binance_latest_1m_interval_seconds),
                    "last_run_ts": int(now_ts),
                    "running": True,
                    "current_coin": None,
                    "coins_done": _coins_done_offset,
                    "coins_total": len(coins),
                    "coins": dict(_prev_bnc),
                }
                try:
                    await self._update_market_data_status("binance_latest_1m", status_bnc)
                except Exception:
                    pass

                _skipping = bool(_resume_after_coin)
                for coin in coins:
                    if _skipping:
                        if coin == _resume_after_coin:
                            _skipping = False
                            _resume_after_coin = ""
                        continue
                    coin_status: dict = {"last_fetch": None, "result": "skipped"}
                    max_lb = int(self._binance_latest_1m_max_lookback_days)
                    lookback_days = int(self._binance_latest_1m_min_lookback_days)

                    try:
                        from binance_best_1m import get_newest_day
                        newest_day = get_newest_day(coin) or ""
                        if newest_day:
                            d_new = datetime.strptime(newest_day, "%Y%m%d").date()
                            days_since = (datetime.utcnow().date() - d_new).days
                            if days_since < 0:
                                days_since = 0
                            lookback_days = max(lookback_days, days_since + 1)
                        else:
                            lookback_days = max_lb
                            coin_status["note"] = "no_local_data"
                    except Exception as e:
                        coin_status["error"] = f"coverage:{type(e).__name__}"

                    if lookback_days > max_lb:
                        coin_status["note"] = "window_limited"
                        lookback_days = max_lb

                    try:
                        res = await asyncio.to_thread(
                            update_latest_binance_1m_for_coin,
                            coin=coin,
                            lookback_days=int(lookback_days),
                            overwrite=True,
                            timeout_s=float(self._binance_latest_1m_api_timeout_seconds),
                        )
                        coin_status["last_fetch"] = datetime.now().isoformat(sep=" ", timespec="seconds")
                        coin_status["result"] = "ok"
                        coin_status["lookback_days"] = int(lookback_days)
                        coin_status["api_result"] = res
                        try:
                            _refresh_inventory_coin("binanceusdm", "1m_api", coin)
                        except Exception:
                            pass
                    except Exception as e:
                        coin_status["last_fetch"] = now.isoformat(sep=" ", timespec="seconds")
                        coin_status["result"] = "error"
                        coin_status["error"] = str(e)

                    status_bnc["coins"][coin] = coin_status
                    status_bnc["coins_done"] = status_bnc.get("coins_done", 0) + 1
                    status_bnc["current_coin"] = coin
                    try:
                        await self._update_market_data_status("binance_latest_1m", status_bnc)
                    except Exception:
                        pass

                    # Check stop flag
                    _bnc_stop = _Path(f"{PBGDIR}/data/logs/binance_latest_1m_stop.flag")
                    if _bnc_stop.exists():
                        try:
                            _bnc_stop.unlink(missing_ok=True)
                        except Exception:
                            pass
                        status_bnc["running"] = False
                        status_bnc["current_coin"] = None
                        status_bnc["stopped"] = True
                        status_bnc["last_run_duration_s"] = int(datetime.now().timestamp() - now_ts)
                        try:
                            await self._update_market_data_status("binance_latest_1m", status_bnc)
                        except Exception:
                            pass
                        break

                    if self._binance_latest_1m_coin_pause_seconds > 0:
                        await asyncio.sleep(float(self._binance_latest_1m_coin_pause_seconds))

                _resume_after_coin = ""  # consumed — next iteration starts fresh
                # Merge into market data status (locked to avoid clobbering other loops)
                status_bnc["running"] = False
                status_bnc["current_coin"] = None
                status_bnc["last_run_duration_s"] = int(datetime.now().timestamp() - now_ts)
                try:
                    await self._update_market_data_status("binance_latest_1m", status_bnc)
                except Exception:
                    pass

            except Exception as e:
                try:
                    _human_log('PBData', f"[market-data] binance_latest_1m loop error: {e}", level='WARNING')
                except Exception:
                    pass

            _bnc_flag = _Path(f"{PBGDIR}/data/logs/binance_latest_1m_run_now.flag")
            try:
                _elapsed = datetime.now().timestamp() - now_ts
                _remaining_wait = max(0.0, float(self._binance_latest_1m_interval_seconds) - _elapsed)
            except Exception:
                _remaining_wait = float(self._binance_latest_1m_interval_seconds)
            if await _wait_for_flag(_bnc_flag, _remaining_wait):
                try:
                    _bnc_flag.unlink(missing_ok=True)
                except Exception:
                    pass

    async def _bybit_latest_1m_loop(self):
        """Background loop: refresh Bybit 1m candles for enabled coins."""
        await asyncio.sleep(16)  # Slight offset from binance loop
        _first_iter = True
        _resume_after_coin: str = ""
        while True:
            try:
                try:
                    self._load_settings()
                except Exception:
                    pass

                if not self._bybit_latest_1m_enabled:
                    await asyncio.sleep(5)
                    continue

                if _first_iter:
                    _first_iter = False
                    try:
                        _saved = json.loads(self._market_data_status_path.read_text(encoding="utf-8")).get("bybit_latest_1m", {})
                        _last_ts = float(_saved.get("last_run_ts") or 0.0)
                        _interval = float(self._bybit_latest_1m_interval_seconds)
                        _remaining = _interval - (datetime.now().timestamp() - _last_ts) - 16.0
                        if _saved.get("running"):
                            _resume_after_coin = str(_saved.get("current_coin") or "")
                            _human_log("PBData", f"[market-data] Bybit latest_1m: resuming after restart, skipping up to {_resume_after_coin!r}", level="INFO")
                        elif _remaining > 2.0:
                            _human_log("PBData", f"[market-data] Bybit latest_1m: {_remaining:.0f}s remaining in cycle — waiting on restart", level="INFO")
                            _bbt_flag_r = _Path(f"{PBGDIR}/data/logs/bybit_latest_1m_run_now.flag")
                            if await _wait_for_flag(_bbt_flag_r, _remaining):
                                try:
                                    _bbt_flag_r.unlink(missing_ok=True)
                                except Exception:
                                    pass
                    except Exception:
                        pass

                cfg = load_market_data_config()
                coins = list(cfg.enabled_coins.get("bybit", []) or [])
                coins = [str(c).strip().upper() for c in coins if str(c).strip()]

                if not coins:
                    await asyncio.sleep(float(self._bybit_latest_1m_interval_seconds))
                    continue

                _coins_done_offset = 0
                if _resume_after_coin:
                    if _resume_after_coin in coins:
                        _coins_done_offset = coins.index(_resume_after_coin) + 1
                    else:
                        _resume_after_coin = ""

                now = datetime.now()
                now_ts = now.timestamp()
                _prev_bbt: dict = {}
                try:
                    _prev_bbt = json.loads(self._market_data_status_path.read_text(encoding="utf-8")).get("bybit_latest_1m", {}).get("coins", {})
                except Exception:
                    pass
                status_bbt = {
                    "exchange": "bybit",
                    "interval_seconds": int(self._bybit_latest_1m_interval_seconds),
                    "last_run_ts": int(now_ts),
                    "running": True,
                    "current_coin": None,
                    "coins_done": _coins_done_offset,
                    "coins_total": len(coins),
                    "coins": dict(_prev_bbt),
                }
                try:
                    await self._update_market_data_status("bybit_latest_1m", status_bbt)
                except Exception:
                    pass

                _skipping = bool(_resume_after_coin)
                for coin in coins:
                    if _skipping:
                        if coin == _resume_after_coin:
                            _skipping = False
                            _resume_after_coin = ""
                        continue
                    coin_status: dict = {"last_fetch": None, "result": "skipped"}
                    max_lb = int(self._bybit_latest_1m_max_lookback_days)
                    lookback_days = int(self._bybit_latest_1m_min_lookback_days)

                    try:
                        from bybit_best_1m import get_newest_day as bybit_get_newest_day
                        newest_day = bybit_get_newest_day(coin) or ""
                        if newest_day:
                            d_new = datetime.strptime(newest_day, "%Y%m%d").date()
                            days_since = (datetime.utcnow().date() - d_new).days
                            if days_since < 0:
                                days_since = 0
                            lookback_days = max(lookback_days, days_since + 1)
                        else:
                            lookback_days = max_lb
                            coin_status["note"] = "no_local_data"
                    except Exception as e:
                        coin_status["error"] = f"coverage:{type(e).__name__}"

                    if lookback_days > max_lb:
                        coin_status["note"] = "window_limited"
                        lookback_days = max_lb

                    try:
                        from bybit_best_1m import update_latest_bybit_1m_for_coin
                        res = await asyncio.to_thread(
                            update_latest_bybit_1m_for_coin,
                            coin=coin,
                            lookback_days=int(lookback_days),
                            overwrite=True,
                            timeout_s=float(self._bybit_latest_1m_api_timeout_seconds),
                        )
                        coin_status["last_fetch"] = datetime.now().isoformat(sep=" ", timespec="seconds")
                        coin_status["result"] = "ok"
                        coin_status["lookback_days"] = int(lookback_days)
                        coin_status["api_result"] = res
                        try:
                            _refresh_inventory_coin("bybit", "1m", coin)
                        except Exception:
                            pass
                    except Exception as e:
                        coin_status["last_fetch"] = datetime.now().isoformat(sep=" ", timespec="seconds")
                        coin_status["result"] = "error"
                        coin_status["error"] = str(e)

                    status_bbt["coins"][coin] = coin_status
                    status_bbt["coins_done"] = status_bbt.get("coins_done", 0) + 1
                    status_bbt["current_coin"] = coin
                    try:
                        await self._update_market_data_status("bybit_latest_1m", status_bbt)
                    except Exception:
                        pass

                    _bbt_stop = _Path(f"{PBGDIR}/data/logs/bybit_latest_1m_stop.flag")
                    if _bbt_stop.exists():
                        try:
                            _bbt_stop.unlink(missing_ok=True)
                        except Exception:
                            pass
                        status_bbt["running"] = False
                        status_bbt["current_coin"] = None
                        status_bbt["stopped"] = True
                        status_bbt["last_run_duration_s"] = int(datetime.now().timestamp() - now_ts)
                        try:
                            await self._update_market_data_status("bybit_latest_1m", status_bbt)
                        except Exception:
                            pass
                        break

                    if self._bybit_latest_1m_coin_pause_seconds > 0:
                        await asyncio.sleep(float(self._bybit_latest_1m_coin_pause_seconds))

                _resume_after_coin = ""
                status_bbt["running"] = False
                status_bbt["current_coin"] = None
                status_bbt["last_run_duration_s"] = int(datetime.now().timestamp() - now_ts)
                try:
                    await self._update_market_data_status("bybit_latest_1m", status_bbt)
                except Exception:
                    pass

            except Exception as e:
                try:
                    _human_log('PBData', f"[market-data] bybit_latest_1m loop error: {e}", level='WARNING')
                except Exception:
                    pass

            _bbt_flag = _Path(f"{PBGDIR}/data/logs/bybit_latest_1m_run_now.flag")
            try:
                _elapsed = datetime.now().timestamp() - now_ts
                _remaining_wait = max(0.0, float(self._bybit_latest_1m_interval_seconds) - _elapsed)
            except Exception:
                _remaining_wait = float(self._bybit_latest_1m_interval_seconds)
            if await _wait_for_flag(_bbt_flag, _remaining_wait):
                try:
                    _bbt_flag.unlink(missing_ok=True)
                except Exception:
                    pass

    async def request_private_client(self, exchange_id: str, user, caller: str = None):
        """Public API for callers to request a private client.

        Returns a client instance or None. Internally queues request to
        manager which performs serialized check+reserve and calls
        Exchange.get_private_ws_client.
        """
        try:
            # Ensure manager task running
            self._start_private_client_manager()
            loop = asyncio.get_running_loop()
            fut = loop.create_future()
            await self._private_client_queue.put((exchange_id, user, caller, fut))
            return await fut
        except Exception:
            return None

    async def _private_client_manager_loop(self):
        """Background manager: serially process creation requests.

        For each request we compute current counts (via Exchange.get_client_metrics()),
        decide whether to reserve a slot or return None, and call
        Exchange.get_private_ws_client while the reservation is held.
        """
        from Exchange import Exchange as _ExchCls
        import Exchange as _ExMod
        while True:
            try:
                exchange_id, user, caller, fut = await self._private_client_queue.get()
            except asyncio.CancelledError:
                break
            except Exception:
                # Shouldn't happen, but protect the loop
                try:
                    fut.set_result(None)
                except Exception:
                    pass
                continue

            key = f"{('kucoinfutures' if exchange_id == 'kucoin' else exchange_id)}:{getattr(user,'name',None)}"
            try:
                # Fast path: if Exchange already has a private client, return it
                try:
                    client = _ExchCls._private_ws_clients.get(key)
                    if client is not None:
                        fut.set_result(client)
                        continue
                except Exception:
                    pass

                # Compute effective caps
                try:
                    global_cap = int(getattr(_ExMod, '_RUNTIME_MAX_PRIVATE_WS_GLOBAL')) if getattr(_ExMod, '_RUNTIME_MAX_PRIVATE_WS_GLOBAL', None) is not None else getattr(_ExMod, 'MAX_PRIVATE_WS_GLOBAL')
                except Exception:
                    global_cap = getattr(_ExMod, 'MAX_PRIVATE_WS_GLOBAL')

                try:
                    runtime_per = getattr(_ExMod, '_RUNTIME_MAX_PRIVATE_WS_PER_EXCHANGE') or {}
                except Exception:
                    runtime_per = {}

                # Count existing private clients
                try:
                    metrics = _ExchCls.get_client_metrics()
                    total_now = sum(v.get('private_count', 0) for v in metrics.values())
                except Exception:
                    total_now = len(_ExchCls._private_ws_clients.keys())

                inflight_count = len(self._private_client_manager_state['inflight'])

                # Global cap check
                if global_cap is not None and (total_now + inflight_count) >= global_cap:
                    # emit warning once (use DEBUG to avoid spamming production logs)
                    if not self._private_client_manager_state.get('warned_global'):
                        try:
                            # Include a compact snapshot to aid debugging: client metrics
                            # and currently reserved inflight keys. Keep it concise.
                            snapshot = dict(metrics) if isinstance(metrics, dict) else {}
                            infl = list(self._private_client_manager_state.get('inflight', []))
                            _human_log('PBData', f"[ws-manager] reached GLOBAL cap ({total_now + inflight_count}/{global_cap}); returning None for user={getattr(user,'name',None)} snapshot_metrics={snapshot} inflight={infl}", level='DEBUG', user=user)
                        except Exception:
                            pass
                        self._private_client_manager_state['warned_global'] = True
                    try:
                        fut.set_result(None)
                    except Exception:
                        pass
                    continue

                # Per-exchange cap
                base_key = 'kucoinfutures' if exchange_id == 'kucoin' else exchange_id
                try:
                    if runtime_per and base_key in runtime_per:
                        cap = int(runtime_per.get(base_key))
                    else:
                        cap = getattr(_ExMod, 'MAX_PRIVATE_WS_PER_EXCHANGE').get(base_key)
                except Exception:
                    cap = None

                if cap is not None:
                    try:
                        # current per-exchange existing
                        current = 0
                        for k in _ExchCls._private_ws_clients.keys():
                            if k.startswith(f"{base_key}:"):
                                current += 1
                    except Exception:
                        current = 0
                    inflight_for_exch = sum(1 for c in self._private_client_manager_state['inflight'] if c.startswith(f"{base_key}:"))
                    if (current + inflight_for_exch) >= cap:
                        warned = self._private_client_manager_state['warned_per_exch'].get(base_key, False)
                        if not warned:
                            try:
                                snapshot = dict(metrics) if isinstance(metrics, dict) else {}
                                infl = list(self._private_client_manager_state.get('inflight', []))
                                _human_log('PBData', f"[ws-manager] reached cap for {base_key} ({current + inflight_for_exch}/{cap}); returning None for user={getattr(user,'name',None)} snapshot_metrics={snapshot} inflight={infl}", level='DEBUG', user=user)
                            except Exception:
                                pass
                            self._private_client_manager_state['warned_per_exch'][base_key] = True
                        try:
                            fut.set_result(None)
                        except Exception:
                            pass
                        continue

                # Reserve an in-flight slot and perform creation
                try:
                    self._private_client_manager_state['inflight'].add(key)
                except Exception:
                    pass

                try:
                    # Call into Exchange to create/get client. Exchange itself may
                    # perform its own protections; manager ensures callers are
                    # serialized enough to avoid duplicate logs and overshoot.
                    client = await _ExchCls.get_private_ws_client(exchange_id, user, caller=caller)
                    try:
                        fut.set_result(client)
                    except Exception:
                        pass
                except Exception:
                    try:
                        fut.set_result(None)
                    except Exception:
                        pass
                finally:
                    try:
                        self._private_client_manager_state['inflight'].discard(key)
                    except Exception:
                        pass
                    # Maybe clear warning flags if capacity freed
                    try:
                        # Recompute counts and clear flags if below caps
                        metrics = _ExchCls.get_client_metrics()
                        total_now = sum(v.get('private_count', 0) for v in metrics.values())
                        if global_cap is not None and total_now + len(self._private_client_manager_state['inflight']) < global_cap:
                            self._private_client_manager_state['warned_global'] = False
                        # per-exchange
                        try:
                            curr_ex = 0
                            for k in _ExchCls._private_ws_clients.keys():
                                if k.startswith(f"{base_key}:"):
                                    curr_ex += 1
                        except Exception:
                            curr_ex = 0
                        if cap is not None and (curr_ex + sum(1 for c in self._private_client_manager_state['inflight'] if c.startswith(f"{base_key}:"))) < cap:
                            self._private_client_manager_state['warned_per_exch'][base_key] = False
                    except Exception:
                        pass
            except Exception:
                try:
                    fut.set_result(None)
                except Exception:
                    pass

    async def _private_manager_maybe_clear_flags_for_exchange(self, exchange_id: str):
        """Recompute client counts and clear manager warning flags if capacity freed.

        This is called when Exchange notifies PBData that a private client was
        closed (or by internal manager after finishing a creation) so that
        the manager's 'warned' flags don't stay stuck.
        """
        try:
            from Exchange import Exchange as _ExchCls
            import Exchange as _ExMod
            base_key = 'kucoinfutures' if exchange_id == 'kucoin' else exchange_id
            try:
                metrics = _ExchCls.get_client_metrics()
                total_now = sum(v.get('private_count', 0) for v in metrics.values())
            except Exception:
                total_now = len(_ExchCls._private_ws_clients.keys())

            # global cap
            try:
                global_cap = int(getattr(_ExMod, '_RUNTIME_MAX_PRIVATE_WS_GLOBAL')) if getattr(_ExMod, '_RUNTIME_MAX_PRIVATE_WS_GLOBAL', None) is not None else getattr(_ExMod, 'MAX_PRIVATE_WS_GLOBAL')
            except Exception:
                global_cap = getattr(_ExMod, 'MAX_PRIVATE_WS_GLOBAL')
            if global_cap is not None and total_now + len(self._private_client_manager_state['inflight']) < global_cap:
                self._private_client_manager_state['warned_global'] = False

            # per-exchange cap
            try:
                runtime_per = getattr(_ExMod, '_RUNTIME_MAX_PRIVATE_WS_PER_EXCHANGE') or {}
            except Exception:
                runtime_per = {}
            try:
                if runtime_per and base_key in runtime_per:
                    cap = int(runtime_per.get(base_key))
                else:
                    cap = getattr(_ExMod, 'MAX_PRIVATE_WS_PER_EXCHANGE').get(base_key)
            except Exception:
                cap = None
            try:
                curr_ex = metrics.get(base_key, {}).get('private_count', 0)
            except Exception:
                curr_ex = 0
            if cap is not None and (curr_ex + sum(1 for c in self._private_client_manager_state['inflight'] if c.startswith(f"{base_key}:"))) < cap:
                try:
                    self._private_client_manager_state['warned_per_exch'][base_key] = False
                except Exception:
                    pass
        except Exception:
            pass


    async def _enqueue_debounce(self, kind: str, user):
        """Enqueue a debounce flush for `kind` ('balances'|'positions'|'orders') for `user`.

        Multiple calls within the debounce interval reset the timer. The actual
        DB write is performed by `_debounce_flusher` which respects `_rest_slot`
        gating and will retry for up to `_debounce_max_retry_seconds` before
        dropping the update.
        """
        try:
            if kind not in self._debounce_buffers:
                return
            buf = self._debounce_buffers[kind]
            name = getattr(user, 'name', None)
            now = datetime.now().timestamp()
            entry = buf.get(name)
            # Cancel existing timer task if present (to reset debounce)
            if entry is None:
                task = asyncio.create_task(self._debounce_flusher(kind, name))
                buf[name] = {'user': user, 'first_ts': now, 'task': task}
            else:
                # update user reference and reset timer
                entry['user'] = user
                entry_task = entry.get('task')
                try:
                    if entry_task and not entry_task.done():
                        entry_task.cancel()
                except Exception:
                    pass
                entry['task'] = asyncio.create_task(self._debounce_flusher(kind, name))
        except Exception:
            try:
                _human_log('PBData', f"[debounce] failed to enqueue {kind} for {getattr(user,'name',None)}", level='DEBUG')
            except Exception:
                pass

    async def _debounce_flusher(self, kind: str, user_name: str):
        """Task that waits the debounce interval, then attempts to persist the
        latest update for `user_name`/`kind`. Respects REST semaphore gating and
        will retry with small backoff until `_debounce_max_retry_seconds` have
        elapsed since the first queued event.
        """
        try:
            buf = self._debounce_buffers.get(kind, {})
            entry = buf.get(user_name)
            if not entry:
                return
            interval = float(self._debounce_interval_by_kind.get(kind, 2.0))
            first_ts = entry.get('first_ts', datetime.now().timestamp())
            # initial debounce sleep
            await asyncio.sleep(interval)
            # retrieve latest user ref
            entry = buf.get(user_name)
            if not entry:
                return
            user = entry.get('user')
            # If user object missing, try to find by name
            if user is None:
                try:
                    user = self.users.find_user(user_name)
                except Exception:
                    user = None
            if not user:
                # Nothing to flush
                try:
                    buf.pop(user_name, None)
                except Exception:
                    pass
                return

            start_retry = datetime.now().timestamp()
            while True:
                tried = False
                try:
                    # Use rest_slot gating for the actual DB write
                    async with self._rest_slot(user.exchange) as got:
                        if got:
                            if kind == 'balances':
                                await asyncio.to_thread(self.db.update_balances, user)
                                asyncio.create_task(_notify_api_balance())
                            elif kind == 'positions':
                                await asyncio.to_thread(self.db.update_positions, user)
                                asyncio.create_task(_notify_api_positions(user.name))
                            elif kind == 'orders':
                                await asyncio.to_thread(self.db.update_orders, user)
                            try:
                                self._last_fetch_ts[(user.name, kind)] = datetime.now().timestamp()
                            except Exception:
                                pass
                            try:
                                self._write_fetch_summary()
                            except Exception:
                                pass
                            # success: remove from buffer and exit
                            try:
                                buf.pop(user_name, None)
                            except Exception:
                                pass
                            return
                        else:
                            tried = True
                            # Could not acquire REST slot; fall through to retry logic
                except asyncio.CancelledError:
                    # task was cancelled (likely due to a reset); stop quietly
                    return
                except Exception as e:
                    tried = True
                    try:
                        tb = traceback.format_exc()
                    except Exception:
                        tb = None
                    # Detect permanent auth errors (missing credentials) — no point retrying
                    is_auth_error = False
                    try:
                        try:
                            from ccxt.base.errors import AuthenticationError as _AuthError
                        except Exception:
                            _AuthError = None
                        if _AuthError is not None and isinstance(e, _AuthError):
                            is_auth_error = True
                    except Exception:
                        pass

                    if is_auth_error:
                        try:
                            _human_log('PBData', f"[debounce] permanent auth error for {kind} {user_name}: {e} (dropped, will not retry)", level='WARNING', meta={'traceback': tb} if tb else None)
                        except Exception:
                            pass
                        try:
                            buf.pop(user_name, None)
                        except Exception:
                            pass
                        return

                    # Detect ccxt InvalidNonce (recv_window / server timestamp mismatch)
                    is_invalid_nonce = False
                    try:
                        # Try import ccxt error class if available
                        try:
                            from ccxt.base.errors import InvalidNonce as _InvalidNonce
                        except Exception:
                            _InvalidNonce = None
                        if _InvalidNonce is not None and isinstance(e, _InvalidNonce):
                            is_invalid_nonce = True
                        else:
                            # fallback: inspect message for recv_window/server timestamp pattern
                            msg = str(e).lower()
                            if 'invalid request' in msg and 'recv_window' in msg:
                                is_invalid_nonce = True
                    except Exception:
                        is_invalid_nonce = False

                    if is_invalid_nonce:
                        # Parse recv_window/server timestamps if possible and include in logs
                        try:
                            parsed = self._parse_recv_window_from_exception(e)
                        except Exception:
                            parsed = {}
                        try:
                            meta = {'traceback': tb} if tb else {}
                            if parsed:
                                meta.update(parsed)
                            if not meta:
                                meta = None
                            _human_log('PBData', f"[debounce] InvalidNonce / recv_window error for {user.exchange} {user_name}: {e}", level='WARNING', meta=meta, user=user)
                        except Exception:
                            pass
                        try:
                            # Choose backoff duration based on observed timestamp delta when available.
                            dur = None
                            if parsed and parsed.get('delta_ms') is not None:
                                # If server timestamp is ahead of request by more than recv_window,
                                # backoff at least long enough for human-visible recovery; clamp to reasonable bounds.
                                delta_s = int(abs(int(parsed.get('delta_ms'))) / 1000)
                                dur = min(max(60, delta_s + 10), 600)
                            self._set_exchange_backoff(user.exchange, reason='invalid_nonce', duration=dur or 60, user=user)
                        except Exception:
                            pass
                        # Drop this buffered entry to avoid repeated immediate retries
                        try:
                            buf.pop(user_name, None)
                        except Exception:
                            pass
                        return
                    else:
                        try:
                            if tb:
                                _human_log('PBData', f"[debounce] write failed for {kind} {user_name}: {e}", level='WARNING', meta={'traceback': tb})
                            else:
                                _human_log('PBData', f"[debounce] write failed for {kind} {user_name}: {e}", level='WARNING')
                        except Exception:
                            try:
                                _human_log('PBData', f"[debounce] write failed for {kind} {user_name}: {e}", level='WARNING')
                            except Exception:
                                pass

                # Decide whether to retry or give up based on elapsed time since first event
                now_ts = datetime.now().timestamp()
                elapsed_since_first = now_ts - first_ts
                if elapsed_since_first >= float(self._debounce_max_retry_seconds):
                    try:
                        _human_log('PBData', f"[debounce] giving up flush for {kind} {user_name} after {int(elapsed_since_first)}s (dropped)", level=self._debounce_giveup_log_level)
                    except Exception:
                        pass
                    try:
                        buf.pop(user_name, None)
                    except Exception:
                        pass
                    return

                # Backoff a bit and retry; include jitter
                await asyncio.sleep(min(1.0 + random.random() * 1.5, self._debounce_max_retry_seconds))
        except Exception:
            try:
                _human_log('PBData', f"[debounce] unexpected error in flusher for {kind} {user_name}", level='ERROR', meta={'traceback': traceback.format_exc()})
            except Exception:
                pass


    @asynccontextmanager
    async def _rest_slot(self, exchange: str, timeout: float = None):
        """Async context manager to acquire a per-exchange REST slot.

        Yields True if slot acquired, False if timed out. Caller is expected
        to skip or defer the REST call when False is yielded.
        """
        # Simpler, safer implementation: do not attempt to yield again when
        # an exception is propagated into this generator (the asynccontextmanager
        # machinery will re-throw the with-block exception into this generator).
        # If that happens we must not perform another yield() from here or
        # Python will raise "generator didn't stop after athrow()". Instead
        # allow the exception to propagate after performing cleanup in
        # `finally`.
        if timeout is None:
            timeout = getattr(self, '_rest_semaphore_acquire_timeout', 5.0)
        sem = self._rest_semaphores.get(exchange)
        if sem is None:
            limit = self._rest_semaphore_limits_by_exchange.get(exchange, getattr(self, '_default_rest_semaphore_limit', 3))
            sem = asyncio.Semaphore(limit)
            self._rest_semaphores[exchange] = sem
        acquired = False
        try:
            await asyncio.wait_for(sem.acquire(), timeout=timeout)
            acquired = True
            yield True
        except asyncio.TimeoutError:
            # Timed out acquiring slot: signal caller to skip the REST update
            yield False
        finally:
            if acquired:
                try:
                    sem.release()
                except Exception:
                    pass

    async def _acquire_rate_budget(self, exchange: str, operation: str, timeout: float = 60.0, weight_override: int = None) -> bool:
        """Acquire rate-limit budget tokens for an API operation.

        Returns True if budget was acquired (or exchange has no budget tracking).
        Returns False if timed out waiting for budget — caller should skip the call.
        If *weight_override* is given it is used instead of the static table lookup.
        """
        budget = self._rate_budgets.get(exchange)
        if not budget:
            return True  # no budget tracking for this exchange
        weight = weight_override if weight_override is not None else get_weight(exchange, operation)
        acquired = await budget.acquire(weight=weight, timeout=timeout, tag=operation)
        if not acquired:
            try:
                _human_log('PBData', f"[budget] Rate budget timeout for {exchange} op={operation} weight={weight}", level='WARNING')
            except Exception:
                pass
        return acquired

    def _set_exchange_backoff(self, exchange: str, reason: str = None, duration: int = None, user=None):
        try:
            now = datetime.now().timestamp()
            dur = duration if duration is not None else self._backoff_duration_seconds
            # long_history_poll only blocks the history poller, not the combined poller
            if reason == 'long_history_poll':
                backoff_dict = self._exchange_history_backoff_until
            else:
                backoff_dict = self._exchange_backoff_until
            # If already in backoff, increase duration (exponential-ish) to avoid
            # repeated tight retry loops. Clamp to a reasonable maximum.
            prev_until = backoff_dict.get(exchange, 0)
            prev_remaining = max(0, prev_until - now)
            # If previously backing off, double the remaining time up to max
            if prev_remaining > 0:
                new_dur = min(max(self._backoff_duration_seconds, dur, int(prev_remaining * 2)), 3600)
            else:
                new_dur = dur
            until = now + new_dur
            backoff_dict[exchange] = until
            # Pass username as `user` kwarg to human_log (human_log supports `user`)
            _human_log('PBData', f"[BACKOFF] Entering backoff for exchange {exchange} for {int(new_dur)}s (reason={reason})", level='WARNING', user=getattr(user, 'name', None))
        except Exception:
            pass

    def _parse_recv_window_from_exception(self, e) -> dict:
        """Attempt to extract req_timestamp and server_timestamp from ccxt/bybit InvalidNonce error messages.

        Returns dict with keys 'req_ts', 'server_ts', 'delta_ms' if found, otherwise empty dict.
        """
        try:
            text = str(e)
            # First try JSON body inside the message
            m = re.search(r"\{.*\}", text)
            if m:
                try:
                    body = json.loads(m.group(0))
                except Exception:
                    body = None
            else:
                body = None
            # Try regex extraction like req_timestamp[123],server_timestamp[456]
            rx = re.search(r"req_timestamp\[(\d+)\].*server_timestamp\[(\d+)\]", text, flags=re.IGNORECASE)
            if rx:
                req_ts = int(rx.group(1))
                server_ts = int(rx.group(2))
                delta = server_ts - req_ts
                return {'req_ts': req_ts, 'server_ts': server_ts, 'delta_ms': delta, 'body': body}
            # Fallback: look for numeric 'time' in JSON body
            if body and isinstance(body, dict):
                if 'time' in body and isinstance(body['time'], int):
                    server_ts = int(body['time'])
                    # try to find req_timestamp in message text
                    rx2 = re.search(r"req_timestamp\[(\d+)\]", text, flags=re.IGNORECASE)
                    if rx2:
                        req_ts = int(rx2.group(1))
                        delta = server_ts - req_ts
                        return {'req_ts': req_ts, 'server_ts': server_ts, 'delta_ms': delta, 'body': body}
            return {}
        except Exception:
            return {}

    def _is_exchange_in_backoff(self, exchange: str) -> bool:
        """Return True if we are currently backing off for this exchange."""
        try:
            if not exchange:
                return False
            until = self._exchange_backoff_until.get(exchange, 0)
            return datetime.now().timestamp() < until
        except Exception:
            return False

    def _is_exchange_in_history_backoff(self, exchange: str) -> bool:
        """Return True if the history poller should back off for this exchange (long_history_poll reason)."""
        try:
            if not exchange:
                return False
            until = self._exchange_history_backoff_until.get(exchange, 0)
            return datetime.now().timestamp() < until
        except Exception:
            return False

    # Small helper: treat some close messages as "normal" (don't immediately demote)
    def _is_normal_ws_close(self, msg: str) -> bool:
        if not msg:
            return False
        lower = msg.lower()
        # Common benign closure patterns (code 1000 = normal closure)
        if "code 1000" in lower or "closing code 1000" in lower or "normal closure" in lower:
            return True
        return False

    # Throttled logging for repeated network errors (reduce log spam)
    def _throttled_log_network(self, key, msg: str, throttle: float = 30.0):
        now = datetime.now().timestamp()
        last = self._last_network_error_log_ts.get(key, 0.0)
        if now - last >= throttle:
            try:
                _human_log('PBData', msg, level='WARNING')
            except Exception:
                pass
            self._last_network_error_log_ts[key] = now

    def _get_reconnect_sem(self, exchange: str) -> 'asyncio.Semaphore':
        """Return a per-exchange semaphore that limits concurrent WS reconnects to 2.

        When a server-side event drops all N private connections at once, all N
        asyncio tasks wake up and race to reconnect simultaneously.  Without
        throttling this overwhelms the event loop and can delay ping-pong frames
        for other exchanges, causing cascading failures.  Limiting each exchange
        to 2 concurrent reconnects spreads the storm over several seconds.
        """
        if exchange not in self._exchange_reconnect_sem:
            self._exchange_reconnect_sem[exchange] = asyncio.Semaphore(2)
        return self._exchange_reconnect_sem[exchange]

    async def _metrics_loop(self):
        """Periodic metrics logger: logs counts of shared/private clients and backoff states."""
        while True:
            try:
                from Exchange import Exchange
                metrics = Exchange.get_client_metrics()
                lines = []
                for exch, vals in metrics.items():
                    lines.append(f"{exch}: shared={vals.get('shared',0)} private={vals.get('private_count',0)}")
                # Also include exchanges in backoff
                backoffs = []
                now = datetime.now().timestamp()
                for exch, until in list(self._exchange_backoff_until.items()):
                    if until > now:
                        backoffs.append(f"{exch}:until={int(until-now)}s")
                for exch, until in list(self._exchange_history_backoff_until.items()):
                    if until > now:
                        backoffs.append(f"{exch}(hist):until={int(until-now)}s")
                if lines or backoffs:
                    _human_log('PBData', f"[METRICS] Clients: {', '.join(lines)}; Backoffs: {', '.join(backoffs) if backoffs else '(none)'}", level='INFO')
                # Write poller metrics JSON for the GUI panel
                try:
                    self._write_poller_metrics()
                except Exception:
                    pass
                # IO debugging removed: process/db IO summary logging disabled
                # Periodic cleanup of stale entries in unbounded data structures
                self._cleanup_stale_state()
            except Exception:
                try:
                    _human_log('PBData', f"[METRICS] failed to collect metrics", level='WARNING')
                except Exception:
                    pass
            await asyncio.sleep(self._metrics_interval)

    def _cleanup_stale_state(self):
        """Prune stale entries from unbounded dicts/sets to prevent slow memory growth.

        Called periodically from _metrics_loop (~every 60s).  Only removes entries
        whose user is no longer in the active fetch_users list, so active state is
        never touched.
        """
        try:
            active_users = set(self.fetch_users) if self.fetch_users else set()
            if not active_users:
                return  # no users → nothing to prune

            # --- _last_fetch_ts: keyed by (user_name, kind) ---
            stale_keys = [k for k in self._last_fetch_ts if k[0] not in active_users]
            for k in stale_keys:
                self._last_fetch_ts.pop(k, None)

            # --- _last_network_error_log_ts: keyed by (exchange, user_name) or similar ---
            stale_keys = []
            for k in self._last_network_error_log_ts:
                # key is (exchange, user_name) — check index 1
                try:
                    uname = k[1] if isinstance(k, tuple) and len(k) >= 2 else None
                    if uname and uname not in active_users:
                        stale_keys.append(k)
                except Exception:
                    pass
            for k in stale_keys:
                self._last_network_error_log_ts.pop(k, None)

            # --- _ws_success_counts: keyed by (exchange, user_name) ---
            stale_keys = [k for k in self._ws_success_counts
                          if isinstance(k, tuple) and len(k) >= 2 and k[1] not in active_users]
            for k in stale_keys:
                self._ws_success_counts.pop(k, None)

            # --- _ws_restarted_once: keyed by (exchange, user_name) ---
            stale_keys = {k for k in self._ws_restarted_once
                          if isinstance(k, tuple) and len(k) >= 2 and k[1] not in active_users}
            self._ws_restarted_once -= stale_keys

            # --- _exchange_network_error_users: exchange -> {user_name: ts} ---
            for exch in list(self._exchange_network_error_users):
                entry = self._exchange_network_error_users.get(exch, {})
                stale = [u for u in entry if u not in active_users]
                for u in stale:
                    entry.pop(u, None)
                if not entry:
                    self._exchange_network_error_users.pop(exch, None)

            if stale_keys:
                _human_log('PBData', f"[METRICS] cleaned stale state entries for inactive users", level='DEBUG')
        except Exception:
            pass

    # fetch_users

    # --- Price buffering and background writer ---
    async def buffer_price(self, user, symbol: str, timestamp: int, price_value: float):
        """Buffer the latest price for (user.name, symbol).

        This stores only the most recent tick per (user,symbol).
        """
        try:
            lock = getattr(self, '_price_buffer_lock', None)
            if lock is None:
                # lazily create if missing
                self._price_buffer_lock = asyncio.Lock()
                lock = self._price_buffer_lock
            if hasattr(lock, '__aenter__'):
                async with lock:
                    self._price_buffer[(user.name, symbol)] = (timestamp, price_value)
            else:
                # synchronous lock (fallback)
                with lock:
                    self._price_buffer[(user.name, symbol)] = (timestamp, price_value)
        except Exception:
            try:
                _human_log('PBData', f"[price_buffer] buffer_price error for {getattr(user,'name',user)} {symbol}", level='DEBUG')
            except Exception:
                pass

    async def _price_writer_loop(self):
        """Background task: periodically flush buffered prices to DB."""
        while True:
            try:
                await asyncio.sleep(self._price_flush_interval)
                try:
                    await self._flush_price_buffer()
                except Exception as e:
                    try:
                        _human_log('PBData', f"[price_writer] flush failed: {e}", level='WARNING')
                    except Exception:
                        pass
            except asyncio.CancelledError:
                # On cancellation, attempt a final flush then exit
                try:
                    await self._flush_price_buffer()
                except Exception:
                    pass
                raise
            except Exception:
                # swallow and continue
                try:
                    _human_log('PBData', "[price_writer] loop encountered error; continuing", level='DEBUG')
                except Exception:
                    pass

    async def _flush_price_buffer(self):
        """Snapshot buffer and flush via blocking batch DB call."""
        try:
            lock = getattr(self, '_price_buffer_lock', None)
            if lock is None:
                self._price_buffer_lock = asyncio.Lock()
                lock = self._price_buffer_lock
            if hasattr(lock, '__aenter__'):
                async with lock:
                    if not self._price_buffer:
                        return
                    snapshot = [(uname, sym, ts, pr) for (uname, sym), (ts, pr) in self._price_buffer.items()]
                    self._price_buffer = {}
            else:
                with lock:
                    if not self._price_buffer:
                        return
                    snapshot = [(uname, sym, ts, pr) for (uname, sym), (ts, pr) in self._price_buffer.items()]
                    self._price_buffer = {}
            # Perform blocking DB batch write in thread
            await asyncio.to_thread(self._write_prices_batch_sync, snapshot)
        except Exception as e:
            try:
                _human_log('PBData', f"[price_writer] _flush_price_buffer error: {e}", level='ERROR')
            except Exception:
                pass

    def _write_prices_batch_sync(self, rows: list):
        """Blocking function run in thread to write batch rows to DB.

        rows: list of (user_name, symbol, timestamp, price)
        """
        try:
            # Database.batch_upsert_prices expects rows as (user, symbol, timestamp, price)
            formatted = [(r[0], r[1], r[2], r[3]) for r in rows]
            try:
                # Use batch method if available
                if hasattr(self.db, 'batch_upsert_prices'):
                    self.db.batch_upsert_prices(formatted)
                else:
                    # Fallback: call upsert_price per row
                    for user, symbol, ts, pr in formatted:
                        try:
                            u = self.users.find_user(user)
                            if u:
                                self.db.upsert_price(u, symbol, ts, pr)
                        except Exception:
                            pass
            except Exception as e:
                try:
                    _human_log('PBData', f"[price_writer] DB batch write failed: {e}", level='ERROR')
                except Exception:
                    pass
        except Exception:
            pass

    @property
    def fetch_users(self):
        return self._fetch_users
    @fetch_users.setter
    def fetch_users(self, new_fetch_users):
        self._fetch_users = new_fetch_users
        self.save_fetch_users()

    @property
    def trades_users(self):
        return self._trades_users

    @trades_users.setter
    def trades_users(self, new_trades_users):
        self._trades_users = new_trades_users
        self.save_trades_users()

    def run(self):
        if not self.is_running():
            cmd = [sys.executable, '-u', str(PurePath(f'{PBGDIR}/PBData.py'))]
            if platform.system() == "Windows":
                creationflags = subprocess.DETACHED_PROCESS
                creationflags |= subprocess.CREATE_NO_WINDOW
                subprocess.Popen(cmd, stdout=None, stderr=None, cwd=PBGDIR, text=True, creationflags=creationflags)
            else:
                subprocess.Popen(cmd, stdout=None, stderr=None, cwd=PBGDIR, text=True, start_new_session=True)
            count = 0
            while True:
                if count > 5:
                    _human_log('PBData', 'Error: Can not start PBData', level='ERROR')
                    break
                sleep(1)
                if self.is_running():
                    break
                count += 1
    

    def stop(self):
        if self.is_running():
            _human_log('PBData', 'Stop: PBData', level='INFO')
            try:
                psutil.Process(self.my_pid).kill()
            except psutil.NoSuchProcess:
                pass

    def restart(self):
        if self.is_running():
            self.stop()
            self.run()

    def is_running(self):
        self.load_pid()
        try:
            if self.my_pid and psutil.pid_exists(self.my_pid) and any(sub.lower().endswith("pbdata.py") for sub in psutil.Process(self.my_pid).cmdline()):
                return True
        except psutil.NoSuchProcess:
            pass
        return False

    def load_pid(self):
        if self.pidfile.exists():
            with open(self.pidfile) as f:
                pid = f.read()
                self.my_pid = int(pid) if pid.isnumeric() else None

    def save_pid(self):
        self.my_pid = os.getpid()
        tmp_path = self.pidfile.with_suffix(self.pidfile.suffix + '.tmp')
        with tmp_path.open('w', encoding='utf-8') as f:
            f.write(str(self.my_pid))
        tmp_path.replace(self.pidfile)
    
    def load_fetch_users(self):
        pb_config = configparser.ConfigParser()
        try:
            pb_config.read('pbgui.ini')
        except Exception as e:
            _human_log('PBData', f"Warning: failed reading pbgui.ini ({e}); keeping previous fetch_users: {self._fetch_users}", level='WARNING')
            return
        if pb_config.has_option("pbdata", "fetch_users"):
            users = ast.literal_eval(pb_config.get("pbdata", "fetch_users"))
            valid = set(self.users.list())
            self._fetch_users = [u for u in users if u in valid]
        else:
            self._fetch_users = []  # Default to empty list if not set

    def load_trades_users(self):
        """Load the user list which is allowed to download/store executions (trades).

        Default: empty list (no trades download) unless explicitly configured.
        """
        pb_config = configparser.ConfigParser()
        try:
            pb_config.read('pbgui.ini')
        except Exception as e:
            _human_log('PBData', f"Warning: failed reading pbgui.ini ({e}); keeping previous trades_users: {self._trades_users}", level='WARNING')
            return

        if pb_config.has_option("pbdata", "trades_users"):
            try:
                raw = pb_config.get("pbdata", "trades_users")
                sval = str(raw).strip() if raw is not None else ''
                if sval == '':
                    users = []
                else:
                    users = ast.literal_eval(sval)
                if not isinstance(users, list):
                    users = []
            except Exception:
                users = []
            valid = set(self.users.list())
            self._trades_users = [u for u in users if u in valid]
        else:
            # Default to empty list unless explicitly configured.
            self._trades_users = []
    
    def save_fetch_users(self):
        pb_config = configparser.ConfigParser()
        pb_config.read('pbgui.ini')
        if not pb_config.has_section("pbdata"):
            pb_config.add_section("pbdata")
        pb_config.set("pbdata", "fetch_users", f'{self.fetch_users}')
        fd, tmp = tempfile.mkstemp(dir='.', suffix='.ini.tmp')
        try:
            with os.fdopen(fd, 'w') as f:
                pb_config.write(f)
            os.replace(tmp, 'pbgui.ini')
        except Exception:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            raise

    def save_trades_users(self):
        pb_config = configparser.ConfigParser()
        pb_config.read('pbgui.ini')
        if not pb_config.has_section("pbdata"):
            pb_config.add_section("pbdata")
        pb_config.set("pbdata", "trades_users", f'{self.trades_users}')
        fd, tmp = tempfile.mkstemp(dir='.', suffix='.ini.tmp')
        try:
            with os.fdopen(fd, 'w') as f:
                pb_config.write(f)
            os.replace(tmp, 'pbgui.ini')
        except Exception:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            raise

    async def _order_poll_loop(self, user, interval_seconds: int = 5):
        _human_log('PBData', f"[poll] Starting orders poller for {user.name}", level='INFO', user=user)
        while True:
            try:
                await asyncio.to_thread(self.db.update_orders, user)
                try:
                    self._last_fetch_ts[(user.name, 'orders')] = datetime.now().timestamp()
                except Exception:
                    pass
                try:
                    self._write_fetch_summary()
                except Exception:
                    pass
            except Exception as e:
                _human_log('PBData', f"[poll] Orders poll failed for {user.name}: {e}", level='ERROR', user=user)
            await asyncio.sleep(interval_seconds)

    async def _position_poll_loop(self, user, interval_seconds: int = 5):
        _human_log('PBData', f"[poll] Starting positions poller for {user.name}", level='INFO', user=user)
        while True:
            try:
                await asyncio.to_thread(self.db.update_positions, user)
                try:
                    self._last_fetch_ts[(user.name, 'positions')] = datetime.now().timestamp()
                except Exception:
                    pass
                try:
                    self._write_fetch_summary()
                except Exception:
                    pass
            except Exception as e:
                _human_log('PBData', f"[poll] Positions poll failed for {user.name}: {e}", level='ERROR', user=user)
            await asyncio.sleep(interval_seconds)

    async def _shared_poll_serial(
        self,
        kind: str,
        interval_seconds: int,
        per_exchange: bool = True,
        start_immediately: bool = False,
        start_delay_seconds: float = 0.0,
        initial_jitter_seconds: float = 0.0,
        exchange_filter: str = None,
    ):
        """Generic serial poller for 'positions', 'history', 'orders', 'balances', or 'executions'.

        By default, the poller sleeps for `interval_seconds` before the first run.
        If `start_immediately=True`, it runs one cycle early, optionally after a
        startup delay. With `start_delay_seconds` and `initial_jitter_seconds`,
        the first run is delayed by a random amount in the range
        `[start_delay_seconds - initial_jitter_seconds, start_delay_seconds + initial_jitter_seconds]`.
        After the first run, the poller follows the regular interval cadence.

        If `exchange_filter` is set, only users of that specific exchange are polled.
        This enables per-exchange parallel tasks — each exchange gets its own
        interval cycle independent of other slow exchanges.
        """
        backoff = 0
        max_backoff = 600
        base_interval = max(10, interval_seconds)
        exch_label = f" exchange={exchange_filter}" if exchange_filter else ""
        _human_log('PBData', f"[poll] Starting shared serial poller kind={kind} interval={base_interval}s{exch_label}", level='INFO')
        while True:
            if start_immediately:
                # Only the first iteration runs early.
                start_immediately = False
                try:
                    center = float(start_delay_seconds or 0.0)
                    jitter = float(initial_jitter_seconds or 0.0)
                    low = max(0.0, center - jitter)
                    high = max(low, center + jitter)
                    if high > 0:
                        await asyncio.sleep(random.uniform(low, high))
                except Exception:
                    pass
            else:
                delay = base_interval + backoff
                await asyncio.sleep(delay)
            users = [u for u in self.users if u.name in self.fetch_users]
            if exchange_filter:
                users = [u for u in users if u.exchange == exchange_filter]
            if kind == 'executions':
                try:
                    self.load_trades_users()
                except Exception:
                    pass
                try:
                    allowed = set(self.trades_users or [])
                except Exception:
                    allowed = set()
                users = [u for u in users if u.name in allowed]
            if not users:
                continue
            if per_exchange:
                groups = {}
                for u in users:
                    groups.setdefault(u.exchange, []).append(u)
                batches = groups.items()
            else:
                batches = [(None, users)]
            had_rate_limit = False
            _serial_cycle_start = datetime.now().timestamp()
            _serial_users_polled = 0
            for exch, batch_users in batches:
                # Skip this exchange while in backoff (not for budget-tracked exchanges).
                # For kind='history', also check history-specific backoff (long_history_poll).
                _has_budget_ex = exch and exch in self._rate_budgets
                in_backoff = exch and not _has_budget_ex and self._is_exchange_in_backoff(exch)
                if not in_backoff and not _has_budget_ex and kind == 'history':
                    in_backoff = exch and self._is_exchange_in_history_backoff(exch)
                if in_backoff:
                    _human_log('PBData', f"[poll] Skipping shared {kind} poll for {exch} due to backoff", level='INFO')
                    continue
                # Acquire per-exchange budget lock so pollers don't starve each other.
                _bpl2 = self._budget_poller_locks.get(exch)
                if _bpl2:
                    await _bpl2.acquire()
                try:
                    for user in batch_users:
                        try:
                            if kind == 'positions':
                                # Gate shared REST calls using per-exchange REST slot
                                exchange_for_slot = exch or user.exchange
                                try:
                                    async with self._rest_slot(exchange_for_slot) as got:
                                        if got:
                                            await asyncio.to_thread(self.db.update_positions, user)
                                            try:
                                                self._last_fetch_ts[(user.name, 'positions')] = datetime.now().timestamp()
                                            except Exception:
                                                pass
                                        else:
                                            # Could not acquire REST slot; skip this update
                                            try:
                                                _human_log('PBData', f"[poll] Skipped positions update for {user.name}; rest slot busy for {exchange_for_slot}", level=self._rest_slot_busy_log_level)
                                            except Exception:
                                                pass
                                            had_rate_limit = True
                                except Exception as e:
                                    # Handle InvalidNonce specially to avoid generator issues and retry storms
                                    try:
                                        msg = str(e).lower()
                                        if 'invalid request' in msg and 'recv_window' in msg:
                                            tb = traceback.format_exc()
                                            try:
                                                parsed = self._parse_recv_window_from_exception(e)
                                            except Exception:
                                                parsed = {}
                                            try:
                                                meta = {'traceback': tb} if tb else {}
                                                if parsed:
                                                    meta.update(parsed)
                                                if not meta:
                                                    meta = None
                                                _human_log('PBData', f"[poll] InvalidNonce / recv_window error for {user.name} ({user.exchange}): {e}", level='WARNING', meta=meta)
                                            except Exception:
                                                pass
                                            try:
                                                dur = None
                                                if parsed and parsed.get('delta_ms') is not None:
                                                    delta_s = int(abs(int(parsed.get('delta_ms'))) / 1000)
                                                    dur = min(max(60, delta_s + 10), 600)
                                                self._set_exchange_backoff(user.exchange, reason='invalid_nonce', duration=dur or 60, user=user)
                                            except Exception:
                                                pass
                                            had_rate_limit = True
                                            continue
                                    except Exception:
                                        pass
                            elif kind == 'orders':
                                exchange_for_slot = exch or user.exchange
                                try:
                                    async with self._rest_slot(exchange_for_slot) as got:
                                        if got:
                                            await asyncio.to_thread(self.db.update_orders, user)
                                            try:
                                                self._last_fetch_ts[(user.name, 'orders')] = datetime.now().timestamp()
                                            except Exception:
                                                pass
                                        else:
                                            try:
                                                _human_log('PBData', f"[poll] Skipped orders update for {user.name}; rest slot busy for {exchange_for_slot}", level=self._rest_slot_busy_log_level)
                                            except Exception:
                                                pass
                                            had_rate_limit = True
                                except Exception as e:
                                    try:
                                        msg = str(e).lower()
                                        if 'invalid request' in msg and 'recv_window' in msg:
                                            tb = traceback.format_exc()
                                            try:
                                                parsed = self._parse_recv_window_from_exception(e)
                                            except Exception:
                                                parsed = {}
                                            try:
                                                meta = {'traceback': tb} if tb else {}
                                                if parsed:
                                                    meta.update(parsed)
                                                if not meta:
                                                    meta = None
                                                _human_log('PBData', f"[poll] InvalidNonce / recv_window error for {user.name} ({user.exchange}): {e}", level='WARNING', meta=meta)
                                            except Exception:
                                                pass
                                            try:
                                                dur = None
                                                if parsed and parsed.get('delta_ms') is not None:
                                                    delta_s = int(abs(int(parsed.get('delta_ms'))) / 1000)
                                                    dur = min(max(60, delta_s + 10), 600)
                                                self._set_exchange_backoff(user.exchange, reason='invalid_nonce', duration=dur or 60, user=user)
                                            except Exception:
                                                pass
                                            had_rate_limit = True
                                            continue
                                    except Exception:
                                        pass
                            elif kind == 'history':
                                # Instrument shared history polling for debugging/tracing
                                key = (user.name, user.exchange)
                                start_ts = datetime.now().timestamp()
                                _human_log('PBData', f"[poll] Shared history poll START for {user.name} ({user.exchange})", level='INFO', user=user)
                                # Time the history update; if it is very long, mark exchange backoff
                                exchange_for_slot = exch or user.exchange
                                try:
                                    async with self._rest_slot(exchange_for_slot) as got:
                                        if not got:
                                            try:
                                                _human_log('PBData', f"[poll] Skipped history update for {user.name}; rest slot busy for {exchange_for_slot}", level=self._rest_slot_busy_log_level)
                                            except Exception:
                                                pass
                                            had_rate_limit = True
                                        else:
                                            # Acquire rate budget before history fetch
                                            _budget_ok = await self._acquire_rate_budget(exchange_for_slot, 'fetch_history')
                                            if not _budget_ok:
                                                pass  # budget timeout — skip but don't set backoff
                                            try:
                                                if _budget_ok:
                                                    await asyncio.to_thread(self.db.update_history, user)
                                                    asyncio.create_task(_notify_api_income(getattr(user, 'name', '')))
                                                    # Mark balance + positions + orders as stale so the next combined
                                                    # cycle refreshes them (fills/funding change both).
                                                    try:
                                                        self._balance_stale_users.add(user.name)
                                                        self._positions_stale_users.add(user.name)
                                                        self._orders_stale_users.add(user.name)
                                                    except Exception:
                                                        pass
                                            except Exception as e:
                                                # Some runtime errors seen under heavy load look like
                                                # "generator didn't stop after athrow()". Treat these
                                                # as transient and escalate to backoff while logging.
                                                msg = str(e)
                                                lower = msg.lower()
                                                if 'invalid request' in lower and 'recv_window' in lower:
                                                    tb = traceback.format_exc()
                                                    try:
                                                        parsed = self._parse_recv_window_from_exception(e)
                                                    except Exception:
                                                        parsed = {}
                                                    try:
                                                        meta = {'traceback': tb} if tb else {}
                                                        if parsed:
                                                            meta.update(parsed)
                                                        if not meta:
                                                            meta = None
                                                        _human_log('PBData', f"[poll] InvalidNonce / recv_window error for {user.name} ({user.exchange}): {e}", level='WARNING', meta=meta)
                                                    except Exception:
                                                        pass
                                                    try:
                                                        dur = None
                                                        if parsed and parsed.get('delta_ms') is not None:
                                                            delta_s = int(abs(int(parsed.get('delta_ms'))) / 1000)
                                                            dur = min(max(60, delta_s + 10), 600)
                                                        self._set_exchange_backoff(user.exchange, reason='invalid_nonce', duration=dur or 60, user=user)
                                                    except Exception:
                                                        pass
                                                    had_rate_limit = True
                                                elif 'generator didn\'t stop after athrow' in msg or 'athrow' in lower:
                                                    try:
                                                        _human_log('PBData', f"[poll] history update failed for {user.name}: {msg}", level='ERROR')
                                                    except Exception:
                                                        pass
                                                    had_rate_limit = True
                                                else:
                                                    # Re-raise unexpected exceptions to be handled by outer handler
                                                    raise
                                except Exception as e:
                                    raise
                                dur_ms = int((datetime.now().timestamp() - start_ts) * 1000)
                                if dur_ms / 1000.0 > self._long_poll_threshold_seconds:
                                    # mark exchange as overloaded and backoff longer (include user for logging)
                                    try:
                                        self._set_exchange_backoff(user.exchange, reason='long_history_poll', duration=self._backoff_duration_seconds * 2, user=user)
                                    except Exception:
                                        pass
                                # record last successful REST history poll time per user/exchange
                                try:
                                    self._history_rest_last[key] = start_ts
                                    try:
                                        self._last_fetch_ts[(user.name, 'history')] = start_ts
                                    except Exception:
                                        pass
                                except Exception:
                                    pass
                                try:
                                    self._write_fetch_summary()
                                except Exception:
                                    pass
                                _human_log('PBData', f"[poll] Shared history poll DONE for {user.name} ({user.exchange}) dur={dur_ms}ms", level='INFO', user=user)
                            elif kind == 'executions':
                                # Re-check allowed executions users before each fetch so that
                                # removing a user in the UI takes effect immediately even
                                # mid-cycle.
                                try:
                                    self.load_trades_users()
                                    if user.name not in (self.trades_users or []):
                                        continue
                                except Exception:
                                    # Fail-safe: if we cannot determine the allow-list, skip.
                                    continue
                                start_ts = datetime.now().timestamp()
                                exchange_for_slot = exch or user.exchange
                                did_run = False
                                fetched = None
                                inserted = None
                                try:
                                    async with self._rest_slot(exchange_for_slot) as got:
                                        if not got:
                                            try:
                                                _human_log('PBData', f"[poll] Skipped executions update for {user.name}; rest slot busy for {exchange_for_slot}", level=self._rest_slot_busy_log_level)
                                            except Exception:
                                                pass
                                            had_rate_limit = True
                                        else:
                                            _exec_budget_ok = await self._acquire_rate_budget(exchange_for_slot, 'fetch_executions')
                                            if _exec_budget_ok:
                                                did_run = True
                                                res = await asyncio.to_thread(self.db.update_executions, user)
                                                try:
                                                    if isinstance(res, dict):
                                                        fetched = res.get('fetched')
                                                        inserted = res.get('inserted')
                                                        if inserted and inserted > 0:
                                                            self._balance_stale_users.add(user.name)
                                                            self._positions_stale_users.add(user.name)
                                                            self._orders_stale_users.add(user.name)
                                                except Exception:
                                                    pass
                                                try:
                                                    self._last_fetch_ts[(user.name, 'executions')] = start_ts
                                                except Exception:
                                                    pass
                                    try:
                                        self._write_fetch_summary()
                                    except Exception:
                                        pass
                                except Exception:
                                    # Let outer handler log traceback + handle rate-limit keywords.
                                    raise
                                if did_run:
                                    try:
                                        dur_ms = int((datetime.now().timestamp() - start_ts) * 1000)
                                        _human_log(
                                            'PBData',
                                            f"[poll] Shared executions poll DONE for {user.name} ({user.exchange}) dur={dur_ms}ms fetched={fetched} inserted={inserted}",
                                            level='INFO',
                                            user=user,
                                        )
                                    except Exception:
                                        pass
                            elif kind == 'balances':
                                exchange_for_slot = exch or user.exchange
                                try:
                                    async with self._rest_slot(exchange_for_slot) as got:
                                        if got:
                                            await asyncio.to_thread(self.db.update_balances, user)
                                            asyncio.create_task(_notify_api_balance())
                                            try:
                                                self._last_fetch_ts[(user.name, 'balances')] = datetime.now().timestamp()
                                            except Exception:
                                                pass
                                            try:
                                                self._write_fetch_summary()
                                            except Exception:
                                                pass
                                        else:
                                            try:
                                                _human_log('PBData', f"[poll] Skipped balances update for {user.name}; rest slot busy for {exchange_for_slot}", level=self._rest_slot_busy_log_level)
                                            except Exception:
                                                pass
                                            had_rate_limit = True
                                except Exception as e:
                                    try:
                                        msg = str(e).lower()
                                        if 'invalid request' in msg and 'recv_window' in msg:
                                                tb = traceback.format_exc()
                                                try:
                                                    parsed = self._parse_recv_window_from_exception(e)
                                                except Exception:
                                                    parsed = {}
                                                try:
                                                    meta = {'traceback': tb} if tb else {}
                                                    if parsed:
                                                        meta.update(parsed)
                                                    if not meta:
                                                        meta = None
                                                    _human_log('PBData', f"[poll] InvalidNonce / recv_window error for {user.name} ({user.exchange}): {e}", level='WARNING', meta=meta)
                                                except Exception:
                                                    pass
                                                try:
                                                    dur = None
                                                    if parsed and parsed.get('delta_ms') is not None:
                                                        delta_s = int(abs(int(parsed.get('delta_ms'))) / 1000)
                                                        dur = min(max(60, delta_s + 10), 600)
                                                    self._set_exchange_backoff(user.exchange, reason='invalid_nonce', duration=dur or 60, user=user)
                                                except Exception:
                                                    pass
                                                had_rate_limit = True
                                                continue
                                    except Exception:
                                        pass
                            # (duplicate branches removed)
                        except Exception as e:
                            msg = str(e)
                            tb = traceback.format_exc()
                            try:
                                _human_log('PBData', f"[poll] Shared {kind} poll failed for {user.name}: {e}", level='ERROR', meta={'traceback': tb})
                            except Exception:
                                pass
                            _err_exch2 = getattr(user, 'exchange', None) or exch or 'unknown'
                            try:
                                self._poller_metrics[_err_exch2]['errors'] += 1
                            except Exception:
                                pass
                            lower = msg.lower()
                            if '429' in lower or 'too many requests' in lower or 'rate limit' in lower:
                                had_rate_limit = True
                                try:
                                    self._poller_metrics[_err_exch2]['rate_limit_429'] += 1
                                except Exception:
                                    pass
                        else:
                            _serial_users_polled += 1
                        # Small stagger between per-user REST requests to avoid bursts
                        # Skip for exchanges with budget tracking (budget handles pacing)
                        _has_budget2 = (exch or getattr(user, 'exchange', None) or '') in self._rate_budgets
                        if not _has_budget2:
                            try:
                                pause_val = self._shared_rest_pause_by_exchange.get(exch, self._shared_rest_user_pause)
                                if pause_val and pause_val > 0:
                                    jitter = random.uniform(0, pause_val * 0.2)
                                    await asyncio.sleep(pause_val + jitter)
                            except Exception:
                                pass
                finally:
                    if _bpl2:
                        _bpl2.release()
            # Record per-exchange history/serial-poller metrics
            if kind == 'history':
                _serial_cycle_end = datetime.now().timestamp()
                _serial_exch = exchange_filter or 'all'
                try:
                    m = self._poller_metrics[_serial_exch]
                    m['history_last_ts'] = _serial_cycle_end
                    m['history_cycle_ms'] = int((_serial_cycle_end - _serial_cycle_start) * 1000)
                    m['history_users'] = _serial_users_polled
                except Exception:
                    pass
            if had_rate_limit:
                # Skip backoff escalation for exchanges with budget tracking
                if exchange_filter and exchange_filter in self._rate_budgets:
                    pass
                else:
                    backoff = min(max_backoff, backoff + 30)
            else:
                if backoff:
                    _human_log('PBData', f"[poll] Shared {kind} poll recovered; resetting backoff", level='INFO')
                backoff = 0

    async def _shared_combined_poll_serial(self, interval_seconds: int = 60, per_exchange: bool = True, exchange_filter: str = None):
        """Shared poller that sequentially runs balances, positions and orders per user.

        This ensures only one REST "connection" worth of work is performed at a time
        for these combined kinds (per exchange), reducing parallel HTTP load.
        History polling remains separate because it can be long-running.
        If `exchange_filter` is set, only users of that specific exchange are polled.
        """
        backoff = 0
        max_backoff = 600
        base_interval = max(10, interval_seconds)
        exch_label = f" exchange={exchange_filter}" if exchange_filter else ""
        _human_log('PBData', f"[poll] Starting shared COMBINED poller interval={base_interval}s{exch_label}", level='INFO')
        # Persistent Exchange instances per user — keeps CCXT's markets cache alive
        # across poll cycles so load_markets() is only called once (not once per cycle).
        _user_exch: dict = {}
        while True:
            delay = base_interval + backoff
            await asyncio.sleep(delay)
            users = [u for u in self.users if u.name in self.fetch_users]
            if exchange_filter:
                users = [u for u in users if u.exchange == exchange_filter]
            if not users:
                continue
            if per_exchange:
                groups = {}
                for u in users:
                    groups.setdefault(u.exchange, []).append(u)
                batches = groups.items()
            else:
                batches = [(None, users)]
            had_rate_limit = False
            _combined_cycle_start = datetime.now().timestamp()
            _combined_users_polled = 0
            # Remove persistent instances for users no longer active
            _active_names = {u.name for u in users}
            for _stale_name in [n for n in list(_user_exch) if n not in _active_names]:
                try:
                    _user_exch.pop(_stale_name).close()
                except Exception:
                    pass
            for exch, batch_users in batches:
                if exch and exch not in self._rate_budgets and self._is_exchange_in_backoff(exch):
                    _human_log('PBData', f"[poll] Skipping shared COMBINED poll for {exch} due to backoff")
                    continue
                exchange_rate_limited = False
                # Acquire per-exchange budget lock so combined/history don't starve each other.
                _bpl = self._budget_poller_locks.get(exch)
                if _bpl:
                    await _bpl.acquire()
                try:
                    for user in batch_users:
                        if exchange_rate_limited:
                            break
                        try:
                            exchange_for_slot = exch or getattr(user, 'exchange', None) or 'unknown'
                            async with self._rest_slot(exchange_for_slot) as got:
                                if not got:
                                    _human_log('PBData', f"[poll] Shared COMBINED poll skipped {user.name}: REST slot timed out", level='DEBUG', user=user.name)
                                    try:
                                        self._poller_metrics[exchange_for_slot]['rest_slot_timeouts'] += 1
                                    except Exception:
                                        pass
                                    continue
                                # ── Balance: skip if timer not due AND user not stale ──
                                _balance_due = True
                                try:
                                    _last_bal = self._last_fetch_ts.get((user.name, 'balances'))
                                    if _last_bal is not None:
                                        _bal_age = datetime.now().timestamp() - _last_bal
                                        _bal_interval = getattr(self, '_poll_interval_balance_seconds', 300)
                                        _is_stale = user.name in getattr(self, '_balance_stale_users', set())
                                        if _bal_age < _bal_interval and not _is_stale:
                                            _balance_due = False
                                except Exception:
                                    pass
                                # ── Positions: skip if timer not due AND user not stale ──
                                _positions_due = True
                                try:
                                    _last_pos = self._last_fetch_ts.get((user.name, 'positions'))
                                    if _last_pos is not None:
                                        _pos_age = datetime.now().timestamp() - _last_pos
                                        _pos_interval = getattr(self, '_poll_interval_positions_seconds', 300)
                                        _is_pos_stale = user.name in getattr(self, '_positions_stale_users', set())
                                        if _pos_age < _pos_interval and not _is_pos_stale:
                                            _positions_due = False
                                except Exception:
                                    pass
                                # ── Orders: skip if timer not due AND user not stale ──
                                _orders_due = True
                                try:
                                    _last_ord = self._last_fetch_ts.get((user.name, 'orders'))
                                    if _last_ord is not None:
                                        _ord_age = datetime.now().timestamp() - _last_ord
                                        _ord_interval = getattr(self, '_poll_interval_orders_seconds', 60)
                                        _is_ord_stale = user.name in getattr(self, '_orders_stale_users', set())
                                        if _ord_age < _ord_interval and not _is_ord_stale:
                                            _orders_due = False
                                except Exception:
                                    pass
                                # Nothing to do for this user
                                if not _balance_due and not _positions_due and not _orders_due:
                                    _combined_users_polled += 1
                                    try:
                                        self._write_fetch_summary()
                                    except Exception:
                                        pass
                                    continue
                                # ── Persistent shared Exchange ──
                                # Reuse the same Exchange instance across poll cycles so that
                                # CCXT's markets cache (self.instance.markets) is not discarded
                                # between cycles.  On Hyperliquid, fetch_positions() calls
                                # load_markets() internally; with a fresh instance that means
                                # 3-4 extra REST calls per user per cycle → 429s.  With a
                                # persistent instance the cache hit skips those calls entirely.
                                _n_due = int(_balance_due) + int(_positions_due) + int(_orders_due)
                                if user.name not in _user_exch:
                                    _user_exch[user.name] = _Exchange(user.exchange, user)
                                _shared_exch = _user_exch[user.name]
                                try:
                                    if _balance_due:
                                        if not await self._acquire_rate_budget(exchange_for_slot, 'fetch_balance'):
                                            continue
                                        await asyncio.to_thread(self.db.update_balances, user, _shared_exch)
                                        asyncio.create_task(_notify_api_balance())
                                        try:
                                            self._last_fetch_ts[(user.name, 'balances')] = datetime.now().timestamp()
                                        except Exception:
                                            pass
                                        try:
                                            self._balance_stale_users.discard(user.name)
                                        except Exception:
                                            pass
                                    if _positions_due:
                                        if not await self._acquire_rate_budget(exchange_for_slot, 'fetch_positions'):
                                            continue
                                        await asyncio.to_thread(self.db.update_positions, user, _shared_exch)
                                        try:
                                            self._last_fetch_ts[(user.name, 'positions')] = datetime.now().timestamp()
                                        except Exception:
                                            pass
                                        try:
                                            self._positions_stale_users.discard(user.name)
                                        except Exception:
                                            pass
                                        # Positions changed → orders probably changed too
                                        try:
                                            self._orders_stale_users.add(user.name)
                                        except Exception:
                                            pass
                                    if not _orders_due:
                                        _combined_users_polled += 1
                                        try:
                                            self._write_fetch_summary()
                                        except Exception:
                                            pass
                                        continue
                                    # Dynamic weight: fetch_open_orders is called once PER position
                                    _n_pos = 1
                                    try:
                                        _pos_rows = self.db.fetch_positions(user)
                                        if _pos_rows:
                                            _n_pos = max(1, len(_pos_rows))
                                    except Exception:
                                        pass
                                    _orders_weight = get_weight(exchange_for_slot, 'fetch_open_orders') * _n_pos
                                    if not await self._acquire_rate_budget(exchange_for_slot, 'fetch_open_orders', weight_override=_orders_weight):
                                        continue
                                    await asyncio.to_thread(self.db.update_orders, user, _shared_exch)
                                    try:
                                        self._last_fetch_ts[(user.name, 'orders')] = datetime.now().timestamp()
                                    except Exception:
                                        pass
                                    try:
                                        self._orders_stale_users.discard(user.name)
                                    except Exception:
                                        pass
                                    _combined_users_polled += 1
                                    try:
                                        self._write_fetch_summary()
                                    except Exception:
                                        pass
                                finally:
                                    # Do NOT close _shared_exch here — we keep it alive
                                    # so CCXT's markets cache survives to the next cycle.
                                    pass
                        except Exception as e:
                            msg = str(e)
                            tb = traceback.format_exc()
                            try:
                                _human_log('PBData', f"[poll] Shared COMBINED poll failed for {user.name}: {e}", level='ERROR', meta={'traceback': tb})
                            except Exception:
                                pass
                            _err_exch = getattr(user, 'exchange', None) or exch or 'unknown'
                            try:
                                self._poller_metrics[_err_exch]['errors'] += 1
                            except Exception:
                                pass
                            lower = msg.lower()
                            if '429' in lower or 'too many requests' in lower or 'rate limit' in lower:
                                had_rate_limit = True
                                exchange_rate_limited = True
                                try:
                                    self._poller_metrics[_err_exch]['rate_limit_429'] += 1
                                except Exception:
                                    pass
                                # Skip backoff for exchanges with budget tracking (budget handles pacing)
                                if _err_exch not in self._rate_budgets:
                                    try:
                                        self._set_exchange_backoff(user.exchange, reason='rate_limit_429', duration=45, user=user)
                                    except Exception:
                                        pass
                        if not exchange_rate_limited:
                            # Skip fixed pause for exchanges with budget tracking (budget handles pacing)
                            _has_budget = (exch or getattr(user, 'exchange', None) or '') in self._rate_budgets
                            if not _has_budget:
                                try:
                                    pause_val = self._shared_rest_pause_by_exchange.get(exch, self._shared_rest_user_pause)
                                    if pause_val and pause_val > 0:
                                        jitter = random.uniform(0, pause_val * 0.2)
                                        await asyncio.sleep(pause_val + jitter)
                                except Exception:
                                    pass
                finally:
                    if _bpl:
                        _bpl.release()
            # Record per-exchange combined-poller metrics
            _combined_cycle_end = datetime.now().timestamp()
            _combined_exch = exchange_filter or 'all'
            try:
                m = self._poller_metrics[_combined_exch]
                m['combined_last_ts'] = _combined_cycle_end
                m['combined_cycle_ms'] = int((_combined_cycle_end - _combined_cycle_start) * 1000)
                m['combined_users'] = _combined_users_polled
            except Exception:
                pass
            if had_rate_limit:
                # Skip backoff escalation for exchanges with budget tracking
                if exchange_filter and exchange_filter in self._rate_budgets:
                    pass
                else:
                    backoff = min(max_backoff, backoff + 30)
            else:
                if backoff:
                    _human_log('PBData', f"[poll] Shared COMBINED poll recovered; resetting backoff")
                backoff = 0

    async def _balance_poll_loop(self, user, interval_seconds: int = 30):
        _human_log('PBData', f"[poll] Starting balance poller for {user.name}", level='INFO', user=user)
        while True:
            try:
                await asyncio.to_thread(self.db.update_balances, user)
                asyncio.create_task(_notify_api_balance())
                try:
                    self._last_fetch_ts[(user.name, 'balances')] = datetime.now().timestamp()
                except Exception:
                    pass
            except Exception as e:
                _human_log('PBData', f"[poll] Balance poll failed for {user.name}: {e}", level='ERROR', user=user)
            await asyncio.sleep(interval_seconds)

    def _to_ccxt_symbol(self, symbol: str) -> str:
        if symbol.endswith('USDT'):
            return f"{symbol[:-4]}/USDT:USDT"
        if symbol.endswith('USDC'):
            return f"{symbol[:-4]}/USDC:USDC"
        return symbol

    async def _price_exchange_ws_loop(self, exchange: str):
        from Exchange import Exchange
        last_price_write_ts = {}
        throttle_interval_sec = 1.0

        while True:
            exch = Exchange(exchange, None)
            ex = await Exchange.get_shared_ws_client(exchange)
            if not ex:
                _human_log('PBData', f"[ws] ccxtpro unavailable (price) for exchange {exchange}; retrying in 10s", level='WARNING')
                await asyncio.sleep(10)
                continue
            _human_log('PBData', f"[ws] Starting price watcher for exchange {exchange}", level='INFO')
            started_logged = False
            last_heartbeat_ts = 0
            self._price_ticks_count[exchange] = 0
            try:
                if hasattr(ex, 'load_markets'):
                    lm = ex.load_markets()
                    import asyncio as _asyncio  # local alias to check coroutine
                    if _asyncio.iscoroutine(lm):
                        await lm
                    else:
                        # Run sync load_markets without blocking loop
                        await asyncio.to_thread(ex.load_markets)
            except Exception as e:
                _human_log('PBData', f"[ws] load_markets failed for exchange {exchange}: {e}", level='WARNING')
            try:
                subscribe_backoff = 0
                while True:
                    cfg = self._price_exchange_config.get(exchange, {})
                    symbols = list(cfg.get('symbols', set()))
                    mapping = cfg.get('mapping', {})
                    # Build a full list of unique users for this mapping so
                    # we can log and react deterministically to subscribe
                    # rejection errors (some exchanges impose a hard limit).
                    unique_users_all = []
                    seen_all = set()
                    for ccxt_sym, lst in mapping.items():
                        for user_name, _ in lst:
                            if user_name not in seen_all:
                                seen_all.add(user_name)
                                unique_users_all.append(user_name)
                    # Apply per-exchange user tracking limits if configured.
                    user_limit = self._price_subscribe_user_limit_by_exchange.get(exchange)
                    if user_limit is not None:
                        # Build ordered list of unique users from mapping and trim if necessary
                        unique_users = []
                        seen = set()
                        for ccxt_sym, lst in mapping.items():
                            for user_name, _ in lst:
                                if user_name not in seen:
                                    seen.add(user_name)
                                    unique_users.append(user_name)
                        if len(unique_users) > user_limit:
                            allowed_list = unique_users[:user_limit]
                            blocked_list = unique_users[user_limit:]
                            allowed = set(allowed_list)
                            # Build reduced mapping that only includes allowed users
                            reduced = {}
                            for ccxt_sym, lst in mapping.items():
                                filtered = [t for t in lst if t[0] in allowed]
                                if filtered:
                                    reduced[ccxt_sym] = filtered
                            # Persist reduced mapping back into the exchange config so
                            # subsequent iterations don't re-evaluate the full mapping
                            try:
                                cfg['mapping'] = reduced
                                cfg['symbols'] = set(reduced.keys())
                                self._price_exchange_config[exchange] = cfg
                            except Exception:
                                pass
                            # Log the trimming only when the allowed set changed
                            prev = self._price_subscribe_trim_snapshot.get(exchange)
                            curr_snapshot = tuple(allowed_list)
                            if prev != curr_snapshot:
                                _human_log('PBData', f"[ws] Exchange {exchange} has {len(unique_users)} users but limit is {user_limit}; subscribing only for {len(allowed_list)} users and falling back to REST for others", level='INFO')
                                try:
                                    _human_log('PBData', f"[ws] Allowed users for {exchange}: {', '.join(allowed_list)}", level='INFO')
                                    _human_log('PBData', f"[ws] Blocked users for {exchange}: {', '.join(blocked_list)}", level='INFO')
                                except Exception:
                                    pass
                                self._price_subscribe_trim_snapshot[exchange] = curr_snapshot
                            mapping = reduced
                    if not symbols:
                        await asyncio.sleep(1)
                        continue
                    supports_batch = False
                    try:
                        if hasattr(ex, 'has'):
                            has = ex.has
                            if isinstance(has, dict):
                                supports_batch = has.get('watchTickers', False)
                            else:
                                supports_batch = getattr(has, 'watchTickers', False)
                    except Exception:
                        supports_batch = False
                    now_ts = int(datetime.now().timestamp())
                    if now_ts - last_heartbeat_ts >= 120:
                        tick_count = self._price_ticks_count.get(exchange, 0)
                        _human_log('PBData', f"[ws] Price heartbeat for exchange {exchange}: {len(symbols)} symbol(s), mode={'batch' if supports_batch else 'per-symbol'}, ticks_since_last={tick_count}", level='INFO')
                        self._price_ticks_count[exchange] = 0
                        last_heartbeat_ts = now_ts
                    if not started_logged:
                        mode = 'batch watch_tickers' if supports_batch else 'per-symbol watch_ticker'
                        _human_log('PBData', f"[ws] Price stream active for exchange {exchange}: {len(symbols)} symbol(s), mode={mode}", level='INFO')
                        started_logged = True

                    if supports_batch:
                        # Ensure we only attempt to subscribe to newly added
                        # symbols to avoid re-subscribing and triggering
                        # 'already subscribed' errors from exchanges.
                        requested_set = set(symbols)
                        subscribed = self._price_subscribed_symbols.get(exchange, set())
                        added = list(requested_set - subscribed)
                        if added:
                            # Subscribe in chunks to avoid sending very large batch
                            # subscriptions which some exchanges (e.g. hyperliquid)
                            # may reject or rate-limit. Allow per-exchange overrides.
                            per_ex_chunk = getattr(self, '_price_subscribe_chunk_size_by_exchange', None)
                            if per_ex_chunk and exchange in per_ex_chunk:
                                chunk_size = per_ex_chunk[exchange]
                            else:
                                chunk_size = getattr(self, '_price_subscribe_chunk_size', 20)
                            try:
                                for i in range(0, len(added), chunk_size):
                                    chunk = added[i:i+chunk_size]
                                    try:
                                        # give a bit more timeout headroom for some exchanges
                                        timeout_val = getattr(self, '_price_watch_timeout', None)
                                        if timeout_val:
                                            await asyncio.wait_for(ex.watch_tickers(chunk), timeout=timeout_val)
                                        else:
                                            await ex.watch_tickers(chunk)
                                        # Merge successful chunk into subscribed set
                                        subscribed = subscribed.union(set(chunk))
                                        self._price_subscribed_symbols[exchange] = subscribed
                                        # reset backoff on success of any chunk
                                        subscribe_backoff = 0
                                    except Exception as e:
                                        raw = str(e)
                                        lower = raw.lower()
                                        # Treat transient/connect timeouts specially (ccxt RequestTimeout / "connection timeout")
                                        if 'timeout' in lower or 'requesttimeout' in lower or 'connection timeout' in lower:
                                            _human_log('PBData', f"[ws] watch_tickers subscribe TIMEOUT for exchange {exchange} (chunk {i}-{i+chunk_size}): {e}; backing off")
                                            subscribe_backoff = min(subscribe_backoff + 1, 6)
                                            # if repeated timeouts, nudge a reconnect of shared client
                                            if subscribe_backoff >= 3:
                                                try:
                                                    await Exchange.close_shared_ws_client(exchange)
                                                except Exception:
                                                    pass
                                                await asyncio.sleep(30)
                                            else:
                                                await asyncio.sleep(min(5 * subscribe_backoff, 30))
                                            # abort subscription iteration to allow outer loop to re-evaluate
                                            raise RuntimeError("subscribe_chunk_failed")
                                        # Treat 'already subscribed' as non-fatal and merge
                                        if 'already subscribed' in lower or 'already subscribed' in raw:
                                            subscribed = subscribed.union(set(chunk))
                                            self._price_subscribed_symbols[exchange] = subscribed
                                            _human_log('PBData', f"[ws] watch_tickers subscribe: already subscribed for exchange {exchange}: {e}; continuing")
                                            continue
                                        # Detect exchange-enforced subscribe limits (e.g. hyperliquid)
                                        if 'cannot track more than' in lower or 'cannot track more than' in raw:
                                            try:
                                                _human_log('PBData', f"[ws] watch_tickers subscribe REJECTED by exchange {exchange}: {e}; users={len(unique_users_all)}; closing shared client and entering backoff")
                                            except Exception:
                                                pass
                                            try:
                                                await Exchange.close_shared_ws_client(exchange)
                                            except Exception:
                                                pass
                                            # Put the exchange into backoff to avoid rapid retry
                                            try:
                                                self._set_exchange_backoff(exchange, reason='subscribe_limit')
                                            except Exception:
                                                pass
                                            # Allow time for exchange/client to settle
                                            await asyncio.sleep(30 + random.uniform(0, 5))
                                            # Abort current subscription loop and continue outer loop
                                            raise RuntimeError("subscribe_chunk_failed")
                                        # For other errors, log full traceback to aid debugging
                                        tb = traceback.format_exc()
                                        _human_log('PBData', f"[ws] watch_tickers subscribe ERROR for exchange {exchange} (chunk {i}-{i+chunk_size}): {e}\n{tb}", level='ERROR')
                                        # If we repeatedly fail, close shared client to force a fresh reconnect
                                        subscribe_backoff = min(subscribe_backoff + 1, 6)
                                        if subscribe_backoff >= 3:
                                            try:
                                                await Exchange.close_shared_ws_client(exchange)
                                            except Exception:
                                                pass
                                            # longer sleep to allow network/exchange to recover
                                            await asyncio.sleep(30)
                                        else:
                                            delay = min(5 * subscribe_backoff, 30)
                                            await asyncio.sleep(delay)
                                        # Abort current subscription loop and continue outer loop
                                        raise RuntimeError("subscribe_chunk_failed")
                            except RuntimeError:
                                # Subscription chunk failed; go to next outer iteration
                                continue
                        # Now wait for tickers for the currently requested set. Use the
                        # requested_set (not 'added') to receive updates for all
                        # symbols of interest.
                        try:
                            timeout_val = getattr(self, '_price_watch_timeout', None)
                            if timeout_val:
                                tickers = await asyncio.wait_for(ex.watch_tickers(list(requested_set)), timeout=timeout_val)
                            else:
                                tickers = await ex.watch_tickers(list(requested_set))
                        except asyncio.TimeoutError:
                            _human_log('PBData', f"[ws] watch_tickers TIMEOUT for exchange {exchange}; reconnecting price client", level='WARNING')
                            ex = await Exchange.get_shared_ws_client(exchange)
                            if not ex:
                                _human_log('PBData', f"[ws] ccxtpro unavailable (price) after reconnect for exchange {exchange}", level='WARNING')
                                return
                            subscribe_backoff = min(subscribe_backoff + 1, 6)
                            delay = min(5 * subscribe_backoff, 30)
                            await asyncio.sleep(delay)
                            continue
                        except Exception as e:
                            raw = str(e)
                            lower = raw.lower()
                            # If this was a benign/normal close (e.g. code 1000) attempt lightweight reconnect
                            if self._is_normal_ws_close(raw):
                                _human_log('PBData', f"[ws] watch_tickers normal websocket close for exchange {exchange}: {e}; attempting reconnect", level='WARNING')
                                ex = await Exchange.get_shared_ws_client(exchange)
                                if not ex:
                                    _human_log('PBData', f"[ws] ccxtpro unavailable (price) after normal close reconnect for exchange {exchange}", level='WARNING')
                                    return
                                await asyncio.sleep(1)
                                continue
                            # If the exchange enforces a hard subscribe limit, close client and backoff
                            if 'cannot track more than' in lower or 'cannot track more than' in raw:
                                try:
                                    _human_log('PBData', f"[ws] watch_tickers subscribe REJECTED by exchange {exchange}: {e}; users={len(unique_users_all)}; closing shared client and entering backoff", level='ERROR')
                                except Exception:
                                    pass
                                try:
                                    await Exchange.close_shared_ws_client(exchange)
                                except Exception:
                                    pass
                                try:
                                    self._set_exchange_backoff(exchange, reason='subscribe_limit')
                                except Exception:
                                    pass
                                await asyncio.sleep(30 + random.uniform(0, 5))
                                continue
                            # Treat ccxt RequestTimeout / connection timeouts as transient here too
                            if 'timeout' in lower or 'requesttimeout' in lower or 'connection timeout' in lower:
                                _human_log('PBData', f"[ws] watch_tickers REQUESTTIMEOUT for exchange {exchange}: {e}; reconnecting price client (transient)", level='WARNING')
                                ex = await Exchange.get_shared_ws_client(exchange)
                                if not ex:
                                    _human_log('PBData', f"[ws] ccxtpro unavailable (price) after reconnect for exchange {exchange}")
                                    return
                                subscribe_backoff = min(subscribe_backoff + 1, 6)
                                delay = min(5 * subscribe_backoff, 30)
                                await asyncio.sleep(delay)
                                continue
                            # Some exchange implementations (e.g. bybit) raise an
                            # 'already subscribed' error when a topic is re-subscribed.
                            # Treat this as non-fatal.
                            if 'already subscribed' in lower:
                                _human_log('PBData', f"[ws] watch_tickers: already subscribed for exchange {exchange}: {e}; ignoring and continuing", level='DEBUG')
                                await asyncio.sleep(1)
                                continue
                            _human_log('PBData', f"[ws] watch_tickers ERROR for exchange {exchange}: {e}", level='ERROR')
                            ex = await Exchange.get_shared_ws_client(exchange)
                            if not ex:
                                _human_log('PBData', f"[ws] ccxtpro unavailable (price) after error reconnect for exchange {exchange}")
                                return
                            subscribe_backoff = min(subscribe_backoff + 1, 6)
                            delay = min(5 * subscribe_backoff, 30)
                            await asyncio.sleep(delay)
                            continue
                        self._price_ticks_count[exchange] = self._price_ticks_count.get(exchange, 0) + len(tickers)
                        ts_now = int(datetime.now().timestamp() * 1000)
                        for ccxt_symbol, ticker in tickers.items():
                            last = ticker.get('last')
                            ts = ticker.get('timestamp') or ts_now
                            if last is None:
                                continue
                            for user_name, internal_symbol in mapping.get(ccxt_symbol, []):
                                try:
                                    user = self.users.find_user(user_name)
                                    if user:
                                        key = (user.name, internal_symbol)
                                        now_sec = datetime.now().timestamp()
                                        if now_sec - last_price_write_ts.get(key, 0.0) >= throttle_interval_sec:
                                            last_price_write_ts[key] = now_sec
                                            await self.buffer_price(user, internal_symbol, ts, last)
                                except Exception as e:
                                    _human_log('PBData', f"[ws] upsert_price failed {user_name} {internal_symbol}: {e}", level='WARNING', user=user_name)
                    else:
                        # Fallback: iterate symbols and watch individually (still single task)
                        ts_now = int(datetime.now().timestamp() * 1000)
                        for ccxt_symbol in symbols:
                            try:
                                timeout_val = getattr(self, '_price_watch_timeout', None)
                                if timeout_val:
                                    try:
                                        ticker = await asyncio.wait_for(ex.watch_ticker(ccxt_symbol), timeout=timeout_val)
                                    except asyncio.TimeoutError:
                                        _human_log('PBData', f"[ws] watch_ticker TIMEOUT exchange {exchange} {ccxt_symbol}; reconnecting price client", level='WARNING')
                                        ex = await Exchange.get_shared_ws_client(exchange)
                                        if not ex:
                                            _human_log('PBData', f"[ws] ccxtpro unavailable (price) after reconnect for exchange {exchange}", level='WARNING')
                                            raise RuntimeError("price client unavailable after watch_ticker timeout")
                                        continue
                                else:
                                    try:
                                        ticker = await ex.watch_ticker(ccxt_symbol)
                                    except Exception as e:
                                        raw = str(e)
                                        lower = raw.lower()
                                        if 'timeout' in lower or 'requesttimeout' in lower or 'connection timeout' in lower:
                                            _human_log('PBData', f"[ws] watch_ticker REQUESTTIMEOUT exchange {exchange} {ccxt_symbol}: {e}; reconnecting price client (transient)", level='WARNING')
                                            ex = await Exchange.get_shared_ws_client(exchange)
                                            if not ex:
                                                _human_log('PBData', f"[ws] ccxtpro unavailable (price) after reconnect for exchange {exchange}", level='WARNING')
                                                raise RuntimeError("price client unavailable after watch_ticker error")
                                            await asyncio.sleep(1)
                                            continue
                                        if self._is_normal_ws_close(raw):
                                            _human_log('PBData', f"[ws] watch_ticker normal websocket close exchange {exchange} {ccxt_symbol}: {e}; attempting reconnect", level='WARNING')
                                        else:
                                            _human_log('PBData', f"[ws] watch_ticker ERROR exchange {exchange} {ccxt_symbol}: {e}; reconnecting price client", level='ERROR')
                                        ex = await Exchange.get_shared_ws_client(exchange)
                                        if not ex:
                                            _human_log('PBData', f"[ws] ccxtpro unavailable (price) after reconnect for exchange {exchange}", level='WARNING')
                                            raise RuntimeError("price client unavailable after watch_ticker error")
                                        await asyncio.sleep(1)
                                        continue

                                last = ticker.get('last')
                                ts = ticker.get('timestamp') or ts_now
                                if last is None:
                                    continue
                                for user_name, internal_symbol in mapping.get(ccxt_symbol, []):
                                    try:
                                        user = self.users.find_user(user_name)
                                        if user:
                                            key = (user.name, internal_symbol)
                                            now_sec = datetime.now().timestamp()
                                            if now_sec - last_price_write_ts.get(key, 0.0) >= throttle_interval_sec:
                                                last_price_write_ts[key] = now_sec
                                                await self.buffer_price(user, internal_symbol, ts, last)
                                                # _human_log('PBData', f"[ws] upsert_price wrote {user.name} {internal_symbol} price={last} ts={ts}")
                                    except Exception as e:
                                        _human_log('PBData', f"[ws] upsert_price failed {user_name} {internal_symbol}: {e}", level='WARNING', user=user_name)
                            except Exception as e:
                                _human_log('PBData', f"[ws] watch_ticker error exchange {exchange} {ccxt_symbol}: {e}", level='ERROR')
                                await asyncio.sleep(0.5)
            except Exception as e:
                raw = str(e)
                # treat normal closes as less severe and attempt graceful reconnect
                if self._is_normal_ws_close(raw):
                    _human_log('PBData', f"[ws] price loop normal websocket close exchange {exchange}: {e}; restarting price watcher after short pause", level='WARNING')
                    await asyncio.sleep(2)
                    continue
                _human_log('PBData', f"[ws] price loop error exchange {exchange}: {e}; restarting price watcher", level='ERROR')
            finally:
                # Do not close the websocket client here. Closing a shared
                # client can abort other watchers that rely on the same
                # underlying connection. Leave cleanup to a centralized
                # manager (Exchange) or process shutdown logic.
                try:
                        _human_log('PBData', f"Leaving price ws client open for exchange {exchange}", level='DEBUG')
                except Exception:
                    pass
            # Small delay before recreating client to avoid tight restart loops
            await asyncio.sleep(5)

    async def _ensure_exchange_price_watcher(self, exchange: str, mapping: dict):
        # mapping: { ccxt_symbol: [(user_name, internal_symbol), ...] }
        symbols = set(mapping.keys())
        # If no symbols for this exchange, stop existing watcher if any
        if not symbols:
            task = self._price_exchange_tasks.pop(exchange, None)
            if task and not task.done():
                try:
                    task.cancel()
                except Exception:
                    pass
            _human_log('PBData', f"[ws] Stopped price watcher for exchange {exchange}: no symbols for fetch_users", level='INFO')
            self._price_exchange_config.pop(exchange, None)
            return
        cfg = self._price_exchange_config.get(exchange)
        if not cfg:
            self._price_exchange_config[exchange] = {'mapping': mapping, 'symbols': symbols}
            _human_log('PBData', f"[ws] Preparing price watcher for exchange {exchange}: {len(symbols)} symbol(s)", level='INFO')
        else:
            # Update mapping and symbols; log changes if any
            old_symbols = cfg.get('symbols', set())
            if old_symbols != symbols:
                added = symbols - old_symbols
                removed = old_symbols - symbols
                _human_log('PBData', f"[ws] Updated symbols for exchange {exchange}: {len(symbols)} now; +{len(added)} / -{len(removed)}", level='INFO')
            cfg['mapping'] = mapping
            cfg['symbols'] = symbols
        task = self._price_exchange_tasks.get(exchange)
        if not task or task.done():
            task = asyncio.create_task(self._price_exchange_ws_loop(exchange))
            self._price_exchange_tasks[exchange] = task

    async def _build_price_mapping_for_exchange(self, exchange: str, users: list):
        """Build and apply price mapping for a single exchange in the background.

        This runs in its own task so that expensive DB work (fetch_positions
        per user) does not block the main update_db_async loop.
        """
        try:
            now_ts = datetime.now().timestamp()
            # Throttle how often we actually rebuild the mapping per exchange.
            # When many background tasks are active, this prevents continuous fetch_positions load and lets the
            # price websocket loop run freely.
            last = self._last_mapping_log_ts_by_exchange.get(exchange, 0.0)
            if now_ts - last < self._mapping_rebuild_min_interval:
                return
            self._last_mapping_log_ts_by_exchange[exchange] = now_ts
            _human_log('PBData', f"[async] Building price mapping for exchange {exchange} with {len(users)} user(s)", level='INFO')
            mapping = {}
            for user in users:
                try:
                    positions = await asyncio.to_thread(self.db.fetch_positions, user)
                except Exception as e:
                    _human_log('PBData', f"[async] fetch_positions failed for {user.name} ({exchange}): {e}", level='ERROR', user=user)
                    continue
                for pos in positions:
                    internal_symbol = pos[1]
                    ccxt_symbol = self._to_ccxt_symbol(internal_symbol)
                    mapping.setdefault(ccxt_symbol, []).append((user.name, internal_symbol))
            # Always log the final symbol count so we know whether the
            # mapping contains anything for this exchange.
            _human_log('PBData', f"[async] Built price mapping for exchange {exchange}: {len(mapping)} symbol(s)", level='INFO')
            await self._ensure_exchange_price_watcher(exchange, mapping)
        except Exception as e:
            _human_log('PBData', f"[async] _build_price_mapping_for_exchange error {exchange}: {e}", level='ERROR')

    async def update_db_async(self):
        # Load users first so filtering in load_fetch_users is correct
        self.users.load()
        self.load_fetch_users()
        # Reload debug setting if pbgui.ini changed
        try:
            self._load_settings()
        except Exception:
            pass

        # If poll intervals changed via settings, restart shared pollers so
        # new intervals take effect without restarting PBData.
        try:
            if getattr(self, '_poll_intervals_changed', False):
                for tname in ("_shared_combined_task", "_shared_history_task", "_shared_executions_task"):
                    try:
                        t = getattr(self, tname, None)
                        if t and not t.done():
                            t.cancel()
                    except Exception:
                        pass
                    try:
                        setattr(self, tname, None)
                    except Exception:
                        pass
                # Also cancel per-exchange history tasks so new interval takes effect.
                for _exch, _t in list(getattr(self, '_shared_history_tasks_by_exchange', {}).items()):
                    try:
                        if _t and not _t.done():
                            _t.cancel()
                    except Exception:
                        pass
                try:
                    self._shared_history_tasks_by_exchange = {}
                except Exception:
                    pass
                # Also cancel per-exchange combined tasks so new interval takes effect.
                for _exch, _t in list(getattr(self, '_shared_combined_tasks_by_exchange', {}).items()):
                    try:
                        if _t and not _t.done():
                            _t.cancel()
                    except Exception:
                        pass
                try:
                    self._shared_combined_tasks_by_exchange = {}
                except Exception:
                    pass
                self._poll_intervals_changed = False
        except Exception:
            pass
        now_ts = datetime.now().timestamp()
        # Only log when the set changes or periodically
        current_users_set = set(self.fetch_users)
        if current_users_set != self._last_fetch_users_snapshot or (now_ts - self._last_queue_log_ts) >= self._queue_log_every_secs:
            _human_log('PBData', f"[async] Will process users: {self.fetch_users}", level='INFO')
            self._last_fetch_users_snapshot = current_users_set
            self._last_queue_log_ts = now_ts
            try:
                # Print a summary of which users use websockets vs shared REST
                self.print_fetch_method_summary()
            except Exception:
                pass

        # Group users by exchange
        users_by_exchange = defaultdict(list)
        for user in self.users:
            if user.name in self.fetch_users:
                users_by_exchange[user.exchange].append(user)

        for exchange, users in users_by_exchange.items():
            # Log only when count changes for this exchange or periodically
            count = len(users)
            prev = self._last_exchange_queue_counts.get(exchange)
            if prev != count or (now_ts - self._last_queue_log_ts) >= self._queue_log_every_secs:
                _human_log('PBData', f"[async] Queueing {count} user(s) for exchange: {exchange}", level='INFO')
                self._last_exchange_queue_counts[exchange] = count

        # Start shared serial pollers only after a grace period so that
        # websocket startup and initial subscriptions don't coincide with
        # heavy REST history/position/order/balance traffic.
        now_ts = datetime.now().timestamp()
        if now_ts >= self._pollers_enabled_after_ts:
            try:
                # Per-exchange combined pollers: one task per exchange so that slow or
                # rate-limited exchanges (e.g. hyperliquid) don't block fast ones (e.g. binance).
                if not hasattr(self, "_shared_combined_tasks_by_exchange"):
                    self._shared_combined_tasks_by_exchange = {}
                # Determine the set of exchanges currently in fetch_users.
                _comb_exchanges = set()
                for _uname in self.fetch_users:
                    _u = self.users.find_user(_uname)
                    if _u and getattr(_u, 'exchange', None):
                        _comb_exchanges.add(_u.exchange)
                for _exch in _comb_exchanges:
                    _t = self._shared_combined_tasks_by_exchange.get(_exch)
                    if _t is None or _t.done():
                        self._shared_combined_tasks_by_exchange[_exch] = asyncio.create_task(
                            self._shared_combined_poll_serial(
                                self._shared_combined_interval_seconds,
                                per_exchange=False,
                                exchange_filter=_exch,
                            )
                        )
                # Legacy single-task attribute: keep for backwards compat (e.g. interval-change restart).
                if not hasattr(self, "_shared_combined_task") or self._shared_combined_task is None:
                    self._shared_combined_task = asyncio.create_task(asyncio.sleep(0))
                # Per-exchange history pollers: one task per exchange so that slow
                # exchanges (e.g. hyperliquid) don't block fast ones (e.g. binance).
                if not hasattr(self, "_shared_history_tasks_by_exchange"):
                    self._shared_history_tasks_by_exchange = {}
                # Determine the set of exchanges currently in fetch_users.
                _hist_exchanges = set()
                for _uname in self.fetch_users:
                    _u = self.users.find_user(_uname)
                    if _u and getattr(_u, 'exchange', None):
                        _hist_exchanges.add(_u.exchange)
                for _exch in _hist_exchanges:
                    _t = self._shared_history_tasks_by_exchange.get(_exch)
                    if _t is None or _t.done():
                        self._shared_history_tasks_by_exchange[_exch] = asyncio.create_task(
                            self._shared_poll_serial(
                                'history',
                                self._shared_history_interval_seconds,
                                per_exchange=False,
                                exchange_filter=_exch,
                            )
                        )
                # Legacy single-task attribute: keep for backwards compat (e.g. interval-change restart).
                # Always set to a dummy done sentinel so the old restart path doesn't spin up a second
                # all-exchanges task.
                if not hasattr(self, "_shared_history_task") or self._shared_history_task is None:
                    self._shared_history_task = asyncio.create_task(asyncio.sleep(0))
                if not hasattr(self, "_shared_executions_task") or self._shared_executions_task is None or self._shared_executions_task.done():
                    # Delay first executions run a bit to avoid a big startup burst.
                    # First run: ~60s ± 15s. Subsequent runs: every 30 minutes.
                    self._shared_executions_task = asyncio.create_task(
                        self._shared_poll_serial(
                            'executions',
                            self._shared_executions_interval_seconds,
                            per_exchange=True,
                            start_immediately=True,
                            start_delay_seconds=60,
                            initial_jitter_seconds=15,
                        )
                    )
                if not hasattr(self, "_latest_1m_task") or self._latest_1m_task is None or self._latest_1m_task.done():
                    self._latest_1m_task = asyncio.create_task(self._latest_1m_loop())
                if not hasattr(self, "_binance_latest_1m_task") or self._binance_latest_1m_task is None or self._binance_latest_1m_task.done():
                    self._binance_latest_1m_task = asyncio.create_task(self._binance_latest_1m_loop())
                if not hasattr(self, "_bybit_latest_1m_task") or self._bybit_latest_1m_task is None or self._bybit_latest_1m_task.done():
                    self._bybit_latest_1m_task = asyncio.create_task(self._bybit_latest_1m_loop())
            except Exception as e:
                _human_log('PBData', f"Error starting shared pollers: {e}", level='DEBUG')
        else:
            try:
                remaining = int(self._pollers_enabled_after_ts - now_ts)
                _human_log('PBData', f"[poll] Shared pollers delayed: starting in {remaining}s (now={int(now_ts)} enable_after={int(self._pollers_enabled_after_ts)})", level='DEBUG')
            except Exception:
                pass

        # Ensure one price watcher per exchange with combined symbols across its users.
        # Run mapping building in background tasks so update_db_async returns quickly.
        for exchange, users in users_by_exchange.items():
                asyncio.create_task(self._build_price_mapping_for_exchange(exchange, users))

    def print_fetch_method_summary(self):
        """Log a summary which users use websocket watchers vs shared history/REST.

        For each configured fetch user, indicate whether balances/positions/orders
        are being updated via a running websocket watcher (ws) or via the
        shared REST poller (rest). History is always 'rest' (shared poller).
        """
        try:
            # Ensure user list is fresh
            try:
                self.users.load()
            except Exception:
                pass
            self.load_fetch_users()
            balances_ws = []
            balances_rest = []
            positions_ws = []
            positions_rest = []
            orders_ws = []
            orders_rest = []
            all_users = []
            for u in self.users:
                if u.name not in self.fetch_users:
                    continue
                all_users.append(u.name)
                balances_rest.append(u.name)
                positions_rest.append(u.name)
                orders_rest.append(u.name)

            # Include last-fetch (minutes ago) per user/kind where available
            now_ts = datetime.now().timestamp()
            def last_minutes(user_name, kind):
                try:
                    ts = self._last_fetch_ts.get((user_name, kind))
                    if not ts:
                        return '(never)'
                    mins = int((now_ts - ts) // 60)
                    if mins <= 0:
                        return '0m'
                    return f'{mins}m'
                except Exception:
                    return '(unknown)'

            # Build a single multi-line, human-friendly summary block
            def join_csv(lst):
                return ', '.join(lst) if lst else '(none)'

            summary_lines = []
            summary_lines.append("[summary] Fetch method summary:")
            summary_lines.append(f"[summary] Balances: ws={len(balances_ws)} rest={len(balances_rest)}")
            summary_lines.append(f"[summary]   ws: {join_csv(balances_ws)}")
            # append last-fetch per user for balances
            summary_lines.append(f"[summary]   last_fetch_balances: " + ", ".join([f"{u}={last_minutes(u,'balances')}" for u in all_users]))
            summary_lines.append(f"[summary]   rest: {join_csv(balances_rest)}")
            summary_lines.append(f"[summary] Positions: ws={len(positions_ws)} rest={len(positions_rest)}")
            summary_lines.append(f"[summary]   ws: {join_csv(positions_ws)}")
            summary_lines.append(f"[summary]   last_fetch_positions: " + ", ".join([f"{u}={last_minutes(u,'positions')}" for u in all_users]))
            summary_lines.append(f"[summary]   rest: {join_csv(positions_rest)}")
            summary_lines.append(f"[summary] Orders: ws={len(orders_ws)} rest={len(orders_rest)}")
            summary_lines.append(f"[summary]   ws: {join_csv(orders_ws)}")
            summary_lines.append(f"[summary]   last_fetch_orders: " + ", ".join([f"{u}={last_minutes(u,'orders')}" for u in all_users]))
            summary_lines.append(f"[summary]   rest: {join_csv(orders_rest)}")
            summary_lines.append(f"[summary] History last_fetch: " + ", ".join([f"{u}={last_minutes(u,'history')}" for u in all_users]))
            summary_lines.append(f"[summary] History (rest): {join_csv(all_users)}")
            try:
                # Also write machine-readable JSON summary for GUI/monitoring
                try:
                    # Build machine-readable summary including last-fetch timestamps
                    lf = {}
                    # Execution-level polling is only meaningful for a subset of exchanges.
                    # Keep in sync with Database.update_executions() support.
                    try:
                        try:
                            self.load_trades_users()
                        except Exception:
                            pass
                        exec_users = [
                            u.name
                            for u in self.users
                            if u.name in self.fetch_users
                            and u.name in (self.trades_users or [])
                            and getattr(u, 'exchange', None) in ('hyperliquid', 'binance', 'bitget', 'bybit', 'okx', 'gateio')
                        ]
                    except Exception:
                        exec_users = []
                    try:
                        for u in all_users:
                            # Prefer the unified _last_fetch_ts mapping, but history
                            # timestamps may also be tracked in _history_rest_last
                            # under keys (user, exchange). Use a fallback to fill
                            # history if the primary mapping is missing.
                            hist_ts = self._last_fetch_ts.get((u, 'history'))
                            if not hist_ts:
                                # search _history_rest_last for any matching user
                                try:
                                    found = None
                                    for (uname, exch), ts in list(self._history_rest_last.items()):
                                        if uname == u and ts:
                                            if not found or (ts and ts > found):
                                                found = ts
                                    if found:
                                        hist_ts = found
                                except Exception:
                                    pass
                            lf[u] = {
                                'balances': self._last_fetch_ts.get((u, 'balances')),
                                'positions': self._last_fetch_ts.get((u, 'positions')),
                                'orders': self._last_fetch_ts.get((u, 'orders')),
                                'history': hist_ts,
                                'executions': self._last_fetch_ts.get((u, 'executions')),
                            }
                    except Exception:
                        pass
                    # Build price watcher info per exchange
                    prices_info = {}
                    for exch_name, task in list(self._price_exchange_tasks.items()):
                        cfg = self._price_exchange_config.get(exch_name, {})
                        sym_count = len(cfg.get('symbols', set()))
                        active = bool(task and not task.done())
                        prices_info[exch_name] = {'symbols': sym_count, 'active': active}
                    summary_obj = {
                        'timestamp': datetime.now().isoformat(sep=' ', timespec='seconds'),
                        'balances': {'ws': balances_ws, 'rest': balances_rest},
                        'positions': {'ws': positions_ws, 'rest': positions_rest},
                        'orders': {'ws': orders_ws, 'rest': orders_rest},
                        'prices': prices_info,
                        'history': all_users,
                        'executions': exec_users,
                        'last_fetch_ts': lf,
                    }
                    logs_dir = _Path(f"{PBGDIR}/data/logs")
                    if not logs_dir.exists():
                        try:
                            logs_dir.mkdir(parents=True, exist_ok=True)
                        except Exception:
                            pass
                    summary_path = logs_dir / 'fetch_summary.json'
                    try:
                        with open(summary_path, 'w') as _f:
                            json.dump(summary_obj, _f, indent=2, ensure_ascii=False)
                    except Exception:
                        pass
                except Exception:
                    pass
                # Note: remove multi-line human-readable summary log to avoid
                # making a single log entry out of many lines (which breaks
                # leading-tag parsing). The machine-readable `fetch_summary.json`
                # remains as the canonical summary.
            except Exception:
                # If writing JSON fails, fall back to logging a minimal error
                try:
                    _human_log('PBData', f"[summary] Failed to write fetch_summary.json", level='WARNING')
                except Exception:
                    pass
        except Exception as e:
            try:
                _human_log('PBData', f"[summary] Failed to build fetch method summary: {e}", level='WARNING')
            except Exception:
                pass

    def _write_poller_metrics(self):
        """Write poller_metrics.json for the GUI metrics panel.

        Collects per-exchange counters from _poller_metrics dict, backoff states,
        semaphore stats, and market-data status. Safe to call from _metrics_loop.
        """
        try:
            now = datetime.now().timestamp()
            # Per-exchange poller counters
            exchanges = {}
            for exch, m in dict(self._poller_metrics).items():
                exchanges[exch] = dict(m)
                # Inject current backoff status
                bo_until = self._exchange_backoff_until.get(exch, 0)
                hbo_until = self._exchange_history_backoff_until.get(exch, 0)
                exchanges[exch]['backoff_remaining_s'] = max(0, int(bo_until - now)) if bo_until > now else 0
                exchanges[exch]['history_backoff_remaining_s'] = max(0, int(hbo_until - now)) if hbo_until > now else 0

            # REST semaphore stats
            semaphores = {}
            for exch, sem in dict(self._rest_semaphores).items():
                limit = self._rest_semaphore_limits_by_exchange.get(exch, self._default_rest_semaphore_limit)
                try:
                    available = sem._value  # asyncio.Semaphore internal counter
                except Exception:
                    available = -1
                semaphores[exch] = {
                    'slots': limit,
                    'available': available,
                    'in_use': max(0, limit - available) if available >= 0 else 0,
                }

            # Market data status (read existing file)
            market_data = {}
            try:
                if self._market_data_status_path.exists():
                    md = json.loads(self._market_data_status_path.read_text(encoding='utf-8'))
                    for key in ('latest_1m', 'binance_latest_1m', 'bybit_latest_1m'):
                        if key in md:
                            entry = md[key]
                            market_data[key] = {
                                'exchange': entry.get('exchange', ''),
                                'running': entry.get('running', False),
                                'coins_done': entry.get('coins_done', 0),
                                'coins_total': entry.get('coins_total', 0),
                                'last_run_ts': entry.get('last_run_ts', 0),
                                'current_coin': entry.get('current_coin'),
                            }
            except Exception:
                pass

            # Rate-limit budgets
            budgets = {}
            for exch_name, budget in self._rate_budgets.items():
                try:
                    budgets[exch_name] = budget.peek()
                except Exception:
                    pass

            obj = {
                'timestamp': datetime.now().isoformat(sep=' ', timespec='seconds'),
                'exchanges': exchanges,
                'semaphores': semaphores,
                'market_data': market_data,
                'budgets': budgets,
            }
            logs_dir = _Path(f"{PBGDIR}/data/logs")
            if not logs_dir.exists():
                try:
                    logs_dir.mkdir(parents=True, exist_ok=True)
                except Exception:
                    pass
            try:
                with open(logs_dir / 'poller_metrics.json', 'w') as _f:
                    json.dump(obj, _f, indent=2, ensure_ascii=False)
            except Exception:
                pass
        except Exception:
            pass

    def _write_fetch_summary(self):
        """Write machine-readable `fetch_summary.json` from current in-memory state.

        Safe to call from websocket/poller loops; exceptions are swallowed.
        """
        try:
            try:
                self.users.load()
            except Exception:
                pass
            self.load_fetch_users()
            try:
                self.load_trades_users()
            except Exception:
                pass
            balances_ws = []
            balances_rest = []
            positions_ws = []
            positions_rest = []
            orders_ws = []
            orders_rest = []
            all_users = []
            for u in self.users:
                if u.name not in self.fetch_users:
                    continue
                all_users.append(u.name)
                balances_rest.append(u.name)
                positions_rest.append(u.name)
                orders_rest.append(u.name)

            # Executions polling is only meaningful for a subset of exchanges.
            # Keep in sync with Database.update_executions() support.
            try:
                exec_users = [
                    u.name
                    for u in self.users
                    if u.name in self.fetch_users
                    and u.name in (self.trades_users or [])
                    and getattr(u, 'exchange', None) in ('hyperliquid', 'binance', 'bitget', 'bybit', 'okx', 'gateio')
                ]
            except Exception:
                exec_users = []

            # Compose last-fetch mapping with history fallback to _history_rest_last
            lf = {}
            for u in all_users:
                hist_ts = self._last_fetch_ts.get((u, 'history'))
                if not hist_ts:
                    try:
                        found = None
                        for (uname, exch), ts in list(self._history_rest_last.items()):
                            if uname == u and ts:
                                if not found or (ts and ts > found):
                                    found = ts
                        if found:
                            hist_ts = found
                    except Exception:
                        pass
                lf[u] = {
                    'balances': self._last_fetch_ts.get((u, 'balances')),
                    'positions': self._last_fetch_ts.get((u, 'positions')),
                    'orders': self._last_fetch_ts.get((u, 'orders')),
                    'history': hist_ts,
                    'executions': self._last_fetch_ts.get((u, 'executions')),
                }

            # Build price watcher info per exchange
            prices_info = {}
            for exch_name, task in list(self._price_exchange_tasks.items()):
                cfg = self._price_exchange_config.get(exch_name, {})
                sym_count = len(cfg.get('symbols', set()))
                active = bool(task and not task.done())
                # Collect unique user-facing (internal) symbols from the mapping
                user_syms: list = []
                try:
                    mapping = cfg.get('mapping', {})
                    seen_syms: set = set()
                    for pairs in mapping.values():
                        for _uname, internal_sym in pairs:
                            if internal_sym not in seen_syms:
                                seen_syms.add(internal_sym)
                                user_syms.append(internal_sym)
                    user_syms.sort()
                except Exception:
                    user_syms = []
                prices_info[exch_name] = {'symbols': sym_count, 'active': active, 'symbol_list': user_syms}
            summary_obj = {
                'timestamp': datetime.now().isoformat(sep=' ', timespec='seconds'),
                'balances': {'ws': balances_ws, 'rest': balances_rest},
                'positions': {'ws': positions_ws, 'rest': positions_rest},
                'orders': {'ws': orders_ws, 'rest': orders_rest},
                'prices': prices_info,
                'history': all_users,
                'executions': exec_users,
                'last_fetch_ts': lf,
            }
            logs_dir = _Path(f"{PBGDIR}/data/logs")
            if not logs_dir.exists():
                try:
                    logs_dir.mkdir(parents=True, exist_ok=True)
                except Exception:
                    pass
            summary_path = logs_dir / 'fetch_summary.json'
            try:
                with open(summary_path, 'w') as _f:
                    json.dump(summary_obj, _f, indent=2, ensure_ascii=False)
            except Exception:
                pass
        except Exception:
            # never raise from this helper
            pass

def main():
    """Entry point kept synchronous; spins up an async loop internally."""
    dest = Path(f'{PBGDIR}/data/logs')
    dest.mkdir(parents=True, exist_ok=True)
    pbdata = PBData()
    if pbdata.is_running():
        _human_log('PBData', 'Error: PBData already started', level='ERROR')
        sys.exit(1)
    _human_log('PBData', 'Start: PBData', level='INFO')
    pbdata.save_pid()

    async def run_loop():
        # Start metrics loop so periodic client/backoff metrics appear in logs
        try:
            if not hasattr(pbdata, '_metrics_task') or pbdata._metrics_task is None or pbdata._metrics_task.done():
                pbdata._metrics_task = asyncio.create_task(pbdata._metrics_loop())
        except Exception:
            pass

        # Start price writer background task if not already running
        try:
            if not hasattr(pbdata, '_price_writer_task') or pbdata._price_writer_task is None or getattr(pbdata, '_price_writer_task').done():
                pbdata._price_writer_task = asyncio.create_task(pbdata._price_writer_loop())
        except Exception:
            pass

        while True:
            try:
                try:
                    await pbdata.update_db_async()
                except Exception as e:
                    try:
                        tb = traceback.format_exc()
                        _human_log('PBData', f"[loop] update_db_async ERROR: {e}", level='ERROR', meta={'traceback': tb})
                    except Exception:
                        pass
                await asyncio.sleep(1)
            except Exception as e:
                try:
                    tb = traceback.format_exc()
                    _human_log('PBData', f"[loop] Unexpected error: {e}", level='ERROR', meta={'traceback': tb})
                except Exception:
                    pass

    async def _shutdown():
        try:
            try:
                _human_log('PBData', '[shutdown] Shutdown requested: closing ws clients', level='INFO')
            except Exception:
                pass
            from Exchange import Exchange
            try:
                await Exchange.close_all_ws_clients()
            except Exception:
                pass
            # Disable further buffering and attempt a final flush before cancelling tasks
            try:
                try:
                    pbdata._price_buffer_enabled = False
                except Exception:
                    pass
                if hasattr(pbdata, '_price_writer_task') and pbdata._price_writer_task is not None:
                    # Ask PBData to flush buffer synchronously with a timeout
                    try:
                        await asyncio.wait_for(pbdata._flush_price_buffer(), timeout=10)
                    except Exception as e:
                        try:
                            _human_log('PBData', f"[shutdown] final price buffer flush failed/timeout: {e}", level='WARNING')
                        except Exception:
                            pass
                    # Cancel the writer task if it's still running and wait briefly
                    try:
                        if not pbdata._price_writer_task.done():
                            pbdata._price_writer_task.cancel()
                            try:
                                await asyncio.wait_for(pbdata._price_writer_task, timeout=5)
                            except Exception:
                                pass
                    except Exception:
                        pass
            except Exception:
                pass
            # Cancel remaining tasks (except current)
            try:
                cur = asyncio.current_task()
                for t in list(asyncio.all_tasks()):
                    if t is not cur:
                        try:
                            t.cancel()
                        except Exception:
                            pass
            except Exception:
                pass
            # Close any per-thread cached DB connections to release fds
            try:
                if hasattr(pbdata, 'db') and hasattr(pbdata.db, 'close_thread_connections'):
                    try:
                        pbdata.db.close_thread_connections()
                    except Exception:
                        pass
            except Exception:
                pass
        except Exception:
            pass
        msg = '[shutdown] Exiting'
        try:
            _human_log('PBData', msg, level='INFO')
        except Exception:
            pass
        try:
            os._exit(0)
        except Exception:
            pass

    async def _runner_with_signals():
        try:
            loop = asyncio.get_running_loop()
            import signal as _signal
            for s in (_signal.SIGINT, _signal.SIGTERM):
                try:
                    loop.add_signal_handler(s, lambda sig=s: asyncio.create_task(_shutdown()))
                except Exception:
                    pass
        except Exception:
            pass
        await run_loop()

    try:
        asyncio.run(_runner_with_signals())
    except Exception:
        try:
            asyncio.run(run_loop())
        except Exception:
            pass

if __name__ == '__main__':
    main()