|
| 1 | +"""Order rejection classification and aggregation. |
| 2 | +
|
| 3 | +The bot's order placement path (``order_manager.py``) historically emits |
| 4 | +an ERROR-level log for *every* rejection returned by the Hyperliquid |
| 5 | +exchange, regardless of whether the rejection signals a routine retry |
| 6 | +condition (e.g. post-only orders that would have crossed the BBO) or a |
| 7 | +genuine error condition. Under live MM load the routine post-only |
| 8 | +rejection accounts for nearly all ERROR records, drowning out real |
| 9 | +anomalies in monitoring. |
| 10 | +
|
| 11 | +This module reclassifies each rejection by matching the API error text |
| 12 | +against a static set of known "routine" patterns. Matched rejections can |
| 13 | +be downgraded to WARNING / INFO via config, and the tracker accumulates |
| 14 | +per-coin counts that are flushed as a single summary line every |
| 15 | +``rejection_summary_interval`` seconds. |
| 16 | +
|
| 17 | +The tracker is *opt-in*: ``order_manager.py`` only invokes it when one |
| 18 | +has been registered via ``set_rejection_tracker``. With no tracker the |
| 19 | +legacy ERROR-level path is preserved exactly. |
| 20 | +
|
| 21 | +Public surface: |
| 22 | +
|
| 23 | +* :class:`OrderRejectionTracker` — main tracker. |
| 24 | +* :func:`classify_rejection` — pure helper used by tests. |
| 25 | +
|
| 26 | +Both are intentionally cheap on the order-placement hot path: ``record`` |
| 27 | +is O(1) under a short-lived lock, and the periodic summary is driven |
| 28 | +externally by the strategy's main loop (no background thread). |
| 29 | +""" |
| 30 | + |
| 31 | +import logging |
| 32 | +import re |
| 33 | +import threading |
| 34 | +import time |
| 35 | +from collections import defaultdict |
| 36 | +from dataclasses import dataclass |
| 37 | +from typing import Dict, Optional, Tuple |
| 38 | + |
| 39 | +logger = logging.getLogger(__name__) |
| 40 | + |
| 41 | + |
| 42 | +# Pattern matcher → tag. Patterns are evaluated in order; current entries |
| 43 | +# are mutually exclusive on the API's error text format. Adding a new |
| 44 | +# routine pattern (e.g. ``Reduce only``) is a one-line edit here plus a |
| 45 | +# regression test in ``tests/test_order_rejection_tracker.py``. |
| 46 | +_ROUTINE_PATTERNS: Tuple[Tuple["re.Pattern[str]", str], ...] = ( |
| 47 | + (re.compile(r"Post only order would have immediately matched"), |
| 48 | + "post_only_match"), |
| 49 | +) |
| 50 | + |
| 51 | +# Mapping from log-level config string to logging level constant. Keys |
| 52 | +# are normalised to lowercase before lookup so config values like |
| 53 | +# ``"WARNING"`` continue to work. |
| 54 | +_LEVEL_MAP: Dict[str, int] = { |
| 55 | + "error": logging.ERROR, |
| 56 | + "warning": logging.WARNING, |
| 57 | + "info": logging.INFO, |
| 58 | + "debug": logging.DEBUG, |
| 59 | +} |
| 60 | + |
| 61 | +# Allowed config string values, exposed for validators. |
| 62 | +ALLOWED_LOG_LEVELS = tuple(_LEVEL_MAP.keys()) |
| 63 | + |
| 64 | +# Tag used for unmatched rejection text. Always logged at ERROR so a |
| 65 | +# format change on the exchange surfaces immediately. |
| 66 | +UNKNOWN_TAG = "unknown" |
| 67 | + |
| 68 | + |
| 69 | +def classify_rejection(msg: str) -> str: |
| 70 | + """Return the routine tag for *msg*, or :data:`UNKNOWN_TAG` if none match. |
| 71 | +
|
| 72 | + Pure function exposed for tests so the pattern map can be exercised |
| 73 | + without instantiating a tracker. |
| 74 | + """ |
| 75 | + for pat, tag in _ROUTINE_PATTERNS: |
| 76 | + if pat.search(msg): |
| 77 | + return tag |
| 78 | + return UNKNOWN_TAG |
| 79 | + |
| 80 | + |
| 81 | +def _extract_bbo(msg: str) -> str: |
| 82 | + """Best-effort extraction of the ``"bid@ask"`` snippet from a reject text. |
| 83 | +
|
| 84 | + Returns an empty string when the pattern is absent (e.g. older HL |
| 85 | + error messages or future format changes). |
| 86 | + """ |
| 87 | + m = re.search(r"bbo was ([\d.]+@[\d.]+)", msg) |
| 88 | + return m.group(1) if m else "" |
| 89 | + |
| 90 | + |
| 91 | +@dataclass |
| 92 | +class _CoinStats: |
| 93 | + count: int = 0 |
| 94 | + first_ts: float = 0.0 |
| 95 | + last_ts: float = 0.0 |
| 96 | + last_bbo: str = "" |
| 97 | + |
| 98 | + |
| 99 | +class OrderRejectionTracker: |
| 100 | + """Classify and aggregate order-rejection events. |
| 101 | +
|
| 102 | + Public methods are thread-safe. ``record`` is on the order-placement |
| 103 | + hot path and intentionally O(1). |
| 104 | + """ |
| 105 | + |
| 106 | + def __init__( |
| 107 | + self, |
| 108 | + routine_log_level: str = "error", |
| 109 | + summary_interval: float = 300.0, |
| 110 | + ) -> None: |
| 111 | + self._level: int = _LEVEL_MAP.get( |
| 112 | + (routine_log_level or "error").lower(), logging.ERROR |
| 113 | + ) |
| 114 | + self._summary_interval: float = float(summary_interval) |
| 115 | + |
| 116 | + # tag -> coin -> _CoinStats. Reset after each summary flush. |
| 117 | + self._stats: Dict[str, Dict[str, _CoinStats]] = defaultdict( |
| 118 | + lambda: defaultdict(_CoinStats) |
| 119 | + ) |
| 120 | + self._unknown_count: int = 0 |
| 121 | + self._lock = threading.Lock() |
| 122 | + self._last_summary_ts: float = time.monotonic() |
| 123 | + |
| 124 | + # ------------------------------------------------------------------ # |
| 125 | + # Recording |
| 126 | + # ------------------------------------------------------------------ # |
| 127 | + def record(self, coin: str, raw_msg: str) -> str: |
| 128 | + """Record a rejection. Returns the matched tag, or :data:`UNKNOWN_TAG`. |
| 129 | +
|
| 130 | + Emits the per-rejection log at the configured level for routine |
| 131 | + matches; unknown patterns are forwarded to ERROR so format |
| 132 | + changes / new reject reasons remain visible. |
| 133 | + """ |
| 134 | + tag = classify_rejection(raw_msg) |
| 135 | + bbo = _extract_bbo(raw_msg) |
| 136 | + now_wall = time.time() |
| 137 | + |
| 138 | + if tag == UNKNOWN_TAG: |
| 139 | + with self._lock: |
| 140 | + self._unknown_count += 1 |
| 141 | + logger.error(f"Order rejected ({coin}): {raw_msg}") |
| 142 | + return tag |
| 143 | + |
| 144 | + with self._lock: |
| 145 | + stats = self._stats[tag][coin] |
| 146 | + if stats.count == 0: |
| 147 | + stats.first_ts = now_wall |
| 148 | + stats.count += 1 |
| 149 | + stats.last_ts = now_wall |
| 150 | + stats.last_bbo = bbo |
| 151 | + |
| 152 | + # Logger call deliberately outside the lock: the lock guards |
| 153 | + # only the in-memory counters. |
| 154 | + logger.log(self._level, f"[reject:{tag}] {coin} — {raw_msg}") |
| 155 | + return tag |
| 156 | + |
| 157 | + # ------------------------------------------------------------------ # |
| 158 | + # Periodic summary |
| 159 | + # ------------------------------------------------------------------ # |
| 160 | + def log_summary_if_due(self, now_monotonic: Optional[float] = None) -> bool: |
| 161 | + """Emit a summary line if the configured interval has elapsed. |
| 162 | +
|
| 163 | + Returns ``True`` iff a summary was emitted. |
| 164 | + ``summary_interval <= 0`` disables the summary entirely. |
| 165 | + """ |
| 166 | + if self._summary_interval <= 0: |
| 167 | + return False |
| 168 | + now = now_monotonic if now_monotonic is not None else time.monotonic() |
| 169 | + if now - self._last_summary_ts < self._summary_interval: |
| 170 | + return False |
| 171 | + |
| 172 | + with self._lock: |
| 173 | + snapshot = self._stats |
| 174 | + unknown = self._unknown_count |
| 175 | + self._stats = defaultdict(lambda: defaultdict(_CoinStats)) |
| 176 | + self._unknown_count = 0 |
| 177 | + self._last_summary_ts = now |
| 178 | + |
| 179 | + self._emit_summary(snapshot, unknown) |
| 180 | + return True |
| 181 | + |
| 182 | + # ------------------------------------------------------------------ # |
| 183 | + # Internal helpers |
| 184 | + # ------------------------------------------------------------------ # |
| 185 | + def _emit_summary( |
| 186 | + self, |
| 187 | + snapshot: Dict[str, Dict[str, _CoinStats]], |
| 188 | + unknown: int, |
| 189 | + ) -> None: |
| 190 | + any_emitted = False |
| 191 | + for tag, by_coin in snapshot.items(): |
| 192 | + if not by_coin: |
| 193 | + continue |
| 194 | + total = sum(s.count for s in by_coin.values()) |
| 195 | + ranked = sorted( |
| 196 | + by_coin.items(), key=lambda kv: kv[1].count, reverse=True |
| 197 | + ) |
| 198 | + top = ", ".join(f"{coin}={s.count}" for coin, s in ranked[:8]) |
| 199 | + sample_bbo = ranked[0][1].last_bbo if ranked else "" |
| 200 | + logger.info( |
| 201 | + f"[reject-summary] tag={tag} window={self._summary_interval:.0f}s " |
| 202 | + f"total={total} coins={len(by_coin)} top=[{top}] " |
| 203 | + f"sample_bbo={sample_bbo}" |
| 204 | + ) |
| 205 | + any_emitted = True |
| 206 | + if unknown > 0: |
| 207 | + logger.warning( |
| 208 | + f"[reject-summary] tag=unknown window={self._summary_interval:.0f}s " |
| 209 | + f"count={unknown} (unknown patterns logged at ERROR individually)" |
| 210 | + ) |
| 211 | + any_emitted = True |
| 212 | + # No-op when both snapshot and unknown are empty: keeps the log |
| 213 | + # quiet during idle periods rather than emitting a meaningless |
| 214 | + # "total=0" line every interval. |
| 215 | + return None if any_emitted else None |
| 216 | + |
| 217 | + # ------------------------------------------------------------------ # |
| 218 | + # Operational helpers (read-only) |
| 219 | + # ------------------------------------------------------------------ # |
| 220 | + def get_stats_snapshot(self) -> Dict[str, Dict[str, int]]: |
| 221 | + """Read-only counter snapshot for tests / external observers. |
| 222 | +
|
| 223 | + Does not reset internal state (use :meth:`log_summary_if_due` |
| 224 | + for flush semantics). |
| 225 | + """ |
| 226 | + with self._lock: |
| 227 | + return { |
| 228 | + tag: {coin: stats.count for coin, stats in by_coin.items()} |
| 229 | + for tag, by_coin in self._stats.items() |
| 230 | + } |
| 231 | + |
| 232 | + def get_unknown_count(self) -> int: |
| 233 | + with self._lock: |
| 234 | + return self._unknown_count |
0 commit comments