feat: Add DYNAMIC_AGE clamp stats and close-reason max_age field

strategies/market_making_strategy.py

@@ -161,6 +161,10 @@ def __init__(self, market_data_manager, order_manager, config: Dict) -> None:
         self._base_max_position_age: float = config.get('max_position_age_seconds', 120.0)
         # coin -> (avg_move_bps, computed_age_seconds) for periodic logging
         self._dynamic_age_recent: Dict[str, Tuple[float, float]] = {}
+        # coin -> {min_clamp, max_clamp, mid, raw_sum, raw_min, raw_max, samples}
+        # Populated by _get_dynamic_position_age and reset on each summary log
+        # so per-coin clamp distribution can be observed over the log interval.
+        self._dynamic_age_clamp_stats: Dict[str, Dict[str, float]] = {}
         self._dynamic_age_log_interval: float = config.get(
             'dynamic_age_log_interval', DYNAMIC_AGE_LOG_INTERVAL
         )
@@ -555,14 +559,42 @@ def _get_dynamic_position_age(self, coin: str) -> Optional[float]:
         # Scale: high vol -> short age, low vol -> long age
         # baseline_vol (bps) = typical move per cycle (calibrated)
         ratio = self._dynamic_age_baseline_vol / max(avg_move_bps, self._dynamic_age_baseline_vol * 0.1)
-        age = self._base_max_position_age * ratio
+        raw_age = self._base_max_position_age * ratio
 
         # Clamp to [min_age, max_age]
-        age = max(self._dynamic_age_min, min(age, self._dynamic_age_max))
+        age = max(self._dynamic_age_min, min(raw_age, self._dynamic_age_max))
 
         # Record latest computation for periodic summary log
         self._dynamic_age_recent[coin] = (avg_move_bps, age)
 
+        # Aggregate clamp distribution per coin so the periodic summary can
+        # show "min_clamp=85% mid=15%" — directly tells operators whether
+        # DYNAMIC_AGE_MIN is biting or the value is moving freely.
+        stats = self._dynamic_age_clamp_stats.setdefault(
+            coin,
+            {
+                "min_clamp": 0,
+                "max_clamp": 0,
+                "mid": 0,
+                "raw_sum": 0.0,
+                "raw_min": float("inf"),
+                "raw_max": 0.0,
+                "samples": 0,
+            },
+        )
+        if raw_age <= self._dynamic_age_min:
+            stats["min_clamp"] += 1
+        elif raw_age >= self._dynamic_age_max:
+            stats["max_clamp"] += 1
+        else:
+            stats["mid"] += 1
+        stats["raw_sum"] += raw_age
+        if raw_age < stats["raw_min"]:
+            stats["raw_min"] = raw_age
+        if raw_age > stats["raw_max"]:
+            stats["raw_max"] = raw_age
+        stats["samples"] += 1
+
         return age
 
     def _place_orders(self, coin: str) -> None:
@@ -770,25 +802,61 @@ def _log_fill_rate(self) -> None:
         self._fills_per_coin.clear()
 
     def _log_dynamic_age(self) -> None:
-        """Log a periodic summary of per-coin dynamic position age."""
+        """Log a periodic summary of per-coin dynamic position age.
+
+        Emits two kinds of lines:
+
+        1. ``[mm] Dynamic age: ...`` — last-seen ``(vol, age)`` per coin,
+           preserved for backward compatibility with existing dashboards.
+        2. ``[mm] dyn-age <coin> samples=N min=X% mid=X% max=X%
+           raw_avg=Ns raw_range=[Ms-Ks]`` — per-coin clamp distribution
+           over the interval.  ``min`` near 100% on a coin is the direct
+           signal that ``DYNAMIC_AGE_MIN`` is biting.
+        """
         if not getattr(self, '_dynamic_age_enabled', False):
             return
         now = time.monotonic()
         if now - self._last_dynamic_age_log < self._dynamic_age_log_interval:
             return
-        self._last_dynamic_age_log = now
 
-        if not self._dynamic_age_recent:
+        if not self._dynamic_age_recent and not self._dynamic_age_clamp_stats:
             return
 
