feat(clock): read time through the clock seam

README.md

@@ -64,6 +64,52 @@ Edit `config/rules.yaml` to:
 - Adjust thresholds (RSI levels, etc.)
 - Set rule weights for ranking
 
+### Replay mode (`CLOCK_MODE`)
+
+| Env var | Default | Purpose |
+|---------|---------|---------|
+| `CLOCK_MODE` | `real` | `replay` reads simulated time from Redis; anything else (including a typo) is real |
+| `CLOCK_SIM_KEY` | `sim:clock` | Redis key holding an ISO-8601 simulated time |
+
+The engine reads "now" through `decision_engine.clock` rather than calling
+`datetime.utcnow()` directly, so the [e2e-replay](../e2e-replay) harness can
+drive it with simulated time. Every relative-duration gate depends on this:
+signal debounce, ranking interval, context staleness, earnings staleness,
+stale-signal and stale-state eviction, tier cache TTL, and the trade plan's
+validity window.
+
+**Production behaviour is unchanged.** `CLOCK_MODE` defaults to `real`, only
+the exact string `replay` switches modes, and the real path never opens a
+Redis connection for the clock. In replay mode `initialize()` fails fast if the
+driver has not published a simulated time, rather than producing a whole run
+quietly stamped with today's date.
+
+Three things deliberately keep the wall clock, because simulated time would
+make them wrong rather than right:
+
+- `EVALUATION_DURATION` — a latency metric; measuring real work in simulated
+  time records zero.
+- The trade planner's Redis circuit-breaker backoff — Redis is a real server
+  during a replay, so its 15-second backoff must elapse in real seconds.
+- `rules_cache` config caching — rules are static across a run, and a
+  simulated TTL would re-read Redis on every simulated minute.
+
+#### Naive vs aware
+
+`decision_engine.clock` exposes three accessors that are exact drop-in
+replacements preserving the tz-awareness each call site already had:
+
+| Call site used | Replace with | Returns |
+|---|---|---|
+| `datetime.utcnow()` | `clock.utcnow()` | naive UTC |
+| `datetime.now(timezone.utc)` | `clock.now()` | aware UTC |
+| `time.time()` | `clock.timestamp()` | float epoch |
+
+This is not cosmetic. Signal timestamps, `last_update`, and `entry_date` are
+naive throughout the service; returning an aware datetime from `utcnow()` would
+raise `TypeError: can't subtract offset-naive and offset-aware datetimes` at
+every comparison.
+
 ## Running
 
 ```bash

decision_engine/checklist.py

@@ -26,14 +26,15 @@
 
 import json
 import logging
-import time
 from dataclasses import dataclass, field
 from typing import Optional
 
 import redis
 
 from .models.trade_plan import TradePlan
 
+from . import clock as _clock
+
 logger = logging.getLogger(__name__)
 
 # Hard gates — trigger BLOCKED status
@@ -326,7 +327,7 @@ def _get_earnings(self, symbol: str) -> Optional[dict]:
             updated_at = data.get("updated_at")
             if updated_at:
                 try:
-                    age_hours = (time.time() - float(updated_at)) / 3600
+                    age_hours = (_clock.timestamp() - float(updated_at)) / 3600
                     if age_hours > EARNINGS_STALENESS_HOURS:
                         logger.warning(
                             f"Earnings data for {symbol} is {age_hours:.1f}h old "

decision_engine/clock.py

@@ -0,0 +1,130 @@
+"""Process-wide time source for the decision engine.
+
+Every relative-duration gate in this service — signal debounce, ranking
+interval, context staleness, earnings staleness, stale-signal and stale-state
+eviction, tier cache TTL — compares *now* against a stored instant. Reading
+that "now" from the wall clock makes the service impossible to replay: driving
+five years of history past a 2026 wall clock means the debounce never
+suppresses, no staleness gate ever fires, and the trade plan's validity window
+is nonsense. See ``e2e-replay`` for what this enables.
+
+Why a module-level accessor rather than constructor injection
+------------------------------------------------------------
+The time reads are spread across module-level functions
+(``trade_planner._end_of_trading_day``) and dataclass methods
+(``Position.add_shares``) that have no constructor to thread a clock through.
+Threading one everywhere would be a large, risky diff across code that sizes
+positions and sets stops. A single process-wide source keeps the change to
+*where* time comes from, never *what* the logic does.
+
+Naive vs aware
+--------------
+This module deliberately offers three accessors that are **exact drop-in
+replacements** preserving the tz-awareness each call site already had:
+
+===========================  ==========================  =================
+Call site used               Replace with                 Returns
+===========================  ==========================  =================
+``datetime.utcnow()``        :func:`utcnow`               naive UTC
+``datetime.now(timezone.utc)``  :func:`now`               aware UTC
+``time.time()``              :func:`timestamp`            float epoch
+===========================  ==========================  =================
+
+Mixing these up raises ``TypeError: can't subtract offset-naive and
+offset-aware datetimes`` at the comparison sites, so the mapping is not
+cosmetic — it is what keeps the refactor behaviour-preserving.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from trading_commons.clock import Clock, SystemClock, from_env
+
+logger = logging.getLogger(__name__)
+
+_lock = threading.RLock()
+_clock: Clock = SystemClock()
+
+
+def get_clock() -> Clock:
+    """Return the process-wide clock (the real system clock by default)."""
+    with _lock:
+        return _clock
+
+
+def set_clock(clock: Clock) -> None:
+    """Replace the process-wide clock.
+
+    Intended for wiring and tests. Passing ``None`` is ignored rather than
+    leaving the process without a clock.
+    """
+    global _clock
+    if clock is None:
+        return
+    with _lock:
+        _clock = clock
+
+
+def reset() -> None:
+    """Restore the real system clock. For tests."""
+    global _clock
+    with _lock:
+        _clock = SystemClock()
+
+
+def configure_from_env(redis_client: Any = None) -> Clock:
+    """Install the clock selected by ``CLOCK_MODE``.
+
+    Returns the real system clock unless ``CLOCK_MODE=replay``, in which case
+    simulated time is read from Redis. Raises if replay is requested but
+    simulated time cannot be read, so a misconfigured replay fails at startup
+    instead of producing a whole run quietly stamped with today's date.
+    """
+    clock = from_env(redis_client)
+    set_clock(clock)
+    if not isinstance(clock, SystemClock):
+        logger.warning(
+            "CLOCK_MODE=replay — decision engine is running on SIMULATED time (%s). "
+            "This must never be set in production.",
+            clock.now().isoformat(),
+        )
+    return clock
+
+
+# ---------------------------------------------------------------------------
+# Drop-in accessors
+# ---------------------------------------------------------------------------
+
+
+def now() -> datetime:
+    """Aware UTC now. Drop-in for ``datetime.now(timezone.utc)``."""
+    return get_clock().now()
+
+
+def utcnow() -> datetime:
+    """Naive UTC now. Drop-in for ``datetime.utcnow()``.
+
+    The naive form is preserved because the values it is compared against
+    across this service (signal timestamps, state ``last_update``, position
+    ``entry_date``) are themselves naive. Returning an aware datetime here
+    would raise ``TypeError`` at every one of those comparisons.
+    """
+    return get_clock().now().replace(tzinfo=None)
+
+
+def timestamp() -> float:
+    """Epoch seconds. Drop-in for ``time.time()``."""
+    return get_clock().now().timestamp()
+
+
+def to_naive_utc(value: Optional[datetime]) -> Optional[datetime]:
+    """Normalise an arbitrary datetime to naive UTC, or pass through ``None``."""
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value
+    return value.astimezone(timezone.utc).replace(tzinfo=None)

decision_engine/kafka_producer.py

@@ -4,7 +4,6 @@
 
 import json
 import logging
-from datetime import datetime
 from typing import Any, Dict, Optional
 
 from kafka import KafkaProducer
@@ -14,6 +13,8 @@
 from .models.signals import AggregatedSignal
 from .models.trade_plan import TradePlan
 
+from . import clock as _clock
+
 logger = logging.getLogger(__name__)
 
 
@@ -96,7 +97,7 @@ def publish_decision(
                 "event_type": "DECISION_UPDATE",
                 "source": "decision-engine",
                 "schema_version": "1.2",
-                "timestamp": datetime.utcnow().isoformat() + "Z",
+                "timestamp": _clock.utcnow().isoformat() + "Z",
                 "data": {
                     "symbol": signal.symbol,
                     "signal": signal.signal_type.value,
@@ -230,7 +231,7 @@ def publish_ranking(self, ranking_result) -> bool:
                 "event_type": "RANKING_UPDATE",
                 "source": "decision-engine",
                 "schema_version": "1.0",
-                "timestamp": datetime.utcnow().isoformat() + "Z",
+                "timestamp": _clock.utcnow().isoformat() + "Z",
                 "data": ranking_result.to_dict(),
             }
 

decision_engine/market_context.py

@@ -22,12 +22,13 @@
 import json
 import logging
 import threading
-import time
 from datetime import datetime, timezone
 from typing import Optional
 
 import redis
 
+from . import clock as _clock
+
 logger = logging.getLogger(__name__)
 
 # Confidence multipliers applied to BUY signals based on market regime.
@@ -179,15 +180,15 @@ def is_stale(self) -> bool:
                 # Never received context — don't penalise during startup.
                 # The regime is UNKNOWN which already gives neutral multiplier.
                 return False
-            age = time.time() - self._context_updated_at
+            age = _clock.timestamp() - self._context_updated_at
             return age > STALE_THRESHOLD_SECONDS
 
     def get_staleness_seconds(self) -> Optional[float]:
         """Return how many seconds since the last context update, or None."""
         with self._lock:
             if self._context_updated_at is None:
                 return None
-            return time.time() - self._context_updated_at
+            return _clock.timestamp() - self._context_updated_at
 
     # ------------------------------------------------------------------
     # Background thread
@@ -223,10 +224,10 @@ def _refresh(self) -> None:
                     new_updated_at = dt.timestamp()
                 except (ValueError, TypeError):
                     # Fall back to "now" — we at least know the key was refreshed.
-                    new_updated_at = time.time()
+                    new_updated_at = _clock.timestamp()
             else:
                 # No timestamp in payload — use current time as best estimate.
-                new_updated_at = time.time()
+                new_updated_at = _clock.timestamp()
 
         except (redis.RedisError, json.JSONDecodeError, ValueError, TypeError) as exc:
             logger.warning(f"Failed to refresh market context from Redis: {exc}")

decision_engine/position_tracker.py

@@ -14,6 +14,8 @@
 from kafka import KafkaConsumer
 from kafka.errors import KafkaError
 
+from . import clock as _clock
+
 logger = logging.getLogger(__name__)
 
 
@@ -155,7 +157,7 @@ def _handle_message(self, raw_msg: bytes):
                         timestamp_str.replace("Z", "+00:00")
                     )
                 except Exception:
-                    timestamp = datetime.utcnow()
+                    timestamp = _clock.utcnow()
 
             # Process the order
             if side == "buy":

decision_engine/ranker.py

@@ -13,6 +13,8 @@
 from .models.signals import AggregatedSignal
 from .rules.base import SignalType
 
+from . import clock as _clock
+
 logger = logging.getLogger(__name__)
 
 
@@ -121,7 +123,7 @@ def rank(
             return RankingResult(
                 signal_type=signal_type,
                 ranked_symbols=[],
-                timestamp=datetime.utcnow(),
+                timestamp=_clock.utcnow(),
                 criteria_used=self.criteria,
             )
 
@@ -149,7 +151,7 @@ def rank(
         result = RankingResult(
             signal_type=signal_type,
             ranked_symbols=ranked,
-            timestamp=datetime.utcnow(),
+            timestamp=_clock.utcnow(),
             criteria_used=self.criteria,
         )