feat: the live per-cycle stop-management step, off by default (#502 stage 2)

keel/agent.py

@@ -61,13 +61,19 @@
 import logging
 import time
 from collections.abc import Callable
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from decimal import Decimal
 from typing import Any, Literal
 
 from keel_broker_api.results import MarketSchedule, SessionState
 from keel_core.products import quote_currency_of
-from keel_core.telemetry import bind_cycle, log_event, new_cycle_id, unbind_cycle
+from keel_core.telemetry import (
+    bind_cycle,
+    log_event,
+    log_exception,
+    new_cycle_id,
+    unbind_cycle,
+)
 
 from keel.config import Config
 from keel.data import freshness, market_feed
@@ -77,6 +83,7 @@
 from keel.execution.executor import ExecutionResult, _fetch_available_quote
 from keel.execution.guards import FEED_STALENESS_CYCLES
 from keel.strategy import engine
+from keel.strategy.exit_policy import EXIT_POLICY_OFF, next_stop, policy_for, trailing_atr
 from keel.strategy.paper import PaperTrader
 from keel.strategy.rules.base import Action, Rule, Setup, Signal
 from keel.strategy.rules.dca import Dca
@@ -819,6 +826,213 @@ def _close_tranches(
         repo.close_position(position["id"], closed_at=now_ts)
 
 
+#: Products whose young-table warmup notice has already been logged THIS PROCESS (#502). The
+#: notice is per product, not per cycle -- a daily deployment a few bars short of the
+#: `4 x atr_period + 1` threshold would otherwise repeat the same line every cycle until
+#: warmed. Process scope is the deliberate bound: an `agent_state` key would outlive the
+#: condition it describes and suppress the notice forever after a restart.
+_WARMUP_LOGGED_PRODUCTS: set[str] = set()
+
+
+def _manage_stops(
+    broker: Any,
+    repo: Repository,
+    config: Config,
+    rules: list[Rule],
+    candles_by_tf_by_product: dict[str, dict[Granularity, list[Any]]],
+    now_ts: int,
+) -> None:
+    """The per-cycle live stop-management step (#502 stage 2): ratchet each held tranche's
+    resting bracket according to its OWNING rule's exit policy -- the rule row that owns the
+    tranche's PRODUCT, never merely a row sharing its family name (the rules table holds one
+    row per (kind, product), so a name-only key would govern a multi-product family's every
+    tranche under whichever same-family row loaded last: an opted-in BTC row managing a
+    knob-less ETH tranche's bracket, or the reverse; `_handle_exits` scopes ownership by
+    product for the same reason).
+
+    DEFAULT OFF, exactly like the sim wiring. The knobs are per-rule-family params
+    (`trail_atr_mult` / `be_roll_rr` on `pullback_continuation` and `rsi_meanrev`), a rule
+    whose params carry neither gets `EXIT_POLICY_OFF` and its position is not touched -- so
+    every rule row that has not opted in trades exactly as before this step existed. The
+    #442 experiment (docs/experiments/2026-08-22-trailing-vs-static-exits.md) measured
+    trailing WORSE and the break-even roll no better than the static exit at the 120 bp fee,
+    so the capability ships dark and the operator opts in per rule; `turtle_breakout` offers
+    the knobs nowhere at all (its exit is the Donchian channel by choice).
+
+    The decision is DELEGATED, not re-derived: `policy_for` + `next_stop` are the same
+    functions the sim/backtest engines apply, so the live level and the simulated level come
+    from one place. The bar is the latest COMPLETED one on the rule's trading timeframe
+    (`market_feed.poll_once` stores closed candles only), so the no-lookahead contract holds:
+    during the bar the venue's bracket rested at the old level, and the new level binds from
+    the next bar. A punctual cycle sees exactly one new bar per run; a MISSED cycle reads only
+    the newest bar, which is conservative in the one arm where it differs -- a break-even
+    trigger on an older bar's high goes unobserved and leaves the stop lower, never looser.
+
+    The ROLL itself is `executor.roll_stop_to`: ONE cancel-and-replace per tranche per cycle
+    (#519's cancel-before-place protocol, the crash ledger, the ratchet and at/above-target
+    refusals -- all in the executor, none re-invented here). Paper cycles never call this:
+    paper entries place no exchange-side brackets, so there is nothing to roll.
+
+    A failed roll is LOUD and ISOLATED, never a dead cycle. The roll's cancel half is a
+    live-money action with ordinary failure modes (`CancelPending` / `CancelUnavailable` are
+    what Coinbase's batch-cancel answers when a fill lands during the roll), so each
+    tranche's roll is wrapped: a raise is logged at CRITICAL -- a possibly-half-completed
+    action on live money must be loud -- and the step CONTINUES to the next tranche. Letting
+    it propagate would abort `run_once` AFTER entries were placed: no `LoopResult`, no
+    post-cycle notify, a nonzero CLI exit, and the live-run wrapper declining to stamp the
+    UTC day -- so the next trigger re-runs the whole cycle into the duplicate-entry window.
+
+    The kill-switch window inside a roll (see `executor._roll_stop`): a cancel is not
+    rail-gated -- it REMOVES risk, so there is nothing for a guard to veto -- and a switch
+    that engages mid-roll, after the cancel, fails only the REPLACEMENT closed (rail 12
+    fails every order), leaving the position unbracketed until the switch clears. That is
+    the correct failure direction, and it is not silent: the CRITICAL is loud, and
+    `reconcile` heals from the crash ledger when cycles resume.
+    """
+    rules_by_owner = {(getattr(rule, "product_id", None), rule.name): rule for rule in rules}
+    for tranche in repo.get_open_positions():
+        rule = rules_by_owner.get((tranche["product_id"], tranche["rule_name"]))
+        if rule is None:
+            continue  # the owning row is not on this cycle's set (demoted/retired): no policy
+        policy = policy_for(rule)
+        if policy is EXIT_POLICY_OFF:
+            continue  # the default: the rule never asked for stop management
+
+        bracket_id = tranche["bracket_order_id"]
+        if bracket_id is None:
+            continue  # no bracket recorded: the unbracketed sweep owns this position (#195)
+        old_order = repo.get_order(bracket_id)
+        if old_order is None or old_order["status"] not in executor.RESTING_STATUSES:
+            continue  # filled or dead: reconciliation owns that bracket, not this step
+        # Per-PRODUCT ratchet state read inside the per-TRANCHE loop: this assumes the
+        # ledger's one-tranche-per-asset shape (the same assumption that lets the defers
+        # treat a product's slots as exclusive). Keying the ratchet per TRANCHE is a
+        # prerequisite for pyramiding, and is deliberately not invented here.
+        current_stop = repo.get_state(f"open_stop:{tranche['product_id']}")
+        if current_stop is None:
+            continue  # no recorded level to ratchet from -- nothing this step may act on
+
+        # `initial_stop` is the tranche's ORIGINAL setup stop (#520) -- the denominator of the
+        # break-even threshold. NULL means "nobody recorded it" (pre-ledger tranche, DCA), and
+        # the ledger's own contract is that the BE arm switches OFF rather than substitutes the
+        # current stop, which is a different (and on a ratcheted position, stale) policy.
+        initial_stop = tranche["initial_stop"]
+        if initial_stop is None:
+            policy = replace(policy, be_roll_rr=None)
+
+        candles_by_tf = candles_by_tf_by_product.get(tranche["product_id"])
+        if not candles_by_tf:
+            # Stale product this cycle: the freshness pre-pass skipped it, so THIS cycle
+            # fetched no bars for it -- which is not "the table holds nothing" (it usually
+            # holds older closed bars; staleness is a claim about the newest one). The
+            # tranche waits a cycle, and says so at the same INFO volume as the
+            # empty-timeframe skip below rather than skipping silently.
+            log_event(
+                logger,
+                logging.INFO,
+                "agent.stop_management_skipped",
+                product=tranche["product_id"],
+                rule=rule.name,
+                reason="product skipped this cycle (stale feed) -- no bars fetched",
+            )
+            continue
+        series = candles_by_tf.get(_management_timeframe(rule, candles_by_tf)) or []
+        if not series:
+            log_event(
+                logger,
+                logging.INFO,
+                "agent.stop_management_skipped",
+                product=tranche["product_id"],
+                rule=rule.name,
+                reason="no candles on the rule's trading timeframe",
+            )
+            continue
+
+        # Young-table warmup (#442 fidelity). `trailing_atr` prices the trail off a
+        # Wilder average that only matches the #442-measured sim behavior once the table
+        # holds `4 x atr_period + 1` closed bars, and the live table does not start there:
+        # `market_feed.poll_once` cold-starts an EMPTY table with one bar, and backfill has
+        # no production caller. Short of the threshold the trail arm is OFF for the cycle.
+        # The BE arm is NOT equally affected -- it reads only the latest bar's HIGH against
+        # thresholds fixed at entry time, which is exactly what the sim does from bar one --
+        # so it stays live. The notice fires once per product, not per cycle.
+        warmup_bars = 4 * policy.atr_period + 1
+        if policy.trail_atr_mult is not None and len(series) < warmup_bars:
+            policy = replace(policy, trail_atr_mult=None)
+            if tranche["product_id"] not in _WARMUP_LOGGED_PRODUCTS:
+                _WARMUP_LOGGED_PRODUCTS.add(tranche["product_id"])
+                log_event(
+                    logger,
+                    logging.INFO,
+                    "agent.stop_management_waiting_for_warmup",
+                    product=tranche["product_id"],
+                    rule=rule.name,
+                    bars=len(series),
+                    needed=warmup_bars,
+                )
+
+        atr = trailing_atr(series, policy.atr_period)
+        level = next_stop(
+            policy,
+            tranche["entry_fill"],
+            # `initial_stop` is only read by the BE arm, which is OFF above when it is NULL;
+            # the entry is the neutral stand-in for the unreachable denominator.
+            initial_stop if initial_stop is not None else tranche["entry_fill"],
+            current_stop,
+            series[-1],
+            atr,
+        )
+        if level <= current_stop:
+            continue  # the ratchet: nothing proposed toward profit this cycle
+
+        try:
+            executor.roll_stop_to(
+                broker,
+                repo,
+                config,
+                product_id=tranche["product_id"],
+                old_stop_order_id=bracket_id,
+                new_stop=level,
+                qty=tranche["qty"],
+                rule_name=tranche["rule_name"],
+                now_ts=now_ts,
+            )
+        except Exception:
+            # Per-tranche isolation, deliberately BROAD: `CancelPending`/`CancelUnavailable`
+            # from the fill-during-roll race is the expected raise, but a possibly-
+            # half-completed action on live money must not be graded by exception type
+            # here. The position may be unprotected RIGHT NOW (the cancel may have landed
+            # before the raise), so this is CRITICAL with the traceback attached -- and the
+            # cycle SURVIVES, per the docstring: a dead `run_once` after entries were
+            # placed re-runs the cycle into the duplicate-entry window.
+            log_exception(
+                logger,
+                "agent.stop_management_roll_failed",
+                level=logging.CRITICAL,
+                product=tranche["product_id"],
+                rule=rule.name,
+                old_stop_order_id=bracket_id,
+                attempted_stop=level,
+            )
+            continue
+
+
+def _management_timeframe(rule: Rule, candles_by_tf: dict[Granularity, list[Any]]) -> Granularity:
+    """The series stop management reads: the rule's own trading timeframe when it declares one
+    (`granularity` on `TurtleBreakout`/`PullbackContinuation`, `timeframe` on `RsiMeanReversion`
+    -- the same attribute order `engine._trading_granularity` and `backtest._rule_trading_tf`
+    use), else the finest series this cycle actually has. One deliberate divergence from the
+    backtester, bounded: `backtest._rule_trading_tf` falls back to a fixed ONE_HOUR, this falls
+    back to whatever the cycle polled finest -- but every knob-carrying family declares its
+    timeframe, so a rule that is ever actually managed (`policy_for` non-OFF) never takes a
+    fallback series at all, and the divergence is a bound, not a path anyone trades on."""
+    for attr in ("granularity", "timeframe"):
+        value = getattr(rule, attr, None)
+        if isinstance(value, Granularity):
+            return value
+    return min(candles_by_tf, key=lambda g: _GRANULARITY_ORDER.get(g, 0))
+
+
 # -- venue session (FR-9: a closed market is not a stale feed) --------------------------------
 
 #: `agent_state` key PREFIXES the cycle writes each run when (and only when) the broker is a
@@ -1665,6 +1879,17 @@ def run_once(
                         initial_stop=signal.setup.stop if signal.setup is not None else None,
                     )
 
+        # == STOP MANAGEMENT: ratchet held positions' brackets per the owning rule's policy ==
+        #
+        # AFTER the trading decisions (a rule exit this cycle already closed its tranches, so
+        # they are no longer `open` and cannot be managed; a fresh entry's bracket rests at its
+        # own setup stop and may trail from this same completed bar, mirroring the sim's
+        # bar-end sequencing) and BEFORE cycle end. LIVE cycles only: paper entries place no
+        # exchange-side brackets, so there is nothing to roll and the paper path must not touch
+        # the real order book. Default-off per rule -- see `_manage_stops`.
+        if paper_trader is None:
+            _manage_stops(broker, repo, config, rules, candles_by_tf_by_product, now_ts)
+
         cycle_result = LoopResult(
             ts=now_ts,
             skipped=False,