-        parts = [
-            f"{coin}: vol={vol:.2f}bps age={age:.0f}s"
-            for coin, (vol, age) in sorted(self._dynamic_age_recent.items())
-        ]
-        logger.info(f"[mm] Dynamic age: {' | '.join(parts)}")
+        # Snapshot line (existing behavior)
+        if self._dynamic_age_recent:
+            parts = [
+                f"{coin}: vol={vol:.2f}bps age={age:.0f}s"
+                for coin, (vol, age) in sorted(self._dynamic_age_recent.items())
+            ]
+            logger.info(f"[mm] Dynamic age: {' | '.join(parts)}")
+
+        # Clamp stats: one line per coin, sorted by min_clamp pct desc so
+        # the most clamped coin appears first.
+        clamp_rows = []
+        for coin, stats in self._dynamic_age_clamp_stats.items():
+            samples = int(stats.get("samples", 0))
+            if samples <= 0:
+                continue
+            min_pct = 100.0 * stats["min_clamp"] / samples
+            mid_pct = 100.0 * stats["mid"] / samples
+            max_pct = 100.0 * stats["max_clamp"] / samples
+            raw_avg = stats["raw_sum"] / samples
+            raw_min = stats["raw_min"] if stats["raw_min"] != float("inf") else raw_avg
+            raw_max = stats["raw_max"]
+            clamp_rows.append((coin, samples, min_pct, mid_pct, max_pct, raw_avg, raw_min, raw_max))
+
+        clamp_rows.sort(key=lambda r: -r[2])  # min_clamp pct desc
+        for coin, samples, min_pct, mid_pct, max_pct, raw_avg, raw_min, raw_max in clamp_rows:
+            logger.info(
+                f"[mm] dyn-age {coin} samples={samples} "
+                f"min={min_pct:.0f}% mid={mid_pct:.0f}% max={max_pct:.0f}% "
+                f"raw_avg={raw_avg:.0f}s raw_range=[{raw_min:.0f}s-{raw_max:.0f}s]"
+            )
 
+        self._last_dynamic_age_log = now
         # Reset so next log reflects the latest window only
         self._dynamic_age_recent.clear()
+        self._dynamic_age_clamp_stats.clear()
 
     # ------------------------------------------------------------------ #
     #  Helpers

strategies/mm_position_closer.py

