""" 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, WEIGHT_CONFIDENCE, ADX_TREND_THRESHOLD, ADX_STRONG_THRESHOLD, REGIME_CONFIDENCE_MIN, ) def compute_structure_score(regime_data: Dict[str, Any]) -> float: trend = regime_data.get("trend", "ranging") structure = regime_data.get("structure", 0) vol_expanding = regime_data.get("vol_expanding", False) vol_contracting = regime_data.get("vol_contracting", False) adx = regime_data.get("adx", 0.0) vol_expanding_from_base = regime_data.get("vol_expanding_from_base", False) if trend == "bullish": base = 0.75 elif trend == "ranging": base = 0.35 else: base = 0.10 # ADX quality modifier if adx >= ADX_STRONG_THRESHOLD: base = min(1.0, base + 0.15) 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) return float(np.clip(base, 0.0, 1.0)) def score_token( regime_data: Dict[str, Any], volume_data: Dict[str, Any], vetoed: bool, ) -> Dict[str, float]: if vetoed: return { "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)) 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 ) # 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), "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]]]: return sorted( scored_map.items(), key=lambda item: item[1].get("total_score", 0.0), reverse=True, ) 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)) bar = "█" * filled + "░" * (width - filled) tier = quality_tier(score) return f"[{bar}] {score:.3f} ({tier})"