From b341b0dd303b3e4a942a72dac68a8cfaa770c947 Mon Sep 17 00:00:00 2001 From: Thomas Rogers Date: Sat, 25 Jul 2026 22:04:55 -0600 Subject: [PATCH 1/2] feat(clock): read time through the clock seam Phase 2b of the e2e-replay workflow, and the first adoption that touches the capital path. The commit changes WHERE time comes from and never WHAT the logic does: all 849 pre-existing tests pass untouched. 25 semantic sites now read decision_engine.clock: - signal debounce and ranking interval (service) - market-context staleness (market_context) - earnings staleness (checklist) - stale-signal and stale-state eviction, entry/scale-in dates (state_manager) - tier cache TTL (tier_reader) - published event timestamps (kafka_producer, ranker, position_tracker) - end-of-trading-day, trading-day arithmetic, and trade plan validity window (trade_planner) 7 sites deliberately keep the wall clock, because simulated time would make them wrong rather than right: - EVALUATION_DURATION latency metric (simulated time records zero) - the trade planner's Redis circuit-breaker backoff (Redis is a real server during a replay, so 15s must elapse in real seconds) - rules_cache config caching (rules are static across a run, and a simulated TTL would re-read Redis every simulated minute) Naive vs aware is preserved exactly. clock.utcnow() returns naive UTC as a drop-in for datetime.utcnow(), clock.now() returns aware UTC for the trade_planner sites. Signal timestamps, last_update and entry_date are naive throughout, so returning aware datetimes would raise TypeError at every comparison. Tests cover both directions. A module-level accessor is used rather than constructor injection because the reads live in module-level functions and dataclass methods with no constructor to thread through; threading one everywhere would be a far larger diff across code that sizes positions and sets stops. Production 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. Replay fails at initialize() if the driver has not published simulated time. Also drops 5 imports this change orphaned, and removes 61 datetime.utcnow() deprecation warnings under Python 3.12. NOTE: requirements.txt carries a TEMPORARY pin at the trading-py-commons branch feat/clock-module, which is not yet pushed or tagged. This branch will not pass CI until that release is cut. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 46 +++++++ decision_engine/checklist.py | 5 +- decision_engine/clock.py | 130 ++++++++++++++++++ decision_engine/kafka_producer.py | 7 +- decision_engine/market_context.py | 11 +- decision_engine/position_tracker.py | 4 +- decision_engine/ranker.py | 6 +- decision_engine/service.py | 41 +++++- decision_engine/state_manager.py | 10 +- decision_engine/tier_reader.py | 7 +- decision_engine/trade_planner.py | 12 +- requirements.txt | 10 +- tests/test_clock.py | 201 ++++++++++++++++++++++++++++ 13 files changed, 458 insertions(+), 32 deletions(-) create mode 100644 decision_engine/clock.py create mode 100644 tests/test_clock.py diff --git a/README.md b/README.md index 4dd3c63..2197588 100644 --- a/README.md +++ b/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 diff --git a/decision_engine/checklist.py b/decision_engine/checklist.py index e4ba745..4ddcdc1 100644 --- a/decision_engine/checklist.py +++ b/decision_engine/checklist.py @@ -26,7 +26,6 @@ import json import logging -import time from dataclasses import dataclass, field from typing import Optional @@ -34,6 +33,8 @@ from .models.trade_plan import TradePlan +from . import clock as _clock + logger = logging.getLogger(__name__) # Hard gates — trigger BLOCKED status @@ -304,7 +305,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 " diff --git a/decision_engine/clock.py b/decision_engine/clock.py new file mode 100644 index 0000000..4d82492 --- /dev/null +++ b/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) diff --git a/decision_engine/kafka_producer.py b/decision_engine/kafka_producer.py index 2428d80..6a5cb4a 100644 --- a/decision_engine/kafka_producer.py +++ b/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(), } diff --git a/decision_engine/market_context.py b/decision_engine/market_context.py index b6ac288..1c6ba9e 100644 --- a/decision_engine/market_context.py +++ b/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,7 +180,7 @@ 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]: @@ -187,7 +188,7 @@ def get_staleness_seconds(self) -> Optional[float]: 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}") diff --git a/decision_engine/position_tracker.py b/decision_engine/position_tracker.py index 4cecab4..a197ed0 100644 --- a/decision_engine/position_tracker.py +++ b/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": diff --git a/decision_engine/ranker.py b/decision_engine/ranker.py index 8b8bab9..f3c9400 100644 --- a/decision_engine/ranker.py +++ b/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, ) diff --git a/decision_engine/service.py b/decision_engine/service.py index 96cb9ad..b133e74 100644 --- a/decision_engine/service.py +++ b/decision_engine/service.py @@ -35,6 +35,8 @@ # Risk engine integration (mandatory for BUY signal gating) from risk_engine import RiskAdapter +from . import clock as _clock + logger = logging.getLogger(__name__) @@ -105,9 +107,38 @@ def __init__(self, settings: Settings): # Pre-trade checklist evaluator self.checklist_evaluator: Optional[ChecklistEvaluator] = None + def _configure_clock(self) -> None: + """Install the process-wide clock before anything reads time. + + In production this is a no-op that selects the real clock without + opening a connection. Only replay mode needs Redis, so the normal path + pays nothing for the seam. + """ + from trading_commons.clock import MODE_REAL, mode_from_env + + if mode_from_env() == MODE_REAL: + _clock.configure_from_env(None) + return + + client = make_redis( + host=self.settings.redis_host, + port=self.settings.redis_port, + db=self.settings.redis_db, + password=self.settings.redis_password or None, + socket_connect_timeout=2, + socket_timeout=2, + )._create_client() + _clock.configure_from_env(client) + def initialize(self) -> bool: """Initialize all connections and load rules.""" try: + # Resolve the clock first, so every component built below reads + # time from the same source. Under CLOCK_MODE=replay this raises + # if the driver has not published simulated time, failing at + # startup rather than producing a run stamped with today's date. + self._configure_clock() + # Load rules configuration self._config = self.settings.load_rules_config() self.rules, self.rule_weights = RuleRegistry.load_rules_from_config(self._config) @@ -354,11 +385,11 @@ def handle_indicator_event(self, event: dict): return # Parse timestamp - time_str = data.get("time", datetime.utcnow().isoformat()) + time_str = data.get("time", _clock.utcnow().isoformat()) try: timestamp = datetime.fromisoformat(time_str.replace("Z", "+00:00")) except Exception: - timestamp = datetime.utcnow() + timestamp = _clock.utcnow() # Update state self.state_manager.update_indicators(symbol, indicators, timestamp) @@ -653,7 +684,7 @@ def handle_indicator_event(self, event: dict): m.SIGNALS_PUBLISHED.labels( signal_type=aggregated_signal.signal_type.value ).inc() - self._last_publish[symbol] = datetime.utcnow() + self._last_publish[symbol] = _clock.utcnow() # Check if we should publish rankings self._maybe_publish_rankings() @@ -935,7 +966,7 @@ def _should_publish(self, symbol: str, signal: AggregatedSignal) -> bool: return False # Evict stale debounce entries to prevent unbounded growth on Pi - now = datetime.utcnow() + now = _clock.utcnow() if len(self._last_publish) > 100: cutoff = now - timedelta(minutes=30) self._last_publish = { @@ -966,7 +997,7 @@ def _should_publish(self, symbol: str, signal: AggregatedSignal) -> bool: def _maybe_publish_rankings(self): """Publish rankings if interval has elapsed.""" - now = datetime.utcnow() + now = _clock.utcnow() if self._last_ranking_publish: elapsed = (now - self._last_ranking_publish).total_seconds() diff --git a/decision_engine/state_manager.py b/decision_engine/state_manager.py index 25ede0c..304a5aa 100644 --- a/decision_engine/state_manager.py +++ b/decision_engine/state_manager.py @@ -12,6 +12,8 @@ from .models.signals import AggregatedSignal, Signal from .rules.base import SignalType +from . import clock as _clock + logger = logging.getLogger(__name__) @@ -58,7 +60,7 @@ def add_shares(self, price: float, shares: float) -> None: self.total_cost += price * shares self.avg_cost_basis = self.total_cost / self.total_shares self.scale_in_count += 1 - self.last_scale_in_date = datetime.utcnow() + self.last_scale_in_date = _clock.utcnow() def to_dict(self) -> Dict: """Convert to dict for passing to rule context.""" @@ -178,7 +180,7 @@ def get_all_symbols(self) -> List[str]: def clear_stale_signals(self, max_age_seconds: int = 300): """Clear signals older than max_age_seconds.""" with self._lock: - now = datetime.utcnow() + now = _clock.utcnow() cleared = 0 for symbol, state in self._states.items(): @@ -201,7 +203,7 @@ def evict_stale_states(self, max_age_seconds: int = 43200): # Proactive eviction: start at 80% capacity, not 100% if len(self._states) <= int(self.MAX_TRACKED_SYMBOLS * 0.8): return - now = datetime.utcnow() + now = _clock.utcnow() to_remove = [] for symbol, state in self._states.items(): if state.has_position: @@ -254,7 +256,7 @@ def open_position( total_shares=shares, total_cost=price * shares, scale_in_count=0, - entry_date=timestamp or datetime.utcnow(), + entry_date=timestamp or _clock.utcnow(), ) logger.info( f"Opened position: {symbol} - {shares} shares @ ${price:.2f}" diff --git a/decision_engine/tier_reader.py b/decision_engine/tier_reader.py index 019c1dd..f9c73a6 100644 --- a/decision_engine/tier_reader.py +++ b/decision_engine/tier_reader.py @@ -20,12 +20,13 @@ import json import logging import threading -import time from dataclasses import dataclass, field from typing import Dict, Optional, List import redis +from . import clock as _clock + logger = logging.getLogger(__name__) # Default multipliers when tier data is unavailable @@ -135,7 +136,7 @@ def _fetch_tier(self, symbol: str) -> Optional[TierData]: blacklisted=bool(data.get("blacklisted", False)), allowed_regimes=data.get("allowed_regimes"), trade_count=int(data.get("trade_count", 0)), - fetched_at=time.time(), + fetched_at=_clock.timestamp(), ) # Trade count floor: cap at C-tier multipliers if insufficient data @@ -162,7 +163,7 @@ def _get_cached(self, symbol: str) -> Optional[TierData]: """Get tier data from cache, fetching from Redis if stale or missing.""" with self._lock: cached = self._cache.get(symbol) - if cached and (time.time() - cached.fetched_at) < self._cache_ttl: + if cached and (_clock.timestamp() - cached.fetched_at) < self._cache_ttl: return cached # Fetch outside lock to avoid blocking other threads during I/O diff --git a/decision_engine/trade_planner.py b/decision_engine/trade_planner.py index 7566a26..a97ac58 100644 --- a/decision_engine/trade_planner.py +++ b/decision_engine/trade_planner.py @@ -9,7 +9,7 @@ import json import logging -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta from decimal import Decimal from typing import Optional @@ -44,6 +44,8 @@ from .redis_factory import make_redis +from . import clock as _clock + # --------------------------------------------------------------------------- # Helpers @@ -51,7 +53,7 @@ def _end_of_trading_day() -> datetime: """Return today's 21:00 UTC (4 PM ET close + buffer) as a UTC datetime.""" - now = datetime.now(timezone.utc) + now = _clock.now() eod = now.replace(hour=21, minute=0, second=0, microsecond=0) if eod <= now: eod += timedelta(days=1) @@ -60,7 +62,7 @@ def _end_of_trading_day() -> datetime: def _trading_days_from_now(days: int) -> datetime: """Rough approximation: add calendar days (weekends excluded best-effort).""" - now = datetime.now(timezone.utc) + now = _clock.now() added = 0 dt = now while added < days: @@ -378,7 +380,7 @@ def _entry_zone( low = close high = close * 1.01 # Breakouts go or fail fast — 2 hours - valid_until = datetime.now(timezone.utc) + timedelta(hours=2) + valid_until = _clock.now() + timedelta(hours=2) else: # SIGNAL entry = close @@ -837,7 +839,7 @@ def _size_position( def _get_account_balance(self) -> float: """Read account balance from Redis, use cache if fresh, fall back to default.""" - now = datetime.now(timezone.utc) + now = _clock.now() # Return cached value if still fresh if ( diff --git a/requirements.txt b/requirements.txt index 57ab1af..45ea1fd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,8 +3,14 @@ # Kafka kafka-python==2.3.0 -# Shared platform infrastructure (config base, metrics shim, redis base) -trading-py-commons @ git+https://github.com/trogers1052/trading-py-commons.git@v0.2.0 +# Shared platform infrastructure (config base, metrics shim, redis base, clock) +# +# TEMPORARY PIN — MUST BE BUMPED BEFORE MERGE. +# decision_engine/clock.py imports trading_commons.clock, which does not exist +# in v0.2.0. It lives on the trading-py-commons branch feat/clock-module, which +# is not yet pushed or tagged. This branch will fail CI until that release is +# cut; bump the ref below to the new tag (v0.3.0) at that point. +trading-py-commons @ git+https://github.com/trogers1052/trading-py-commons.git@feat/clock-module # Configuration pydantic==2.12.5 diff --git a/tests/test_clock.py b/tests/test_clock.py new file mode 100644 index 0000000..764c83e --- /dev/null +++ b/tests/test_clock.py @@ -0,0 +1,201 @@ +"""Tests for the decision engine's clock seam. + +The point of these is not that the clock works — trading-py-commons covers +that — but that the *service* reads time through it, and that the naive/aware +distinction is preserved. Getting that wrong raises TypeError at the +comparison sites, or silently makes a gate compare against the wall. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest +from trading_commons.clock import ManualClock, SystemClock + +from decision_engine import clock + + +SIMULATED = datetime(2021, 3, 1, 14, 30, tzinfo=UTC) + + +@pytest.fixture(autouse=True) +def restore_clock(): + """Never leak a simulated clock into another test.""" + yield + clock.reset() + + +# --- accessors --------------------------------------------------------------- + + +def test_defaults_to_real_clock(): + assert isinstance(clock.get_clock(), SystemClock) + assert abs((datetime.now(UTC) - clock.now()).total_seconds()) < 60 + + +def test_now_is_aware_utcnow_is_naive(): + """The whole refactor hinges on this distinction.""" + clock.set_clock(ManualClock(SIMULATED)) + + assert clock.now().tzinfo is not None, "now() must stay aware" + assert clock.utcnow().tzinfo is None, "utcnow() must stay naive" + assert clock.now() == SIMULATED + assert clock.utcnow() == SIMULATED.replace(tzinfo=None) + + +def test_naive_and_aware_values_are_usable_in_their_own_comparisons(): + """A mismatch here is the TypeError the mapping exists to prevent.""" + clock.set_clock(ManualClock(SIMULATED)) + + # naive vs naive (state_manager, service, ranker style) + stored_naive = clock.utcnow() - timedelta(minutes=5) + assert (clock.utcnow() - stored_naive).total_seconds() == 300 + + # aware vs aware (trade_planner style) + stored_aware = clock.now() - timedelta(hours=2) + assert (clock.now() - stored_aware).total_seconds() == 7200 + + +def test_timestamp_matches_the_simulated_instant(): + clock.set_clock(ManualClock(SIMULATED)) + + assert clock.timestamp() == SIMULATED.timestamp() + # If it had fallen through to the wall clock this would be ~now. + assert datetime.now(UTC).timestamp() - clock.timestamp() > 365 * 24 * 3600 + + +def test_set_clock_ignores_none(): + clock.set_clock(ManualClock(SIMULATED)) + clock.set_clock(None) + + assert clock.now() == SIMULATED + + +def test_reset_restores_real_clock(): + clock.set_clock(ManualClock(SIMULATED)) + clock.reset() + + assert isinstance(clock.get_clock(), SystemClock) + + +def test_to_naive_utc(): + assert clock.to_naive_utc(None) is None + assert clock.to_naive_utc(SIMULATED) == SIMULATED.replace(tzinfo=None) + naive = datetime(2021, 3, 1, 14, 30) + assert clock.to_naive_utc(naive) == naive + + +# --- configure_from_env ------------------------------------------------------ + + +def test_configure_from_env_defaults_to_real(monkeypatch): + monkeypatch.delenv("CLOCK_MODE", raising=False) + + installed = clock.configure_from_env(None) + + assert isinstance(installed, SystemClock) + assert isinstance(clock.get_clock(), SystemClock) + + +def test_configure_from_env_replay_installs_simulated_clock(monkeypatch): + monkeypatch.setenv("CLOCK_MODE", "replay") + + class FakeRedis: + def get(self, key): + return "2021-03-01T14:30:00Z" + + installed = clock.configure_from_env(FakeRedis()) + + assert not isinstance(installed, SystemClock) + assert clock.now() == SIMULATED + assert clock.utcnow() == SIMULATED.replace(tzinfo=None) + + +def test_configure_from_env_replay_raises_when_time_unavailable(monkeypatch): + """A misconfigured replay must die at startup, not run on today's date.""" + monkeypatch.setenv("CLOCK_MODE", "replay") + + class EmptyRedis: + def get(self, key): + return None + + with pytest.raises(Exception, match="is not set"): + clock.configure_from_env(EmptyRedis()) + + +# --- the gates actually read it --------------------------------------------- + + +def test_stale_signal_eviction_uses_simulated_time(): + """The gate must age off simulated time, not the wall.""" + from decision_engine.state_manager import StateManager + + manual = ManualClock(SIMULATED) + clock.set_clock(manual) + sm = StateManager() + sm.record_signal("AAPL", _signal()) + assert sm.get_state("AAPL").current_signal is not None + + # Advance simulated time past the staleness window. The wall clock has + # barely moved, so this only passes if the gate reads the clock. + manual.advance(hours=1) + sm.clear_stale_signals(max_age_seconds=300) + + assert sm.get_state("AAPL").current_signal is None + + +def test_stale_signal_survives_when_simulated_time_has_not_advanced(): + """The mirror case: no simulated time passing means no eviction.""" + from decision_engine.state_manager import StateManager + + clock.set_clock(ManualClock(SIMULATED)) + sm = StateManager() + sm.record_signal("AAPL", _signal()) + + sm.clear_stale_signals(max_age_seconds=300) + + assert sm.get_state("AAPL").current_signal is not None + + +def test_position_entry_date_uses_simulated_time(): + from decision_engine.state_manager import StateManager + + clock.set_clock(ManualClock(SIMULATED)) + sm = StateManager() + + sm.open_position("AAPL", price=100.0, shares=10.0) + + pos = sm.get_position("AAPL") + assert pos is not None + assert pos.entry_date == SIMULATED.replace(tzinfo=None) + assert pos.entry_date.tzinfo is None, "entry_date must stay naive" + + +def test_scale_in_date_uses_simulated_time(): + from decision_engine.state_manager import StateManager + + manual = ManualClock(SIMULATED) + clock.set_clock(manual) + sm = StateManager() + sm.open_position("AAPL", price=100.0, shares=10.0) + + manual.advance(days=3) + sm.add_to_position("AAPL", price=105.0, shares=5.0) + + pos = sm.get_position("AAPL") + assert pos.last_scale_in_date == (SIMULATED + timedelta(days=3)).replace(tzinfo=None) + + +def _signal(): + """Minimal aggregated signal for state-manager tests.""" + from decision_engine.models.signals import AggregatedSignal, SignalType + + return AggregatedSignal( + symbol="AAPL", + signal_type=SignalType.BUY, + aggregate_confidence=0.9, + primary_reasoning="test", + contributing_signals=[], + timestamp=clock.utcnow(), + ) From 72e97b911489d7966df84ec7240024f4de2304cb Mon Sep 17 00:00:00 2001 From: Thomas Rogers Date: Sun, 26 Jul 2026 06:25:04 -0600 Subject: [PATCH 2/2] deps: pin trading-py-commons v0.3.0 The clock module is released, so the temporary branch pin is replaced with the tag. Full suite (863 tests) passes against the published release. Co-Authored-By: Claude Opus 5 (1M context) --- requirements.txt | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/requirements.txt b/requirements.txt index 45ea1fd..af78d18 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,13 +4,7 @@ kafka-python==2.3.0 # Shared platform infrastructure (config base, metrics shim, redis base, clock) -# -# TEMPORARY PIN — MUST BE BUMPED BEFORE MERGE. -# decision_engine/clock.py imports trading_commons.clock, which does not exist -# in v0.2.0. It lives on the trading-py-commons branch feat/clock-module, which -# is not yet pushed or tagged. This branch will fail CI until that release is -# cut; bump the ref below to the new tag (v0.3.0) at that point. -trading-py-commons @ git+https://github.com/trogers1052/trading-py-commons.git@feat/clock-module +trading-py-commons @ git+https://github.com/trogers1052/trading-py-commons.git@v0.3.0 # Configuration pydantic==2.12.5