Source code for trader.daemon.signals

"""Shutdown signalling for the daemon.

A flag rather than an exception raised from the handler: the loop must be
allowed to *finish the cycle it is in*. A `KeyboardInterrupt` landing between
"cancel the protective stop" and "submit the sell" is exactly the interruption
this app's ordering rules exist to prevent.
"""

import signal
from collections.abc import Callable
from types import FrameType

__all__ = ["ShutdownFlag", "install_handlers"]




class ShutdownFlag:
    """Latching "stop after this cycle" flag, set from a signal handler.

    Latching on purpose: a second SIGTERM must not clear it. An operator
    pressing Ctrl-C twice means "stop harder", never "carry on".
    """

    def __init__(self) -> None:
        self._set = False
        self._reason = ""



    def is_set(self) -> bool:
        """Whether shutdown has been requested."""
        return self._set


    @property
    def reason(self) -> str:
        """What asked the daemon to stop, for the final log line."""
        return self._reason



    def request(self, signum: int | None = None, frame: FrameType | None = None) -> None:
        """Request shutdown. Signature matches `signal.signal`'s handler."""
        if self._set:
            return
        self._set = True
        self._reason = signal.Signals(signum).name if signum else "requested"






def install_handlers(flag: ShutdownFlag) -> Callable[[], None]:
    """Route SIGINT and SIGTERM to `flag`, returning a restore function.

    Returning the restore rather than installing permanently keeps a test
    process from inheriting the daemon's handlers, and makes the installation
    itself assertable.
    """
    previous = {sig: signal.getsignal(sig) for sig in (signal.SIGINT, signal.SIGTERM)}
    for sig in previous:
        signal.signal(sig, flag.request)

    def restore() -> None:
        for sig, handler in previous.items():
            signal.signal(sig, handler)

    return restore