Update scorer.py

scorer.py

@@ -1,8 +1,27 @@
+"""
+scorer.py — Multi-factor scoring with regime confidence as 4th dimension.
+
+Key fixes vs prior version:
+- WEIGHT_CONFIDENCE (0.15) added as explicit 4th score axis
+- Absorption hard-zeroes volume_score regardless of other signals
+- Failed breakout penalty applied at scoring level (defence in depth)
+- Structure score uses ADX to weight trend quality, not just HH/HL
+- format_score_bar returns richer display with quality tier label
+"""
+
 from typing import Dict, Any, List, Tuple
 
 import numpy as np
 
-from config import WEIGHT_REGIME, WEIGHT_VOLUME, WEIGHT_STRUCTURE
+from config import (
+    WEIGHT_REGIME,
+    WEIGHT_VOLUME,
+    WEIGHT_STRUCTURE,
+    WEIGHT_CONFIDENCE,
+    ADX_TREND_THRESHOLD,
+    ADX_STRONG_THRESHOLD,
+    REGIME_CONFIDENCE_MIN,
+)
 
 
 def compute_structure_score(regime_data: Dict[str, Any]) -> float:
@@ -10,29 +29,38 @@ def compute_structure_score(regime_data: Dict[str, Any]) -> float:
     structure = regime_data.get("structure", 0)
     vol_expanding = regime_data.get("vol_expanding", False)
     vol_contracting = regime_data.get("vol_contracting", False)
-    atr_trend = regime_data.get("atr_trend", "falling")
+    adx = regime_data.get("adx", 0.0)
+    vol_expanding_from_base = regime_data.get("vol_expanding_from_base", False)
 
     if trend == "bullish":
-        base = 0.8
+        base = 0.75
     elif trend == "ranging":
-        base = 0.4
+        base = 0.35
     else:
-        base = 0.1
+        base = 0.10
 
-    if structure == 1:
+    # ADX quality modifier
+    if adx >= ADX_STRONG_THRESHOLD:
         base = min(1.0, base + 0.15)
-    elif structure == -1:
-        base = max(0.0, base - 0.15)
-
-    if vol_expanding:
-        base = max(0.0, base - 0.1)
-
-    if vol_contracting and trend == "bullish":
+    elif adx < ADX_TREND_THRESHOLD:
+        base = max(0.0, base - 0.20)
+
+    # Structure alignment
+    if structure == 1 and trend == "bullish":
+        base = min(1.0, base + 0.12)
+    elif structure == -1 and trend == "bullish":
+        base = max(0.0, base - 0.20)
+    elif structure == -1 and trend == "bearish":
+        base = min(1.0, base + 0.12)
+
+    # Volatility context
+    if vol_expanding_from_base:
+        base = min(1.0, base + 0.08)
+    if vol_expanding and not vol_expanding_from_base:
+        base = max(0.0, base - 0.10)
+    if vol_contracting:
         base = max(0.0, base - 0.05)
 
-    if atr_trend == "rising" and trend == "bullish":
-        base = min(1.0, base + 0.05)
-
     return float(np.clip(base, 0.0, 1.0))
 
 
@@ -46,33 +74,50 @@ def score_token(
             "regime_score": 0.0,
             "volume_score": 0.0,
             "structure_score": 0.0,
+            "confidence_score": 0.0,
             "total_score": 0.0,
         }
 
     regime_score = float(np.clip(regime_data.get("regime_score", 0.0), 0.0, 1.0))
-    volume_score = float(np.clip(volume_data.get("volume_score", 0.0), 0.0, 1.0))
+    confidence_score = float(np.clip(regime_data.get("regime_confidence", 0.0), 0.0, 1.0))
     structure_score = compute_structure_score(regime_data)
 
+    raw_volume_score = float(np.clip(volume_data.get("volume_score", 0.0), 0.0, 1.0))
+
+    # Absorption hard-zeroes the volume signal regardless of other factors
+    if volume_data.get("absorption", False):
+        volume_score = 0.0
+    elif volume_data.get("failed_breakout", False):
+        # Failed breakout halves the volume score
+        volume_score = raw_volume_score * 0.5
+    else:
+        volume_score = raw_volume_score
+
+    # Climax penalty (not a veto here — defence in depth after veto layer)
+    if volume_data.get("climax", False):
+        volume_score = min(volume_score, 0.30)
+
     total_score = (
         regime_score * WEIGHT_REGIME
         + volume_score * WEIGHT_VOLUME
         + structure_score * WEIGHT_STRUCTURE
+        + confidence_score * WEIGHT_CONFIDENCE
     )
 
-    climax_penalty = 0.15 if volume_data.get("climax", False) else 0.0
-    total_score = float(np.clip(total_score - climax_penalty, 0.0, 1.0))
+    # Confidence multiplier: low confidence compresses total score
+    if confidence_score < REGIME_CONFIDENCE_MIN:
+        total_score *= confidence_score / REGIME_CONFIDENCE_MIN
 
     return {
         "regime_score": round(regime_score, 4),
         "volume_score": round(volume_score, 4),
         "structure_score": round(structure_score, 4),
-        "total_score": round(total_score, 4),
+        "confidence_score": round(confidence_score, 4),
+        "total_score": round(float(np.clip(total_score, 0.0, 1.0)), 4),
     }
 
 
-def rank_tokens(
-    scored_map: Dict[str, Dict[str, Any]],
-) -> List[Tuple[str, Dict[str, Any]]]:
+def rank_tokens(scored_map: Dict[str, Dict[str, Any]]) -> List[Tuple[str, Dict[str, Any]]]:
     return sorted(
         scored_map.items(),
         key=lambda item: item[1].get("total_score", 0.0),
@@ -80,6 +125,20 @@ def rank_tokens(
     )
 
 
-def format_score_bar(score: float, width: int = 20) -> str:
+def quality_tier(score: float) -> str:
+    if score >= 0.80:
+        return "A+"
+    if score >= 0.65:
+        return "A"
+    if score >= 0.50:
+        return "B"
+    if score >= 0.35:
+        return "C"
+    return "D"
+
+
+def format_score_bar(score: float, width: int = 18) -> str:
     filled = int(round(score * width))
-    return "[" + "█" * filled + "░" * (width - filled) + f"] {score:.2f}"
+    bar = "█" * filled + "░" * (width - filled)
+    tier = quality_tier(score)
+    return f"[{bar}] {score:.3f} ({tier})"