From bf97ea9498e08c621c815e81d78ad65c7d5a5103 Mon Sep 17 00:00:00 2001 From: ArthurBernard Date: Tue, 14 Jul 2026 15:02:14 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20per-unit=20event=20trail=20?= =?UTF-8?q?=E2=80=94=20rebalance=20summaries,=20leg=20decisions,=20orders,?= =?UTF-8?q?=20fills?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the daemon log tell each unit's story so "why hasn't alloc1-kraken traded since genesis?" is answerable by grepping logs/daemon.log, while an idle steady-state tick still adds zero per-unit INFO lines. - PortfolioRunner: one INFO rebalance-summary per evaluated tick (unit / asof / legs / submitted / round_up / skipped / on_target — the four counters partition the universe); elevate round_up/skip leg decisions to pinned INFO lines carrying the LegDecision.reason; INFO on each accepted order (cid/side/qty/symbol/px/venue_id) and WARN on a refused leg. Plain submits stay summary-only (no double INFO). Idle ticks are gated before rebalance() runs, so they emit nothing. - OrderFillSync: optional unit_name; INFO per applied fill (cid / signed qty / price / fee / fee_asset-or-quote), prefixed unit= when known, threaded from service_factory for a single-unit slice. - StrategySupervisor: INFO on unit start/stop, ERROR (logger.exception) when a unit's step raises, DEBUG on a no-new-bar tick. Verified on today's real dccd bars via a scratch daemon (fresh empty books, port 8099): both units' genesis rebalance summaries, skip lines with full reasons, submit lines and their matching fills — one fill cross-checked to the store (cid/qty/price exact) — and an idle tick adding only the heartbeat. --- trading_bot/application/order_fill_sync.py | 29 +++- trading_bot/application/portfolio_runner.py | 146 ++++++++++++++---- trading_bot/application/service_factory.py | 28 +++- trading_bot/application/supervisor.py | 40 ++++- .../tests/application/test_order_fill_sync.py | 77 +++++++++ .../application/test_portfolio_runner.py | 97 ++++++++++++ .../tests/application/test_supervisor.py | 61 ++++++++ 7 files changed, 441 insertions(+), 37 deletions(-) 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..bc59a59 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 @@ -850,6 +853,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 +1035,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 +1525,25 @@ 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() + 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 + 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).""" 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..9e5420b 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 @@ -2282,3 +2283,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 From f2b26b0c02f1399d78e2ef4383b3aae92ea1b88d Mon Sep 17 00:00:00 2001 From: ArthurBernard Date: Tue, 14 Jul 2026 15:04:56 +0200 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20unit=20event=20trail=20closeout=20?= =?UTF-8?q?=E2=80=94=20changelog,=20ADR,=20status,=20plan=20tick?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 10 ++ doc/dev/03-decisions.md | 18 +++ doc/dev/06-status.md | 8 +- doc/dev/plans/daemon-logging/00-plan.md | 2 +- .../daemon-logging/02-unit-event-trail.md | 113 ------------------ 5 files changed, 34 insertions(+), 117 deletions(-) delete mode 100644 doc/dev/plans/daemon-logging/02-unit-event-trail.md diff --git a/CHANGELOG.md b/CHANGELOG.md index ad9cb0d..389fd40 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (`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) ### Changed diff --git a/doc/dev/03-decisions.md b/doc/dev/03-decisions.md index 791085e..13c78ac 100644 --- a/doc/dev/03-decisions.md +++ b/doc/dev/03-decisions.md @@ -6,6 +6,24 @@ rejected approaches as tombstones. --- +### 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 diff --git a/doc/dev/06-status.md b/doc/dev/06-status.md index b34334c..813698a 100644 --- a/doc/dev/06-status.md +++ b/doc/dev/06-status.md @@ -183,10 +183,12 @@ 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). -- **`daemon-logging` in flight (1/3)**: the logging spine landed (rotated, +- **`daemon-logging` in flight (2/3)**: the logging spine (rotated, tz-offset-timestamped `logs/daemon.log`, manifest `logging:` section, tick - tracebacks). Next: the per-unit event trail (leaf 02), tick-timing metrics - (leaf 03). Plan: `plans/daemon-logging/`. + tracebacks) and the per-unit event trail (rebalance summaries, skip + reasons, order/fill lines — a unit's inactivity is now explained by the + log) have landed. Next: tick-timing metrics (leaf 03). Plan: + `plans/daemon-logging/`. ## Known gaps / deferred diff --git a/doc/dev/plans/daemon-logging/00-plan.md b/doc/dev/plans/daemon-logging/00-plan.md index e80b436..3c66bc9 100644 --- a/doc/dev/plans/daemon-logging/00-plan.md +++ b/doc/dev/plans/daemon-logging/00-plan.md @@ -54,7 +54,7 @@ What the audit measured (the evidence this epic answers): ## Leaf checklist - [x] 01 log-setup — `feat/daemon-log-setup` — medium -- [ ] 02 unit-event-trail — `feat/daemon-unit-event-logs` — medium +- [x] 02 unit-event-trail — `feat/daemon-unit-event-logs` — medium - [ ] 03 tick-timing-metrics — `feat/tick-timing-metrics` — medium ## Dependencies diff --git a/doc/dev/plans/daemon-logging/02-unit-event-trail.md b/doc/dev/plans/daemon-logging/02-unit-event-trail.md deleted file mode 100644 index f8e35f8..0000000 --- a/doc/dev/plans/daemon-logging/02-unit-event-trail.md +++ /dev/null @@ -1,113 +0,0 @@ ---- -plan: daemon-logging/02-unit-event-trail -kind: leaf -status: planned -complexity: medium -depends: [01] -parallel: false -branch: feat/daemon-unit-event-logs -pr: "" ---- - -# 02 — Per-unit event trail: evaluations, orders, skips, fills - -## Goal - -The log tells each unit's story. After this leaf, "why hasn't alloc1-kraken -traded since genesis?" is answered by grepping `logs/daemon.log`: every -completed rebalance logs a summary, every round-up/skip logs its -`LegDecision.reason` (the exact Decimals and the binding minimum), every -submit and fill logs identifiers and amounts. Pinned anti-spam rule: a -steady-state tick where nothing happened adds **zero** INFO lines. - -## Files to change - -- `trading_bot/application/portfolio_runner.py` — summary line + elevate the - leg-decision/submit DEBUGs (`:704`, `:715`) to the pinned INFO formats -- `trading_bot/application/order_fill_sync.py` — INFO per fill applied -- `trading_bot/application/supervisor.py` — unit state-transition lines - (started / stopped / step error) -- `trading_bot/application/service_factory.py` — thread the unit name where a - component lacks it (see step 4) -- `trading_bot/tests/application/` — extend the runner / fill-sync / - supervisor tests with `caplog` assertions - -## Steps - -Pinned line formats (key=value prefix, grep-stable; all through the module -loggers `logging.getLogger(__name__)` that already exist): - -1. **Rebalance summary** — in `PortfolioRunner`, one INFO at the end of each - *evaluated* rebalance (a new bar was consumed, whether or not orders - resulted): - `unit= rebalance asof= legs= submitted= round_up= skipped= on_target=` - The runner already counts these outcomes to return the submitted count — - accumulate the four counters alongside. -2. **Leg decisions** — in `_prepare_delta` (`portfolio_runner.py:704/715` - today): `round_up` and `skip` decisions log **INFO** - `unit= leg : `; - plain `submit` passthroughs stay DEBUG (the router/summary already tell - that story — no double INFO per order). -3. **Order lifecycle** — the runner is per-unit and awaits the router, so it - logs (no router signature change): - - accepted: `unit= order submitted cid= @ venue_id=` - (INFO — replaces/absorbs the `:715` DEBUG); - - refused/rejected: WARN with the same prefix + `reason=<…>`. -4. **Fills** — `OrderFillSync` applies every fill; give it an optional - `unit_name: str | None = None` constructor param (additive, default keeps - current signature working), threaded from `service_factory` where each - unit's sync is built. On apply: INFO - `unit= fill cid= @ fee= `. -5. **Supervisor transitions** — INFO on unit start/stop, ERROR (with - `logger.exception`) when a unit's `step` raises. A tick that evaluates - nothing (no new bar) logs at DEBUG only. -6. **Anti-spam check** — with INFO level and no new daily bar, one hour of - ticks must add only the ~60 heartbeat lines from leaf 01 — nothing per - unit. (Two units × 1440 min/day of per-unit INFO would bury the signal.) -7. Secrets: none of these lines may include tokens/keys — amounts, ids, - symbols, reasons only (`LegDecision.reason` carries only Decimals and - minimum names by construction). - -## Tests - -- Runner rebalance with a mixed outcome (≥1 submit, ≥1 skip via a venue-min - spec): `caplog` shows the summary line with correct counters, the skip's - INFO line containing the `LegDecision.reason` text, the submit's INFO line - with cid; counters sum to `legs`. -- No-new-bar tick: `caplog` at INFO gains **zero** unit lines. -- `OrderFillSync` with `unit_name="u1"`: fill application logs the pinned - format incl. `fee_asset`; with `unit_name=None`: line still valid (prefix - `unit=?` or omitted — pick one and test it). -- Supervisor: step exception → ERROR with traceback in `caplog`. -- Full suite + `ruff check trading_bot/` green. - -## Verification on real data - -Scratch instance (same recipe as leaf 01 — port 8099, scratch book copies, -read-only real dccd store, **never** port 8000 / `var/` originals): - -1. Point the scratch manifest at **fresh empty books** (no copies) so genesis - funding + the first rebalance fire immediately on start — the same real - signal path as 2026-07-12, on today's real bars. -2. Run ≥ 3 minutes; SIGINT. -3. Report verbatim from the scratch log: both units' rebalance summary lines, - at least one real `skip` line with its full `LegDecision.reason` (expected - in numbers: at capital 100, most kraken legs sit below the venue minimum — - this excerpt IS the documented answer to the kraken-silence question), at - least one submitted-order line and its fill line with matching cid. -4. Cross-check one logged fill against the scratch book - (`sqlite3 .sqlite 'select …'`): cid, qty, price match — the log - never claims a fill the store does not have. -5. Confirm the live daemon untouched (port 8000 responds, pid unchanged). - -## Closeout - -- CHANGELOG `### Added`: "per-unit daemon log trail: rebalance summaries, - round-up/skip reasons (venue minimums), order submits and fills with ids — - a unit's inactivity is now explained by the log (#XX)". -- ADR if judged non-trivial: lifecycle logging lives in the runner (per-unit, - sees router outcomes) instead of the router (shared, unit-blind) — rejected: - threading a LoggerAdapter through the router. -- `06-status.md`: soak observability note — kraken dust-skip visibility. -- Tick leaf 02 in `00-plan.md`; archive this leaf file. Roadmap line stays - (leaf 03 open).