orderbook_watcher.aggregator

orderbook_watcher.aggregator

Orderbook aggregation logic across multiple directory nodes.

Classes

DirectoryNodeStatus

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

class DirectoryNodeStatus:
    def __init__(
        self,
        node_id: str,
        tracking_started: datetime | None = None,
        grace_period_seconds: int = 0,
    ) -> None:
        self.node_id = node_id
        self.connected = False
        self.last_connected: datetime | None = None
        self.last_disconnected: datetime | None = None
        self.connection_attempts = 0
        self.successful_connections = 0
        self.total_uptime_seconds = 0.0
        self.current_session_start: datetime | None = None
        self.tracking_started = tracking_started or datetime.now(UTC)
        self.grace_period_seconds = grace_period_seconds
        # Exponential backoff state
        self.retry_delay: float = 4.0  # Initial retry delay in seconds
        self.retry_delay_max: float = 3600.0  # Max retry delay (1 hour)
        self.retry_delay_multiplier: float = 1.5

    def mark_connected(self, current_time: datetime | None = None) -> None:
        now = current_time or datetime.now(UTC)
        self.connected = True
        self.last_connected = now
        self.current_session_start = now
        self.successful_connections += 1
        # Reset retry delay on successful connection
        self.retry_delay = 4.0

    def mark_disconnected(self, current_time: datetime | None = None) -> None:
        now = current_time or datetime.now(UTC)
        if self.connected and self.current_session_start:
            # Only count uptime after grace period
            grace_end_ts = self.tracking_started.timestamp() + self.grace_period_seconds
            session_start_ts = self.current_session_start.timestamp()
            now_ts = now.timestamp()

            # Calculate the actual uptime to record (only after grace period)
            if now_ts > grace_end_ts:
                # Some or all of the session is after grace period
                counted_start = max(session_start_ts, grace_end_ts)
                session_duration = now_ts - counted_start
                self.total_uptime_seconds += max(0, session_duration)

        self.connected = False
        self.last_disconnected = now
        self.current_session_start = None

    def get_uptime_percentage(self, current_time: datetime | None = None) -> float:
        if not self.tracking_started:
            return 0.0
        now = current_time or datetime.now(UTC)
        elapsed = (now - self.tracking_started).total_seconds()

        # If we're still in grace period, return 100% uptime
        if elapsed < self.grace_period_seconds:
            return 100.0

        # Calculate total time excluding grace period
        total_time = elapsed - self.grace_period_seconds
        if total_time == 0:
            return 0.0

        # Calculate uptime, but only count time after grace period ends
        grace_end = self.tracking_started.timestamp() + self.grace_period_seconds
        current_uptime = self.total_uptime_seconds

        if self.connected and self.current_session_start:
            # Only count uptime after grace period ended
            session_start_ts = self.current_session_start.timestamp()
            if session_start_ts < grace_end:
                # Session started during grace period, only count time after grace ended
                uptime_duration = now.timestamp() - grace_end
            else:
                # Session started after grace period
                uptime_duration = (now - self.current_session_start).total_seconds()
            current_uptime += max(0, uptime_duration)

        return (current_uptime / total_time) * 100.0

    def to_dict(self, current_time: datetime | None = None) -> dict[str, Any]:
        return {
            "node_id": self.node_id,
            "connected": self.connected,
            "last_connected": self.last_connected.isoformat() if self.last_connected else None,
            "last_disconnected": self.last_disconnected.isoformat()
            if self.last_disconnected
            else None,
            "connection_attempts": self.connection_attempts,
            "successful_connections": self.successful_connections,
            "uptime_percentage": round(self.get_uptime_percentage(current_time), 2),
            "tracking_started": self.tracking_started.isoformat()
            if self.tracking_started
            else None,
            "retry_delay": self.retry_delay,
        }

    def get_next_retry_delay(self) -> float:
        """Get the next retry delay using exponential backoff."""
        current_delay = self.retry_delay
        # Increase delay for next retry
        self.retry_delay = min(self.retry_delay * self.retry_delay_multiplier, self.retry_delay_max)
        return current_delay

Attributes

connected = False instance-attribute

connection_attempts = 0 instance-attribute

current_session_start: datetime | None = None instance-attribute

grace_period_seconds = grace_period_seconds instance-attribute

last_connected: datetime | None = None instance-attribute

last_disconnected: datetime | None = None instance-attribute

node_id = node_id instance-attribute

retry_delay: float = 4.0 instance-attribute

retry_delay_max: float = 3600.0 instance-attribute

retry_delay_multiplier: float = 1.5 instance-attribute

successful_connections = 0 instance-attribute

total_uptime_seconds = 0.0 instance-attribute

tracking_started = tracking_started or datetime.now(UTC) instance-attribute

Methods:

__init__(node_id: str, tracking_started: datetime | None = None, grace_period_seconds: int = 0) -> None

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

def __init__(
    self,
    node_id: str,
    tracking_started: datetime | None = None,
    grace_period_seconds: int = 0,
) -> None:
    self.node_id = node_id
    self.connected = False
    self.last_connected: datetime | None = None
    self.last_disconnected: datetime | None = None
    self.connection_attempts = 0
    self.successful_connections = 0
    self.total_uptime_seconds = 0.0
    self.current_session_start: datetime | None = None
    self.tracking_started = tracking_started or datetime.now(UTC)
    self.grace_period_seconds = grace_period_seconds
    # Exponential backoff state
    self.retry_delay: float = 4.0  # Initial retry delay in seconds
    self.retry_delay_max: float = 3600.0  # Max retry delay (1 hour)
    self.retry_delay_multiplier: float = 1.5

get_next_retry_delay() -> float

Get the next retry delay using exponential backoff.

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

def get_next_retry_delay(self) -> float:
    """Get the next retry delay using exponential backoff."""
    current_delay = self.retry_delay
    # Increase delay for next retry
    self.retry_delay = min(self.retry_delay * self.retry_delay_multiplier, self.retry_delay_max)
    return current_delay

get_uptime_percentage(current_time: datetime | None = None) -> float

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

def get_uptime_percentage(self, current_time: datetime | None = None) -> float:
    if not self.tracking_started:
        return 0.0
    now = current_time or datetime.now(UTC)
    elapsed = (now - self.tracking_started).total_seconds()

    # If we're still in grace period, return 100% uptime
    if elapsed < self.grace_period_seconds:
        return 100.0

    # Calculate total time excluding grace period
    total_time = elapsed - self.grace_period_seconds
    if total_time == 0:
        return 0.0

    # Calculate uptime, but only count time after grace period ends
    grace_end = self.tracking_started.timestamp() + self.grace_period_seconds
    current_uptime = self.total_uptime_seconds

    if self.connected and self.current_session_start:
        # Only count uptime after grace period ended
        session_start_ts = self.current_session_start.timestamp()
        if session_start_ts < grace_end:
            # Session started during grace period, only count time after grace ended
            uptime_duration = now.timestamp() - grace_end
        else:
            # Session started after grace period
            uptime_duration = (now - self.current_session_start).total_seconds()
        current_uptime += max(0, uptime_duration)

    return (current_uptime / total_time) * 100.0

mark_connected(current_time: datetime | None = None) -> None

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

def mark_connected(self, current_time: datetime | None = None) -> None:
    now = current_time or datetime.now(UTC)
    self.connected = True
    self.last_connected = now
    self.current_session_start = now
    self.successful_connections += 1
    # Reset retry delay on successful connection
    self.retry_delay = 4.0

mark_disconnected(current_time: datetime | None = None) -> None

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

def mark_disconnected(self, current_time: datetime | None = None) -> None:
    now = current_time or datetime.now(UTC)
    if self.connected and self.current_session_start:
        # Only count uptime after grace period
        grace_end_ts = self.tracking_started.timestamp() + self.grace_period_seconds
        session_start_ts = self.current_session_start.timestamp()
        now_ts = now.timestamp()

        # Calculate the actual uptime to record (only after grace period)
        if now_ts > grace_end_ts:
            # Some or all of the session is after grace period
            counted_start = max(session_start_ts, grace_end_ts)
            session_duration = now_ts - counted_start
            self.total_uptime_seconds += max(0, session_duration)

    self.connected = False
    self.last_disconnected = now
    self.current_session_start = None

