""" lib/features.py — V5 JD-Driven Feature Extraction ~40 features organized in 9 groups. Every feature is JD-driven: the JD parser and domain taxonomy determine what skills matter, not hardcoded lists. Feature groups: G1: JD Fit (7 features) — How well does the candidate match JD requirements G2: Impact & Ownership (4) — Quantified outcomes and ownership level G3: Production & Scale (3) — Real-world deployment evidence G4: Experience & Career (8) — Career trajectory, depth, stability G5: Retrieval & Eval (3) — Domain-specific retrieval/evaluation experience G6: Behavioural (7) — Platform signals and availability G7: Resume Quality (4) — Evidence density, truthiness, stuffing risk G8: Safety (2) — Honeypot detection, disqualifiers G9: Location (1) — Location match G10: Embedding (1) — Semantic similarity (computed in precompute) """ from __future__ import annotations import re from lib import schema from lib.jd_parser import get_jd, JDUnderstanding from lib.domain import get_taxonomy, DomainTaxonomy from lib.evidence import extract_all_evidence, get_evidence_summary, _OWNERSHIP_RE, _OWNERSHIP_TIERS from lib.constants import REFERENCE_DATE from lib import company_tier, title_scoring, honeypot PROF_WEIGHT = {"beginner": 0.25, "intermediate": 0.5, "advanced": 0.75, "expert": 1.0} # ML-relevant skill assessment keys _ML_ASSESSMENT_KEYS = { "python", "machine learning", "nlp", "natural language", "deep learning", "statistics", "pytorch", "tensorflow", "retrieval", "ranking", "recommendation", } # =========================================================================== # Feature names (ordered, for documentation and parquet columns) # =========================================================================== FEATURE_NAMES = [ # G1: JD Fit "skill_coverage", # 1 — Required skill coverage "preferred_coverage", # 2 — Preferred skill coverage "domain_specialization", # 3 — Depth in JD's primary domain "skill_trust_avg", # 4 — Avg trusted skill signal "title_relevance", # 5 — Title match to JD "seniority", # 6 — Seniority level "jd_skill_count", # 7 — Raw count of JD skills found # G2: Impact & Ownership "ownership_hierarchy", # 8 — Hierarchical ownership scoring "impact_magnitude", # 9 — Best quantified impact "impact_signals", # 10 — Non-quantified impact language "evidence_strength", # 11 — Evidence density/quality # G3: Production & Scale "production_strength", # 12 — Production deployment evidence "production_diversity", # 13 — Variety of production signals "scale_evidence", # 14 — System scale metrics # G4: Experience & Career "yoe_band_score", # 15 — YoE match to JD range "career_depth_ratio", # 16 — Fraction of career in domain "pre_llm_months", # 17 — Pre-2022 IR experience (normalized) "career_trajectory", # 18 — Career progression quality "company_quality", # 19 — Current company tier "company_quality_avg", # 20 — Average company quality "career_stability", # 21 — Average tenure length "promotion_velocity", # 22 — Speed of promotions # G5: Retrieval & Evaluation "retrieval_depth", # 23 — Retrieval system sophistication "evaluation_experience", # 24 — Evaluation framework experience "system_design_evidence", # 25 — System design work # G6: Behavioural "recency", # 26 — How recently active "responsiveness", # 27 — Response rate + speed "market_demand", # 28 — Recruiter interest "github_activity", # 29 — Code activity "availability_score", # 30 — Open to work + notice "interview_completion", # 31 — Interview follow-through "platform_trust", # 32 — Verification + completeness # G7: Resume Quality "quantified_outcomes", # 33 — Number of quantified achievements "truthiness", # 34 — Cross-validation of claims "keyword_stuffing_risk", # 35 — Probability of keyword stuffing "profile_completeness", # 36 — Profile completeness score # G8: Safety "disqualifier_penalty", # 37 — Multi-factor penalty (multiplicative) "is_honeypot", # 38 — Synthetic profile flag # G9: Location "location_score", # 39 — Location match # G10: Embedding (computed in precompute) # "embedding_sim", # 40 — Semantic similarity to ideal candidate ] # =========================================================================== # G1: JD FIT # =========================================================================== def skill_coverage(c: dict, text: str | None = None) -> tuple[float, dict]: """G1-1: Fraction of REQUIRED (Tier 1) JD skills found in context.""" if text is None: text = schema.unified_text_blob(c) jd = get_jd() tax = get_taxonomy() total_skills = 0 matched_skills = 0 evidence = {"matched": [], "missing": []} for domain, skills in jd.required_skills.items(): for skill in skills: total_skills += 1 if skill in text: matched_skills += 1 evidence["matched"].append((skill, domain)) else: evidence["missing"].append((skill, domain)) if total_skills == 0: return 0.0, evidence score = min(1.0, matched_skills / total_skills) return score, evidence def preferred_coverage(c: dict, text: str | None = None) -> tuple[float, dict]: """G1-2: Fraction of PREFERRED (Tier 2) JD skills found in context.""" if text is None: text = schema.unified_text_blob(c) jd = get_jd() total_skills = 0 matched_skills = 0 evidence = {"matched": []} for domain, skills in jd.preferred_skills.items(): for skill in skills: total_skills += 1 if skill in text: matched_skills += 1 evidence["matched"].append((skill, domain)) if total_skills == 0: return 0.0, evidence return min(1.0, matched_skills / total_skills), evidence def domain_specialization(c: dict, text: str | None = None) -> tuple[float, dict]: """G1-3: How deeply the candidate is in the JD's primary domain.""" if text is None: text = schema.unified_text_blob(c) jd = get_jd() tax = get_taxonomy() # Get the JD's primary domain skills (all tiers) primary_domain = jd.domain all_domain_skills = set() for tier_dict in [tax.tier1, tax.tier2, tax.tier3]: if primary_domain in tier_dict: all_domain_skills.update(tier_dict[primary_domain]) if not all_domain_skills: return 0.0, {"domain": primary_domain} # Count how many domain skills the candidate has in context found = [s for s in all_domain_skills if s in text] score = min(1.0, len(found) / max(len(all_domain_skills) * 0.4, 1)) # Bonus for Tier 1 skills tier1_skills = tax.tier1.get(primary_domain, []) tier1_found = [s for s in tier1_skills if s in text] if tier1_found: score = min(1.0, score + 0.15) return score, {"domain": primary_domain, "found_count": len(found), "total_domain_skills": len(all_domain_skills)} def skill_trust_avg(c: dict) -> tuple[float, dict]: """G1-4: Weighted average of trusted skill signals for JD-relevant skills.""" jd = get_jd() tax = get_taxonomy() # Collect all JD skills (Tier 1 + 2) jd_skills = set() for d in [tax.tier1, tax.tier2]: for skills in d.values(): jd_skills.update(skills) if not jd_skills: return 0.5, {"matched_count": 0} candidate_skills = schema.skills(c) scores = [] for s in candidate_skills: name = (s.get("name") or "").lower() if any(jd_sk in name for jd_sk in jd_skills): prof_w = PROF_WEIGHT.get(s.get("proficiency"), 0.25) dur_gate = 1.0 if (s.get("duration_months") or 0) > 0 else 0.4 endorsements = min((s.get("endorsements") or 0), 50) / 50.0 trust = 0.40 * prof_w + 0.30 * dur_gate + 0.30 * endorsements scores.append(trust) if not scores: return 0.0, {"matched_count": 0} return sum(scores) / len(scores), {"matched_count": len(scores)} def jd_skill_count(c: dict, text: str | None = None) -> tuple[float, dict]: """G1-7: Raw count of JD skills found, normalized.""" if text is None: text = schema.unified_text_blob(c) jd = get_jd() all_jd_skills = set() for d in [jd.required_skills, jd.preferred_skills]: for skills in d.values(): all_jd_skills.update(skills) found = [s for s in all_jd_skills if s in text] # Normalize: ~5 skills is "full coverage" for this JD return min(1.0, len(found) / 5.0), {"count": len(found)} # =========================================================================== # G2: IMPACT & OWNERSHIP # =========================================================================== def ownership_hierarchy(c: dict, text: str | None = None) -> tuple[float, dict]: """G2-1: Hierarchical ownership scoring from career descriptions.""" if text is None: text = schema.unified_text_blob(c) matches = _OWNERSHIP_RE.findall(text) if not matches: return 0.0, {"best_verb": "none"} # Weighted average of ownership verbs found weights = [] best_verb = "" best_weight = 0 for m in matches: v = m.lower() w = _OWNERSHIP_TIERS.get(v, 0.15) weights.append(w) if w > best_weight: best_weight = w best_verb = v avg_weight = sum(weights) / len(weights) # Bonus for top-tier ownership (architected/spearheaded/owned) top_tier_bonus = 1.15 if best_weight >= 0.85 else 1.0 score = min(1.0, avg_weight * top_tier_bonus) return score, {"best_verb": best_verb, "best_weight": best_weight, "verb_count": len(weights)} def impact_magnitude(c: dict, text: str | None = None) -> tuple[float, dict]: """G2-2: Strength of best quantified impact metric.""" if text is None: text = schema.unified_text_blob(c) patterns = [ (r'improved\s+\w+\s+by\s+(\d+(?:\.\d+)?)\s*%', 0.9), (r'reduced\s+latency\s+from\s+(\d+)\s*ms\s+to\s+(\d+)\s*ms', 1.0), (r'(?:p99|p95)\s*(?:latency\s*)?(?:of\s*)?(\d+)\s*ms', 0.85), (r'ndcg.*?(0\.\d{2,3})', 0.95), (r'recall@?\d+\s*(?:improved\s*)?(?:to\s*)?(\d+(?:\.\d+)?)\s*%', 0.90), (r'(\d+(?:\.\d+)?)\s*million\s+(?:daily\s+)?(?:active\s+)?users', 0.80), (r'(\d+(?:\.\d+)?)\s*k?\s*(?:qps|rps|requests?\s*per\s*sec)', 0.85), ] best_strength = 0.0 best_metric = "" for pattern, weight in patterns: m = re.search(pattern, text, re.IGNORECASE) if m: strength = weight groups = [g for g in m.groups() if g is not None] metric = " / ".join(groups) if strength > best_strength: best_strength = strength best_metric = metric return best_strength, {"best_metric": best_metric} def impact_signals(c: dict, text: str | None = None) -> tuple[float, dict]: """G2-3: Non-quantified impact language.""" if text is None: text = schema.unified_text_blob(c) impact_phrases = [ "significantly improved", "dramatically improved", "substantially improved", "increased engagement", "boosted conversion", "reduced churn", "improved ranking quality", "better relevance", "higher click-through", "reduced false positives", "improved precision", "increased recall", "faster inference", "lower latency", "higher throughput", "scaled to", "grew to", "expanded to", ] found = [p for p in impact_phrases if p in text] score = min(1.0, len(found) / 3.0) return score, {"count": len(found)} def evidence_strength(c: dict) -> tuple[float, dict]: """G2-4: Evidence density and quality (from evidence engine).""" summary = get_evidence_summary(c) score = min(1.0, summary["top3_avg"] / 18.0) # 18 = very strong top-3 return score, { "best_score": summary["best_score"], "count": summary["count"], "top3_avg": round(summary["top3_avg"], 1), } # =========================================================================== # G3: PRODUCTION & SCALE # =========================================================================== def production_strength(c: dict, text: str | None = None) -> tuple[float, list[str]]: """G3-1: Production deployment evidence.""" if text is None: text = schema.unified_text_blob(c) jd = get_jd() hits = [p for p in jd.production_evidence if p in text] unique_hits = list(set(hits)) score = min(1.0, len(unique_hits) / 5.0) return score, unique_hits[:3] def production_diversity(c: dict, text: str | None = None) -> tuple[float, dict]: """G3-2: Variety of production signals (different categories).""" if text is None: text = schema.unified_text_blob(c) categories = { "deployment": ["deployed", "shipped", "launched", "rollout"], "live_traffic": ["live traffic", "real users", "production"], "scale": ["at scale", "throughput", "qps", "latency"], "operations": ["on-call", "monitoring", "sla", "p99"], "testing": ["a/b test", "ab test", "canary", "integration test"], } found_cats = [] for cat, phrases in categories.items(): if any(p in text for p in phrases): found_cats.append(cat) return min(1.0, len(found_cats) / 3.0), {"categories": found_cats} def scale_evidence(c: dict, text: str | None = None) -> tuple[float, dict]: """G3-3: Evidence of system scale.""" if text is None: text = schema.unified_text_blob(c) scale_patterns = [ (r'(\d+(?:\.\d+)?)\s*million', 0.9, "user_scale"), (r'(\d[\d,]*)\s*(?:requests?|queries?)\s*(?:per\s*(?:sec|second|min|minute)|\/\s*(?:sec|s))', 0.85, "qps"), (r'(\d+)\s*(?:tb|gb|pb)\s+of\s+data', 0.70, "data_scale"), (r'(\d[\d,]*)\s*concurrent', 0.60, "concurrency"), ] best = 0.0 best_metric = "" for pattern, weight, label in scale_patterns: m = re.search(pattern, text, re.IGNORECASE) if m and weight > best: best = weight best_metric = f"{m.group(1)} ({label})" return best, {"best_metric": best_metric} # =========================================================================== # G4: EXPERIENCE & CAREER # =========================================================================== def yoe_band_score(c: dict, jd: JDUnderstanding | None = None) -> tuple[float, dict]: """G4-1: YoE match to JD's experience range.""" if jd is None: jd = get_jd() yoe = schema.years_of_experience(c) low, high = jd.yoe_low, jd.yoe_high soft_low, soft_high = low - 1, high + 2 if low <= yoe <= high: score = 1.0 elif soft_low <= yoe < low: score = 0.55 + 0.45 * (yoe - soft_low) / (low - soft_low) elif high < yoe <= soft_high: score = 1.0 - 0.45 * (yoe - high) / (soft_high - high) else: score = 0.2 return max(0.0, min(1.0, score)), {"yoe": yoe, "band": f"{low}-{high}"} def career_depth_ratio(c: dict, jd: JDUnderstanding | None = None) -> tuple[float, dict]: """G4-2: Fraction of career time in JD-relevant domain.""" if jd is None: jd = get_jd() ch = schema.career_history(c) total_months = sum((r.get("duration_months") or 0) for r in ch) or 1 relevant_kw = set() for skills in jd.required_skills.values(): relevant_kw.update(skills) for skills in jd.preferred_skills.values(): relevant_kw.update(skills) relevant_kw.update(jd.pre_llm_keywords) relevant_months = 0 for r in ch: role_text = f"{r.get('title','')} {r.get('description','')}".lower() if any(kw in role_text for kw in relevant_kw): relevant_months += (r.get("duration_months") or 0) ratio = min(1.0, relevant_months / total_months) return ratio, {"ratio": round(ratio, 2), "relevant_months": relevant_months} def pre_llm_months(c: dict, jd: JDUnderstanding | None = None) -> tuple[float, dict]: """G4-3: Pre-2022 IR/search experience, normalized.""" if jd is None: jd = get_jd() ch = schema.career_history(c) evidence = {} for role in ch: sd = schema.parse_date(role.get("start_date")) if sd and sd.year < jd.pre_llm_cutoff_year: role_text = f"{role.get('title','')} {role.get('description','')}".lower() if any(marker in role_text for marker in jd.post_llm_markers): continue hits = [kw for kw in jd.pre_llm_keywords if kw in role_text] if hits: months = role.get("duration_months") or 0 # Normalize: 36+ months = full score score = min(1.0, months / 36.0) evidence["company"] = role.get("company", "") evidence["keywords"] = hits[:2] return score, evidence return 0.0, evidence def career_trajectory(c: dict) -> tuple[float, dict]: """G4-4: Career progression quality (title upgrades, company quality trend).""" ch = schema.career_history(c) if len(ch) < 2: return 0.5, {"reason": "single_role"} # Check title progression (from oldest to newest) seniority_scores = [] for role in reversed(ch): # oldest first sr, _ = title_scoring.seniority_score(role.get("title", "")) seniority_scores.append(sr) # Score upward trajectory upward_moves = 0 for i in range(1, len(seniority_scores)): if seniority_scores[i] > seniority_scores[i-1] + 0.05: upward_moves += 1 traj_score = min(1.0, 0.3 + upward_moves * 0.25) # Company quality trend cq_scores = [company_tier.company_quality_score(r.get("company", "")) for r in reversed(ch)] improving = sum(1 for i in range(1, len(cq_scores)) if cq_scores[i] > cq_scores[i-1] + 0.05) score = 0.6 * traj_score + 0.4 * min(1.0, 0.3 + improving * 0.25) return min(1.0, score), {"upward_moves": upward_moves, "company_improving": improving} def company_quality_avg(c: dict) -> tuple[float, dict]: """G4-6: Average company quality across career.""" ch = schema.career_history(c) if not ch: return 0.5, {"count": 0} scores = [company_tier.company_quality_score(r.get("company", "")) for r in ch] return sum(scores) / len(scores), {"count": len(scores)} def career_stability(c: dict) -> tuple[float, dict]: """G4-7: Average tenure length (penalize job hopping).""" ch = schema.career_history(c) if len(ch) < 2: return 0.6, {"avg_tenure_months": ch[0].get("duration_months", 0) if ch else 0} total_months = sum((r.get("duration_months") or 0) for r in ch) avg_tenure = total_months / len(ch) # Score: 24+ months = 1.0, 12 months = 0.6, <12 = 0.3 if avg_tenure >= 24: score = 1.0 elif avg_tenure >= 12: score = 0.6 + 0.4 * (avg_tenure - 12) / 12 else: score = 0.3 + 0.3 * avg_tenure / 12 return min(1.0, score), {"avg_tenure_months": round(avg_tenure), "roles": len(ch)} def promotion_velocity(c: dict) -> tuple[float, dict]: """G4-8: Speed of title promotions.""" ch = schema.career_history(c) if len(ch) < 2: return 0.5, {"promotions": 0} # Count title level upgrades levels = [] for role in reversed(ch): # oldest first title = role.get("title", "").lower() if any(w in title for w in ["principal", "distinguished", "fellow", "vp", "chief"]): levels.append(5) elif any(w in title for w in ["staff", "lead", "director", "head of"]): levels.append(4) elif "senior" in title or "sr." in title or "sr " in title: levels.append(3) elif any(w in title for w in ["junior", "jr.", "jr ", "intern", "associate"]): levels.append(1) else: levels.append(2) promotions = sum(1 for i in range(1, len(levels)) if levels[i] > levels[i-1]) total_years = schema.years_of_experience(c) or 1 score = min(1.0, 0.4 + promotions * 0.2) return score, {"promotions": promotions, "years": total_years} # =========================================================================== # G5: RETRIEVAL & EVALUATION # =========================================================================== def retrieval_depth(c: dict, text: str | None = None) -> tuple[float, dict]: """G5-1: Retrieval system sophistication.""" if text is None: text = schema.unified_text_blob(c) retrieval_keywords = [ "bm25", "elasticsearch", "opensearch", "faiss", "pinecone", "weaviate", "qdrant", "milvus", "vector database", "vector db", "hybrid search", "hybrid retrieval", "semantic search", "dense retrieval", "ann", "embedding index", "cross-encoder", "bi-encoder", "colbert", "re-ranking", "query understanding", "relevance feedback", ] found = [kw for kw in retrieval_keywords if kw in text] # Score based on sophistication has_vector = any(kw in found for kw in ["faiss", "pinecone", "weaviate", "qdrant", "milvus", "vector db", "vector database"]) has_hybrid = any(kw in found for kw in ["hybrid search", "hybrid retrieval", "bm25", "elasticsearch", "opensearch"]) has_dense = any(kw in found for kw in ["dense retrieval", "semantic search", "embedding index", "ann"]) has_advanced = any(kw in found for kw in ["cross-encoder", "bi-encoder", "colbert", "re-ranking"]) score = 0.0 if found: score += min(0.4, len(found) * 0.1) if has_vector: score += 0.2 if has_hybrid: score += 0.2 if has_dense: score += 0.1 if has_advanced: score += 0.1 return min(1.0, score), {"found": found[:5], "has_hybrid": has_hybrid, "has_vector": has_vector, "has_advanced": has_advanced} def evaluation_experience(c: dict, text: str | None = None) -> tuple[float, dict]: """G5-2: Evaluation framework experience.""" if text is None: text = schema.unified_text_blob(c) eval_keywords = [ "ndcg", "mrr", "map@", "precision@", "recall@", "a/b test", "ab test", "offline evaluation", "online evaluation", "evaluation framework", "evaluation pipeline", "offline-to-online", "ranking quality", "relevance judgment", ] found = [kw for kw in eval_keywords if kw in text] has_ndcg = "ndcg" in found has_ab = any("a/b" in kw or "ab" in kw for kw in found) has_offline = "offline evaluation" in text or "offline-to-online" in text score = min(1.0, len(found) * 0.2) if has_ndcg and has_ab: score += 0.2 if has_offline: score += 0.1 return min(1.0, score), {"found": found[:5], "has_ndcg": has_ndcg, "has_ab": has_ab} def system_design_evidence(c: dict, text: str | None = None) -> tuple[float, dict]: """G5-3: System design / architecture work.""" if text is None: text = schema.unified_text_blob(c) design_keywords = [ "architecture", "designed", "system design", "tech design", "end-to-end", "microservice", "service-oriented", "data pipeline", "ml pipeline", "feature pipeline", "api design", "scalable", "distributed", ] found = [kw for kw in design_keywords if kw in text] score = min(1.0, len(found) * 0.15) return score, {"found": found[:5]} # =========================================================================== # G6: BEHAVIOURAL # =========================================================================== def recency(c: dict) -> tuple[float, dict]: """G6-1: How recently active on the platform.""" s = schema.signals(c) last_active = schema.parse_date(s.get("last_active_date")) evidence = {} if last_active: days = (REFERENCE_DATE - last_active).days evidence["days_since_active"] = days if days <= 30: score = 1.00 elif days <= 60: score = 0.92 elif days <= 90: score = 0.82 elif days <= 180: score = 0.70 else: score = 0.55 else: evidence["days_since_active"] = 90 score = 0.82 return score, evidence def responsiveness(c: dict) -> tuple[float, dict]: """G6-2: Response rate + speed.""" s = schema.signals(c) evidence = {} rr = s.get("recruiter_response_rate") rr = float(rr) if isinstance(rr, (int, float)) else 0.3 evidence["recruiter_response_rate"] = rr if rr >= 0.70: resp = 1.00 elif rr >= 0.50: resp = 0.92 elif rr >= 0.30: resp = 0.82 elif rr >= 0.15: resp = 0.68 else: resp = 0.55 speed_h = s.get("avg_response_time_hours") speed_h = float(speed_h) if isinstance(speed_h, (int, float)) else 48.0 speed_sc = max(0.0, 1.0 - speed_h / 168.0) score = 0.65 * resp + 0.35 * speed_sc return score, evidence def market_demand(c: dict) -> tuple[float, dict]: """G6-3: Recruiter interest signals.""" s = schema.signals(c) saves = min(float(s.get("saved_by_recruiters_30d") or 0), 20.0) / 20.0 appearances = min(float(s.get("search_appearance_30d") or 0), 200.0) / 200.0 return 0.6 * saves + 0.4 * appearances, {"saves": s.get("saved_by_recruiters_30d", 0)} def github_activity(c: dict) -> tuple[float, dict]: """G6-4: GitHub coding activity.""" s = schema.signals(c) gh = s.get("github_activity_score") gh = float(gh) if isinstance(gh, (int, float)) else -1.0 if gh >= 50: score = 1.00 elif gh >= 20: score = 0.92 elif gh >= 5: score = 0.82 elif gh >= 0: score = 0.72 else: score = 0.65 # No GitHub linked, not penalized to zero return score, {"github_activity_score": gh} def availability_score(c: dict) -> tuple[float, dict]: """G6-5: Open to work + notice period.""" s = schema.signals(c) evidence = {} # Open to work otw = 1.05 if s.get("open_to_work_flag") else 0.95 # Notice period notice = s.get("notice_period_days") notice = int(notice) if isinstance(notice, (int, float)) else 45 evidence["notice_period_days"] = notice jd = get_jd() preferred = jd.notice_preferred_days if notice <= preferred: notice_sc = 1.00 elif notice <= 60: notice_sc = 0.95 elif notice <= 90: notice_sc = 0.85 else: notice_sc = 0.75 raw = 0.6 * notice_sc + 0.4 * (1.0 if s.get("open_to_work_flag") else 0.5) return max(0.50, min(1.10, raw)), evidence def interview_completion(c: dict) -> tuple[float, dict]: """G6-6: Interview follow-through rate.""" s = schema.signals(c) rate = s.get("interview_completion_rate") rate = float(rate) if isinstance(rate, (int, float)) else 0.7 return rate, {"interview_completion_rate": rate} def platform_trust(c: dict) -> tuple[float, dict]: """G6-7: Profile verification and completeness.""" s = schema.signals(c) completeness = s.get("profile_completeness_score") completeness = float(completeness) if isinstance(completeness, (int, float)) else 50.0 verification = sum([ bool(s.get("verified_email")), bool(s.get("verified_phone")), bool(s.get("linkedin_connected")), ]) / 3.0 interview_rate = s.get("interview_completion_rate") interview_rate = float(interview_rate) if isinstance(interview_rate, (int, float)) else 0.7 score = 0.40 * (completeness / 100.0) + 0.35 * verification + 0.25 * interview_rate return score, {"completeness": completeness} # =========================================================================== # G7: RESUME QUALITY # =========================================================================== def quantified_outcomes(c: dict, text: str | None = None) -> tuple[float, dict]: """G7-1: Number of quantified achievements.""" if text is None: text = schema.unified_text_blob(c) # Count numbers with context patterns = [ r'\d+(?:\.\d+)?%', r'\d+\s*ms', r'\d+\s*(?:million|billion)', r'\d+x\s+(?:improvement|speedup|increase)', r'\d+[\d,]*\s*(?:users|requests|queries|events)', ] count = 0 for p in patterns: count += len(re.findall(p, text, re.IGNORECASE)) score = min(1.0, count / 5.0) return score, {"count": count} def truthiness(c: dict, text: str | None = None) -> tuple[float, dict]: """G7-2: Cross-validation of skill claims vs career evidence.""" if text is None: text = schema.unified_text_blob(c) jd = get_jd() # Get skills listed in skills[] that are JD-relevant jd_skills = set() for d in [jd.required_skills, jd.preferred_skills]: for skills in d.values(): jd_skills.update(skills) listed_jd_skills = [] for s in schema.skills(c): name = (s.get("name") or "").lower() if any(jd_sk in name for jd_sk in jd_skills): listed_jd_skills.append(name) # Check how many are actually backed by career context backed = 0 unbacked = 0 for skill_name in listed_jd_skills: if skill_name in text: backed += 1 else: unbacked += 1 total = backed + unbacked if total == 0: return 0.5, {"backed": 0, "unbacked": 0} score = backed / total return score, {"backed": backed, "unbacked": unbacked} def keyword_stuffing_risk(c: dict) -> tuple[float, dict]: """G7-3: Probability of keyword stuffing.""" skills = schema.skills(c) if len(skills) < 12: return 0.0, {"risk": "low", "skill_count": len(skills)} # High skill count + most with zero duration and high proficiency zero_dur_expert = sum( 1 for s in skills if (s.get("duration_months") or 0) == 0 and s.get("proficiency") in ("advanced", "expert") ) risk = zero_dur_expert / len(skills) return min(1.0, risk), {"risk": "high" if risk > 0.3 else "medium" if risk > 0.1 else "low", "zero_dur_expert": zero_dur_expert} def profile_completeness(c: dict) -> tuple[float, dict]: """G7-4: Profile completeness score.""" s = schema.signals(c) score = s.get("profile_completeness_score") score = float(score) if isinstance(score, (int, float)) else 50.0 return score / 100.0, {"raw": score} # =========================================================================== # G8: SAFETY # =========================================================================== def disqualifier_penalty(c: dict, text: str | None = None) -> tuple[float, list[str]]: """G8-1: Multi-factor disqualifier penalty (multiplicative).""" if text is None: text = schema.unified_text_blob(c) jd = get_jd() penalty = 1.0 reasons = [] title = schema.current_title(c).lower() ch = schema.career_history(c) # Non-engineering title if any(bt in title for bt in jd.bad_title_patterns): ml_hits = _count_hits(text, list(jd.required_skills.keys())) context_hits = 0 for skills in jd.required_skills.values(): context_hits += len([s for s in skills if s in text]) if context_hits < 4: penalty *= 0.05 reasons.append("non_engineering_title") # Research-only title + zero production if any(h in title for h in jd.research_only_titles): prod_score, _ = production_strength(c, text) if prod_score == 0.0: penalty *= 0.15 reasons.append("research_only_no_production") # Entire career at consulting def _is_consulting(r: dict) -> bool: comp = (r.get("company") or "").lower() ind = (r.get("industry") or "").lower() return (any(f in comp for f in jd.consulting_firms) or any(k in ind for k in jd.consulting_industries)) current_ind = (schema.profile(c).get("current_industry") or "").lower() current_is_consulting = ( any(f in schema.current_company(c).lower() for f in jd.consulting_firms) or any(k in current_ind for k in jd.consulting_industries) ) role_flags = [_is_consulting(r) for r in ch] if len(ch) >= 2 and all(role_flags) and (not schema.current_company(c) or current_is_consulting): penalty *= 0.50 reasons.append("consulting_only_career") # Non-target domain without rescue if any(d in text for d in jd.non_target_domains): if not any(k in text for k in jd.non_target_rescue): penalty *= 0.40 reasons.append("non_nlp_domain") # Architect drift if any(h in title for h in jd.architect_titles): current_role = next((r for r in ch if r.get("is_current")), None) if current_role and (current_role.get("duration_months") or 0) >= 18: role_text = (current_role.get("description") or "").lower() if not _count_hits(role_text, jd.production_evidence + ["code", "coding", "implement"]): penalty *= 0.60 reasons.append("architect_no_recent_code") # Title-chaser if len(ch) >= 3: avg_tenure = sum((r.get("duration_months") or 0) for r in ch) / len(ch) if avg_tenure < 16: penalty *= 0.75 reasons.append("short_average_tenure") # Current services company penalty comp = schema.current_company(c).lower() comp_score = company_tier.company_quality_score(comp) if comp_score <= 0.35: penalty *= 0.65 reasons.append("current_services_company") return penalty, reasons # =========================================================================== # G9: LOCATION # =========================================================================== def location_score(c: dict) -> tuple[float, dict]: """G9-1: Location match to JD preferences.""" jd = get_jd() p = schema.profile(c) loc = (p.get("location") or "").lower() country = (p.get("country") or "").lower() relocate = bool(schema.signals(c).get("willing_to_relocate", False)) if any(city in loc for city in jd.preferred_locations): return 1.0, {"match": "preferred"} if any(city in loc for city in jd.welcome_locations): return 0.90, {"match": "welcome"} if "india" in country: return (0.85 if relocate else 0.75), {"match": "india"} return (0.50 if relocate else 0.20), {"match": "international"} # =========================================================================== # HELPERS # =========================================================================== def _count_hits(text: str, phrases: list[str]) -> list[str]: return [p for p in phrases if p in text] def title_relevance(c: dict) -> tuple[float, str]: """G1-5: Title match to JD (delegates to title_scoring).""" return title_scoring.title_relevance_score(schema.current_title(c)) def seniority_feature(c: dict) -> tuple[float, str]: """G1-6: Seniority level (delegates to title_scoring).""" return title_scoring.seniority_score(schema.current_title(c)) def company_quality_feature(c: dict) -> float: """G4-5: Current company quality (delegates to company_tier).""" return company_tier.company_quality_score(schema.current_company(c)) # =========================================================================== # V6.1 NEW FEATURES — winning differentiators # Each addresses a specific trap or signal the JD calls out but V6 misses. # =========================================================================== # Tier-5 signature: a senior engineer at a product company who built and owned # a real system, even if they don't use the JD's exact keywords. # The JD says: "A Tier 5 candidate may not use the words 'RAG' or 'Pinecone' # in their profile, but if their career history shows they built a recommendation # system at a product company, they're a fit." def tier5_signature(c: dict, text: str | None = None, features: dict | None = None) -> tuple[float, dict]: """V6.1-1: Tier-5 candidate signature — strong career evidence even without JD keywords.""" if text is None: text = schema.unified_text_blob(c) if features is None: features = {} title_rel = features.get("title_relevance", 0) company_q = features.get("company_quality", 0) career_depth = features.get("career_depth_ratio", 0) ownership = features.get("ownership_hierarchy", 0) production = features.get("production_strength", 0) skill_cov = features.get("skill_coverage", 0) yoe_band = features.get("yoe_band_score", 0) # Tier-5 conditions (all must be true) cond_title = title_rel >= 0.85 cond_company = company_q >= 0.80 cond_depth = career_depth >= 0.50 cond_ownership = ownership >= 0.65 cond_production = production >= 0.40 cond_yoe = yoe_band >= 0.55 # Tier-5 fires when 5+ of 6 conditions are met AND skill_coverage is moderate # (the whole point: Tier-5 candidates don't have perfect keyword coverage) conditions_met = sum([cond_title, cond_company, cond_depth, cond_ownership, cond_production, cond_yoe]) # Bonus when skill_coverage is low but other signals are strong — exactly # the "doesn't use the words RAG or Pinecone" pattern from the JD. if conditions_met >= 5 and skill_cov < 0.50: score = 1.0 # full Tier-5 signature sig_type = "pure_tier5" elif conditions_met >= 5: score = 0.85 sig_type = "tier5_with_keywords" elif conditions_met >= 4: score = 0.50 sig_type = "partial_tier5" else: score = 0.0 sig_type = "none" return score, {"type": sig_type, "conditions_met": conditions_met, "skill_coverage": skill_cov} # Behavioral twin trap: perfect-on-paper candidate who is behaviorally unavailable. # JD: "A perfect-on-paper candidate who hasn't logged in for 6 months and has a 5% # recruiter response rate is, for hiring purposes, not actually available." def behavioral_twin(c: dict, features: dict | None = None) -> tuple[float, dict]: """V6.1-2: Detect 'behavioral twin' — looks perfect on paper but is unavailable. Returns a PENALTY score in [0, 1] where 1.0 = no penalty, 0.0 = severe penalty. Multiply the final composite by this penalty. """ if features is None: features = {} s = schema.signals(c) days_active = features.get("recency", 0) # not days, but score last_active = schema.parse_date(s.get("last_active_date")) if last_active: actual_days = (REFERENCE_DATE - last_active).days else: actual_days = 180 rr = float(s.get("recruiter_response_rate") or 0) otw = bool(s.get("open_to_work_flag")) notice = int(s.get("notice_period_days") or 45) interview_rate = float(s.get("interview_completion_rate") or 0.7) offer_rate = s.get("offer_acceptance_rate") offer_rate = float(offer_rate) if isinstance(offer_rate, (int, float)) and offer_rate >= 0 else 0.5 # Build a penalty in [0, 1] (1 = no penalty, 0 = severe penalty) penalty = 1.0 reasons = [] # Stale activity (>180 days) if actual_days > 180: penalty *= 0.75 reasons.append(f"inactive_{actual_days}d") elif actual_days > 120: penalty *= 0.90 reasons.append(f"inactive_{actual_days}d") # Very low recruiter response rate if rr < 0.15: penalty *= 0.70 reasons.append(f"low_response_rate_{rr:.2f}") elif rr < 0.30: penalty *= 0.90 reasons.append(f"low_response_rate_{rr:.2f}") # Long notice period (>90 days, the JD says "bar gets higher") if notice > 120: penalty *= 0.80 reasons.append(f"long_notice_{notice}d") elif notice > 90: penalty *= 0.92 reasons.append(f"long_notice_{notice}d") # Low interview completion if interview_rate < 0.50: penalty *= 0.85 reasons.append(f"low_interview_completion_{interview_rate:.2f}") # Low offer acceptance (signal of being unserious) if 0 <= offer_rate < 0.30: penalty *= 0.88 reasons.append(f"low_offer_acceptance_{offer_rate:.2f}") return max(0.30, penalty), {"reasons": reasons, "days_inactive": actual_days, "response_rate": rr, "notice_days": notice} # LangChain-only recent AI experience — JD explicit disqualifier: # "If your 'AI experience' consists primarily of recent (under 12 months) projects # using LangChain to call OpenAI — we will probably not move forward" def langchain_only_recent(c: dict, text: str | None = None, features: dict | None = None) -> tuple[float, dict]: """V6.1-3: Detect LangChain-only recent AI experience. Returns a PENALTY multiplier in [0, 1] where 1 = no penalty. """ if text is None: text = schema.unified_text_blob(c) if features is None: features = {} ch = schema.career_history(c) pre_llm = features.get("pre_llm_months", 0) total_yoe = schema.years_of_experience(c) # Count LangChain-only signals across career langchain_kw = ["langchain", "llamaindex", "openai api", "chatgpt", "gpt-4", "gpt-3.5", "claude", "gemini", "anthropic"] production_ml_kw = ["production", "deployed", "shipped", "serving", "live traffic"] # Recent role (current or most recent) description current_role = next((r for r in ch if r.get("is_current")), ch[0] if ch else None) if not current_role: return 1.0, {"reasons": []} current_desc = (current_role.get("description") or "").lower() current_duration = current_role.get("duration_months") or 0 # Is recent role LangChain-heavy? has_langchain = any(kw in current_desc for kw in langchain_kw) has_production = any(kw in current_desc for kw in production_ml_kw) # Recent role < 12 months AND LangChain-heavy AND no pre-LLM experience if has_langchain and current_duration < 12 and pre_llm < 0.10 and total_yoe < 36: return 0.35, {"reasons": ["langchain_only_recent_no_pre_llm"], "duration": current_duration, "pre_llm": pre_llm} # LangChain-heavy but with some production evidence — partial penalty if has_langchain and not has_production and pre_llm < 0.20: return 0.65, {"reasons": ["langchain_no_production_evidence"], "duration": current_duration, "pre_llm": pre_llm} return 1.0, {"reasons": []} # Closed-source 5+ years without external validation — JD explicit disqualifier: # "People whose work has been entirely on closed-source proprietary systems for 5+ # years without external validation (papers, talks, open-source)." def closed_source_isolation(c: dict, text: str | None = None, features: dict | None = None) -> tuple[float, dict]: """V6.1-4: Detect closed-source isolation — 5+ years with no external validation. Returns a PENALTY multiplier in [0, 1]. """ if text is None: text = schema.unified_text_blob(c) if features is None: features = {} yoe = schema.years_of_experience(c) s = schema.signals(c) # External validation signals github = s.get("github_activity_score") github = float(github) if isinstance(github, (int, float)) and github >= 0 else -1 external_val_kw = ["open source", "open-source", "published", "paper", "conference talk", "blog post", "github.com", "patent", "arxiv", "workshop", "neurips", "icml", "iclr", "acl", "emnlp", "kdd", "www ", "sigir"] has_external = any(kw in text.lower() for kw in external_val_kw) # 5+ years experience AND no external validation AND no GitHub activity if yoe >= 5 and not has_external and github < 5: # Severity scales with YoE if yoe >= 10: return 0.55, {"reasons": ["closed_source_5yr_isolation_severe"], "yoe": yoe, "github": github} else: return 0.75, {"reasons": ["closed_source_5yr_isolation"], "yoe": yoe, "github": github} return 1.0, {"reasons": []} # Pre-LLM × Ownership interaction — rare and valuable. # The JD wants "people who understood retrieval and ranking before it became fashionable" # AND "shipped at least one end-to-end ranking system." This interaction captures both. def pre_llm_x_ownership(c: dict, features: dict | None = None) -> tuple[float, dict]: """V6.1-5: Pre-LLM IR experience × Ownership interaction.""" if features is None: features = {} pre_llm = features.get("pre_llm_months", 0) ownership = features.get("ownership_hierarchy", 0) # Both must be present if pre_llm < 0.20 or ownership < 0.50: return 0.0, {"pre_llm": pre_llm, "ownership": ownership} # Geometric mean — rewards having BOTH score = (pre_llm * ownership) ** 0.5 return min(1.0, score * 1.2), {"pre_llm": pre_llm, "ownership": ownership, "interaction": score} # Salary compatibility — V6 missed this signal entirely. # JD says "Notice period: sub-30-day preferred. 30+ day notice candidates are still # in scope but the bar gets higher." Salary mismatch is an implicit disqualifier. def salary_compatibility(c: dict) -> tuple[float, dict]: """V6.1-6: Salary expectations vs typical Senior AI Engineer range in India. Senior AI Engineer in India (Series A): 40-80 LPA typical. Below 25 LPA suggests junior-level expectations (mismatch with senior role). Above 120 LPA suggests they're at a level beyond this role. """ s = schema.signals(c) sal = s.get("expected_salary_range_inr_lpa") or {} sal_min = float(sal.get("min", 0) or 0) sal_max = float(sal.get("max", 0) or 0) if sal_min == 0 and sal_max == 0: return 0.70, {"reason": "no_salary_data"} # Use the midpoint mid = (sal_min + sal_max) / 2 # Sweet spot: 40-90 LPA if 40 <= mid <= 90: return 1.0, {"mid": mid, "reason": "sweet_spot"} if 30 <= mid < 40: return 0.85, {"mid": mid, "reason": "slightly_low"} if 90 < mid <= 110: return 0.85, {"mid": mid, "reason": "slightly_high"} if 25 <= mid < 30: return 0.65, {"mid": mid, "reason": "low_expectations"} if 110 < mid <= 150: return 0.70, {"mid": mid, "reason": "high_expectations"} if mid < 25: return 0.40, {"mid": mid, "reason": "junior_level_salary"} if mid > 150: return 0.50, {"mid": mid, "reason": "overqualified_salary"} return 0.70, {"mid": mid, "reason": "default"}