"""The daemon loop: when to run a cycle, and how to survive one failing.
This module knows nothing about strategies, orders, or persistence. It is
handed a `cycle` callable and decides when to call it — which is what keeps the
trading rules in `pipeline/run_once.py` where they can be reasoned about
without a scheduler in the picture.
Three things here are deliberate and easy to undo by accident:
**Shutdown finishes the cycle it is in.** The flag is checked between phases,
never raised through one. A signal landing between "cancel the protective
stop" and "submit the sell" is the interruption this app's ordering rules
exist to prevent.
**Positions are carried across cycles.** `reconcile_fills` needs the positions
as they were *before* the fills it is reconciling. Re-reading at the start of
the next cycle would already include the fill, so a newly filled position would
look like one that had always been there — and a fill invisible to
reconciliation is a round trip that never opens.
**A `ConfigError` is fatal.** Waiting does not fix a misconfiguration, and
`UnsafeConfigError` means the safety gate refused something. Exiting non-zero
is both more honest and what LaunchAgent's throttling understands.
"""
import logging
from collections.abc import Callable
from datetime import UTC, datetime
from trader.brokers.base import BrokerAdapter
from trader.config.schema import DaemonConfig
from trader.daemon.backoff import Backoff
from trader.daemon.schedule import MarketSchedule, SleepFor
from trader.daemon.signals import ShutdownFlag
from trader.daemon.sleeper import Sleeper
from trader.domain import Position
from trader.errors import ConfigError
from trader.pipeline.run_once import CycleReport
__all__ = ["run_forever"]
_logger = logging.getLogger("trader.daemon")
_EXIT_OK = 0
_EXIT_FATAL = 1
[docs]
def run_forever(
*,
broker: BrokerAdapter,
cycle: Callable[[dict[str, Position] | None], CycleReport],
schedule: MarketSchedule,
backoff: Backoff,
sleeper: Sleeper,
shutdown: ShutdownFlag,
config: DaemonConfig,
now: Callable[[], datetime] = lambda: datetime.now(UTC),
max_cycles: int | None = None,
) -> int:
"""Run cycles until stopped, and return a process exit code.
Args:
broker: a *read-only* adapter — in production the retrying wrapper.
The loop only ever reads the clock and positions through it; the
order-placing broker is inside `cycle`, where it belongs.
cycle: called with the previous cycle's positions (or `None`), returns
a `CycleReport`. In production this is `run_once` with its graph
already bound.
max_cycles: stop after this many *successfully completed* cycles.
Failed cycles do not count towards it, so `--max-cycles 2` means
two real cycles rather than two attempts. For verification runs
and tests; `None` means run until shutdown.
Returns:
0 for a clean stop, 1 for a fatal configuration or safety failure.
"""
interval_seconds = float(config.cycle_interval_minutes * 60)
previous_positions: dict[str, Position] | None = None
cycles_run = 0
# The last cycle's report, kept so every exit from this loop — clean or
# fatal — can repeat its safety-relevant summary once more right before
# the process actually returns. See `_log_shutdown_summary`.
last_report: CycleReport | None = None
# Whether the most recent *attempt* at a cycle raised. `last_report` alone
# cannot say: it is only ever assigned on success, so after a cycle that
# raised "possibly after submitting an order" the shutdown summary
# cheerfully repeated the previous clean cycle at INFO. Tracked separately
# rather than by clearing `last_report`, because the last successful
# cycle's summary is still worth printing — it just must not be presented
# as the current state.
last_cycle_failed = False
while not shutdown.is_set():
try:
clock = broker.get_clock()
except ConfigError as exc:
_logger.critical("Refusing to continue: %s", exc)
_log_shutdown_summary(last_report, cycles_run)
return _EXIT_FATAL
except Exception:
delay = backoff.next_delay()
_logger.exception(
"Could not read the market clock (failure %d); waiting %ss",
backoff.consecutive_failures,
delay,
)
sleeper.sleep(delay, shutdown)
continue
action = schedule.next_action(clock, now())
if isinstance(action, SleepFor):
_logger.info("Idle: %s (%.0fs)", action.reason, action.seconds)
sleeper.sleep(action.seconds, shutdown)
continue
try:
report = cycle(previous_positions)
except ConfigError as exc:
# Not retried and not backed off: this will not fix itself, and a
# process that sleeps on it hides the reason it is doing nothing.
_logger.critical("Refusing to continue: %s", exc)
_log_shutdown_summary(last_report, cycles_run, cycle_failed=True)
return _EXIT_FATAL
except Exception:
last_cycle_failed = True
delay = backoff.next_delay()
_logger.exception(
"Cycle failed (failure %d); waiting %ss before the next one",
backoff.consecutive_failures,
delay,
)
# Dropped rather than reused: the cycle raised somewhere unknown,
# possibly after submitting an order, so this snapshot is no longer
# a safe "before". A wrong previous is worse than none.
previous_positions = None
sleeper.sleep(delay, shutdown)
continue
cycles_run += 1
backoff.reset()
_log_report(report, cycles_run)
last_report = report
last_cycle_failed = False
previous_positions = _positions_now(broker)
if max_cycles is not None and cycles_run >= max_cycles:
_logger.info("Reached the requested %d cycle(s); stopping.", max_cycles)
_log_shutdown_summary(last_report, cycles_run)
return _EXIT_OK
sleeper.sleep(interval_seconds, shutdown)
_logger.info("Shutting down (%s).", shutdown.reason or "requested")
_log_shutdown_summary(last_report, cycles_run, cycle_failed=last_cycle_failed)
return _EXIT_OK
def _positions_now(broker: BrokerAdapter) -> dict[str, Position] | None:
"""Snapshot positions for the next cycle's reconciliation.
A failure here is not worth failing a cycle over: `run_once` falls back to
reading positions itself when handed `None`, which is exactly the
pre-daemon behaviour.
"""
try:
return {p.symbol: p for p in broker.get_positions()}
except Exception:
_logger.warning(
"Could not snapshot positions after the cycle; the next cycle will "
"read them itself.",
exc_info=True,
)
return None
def _cycle_summary(report: CycleReport) -> str:
"""One line an operator can scan, not a dump of every decision.
Shared by the per-cycle log line and its repeat at shutdown
(`_log_shutdown_summary`), so the two can never say different things about
the same `CycleReport`.
"""
actions: dict[str, int] = {}
for outcome in report.outcomes:
actions[outcome.action] = actions.get(outcome.action, 0) + 1
summary = ", ".join(f"{count} {action}" for action, count in sorted(actions.items()))
return (
f"{summary or 'no decisions'}"
+ (
f"; stale entries cancelled for {', '.join(report.entries_cancelled)}"
if report.entries_cancelled
else ""
)
+ (
f"; stops placed for {', '.join(report.stops_placed)}"
if report.stops_placed
else ""
)
+ (
f"; stops raised for {', '.join(report.stops_ratcheted)}"
if report.stops_ratcheted
else ""
)
+ ("; DAILY LOSS LIMIT BREACHED, no new entries" if report.halted else "")
+ (
f"; UNPROTECTED position(s): {', '.join(report.unprotected)}"
if report.unprotected
else ""
)
+ (
f"; DOUBLE PROTECTED position(s): {', '.join(report.double_protected)}"
if report.double_protected
else ""
)
+ (
f"; outstanding unfilled entries for {', '.join(report.outstanding_entries)}"
if report.outstanding_entries
else ""
)
)
def _log_report(report: CycleReport, cycle_number: int) -> None:
"""Log the cycle's summary, at WARNING when it leaves anything unprotected,
double-protected or outstanding — never at INFO, where an operator who only
watches warnings would never see it."""
level = (
logging.WARNING
if (report.unprotected or report.double_protected or report.outstanding_entries)
else logging.INFO
)
_logger.log(level, "Cycle %d complete: %s", cycle_number, _cycle_summary(report))
def _log_shutdown_summary(
report: CycleReport | None, cycle_number: int, *, cycle_failed: bool = False
) -> None:
"""Repeat the last cycle's summary once more, right before this process
actually returns its exit code.
Requirements §8: "The app must never terminate quietly while it believes
money is unprotected. Silence must mean 'protected', never 'not
checked'." On 2026-08-07 a daemon cycle submitted two buys, logged a
clean summary, and exited 0 while both sat unfilled at the broker; the
one line that named it scrolled by and nobody looked again. Called from
every `return` in `run_forever` — the two ordinary ones and the two fatal
`ConfigError` ones — so it is the last thing in the log on every path out,
not only the ones anyone expected to need it.
`report is None` means no cycle ever completed this run (a startup
failure, say): there is nothing to have left unprotected, so this logs
that plainly rather than warning about a report that does not exist.
`cycle_failed` says the most recent *attempt* raised, which is a different
claim from anything in `report` — `report` is the last cycle that
*succeeded*, and after a failure it describes a state two cycles old. The
daemon drops `previous_positions` on a failed cycle precisely because the
cycle "raised somewhere unknown, possibly after submitting an order", and
this says so, at WARNING: a summary of an older clean cycle logged at INFO
read as "all is well" when the truth was "unknown". The exit code is
deliberately unchanged — a failed cycle is retried, not fatal, and
`--max-cycles` runs count successes only.
"""
if cycle_failed:
_logger.warning(
"Shutting down after a cycle that FAILED: its effects are UNKNOWN "
"— it may have submitted an order before raising, so positions and "
"protection at the broker are unverified. Check the broker "
"directly. %s",
(
f"The last cycle that completed was cycle {cycle_number}: "
f"{_cycle_summary(report)}"
if report is not None
else "No cycle completed successfully this run."
),
)
return
if report is None:
_logger.info("Shutting down; no cycle completed this run.")
return
level = (
logging.WARNING
if (report.unprotected or report.double_protected or report.outstanding_entries)
else logging.INFO
)
_logger.log(
level, "Shutting down after cycle %d: %s", cycle_number, _cycle_summary(report)
)