to_dict(current_time: datetime | None = None) -> dict[str, Any]

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

def to_dict(self, current_time: datetime | None = None) -> dict[str, Any]:
    return {
        "node_id": self.node_id,
        "connected": self.connected,
        "last_connected": self.last_connected.isoformat() if self.last_connected else None,
        "last_disconnected": self.last_disconnected.isoformat()
        if self.last_disconnected
        else None,
        "connection_attempts": self.connection_attempts,
        "successful_connections": self.successful_connections,
        "uptime_percentage": round(self.get_uptime_percentage(current_time), 2),
        "tracking_started": self.tracking_started.isoformat()
        if self.tracking_started
        else None,
        "retry_delay": self.retry_delay,
    }

OrderbookAggregator

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

class OrderbookAggregator:
    def __init__(
        self,
        directory_nodes: list[tuple[str, int]],
        network: str,
        mempool_api_url: str,
        mempool_api_use_tor: bool = True,
        socks_host: str = "127.0.0.1",
        socks_port: int = 9050,
        timeout: float = 30.0,
        max_retry_attempts: int = 3,
        retry_delay: float = 5.0,
        max_message_size: int = 2097152,
        uptime_grace_period: int = 60,
        stream_isolation: bool = False,
        blockchain_backend: BlockchainBackend | None = None,
    ) -> None:
        self.directory_nodes = directory_nodes
        self.network = network
        self.socks_host = socks_host
        self.socks_port = socks_port
        self.stream_isolation = stream_isolation
        self.timeout = timeout
        self.mempool_api_url = mempool_api_url
        self.mempool_api_use_tor = mempool_api_use_tor
        self.max_retry_attempts = max_retry_attempts
        self.retry_delay = retry_delay
        self.max_message_size = max_message_size
        self.uptime_grace_period = uptime_grace_period
        self.blockchain_backend = blockchain_backend

        # Build the optional mempool proxy URL and pre-compute isolation credentials.
        self._dir_username: str | None = None
        self._dir_password: str | None = None
        self._hc_username: str | None = None
        self._hc_password: str | None = None
        socks_proxy: str | None = None
        if stream_isolation:
            from jmcore.tor_isolation import (  # noqa: PLC0415
                IsolationCategory,
                build_isolated_proxy_url,
                get_isolation_credentials,
            )

            if mempool_api_use_tor:
                socks_proxy = build_isolated_proxy_url(
                    socks_host, socks_port, IsolationCategory.MEMPOOL
                )
            dir_c = get_isolation_credentials(IsolationCategory.DIRECTORY)
            self._dir_username = dir_c.username
            self._dir_password = dir_c.password
            hc_c = get_isolation_credentials(IsolationCategory.HEALTH_CHECK)
            self._hc_username = hc_c.username
            self._hc_password = hc_c.password
        elif mempool_api_use_tor:
            socks_proxy = f"socks5h://{socks_host}:{socks_port}"
        self.mempool_api: MempoolAPI | None = None
        self._mempool_test_task: asyncio.Task[Any] | None = None
        if mempool_api_url:
            mempool_timeout = 60.0
            if mempool_api_use_tor:
                logger.info("Mempool API configured with Tor routing")
            else:
                logger.info("Mempool API configured for direct access")
            self.mempool_api = MempoolAPI(
                base_url=mempool_api_url,
                socks_proxy=socks_proxy,
                timeout=mempool_timeout,
                trust_env=False,
            )
            self._mempool_test_task = asyncio.create_task(self._test_mempool_connection())
        else:
            logger.info("Mempool API disabled by configuration; external mempool lookups are off")
        self.current_orderbook: OrderBook = OrderBook()
        self._lock = asyncio.Lock()
        self.clients: dict[str, DirectoryClient] = {}
        self.listener_tasks: list[asyncio.Task[Any]] = []
        self._bond_calculation_task: asyncio.Task[Any] | None = None
        self._bond_queue: asyncio.Queue[OrderBook] = asyncio.Queue()
        self._bond_cache: dict[str, FidelityBond] = {}
        self._last_offers_hash: int = 0
        self._mempool_semaphore = asyncio.Semaphore(5)
        self.node_statuses: dict[str, DirectoryNodeStatus] = {}
        self._retry_tasks: list[asyncio.Task[Any]] = []

        # Maker health checker for direct reachability verification
        self.health_checker = MakerHealthChecker(
            network=network,
            socks_host=socks_host,
            socks_port=socks_port,
            timeout=timeout,
            check_interval=600.0,  # Check each maker at most once per 10 minutes
            max_concurrent_checks=5,
            socks_username=self._hc_username,
            socks_password=self._hc_password,
        )

        for onion_address, port in directory_nodes:
            node_id = f"{onion_address}:{port}"
            self.node_statuses[node_id] = DirectoryNodeStatus(
                node_id, grace_period_seconds=uptime_grace_period
            )

    def _handle_client_disconnect(self, onion_address: str, port: int) -> None:
        node_id = f"{onion_address}:{port}"
        client = self.clients.pop(node_id, None)
        if client:
            client.stop()
        self._schedule_reconnect(onion_address, port)

    def _schedule_reconnect(self, onion_address: str, port: int) -> None:
        node_id = f"{onion_address}:{port}"
        self._retry_tasks = [task for task in self._retry_tasks if not task.done()]
        if any(task.get_name() == f"retry:{node_id}" for task in self._retry_tasks):
            logger.debug(f"Retry already scheduled for {node_id}")
            return
        retry_task = asyncio.create_task(self._retry_failed_connection(onion_address, port))
        retry_task.set_name(f"retry:{node_id}")
        self._retry_tasks.append(retry_task)
        logger.info(f"Scheduled retry task for {node_id}")

    async def fetch_from_directory(
        self, onion_address: str, port: int
    ) -> tuple[list[Offer], list[FidelityBond], str]:
        node_id = f"{onion_address}:{port}"
        logger.info(f"Fetching orderbook from directory: {node_id}")
        client = DirectoryClient(
            onion_address,
            port,
            self.network,
            socks_host=self.socks_host,
            socks_port=self.socks_port,
            timeout=self.timeout,
            max_message_size=self.max_message_size,
            socks_username=self._dir_username,
            socks_password=self._dir_password,
        )
        try:
            await client.connect()
            offers, bonds = await client.fetch_orderbooks()

            for offer in offers:
                offer.directory_node = node_id
            for bond in bonds:
                bond.directory_node = node_id

            return offers, bonds, node_id
        except Exception as e:
            logger.error(f"Failed to fetch from directory {node_id}: {e}")
            return [], [], node_id
        finally:
            await client.close()

    async def update_orderbook(self) -> OrderBook:
        tasks = [
            self.fetch_from_directory(onion_address, port)
            for onion_address, port in self.directory_nodes
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        new_orderbook = OrderBook(timestamp=datetime.now(UTC))

        for result in results:
            if isinstance(result, BaseException):
                logger.error(f"Directory fetch failed: {result}")
                continue

            offers, bonds, node_id = result
            if offers or bonds:
                new_orderbook.add_offers(offers, node_id)
                new_orderbook.add_fidelity_bonds(bonds, node_id)

        await self._calculate_bond_values(new_orderbook)

        async with self._lock:
            self.current_orderbook = new_orderbook

        logger.info(
            f"Updated orderbook: {len(new_orderbook.offers)} offers, "
            f"{len(new_orderbook.fidelity_bonds)} bonds from "
            f"{len(new_orderbook.directory_nodes)} directory nodes"
        )

        return new_orderbook

    async def get_orderbook(self) -> OrderBook:
        async with self._lock:
            return self.current_orderbook

    async def _background_bond_calculator(self) -> None:
        while True:
            try:
                orderbook = await self._bond_queue.get()
                await self._calculate_bond_values(orderbook)
                for offer in orderbook.offers:
                    if offer.fidelity_bond_data:
                        matching_bonds = [
                            b
                            for b in orderbook.fidelity_bonds
                            if b.counterparty == offer.counterparty
                            and b.utxo_txid == offer.fidelity_bond_data.get("utxo_txid")
                        ]
                        if matching_bonds and matching_bonds[0].bond_value is not None:
                            offer.fidelity_bond_value = matching_bonds[0].bond_value
                logger.debug("Background bond calculation completed")
            except Exception as e:
                logger.error(f"Error in background bond calculator: {e}")

    async def _periodic_directory_connection_status(self) -> None:
        """Background task to periodically log directory connection status.

        This runs every 10 minutes to provide visibility into orderbook
        connectivity. Shows:
        - Total directory servers configured
        - Currently connected servers with uptime percentage
        - Disconnected servers (if any)
        """
        # First log after 5 minutes (give time for initial connection)
        await asyncio.sleep(300)

        while True:
            try:
                total_servers = len(self.directory_nodes)
                connected_nodes = []
                disconnected_nodes = []

                for node_id, status in self.node_statuses.items():
                    if status.connected:
                        uptime_pct = status.get_uptime_percentage()
                        connected_nodes.append(f"{node_id} ({uptime_pct:.1f}% uptime)")
                    else:
                        disconnected_nodes.append(node_id)

                connected_count = len(connected_nodes)

                if disconnected_nodes:
                    disconnected_str = ", ".join(disconnected_nodes[:5])
                    if len(disconnected_nodes) > 5:
                        disconnected_str += f", ... and {len(disconnected_nodes) - 5} more"
                    logger.warning(
                        f"Directory connection status: {connected_count}/{total_servers} connected. "
                        f"Disconnected: [{disconnected_str}]"
                    )
                else:
                    connected_str = ", ".join(connected_nodes[:5])
                    if len(connected_nodes) > 5:
                        connected_str += f", ... and {len(connected_nodes) - 5} more"
                    logger.info(
                        f"Directory connection status: {connected_count}/{total_servers} connected "
                        f"[{connected_str}]"
                    )

                # Log again in 10 minutes
                await asyncio.sleep(600)

            except asyncio.CancelledError:
                logger.info("Directory connection status task cancelled")
                break
            except Exception as e:
                logger.error(f"Error in directory connection status task: {e}")
                await asyncio.sleep(600)

        logger.info("Directory connection status task stopped")

    async def _periodic_peerlist_refresh(self) -> None:
        """Background task to periodically refresh peerlists and clean up stale offers.

        Runs every 5 minutes and applies a per-directory trust model:

        1. Refresh each connected directory's peerlist.
        2. For directories that support GETPEERLIST and responded successfully,
           remove offers from that client whose nicks are NOT in that
           directory's current peerlist. This honours explicit disconnect
           announcements (";D") as well as silent drops the directory has
           since pruned from its registry.
        3. For directories without GETPEERLIST support (e.g. legacy reference
           implementation), fall back to age-based cleanup via
           ``cleanup_stale_offers``.
        4. After peerlist refresh, run feature discovery for makers we still
           don't have features for. Direct probes never remove offers; they
           are strictly informational / for feature population.

        This replaces the previous cross-directory union model, where an
        offer was kept as long as ANY directory still reported the maker.
        That model retained orphan offers for makers that had disconnected
        from a directory but were still listed by another, which inflated
        per-directory offer counts.
        """
        # Initial wait to let connections stabilize
        await asyncio.sleep(120)

        refresh_interval = 300.0
        stale_offer_max_age = 1800.0

        while True:
            try:
                total_removed = 0
                refreshed = 0
                refresh_failures = 0
                fallback_cleanups = 0

                for node_id, client in list(self.clients.items()):
                    refresh_ok = False
                    try:
                        await client.get_peerlist_with_features()
                        refresh_ok = True
                        refreshed += 1
                    except Exception as e:
                        logger.debug(f"Failed to refresh peerlist from {node_id}: {e}")
                        refresh_failures += 1

                    if refresh_ok and client._peerlist_supported:
                        # Per-directory trust: drop offers from nicks the
                        # directory no longer reports as connected.
                        active_nicks = client.get_active_nicks()
                        client_nicks = {key[0] for key in client.offers}
                        stale_nicks = client_nicks - active_nicks
                        for nick in stale_nicks:
                            total_removed += client.remove_offers_for_nick(nick)
                        logger.debug(
                            f"Directory {node_id}: {len(active_nicks)} active nicks, "
                            f"removed offers for {len(stale_nicks)} disconnected nicks"
                        )
                    elif client._peerlist_supported is False:
                        # Reference implementation fallback: prune by age.
                        removed = client.cleanup_stale_offers(max_age_seconds=stale_offer_max_age)
                        if removed:
                            fallback_cleanups += removed
                            logger.debug(
                                f"Directory {node_id}: fallback removed {removed} stale offers "
                                f"(no GETPEERLIST support)"
                            )

                if refreshed or refresh_failures:
                    logger.info(
                        f"Peerlist refresh: {refreshed} directories refreshed, "
                        f"{refresh_failures} failed, "
                        f"{total_removed} offers pruned from disconnected nicks, "
                        f"{fallback_cleanups} stale offers pruned by age"
                    )

                # After peerlist refresh, populate features for makers we're
                # still missing. This is informational and never removes offers.
                try:
                    await self._check_makers_without_features()
                except Exception as e:
                    logger.debug(f"Error checking makers without features: {e}")

                await asyncio.sleep(refresh_interval)

            except asyncio.CancelledError:
                logger.info("Peerlist refresh task cancelled")
                break
            except Exception as e:
                logger.error(f"Error in peerlist refresh task: {e}")
                await asyncio.sleep(refresh_interval)

        logger.info("Peerlist refresh task stopped")

    def _prioritize_makers_for_scan(
        self,
        makers: list[tuple[str, str]],
    ) -> list[tuple[str, str]]:
        """Sort makers for scanning: bonded (desc bond value), then bondless (asc fee).

        During sybil attacks the orderbook may contain thousands of fake offers.
        Scanning them in arbitrary order wastes the limited concurrent-check
        slots on attackers while legitimate makers never get their features
        discovered.  By processing bonded makers first (highest bond value first)
        and then bondless makers in ascending fee order we ensure the most
        trustworthy and cheapest liquidity is scanned before the spam.

        Args:
            makers: List of (nick, location) tuples to prioritize.

        Returns:
            Sorted list of (nick, location) tuples.
        """
        # Build lookup: nick -> (best_bond_value, lowest_fee_rate)
        # We scan all clients for the best bond value and lowest fee for each nick.
        nick_bond_value: dict[str, int] = {}
        nick_fee_rate: dict[str, float] = {}

        for client in self.clients.values():
            for key, offer_ts in client.offers.items():
                nick = key[0]
                offer = offer_ts.offer

                # Track best (highest) bond value
                bond_val = offer.fidelity_bond_value
                if bond_val > nick_bond_value.get(nick, 0):
                    nick_bond_value[nick] = bond_val

                # Track lowest fee rate for ordering bondless makers.
                # Normalise both offer types to a comparable float:
                #   - Relative offers: the cjfee string is already a rate
                #     (e.g. "0.003" = 0.3%)
                #   - Absolute offers: express as rate relative to maxsize
                #     so that we have a rough comparison metric.
                try:
                    if offer.is_absolute_fee():
                        fee_rate = float(int(offer.cjfee)) / max(offer.maxsize, 1)
                    else:
                        fee_rate = float(offer.cjfee)
                except (ValueError, TypeError, ZeroDivisionError):
                    fee_rate = float("inf")

                if fee_rate < nick_fee_rate.get(nick, float("inf")):
                    nick_fee_rate[nick] = fee_rate

        def sort_key(item: tuple[str, str]) -> tuple[int, float, float]:
            nick = item[0]
            bond = nick_bond_value.get(nick, 0)
            fee = nick_fee_rate.get(nick, float("inf"))
            # Primary: bonded first (group 0) vs bondless (group 1)
            # Secondary for bonded: descending bond value (negate)
            # Secondary for bondless: ascending fee rate
            if bond > 0:
                return (0, -bond, fee)
            return (1, 0, fee)

        sorted_makers = sorted(makers, key=sort_key)

        # Log a brief summary
        bonded_count = sum(1 for n, _ in sorted_makers if nick_bond_value.get(n, 0) > 0)
        logger.info(
            f"Scan priority: {bonded_count} bonded makers first, "
            f"{len(sorted_makers) - bonded_count} bondless (fee-ascending)"
        )
        return sorted_makers

    async def _check_makers_without_features(self) -> None:
        """Check makers that have offers but no features discovered yet.

        This is called after peerlist refresh to ensure we discover features
        for makers whose features weren't included in the peerlist (e.g., from
        reference implementation directories that don't support peerlist_features).

        Makers are scanned in priority order: bonded makers first (descending
        bond value), then bondless makers in ascending fee order.  This ensures
        legitimate makers get their features discovered before sybil spam.
        """
        makers_to_check: list[tuple[str, str]] = []

        for _node_id, client in self.clients.items():
            for key, _offer_ts in client.offers.items():
                nick = key[0]
                # Check if this peer has no features
                peer_features = client.peer_features.get(nick, {})
                if not peer_features:
                    # Try to find location
                    location = client._active_peers.get(nick)
                    if location and location != "NOT-SERVING-ONION":
                        makers_to_check.append((nick, location))

        # Deduplicate by location (keep first nick per location after priority sort)
        unique_makers = {loc: (nick, loc) for nick, loc in makers_to_check}
        makers_list = self._prioritize_makers_for_scan(list(unique_makers.values()))

        if not makers_list:
            return

        logger.info(f"Feature discovery: Checking {len(makers_list)} makers without features")

        # Check makers and extract features from handshake
        health_statuses = await self.health_checker.check_makers_batch(makers_list, force=True)

        # Update features in directory clients' peer_features cache
        features_discovered = 0
        for _location, status in health_statuses.items():
            if status.reachable and status.features.features:
                features_dict = status.features.to_dict()
                features_discovered += 1
                # Update all clients with this peer's features using merge
                # (never overwrite/downgrade existing features)
                for client in self.clients.values():
                    if status.nick in client._active_peers:
                        client._merge_peer_features(status.nick, features_dict)
                        # Also update cached offers
                        client._update_offer_features(status.nick, features_dict)

        if features_discovered > 0:
            logger.info(f"Feature discovery: Discovered features for {features_discovered} makers")

    async def _periodic_maker_health_check(self) -> None:
        """Background task to periodically probe makers for feature discovery.

        This task is purely informational and NEVER removes offers. Offer
        presence is governed exclusively by the directories' peerlists (see
        ``_periodic_peerlist_refresh``).

        It exists so we can still learn maker capabilities (e.g.
        ``neutrino_compat``) when a directory does not advertise
        ``peerlist_features`` (notably the legacy reference implementation).
        On healthy jm-ng directories the peerlist already carries features
        and this probe will mostly be a no-op since makers without missing
        features are skipped by ``_check_makers_without_features``.
        """
        # Initial wait to let orderbook populate; feature discovery also
        # runs after each peerlist refresh, so this loop is the slow path.
        await asyncio.sleep(120)

        check_interval = 900.0

        while True:
            try:
                await self._check_makers_without_features()
                await asyncio.sleep(check_interval)
            except asyncio.CancelledError:
                logger.info("Maker feature-discovery task cancelled")
                break
            except Exception as e:
                logger.error(f"Error in maker feature-discovery task: {e}")
                await asyncio.sleep(check_interval)

        logger.info("Maker feature-discovery task stopped")

    async def _connect_to_node(self, onion_address: str, port: int) -> DirectoryClient | None:
        node_id = f"{onion_address}:{port}"
        status = self.node_statuses[node_id]
        status.connection_attempts += 1

        logger.info(f"Connecting to directory: {node_id}")

        def on_disconnect() -> None:
            logger.info(f"Directory node {node_id} disconnected")
            status.mark_disconnected()
            self._handle_client_disconnect(onion_address, port)

        client = DirectoryClient(
            onion_address,
            port,
            self.network,
            socks_host=self.socks_host,
            socks_port=self.socks_port,
            timeout=self.timeout,
            max_message_size=self.max_message_size,
            on_disconnect=on_disconnect,
            socks_username=self._dir_username,
            socks_password=self._dir_password,
        )

        try:
            await client.connect()
            status.mark_connected()
            logger.info(f"Successfully connected to directory: {node_id} (our nick: {client.nick})")
            return client

        except Exception as e:
            logger.warning(f"Connection to directory {node_id} failed: {e}")
            await client.close()
            status.mark_disconnected()
            self._schedule_reconnect(onion_address, port)
            return None

    async def _retry_failed_connection(self, onion_address: str, port: int) -> None:
        """Retry connecting to a directory with exponential backoff."""
        node_id = f"{onion_address}:{port}"
        status = self.node_statuses[node_id]

        while True:
            # Get next retry delay with exponential backoff
            retry_delay = status.get_next_retry_delay()
            logger.info(f"Waiting {retry_delay:.1f}s before retrying connection to {node_id}")
            await asyncio.sleep(retry_delay)

            if node_id in self.clients:
                logger.debug(f"Node {node_id} already connected, stopping retry")
                return

            logger.info(f"Retrying connection to directory {node_id}...")
            client = await self._connect_to_node(onion_address, port)

            if client:
                self.clients[node_id] = client
                task = asyncio.create_task(client.listen_continuously())
                self.listener_tasks.append(task)
                logger.info(f"Successfully reconnected to directory: {node_id}")
                return

    async def start_continuous_listening(self) -> None:
        logger.info("Starting continuous listening on all directory nodes")

        self._bond_calculation_task = asyncio.create_task(self._background_bond_calculator())

        # Start periodic directory connection status logging task
        status_task = asyncio.create_task(self._periodic_directory_connection_status())
        self.listener_tasks.append(status_task)

        # Start periodic peerlist refresh task for cleanup
        peerlist_task = asyncio.create_task(self._periodic_peerlist_refresh())
        self.listener_tasks.append(peerlist_task)

        # Start periodic feature-discovery task (direct onion handshake to learn
        # maker capabilities when directories don't advertise peerlist_features).
        # This task NEVER removes offers; offer presence is governed by peerlist.
        health_check_task = asyncio.create_task(self._periodic_maker_health_check())
        self.listener_tasks.append(health_check_task)

        connection_tasks = [
            self._connect_to_node(onion_address, port)
            for onion_address, port in self.directory_nodes
        ]

        clients = await asyncio.gather(*connection_tasks, return_exceptions=True)

        for (onion_address, port), result in zip(self.directory_nodes, clients, strict=True):
            node_id = f"{onion_address}:{port}"

            if isinstance(result, BaseException):
                logger.error(f"Connection to {node_id} raised exception: {result}")
                retry_task = asyncio.create_task(self._retry_failed_connection(onion_address, port))
                self._retry_tasks.append(retry_task)
                logger.info(f"Scheduled retry task for {node_id}")
            elif result is not None:
                self.clients[node_id] = result
                task = asyncio.create_task(result.listen_continuously())
                self.listener_tasks.append(task)
                logger.info(f"Started listener task for {node_id}")
            else:
                retry_task = asyncio.create_task(self._retry_failed_connection(onion_address, port))
                self._retry_tasks.append(retry_task)
                logger.info(f"Scheduled retry task for {node_id}")

        # Start early feature discovery task - runs once after initial connections settle
        early_feature_task = asyncio.create_task(self._early_feature_discovery())
        self.listener_tasks.append(early_feature_task)

    async def _early_feature_discovery(self) -> None:
        """Run feature discovery shortly after startup to populate features quickly.

        This is a one-shot task that runs after a brief delay to allow initial
        offers to be received, then checks makers without features.
        """
        try:
            # Wait for initial offers to arrive (30 seconds should be enough for !orderbook responses)
            await asyncio.sleep(30)
            logger.info("Running early feature discovery for makers without features")
            await self._check_makers_without_features()
        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.debug(f"Early feature discovery error: {e}")

    async def stop_listening(self) -> None:
        logger.info("Stopping all directory listeners")

        if self._bond_calculation_task:
            self._bond_calculation_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._bond_calculation_task

        for task in self._retry_tasks:
            task.cancel()

        if self._retry_tasks:
            await asyncio.gather(*self._retry_tasks, return_exceptions=True)

        for client in self.clients.values():
            client.stop()

        for task in self.listener_tasks:
            task.cancel()

        if self.listener_tasks:
            await asyncio.gather(*self.listener_tasks, return_exceptions=True)

        for node_id, client in self.clients.items():
            self.node_statuses[node_id].mark_disconnected()
            await client.close()

        self.clients.clear()
        self.listener_tasks.clear()
        self._retry_tasks.clear()

    async def get_live_orderbook(self, calculate_bonds: bool = True) -> OrderBook:
        """Get the live orderbook with deduplication and bond value calculation.

        This method implements several key cleanup mechanisms:
        1. Deduplicates offers by bond UTXO - if multiple nicks use the same bond,
           only the most recently received offer is kept
        2. Deduplicates bonds by UTXO key
        3. Links bonds to offers and calculates bond values

        Returns:
            OrderBook with deduplicated offers and calculated bond values
        """
        orderbook = OrderBook(timestamp=datetime.now(UTC))
        all_offers_with_timestamps = self._collect_live_offers_and_bonds(orderbook)
        deduplicated_offers = self._deduplicate_bond_backed_offers(all_offers_with_timestamps)
        self._merge_offers_into_orderbook(orderbook, deduplicated_offers)
        self._deduplicate_bonds(orderbook)

        if calculate_bonds:
            self._apply_bond_cache(orderbook)
            await self._calculate_bond_values(orderbook)
            self._update_bond_cache(orderbook)
            self._link_bonds_to_offers(orderbook)

        self._populate_bond_directory_nodes(orderbook)
        self._enrich_offers_with_health_status(orderbook)
        return orderbook

    def _collect_live_offers_and_bonds(
        self, orderbook: OrderBook
    ) -> list[tuple[Offer, float, str | None, str]]:
        all_offers_with_timestamps: list[tuple[Offer, float, str | None, str]] = []
        total_offers_from_directories = 0
        total_bonds_from_directories = 0

        for node_id, client in self.clients.items():
            offers_with_ts = client.get_offers_with_timestamps()
            bonds = client.get_current_bonds()
            total_offers_from_directories += len(offers_with_ts)
            total_bonds_from_directories += len(bonds)

            offers_with_bonds_count = sum(
                1 for offer_ts in offers_with_ts if offer_ts.bond_utxo_key
            )
            logger.debug(
                f"Directory {node_id}: {len(offers_with_ts)} offers "
                f"({offers_with_bonds_count} with bonds), {len(bonds)} bonds"
            )

            for offer_ts in offers_with_ts:
                offer = offer_ts.offer
                offer.directory_node = node_id
                all_offers_with_timestamps.append(
                    (offer, offer_ts.received_at, offer_ts.bond_utxo_key, node_id)
                )

            for bond in bonds:
                bond.directory_node = node_id
            orderbook.add_fidelity_bonds(bonds, node_id)

        logger.info(
            f"Collected {total_offers_from_directories} offers and "
            f"{total_bonds_from_directories} bonds from {len(self.clients)} directories"
        )
        return all_offers_with_timestamps

    def _deduplicate_bond_backed_offers(
        self, all_offers_with_timestamps: list[tuple[Offer, float, str | None, str]]
    ) -> list[Offer]:
        bond_oid_to_best_offer: dict[tuple[str, int], tuple[Offer, float, list[str], str]] = {}
        offers_without_bond: list[Offer] = []

        total_offers_processed = len(all_offers_with_timestamps)
        offers_with_bonds = 0
        offers_without_bonds = 0
        bond_replacements = 0

        for offer, timestamp, bond_utxo_key, _node_id in all_offers_with_timestamps:
            if not bond_utxo_key:
                offers_without_bonds += 1
                logger.debug(
                    f"Offer without bond from {offer.counterparty} (oid={offer.oid}, "
                    f"ordertype={offer.ordertype.value})"
                )
                offers_without_bond.append(offer)
                continue

            offers_with_bonds += 1
            dedup_key = (bond_utxo_key, offer.oid)
            existing = bond_oid_to_best_offer.get(dedup_key)
            if existing is None:
                directory_nodes = [offer.directory_node] if offer.directory_node else []
                logger.debug(
                    f"Bond deduplication: First offer for bond {bond_utxo_key[:20]}... "
                    f"from {offer.counterparty} (oid={offer.oid})"
                )
                bond_oid_to_best_offer[dedup_key] = (
                    offer,
                    timestamp,
                    directory_nodes,
                    offer.counterparty,
                )
                continue

            old_offer, old_timestamp, directory_nodes, old_counterparty = existing
            if old_counterparty == offer.counterparty:
                if offer.directory_node and offer.directory_node not in directory_nodes:
                    directory_nodes.append(offer.directory_node)
                if timestamp > old_timestamp:
                    bond_oid_to_best_offer[dedup_key] = (
                        offer,
                        timestamp,
                        directory_nodes,
                        offer.counterparty,
                    )
                logger.debug(
                    f"Bond deduplication: Same maker {offer.counterparty} (oid={offer.oid}) "
                    f"seen on {len(directory_nodes)} directories for bond {bond_utxo_key[:20]}..."
                )
                continue

            time_diff = timestamp - old_timestamp
            if abs(time_diff) < 60:
                # Routine and self-healing: the same bond seen under two nicks
                # within the skew window keeps the existing offer. This recurs
                # on every aggregation cycle for as long as both nicks are
                # announced, so it is diagnostic detail, not a warning.
                logger.debug(
                    f"Bond deduplication: Ignoring potential duplicate from {offer.counterparty} "
                    f"(oid={offer.oid}) - same bond as {old_offer.counterparty} "
                    f"(oid={old_offer.oid}) with only {abs(time_diff):.1f}s difference "
                    f"[bond UTXO: {bond_utxo_key[:20]}...]. Likely clock skew."
                )
                continue
            if time_diff <= 0:
                continue

            bond_replacements += 1
            logger.debug(
                f"Bond deduplication: Replacing offer from {old_offer.counterparty} "
                f"(oid={old_offer.oid}) with {offer.counterparty} (oid={offer.oid}) "
                f"[same bond UTXO: {bond_utxo_key[:20]}..., age_diff={time_diff:.1f}s]"
            )
            new_directory_nodes = [offer.directory_node] if offer.directory_node else []
            bond_oid_to_best_offer[dedup_key] = (
                offer,
                timestamp,
                new_directory_nodes,
                offer.counterparty,
            )

        if bond_replacements > 0:
            logger.info(
                f"Bond deduplication: Replaced {bond_replacements} offers from makers who "
                f"restarted with same bond. Result: {len(bond_oid_to_best_offer)} unique bond offers + "
                f"{len(offers_without_bond)} non-bond offers"
            )
        else:
            logger.debug(
                f"Bond deduplication: Processed {total_offers_processed} offers "
                f"({offers_with_bonds} with bonds, {offers_without_bonds} without bonds). "
                f"Result: {len(bond_oid_to_best_offer)} unique bond offers + "
                f"{len(offers_without_bond)} non-bond offers"
            )

        deduplicated_offers: list[Offer] = []
        for offer, _ts, directory_nodes, _counterparty in bond_oid_to_best_offer.values():
            offer.directory_nodes = directory_nodes
            deduplicated_offers.append(offer)
        deduplicated_offers.extend(offers_without_bond)
        return deduplicated_offers

    def _merge_offers_into_orderbook(
        self, orderbook: OrderBook, deduplicated_offers: list[Offer]
    ) -> None:
        offer_key_to_offer: dict[tuple[str, int], Offer] = {}
        for offer in deduplicated_offers:
            key = (offer.counterparty, offer.oid)
            if key not in offer_key_to_offer:
                if not offer.directory_nodes and offer.directory_node:
                    offer.directory_nodes = [offer.directory_node]
                offer_key_to_offer[key] = offer
                continue

            existing_offer = offer_key_to_offer[key]
            if offer.directory_node and offer.directory_node not in existing_offer.directory_nodes:
                existing_offer.directory_nodes.append(offer.directory_node)
            for feature, value in offer.features.items():
                if value and not existing_offer.features.get(feature):
                    existing_offer.features[feature] = value

        for offer in offer_key_to_offer.values():
            orderbook.offers.append(offer)
            for node in offer.directory_nodes:
                if node not in orderbook.directory_nodes:
                    orderbook.directory_nodes.append(node)

    def _deduplicate_bonds(self, orderbook: OrderBook) -> None:
        unique_bonds: dict[str, FidelityBond] = {}
        for bond in orderbook.fidelity_bonds:
            cache_key = f"{bond.utxo_txid}:{bond.utxo_vout}"
            if cache_key not in unique_bonds:
                if bond.directory_node:
                    bond.directory_nodes = [bond.directory_node]
                unique_bonds[cache_key] = bond
                continue
            existing_bond = unique_bonds[cache_key]
            if bond.directory_node and bond.directory_node not in existing_bond.directory_nodes:
                existing_bond.directory_nodes.append(bond.directory_node)
        orderbook.fidelity_bonds = list(unique_bonds.values())

    def _apply_bond_cache(self, orderbook: OrderBook) -> None:
        cached_count = 0
        for bond in orderbook.fidelity_bonds:
            cache_key = f"{bond.utxo_txid}:{bond.utxo_vout}"
            if cache_key not in self._bond_cache:
                continue
            cached_bond = self._bond_cache[cache_key]
            bond.bond_value = cached_bond.bond_value
            bond.amount = cached_bond.amount
            bond.utxo_confirmation_timestamp = cached_bond.utxo_confirmation_timestamp
            cached_count += 1
        if cached_count > 0:
            logger.debug(f"Loaded {cached_count}/{len(orderbook.fidelity_bonds)} bonds from cache")

    def _update_bond_cache(self, orderbook: OrderBook) -> None:
        for bond in orderbook.fidelity_bonds:
            if bond.bond_value is not None:
                cache_key = f"{bond.utxo_txid}:{bond.utxo_vout}"
                self._bond_cache[cache_key] = bond

    def _link_bonds_to_offers(self, orderbook: OrderBook) -> None:
        for offer in orderbook.offers:
            if offer.fidelity_bond_data:
                matching_bonds = [
                    bond
                    for bond in orderbook.fidelity_bonds
                    if bond.counterparty == offer.counterparty
                    and bond.utxo_txid == offer.fidelity_bond_data.get("utxo_txid")
                ]
                if matching_bonds and matching_bonds[0].bond_value is not None:
                    offer.fidelity_bond_value = matching_bonds[0].bond_value
                continue

            matching_bonds = [
                bond for bond in orderbook.fidelity_bonds if bond.counterparty == offer.counterparty
            ]
            if not matching_bonds:
                continue

            bond = max(
                matching_bonds,
                key=lambda candidate: (
                    candidate.bond_value if candidate.bond_value is not None else 0
                ),
            )
            if bond.fidelity_bond_data:
                offer.fidelity_bond_data = bond.fidelity_bond_data
                if bond.bond_value is not None:
                    offer.fidelity_bond_value = bond.bond_value
            logger.debug(
                f"Linked standalone bond from {bond.counterparty} "
                f"(txid={bond.utxo_txid[:16]}...) to offer oid={offer.oid}"
            )

    def _populate_bond_directory_nodes(self, orderbook: OrderBook) -> None:
        offers_by_counterparty: dict[str, list[Offer]] = {}
        for offer in orderbook.offers:
            offers_by_counterparty.setdefault(offer.counterparty, []).append(offer)

        for bond in orderbook.fidelity_bonds:
            maker_offers = offers_by_counterparty.get(bond.counterparty)
            if not maker_offers:
                continue
            all_directories: set[str] = set(bond.directory_nodes)
            for offer in maker_offers:
                all_directories.update(offer.directory_nodes)
            bond.directory_nodes = sorted(all_directories)

    def _enrich_offers_with_health_status(self, orderbook: OrderBook) -> None:
        location_by_counterparty: dict[str, str] = {}
        for offer in orderbook.offers:
            for client in self.clients.values():
                location = client._active_peers.get(offer.counterparty)
                if location and location != "NOT-SERVING-ONION":
                    location_by_counterparty[offer.counterparty] = location
                    break

        for offer in orderbook.offers:
            location = location_by_counterparty.get(offer.counterparty)
            if not location:
                continue
            health_status = self.health_checker.health_status.get(location)
            if health_status is None:
                continue

            offer.directly_reachable = health_status.reachable
            if not health_status.features:
                continue

            health_features = health_status.features.to_dict()
            for feature, value in health_features.items():
                if value:
                    offer.features[feature] = value

    async def _calculate_bond_value_single(
        self, bond: FidelityBond, current_time: int
    ) -> FidelityBond:
        if bond.bond_value is not None:
            return bond

        if self.mempool_api is None:
            return bond

        async with self._mempool_semaphore:
            try:
                tx_data = await self.mempool_api.get_transaction(bond.utxo_txid)
                if not tx_data or not tx_data.status.confirmed:
                    logger.debug(f"Bond {bond.utxo_txid}:{bond.utxo_vout} not confirmed")
                    return bond

                if bond.utxo_vout >= len(tx_data.vout):
                    logger.warning(
                        f"Invalid vout {bond.utxo_vout} for tx {bond.utxo_txid} "
                        f"(only {len(tx_data.vout)} outputs)"
                    )
                    return bond

                utxo = tx_data.vout[bond.utxo_vout]
                amount = utxo.value
                confirmation_time = tx_data.status.block_time or current_time

                bond_value = calculate_timelocked_fidelity_bond_value(
                    amount, confirmation_time, bond.locktime, current_time
                )

                bond.bond_value = bond_value
                bond.amount = amount
                bond.utxo_confirmation_timestamp = confirmation_time

                logger.debug(
                    f"Bond {bond.counterparty}: value={bond_value}, "
                    f"amount={amount}, locktime={datetime.utcfromtimestamp(bond.locktime)}, "
                    f"confirmed={datetime.utcfromtimestamp(confirmation_time)}"
                )

            except Exception as e:
                logger.error(f"Failed to calculate bond value for {bond.utxo_txid}: {e}")
                logger.debug(
                    f"Bond data: txid={bond.utxo_txid}, vout={bond.utxo_vout}, amount={bond.amount}"
                )

        return bond

    async def _calculate_bond_values(self, orderbook: OrderBook) -> None:
        """Calculate bond values, using blockchain backend if available, else mempool API."""
        if self.blockchain_backend is not None:
            await self._calculate_bond_values_via_backend(orderbook)
        else:
            await self._calculate_bond_values_via_mempool(orderbook)

    async def _calculate_bond_values_via_backend(self, orderbook: OrderBook) -> None:
        """Calculate bond values using the blockchain backend's verify_bonds().

        This path is used when a full node or neutrino backend is configured,
        providing trustless and private bond verification without external API calls.
        """
        current_time = int(datetime.now(UTC).timestamp())

        # Build verification requests from bonds that need values
        bonds_to_verify: list[tuple[FidelityBond, BondVerificationRequest]] = []

        for bond in orderbook.fidelity_bonds:
            if bond.bond_value is not None:
                continue

            # Get utxo_pub and locktime from bond data
            bond_data = bond.fidelity_bond_data
            utxo_pub_hex: str | None = None
            locktime: int = bond.locktime

            if bond_data:
                utxo_pub_hex = bond_data.get("utxo_pub")

            if not utxo_pub_hex and bond.script:
                # Fallback: the `script` field stores utxo_pub hex
                utxo_pub_hex = bond.script

            if not utxo_pub_hex:
                logger.debug(f"Bond {bond.utxo_txid}:{bond.utxo_vout} missing utxo_pub, skipping")
                continue

            try:
                utxo_pub_bytes = bytes.fromhex(utxo_pub_hex)
                bond_addr = derive_bond_address(utxo_pub_bytes, locktime, self.network)
            except Exception as e:
                logger.debug(
                    f"Failed to derive bond address for {bond.utxo_txid}:{bond.utxo_vout}: {e}"
                )
                continue

            request = BondVerificationRequest(
                txid=bond.utxo_txid,
                vout=bond.utxo_vout,
                utxo_pub=utxo_pub_bytes,
                locktime=locktime,
                address=bond_addr.address,
                scriptpubkey=bond_addr.scriptpubkey.hex(),
            )
            bonds_to_verify.append((bond, request))

        if not bonds_to_verify:
            return

        logger.info(f"Verifying {len(bonds_to_verify)} bonds via blockchain backend...")

        try:
            assert self.blockchain_backend is not None
            results = await self.blockchain_backend.verify_bonds(
                [req for _, req in bonds_to_verify]
            )
        except Exception as e:
            logger.error(f"Backend bond verification failed: {e}")
            return

        # Update bond objects with results
        for (bond, _request), result in zip(bonds_to_verify, results, strict=True):
            if not result.valid:
                logger.debug(f"Bond {bond.utxo_txid}:{bond.utxo_vout} invalid: {result.error}")
                continue

            bond_value = calculate_timelocked_fidelity_bond_value(
                result.value, result.block_time, bond.locktime, current_time
            )

            bond.bond_value = bond_value
            bond.amount = result.value
            bond.utxo_confirmation_timestamp = result.block_time
            bond.utxo_confirmations = result.confirmations

            logger.debug(
                f"Bond {bond.counterparty}: value={bond_value}, "
                f"amount={result.value}, locktime={datetime.utcfromtimestamp(bond.locktime)}, "
                f"confirmed={datetime.utcfromtimestamp(result.block_time)}"
            )

        valid_count = sum(1 for r in results if r.valid)
        logger.info(f"Bond verification complete: {valid_count}/{len(bonds_to_verify)} verified")

    async def _calculate_bond_values_via_mempool(self, orderbook: OrderBook) -> None:
        """Calculate bond values via mempool API (legacy path)."""
        if self.mempool_api is None:
            logger.debug("Skipping mempool bond calculation (mempool API disabled)")
            return

        current_time = int(datetime.now(UTC).timestamp())

        tasks = [
            self._calculate_bond_value_single(bond, current_time)
            for bond in orderbook.fidelity_bonds
        ]
        await asyncio.gather(*tasks, return_exceptions=True)

    async def _test_mempool_connection(self) -> None:
        """Test the configured mempool connection on startup."""
        if self.mempool_api is None:
            return

        try:
            success = await self.mempool_api.test_connection()
            if success:
                logger.info("Mempool API connection test successful")
            else:
                logger.warning(
                    "Mempool API connection test failed - bond value calculation may not work"
                )
        except Exception as e:
            logger.error(f"Mempool API connection test error: {e}")
            logger.warning("Bond value calculation may not work without mempool API access")

Attributes

blockchain_backend = blockchain_backend instance-attribute

clients: dict[str, DirectoryClient] = {} instance-attribute

current_orderbook: OrderBook = OrderBook() instance-attribute

directory_nodes = directory_nodes instance-attribute

health_checker = MakerHealthChecker(network=network, socks_host=socks_host, socks_port=socks_port, timeout=timeout, check_interval=600.0, max_concurrent_checks=5, socks_username=(self._hc_username), socks_password=(self._hc_password)) instance-attribute

listener_tasks: list[asyncio.Task[Any]] = [] instance-attribute

max_message_size = max_message_size instance-attribute

max_retry_attempts = max_retry_attempts instance-attribute

mempool_api: MempoolAPI | None = None instance-attribute

mempool_api_url = mempool_api_url instance-attribute

mempool_api_use_tor = mempool_api_use_tor instance-attribute

network = network instance-attribute

node_statuses: dict[str, DirectoryNodeStatus] = {} instance-attribute

retry_delay = retry_delay instance-attribute

socks_host = socks_host instance-attribute

socks_port = socks_port instance-attribute

stream_isolation = stream_isolation instance-attribute

timeout = timeout instance-attribute

uptime_grace_period = uptime_grace_period instance-attribute

Methods:

__init__(directory_nodes: list[tuple[str, int]], network: str, mempool_api_url: str, mempool_api_use_tor: bool = True, socks_host: str = '127.0.0.1', socks_port: int = 9050, timeout: float = 30.0, max_retry_attempts: int = 3, retry_delay: float = 5.0, max_message_size: int = 2097152, uptime_grace_period: int = 60, stream_isolation: bool = False, blockchain_backend: BlockchainBackend | None = None) -> None

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

def __init__(
    self,
    directory_nodes: list[tuple[str, int]],
    network: str,
    mempool_api_url: str,
    mempool_api_use_tor: bool = True,
    socks_host: str = "127.0.0.1",
    socks_port: int = 9050,
    timeout: float = 30.0,
    max_retry_attempts: int = 3,
    retry_delay: float = 5.0,
    max_message_size: int = 2097152,
    uptime_grace_period: int = 60,
    stream_isolation: bool = False,
    blockchain_backend: BlockchainBackend | None = None,
) -> None:
    self.directory_nodes = directory_nodes
    self.network = network
    self.socks_host = socks_host
    self.socks_port = socks_port
    self.stream_isolation = stream_isolation
    self.timeout = timeout
    self.mempool_api_url = mempool_api_url
    self.mempool_api_use_tor = mempool_api_use_tor
    self.max_retry_attempts = max_retry_attempts
    self.retry_delay = retry_delay
    self.max_message_size = max_message_size
    self.uptime_grace_period = uptime_grace_period
    self.blockchain_backend = blockchain_backend

    # Build the optional mempool proxy URL and pre-compute isolation credentials.
    self._dir_username: str | None = None
    self._dir_password: str | None = None
    self._hc_username: str | None = None
    self._hc_password: str | None = None
    socks_proxy: str | None = None
    if stream_isolation:
        from jmcore.tor_isolation import (  # noqa: PLC0415
            IsolationCategory,
            build_isolated_proxy_url,
            get_isolation_credentials,
        )

        if mempool_api_use_tor:
            socks_proxy = build_isolated_proxy_url(
                socks_host, socks_port, IsolationCategory.MEMPOOL
            )
        dir_c = get_isolation_credentials(IsolationCategory.DIRECTORY)
        self._dir_username = dir_c.username
        self._dir_password = dir_c.password
        hc_c = get_isolation_credentials(IsolationCategory.HEALTH_CHECK)
        self._hc_username = hc_c.username
        self._hc_password = hc_c.password
    elif mempool_api_use_tor:
        socks_proxy = f"socks5h://{socks_host}:{socks_port}"
    self.mempool_api: MempoolAPI | None = None
    self._mempool_test_task: asyncio.Task[Any] | None = None
    if mempool_api_url:
        mempool_timeout = 60.0
        if mempool_api_use_tor:
            logger.info("Mempool API configured with Tor routing")
        else:
            logger.info("Mempool API configured for direct access")
        self.mempool_api = MempoolAPI(
            base_url=mempool_api_url,
            socks_proxy=socks_proxy,
            timeout=mempool_timeout,
            trust_env=False,
        )
        self._mempool_test_task = asyncio.create_task(self._test_mempool_connection())
    else:
        logger.info("Mempool API disabled by configuration; external mempool lookups are off")
    self.current_orderbook: OrderBook = OrderBook()
    self._lock = asyncio.Lock()
    self.clients: dict[str, DirectoryClient] = {}
    self.listener_tasks: list[asyncio.Task[Any]] = []
    self._bond_calculation_task: asyncio.Task[Any] | None = None
    self._bond_queue: asyncio.Queue[OrderBook] = asyncio.Queue()
    self._bond_cache: dict[str, FidelityBond] = {}
    self._last_offers_hash: int = 0
    self._mempool_semaphore = asyncio.Semaphore(5)
    self.node_statuses: dict[str, DirectoryNodeStatus] = {}
    self._retry_tasks: list[asyncio.Task[Any]] = []

    # Maker health checker for direct reachability verification
    self.health_checker = MakerHealthChecker(
        network=network,
        socks_host=socks_host,
        socks_port=socks_port,
        timeout=timeout,
        check_interval=600.0,  # Check each maker at most once per 10 minutes
        max_concurrent_checks=5,
        socks_username=self._hc_username,
        socks_password=self._hc_password,
    )

    for onion_address, port in directory_nodes:
        node_id = f"{onion_address}:{port}"
        self.node_statuses[node_id] = DirectoryNodeStatus(
            node_id, grace_period_seconds=uptime_grace_period
        )

fetch_from_directory(onion_address: str, port: int) -> tuple[list[Offer], list[FidelityBond], str] async

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

async def fetch_from_directory(
    self, onion_address: str, port: int
) -> tuple[list[Offer], list[FidelityBond], str]:
    node_id = f"{onion_address}:{port}"
    logger.info(f"Fetching orderbook from directory: {node_id}")
    client = DirectoryClient(
        onion_address,
        port,
        self.network,
        socks_host=self.socks_host,
        socks_port=self.socks_port,
        timeout=self.timeout,
        max_message_size=self.max_message_size,
        socks_username=self._dir_username,
        socks_password=self._dir_password,
    )
    try:
        await client.connect()
        offers, bonds = await client.fetch_orderbooks()

        for offer in offers:
            offer.directory_node = node_id
        for bond in bonds:
            bond.directory_node = node_id

        return offers, bonds, node_id
    except Exception as e:
        logger.error(f"Failed to fetch from directory {node_id}: {e}")
        return [], [], node_id
    finally:
        await client.close()

get_live_orderbook(calculate_bonds: bool = True) -> OrderBook async

Get the live orderbook with deduplication and bond value calculation.

This method implements several key cleanup mechanisms: 1. Deduplicates offers by bond UTXO - if multiple nicks use the same bond, only the most recently received offer is kept 2. Deduplicates bonds by UTXO key 3. Links bonds to offers and calculates bond values

Returns: OrderBook with deduplicated offers and calculated bond values

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

async def get_live_orderbook(self, calculate_bonds: bool = True) -> OrderBook:
    """Get the live orderbook with deduplication and bond value calculation.

    This method implements several key cleanup mechanisms:
    1. Deduplicates offers by bond UTXO - if multiple nicks use the same bond,
       only the most recently received offer is kept
    2. Deduplicates bonds by UTXO key
    3. Links bonds to offers and calculates bond values

    Returns:
        OrderBook with deduplicated offers and calculated bond values
    """
    orderbook = OrderBook(timestamp=datetime.now(UTC))
    all_offers_with_timestamps = self._collect_live_offers_and_bonds(orderbook)
    deduplicated_offers = self._deduplicate_bond_backed_offers(all_offers_with_timestamps)
    self._merge_offers_into_orderbook(orderbook, deduplicated_offers)
    self._deduplicate_bonds(orderbook)

    if calculate_bonds:
        self._apply_bond_cache(orderbook)
        await self._calculate_bond_values(orderbook)
        self._update_bond_cache(orderbook)
        self._link_bonds_to_offers(orderbook)

    self._populate_bond_directory_nodes(orderbook)
    self._enrich_offers_with_health_status(orderbook)
    return orderbook

get_orderbook() -> OrderBook async

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

async def get_orderbook(self) -> OrderBook:
    async with self._lock:
        return self.current_orderbook

start_continuous_listening() -> None async

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

async def start_continuous_listening(self) -> None:
    logger.info("Starting continuous listening on all directory nodes")

    self._bond_calculation_task = asyncio.create_task(self._background_bond_calculator())

    # Start periodic directory connection status logging task
    status_task = asyncio.create_task(self._periodic_directory_connection_status())
    self.listener_tasks.append(status_task)

    # Start periodic peerlist refresh task for cleanup
    peerlist_task = asyncio.create_task(self._periodic_peerlist_refresh())
    self.listener_tasks.append(peerlist_task)

    # Start periodic feature-discovery task (direct onion handshake to learn
    # maker capabilities when directories don't advertise peerlist_features).
    # This task NEVER removes offers; offer presence is governed by peerlist.
    health_check_task = asyncio.create_task(self._periodic_maker_health_check())
    self.listener_tasks.append(health_check_task)

    connection_tasks = [
        self._connect_to_node(onion_address, port)
        for onion_address, port in self.directory_nodes
    ]

    clients = await asyncio.gather(*connection_tasks, return_exceptions=True)

    for (onion_address, port), result in zip(self.directory_nodes, clients, strict=True):
        node_id = f"{onion_address}:{port}"

        if isinstance(result, BaseException):
            logger.error(f"Connection to {node_id} raised exception: {result}")
            retry_task = asyncio.create_task(self._retry_failed_connection(onion_address, port))
            self._retry_tasks.append(retry_task)
            logger.info(f"Scheduled retry task for {node_id}")
        elif result is not None:
            self.clients[node_id] = result
            task = asyncio.create_task(result.listen_continuously())
            self.listener_tasks.append(task)
            logger.info(f"Started listener task for {node_id}")
        else:
            retry_task = asyncio.create_task(self._retry_failed_connection(onion_address, port))
            self._retry_tasks.append(retry_task)
            logger.info(f"Scheduled retry task for {node_id}")

    # Start early feature discovery task - runs once after initial connections settle
    early_feature_task = asyncio.create_task(self._early_feature_discovery())
    self.listener_tasks.append(early_feature_task)

stop_listening() -> None async

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

async def stop_listening(self) -> None:
    logger.info("Stopping all directory listeners")

    if self._bond_calculation_task:
        self._bond_calculation_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._bond_calculation_task

    for task in self._retry_tasks:
        task.cancel()

    if self._retry_tasks:
        await asyncio.gather(*self._retry_tasks, return_exceptions=True)

    for client in self.clients.values():
        client.stop()

    for task in self.listener_tasks:
        task.cancel()

    if self.listener_tasks:
        await asyncio.gather(*self.listener_tasks, return_exceptions=True)

    for node_id, client in self.clients.items():
        self.node_statuses[node_id].mark_disconnected()
        await client.close()

    self.clients.clear()
    self.listener_tasks.clear()
    self._retry_tasks.clear()

update_orderbook() -> OrderBook async

Source code in orderbook_watcher/src/orderbook_watcher/aggregator.py

async def update_orderbook(self) -> OrderBook:
    tasks = [
        self.fetch_from_directory(onion_address, port)
        for onion_address, port in self.directory_nodes
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    new_orderbook = OrderBook(timestamp=datetime.now(UTC))

    for result in results:
        if isinstance(result, BaseException):
            logger.error(f"Directory fetch failed: {result}")
            continue

        offers, bonds, node_id = result
        if offers or bonds:
            new_orderbook.add_offers(offers, node_id)
            new_orderbook.add_fidelity_bonds(bonds, node_id)

    await self._calculate_bond_values(new_orderbook)

    async with self._lock:
        self.current_orderbook = new_orderbook

    logger.info(
        f"Updated orderbook: {len(new_orderbook.offers)} offers, "
        f"{len(new_orderbook.fidelity_bonds)} bonds from "
        f"{len(new_orderbook.directory_nodes)} directory nodes"
    )

    return new_orderbook

Functions: