feat(trade-executor): defer outside_market_hours signals to next open

Kevin's signals are mid-long term (weeks/months) and he uploads almost
exclusively pre-market or evenings. Before this change, every such
signal hit RiskManager.outside_market_hours, got consumed off the
Redis stream, and was lost. End result: 71 emitted signals, 0 trades.

New behaviour: when RiskManager rejects with outside_market_hours,
push the signal into a Redis sorted-set keyed by next_market_open
(via Alpaca's clock API — handles weekends + holidays). A background
drain task polls the set every kevin_defer_drain_interval_s (60s);
any signal whose target <= now gets re-run through process_signal.

Safety:
  - kevin_max_defer_hours (default 72h) caps signal staleness so we
    don't trade on week-old views.
  - Other RiskManager rejections (cooldown, kill-switch, drawdown
    halt) fall through to the existing drop path.
  - kevin_defer_outside_market_hours toggle defaults True; flip to
    false for legacy behaviour.

Slack: new notify_deferred() emits "🕒 Meet Kevin: DEFERRED
NVDA until Mon 13:30 UTC (market closed; conviction 0.85)" instead
of the noisy outside_market_hours rejection spam.

Tests: 5 queue + 4 integration = 9 new, all 32 trade-executor tests
GREEN.

services/trade_executor/main.py

@@ -13,11 +13,13 @@ import logging
 import signal
 import time
 import uuid
+from datetime import datetime, timezone
 
 from redis.asyncio import Redis
 from sqlalchemy.ext.asyncio import async_sessionmaker
 
 from services.trade_executor.config import TradeExecutorConfig
+from services.trade_executor.deferred_queue import DeferredSignalQueue
 from services.trade_executor.risk_manager import RiskManager
 from services.trade_executor.slack_notifier import SlackNotifier
 from shared.broker.alpaca_broker import AlpacaBroker
@@ -39,6 +41,21 @@ from shared.telemetry import setup_telemetry
 logger = logging.getLogger(__name__)
 
 
+async def _next_market_open(broker: AlpacaBroker) -> datetime:
+    """Alpaca's clock API knows weekends + holidays; use it instead of
+    hardcoding 9:30 ET."""
+    from datetime import timedelta
+
+    clock = await asyncio.to_thread(broker._client.get_clock)
+    next_open = getattr(clock, "next_open", None)
+    if next_open is None:
+        # Defensive fallback — defer by 1 hour and re-check.
+        return datetime.now(timezone.utc) + timedelta(hours=1)
+    if next_open.tzinfo is None:
+        next_open = next_open.replace(tzinfo=timezone.utc)
+    return next_open.astimezone(timezone.utc)
+
+
 async def process_signal(
     signal: TradeSignal,
     risk_manager: RiskManager,
@@ -47,6 +64,8 @@ async def process_signal(
     counters: dict,
     db_session_factory: async_sessionmaker | None = None,
     slack_notifier: SlackNotifier | None = None,
+    deferred_queue: DeferredSignalQueue | None = None,
+    config: TradeExecutorConfig | None = None,
 ) -> None:
     """Process a single trade signal: risk check, order, record, publish.
 
@@ -68,6 +87,36 @@ async def process_signal(
     # --- Step 1: risk check ---
     approved, reason = await risk_manager.check_risk(signal)
     if not approved:
+        # v2: defer outside-market-hours instead of dropping. Kevin's
+        # signals are mid/long-term so a Sunday-evening signal should turn
+        # into a Monday paper trade. Skip if signal is already stale beyond
+        # kevin_max_defer_hours.
+        if (
+            reason == "outside_market_hours"
+            and deferred_queue is not None
+            and config is not None
+            and config.kevin_defer_outside_market_hours
+        ):
+            age_seconds = (
+                datetime.now(timezone.utc) - signal.timestamp
+            ).total_seconds()
+            max_defer_seconds = config.kevin_max_defer_hours * 3600
+            if age_seconds < max_defer_seconds:
+                target = await _next_market_open(broker)
+                await deferred_queue.defer(signal, target)
+                logger.info(
+                    "Signal DEFERRED for %s until %s", signal.ticker, target.isoformat()
+                )
+                counters["rejections"].add(1, {"reason": "deferred"})
+                if slack_notifier is not None:
+                    await slack_notifier.notify_deferred(signal, target)
+                return
+            logger.info(
+                "Signal NOT DEFERRED for %s — age %.1fh exceeds max_defer_hours %.1fh",
+                signal.ticker,
+                age_seconds / 3600,
+                config.kevin_max_defer_hours,
+            )
         logger.info("Signal REJECTED for %s: %s", signal.ticker, reason)
         counters["rejections"].add(1, {"reason": reason.split(" ")[0]})
         if slack_notifier is not None:
@@ -228,12 +277,59 @@ async def run(config: TradeExecutorConfig | None = None) -> None:
 
     logger.info("Consuming from signals:generated, publishing to trades:executed")
 
+    # --- Deferred-signal queue + drain task ---
+    deferred_queue = DeferredSignalQueue(redis)
+    if config.kevin_defer_outside_market_hours:
+        logger.info(
+            "Deferred-signal queue enabled (max_defer=%.1fh, drain_interval=%ds)",
+            config.kevin_max_defer_hours,
+            config.kevin_defer_drain_interval_s,
+        )
+
     # Graceful shutdown on SIGTERM/SIGINT
     shutdown_event = asyncio.Event()
     loop = asyncio.get_running_loop()
     for sig in (signal.SIGTERM, signal.SIGINT):
         loop.add_signal_handler(sig, shutdown_event.set)
 
+    async def _drain_deferred_loop() -> None:
+        """Poll the deferred-signal sorted-set every drain_interval_s and
+        re-run process_signal for any due signals."""
+        while not shutdown_event.is_set():
+            try:
+                due = await deferred_queue.pop_due()
+                for sig_obj, _queued_at in due:
+                    logger.info("Draining deferred signal for %s", sig_obj.ticker)
+                    try:
+                        await process_signal(
+                            sig_obj,
+                            risk_manager,
+                            broker,
+                            publisher,
+                            counters,
+                            db_session_factory,
+                            slack_notifier,
+                            deferred_queue,
+                            config,
+                        )
+                    except Exception:
+                        logger.exception("Drained signal processing failed")
+            except Exception:
+                logger.exception("Drain loop error")
+            try:
+                await asyncio.wait_for(
+                    shutdown_event.wait(),
+                    timeout=config.kevin_defer_drain_interval_s,
+                )
+            except asyncio.TimeoutError:
+                continue
+
+    drain_task = (
+        asyncio.create_task(_drain_deferred_loop())
+        if config.kevin_defer_outside_market_hours
+        else None
+    )
+
     # --- Consume loop ---
     try:
         async for _msg_id, data in consumer.consume():
@@ -249,10 +345,18 @@ async def run(config: TradeExecutorConfig | None = None) -> None:
                     counters,
                     db_session_factory,
                     slack_notifier,
+                    deferred_queue,
+                    config,
                 )
             except Exception:
                 logger.exception("Error processing signal: %s", data)
     finally:
+        if drain_task is not None:
+            drain_task.cancel()
+            try:
+                await drain_task
+            except (asyncio.CancelledError, Exception):
+                pass
         await redis.aclose()
         logger.info("Trade executor stopped gracefully")