diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e1c249..2066543 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Removed +## [0.15.0] - 2026-07-14 + +### Added + +- **Daemon logs are real logs** — the daemon path writes `logs/daemon.log` + through a midnight-rotating handler (14-day retention) with ISO-8601 local + timestamps carrying the numeric tz offset; level/dir/retention configurable + via the manifest's new optional `logging:` section + (`application/log_setup.py`); tick errors now log their traceback + (previously invisible — no handler existed). Interactive CLI output + unchanged. (#226) +- **Per-unit daemon log trail** — every evaluated rebalance logs one + grep-stable summary (`unit=… rebalance asof=… legs/submitted/round_up/ + skipped/on_target`, counters partitioning the universe); round-up/skip leg + decisions log their full `LegDecision.reason` (the exact Decimals and the + binding venue minimum) at INFO; order submits (cid, side, qty, px, + venue id) at INFO and refusals at WARN; each applied fill logs + cid/qty/price/fee/fee-asset (`OrderFillSync.unit_name`, additive); unit + start/stop/step-error transitions logged. Anti-spam pinned: an idle tick + (no new bar) adds zero per-unit lines — a unit's inactivity is now + explained by the log. (#227) +- **Tick timing measured and pinned (closes the `daemon-logging` epic)** — + every tick's duration is logged on the heartbeat (`in s`, WARN on + interval overrun) and per-unit step durations captured; the tick job's + APScheduler semantics are now explicit (`coalesce=True`, `max_instances=1`, + `misfire_grace_time=interval` — the 1 s default grace silently skipped late + ticks, the measured source of the audit's ~74 s-vs-60 s cadence shortfall); + additive API fields: `/api/health` `last_tick_duration_ms` / `last_tick_ts` / + `ticks_total` / `ticks_overrun`, `/api/strategies` `last_step_duration_ms`. + Real-data check: steady-state ticks ~2 s, first tick (data load) ~46 s — + the duration hypothesis refuted, the misfire one confirmed as the fix. (#228) + +### Changed + +### Fixed + +### Deprecated + +### Removed + ## [0.14.0] - 2026-07-12 ### Added diff --git a/doc/dev/03-decisions.md b/doc/dev/03-decisions.md index 2830c08..8b017d3 100644 --- a/doc/dev/03-decisions.md +++ b/doc/dev/03-decisions.md @@ -6,6 +6,61 @@ rejected approaches as tombstones. --- +### 2026-07-14 Tick scheduler semantics pinned; timing is a health signal (PR #228) [accepted] +- **Choice**: the daemon tick job runs with explicit `coalesce=True`, + `max_instances=1`, `misfire_grace_time=interval` (cron triggers: fixed + 30 s); every tick's duration is measured (heartbeat `in s`, WARN on + interval overrun) and exposed additively on `/api/health` + (`last_tick_duration_ms`/`last_tick_ts`/`ticks_total`/`ticks_overrun`) and + `/api/strategies` (`last_step_duration_ms`). Per-unit duration lives on + the API field only, measured in the supervisor's `step` `finally` (counted + even when the step raises). +- **Why**: APScheduler's **1 s default misfire grace silently skips** a tick + that fires late — the measured cause of the audit's ~74 s realised cadence + vs the 60 s nominal (real-data check: steady-state ticks ~2 s, first tick + ~46 s data load — duration never exceeds the interval, so skips, not + overruns, explain the shortfall). `max_instances=1` guarantees two ticks + never race one engine. +- **Rejected alternatives**: raising the tick interval (hides the problem); + gathering units concurrently inside a tick (the per-unit lock serializes + anyway; complexity without cadence benefit); putting per-unit duration on + the leaf-02 rebalance line (emitted inside the runner, before the outer + step timing exists). + +### 2026-07-14 Unit event trail: runner-attributed lines, bus and log are two sinks (PR #227) [accepted] +- **Choice**: order-lifecycle log lines live in the **runner** (per-unit, sees + the router's outcome and the unit name) — not the shared router; the new + module-logger INFO/WARN lines are emitted **alongside** the existing + EventBus `LogEvent`s, not instead (the bus feeds the SSE dashboard, the + logger feeds `daemon.log` — two sinks, two audiences). `OrderFillSync` + gains an additive `unit_name` for fill attribution; supervisor step errors + are logged with traceback then **re-raised** (attribution without + swallowing the CLI tick's safety net). +- **Why**: threading unit identity through the shared router would touch its + signature for a logging concern; the runner already owns the name and the + outcome. Replacing the bus emits would have silently unplugged the + dashboard's live log feed. +- **Rejected alternatives**: per-unit `LoggerAdapter` injected into the + router (signature churn); replacing `LogEvent`s with logging (breaks SSE); + logging every idle tick per unit (2 units × 1440 min/day of noise — the + anti-spam rule is pinned in the plan). + +### 2026-07-14 Daemon logging spine: root handlers, offset timestamps (PR #226) [accepted] +- **Choice**: the daemon's logging config lives in `application/log_setup.py` + and wires the **root** logger with two owned, tagged handlers (midnight + `TimedRotatingFileHandler` + stderr stream); third-party namespaces + (`apscheduler`/`uvicorn`/`httpx`/`websockets`) are capped at WARNING; + timestamps are **local ISO-8601 with the numeric UTC offset**. +- **Why**: one spine, N emitters — the runner/router/feed module loggers + (some already emitting into the void) flow into the rotated file without + knowing it exists; the owned-handler tag makes re-configuration idempotent + (no double-writes under tests or embedding); the explicit offset kills the + UTC-vs-CEST ambiguity that burned the 2026-07-14 audit. +- **Rejected alternatives**: UTC-only timestamps (the operator reads local + time; the offset carries both); per-module handlers (drift + duplication); + wiring in `interfaces/cli` (the seam is application-level composition, + reusable by a future systemd entrypoint). + ### 2026-07-12 The venue oracle is an accounting identity; fees carry their asset (PR #223) [accepted] - **Choice**: on a real venue the canary asserts the **identity** — our fills == venue-reported fills, per-asset balance deltas == the fills math diff --git a/doc/dev/06-status.md b/doc/dev/06-status.md index 984f2e3..5468618 100644 --- a/doc/dev/06-status.md +++ b/doc/dev/06-status.md @@ -120,6 +120,17 @@ and `CHANGELOG.md` for what shipped. - **E1–E11 + the multi-asset/portfolio unit** shipped; the pre-production safety hardening (audit) is wired in (see *Where things stand*). `CHANGELOG.md` + git log are authoritative for what shipped per release. +- **`daemon-logging` (2026-07-14, PRs #226–#228)**: the daemon is auditable + from its log alone — rotated, tz-offset-timestamped `logs/daemon.log` + (manifest `logging:` section), per-unit rebalance summaries + round-up/skip + `LegDecision.reason`s + order/fill lines (a unit's inactivity is now + explained by the log), tick durations on the heartbeat + additive + `/api/health`//`api/strategies` timing fields. Root-caused the 2026-07-14 + audit's ~74 s-vs-60 s cadence shortfall: APScheduler's 1 s default misfire + grace silently skipped late ticks (steady-state ticks measure ~2 s; first + tick ~46 s data load) — the job now runs `coalesce=True`, `max_instances=1`, + `misfire_grace_time=interval`. The legacy unrotated log gets archived at the + post-release daemon restart. - **`paper-integrity` (2026-07-11, PRs #190–#194)**: paper ids unique across engine lifetimes, fills reach the tracked order + store row (`OrderFillSync`, startup healing), reconcile persists orphan-closes, restart/replay E2E. @@ -183,6 +194,7 @@ engine code: **real-key live enablement** (validate Kraken private endpoints + venue-level idempotency against a real-key sandbox, then flip `live_enabled`) — the one maintainer step in [`07-roadmap.md`](07-roadmap.md). + ## Known gaps / deferred - ~~**Final project name**~~ — **decided**: kept as `trading_bot` (with `dccd` / diff --git a/pyproject.toml b/pyproject.toml index bc9dee9..ac59d81 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "trading_bot" -version = "0.14.0" +version = "0.15.0" description = "Execution & orchestration layer of the trading triptych — live strategies, order routing, risk. Hexagonal, async-first." readme = "README.md" license = { text = "MIT" } diff --git a/trading_bot/application/config.py b/trading_bot/application/config.py index 5374385..e35ea9e 100644 --- a/trading_bot/application/config.py +++ b/trading_bot/application/config.py @@ -58,6 +58,7 @@ from __future__ import annotations +import logging import os import pathlib from decimal import Decimal @@ -77,6 +78,7 @@ "StrategyConfig", "PortfolioStrategyConfig", "RiskConfig", + "LoggingConfig", "AppConfig", ] @@ -671,6 +673,54 @@ def _require_token_for_non_loopback(self) -> UIConfig: return self +class LoggingConfig(BaseModel): + """How the **daemon** logs — level, destination directory, retention. + + The daemon path (``trading-bot start [--serve]``) routes its lifecycle and + per-tick lines through the standard-library :mod:`logging` root logger, + wired by :func:`~trading_bot.application.log_setup.configure_daemon_logging` + to a midnight-rotated ``daemon.log`` (plus ``stderr``, so systemd/journal + still sees output). This model is the manifest's optional ``logging:`` + section that tunes that wiring; interactive CLI commands (``status`` / + ``kpi`` / ``canary`` / ...) keep their rich-console output and ignore it. + The section is **additive** — a manifest with no ``logging:`` key keeps + every default below. + + Parameters + ---------- + level : str, optional + Root log level, validated case-insensitively against the standard + library's level names (:func:`logging.getLevelNamesMapping` — ``DEBUG`` + / ``INFO`` / ``WARNING`` / ``ERROR`` / ``CRITICAL`` and their aliases) + and stored **upper-case**. Defaults to ``"INFO"`` — the engine's own + story, with the noisy third-party namespaces capped at ``WARNING`` by + the setup. + dir : pathlib.Path, optional + Directory the rotated ``daemon.log`` (and its dated backups) is written + to. A relative path (the default ``Path("logs")``) resolves against the + working directory the daemon runs from, like the other manifest paths; + it is created if missing. + retention_days : int, optional + Number of dated backups the midnight rotation keeps (``backupCount``); + older files are pruned. Must be ``>= 1``. Defaults to ``14``. + + """ + + level: str = "INFO" + dir: pathlib.Path = pathlib.Path("logs") + retention_days: int = Field(default=14, ge=1) + + @field_validator("level") + @classmethod + def _valid_level(cls, v: str) -> str: + """Accept any standard level name case-insensitively; store it upper-case.""" + upper = v.strip().upper() + if upper not in logging.getLevelNamesMapping(): + valid = ", ".join(sorted(logging.getLevelNamesMapping())) + raise ValueError(f"level {v!r} is not a valid log level (one of: {valid})") + return upper + + class AppConfig(BaseModel): """Top-level engine configuration — brokers, strategies and risk. @@ -757,6 +807,11 @@ class AppConfig(BaseModel): storage : StorageConfig, optional Where state is persisted (SQLite) and where the bars feed reads data from (dccd dir). Defaults to all-unset (each layer's own default). + logging : LoggingConfig, optional + How the **daemon** path logs (level / directory / retention) — consumed + by :func:`~trading_bot.application.log_setup.configure_daemon_logging`. + Additive: a manifest with no ``logging:`` section keeps the defaults + (INFO, ``logs/``, 14 days). Interactive CLI commands ignore it. Examples -------- @@ -787,6 +842,7 @@ class AppConfig(BaseModel): risk: RiskConfig = Field(default_factory=RiskConfig) storage: StorageConfig = Field(default_factory=StorageConfig) ui: UIConfig = Field(default_factory=UIConfig) + logging: LoggingConfig = Field(default_factory=LoggingConfig) @field_validator("display_currency") @classmethod diff --git a/trading_bot/application/log_setup.py b/trading_bot/application/log_setup.py new file mode 100644 index 0000000..d25b14f --- /dev/null +++ b/trading_bot/application/log_setup.py @@ -0,0 +1,148 @@ +"""Daemon logging spine — rotated, ISO-8601-timestamped, manifest-configurable. + +The interactive CLI commands (``status`` / ``kpi`` / ``canary`` / ...) render to +a rich console and are unaffected by this module. The **daemon** path +(``trading-bot start [--serve]``) is different: it runs headless for days, so its +story has to land somewhere durable. :func:`configure_daemon_logging` wires the +standard-library :mod:`logging` root logger — driven by the manifest's optional +``logging:`` section (:class:`~trading_bot.application.config.LoggingConfig`) — to + +* a :class:`~logging.handlers.TimedRotatingFileHandler` that rolls + ``/daemon.log`` at **midnight** and keeps ``retention_days`` dated + backups, and +* a :class:`~logging.StreamHandler` on ``stderr`` (so a systemd/journal unit — + or a plain ``nohup`` redirect — still captures the same lines). + +Both share a formatter whose timestamp is **local time with a numeric UTC +offset** (e.g. ``2026-07-14T13:35:40.123+02:00``): unambiguous across a DST +change and directly sortable/greppable. The engine's own namespaces log at the +configured ``level`` (INFO by default); the noisy third-party namespaces +(:data:`NOISY_NAMESPACES`) are capped at ``WARNING`` so INFO stays the engine's +story (an APScheduler *misfire* WARNING still flows). + +The setup is **idempotent**: it tags the handlers it installs +(:data:`OWNED_HANDLER_ATTR`) and removes its own stale handlers on a re-call, so +a second invocation never double-writes and never disturbs handlers attached by +anything else (pytest, an embedding host, ...). + +Secrets discipline: the formatter adds nothing beyond time / level / logger name +/ message — no request bodies, no credentials. Callers stay responsible for never +passing a credential-adjacent value into a log record. +""" + +from __future__ import annotations + +# Built-in +import logging +import sys +from datetime import datetime +from logging.handlers import TimedRotatingFileHandler + +# Local +from trading_bot.application.config import LoggingConfig + +__all__ = [ + "configure_daemon_logging", + "OWNED_HANDLER_ATTR", + "NOISY_NAMESPACES", +] + +#: Attribute stamped ``True`` on every handler this module installs, so a re-call +#: can find and drop exactly its own handlers (idempotence) without touching any +#: handler attached by something else. +OWNED_HANDLER_ATTR = "_trading_bot_daemon_owned" + +#: Third-party logger namespaces capped at ``WARNING`` so INFO stays the engine's +#: story rather than framework chatter. An APScheduler misfire WARNING (a real +#: signal the daemon fell behind) is >= WARNING, so it still reaches the file. +NOISY_NAMESPACES: tuple[str, ...] = ( + "apscheduler", + "uvicorn", + "uvicorn.access", + "httpx", + "websockets", +) + +#: The shared log-line layout: ISO timestamp, padded level, logger name, message. +_LOG_FORMAT = "%(asctime)s %(levelname)-8s %(name)s — %(message)s" + + +class _OffsetIsoFormatter(logging.Formatter): + """A :class:`logging.Formatter` whose ``asctime`` is local ISO-8601 with offset. + + Overrides :meth:`logging.Formatter.formatTime` to render the record's + creation time as local time carrying a **numeric UTC offset** (e.g. + ``2026-07-14T13:35:40.123+02:00``), millisecond precision. The offset is what + makes a line unambiguous across a DST change; ``datefmt`` is ignored on + purpose so every line is uniform. + """ + + def formatTime( # noqa: N802 - overriding the stdlib camelCase hook + self, record: logging.LogRecord, datefmt: str | None = None + ) -> str: + """Return the record's local creation time as ISO-8601 with numeric offset.""" + # A naive local datetime, then `.astimezone()` attaches the local offset. + return ( + datetime.fromtimestamp(record.created) + .astimezone() + .isoformat(timespec="milliseconds") + ) + + +def configure_daemon_logging(cfg: LoggingConfig) -> None: + """Wire the root logger for the daemon: rotated file + stderr, per ``cfg``. + + Attaches an owned :class:`~logging.handlers.TimedRotatingFileHandler` (rolls + ``cfg.dir / "daemon.log"`` at midnight, keeps ``cfg.retention_days`` backups) + and an owned :class:`~logging.StreamHandler` on ``stderr`` to the root logger, + both using the :class:`_OffsetIsoFormatter`; sets the root level to + ``cfg.level`` and caps :data:`NOISY_NAMESPACES` at ``WARNING``. Creates + ``cfg.dir`` if missing. + + Idempotent: any handler this module previously installed (tagged + :data:`OWNED_HANDLER_ATTR`) is removed and closed first, so a second call + leaves exactly one owned handler of each kind — no double-writing — and + handlers owned by anything else are left untouched. + + Parameters + ---------- + cfg : LoggingConfig + The resolved ``logging:`` section (level / directory / retention). + + Returns + ------- + None + + """ + cfg.dir.mkdir(parents=True, exist_ok=True) + root = logging.getLogger() + + # Idempotence: drop (and close) only the handlers this module installed + # before, so a re-call never accumulates duplicates. + for handler in list(root.handlers): + if getattr(handler, OWNED_HANDLER_ATTR, False): + root.removeHandler(handler) + handler.close() + + formatter = _OffsetIsoFormatter(_LOG_FORMAT) + + file_handler = TimedRotatingFileHandler( + cfg.dir / "daemon.log", + when="midnight", + backupCount=cfg.retention_days, + encoding="utf-8", + ) + file_handler.setFormatter(formatter) + setattr(file_handler, OWNED_HANDLER_ATTR, True) + root.addHandler(file_handler) + + stream_handler = logging.StreamHandler(sys.stderr) + stream_handler.setFormatter(formatter) + setattr(stream_handler, OWNED_HANDLER_ATTR, True) + root.addHandler(stream_handler) + + # `cfg.level` is a validated, upper-cased level name — `setLevel` accepts it. + root.setLevel(cfg.level) + + for namespace in NOISY_NAMESPACES: + logging.getLogger(namespace).setLevel(logging.WARNING) diff --git a/trading_bot/application/order_fill_sync.py b/trading_bot/application/order_fill_sync.py index b32519d..1c66e46 100644 --- a/trading_bot/application/order_fill_sync.py +++ b/trading_bot/application/order_fill_sync.py @@ -123,6 +123,13 @@ class OrderFillSync: :class:`~trading_bot.domain.order.Order` aggregates to update. bus : EventBus The shared bus to listen on and re-emit updated orders onto. + unit_name : str or None, optional + The managed unit this sync belongs to, threaded from + :func:`~trading_bot.application.service_factory.build_engine` for a + single-unit slice. When given, every applied-fill INFO line is prefixed + ``unit= ``; when ``None`` (the whole-system / multi-unit path, the + default — keeps the historic signature working) the same line is logged + without the prefix (starting at ``fill cid=…``). Examples -------- @@ -136,9 +143,12 @@ class OrderFillSync: """ - def __init__(self, router: OrderRouter, bus: EventBus) -> None: + def __init__( + self, router: OrderRouter, bus: EventBus, *, unit_name: str | None = None + ) -> None: self._router = router self._bus = bus + self._unit_name = unit_name # Fill-id dedup: every fill this component has consumed — applied, or # deliberately skipped (untracked/terminal on replay) — so a re-emitted # event or a replayed history never double-applies. Fills are the PnL @@ -302,4 +312,21 @@ def _apply(self, order: Order, fill: Fill) -> bool: exc, ) return False + # One INFO per applied fill — the execution the log's fill line reports + # (signed qty, price, fee) is exactly what reached the tracked order, so a + # reader can tie the daemon log to the store's fills table. The fee is + # denominated in ``fee_asset`` when the venue set one, else the quote + # currency. Prefixed ``unit= `` when this sync knows its unit; the + # same line without the prefix otherwise (see the ``unit_name`` param). + fee_ccy = fill.fee_asset or fill.instrument.symbol.quote + prefix = f"unit={self._unit_name} " if self._unit_name is not None else "" + logger.info( + "%sfill cid=%s %s @ %s fee=%s %s", + prefix, + fill.client_order_id, + fill.signed_qty, + fill.price, + fill.fee, + fee_ccy, + ) return True diff --git a/trading_bot/application/portfolio_runner.py b/trading_bot/application/portfolio_runner.py index 9a70508..ce7d49c 100644 --- a/trading_bot/application/portfolio_runner.py +++ b/trading_bot/application/portfolio_runner.py @@ -118,6 +118,7 @@ import time from collections.abc import Callable, Mapping from dataclasses import dataclass, field, replace +from datetime import datetime, timezone from typing import TYPE_CHECKING from trading_bot.application.events import EventBus, LogEvent @@ -169,6 +170,21 @@ def _now_ms() -> int: return int(time.time() * 1000) +def _asof_iso(asof_ms: int) -> str: + """Render an epoch-ms as-of as ISO-8601 UTC (e.g. ``2026-07-14T00:00:00+00:00``). + + The grep-stable timestamp the per-unit rebalance-summary log line carries — a + bar's as-of, not wall-clock — so a reader can tie the summary back to the exact + bar the tick evaluated. + """ + return datetime.fromtimestamp(asof_ms / 1000, tz=timezone.utc).isoformat() + + +def _order_px(order: Order) -> str: + """The order's price for a log line: its limit price, or ``mkt`` for a market order.""" + return "mkt" if order.limit_price is None else str(order.limit_price) + + @dataclass(frozen=True, slots=True) class RebalanceFailure: """One coin's leg that failed to route during a rebalance. @@ -526,7 +542,15 @@ async def rebalance(self, frames: Mapping[Symbol, pl.DataFrame]) -> RebalanceRes ) signal_by_symbol = {sig.instrument.symbol: sig for sig in signals} - submitted = 0 + # Per-outcome counters for the pinned rebalance-summary line. The four + # visible counters plus the (invisible) failure count partition the + # universe exactly: every coin is on-target, skipped, round-up, submitted + # or failed. ``submitted`` on the returned result stays the number of legs + # that actually routed (plain submits + round-ups), unchanged. + n_submit = 0 + n_round_up = 0 + n_skip = 0 + n_on_target = 0 failures: list[RebalanceFailure] = [] # Route in universe order for a deterministic per-tick leg sequence. for symbol in self._strategy.universe: @@ -538,20 +562,23 @@ async def rebalance(self, frames: Mapping[Symbol, pl.DataFrame]) -> RebalanceRes if delta == 0: # Already on target (incl. a flat target against a flat position): # no leg. + n_on_target += 1 continue + action = "submit" if self._spec_resolver is not None: # Venue-minimum order preparation (see the module docstring): # resolve the venue spec (cached after the first tick), run the # pure policy, and act on its decision. The order below still # carries the BARE instrument — the resolved spec shapes the # quantity only, never the tracker/risk keying. - delta = await self._prepare_delta( + delta, action = await self._prepare_delta( symbol, delta, position.net_qty, prices[symbol], step ) if delta == 0: # The policy skipped the leg (dust / capped below the # minimum / quantized to zero); already logged. + n_skip += 1 continue order = self._build_order(symbol, instrument, delta, prices[symbol], step) @@ -561,6 +588,16 @@ async def rebalance(self, frames: Mapping[Symbol, pl.DataFrame]) -> RebalanceRes # Per-leg failure: record it and continue the other legs (the # rebalance is not all-or-nothing — see the module docstring). failures.append(RebalanceFailure(symbol=symbol, error=exc)) + logger.warning( + "unit=%s order refused cid=%s %s %s %s @ %s reason=%s", + self._strategy.name, + order.client_order_id, + order.side.value, + order.qty, + symbol, + _order_px(order), + f"{type(exc).__name__}: {exc}", + ) if self._bus is not None: self._bus.emit( LogEvent( @@ -574,7 +611,20 @@ async def rebalance(self, frames: Mapping[Symbol, pl.DataFrame]) -> RebalanceRes ) continue - submitted += 1 + if action == "round_up": + n_round_up += 1 + else: + n_submit += 1 + logger.info( + "unit=%s order submitted cid=%s %s %s %s @ %s venue_id=%s", + self._strategy.name, + routed.client_order_id, + routed.side.value, + routed.qty, + symbol, + _order_px(routed), + routed.venue_order_id, + ) if self._bus is not None: self._bus.emit( LogEvent( @@ -587,7 +637,24 @@ async def rebalance(self, frames: Mapping[Symbol, pl.DataFrame]) -> RebalanceRes ) ) - return RebalanceResult(submitted=submitted, failures=failures) + # One INFO summary per *evaluated* rebalance (a new bar was consumed, + # whether or not orders resulted): the unit's story in a single grep-stable + # line. Idle daemon ticks never reach here — the freshness gate in + # `rebalance_latest` returns before calling `rebalance`, so a steady-state + # tick that consumes no new bar adds zero INFO lines (leaf 02 anti-spam). + logger.info( + "unit=%s rebalance asof=%s legs=%d submitted=%d round_up=%d " + "skipped=%d on_target=%d", + self._strategy.name, + _asof_iso(asof), + len(self._strategy.universe), + n_submit, + n_round_up, + n_skip, + n_on_target, + ) + + return RebalanceResult(submitted=n_submit + n_round_up, failures=failures) async def rebalance_latest(self) -> RebalanceResult | None: """Rebalance the book over the feed's **latest** cross-section — for a daemon. @@ -728,28 +795,33 @@ async def _prepare_delta( position_qty: Money, close: Money, step: int, - ) -> Money: - """Run one leg through the venue-minimum policy; return the final delta. + ) -> tuple[Money, str]: + """Run one leg through the venue-minimum policy; return ``(delta, action)``. Resolves the venue spec for ``(exchange, symbol)`` (cached by the resolver after the first tick), applies :func:`~trading_bot.application.order_prep.prepare_leg` to the signed ``delta`` and acts on the decision: - * ``skip`` → emits one info :class:`LogEvent` with the policy's reason - and returns ``0`` (the caller routes nothing — the residual is - recomputed naturally on the next rebalance); - * ``round_up`` → emits one info :class:`LogEvent` and returns the - bumped quantity on the original delta's side; + * ``skip`` → logs one INFO leg-decision line (module logger) + one info + :class:`LogEvent`, and returns ``(0, "skip")`` (the caller routes + nothing — the residual is recomputed naturally on the next rebalance); + * ``round_up`` → logs one INFO leg-decision line + one info + :class:`LogEvent`, and returns the bumped quantity on the original + delta's side with ``"round_up"``; * ``submit`` → returns the (lot-quantized, possibly sell-capped) - quantity on the original delta's side. - - A degraded resolver (a venue metadata fetch failed and fell back to a - bare instrument) additionally emits ONE warning :class:`LogEvent` per - runner lifetime; the leg itself degrades to the permissive path via the - bare spec. Never raises on venue-metadata trouble — the resolver - swallows fetch failures by contract, so a leg can never abort the - rebalance from here. + quantity on the original delta's side with ``"submit"`` (a plain + submit is *not* logged here — the summary line already tells that + story, so there is no double INFO per order). + + The returned ``action`` (``"submit"`` / ``"round_up"`` / ``"skip"``) lets + :meth:`rebalance` bucket the leg into the summary counters. A degraded + resolver (a venue metadata fetch failed and fell back to a bare + instrument) additionally emits ONE warning :class:`LogEvent` per runner + lifetime; the leg itself degrades to the permissive path via the bare + spec. Never raises on venue-metadata trouble — the resolver swallows + fetch failures by contract, so a leg can never abort the rebalance from + here. """ assert self._spec_resolver is not None and self._exchange is not None spec = await self._spec_resolver.resolve(self._exchange, symbol) @@ -778,6 +850,12 @@ async def _prepare_delta( price=close, ) if decision.action == "skip": + logger.info( + "unit=%s leg %s skip: %s", + self._strategy.name, + symbol, + decision.reason, + ) if self._bus is not None: self._bus.emit( LogEvent( @@ -788,20 +866,28 @@ async def _prepare_delta( level="info", ) ) - return _ZERO - if decision.action == "round_up" and self._bus is not None: - self._bus.emit( - LogEvent( - message=( - f"{self._strategy.name} step {step}: leg {symbol} " - f"rounded up to the venue minimum — {decision.reason}" - ), - level="info", - ) + return _ZERO, "skip" + if decision.action == "round_up": + logger.info( + "unit=%s leg %s round_up: %s", + self._strategy.name, + symbol, + decision.reason, ) + if self._bus is not None: + self._bus.emit( + LogEvent( + message=( + f"{self._strategy.name} step {step}: leg {symbol} " + f"rounded up to the venue minimum — {decision.reason}" + ), + level="info", + ) + ) # The policy returns the final ABSOLUTE quantity; it rides the original # delta's side (the policy never flips a leg's direction). - return decision.qty if delta > 0 else -decision.qty + final = decision.qty if delta > 0 else -decision.qty + return final, decision.action def _build_order( self, diff --git a/trading_bot/application/service_factory.py b/trading_bot/application/service_factory.py index 0c43075..95475e0 100644 --- a/trading_bot/application/service_factory.py +++ b/trading_bot/application/service_factory.py @@ -256,8 +256,11 @@ def build_engine( # confirmed fill to the router's *tracked* Order, so every filled order # freezes at OPEN/0 in the store and UI. Constructed right after the router # (it subscribes itself to the bus on construction), before the store - # attaches, so its re-emitted OrderEvent is what the store persists. - fill_sync = OrderFillSync(router, bus) + # attaches, so its re-emitted OrderEvent is what the store persists. Threaded + # the unit name (when this config is a single-unit slice — the supervisor's + # shape) so each applied-fill INFO line names its unit; None on the + # whole-system / multi-unit path (the line drops the ``unit=`` prefix). + fill_sync = OrderFillSync(router, bus, unit_name=_unit_name_of(config)) store: SqliteStore | None = None if db_path is not None: @@ -336,6 +339,27 @@ def genesis_v0(config: AppConfig) -> Money: return config.starting_capital +def _unit_name_of(config: AppConfig) -> str | None: + """The logical unit name of a single-unit slice, else ``None``. + + Mirrors :func:`genesis_v0`'s single-unit detection: when ``config`` declares + exactly one strategy (and no portfolio) or exactly one portfolio (and no + strategy) — the shape the + :class:`~trading_bot.application.supervisor.StrategySupervisor` builds every + unit's engine from — the unit's name is returned. A multi-unit or empty config + (the whole-system :func:`~trading_bot.application.run_app.run_app` path) + yields ``None``, so its :class:`~trading_bot.application.order_fill_sync. + OrderFillSync` fill lines carry no unit prefix. + """ + strategies = config.strategies + portfolios = config.portfolios + if len(strategies) == 1 and not portfolios: + return strategies[0].name + if len(portfolios) == 1 and not strategies: + return portfolios[0].name + return None + + def _build_broker(config: AppConfig, bus: EventBus) -> Broker: """Select and construct the broker for ``config`` — paper-by-default. diff --git a/trading_bot/application/supervisor.py b/trading_bot/application/supervisor.py index 13a9c56..def9101 100644 --- a/trading_bot/application/supervisor.py +++ b/trading_bot/application/supervisor.py @@ -36,6 +36,7 @@ from __future__ import annotations import asyncio +import logging import time from collections.abc import Callable, Sequence from dataclasses import dataclass, field @@ -109,6 +110,8 @@ _ZERO: Money = money("0") +logger = logging.getLogger(__name__) + #: The accounting checker's cache TTL, in milliseconds (see ``_accounting_of``). #: The dashboard polls every 10 s; a 60 s window means the checker recomputes #: at most once a minute per unit, and only while something is actually @@ -164,6 +167,14 @@ class StrategyStatus: .last_asof_ms` / :attr:`~trading_bot.application.portfolio_runner .PortfolioRunner.last_asof_ms`), i.e. "the data this strategy last computed on". ``None`` before the first completed evaluation. + last_step_duration_ms : int or None + Wall-clock duration (**milliseconds**) of the unit's most recent + :meth:`StrategySupervisor.step` — the time its runner spent evaluating + one tick (data drain + signal + any routing), measured with a monotonic + clock. ``None`` before the unit has ever been stepped. Recorded on every + step (even one that raised), so a dashboard can spot a unit whose + evaluation is creeping toward the tick interval (the tick-timing health + signal — see ``doc/dev/plans/daemon-logging/03-tick-timing-metrics.md``). allocation : Money or None The unit's **genesis** capital — its declared ``allocation`` (or a portfolio's ``allocation``/``capital``). ``None`` for a single-instrument @@ -215,6 +226,7 @@ class StrategyStatus: open_orders: int last_eval_ts: int | None = None last_asof_ts: int | None = None + last_step_duration_ms: int | None = None allocation: Money | None = None contributed: Money | None = None unrealised: Money | None = None @@ -501,6 +513,12 @@ class _Unit: #: are cleared by :meth:`StrategySupervisor._teardown`) so a stopped unit still #: reports its last-known state. accounting: _AccountingState | None = None + #: Wall-clock duration (ms) of this unit's most recent :meth:`StrategySupervisor + #: .step` — the time its runner spent on one evaluation. ``None`` before the + #: unit has ever been stepped; recorded on every step (incl. one that raised), + #: and (like :attr:`accounting`) it survives ``stop`` so a stopped unit still + #: reports the last duration it measured. + last_step_duration_ms: int | None = None class StrategySupervisor: @@ -850,6 +868,13 @@ async def _start_locked(self, unit: _Unit) -> None: unit.runner = pruns[0] unit.engine = engine unit.running = True + logger.info( + "unit=%s started mode=%s exchange=%s kind=%s", + unit.name, + unit.mode, + unit.exchange, + unit.kind, + ) @staticmethod def _replay_paper_book(engine: Engine) -> None: @@ -1025,7 +1050,14 @@ def _teardown(unit: _Unit) -> None: (drop the runner + engine, flip ``running`` off); the store is drained first by :meth:`_drain_store` (async paths) or a direct blocking close (:meth:`remove_unit`) so no enqueued write is lost. + + Logs one INFO transition line **only when the unit was actually running** + — the single teardown path covers ``stop`` / ``set_mode`` restart / + ``remove_unit``, and an idempotent stop of an already-stopped unit stays + silent (no phantom transition). """ + if unit.running: + logger.info("unit=%s stopped", unit.name) unit.running = False unit.runner = None unit.engine = None @@ -1508,10 +1540,36 @@ async def step(self, name: str) -> Order | object | None: if not unit.running or unit.runner is None: return None runner = unit.runner - if isinstance(runner, StrategyRunner): - return await runner.step_latest() - assert isinstance(runner, PortfolioRunner) - return await runner.rebalance_latest() + # Time the runner's evaluation (data drain + signal + any routing) with a + # monotonic clock, and stamp it on the unit's runtime state in a `finally` + # so a slow-and-then-raising step is still measured (the duration is a + # health signal, not a success signal). Per-unit duration lives ONLY on + # this field, not the leaf-02 rebalance-summary line: that line is emitted + # from *inside* the runner, before this outer step timing exists, so + # threading the total back into it would mean the summary timing itself — + # the runner is deliberately not contorted for it (leaf 03, step 3). + started = time.monotonic() + try: + if isinstance(runner, StrategyRunner): + result: Order | object | None = await runner.step_latest() + else: + assert isinstance(runner, PortfolioRunner) + result = await runner.rebalance_latest() + except Exception: + # A unit's evaluation blew up: attribute the traceback to this unit + # (the CLI tick's own catch only knows "a tick failed"), then re-raise + # so the daemon's outer safety net still sees it. A step error is rare, + # never steady-state — this is not an anti-spam concern. + logger.exception("unit=%s step error", name) + raise + finally: + unit.last_step_duration_ms = int((time.monotonic() - started) * 1000) + if result is None: + # Evaluated nothing this tick (freshness gate skip / no new bar): a + # DEBUG line only, so a steady-state INFO log stays free of per-unit + # idle chatter (leaf 02 anti-spam). + logger.debug("unit=%s step evaluated nothing (no new bar)", name) + return result async def start_all(self) -> None: """Start every managed unit (the daemon's boot — each in its config mode).""" @@ -1594,6 +1652,7 @@ def _status_of(self, unit: _Unit) -> StrategyStatus: open_orders=open_orders, last_eval_ts=last_eval_ts, last_asof_ts=last_asof_ts, + last_step_duration_ms=unit.last_step_duration_ms, allocation=allocation, contributed=contributed, unrealised=unrealised, diff --git a/trading_bot/interfaces/api/app.py b/trading_bot/interfaces/api/app.py index 8c4775d..394705b 100644 --- a/trading_bot/interfaces/api/app.py +++ b/trading_bot/interfaces/api/app.py @@ -1163,7 +1163,9 @@ def _status_dict(status: StrategyStatus, config: AppConfig) -> dict[str, Any]: ``last_eval_ts`` / ``last_asof_ts`` are already integer epoch ms (or ``None``) on the domain side — no stringification needed, unlike the money - fields. + fields. ``last_step_duration_ms`` (the unit's most-recent step duration in + ms, ``None`` before its first step) is likewise a bare int — an additive + tick-timing field (leaf 03), never displacing a pre-existing one. ``display_currency`` (resolved for the unit's ``exchange``) + ``total_value_display`` / ``unrealised_display`` (converted from the @@ -1186,6 +1188,7 @@ def _status_dict(status: StrategyStatus, config: AppConfig) -> dict[str, Any]: "open_orders": status.open_orders, "last_eval_ts": status.last_eval_ts, "last_asof_ts": status.last_asof_ts, + "last_step_duration_ms": status.last_step_duration_ms, "allocation": _money_str(status.allocation), "contributed": _money_str(status.contributed), "unrealised": _money_str(status.unrealised), @@ -1738,12 +1741,15 @@ def create_dashboard_app( written. Typically ``lambda: supervisor.manifest().to_yaml(path)``. schedule_info : Callable[[], dict] or None, optional A hook returning ``{"next_tick_ts": , "tick": }`` — the daemon's scheduler cadence, surfaced on ``/api/health``. + or None>, "last_tick_duration_ms": , "last_tick_ts": , "ticks_total": , "ticks_overrun": }`` — the daemon's + scheduler cadence + tick-timing metrics, surfaced on ``/api/health``. Keeps this app **scheduler-agnostic**: only the daemon (``_run_daemon``) has an ``apscheduler`` job to report, so it injects this hook; the plain - ``dashboard`` command (no scheduler) passes ``None`` and both fields stay - ``null``. The hook is called under a ``try``/``except`` — a raising or - absent hook degrades to ``null``/``null``, never breaking health. + ``dashboard`` command (no scheduler) passes ``None`` and the cadence + fields stay ``null``/``0``. The hook is called under a ``try``/``except`` + — a raising or absent hook degrades to those same defaults, never + breaking health. Returns ------- @@ -1849,7 +1855,10 @@ async def health(request: Request) -> dict[str, Any]: from the ``schedule_info`` hook when one was injected (the daemon's cadence); both stay ``null`` with no hook, and a hook that raises degrades to ``null``/``null`` too — health must never 500 because the - scheduler hiccuped. + scheduler hiccuped. The same hook feeds the additive tick-timing fields + ``last_tick_duration_ms`` / ``last_tick_ts`` (``null`` before the first + tick / no hook) and ``ticks_total`` / ``ticks_overrun`` (``0`` before the + first tick / no hook) — the realised-vs-nominal cadence, in numbers. ``worst`` is the worst per-unit :attr:`~trading_bot.application. supervisor.StrategyStatus.health` among **running** units @@ -1861,15 +1870,31 @@ async def health(request: Request) -> dict[str, Any]: sup = _sup(request) next_tick_ts: int | None = None tick: str | None = None + # Additive tick-timing fields (leaf 03), fed by the same daemon hook: the + # last tick's duration + wall-clock, and the running / overrun tick counts. + # `None`/`0` with no hook (the scheduler-agnostic `dashboard` command) or a + # hook that raises — health degrades, never 500s. + last_tick_duration_ms: int | None = None + last_tick_ts: int | None = None + ticks_total = 0 + ticks_overrun = 0 hook = request.app.state.schedule_info if hook is not None: try: info = hook() or {} next_tick_ts = info.get("next_tick_ts") tick = info.get("tick") + last_tick_duration_ms = info.get("last_tick_duration_ms") + last_tick_ts = info.get("last_tick_ts") + ticks_total = info.get("ticks_total", 0) + ticks_overrun = info.get("ticks_overrun", 0) except Exception: # noqa: BLE001 - health must degrade, never 500 next_tick_ts = None tick = None + last_tick_duration_ms = None + last_tick_ts = None + ticks_total = 0 + ticks_overrun = 0 statuses = sup.status() worst = _worst_health(s.health for s in statuses if s.running) unhealthy = sum(1 for s in statuses if s.health != "ok") @@ -1880,6 +1905,10 @@ async def health(request: Request) -> dict[str, Any]: "read_only": request.app.state.read_only, "next_tick_ts": next_tick_ts, "tick": tick, + "last_tick_duration_ms": last_tick_duration_ms, + "last_tick_ts": last_tick_ts, + "ticks_total": ticks_total, + "ticks_overrun": ticks_overrun, "worst": worst, "unhealthy": unhealthy, } diff --git a/trading_bot/interfaces/cli/main.py b/trading_bot/interfaces/cli/main.py index 42a52f6..3c88793 100644 --- a/trading_bot/interfaces/cli/main.py +++ b/trading_bot/interfaces/cli/main.py @@ -47,12 +47,14 @@ import asyncio import contextlib import dataclasses +import logging import math import os import pathlib import signal import sys import tempfile +import time from collections.abc import Callable from decimal import Decimal from typing import TYPE_CHECKING, Any @@ -71,6 +73,7 @@ from trading_bot.application.config import AppConfig, BrokerConfig, StrategyConfig from trading_bot.application.data_feed import BARS_SCHEMA, InMemoryFeed from trading_bot.application.instrument_specs import InstrumentSpecResolver +from trading_bot.application.log_setup import configure_daemon_logging from trading_bot.application.performance_service import PerformanceService from trading_bot.application.run_app import run_app from trading_bot.application.service_factory import Engine, build_engine @@ -105,6 +108,11 @@ #: The shared rich console every command prints through. _console = Console() +#: The daemon's lifecycle/tick logger. Only the daemon path (``start [--serve]``) +#: emits through it, after :func:`~trading_bot.application.log_setup.configure_daemon_logging` +#: has wired the root handlers; interactive commands print to ``_console`` instead. +_daemon_logger = logging.getLogger("trading_bot.daemon") + #: Default synthetic-feed length (bars) when no ``--bars`` file is given — long #: enough for the default 10/30 MA windows to warm up and cross at least once. _SYNTHETIC_BARS = 80 @@ -1210,6 +1218,69 @@ def serve( # --- start (daemon) -------------------------------------------------------- # +@dataclasses.dataclass +class _TickMetrics: + """The daemon's rolling per-tick timing counters (``_run_daemon`` closure state). + + Owned by the daemon (not the served app): the plain ``dashboard`` command has + no scheduler and so no metrics, exactly like ``next_tick_ts``. A single + instance lives in :func:`_run_daemon`; each ``_tick`` folds its measured + duration in via :meth:`record`, and :func:`_run_daemon`'s ``_schedule_info`` + hook reads the fields out onto ``/api/health``. + + Attributes + ---------- + last_tick_duration_ms : int or None + Wall-clock duration (ms) of the most recent tick's ``step_all``. ``None`` + before the first tick. + last_tick_ts : int or None + Epoch-ms wall-clock time the most recent tick finished. ``None`` before + the first tick. + ticks_total : int + How many ticks have run since the daemon started (a tick that raised in + ``step_all`` still counts — it did fire). ``0`` before the first tick. + ticks_overrun : int + How many of those ticks ran **longer than the configured interval** + (never incremented for a cron trigger, where an interval overrun is not + defined). ``0`` before the first tick / when no tick has overrun. + """ + + last_tick_duration_ms: int | None = None + last_tick_ts: int | None = None + ticks_total: int = 0 + ticks_overrun: int = 0 + + def record( + self, *, duration_ms: int, now_ms: int, interval_seconds: float | None + ) -> bool: + """Fold one completed tick's timing in; return whether it overran the interval. + + Parameters + ---------- + duration_ms : int + The tick's measured wall-clock duration in milliseconds. + now_ms : int + Epoch-ms wall-clock time the tick finished. + interval_seconds : float or None + The configured interval in seconds; ``None`` for a cron trigger, where + an "overrun" against a fixed interval is undefined — the duration and + counters are still recorded, but ``ticks_overrun`` never increments. + + Returns + ------- + bool + ``True`` when the tick overran (``duration_ms`` exceeds + ``interval_seconds``), else ``False``. + """ + self.last_tick_duration_ms = duration_ms + self.last_tick_ts = now_ms + self.ticks_total += 1 + overran = interval_seconds is not None and duration_ms > interval_seconds * 1000 + if overran: + self.ticks_overrun += 1 + return overran + + async def _run_daemon( config: AppConfig, *, @@ -1265,18 +1336,56 @@ async def _run_daemon( from trading_bot.application.supervisor import StrategySupervisor + # Wire the durable logging spine before anything logs: the daemon runs + # headless for days, so its lifecycle/tick lines (and any WARNING from the + # runners/router) must land in the rotated file, not just fly past on stderr. + configure_daemon_logging(config.logging) + supervisor = StrategySupervisor(config, dccd_client=dccd_client) # type: ignore[arg-type] await supervisor.start_all() + # The tick-timing counters (health signal, exposed on `/api/health` via the + # `_schedule_info` hook below). An interval trigger has a well-defined overrun + # threshold (the interval); a cron trigger does not, so its overrun check is + # disabled (`interval_seconds = None`) while duration/counters still record. + metrics = _TickMetrics() + interval_seconds: float | None = None if cron is not None else interval + async def _tick() -> None: + started = time.monotonic() + stepped = 0 try: stepped = await supervisor.step_all() - if stepped: - _console.print( - f"[dim]daemon tick: stepped {stepped} strategy(ies)[/dim]" - ) except Exception as exc: # noqa: BLE001 - never let a tick kill the daemon - _console.print(f"[red]daemon tick error:[/red] {exc}") + # `exception` captures the traceback into the file (invisible before). + _daemon_logger.exception("daemon tick error: %s", exc) + # Measure every tick — even one that raised (the duration is a health + # signal, not a success signal); the errored tick still counts as a tick. + duration_ms = int((time.monotonic() - started) * 1000) + overran = metrics.record( + duration_ms=duration_ms, + now_ms=int(time.time() * 1000), + interval_seconds=interval_seconds, + ) + duration_s = duration_ms / 1000 + if stepped: + # Daemon-only noise: to the log file, not the console. The leaf-01 + # heartbeat now carries the tick's duration. + _daemon_logger.info( + "daemon tick: stepped %d strategy(ies) in %.3fs", stepped, duration_s + ) + if overran: + # An interval tick that outran its interval: the next tick would have + # queued behind it — `max_instances=1` + `coalesce=True` collapse the + # backlog into one run, but the operator should know the cadence is + # slipping (the 74 s-vs-60 s health question this leaf answers). + _daemon_logger.warning( + "daemon tick overrun: %.3fs > %gs " + "(evaluation outran the tick interval; the backlog coalesces " + "under max_instances=1)", + duration_s, + interval_seconds, + ) trigger = ( CronTrigger.from_crontab(cron) @@ -1284,26 +1393,62 @@ async def _tick() -> None: else IntervalTrigger(seconds=interval) ) scheduler = AsyncIOScheduler() - job = scheduler.add_job(_tick, trigger) + # Pin the tick's overlap/misfire semantics explicitly (previously all + # APScheduler defaults): + # max_instances=1 — two ticks must never race one engine; + # coalesce=True — a backlog of missed runs collapses into a single run; + # misfire_grace_time — how late a tick may still run. APScheduler's default + # grace is 1 s, so a tick even ~1 s late is silently + # *skipped* (the presumed source of the ~74 s-vs-60 s + # shortfall a health audit measured); grant the whole + # interval instead so a late tick still runs. A cron + # trigger has no fixed interval to derive a grace from, + # so use a fixed, sensible 30 s. APScheduler requires a + # positive int, so floor the interval at 1 s. + misfire_grace_time = 30 if cron is not None else max(1, int(interval)) + job = scheduler.add_job( + _tick, + trigger, + coalesce=True, + max_instances=1, + misfire_grace_time=misfire_grace_time, + ) scheduler.start() + _tick_desc = cron or f"every {interval:g}s" + _daemon_logger.info( + "daemon started (mode=%s): %d strateg(ies), tick=%s", + config.mode, + len(supervisor.names()), + _tick_desc, + ) _console.print( f"[green]daemon started[/green] (mode={config.mode}): " f"{len(supervisor.names())} strateg(ies), " - f"tick={cron or f'every {interval:g}s'}" + f"tick={_tick_desc}" ) def _schedule_info() -> dict[str, Any]: - """The scheduler's cadence, for the dashboard's ``/api/health`` hook. + """The scheduler's cadence + tick-timing metrics, for the ``/api/health`` hook. ``next_run_time`` is a tz-aware ``datetime`` (or ``None`` between ticks / once exhausted); converted to epoch **ms** for JSON. ``tick`` is the same - human trigger description the startup banner above prints. + human trigger description the startup banner above prints. The four + ``*_tick_*`` / ``ticks_*`` keys are the live :class:`_TickMetrics` counters + (``None``/``0`` before the first tick) — additive fields on + ``/api/health`` proving the realised cadence against the nominal one. """ next_run = job.next_run_time next_tick_ts = ( int(next_run.timestamp() * 1000) if next_run is not None else None ) - return {"next_tick_ts": next_tick_ts, "tick": cron or f"every {interval:g}s"} + return { + "next_tick_ts": next_tick_ts, + "tick": cron or f"every {interval:g}s", + "last_tick_duration_ms": metrics.last_tick_duration_ms, + "last_tick_ts": metrics.last_tick_ts, + "ticks_total": metrics.ticks_total, + "ticks_overrun": metrics.ticks_overrun, + } try: if serve: @@ -1397,6 +1542,7 @@ def _on_shutdown_signal() -> None: finally: scheduler.shutdown(wait=False) await supervisor.shutdown() + _daemon_logger.info("daemon stopped (all strategies shut down)") _console.print("[green]daemon stopped[/green] (all strategies shut down)") diff --git a/trading_bot/tests/application/test_config.py b/trading_bot/tests/application/test_config.py index 9dd9a58..bec4f6b 100644 --- a/trading_bot/tests/application/test_config.py +++ b/trading_bot/tests/application/test_config.py @@ -26,7 +26,7 @@ StorageConfig, StrategyConfig, ) -from trading_bot.application.config import PortfolioStrategyConfig +from trading_bot.application.config import LoggingConfig, PortfolioStrategyConfig #: The shipped runnable paper config (resolved from the repo root). EXAMPLE_CONFIG = ( @@ -860,3 +860,89 @@ def test_ui_config_loopback_without_token_is_allowed() -> None: cfg = AppConfig.model_validate({"mode": "paper", "ui": {"host": host}}) assert cfg.ui.host == host assert cfg.ui.token is None + + +# --- LoggingConfig (daemon logging spine — level/dir/retention) ------------ # + + +def test_logging_config_defaults() -> None: + """A bare ``LoggingConfig`` is INFO / ``logs`` / 14-day retention.""" + cfg = LoggingConfig() + assert cfg.level == "INFO" + assert cfg.dir == pathlib.Path("logs") + assert cfg.retention_days == 14 + + +def test_app_config_logging_defaults_when_section_absent() -> None: + """A manifest with no ``logging:`` section keeps the defaults (additive).""" + cfg = AppConfig.model_validate({"mode": "paper"}) + assert isinstance(cfg.logging, LoggingConfig) + assert cfg.logging.level == "INFO" + assert cfg.logging.dir == pathlib.Path("logs") + assert cfg.logging.retention_days == 14 + + +def test_logging_level_is_upper_cased_case_insensitively() -> None: + """A lower-case ``level`` validates and is stored upper-case.""" + cfg = AppConfig.model_validate({"logging": {"level": "debug"}}) + assert cfg.logging.level == "DEBUG" + + +def test_logging_invalid_level_rejected() -> None: + """A ``level`` that is not a standard log-level name is rejected.""" + with pytest.raises(ValidationError): + AppConfig.model_validate({"logging": {"level": "verbose"}}) + + +def test_logging_retention_days_zero_rejected() -> None: + """``retention_days`` must be ``>= 1`` — zero is rejected.""" + with pytest.raises(ValidationError): + AppConfig.model_validate({"logging": {"retention_days": 0}}) + + +def test_logging_retention_days_negative_rejected() -> None: + """A negative ``retention_days`` is rejected.""" + with pytest.raises(ValidationError): + AppConfig.model_validate({"logging": {"retention_days": -1}}) + + +def test_logging_section_parses_from_yaml(tmp_path) -> None: # noqa: ANN001 + """A ``logging:`` section parses (level upper-cased, dir/retention set).""" + path = tmp_path / "with-logging.yml" + path.write_text( + textwrap.dedent( + """\ + mode: paper + logging: + level: debug + dir: /var/log/trading_bot + retention_days: 30 + """ + ) + ) + cfg = AppConfig.from_yaml(path) + assert cfg.logging.level == "DEBUG" + assert cfg.logging.dir == pathlib.Path("/var/log/trading_bot") + assert cfg.logging.retention_days == 30 + + +def test_logging_absent_yaml_section_yields_defaults(tmp_path) -> None: # noqa: ANN001 + """A YAML manifest without ``logging:`` yields the default LoggingConfig.""" + path = tmp_path / "no-logging.yml" + path.write_text("mode: paper\n") + cfg = AppConfig.from_yaml(path) + assert cfg.logging.level == "INFO" + assert cfg.logging.dir == pathlib.Path("logs") + assert cfg.logging.retention_days == 14 + + +def test_logging_section_round_trips_through_yaml(tmp_path) -> None: # noqa: ANN001 + """The ``logging:`` section survives a ``to_yaml``/``from_yaml`` round-trip.""" + cfg = AppConfig.model_validate( + {"mode": "paper", "logging": {"level": "warning", "retention_days": 7}} + ) + path = tmp_path / "m.yaml" + cfg.to_yaml(path) + reloaded = AppConfig.from_yaml(path) + assert reloaded.logging.level == "WARNING" + assert reloaded.logging.retention_days == 7 diff --git a/trading_bot/tests/application/test_display_ccy.py b/trading_bot/tests/application/test_display_ccy.py index 88af182..6373c58 100644 --- a/trading_bot/tests/application/test_display_ccy.py +++ b/trading_bot/tests/application/test_display_ccy.py @@ -384,6 +384,7 @@ def test_api_strategies_display_fields_and_contract_regression() -> None: "open_orders", "last_eval_ts", "last_asof_ts", + "last_step_duration_ms", "allocation", "contributed", "unrealised", diff --git a/trading_bot/tests/application/test_log_setup.py b/trading_bot/tests/application/test_log_setup.py new file mode 100644 index 0000000..ecd2d4b --- /dev/null +++ b/trading_bot/tests/application/test_log_setup.py @@ -0,0 +1,167 @@ +"""Tests for :func:`~trading_bot.application.log_setup.configure_daemon_logging`. + +These prove the daemon logging spine wires as intended: the root logger gains +exactly one owned rotated-file handler and one owned stderr handler, a second +call does not double them (idempotence), each line is ISO-8601 with a numeric UTC +offset, the configured level reaches the file while the noisy third-party +namespaces are capped at WARNING, and the log directory/file are created under +the manifest-declared ``dir``. + +Test hygiene: :func:`configure_daemon_logging` mutates *process-global* logging +state (the root logger's handlers/level and a few namespace levels). The +``restore_root_logging`` fixture snapshots and restores all of it around every +test so the wider suite's own logging is never polluted. +""" + +from __future__ import annotations + +import logging +import re +from collections.abc import Iterator +from logging.handlers import TimedRotatingFileHandler +from pathlib import Path + +import pytest + +from trading_bot.application.config import LoggingConfig +from trading_bot.application.log_setup import ( + NOISY_NAMESPACES, + OWNED_HANDLER_ATTR, + configure_daemon_logging, +) + +#: A regex for one emitted line: ISO-8601 local time with a numeric UTC offset +#: (e.g. ``2026-07-14T13:35:40.123+02:00``) followed by a space then the level. +_ISO_OFFSET_LINE = re.compile( + r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}[+-]\d{2}:\d{2} " +) + + +@pytest.fixture +def restore_root_logging() -> Iterator[None]: + """Snapshot and restore the process-global logging state around a test. + + Saves the root logger's handlers + level and every capped-namespace level, + then on teardown removes/closes any handler the test installed and puts the + original handlers, root level and namespace levels back — so a test that + wires the daemon logging never leaks handlers into the rest of the suite. + """ + root = logging.getLogger() + saved_handlers = root.handlers[:] + saved_level = root.level + saved_ns_levels = {ns: logging.getLogger(ns).level for ns in NOISY_NAMESPACES} + try: + yield + finally: + for handler in root.handlers[:]: + if handler not in saved_handlers: + root.removeHandler(handler) + handler.close() + for handler in saved_handlers: + if handler not in root.handlers: + root.addHandler(handler) + root.setLevel(saved_level) + for ns, level in saved_ns_levels.items(): + logging.getLogger(ns).setLevel(level) + + +def _owned(kind: type[logging.Handler]) -> list[logging.Handler]: + """The root logger's owned handlers whose *exact* type is ``kind``.""" + return [ + h + for h in logging.getLogger().handlers + if getattr(h, OWNED_HANDLER_ATTR, False) and type(h) is kind + ] + + +def test_configure_wires_one_owned_file_and_stream_handler( + tmp_path: Path, restore_root_logging: None +) -> None: + """After configure, root has exactly one owned rotated-file + one stderr handler.""" + cfg = LoggingConfig(dir=tmp_path / "logs", retention_days=7) + configure_daemon_logging(cfg) + + file_handlers = _owned(TimedRotatingFileHandler) + stream_handlers = _owned(logging.StreamHandler) + assert len(file_handlers) == 1 + assert len(stream_handlers) == 1 + + fh = file_handlers[0] + assert isinstance(fh, TimedRotatingFileHandler) + assert fh.when == "MIDNIGHT" + assert fh.backupCount == 7 + + +def test_configure_is_idempotent(tmp_path: Path, restore_root_logging: None) -> None: + """A second configure leaves exactly one owned handler of each kind (no doubling).""" + cfg = LoggingConfig(dir=tmp_path / "logs") + configure_daemon_logging(cfg) + configure_daemon_logging(cfg) + + assert len(_owned(TimedRotatingFileHandler)) == 1 + assert len(_owned(logging.StreamHandler)) == 1 + + +def test_formatter_is_iso_with_numeric_offset( + tmp_path: Path, restore_root_logging: None +) -> None: + """An emitted line is ISO-8601 with a numeric offset and carries level + name.""" + configure_daemon_logging(LoggingConfig(dir=tmp_path / "logs")) + logging.getLogger("trading_bot.daemon.test").info("hello world") + + line = _last_log_line(tmp_path / "logs" / "daemon.log") + assert _ISO_OFFSET_LINE.match(line), line + assert "INFO" in line + assert "trading_bot.daemon.test" in line + assert "hello world" in line + + +def test_level_debug_reaches_file_and_noisy_namespaces_capped( + tmp_path: Path, restore_root_logging: None +) -> None: + """``level=DEBUG`` lets a DEBUG record reach the file; apscheduler stays WARNING.""" + configure_daemon_logging(LoggingConfig(dir=tmp_path / "logs", level="DEBUG")) + + logging.getLogger("trading_bot.daemon.test").debug("a debug line") + contents = (tmp_path / "logs" / "daemon.log").read_text() + assert "a debug line" in contents + + # The noisy namespaces are floored at WARNING regardless of the root level. + assert logging.getLogger("apscheduler").level == logging.WARNING + for ns in NOISY_NAMESPACES: + assert logging.getLogger(ns).level == logging.WARNING + + +def test_default_level_info_drops_debug( + tmp_path: Path, restore_root_logging: None +) -> None: + """At the default INFO level, a DEBUG record does not reach the file.""" + configure_daemon_logging(LoggingConfig(dir=tmp_path / "logs")) + + logging.getLogger("trading_bot.daemon.test").debug("should-not-appear") + logging.getLogger("trading_bot.daemon.test").info("should-appear") + contents = (tmp_path / "logs" / "daemon.log").read_text() + assert "should-not-appear" not in contents + assert "should-appear" in contents + + +def test_log_dir_and_file_created(tmp_path: Path, restore_root_logging: None) -> None: + """A missing (nested) ``dir`` is created and the log file appears on first emit.""" + log_dir = tmp_path / "nested" / "logs" + assert not log_dir.exists() + + configure_daemon_logging(LoggingConfig(dir=log_dir)) + assert log_dir.is_dir() + + logging.getLogger("trading_bot.daemon.test").info("x") + assert (log_dir / "daemon.log").is_file() + + +def _last_log_line(path: Path) -> str: + """Flush the owned handlers and return the last non-empty line of ``path``.""" + for handler in logging.getLogger().handlers: + if getattr(handler, OWNED_HANDLER_ATTR, False): + handler.flush() + lines = [ln for ln in path.read_text().splitlines() if ln.strip()] + assert lines, f"no log lines written to {path}" + return lines[-1] diff --git a/trading_bot/tests/application/test_order_fill_sync.py b/trading_bot/tests/application/test_order_fill_sync.py index 02ee48c..be2c9f7 100644 --- a/trading_bot/tests/application/test_order_fill_sync.py +++ b/trading_bot/tests/application/test_order_fill_sync.py @@ -26,8 +26,12 @@ from __future__ import annotations # Built-in +import logging import pathlib +# Third-party +import pytest + # Local from trading_bot.application.events import EventBus, FillEvent, OrderEvent from trading_bot.application.order_fill_sync import OrderFillSync @@ -199,6 +203,79 @@ async def test_no_event_loop() -> None: assert all(e.order.client_order_id == "cid-1" for e in events) +# --- leaf 02: per-fill INFO trail (module logger) --------------------------- # + +_FILL_SYNC_LOGGER = "trading_bot.application.order_fill_sync" + + +async def test_applied_fill_logs_unit_prefixed_pinned_line( + caplog: pytest.LogCaptureFixture, +) -> None: + """A unit-named sync logs the pinned ``unit= fill ...`` line incl fee_asset. + + A partial submit leaves the order tracked & fillable; a controlled follow-up + fill (explicit ``fee_asset``) then applies on the live path and logs one INFO + line: ``unit=u1 fill cid=… @ fee= ``. + """ + bus = EventBus() + broker = PaperBroker( + event_bus=bus, + fill_model="partial", + partial_chunks=2, + partial_fill_ratio=money("0.5"), + ) + router = OrderRouter(broker, bus) + OrderFillSync(router, bus, unit_name="u1") + + with caplog.at_level(logging.INFO, logger=_FILL_SYNC_LOGGER): + order = _order(qty="2") + await router.submit(order) # → PARTIALLY_FILLED, tracked & fillable + assert order.status is OrderStatus.PARTIALLY_FILLED + caplog.clear() # drop the paper partial-fill lines; keep only ours + bus.emit( + FillEvent( + Fill( + fill_id="F-EXPLICIT", + client_order_id="cid-1", + instrument=BTC_USD, + side=OrderSide.BUY, + qty=money("1"), + price=money("30100"), + fee=money("0.5"), + fee_asset="BNB", + ts=1_700_000_000_100, + ) + ) + ) + + lines = [r.getMessage() for r in caplog.records if r.name == _FILL_SYNC_LOGGER] + fill_lines = [m for m in lines if m.startswith("unit=u1 fill cid=cid-1")] + assert len(fill_lines) == 1 + assert "@ 30100" in fill_lines[0] + assert "fee=0.5 BNB" in fill_lines[0] # the explicit fee_asset, not the quote + + +async def test_applied_fill_logs_without_prefix_when_no_unit_name( + caplog: pytest.LogCaptureFixture, +) -> None: + """A ``unit_name=None`` sync logs the same line without the ``unit=`` prefix. + + Fee asset defaults to the instrument's quote currency (``USD``) when the fill + carries none. + """ + _bus, _broker, router, _sync = _wire() # default sync: unit_name=None + + with caplog.at_level(logging.INFO, logger=_FILL_SYNC_LOGGER): + await router.submit(_order()) # paper immediate fill → applied + + lines = [r.getMessage() for r in caplog.records if r.name == _FILL_SYNC_LOGGER] + fill_lines = [m for m in lines if m.startswith("fill cid=cid-1")] + assert len(fill_lines) == 1 + assert "unit=" not in fill_lines[0] # no prefix when unit_name is None + assert "@ 30000" in fill_lines[0] + assert fill_lines[0].endswith("USD") # fee_asset None → quote currency + + # --- restart healing: replay ------------------------------------------------ # diff --git a/trading_bot/tests/application/test_portfolio_runner.py b/trading_bot/tests/application/test_portfolio_runner.py index 9135297..a522d5e 100644 --- a/trading_bot/tests/application/test_portfolio_runner.py +++ b/trading_bot/tests/application/test_portfolio_runner.py @@ -27,11 +27,14 @@ from __future__ import annotations import asyncio +import logging +import re import time from collections.abc import Mapping from decimal import Decimal import polars as pl +import pytest from trading_bot.application import ( EventBus, @@ -1264,3 +1267,97 @@ async def test_no_mark_cache_is_a_legacy_noop() -> None: assert result.submitted == 2 assert result.failed == 0 + + +# --- leaf 02: per-unit event-trail logging (module loggers) ----------------- # + +_RUNNER_LOGGER = "trading_bot.application.portfolio_runner" + + +async def test_mixed_rebalance_logs_summary_skip_and_submit_lines( + caplog: pytest.LogCaptureFixture, +) -> None: + """A mixed rebalance logs the summary, the skip's reason, and the submit's cid. + + BTC (weight 1e-5 → 0.00002 BTC ≈ 1 USDT) is skipped below the 5-USDT notional + minimum; ETH (weight 0.5) submits. The pinned INFO lines land through the module + logger: the summary with correct counters (which sum to ``legs``), the skip line + carrying the ``LegDecision.reason``, and the submit line naming the cid. + """ + weights = {BTC: money("0.00001"), ETH: money("0.5")} + router, tracker, bus, _broker = _engine() + resolver = _FakeResolver({BTC: _BTC_SPEC, ETH: _ETH_SPEC}) + runner = PortfolioRunner( + _strategy(_weights_signal(weights)), + _ListFeed([], asof=1_700), + router, + tracker, + event_bus=bus, + spec_resolver=resolver, + exchange="binance", + ) + + with caplog.at_level(logging.INFO, logger=_RUNNER_LOGGER): + result = await runner.rebalance(_frames()) + + assert result.submitted == 1 + msgs = [r.getMessage() for r in caplog.records if r.name == _RUNNER_LOGGER] + + # The one summary line, with the exact pinned counters. + summary = [m for m in msgs if " rebalance asof=" in m] + assert len(summary) == 1 + assert summary[0].startswith("unit=book rebalance asof=") + m = re.search( + r"legs=(\d+) submitted=(\d+) round_up=(\d+) skipped=(\d+) on_target=(\d+)", + summary[0], + ) + assert m is not None + legs, sub, ru, sk, ot = (int(g) for g in m.groups()) + assert (sub, ru, sk, ot) == (1, 0, 1, 0) + assert sub + ru + sk + ot == legs # the four counters partition the universe + + # The skip line carries the LegDecision.reason (Decimals + the binding min). + skips = [m for m in msgs if "leg BTC/USDT skip:" in m] + assert len(skips) == 1 + assert "venue minimum" in skips[0] + + # The submit line names the leg's deterministic cid. + submits = [m for m in msgs if m.startswith("unit=book order submitted")] + assert len(submits) == 1 + assert "cid=book-ETH/USDT-0" in submits[0] + + +async def test_idle_gated_tick_adds_no_unit_info_lines( + caplog: pytest.LogCaptureFixture, +) -> None: + """A freshness-gate-skipped tick adds zero unit INFO lines (leaf 02 anti-spam). + + The first tick evaluates (genesis) and logs its summary; a second tick over an + unchanged store is skipped by the freshness gate before ``rebalance`` runs — so + it must add **no** INFO line from the runner at all. + """ + client = _counting_client([100.0], [50.0]) # a single common day only + feed = _portfolio_feed_over(client) + strat = _strategy(_weights_signal({BTC: money("0.5"), ETH: money("-0.25")})) + router, tracker, bus, _broker = _engine() + runner = PortfolioRunner(strat, feed, router, tracker, event_bus=bus) + + with caplog.at_level(logging.INFO, logger=_RUNNER_LOGGER): + first = await runner.rebalance_latest() + assert first is not None # first-ever tick evaluates + assert any( + " rebalance asof=" in r.getMessage() + for r in caplog.records + if r.name == _RUNNER_LOGGER + ) + + caplog.clear() + second = await runner.rebalance_latest() + assert second is None # the gate skipped this tick + + idle_info = [ + r + for r in caplog.records + if r.name == _RUNNER_LOGGER and r.levelno >= logging.INFO + ] + assert idle_info == [] diff --git a/trading_bot/tests/application/test_supervisor.py b/trading_bot/tests/application/test_supervisor.py index d1f9dc3..8938ee8 100644 --- a/trading_bot/tests/application/test_supervisor.py +++ b/trading_bot/tests/application/test_supervisor.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio +import logging import polars as pl import pytest @@ -132,6 +133,30 @@ async def test_start_step_stop_lifecycle() -> None: assert await sup.step("btc-ma") is None +async def test_step_records_last_step_duration_ms() -> None: + """`step` times the unit's evaluation onto its status (`None` before any step). + + Leaf 03 tick-timing: the per-unit step duration is `None` on a fresh / + never-stepped unit, and after one `step` the status carries a non-negative + int (milliseconds). It survives `stop` (like the cached accounting report), + so a stopped unit still reports the last duration it measured. + """ + pytest.importorskip("fynance") # ma_crossover evaluates fynance.sma + sup = _supervisor() + + # Never stepped -> None. + assert sup.status("btc-ma")[0].last_step_duration_ms is None + + await sup.start("btc-ma") + await sup.step("btc-ma") + duration = sup.status("btc-ma")[0].last_step_duration_ms + assert isinstance(duration, int) and duration >= 0 + + # Survives stop (the measured duration is retained, like `accounting`). + await sup.stop("btc-ma") + assert sup.status("btc-ma")[0].last_step_duration_ms == duration + + async def test_set_mode_paper_testnet_roundtrip() -> None: """paper ↔ testnet switch needs no confirmation and updates the unit's mode.""" sup = _supervisor() @@ -2282,3 +2307,63 @@ async def test_kill_switch_folds_to_error(tmp_path) -> None: # noqa: ANN001 status = sup.status("btc-ma")[0] assert status.health == "error" assert status.health_detail == ("daily loss limit breached",) + + +# --- leaf 02: unit state-transition + step-error logging (module logger) ---- # + +_SUP_LOGGER = "trading_bot.application.supervisor" + + +class _RaisingRunner(StrategyRunner): + """A ``StrategyRunner`` whose ``step_latest`` always raises (installed directly). + + Bypasses the real ``__init__`` (no engine wiring) like ``_BarrierRunner``, so a + test can drop it onto a unit's ``runner`` and drive ``step`` into its error path + without a real engine / fynance signal. + """ + + def __init__(self) -> None: + # Deliberately do NOT call super().__init__ — no engine to wire here. + pass + + async def step_latest(self): # type: ignore[override] # noqa: ANN201 + raise RuntimeError("boom") + + +async def test_step_error_logs_error_with_traceback( + caplog: pytest.LogCaptureFixture, +) -> None: + """A unit whose ``step`` raises logs one ERROR (with traceback) and re-raises.""" + sup = _supervisor() + unit = sup._units["btc-ma"] # noqa: SLF001 + unit.running = True + unit.runner = _RaisingRunner() + + with caplog.at_level(logging.ERROR, logger=_SUP_LOGGER): + with pytest.raises(RuntimeError, match="boom"): + await sup.step("btc-ma") + + errors = [ + r + for r in caplog.records + if r.name == _SUP_LOGGER and r.levelno >= logging.ERROR + ] + assert len(errors) == 1 + assert "unit=btc-ma step error" in errors[0].getMessage() + assert errors[0].exc_info is not None # logger.exception captured the traceback + + +async def test_start_and_stop_log_unit_transitions( + caplog: pytest.LogCaptureFixture, +) -> None: + """Starting and stopping a unit each log one INFO transition line.""" + pytest.importorskip("fynance") # start builds a real engine (ma_crossover) + sup = _supervisor() + + with caplog.at_level(logging.INFO, logger=_SUP_LOGGER): + await sup.start("btc-ma") + await sup.stop("btc-ma") + + msgs = [r.getMessage() for r in caplog.records if r.name == _SUP_LOGGER] + assert any(m.startswith("unit=btc-ma started mode=paper") for m in msgs) + assert "unit=btc-ma stopped" in msgs diff --git a/trading_bot/tests/interfaces/test_cli_commands.py b/trading_bot/tests/interfaces/test_cli_commands.py index 6d3b7d9..46481fb 100644 --- a/trading_bot/tests/interfaces/test_cli_commands.py +++ b/trading_bot/tests/interfaces/test_cli_commands.py @@ -842,3 +842,228 @@ async def _step_all_signalling(self: StrategySupervisor) -> int: task.cancel() with contextlib.suppress(asyncio.CancelledError): await task + + +# --- tick-timing metrics (leaf 03) ----------------------------------------- # + + +def test_tick_metrics_record_counts_durations_and_overruns() -> None: + """`_TickMetrics.record` folds durations in, counts ticks, flags interval overruns. + + All fields start `None`/`0`; each `record` stamps the last duration/wall-clock + and bumps `ticks_total`; a duration over the interval flags an overrun and + bumps `ticks_overrun`, returning `True` (a fast tick returns `False`). + """ + from trading_bot.interfaces.cli.main import _TickMetrics + + m = _TickMetrics() + # None / 0 before any tick. + assert m.last_tick_duration_ms is None + assert m.last_tick_ts is None + assert m.ticks_total == 0 + assert m.ticks_overrun == 0 + + # A fast tick under the 60 s interval: recorded, not an overrun. + assert m.record(duration_ms=2_000, now_ms=1_000, interval_seconds=60.0) is False + assert m.last_tick_duration_ms == 2_000 + assert m.last_tick_ts == 1_000 + assert m.ticks_total == 1 + assert m.ticks_overrun == 0 + + # A second, slow tick over the interval: recorded, flagged, counted. + assert m.record(duration_ms=61_000, now_ms=2_000, interval_seconds=60.0) is True + assert m.last_tick_duration_ms == 61_000 + assert m.last_tick_ts == 2_000 + assert m.ticks_total == 2 # after two ticks + assert m.ticks_overrun == 1 + + +def test_tick_metrics_record_cron_trigger_never_overruns() -> None: + """A cron trigger (``interval_seconds=None``) records but never flags an overrun. + + An "overrun" against a fixed interval is undefined for a cron schedule, so + the duration + counters still advance while `ticks_overrun` stays `0`. + """ + from trading_bot.interfaces.cli.main import _TickMetrics + + m = _TickMetrics() + assert m.record(duration_ms=10_000_000, now_ms=5, interval_seconds=None) is False + assert m.last_tick_duration_ms == 10_000_000 + assert m.ticks_total == 1 + assert m.ticks_overrun == 0 + + +async def test_daemon_add_job_pins_scheduler_semantics( + monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path +) -> None: + """`_run_daemon` schedules the tick with explicit coalesce / max_instances / grace. + + Spies on `AsyncIOScheduler.add_job` to prove the daemon pins the semantics the + leaf requires: `coalesce=True`, `max_instances=1`, and a `misfire_grace_time` + equal to the configured interval (so a late tick still runs instead of being + silently skipped by APScheduler's 1 s default grace). + """ + import asyncio + import contextlib + + import apscheduler.schedulers.asyncio as aio_sched + + from trading_bot.application.config import AppConfig + from trading_bot.interfaces.cli.main import _run_daemon + + captured: dict[str, object] = {} + called = asyncio.Event() + original_add_job = aio_sched.AsyncIOScheduler.add_job + + def _spy_add_job(self, func, trigger=None, *args, **kwargs): # type: ignore[no-untyped-def] + captured["kwargs"] = dict(kwargs) + job = original_add_job(self, func, trigger, *args, **kwargs) + called.set() + return job + + monkeypatch.setattr(aio_sched.AsyncIOScheduler, "add_job", _spy_add_job) + + cfg = AppConfig.model_validate({"logging": {"dir": str(tmp_path / "logs")}}) + task = asyncio.create_task(_run_daemon(cfg, interval=60.0, cron=None)) + try: + await asyncio.wait_for(called.wait(), timeout=5.0) + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + kwargs = captured["kwargs"] + assert kwargs["coalesce"] is True + assert kwargs["max_instances"] == 1 + # Derived from the interval (a positive int), so `== 60` and `== the interval`. + assert kwargs["misfire_grace_time"] == 60 + assert kwargs["misfire_grace_time"] == int(60.0) + + +async def test_daemon_tick_heartbeat_carries_duration_and_counts( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + tmp_path: pathlib.Path, +) -> None: + """Each daemon tick logs its duration in the heartbeat and advances the counters. + + Drives the real daemon over a fake `step_all` (reports two strategies stepped + so the heartbeat fires); captures the daemon's own `_TickMetrics` instance to + prove its rolling counters advance with the ticks, and reads the heartbeat log + line to prove it now carries `in s` (the leaf-01 line + leaf-03 + duration). + """ + import asyncio + import contextlib + import logging + import re + + from trading_bot.application.config import AppConfig + from trading_bot.application.supervisor import StrategySupervisor + from trading_bot.interfaces.cli.main import _run_daemon + + created: list[cli_main._TickMetrics] = [] + real_cls = cli_main._TickMetrics + + def _capturing_factory() -> cli_main._TickMetrics: + instance = real_cls() + created.append(instance) + return instance + + monkeypatch.setattr(cli_main, "_TickMetrics", _capturing_factory) + + async def _fake_step_all(self: StrategySupervisor) -> int: + return 2 # pretend two strategies stepped -> the heartbeat logs + + monkeypatch.setattr(StrategySupervisor, "step_all", _fake_step_all) + + def _heartbeats() -> list[str]: + return [ + r.getMessage() + for r in caplog.records + if "daemon tick: stepped" in r.getMessage() + ] + + caplog.set_level(logging.INFO, logger="trading_bot.daemon") + cfg = AppConfig.model_validate({"logging": {"dir": str(tmp_path / "logs")}}) + task = asyncio.create_task(_run_daemon(cfg, interval=0.01, cron=None)) + try: + # Bounded poll for two ticks' worth of heartbeats (no fixed sleep race). + for _ in range(500): + if len(_heartbeats()) >= 2: + break + await asyncio.sleep(0.01) + heartbeats = _heartbeats() + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + assert len(heartbeats) >= 2 + # The heartbeat carries the tick's measured duration: `in s`. + assert re.search( + r"daemon tick: stepped 2 strategy\(ies\) in \d+\.\d{3}s", heartbeats[0] + ) + # The daemon's own rolling counters advanced with the ticks. + assert created[0].ticks_total >= 2 + assert isinstance(created[0].last_tick_duration_ms, int) + assert created[0].last_tick_duration_ms >= 0 + + +async def test_daemon_tick_overrun_warns_and_counts( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + tmp_path: pathlib.Path, +) -> None: + """A tick slower than the interval increments `ticks_overrun` and logs a WARN. + + A stubbed `step_all` that sleeps longer than the (tiny) interval forces an + overrun: the daemon's `ticks_overrun` counter advances and it emits the + `daemon tick overrun: s > s` WARNING. + """ + import asyncio + import contextlib + import logging + import re + + from trading_bot.application.config import AppConfig + from trading_bot.application.supervisor import StrategySupervisor + from trading_bot.interfaces.cli.main import _run_daemon + + created: list[cli_main._TickMetrics] = [] + real_cls = cli_main._TickMetrics + + def _capturing_factory() -> cli_main._TickMetrics: + instance = real_cls() + created.append(instance) + return instance + + monkeypatch.setattr(cli_main, "_TickMetrics", _capturing_factory) + + async def _slow_step_all(self: StrategySupervisor) -> int: + await asyncio.sleep(0.03) # > the 0.01 s interval -> overrun + return 1 + + monkeypatch.setattr(StrategySupervisor, "step_all", _slow_step_all) + + caplog.set_level(logging.INFO, logger="trading_bot.daemon") + cfg = AppConfig.model_validate({"logging": {"dir": str(tmp_path / "logs")}}) + task = asyncio.create_task(_run_daemon(cfg, interval=0.01, cron=None)) + try: + for _ in range(500): + if created and created[0].ticks_overrun >= 1: + break + await asyncio.sleep(0.01) + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + assert created[0].ticks_overrun >= 1 + warns = [ + r.getMessage() + for r in caplog.records + if "daemon tick overrun" in r.getMessage() + ] + assert warns + assert re.search(r"daemon tick overrun: \d+\.\d{3}s > 0\.01s", warns[0]) diff --git a/trading_bot/tests/interfaces/test_control_api.py b/trading_bot/tests/interfaces/test_control_api.py index c22b3b9..7fc98ed 100644 --- a/trading_bot/tests/interfaces/test_control_api.py +++ b/trading_bot/tests/interfaces/test_control_api.py @@ -162,8 +162,9 @@ def test_control_wrapper_detail_page_has_the_control_surface() -> None: def test_control_wrapper_health_is_the_dashboard_shape() -> None: """The wrapper's `/api/health` is the unified dashboard's shape (mode + read_only). - `create_control_app` wires no `schedule_info` hook, so the cadence fields stay - `null` — same scheduler-agnostic default as the plain `dashboard` command. + `create_control_app` wires no `schedule_info` hook, so the cadence + tick-timing + fields stay `null`/`0` — same scheduler-agnostic default as the plain + `dashboard` command. """ body = _client().get("/api/health").json() assert body == { @@ -173,6 +174,10 @@ def test_control_wrapper_health_is_the_dashboard_shape() -> None: "read_only": False, "next_tick_ts": None, "tick": None, + "last_tick_duration_ms": None, + "last_tick_ts": None, + "ticks_total": 0, + "ticks_overrun": 0, "worst": "ok", "unhealthy": 0, } diff --git a/trading_bot/tests/interfaces/test_dashboard.py b/trading_bot/tests/interfaces/test_dashboard.py index 4bf6c6c..b3df241 100644 --- a/trading_bot/tests/interfaces/test_dashboard.py +++ b/trading_bot/tests/interfaces/test_dashboard.py @@ -238,10 +238,11 @@ def test_active_tab_is_highlighted() -> None: def test_health_shape_and_values() -> None: - """`GET /api/health` returns the health shape; `next_tick_ts`/`tick` null by default. + """`GET /api/health` returns the health shape; cadence fields null/0 by default. With no `schedule_info` hook (the plain `dashboard` command has no scheduler), - the cadence fields stay `null` — a scheduler-agnostic health payload. + the cadence + tick-timing fields stay `null`/`0` — a scheduler-agnostic + health payload. """ resp = _client().get("/api/health") assert resp.status_code == 200 @@ -253,20 +254,35 @@ def test_health_shape_and_values() -> None: "read_only": False, "next_tick_ts": None, "tick": None, + "last_tick_duration_ms": None, + "last_tick_ts": None, + "ticks_total": 0, + "ticks_overrun": 0, "worst": "ok", "unhealthy": 0, } def test_health_schedule_info_hook_surfaces_cadence() -> None: - """A `schedule_info` hook's `next_tick_ts` / `tick` surface on `/api/health`.""" + """A `schedule_info` hook's cadence + tick-timing fields surface on `/api/health`.""" app = create_dashboard_app( _supervisor(), - schedule_info=lambda: {"next_tick_ts": 1_700_000_000_000, "tick": "every 60s"}, + schedule_info=lambda: { + "next_tick_ts": 1_700_000_000_000, + "tick": "every 60s", + "last_tick_duration_ms": 2100, + "last_tick_ts": 1_699_999_940_000, + "ticks_total": 42, + "ticks_overrun": 1, + }, ) body = TestClient(app).get("/api/health").json() assert body["next_tick_ts"] == 1_700_000_000_000 assert body["tick"] == "every 60s" + assert body["last_tick_duration_ms"] == 2100 + assert body["last_tick_ts"] == 1_699_999_940_000 + assert body["ticks_total"] == 42 + assert body["ticks_overrun"] == 1 def test_health_schedule_info_hook_that_raises_degrades_to_nulls() -> None: @@ -281,6 +297,11 @@ def _boom() -> dict[str, object]: body = resp.json() assert body["next_tick_ts"] is None assert body["tick"] is None + # The tick-timing fields degrade to their no-hook defaults too, never 500. + assert body["last_tick_duration_ms"] is None + assert body["last_tick_ts"] is None + assert body["ticks_total"] == 0 + assert body["ticks_overrun"] == 0 async def test_api_health_worst_and_count() -> None: @@ -336,6 +357,11 @@ async def test_api_health_worst_and_count() -> None: assert body["read_only"] is False assert body["next_tick_ts"] is None assert body["tick"] is None + # Tick-timing fields: no scheduler hook here → null/0 defaults, unchanged. + assert body["last_tick_duration_ms"] is None + assert body["last_tick_ts"] is None + assert body["ticks_total"] == 0 + assert body["ticks_overrun"] == 0 # Both units running: worst is the warn one; one unhealthy unit. assert body["worst"] == "warn" assert body["unhealthy"] == 1 @@ -1311,6 +1337,7 @@ async def test_api_completeness_contract_sweep(tmp_path) -> None: # noqa: ANN00 "open_orders", "last_eval_ts", "last_asof_ts", + "last_step_duration_ms", "allocation", "contributed", "unrealised", @@ -1405,6 +1432,10 @@ async def test_api_completeness_contract_sweep(tmp_path) -> None: # noqa: ANN00 "read_only", "next_tick_ts", "tick", + "last_tick_duration_ms", + "last_tick_ts", + "ticks_total", + "ticks_overrun", "worst", "unhealthy", } @@ -2048,22 +2079,25 @@ def test_api_strategies_carries_health() -> None: async def test_strategies_endpoint_last_eval_and_asof_ts() -> None: - """`GET /api/strategies` carries `last_eval_ts`/`last_asof_ts`, set after a tick. + """`GET /api/strategies` carries eval/asof/step-duration fields, set after a tick. - Both are `None` before the unit has ever ticked via its `*_latest` path - (incl. a never-started unit — PR #168's diagnostic fields). Driving one - tick through the supervisor's `step_all` (the daemon's own path) stamps - both: `last_eval_ts` is the wall-clock of the attempt, `last_asof_ts` the - as-of of the data actually evaluated. + `last_eval_ts` / `last_asof_ts` / `last_step_duration_ms` are all `None` + before the unit has ever ticked via its `*_latest` path (incl. a + never-started unit — PR #168's diagnostic fields + leaf 03's step duration). + Driving one tick through the supervisor's `step_all` (the daemon's own path) + stamps them: `last_eval_ts` is the wall-clock of the attempt, `last_asof_ts` + the as-of of the data actually evaluated, `last_step_duration_ms` how long + the evaluation took (a non-negative int). """ pytest.importorskip("fynance") # ma_crossover evaluates fynance.sma sup = StrategySupervisor(_config(), dccd_client=_FakeStartClient()) client = TestClient(create_dashboard_app(sup)) - # Never started/ticked -> both null. + # Never started/ticked -> all null. [before] = client.get("/api/strategies").json() assert before["last_eval_ts"] is None assert before["last_asof_ts"] is None + assert before["last_step_duration_ms"] is None await sup.start("btc-ma") assert await sup.step_all() == 1 # the one running unit stepped once @@ -2071,6 +2105,10 @@ async def test_strategies_endpoint_last_eval_and_asof_ts() -> None: [after] = client.get("/api/strategies").json() assert isinstance(after["last_eval_ts"], int) and after["last_eval_ts"] > 0 assert isinstance(after["last_asof_ts"], int) and after["last_asof_ts"] > 0 + assert ( + isinstance(after["last_step_duration_ms"], int) + and after["last_step_duration_ms"] >= 0 + ) def test_set_mode_testnet_then_paper() -> None: