#!/usr/bin/env python3 """ precompute.py — V7 Winning-Level Intelligent Recruitment Engine Offline preprocessing. Extracts ~55 JD-driven features: - V5's ~40 base features - V6's 8 interaction/narrative features - V6.1's 6 winning-differentiator features (tier5_signature, behavioral_twin, langchain_only, closed_source_isolation, pre_llm_x_ownership, salary_compatibility) - V7's 4 new features (assessment_signal, endorsement_signal, education_tier, cross_validation_signal) Builds canonical candidate profiles, fits TF-IDF+SVD embeddings with query expansion, computes adaptive RRF retrieval scores, and writes a Parquet artifact. V7 pipeline: JD -> Hiring Intent -> Domain Builder -> Query Expansion Candidates -> Canonical Profile -> Feature Registry -> Interaction Features -> V6.1 Features -> V7 Features -> All features + embeddings -> Parquet """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path import numpy as np import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent)) from lib import schema, features, honeypot, embeddings as emb_mod from lib.jd_parser import get_jd from lib.hiring_intent import get_intent from lib.query_expansion import get_expanded_text def load_candidates(path: str): """Load candidates from JSONL (one JSON object per line).""" import gzip as _gzip if path.endswith(".gz"): f = _gzip.open(path, "rt", encoding="utf-8") else: f = open(path, "r", encoding="utf-8") try: for line in f: line = line.strip() if not line: continue if line.startswith("candidate_id,") or line.startswith("#"): continue try: yield json.loads(line) except json.JSONDecodeError: continue finally: f.close() def extract_features_v7(c: dict) -> dict: """Extract all V7 features for one candidate (~55 features).""" text = schema.unified_text_blob(c) jd = get_jd() # === V5 base features (all ~40) === skill_cov, _ = features.skill_coverage(c, text) pref_cov, _ = features.preferred_coverage(c, text) domain_spec, _ = features.domain_specialization(c, text) skill_trust, _ = features.skill_trust_avg(c) title_rel, _ = features.title_relevance(c) seniority, _ = features.seniority_feature(c) jd_skill_cnt, _ = features.jd_skill_count(c, text) own_hier, _ = features.ownership_hierarchy(c, text) imp_mag, _ = features.impact_magnitude(c, text) imp_sig, _ = features.impact_signals(c, text) ev_str, _ = features.evidence_strength(c) prod_str, prod_hits = features.production_strength(c, text) prod_div, _ = features.production_diversity(c, text) scale_ev, _ = features.scale_evidence(c, text) yoe_band, _ = features.yoe_band_score(c, jd) depth_ratio, _ = features.career_depth_ratio(c, jd) pre_llm, _ = features.pre_llm_months(c, jd) career_traj, _ = features.career_trajectory(c) comp_qual = features.company_quality_feature(c) comp_qual_avg, _ = features.company_quality_avg(c) career_stab, _ = features.career_stability(c) promo_vel, _ = features.promotion_velocity(c) ret_depth, _ = features.retrieval_depth(c, text) eval_exp, _ = features.evaluation_experience(c, text) sys_design, _ = features.system_design_evidence(c, text) recency_sc, recency_ev = features.recency(c) resp_sc, resp_ev = features.responsiveness(c) market_sc, _ = features.market_demand(c) github_sc, _ = features.github_activity(c) avail_sc, avail_ev = features.availability_score(c) interview_sc, _ = features.interview_completion(c) trust_sc, _ = features.platform_trust(c) quant_out, _ = features.quantified_outcomes(c, text) truth_sc, _ = features.truthiness(c, text) kw_risk, _ = features.keyword_stuffing_risk(c) prof_comp, _ = features.profile_completeness(c) disq_pen, disq_reasons = features.disqualifier_penalty(c, text) is_hp, hp_reasons = honeypot.is_honeypot(c) loc_sc, _ = features.location_score(c) # === V6 Career narrative === from lib.career_narrative import analyze as analyze_narrative narrative = analyze_narrative(c) # === V6 Interaction features === ownership_x_prod = own_hier * prod_str skill_x_yoe = skill_cov * (0.5 + 0.5 * yoe_band) impact_x_domain = imp_mag * (0.3 + 0.7 * depth_ratio) trajectory_x_company = career_traj * comp_qual_avg # === V6.1 NEW FEATURES === # Build a features dict for the V6.1 feature functions that depend on other features base_features = { "title_relevance": title_rel, "company_quality": comp_qual, "career_depth_ratio": depth_ratio, "ownership_hierarchy": own_hier, "production_strength": prod_str, "skill_coverage": skill_cov, "yoe_band_score": yoe_band, "pre_llm_months": pre_llm, } tier5_sig, tier5_ev = features.tier5_signature(c, text, base_features) beh_twin_pen, beh_twin_ev = features.behavioral_twin(c, base_features) lc_pen, lc_ev = features.langchain_only_recent(c, text, base_features) cs_pen, cs_ev = features.closed_source_isolation(c, text, base_features) pllm_own, pllm_own_ev = features.pre_llm_x_ownership(c, base_features) sal_compat, sal_ev = features.salary_compatibility(c) # === V7 NEW FEATURES === # V7-1: Assessment signal — platform skill assessment scores assessment_signal, assessment_ev = _v7_assessment_signal(c) # V7-2: Endorsement signal — volume and quality of endorsements endorsement_signal, endorsement_ev = _v7_endorsement_signal(c) # V7-3: Education tier — IIT/NIT/IIIT boost edu_tier, edu_ev = _v7_education_tier(c) # V7-4: Cross-validation signal — multiple evidence types agreeing cross_val, cross_val_ev = _v7_cross_validation(c, text, base_features) return { "candidate_id": c["candidate_id"], # G1: JD Fit "skill_coverage": skill_cov, "preferred_coverage": pref_cov, "domain_specialization": domain_spec, "skill_trust_avg": skill_trust, "title_relevance": title_rel, "seniority": seniority, "jd_skill_count": jd_skill_cnt, # G2: Impact & Ownership "ownership_hierarchy": own_hier, "impact_magnitude": imp_mag, "impact_signals": imp_sig, "evidence_strength": ev_str, # G3: Production & Scale "production_strength": prod_str, "production_diversity": prod_div, "scale_evidence": scale_ev, # G4: Experience & Career "yoe_band_score": yoe_band, "career_depth_ratio": depth_ratio, "pre_llm_months": pre_llm, "career_trajectory": career_traj, "company_quality": comp_qual, "company_quality_avg": comp_qual_avg, "career_stability": career_stab, "promotion_velocity": promo_vel, # G5: Retrieval & Evaluation "retrieval_depth": ret_depth, "evaluation_experience": eval_exp, "system_design_evidence": sys_design, # G6: Behavioural "recency": recency_sc, "responsiveness": resp_sc, "market_demand": market_sc, "github_activity": github_sc, "availability_score": avail_sc, "interview_completion": interview_sc, "platform_trust": trust_sc, # G7: Resume Quality "quantified_outcomes": quant_out, "truthiness": truth_sc, "keyword_stuffing_risk": kw_risk, "profile_completeness": prof_comp, # G8: Safety "disqualifier_penalty": disq_pen, "is_honeypot": is_hp, # G9: Location "location_score": loc_sc, # G10: Career Narrative (V6) "career_coherence": narrative.coherence, "narrative_type": narrative.trajectory_type, "narrative_suspicious": json.dumps(narrative.suspicious_patterns), # G11: Interaction Features (V6) "ownership_x_production": ownership_x_prod, "skill_x_yoe": skill_x_yoe, "impact_x_domain": impact_x_domain, "trajectory_x_company": trajectory_x_company, # G12: V6.1 Winning Features "tier5_signature": tier5_sig, "behavioral_twin_penalty": beh_twin_pen, "langchain_only_penalty": lc_pen, "closed_source_penalty": cs_pen, "pre_llm_x_ownership": pllm_own, "salary_compatibility": sal_compat, # G13: V7 New Features "assessment_signal": assessment_signal, "endorsement_signal": endorsement_signal, "education_tier": edu_tier, "cross_validation": cross_val, # Metadata for reasoning "current_title": schema.current_title(c), "current_company": schema.current_company(c), "years_of_experience": schema.years_of_experience(c), # Evidence JSON for reasoning "_candidate_json": json.dumps(c, default=str), # Disq reasons for reasoning "_disq_reasons": json.dumps(disq_reasons), # Behavioural evidence for reasoning "_behaviour_evidence": json.dumps({ **recency_ev, **resp_ev, **avail_ev, "notice_period_days": avail_ev.get("notice_period_days", 45), "days_since_active": recency_ev.get("days_since_active", 90), "recruiter_response_rate": resp_ev.get("recruiter_response_rate", 0.3), }), # V6.1 evidence columns for reasoning "_beh_twin_evidence": json.dumps(beh_twin_ev), "_langchain_evidence": json.dumps(lc_ev), "_closed_source_evidence": json.dumps(cs_ev), "_salary_evidence": json.dumps(sal_ev), "_tier5_evidence": json.dumps(tier5_ev), "_assessment_evidence": json.dumps(assessment_ev), "_endorsement_evidence": json.dumps(endorsement_ev), "_education_evidence": json.dumps(edu_ev), } # =========================================================================== # V7 NEW FEATURE FUNCTIONS # =========================================================================== def _v7_assessment_signal(c: dict) -> tuple[float, dict]: """V7-1: Platform skill assessment scores — validated skill proficiency. The platform has skill_assessment_scores which are actual test results, not self-reported. This is a much stronger signal than listed skills. """ s = schema.signals(c) assessments = s.get("skill_assessment_scores") or {} if not assessments: return 0.50, {"reason": "no_assessments", "count": 0} # Only count ML-relevant assessments relevant_keys = {"python", "machine learning", "nlp", "natural language", "deep learning", "statistics", "pytorch", "tensorflow", "retrieval", "ranking", "recommendation", "data science", "computer vision", "reinforcement learning"} relevant_scores = [] for k, v in assessments.items(): if k.lower() in relevant_keys: relevant_scores.append(float(v) if isinstance(v, (int, float)) else 0) if not relevant_scores: return 0.50, {"reason": "no_relevant_assessments", "count": len(assessments)} avg_score = sum(relevant_scores) / len(relevant_scores) # Normalize: >70 is strong, >50 is decent, <30 is concerning if avg_score >= 70: signal = 0.80 + 0.20 * min(1, (avg_score - 70) / 30) elif avg_score >= 50: signal = 0.60 + 0.20 * (avg_score - 50) / 20 elif avg_score >= 30: signal = 0.40 + 0.20 * (avg_score - 30) / 20 else: signal = 0.20 + 0.20 * avg_score / 30 # Bonus for taking multiple assessments (shows engagement) count_bonus = min(0.10, len(relevant_scores) * 0.025) signal = min(1.0, signal + count_bonus) return signal, {"avg_score": avg_score, "count": len(relevant_scores), "reason": "assessed"} def _v7_endorsement_signal(c: dict) -> tuple[float, dict]: """V7-2: Endorsement volume and quality — peer-validated skills. High endorsement counts on relevant skills indicate genuine expertise, not just keyword stuffing. The JD values 'proven' skills. """ skills = schema.skills(c) if not skills: return 0.30, {"reason": "no_skills", "total_endorsements": 0} # Get domain taxonomy for relevant skills try: taxonomy = __import__("lib.domain", fromlist=["get_taxonomy"]).get_taxonomy() relevant_skill_names = set(taxonomy.skill_tier.keys()) except Exception: relevant_skill_names = { "python", "machine learning", "nlp", "deep learning", "pytorch", "tensorflow", "elasticsearch", "faiss", "pinecone", "rag", "embeddings", "retrieval", "ranking", "recommendation", "ann", "hnsw", "vector database", "semantic search", "bm25", "learning to rank", "numpy", "pandas", "scikit-learn", } total_endorsements = 0 relevant_endorsements = 0 relevant_count = 0 max_endorsement = 0 for sk in skills: name = (sk.get("name") or "").lower() endorsements = int(sk.get("endorsements") or 0) proficiency = (sk.get("proficiency") or "").lower() total_endorsements += endorsements if name in relevant_skill_names: relevant_endorsements += endorsements relevant_count += 1 max_endorsement = max(max_endorsement, endorsements) elif any(partial in name for partial in ["embed", "retriev", "rank", "search", "vector", "nlp", "ml ", "machine learn", "deep learn", "pytorch", "tensor", "faiss", "pinecone", "ann", "rag", "bm25", "elasticsearch"]): relevant_endorsements += endorsements relevant_count += 1 max_endorsement = max(max_endorsement, endorsements) if relevant_count == 0: # No relevant skills endorsed at all return 0.30, {"reason": "no_relevant_skills", "total_endorsements": total_endorsements} avg_rel = relevant_endorsements / relevant_count # Signal based on endorsement volume if avg_rel >= 30 and relevant_count >= 3: signal = 0.85 + 0.15 * min(1, (avg_rel - 30) / 50) elif avg_rel >= 15 and relevant_count >= 2: signal = 0.65 + 0.20 * (avg_rel - 15) / 15 elif avg_rel >= 5: signal = 0.45 + 0.20 * (avg_rel - 5) / 10 else: signal = 0.25 + 0.20 * avg_rel / 5 return min(1.0, signal), {"avg_relevant": avg_rel, "relevant_count": relevant_count, "max_endorsement": max_endorsement, "total_endorsements": total_endorsements} def _v7_education_tier(c: dict) -> tuple[float, dict]: """V7-3: Education tier — IIT/NIT/IIIT boost for Indian candidates. Stage 4 reviewers may consider educational pedigree. This is a mild signal (not a primary ranking factor) that breaks ties. """ education = schema.education(c) if not education: return 0.50, {"reason": "no_education"} # Get highest degree best_tier = 3 # default tier_3 best_degree = None best_institution = None tier_map = {"tier_1": 1, "tier_2": 2, "tier_3": 3, "tier_4": 4} for edu in education: tier_str = (edu.get("tier") or "tier_3").lower() tier_num = tier_map.get(tier_str, 3) if tier_num < best_tier: best_tier = tier_num best_degree = edu.get("degree", "") best_institution = edu.get("institution", "") # Tier-1 (IITs, top institutions): 0.80 # Tier-2 (NITs, IIITs, good colleges): 0.70 # Tier-3 (other): 0.50 # Tier-4 or unknown: 0.40 tier_scores = {1: 0.80, 2: 0.70, 3: 0.50, 4: 0.40} score = tier_scores.get(best_tier, 0.50) # Bonus for CS/AI/ML field of study best_field = "" for edu in education: field = (edu.get("field_of_study") or "").lower() if any(kw in field for kw in ["computer science", "artificial intelligence", "machine learning", "data science", "information technology"]): best_field = edu.get("field_of_study", "") score = min(1.0, score + 0.10) break return score, {"tier": best_tier, "degree": best_degree, "institution": best_institution, "field": best_field} def _v7_cross_validation(c: dict, text: str, base_features: dict) -> tuple[float, dict]: """V7-4: Cross-validation — when multiple evidence types agree on quality. A candidate who has high impact AND high ownership AND high production AND high retrieval depth is much more likely to be genuinely strong than someone who scores high on just one dimension. """ imp = base_features.get("impact_magnitude", 0) own = base_features.get("ownership_hierarchy", 0) prod = base_features.get("production_strength", 0) depth = base_features.get("career_depth_ratio", 0) eval_exp = base_features.get("evaluation_experience", 0) if "evaluation_experience" in base_features else 0 # Count how many dimensions are "strong" (>= 0.50) strong_dims = sum(1 for v in [imp, own, prod, depth, eval_exp] if v >= 0.50) # Score based on cross-validation if strong_dims >= 5: signal = 0.95 elif strong_dims >= 4: signal = 0.80 elif strong_dims >= 3: signal = 0.60 elif strong_dims >= 2: signal = 0.40 else: signal = 0.20 # Also check for "no weak dimensions" bonus no_weak = all(v >= 0.20 for v in [imp, own, prod, depth]) if no_weak and strong_dims >= 3: signal = min(1.0, signal + 0.10) return signal, {"strong_dimensions": strong_dims, "dimensions": {"impact": imp, "ownership": own, "production": prod, "depth": depth, "evaluation": eval_exp}} def main(): ap = argparse.ArgumentParser() ap.add_argument("--candidates", required=True) ap.add_argument("--out", default="./artifacts/features.parquet") args = ap.parse_args() t0 = time.time() candidates = list(load_candidates(args.candidates)) print(f"[precompute] loaded {len(candidates)} candidates in {time.time()-t0:.1f}s") # Print hiring intent intent = get_intent() print(f"[precompute] Hiring Intent: philosophy={intent.philosophy}, " f"need={intent.primary_need}, ownership={intent.ownership_expectation:.2f}, " f"culture={intent.shipping_culture}, team={intent.team_context}") # Feature extraction rows = [] texts = [] t1 = time.time() for i, c in enumerate(candidates): row = extract_features_v7(c) rows.append(row) texts.append(schema.unified_text_blob(c)) if (i + 1) % 20000 == 0: print(f"[precompute] processed {i+1}/{len(candidates)} candidates " f"({time.time()-t1:.1f}s)") print(f"[precompute] extracted V7 features for {len(rows)} candidates in {time.time()-t1:.1f}s") # TF-IDF + SVD embeddings (with query expansion) t2 = time.time() jd = get_jd() expanded_ideal = get_expanded_text(jd.ideal_text) all_texts = texts + [expanded_ideal] embedder = emb_mod.TfidfSvdEmbedder(n_components=min(100, len(all_texts) - 1)) embedder.fit(all_texts) doc_emb = embedder.transform(all_texts)[:len(texts)] sims = embedder.similarity_to_query(doc_emb, expanded_ideal) for row, sim in zip(rows, sims): row["embedding_sim"] = float(sim) print(f"[precompute] TF-IDF+SVD embeddings (expanded query) in {time.time()-t2:.1f}s") df = pd.DataFrame(rows) out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) df.to_parquet(out_path, index=False) print(f"[precompute] wrote {len(df)} rows -> {out_path} ({out_path.stat().st_size/1e6:.1f} MB)") print(f"[precompute] total time: {time.time()-t0:.1f}s") print(f"[precompute] honeypots flagged: {int(df['is_honeypot'].sum())} / {len(df)}") print(f"[precompute] V7 features: {len(df.columns)} columns") # Feature statistics for col in ["tier5_signature", "behavioral_twin_penalty", "langchain_only_penalty", "closed_source_penalty", "pre_llm_x_ownership", "salary_compatibility", "assessment_signal", "endorsement_signal", "education_tier", "cross_validation"]: if col in df.columns: vals = df[col].dropna() print(f" {col}: mean={vals.mean():.3f} " f"p50={vals.median():.3f} p90={vals.quantile(0.90):.3f}") if __name__ == "__main__": main()