refactor(winrate): structured conf_key, improve ConfirmationStore.clear logging, dedupe log fixes, time_to_end instrumentation

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: VirilePeak

src/utils/winrate_upgrade.py

@@ -1,6 +1,7 @@
 import json
 import os
 import time
+import logging
 from threading import Lock
 from typing import Any, Dict, Optional, Tuple
 
@@ -128,17 +129,23 @@ def expire_all_older_than(self, ttl: int) -> int:
 
     def clear(self, key: str) -> bool:
         """Remove a pending confirmation key if present. Returns True if removed."""
+        logger = logging.getLogger("webhook_server_fastapi")
         with self.lock:
+            exists_before = key in self._data
+            logger.info(f"confirmation_clear_attempt: key={key} exists_before={exists_before}")
             if key in self._data:
                 try:
                     del self._data[key]
                     try:
                         self._save()
                     except Exception:
                         pass
+                    logger.info(f"confirmation_cleared: key={key} removed=True")
                     return True
                 except Exception:
+                    logger.exception(f"confirmation_cleared: key={key} removed=False (exception)")
                     return False
+        logger.info(f"confirmation_cleared: key={key} removed=False")
         return False
 
 
@@ -191,5 +198,22 @@ def compute_time_to_market_end(market: Optional[Dict[str, Any]]) -> Tuple[Option
         seconds = int((end - now).total_seconds())
         return seconds, "ok"
     except Exception:
+        # Log end_time_unavailable once per market to avoid spam
+        logger = logging.getLogger("webhook_server_fastapi")
+        try:
+            market_id = None
+            attempted_fields = []
+            if isinstance(market, dict):
+                market_id = market.get("id") or market.get("market_id") or market.get("slug")
+                attempted_fields = [k for k in ("end_time", "close_time", "end") if k in market]
+            key = f"end_unavailable:{market_id or 'unknown'}"
+            # use module-level cache
+            if not hasattr(compute_time_to_market_end, "_logged_keys"):
+                compute_time_to_market_end._logged_keys = set()
+            if key not in compute_time_to_market_end._logged_keys:
+                logger.info(f"end_time_unavailable: market_id/token={market_id} attempted_fields={attempted_fields}")
+                compute_time_to_market_end._logged_keys.add(key)
+        except Exception:
+            logger.exception("error logging end_time_unavailable")
         return None, "end_time_unavailable"
 

webhook_server_fastapi.py

@@ -3469,24 +3469,20 @@ def _check_signal_id_duplicate(signal_id: str, signal: str, request_id: str) ->
         if cache_age < _signal_id_cache_ttl_seconds:
             # Duplicate found within TTL
             _duplicate_signals_count += 1
+            # If confirmation flow is enabled, allow duplicate to proceed so confirmation logic can run
+            if getattr(settings, "WINRATE_UPGRADE_ENABLED", False) and getattr(settings, "REQUIRE_CONFIRMATION", False):
+                # Refresh cache timestamp and allow processing so confirmation handling can run
+                _signal_id_cache[cache_key] = current_time
+                logger.info(
+                    f"[{request_id}] DUPLICATE_BUT_ALLOWED_FOR_CONFIRMATION: signal_id={signal_id}, signal={signal_upper}, cache_key={cache_key} "
+                    f"(age={cache_age:.1f}s, TTL={_signal_id_cache_ttl_seconds}s)"
+                )
+                return False, "duplicate_allowed_for_confirmation"
+            # Default duplicate behavior: skip processing
             logger.info(
-                f"[{request_id}] DUPLICATE SIGNAL: signal_id={signal_id}, signal={signal_upper}, cache_key={cache_key} "
+                f"[{request_id}] DUPLICATE_SIGNAL_SKIPPED: signal_id={signal_id}, signal={signal_upper}, cache_key={cache_key} "
                 f"(age={cache_age:.1f}s, TTL={_signal_id_cache_ttl_seconds}s) - SKIPPED"
             )
-            # If confirmation flow is enabled, allow duplicate to proceed if it's awaiting confirmation
-            try:
-                if getattr(settings, "WINRATE_UPGRADE_ENABLED", False) and getattr(settings, "REQUIRE_CONFIRMATION", False):
-                    conf_key = f"{signal_id}|{signal_upper}"
-                    if hasattr(_confirmation_store, "_data"):
-                        keys_preview = list(_confirmation_store._data.keys())[:20]
-                        logger.debug(f"[{request_id}] confirmation_store keys_preview={keys_preview}")
-                        if conf_key in _confirmation_store._data:
-                            logger.info(f"[{request_id}] DUPLICATE SIGNAL but confirmation pending: allowing for confirmation (key={conf_key})")
-                        # refresh cache timestamp and allow processing so confirmation handling can run
-                        _signal_id_cache[cache_key] = current_time
-                        return False, "duplicate_allowed_for_confirmation"
-            except Exception:
-                pass
             return True, "duplicate"
         else:
             # Expired entry, remove it
@@ -3554,9 +3550,20 @@ def webhook(payload: WebhookPayload):
                                 except Exception:
                                     market = None
                                 up_token, down_token = resolve_up_down_tokens(market) if market else (None, None)
-                                token_or_market = up_token or slug or "unknown_market"
+                                # Build structured confirmation key:
+                                # Prefer market id, else token id, include direction and signal_id to avoid collisions.
                                 sig_id = payload.signal_id or signal_id_for_logging or "no-signal-id"
-                                conf_key = f"{token_or_market}:{sig_for_dedupe}:{sig_id}"
+                                market_id = None
+                                if market and isinstance(market, dict):
+                                    market_id = market.get("id")
+                                token_id = up_token or down_token or None
+                                if market_id:
+                                    conf_key = f\"pm:market:{market_id}:{sig_for_dedupe}:{sig_id}\"
+                                elif token_id:
+                                    conf_key = f\"pm:token:{token_id}:{sig_for_dedupe}:{sig_id}\"
+                                else:
+                                    # Fallback to slug-based key (least preferred)
+                                    conf_key = f\"pm:slug:{slug}:{sig_for_dedupe}:{sig_id}\"
 
                                 # Use new high-level handle API (returns pending/expired/confirmed)
                                 result = _confirmation_store.handle(