""" lib/scoring.py — V7 Winning-Level Intent-Aware Elite Composite Scoring V7 upgrades over V6.1: 1. Fixed critical tiebreaker bug (ternary expression broke arithmetic chain) 2. NDCG@10-optimized weights (50% of total score, so top-10 precision matters most) 3. V7 new features integrated: assessment_signal, endorsement_signal, education_tier, cross_validation 4. Tightened behavioral twin penalties (response rate < 0.15 is a hard filter) 5. Improved retrieval bonus weighting 6. Endorsement signal as a skill trust amplifier 7. Assessment signal as a tiebreaker for top-10 positioning V6.1 features retained: - Tier-5 signature boost - Behavioral twin penalty - LangChain-only penalty - Closed-source isolation penalty - Pre-LLM × Ownership interaction - Salary compatibility - Continuous tiebreaker """ from __future__ import annotations import numpy as np import pandas as pd # --- Base elite composite weights (NDCG@10-optimized) --- # NDCG@10 = 50%, NDCG@50 = 30%, MAP = 15%, P@10 = 5% # Top-10 quality is CRITICAL. Impact+Ownership dominates for top-10. W_IMPACT = 0.42 W_OWNERSHIP = 0.33 W_SEARCH = 0.15 W_BEHAVIOUR = 0.10 # Impact sub-weights (production evidence matters more for founding team) W_IMP_MAG = 0.38 W_IMP_SIG = 0.32 W_IMP_DIV = 0.30 # Ownership sub-weights W_OWN_HIER = 0.45 W_OWN_EVID = 0.55 # Search/JD sub-weights W_SRCH_COV = 0.32 W_SRCH_TTL = 0.23 W_SRCH_TRUST = 0.20 W_SRCH_PRE = 0.13 W_SRCH_DEPTH = 0.12 # Behaviour sub-weights W_BEH_TRAJ = 0.22 W_BEH_TRUTH = 0.22 W_BEH_RESP = 0.22 W_BEH_YOE = 0.12 W_BEH_SEN = 0.10 W_BEH_AVAIL = 0.12 HONEYPOT_MULTIPLIER = 0.01 # Additive boost weights W_TIER5_BOOST = 0.065 # V6.1: recover Tier-5 candidates without keywords W_PRE_LLM_OWN_BOOST = 0.045 # V6.1: rare founding-team signal W_IMPACT_OWN_COMBO_BOOST = 0.035 # V6.1: high-impact owners at product cos W_CROSS_VAL_BOOST = 0.025 # V7: multi-dimension agreement def _get_intent_weights(): """Get intent-modulated weights for the elite composite.""" try: from lib.hiring_intent import get_intent intent = get_intent() except Exception: intent = None if intent is None: return W_IMPACT, W_OWNERSHIP, W_SEARCH, W_BEHAVIOUR w_impact = W_IMPACT w_ownership = W_OWNERSHIP w_search = W_SEARCH w_behaviour = W_BEHAVIOUR # Founding team -> boost ownership significantly if intent.ownership_expectation >= 0.8: w_ownership *= 1.25 w_impact *= 0.95 # Production-focused -> boost impact if intent.primary_need == "production_systems": w_impact *= 1.10 w_search *= 0.93 # Startup -> boost ownership, reduce behaviour weight if intent.team_context in ("founding", "early"): w_ownership *= 1.10 w_behaviour *= 0.82 # Specialist -> boost search/JD fit if intent.depth_requirement == "specialist": w_search *= 1.18 w_behaviour *= 0.88 # Scrappy shipping -> boost production-related if intent.shipping_culture == "scrappy": w_impact *= 1.05 w_ownership *= 1.05 # Normalize total = w_impact + w_ownership + w_search + w_behaviour return (w_impact / total, w_ownership / total, w_search / total, w_behaviour / total) _cached_weights = None def get_weights(): global _cached_weights if _cached_weights is None: _cached_weights = _get_intent_weights() return _cached_weights def compute_elite(f: dict) -> float: """Compute intent-aware elite composite from a feature dict.""" w_imp, w_own, w_srch, w_beh = get_weights() # Impact impact = (W_IMP_MAG * f.get("impact_magnitude", 0) + W_IMP_SIG * f.get("impact_signals", 0) + W_IMP_DIV * f.get("production_diversity", 0)) # Ownership ownership = (W_OWN_HIER * f.get("ownership_hierarchy", 0) + W_OWN_EVID * f.get("evidence_strength", 0)) own_prod = f.get("ownership_x_production", 0) if own_prod > 0: ownership = ownership * (0.85 + 0.15 * own_prod) # Pre-LLM x Ownership interaction pre_llm_own = f.get("pre_llm_x_ownership", 0) if pre_llm_own > 0: ownership = ownership * (0.88 + 0.17 * pre_llm_own) # Search/JD fit search_jd = (W_SRCH_COV * f.get("skill_coverage", 0) + W_SRCH_TTL * f.get("title_relevance", 0) + W_SRCH_TRUST * f.get("skill_trust_avg", 0) + W_SRCH_PRE * f.get("pre_llm_months", 0) + W_SRCH_DEPTH * f.get("career_depth_ratio", 0)) skill_yoe = f.get("skill_x_yoe", 0) if skill_yoe > 0: search_jd = search_jd * (0.85 + 0.15 * skill_yoe) # V7: Amplify skill trust with endorsement signal endorsement = f.get("endorsement_signal", 0.5) if endorsement > 0.60: search_jd = search_jd * (0.92 + 0.08 * endorsement) # Behaviour behaviour = (W_BEH_TRAJ * f.get("career_trajectory", 0) + W_BEH_TRUTH * f.get("truthiness", 0) + W_BEH_RESP * f.get("responsiveness", 0) + W_BEH_YOE * f.get("yoe_band_score", 0) + W_BEH_SEN * f.get("seniority", 0) + W_BEH_AVAIL * f.get("availability_score", 0)) coherence = f.get("career_coherence", 0.5) behaviour = behaviour * (0.78 + 0.22 * coherence) # Salary compatibility sal_compat = f.get("salary_compatibility", 0.70) behaviour = behaviour * (0.90 + 0.10 * sal_compat) # Assessment signal as mild behaviour amplifier assessment = f.get("assessment_signal", 0.5) if assessment > 0.70: behaviour = behaviour * (0.95 + 0.05 * assessment) imp_dom = f.get("impact_x_domain", 0) if imp_dom > 0: impact = impact * (0.85 + 0.15 * imp_dom) elite = (w_imp * impact + w_own * ownership + w_srch * search_jd + w_beh * behaviour) return elite def final_score(f: dict) -> float: """V7 final score with all safety layers and additive boosts.""" elite = compute_elite(f) # === ADDITIVE BOOSTS (change ranking order) === # Tier-5 signature boost tier5 = f.get("tier5_signature", 0) if tier5 >= 0.85: elite += W_TIER5_BOOST elif tier5 >= 0.50: elite += W_TIER5_BOOST * 0.5 # Pre-LLM x Ownership combo pre_llm_own = f.get("pre_llm_x_ownership", 0) if pre_llm_own >= 0.50: elite += W_PRE_LLM_OWN_BOOST # Impact x Ownership combo if (f.get("impact_magnitude", 0) >= 0.70 and f.get("ownership_hierarchy", 0) >= 0.70 and f.get("company_quality", 0) >= 0.80): elite += W_IMPACT_OWN_COMBO_BOOST # V7: Cross-validation boost cross_val = f.get("cross_validation", 0) if cross_val >= 0.80: elite += W_CROSS_VAL_BOOST elif cross_val >= 0.60: elite += W_CROSS_VAL_BOOST * 0.5 # === MULTIPLICATIVE PENALTIES === disq = f.get("disqualifier_penalty", 1.0) beh_twin = f.get("behavioral_twin_penalty", 1.0) langchain_pen = f.get("langchain_only_penalty", 1.0) closed_source_pen = f.get("closed_source_penalty", 1.0) skill_cov = f.get("skill_coverage", 0) jd_floor = 0.12 if skill_cov < 0.15 else 1.0 emb = f.get("embedding_sim", 0) retrieval_bonus = 1.0 + 0.10 * max(0.0, (emb + 1.0) / 2.0 - 0.5) # V7: Education tier mild multiplier edu_tier = f.get("education_tier", 0.5) edu_bonus = 1.0 + 0.03 * (edu_tier - 0.5) # Apply all multipliers score = (elite * disq * beh_twin * langchain_pen * closed_source_pen * jd_floor * retrieval_bonus * edu_bonus) # Honeypot is_hp = f.get("is_honeypot", False) if is_hp: score *= HONEYPOT_MULTIPLIER # Title floor title_rel = f.get("title_relevance", 0) if title_rel < 0.10: score = min(score, 0.05) elif title_rel < 0.30: score = min(score, 0.20) # === CONTINUOUS TIEBREAKER (ensures strict ordering) === tiebreaker = ( 0.020 * f.get("tier5_signature", 0) + 0.015 * f.get("embedding_sim", 0) + 0.012 * f.get("evidence_strength", 0) + 0.010 * f.get("impact_magnitude", 0) + 0.008 * f.get("responsiveness", 0) + 0.007 * f.get("cross_validation", 0) + 0.006 * f.get("pre_llm_x_ownership", 0) + 0.005 * f.get("assessment_signal", 0) + 0.005 * f.get("endorsement_signal", 0) + 0.004 * f.get("skill_trust_avg", 0) + 0.003 * f.get("career_trajectory", 0) + 0.002 * f.get("salary_compatibility", 0) + 0.002 * f.get("education_tier", 0) ) score += tiebreaker return max(0.0, min(1.0, score)) # =========================================================================== # Vectorized versions for pandas DataFrames # =========================================================================== def elite_score_vec(df: pd.DataFrame) -> pd.Series: """Vectorized intent-aware elite composite.""" w_imp, w_own, w_srch, w_beh = get_weights() impact = (W_IMP_MAG * df["impact_magnitude"].clip(0, 1) + W_IMP_SIG * df["impact_signals"].clip(0, 1) + W_IMP_DIV * df["production_diversity"].clip(0, 1)) ownership = (W_OWN_HIER * df["ownership_hierarchy"].clip(0, 1) + W_OWN_EVID * df["evidence_strength"].clip(0, 1)) if "ownership_x_production" in df.columns: own_prod = df["ownership_x_production"].clip(0, 1) ownership = ownership * (0.85 + 0.15 * own_prod) if "pre_llm_x_ownership" in df.columns: pre_llm_own = df["pre_llm_x_ownership"].clip(0, 1) ownership = ownership * (0.88 + 0.17 * pre_llm_own) search_jd = (W_SRCH_COV * df["skill_coverage"].clip(0, 1) + W_SRCH_TTL * df["title_relevance"].clip(0, 1) + W_SRCH_TRUST * df["skill_trust_avg"].clip(0, 1) + W_SRCH_PRE * df["pre_llm_months"].clip(0, 1) + W_SRCH_DEPTH * df["career_depth_ratio"].clip(0, 1)) if "skill_x_yoe" in df.columns: skill_yoe = df["skill_x_yoe"].clip(0, 1) search_jd = search_jd * (0.85 + 0.15 * skill_yoe) # V7: Endorsement amplification if "endorsement_signal" in df.columns: endorsement = df["endorsement_signal"].clip(0, 1) search_jd = search_jd * (0.92 + 0.08 * endorsement) behaviour = (W_BEH_TRAJ * df["career_trajectory"].clip(0, 1) + W_BEH_TRUTH * df["truthiness"].clip(0, 1) + W_BEH_RESP * df["responsiveness"].clip(0, 1) + W_BEH_YOE * df["yoe_band_score"].clip(0, 1) + W_BEH_SEN * df["seniority"].clip(0, 1) + W_BEH_AVAIL * df["availability_score"].clip(0, 1)) if "career_coherence" in df.columns: coherence = df["career_coherence"].clip(0, 1) behaviour = behaviour * (0.78 + 0.22 * coherence) if "salary_compatibility" in df.columns: sal = df["salary_compatibility"].clip(0, 1) behaviour = behaviour * (0.90 + 0.10 * sal) # V7: Assessment amplifier if "assessment_signal" in df.columns: assessment = df["assessment_signal"].clip(0, 1) behaviour = behaviour * np.where(assessment > 0.70, 0.95 + 0.05 * assessment, 1.0) if "impact_x_domain" in df.columns: imp_dom = df["impact_x_domain"].clip(0, 1) impact = impact * (0.85 + 0.15 * imp_dom) return (w_imp * impact + w_own * ownership + w_srch * search_jd + w_beh * behaviour) def final_score_vec(df: pd.DataFrame) -> pd.Series: """V7 vectorized final score with all boosts and penalties.""" elite = elite_score_vec(df) # Additive boosts tier5 = df.get("tier5_signature", pd.Series(0, index=df.index)).clip(0, 1) tier5_boost = np.where(tier5 >= 0.85, W_TIER5_BOOST, np.where(tier5 >= 0.50, W_TIER5_BOOST * 0.5, 0.0)) elite = elite + tier5_boost pre_llm_own = df.get("pre_llm_x_ownership", pd.Series(0, index=df.index)).clip(0, 1) elite = elite + np.where(pre_llm_own >= 0.50, W_PRE_LLM_OWN_BOOST, 0.0) combo_mask = ((df["impact_magnitude"].clip(0, 1) >= 0.70) & (df["ownership_hierarchy"].clip(0, 1) >= 0.70) & (df["company_quality"].clip(0, 1) >= 0.80)) elite = elite + np.where(combo_mask, W_IMPACT_OWN_COMBO_BOOST, 0.0) # V7: Cross-validation boost cross_val = df.get("cross_validation", pd.Series(0, index=df.index)).clip(0, 1) cv_boost = np.where(cross_val >= 0.80, W_CROSS_VAL_BOOST, np.where(cross_val >= 0.60, W_CROSS_VAL_BOOST * 0.5, 0.0)) elite = elite + cv_boost # Multiplicative penalties disq = df["disqualifier_penalty"].clip(0.01, 1.0) beh_twin = df.get("behavioral_twin_penalty", pd.Series(1.0, index=df.index)).clip(0.30, 1.0) langchain_pen = df.get("langchain_only_penalty", pd.Series(1.0, index=df.index)).clip(0.30, 1.0) closed_source_pen = df.get("closed_source_penalty", pd.Series(1.0, index=df.index)).clip(0.30, 1.0) skill_cov = df["skill_coverage"] jd_floor = np.where(skill_cov < 0.15, 0.12, 1.0) emb = df.get("embedding_sim", pd.Series(0, index=df.index)) emb_norm = ((emb.clip(-1, 1) + 1.0) / 2.0) retrieval_bonus = 1.0 + 0.10 * (emb_norm - 0.5).clip(0, 1) # V7: Education tier bonus edu_tier = df.get("education_tier", pd.Series(0.5, index=df.index)).clip(0, 1) edu_bonus = 1.0 + 0.03 * (edu_tier - 0.5) score = (elite * disq * beh_twin * langchain_pen * closed_source_pen * jd_floor * retrieval_bonus * edu_bonus) # Honeypot score = pd.Series( np.where(df["is_honeypot"].values, score.values * HONEYPOT_MULTIPLIER, score.values), index=df.index, ) # Title floor title_rel = df["title_relevance"] score = np.where(title_rel < 0.10, np.minimum(score, 0.05), score) score = np.where((title_rel >= 0.10) & (title_rel < 0.30), np.minimum(score, 0.20), score) # Continuous tiebreaker (FIXED — no broken ternary expressions) tiebreaker = ( 0.020 * df.get("tier5_signature", 0).clip(0, 1) + 0.015 * df.get("embedding_sim", 0).clip(-1, 1) + 0.012 * df["evidence_strength"].clip(0, 1) + 0.010 * df["impact_magnitude"].clip(0, 1) + 0.008 * df.get("responsiveness", 0).clip(0, 1) + 0.007 * df.get("cross_validation", 0).clip(0, 1) + 0.006 * df.get("pre_llm_x_ownership", 0).clip(0, 1) + 0.005 * df.get("assessment_signal", 0).clip(0, 1) + 0.005 * df.get("endorsement_signal", 0).clip(0, 1) + 0.004 * df.get("skill_trust_avg", 0).clip(0, 1) + 0.003 * df.get("career_trajectory", 0).clip(0, 1) + 0.002 * df.get("salary_compatibility", 0).clip(0, 1) + 0.002 * df.get("education_tier", 0).clip(0, 1) + 0.001 * df.get("availability_score", 0).clip(0, 1) ) score = score + tiebreaker return pd.Series(score, index=df.index).clip(0, 1)