@@ -81,6 +81,11 @@ def __init__(
         self._open_positions: Dict[str, Tuple[float, Optional[int], int]] = {}
         # coin -> consecutive cycles where manage() failed to place a close order
         self._consecutive_close_failures: Dict[str, int] = {}
+        # coin -> last effective_max_age seen by manage(); used so close-event
+        # records driven from outside manage() (cleanup_closed,
+        # on_position_closed) can still report the dynamic max_age the bot
+        # was using at the moment the position was alive.
+        self._last_effective_max_age: Dict[str, float] = {}
 
         # Close reason statistics: reason -> count, coin-level: (coin, reason) -> count
         self._close_stats: Dict[str, int] = defaultdict(int)
@@ -97,14 +102,31 @@ def close_stats(self) -> Dict[str, int]:
         """Read-only access to close reason counters."""
         return dict(self._close_stats)
 
-    def _record_close(self, coin: str, reason: str, age: float, tier: int) -> None:
-        """Record a position close event with reason and context."""
+    def _record_close(
+        self,
+        coin: str,
+        reason: str,
+        age: float,
+        tier: int,
+        effective_max_age: Optional[float] = None,
+    ) -> None:
+        """Record a position close event with reason and context.
+
+        ``effective_max_age`` is the max-age the bot was using when the
+        close fired (post-DYNAMIC_AGE). Appended as ``max_age=Ns`` to
+        the log line so post-hoc analysis can tell whether a long
+        ``age`` was driven by a small ``effective_max_age`` (DYNAMIC_AGE
+        clamped low) or by the full grace window. Optional and
+        backward-compatible — the existing log regex
+        ``last_tier=(\\S+)`` still matches the same group.
+        """
         self._close_stats[reason] += 1
         self._close_stats_by_coin[coin][reason] += 1
         tier_name = _TIER_NAMES.get(tier, f"tier{tier}")
+        suffix = f" max_age={effective_max_age:.0f}s" if effective_max_age is not None else ""
         logger.info(
             f"[close-reason] {coin} reason={reason} age={age:.0f}s "
-            f"last_tier={tier_name}"
+            f"last_tier={tier_name}{suffix}"
         )
 
     def log_close_stats(self) -> None:
@@ -148,18 +170,22 @@ def cleanup_closed(self, coin: str) -> None:
             return
         entry_time, close_oid, tier = self._open_positions[coin]
         age = time.monotonic() - entry_time
+        effective_max_age = self._last_effective_max_age.get(coin)
         # Position was closed while we had a close order → likely maker fill
         if close_oid is not None:
-            self._record_close(coin, CLOSE_REASON_MAKER, age, tier)
+            self._record_close(coin, CLOSE_REASON_MAKER, age, tier,
+                               effective_max_age=effective_max_age)
             try:
                 self.order_manager.cancel_order(close_oid, coin)
             except API_ERRORS as e:
                 logger.debug(f"[mm] Could not cancel leftover close order for {coin}: {e}")
         else:
             # No close order was pending → external close (risk manager, etc.)
-            self._record_close(coin, CLOSE_REASON_EXTERNAL, age, tier)
+            self._record_close(coin, CLOSE_REASON_EXTERNAL, age, tier,
+                               effective_max_age=effective_max_age)
         self._open_positions.pop(coin, None)
         self._consecutive_close_failures.pop(coin, None)
+        self._last_effective_max_age.pop(coin, None)
 
     def invalidate_close_order(self, coin: str) -> bool:
         """Cancel and clear the current close order so manage() re-places it.
@@ -196,9 +222,12 @@ def on_position_closed(self, coin: str) -> None:
             entry_time, close_oid, tier = entry
             age = time.monotonic() - entry_time
             reason = CLOSE_REASON_MAKER if close_oid is not None else CLOSE_REASON_EXTERNAL
-            self._record_close(coin, reason, age, tier)
+            effective_max_age = self._last_effective_max_age.get(coin)
+            self._record_close(coin, reason, age, tier,
+                               effective_max_age=effective_max_age)
         self._open_positions.pop(coin, None)
         self._consecutive_close_failures.pop(coin, None)
+        self._last_effective_max_age.pop(coin, None)
 
     def manage(self, coin: str, position: Dict, close_position_fn,
                max_age_override: Optional[float] = None) -> None:
@@ -229,6 +258,9 @@ def manage(self, coin: str, position: Dict, close_position_fn,
         entry_time, close_oid, current_tier = self._open_positions[coin]
         age = now - entry_time
         effective_max_age = max_age_override if max_age_override is not None else self.max_position_age_seconds
+        # Snapshot for close-event logging from cleanup_closed /
+        # on_position_closed paths that don't otherwise know it.
+        self._last_effective_max_age[coin] = effective_max_age
 
         # Unrealized loss early close: taker close when loss exceeds threshold
         if self.unrealized_loss_close_bps > 0 and entry_price > 0:
@@ -256,9 +288,11 @@ def manage(self, coin: str, position: Dict, close_position_fn,
                         f"exceeds threshold {self.unrealized_loss_close_bps}bps -- "
                         f"early taker close (age={age:.0f}s)"
                     )
-                    self._record_close(coin, CLOSE_REASON_UNREALIZED_LOSS, age, current_tier)
+                    self._record_close(coin, CLOSE_REASON_UNREALIZED_LOSS, age, current_tier,
+                                       effective_max_age=effective_max_age)
                     close_position_fn(coin)
                     self._open_positions.pop(coin, None)
+                    self._last_effective_max_age.pop(coin, None)
                     return
 
         # Check if max age exceeded -- force close
@@ -376,9 +410,11 @@ def _handle_force_close(
                 f"[mm] Position {coin} held {age:.0f}s -- force closing with taker order "
                 f"(last_tier={tier_name}, had_close_order={close_oid is not None})"
             )
-            self._record_close(coin, CLOSE_REASON_TAKER_AGE, age, current_tier)
+            self._record_close(coin, CLOSE_REASON_TAKER_AGE, age, current_tier,
+                               effective_max_age=effective_max_age)
             close_position_fn(coin)
             self._open_positions.pop(coin, None)
+            self._last_effective_max_age.pop(coin, None)
             return
 
         # Maker-only close with progressive loss acceptance

tests/test_close_reason_logging.py

@@ -56,6 +56,111 @@ def test_record_close_tracks_per_coin(self):
         assert closer._close_stats_by_coin["ETH"][CLOSE_REASON_TAKER_AGE] == 1
 
 
+class TestRecordCloseEffectiveMaxAge:
+    """_record_close emits the trailing ``max_age=Ns`` field when given an
+    effective_max_age, and omits it otherwise. The existing log regex
+    ``last_tier=(\\S+)`` must still match the same group either way.
+    """
+
+    @staticmethod
+    def _capture_logs(closer, **kwargs) -> str:
+        with patch('strategies.mm_position_closer.logger') as mock_logger:
+            closer._record_close(**kwargs)
+            calls = [str(c) for c in mock_logger.info.call_args_list]
+            close_calls = [c for c in calls if '[close-reason]' in c]
+            return close_calls[0] if close_calls else ""
+
+    def test_max_age_appended_when_provided(self):
+        closer, _, _ = _make_closer()
+        msg = self._capture_logs(
+            closer, coin="BTC", reason=CLOSE_REASON_TAKER_AGE,
+            age=235.0, tier=_TIER_AGGRESSIVE, effective_max_age=120.0,
+        )
+        assert 'reason=taker_age' in msg
+        assert 'age=235s' in msg
+        assert 'last_tier=aggressive' in msg
+        assert 'max_age=120s' in msg
+
+    def test_max_age_omitted_when_none(self):
+        closer, _, _ = _make_closer()
+        msg = self._capture_logs(
+            closer, coin="BTC", reason=CLOSE_REASON_MAKER,
+            age=45.0, tier=_TIER_NORMAL, effective_max_age=None,
+        )
+        assert 'last_tier=normal' in msg
+        assert 'max_age=' not in msg
+
+    def test_existing_close_reason_regex_still_matches(self):
+        # The /review skill greps with this regex; the trailing max_age=
+        # field must NOT break the existing capture groups.
+        import re
+        closer, _, _ = _make_closer()
+        msg = self._capture_logs(
+            closer, coin="xyz:TSLA", reason=CLOSE_REASON_TAKER_AGE,
+            age=235.0, tier=_TIER_AGGRESSIVE, effective_max_age=90.0,
+        )
+        m = re.search(
+            r"\[close-reason\]\s+(\S+)\s+reason=(\S+)\s+age=(\d+)s\s+last_tier=(\S+)",
+            msg,
+        )
+        assert m is not None
+        assert m.group(1) == 'xyz:TSLA'
+        assert m.group(2) == 'taker_age'
+        assert m.group(3) == '235'
+        # group(4) is "aggressive" — \S+ stops at the space before max_age
+        assert m.group(4) == 'aggressive'
+
+    def test_dynamic_max_age_visible_when_clamped_low(self):
+        # The headline use case: distinguish an age=235s close that
+        # happened with effective_max_age=90s (DYNAMIC_AGE_MIN biting)
+        # from one with effective_max_age=120s (default).
+        closer, _, _ = _make_closer()
+        clamped = self._capture_logs(
+            closer, coin="xyz:TSLA", reason=CLOSE_REASON_TAKER_AGE,
+            age=210.0, tier=_TIER_AGGRESSIVE, effective_max_age=90.0,
+        )
+        normal = self._capture_logs(
+            closer, coin="xyz:TSLA", reason=CLOSE_REASON_TAKER_AGE,
+            age=240.0, tier=_TIER_AGGRESSIVE, effective_max_age=120.0,
+        )
+        assert 'max_age=90s' in clamped
+        assert 'max_age=120s' in normal
+
+
+class TestEffectiveMaxAgePropagation:
+    """The effective_max_age recorded in the log line should match what
+    ``manage()`` actually used, and propagate to externally-driven close
+    paths (cleanup_closed, on_position_closed) via _last_effective_max_age.
+    """
+
+    def test_manage_records_last_effective_max_age(self):
+        # max_age_override=90 should be remembered for subsequent
+        # cleanup_closed / on_position_closed that don't see it directly.
+        closer, om, _ = _make_closer(max_age=120, taker_fallback=120)
+        position = {'size': 1.0, 'entry_price': 100.0}
+
+        # The position must already be tracked, so simulate a prior call
+        # by registering it manually.
+        closer._open_positions['BTC'] = (time.monotonic() - 5, None, _TIER_NORMAL)
+
+        # This call sets _last_effective_max_age and then proceeds to
+        # try placing a close order; the tracked value is what we check.
+        closer.manage('BTC', position, lambda c: None, max_age_override=90.0)
+        assert closer._last_effective_max_age.get('BTC') == 90.0
+
+    def test_cleanup_closed_uses_last_effective_max_age(self):
+        closer, _, _ = _make_closer()
+        closer._open_positions['BTC'] = (time.monotonic() - 30, 42, _TIER_BREAKEVEN)
+        closer._last_effective_max_age['BTC'] = 90.0
+
+        with patch('strategies.mm_position_closer.logger') as mock_logger:
+            closer.cleanup_closed('BTC')
+            calls = [str(c) for c in mock_logger.info.call_args_list]
+            assert any('max_age=90s' in c for c in calls)
+        # Cleanup also evicts the snapshot so a re-opened position starts fresh.
+        assert 'BTC' not in closer._last_effective_max_age
+
+
 class TestCleanupClosedReason:
     """cleanup_closed records maker or external close reason."""