Upload 11 files

src/.gitkeep

@@ -0,0 +1 @@
+# Keep directory

src/availability.py

@@ -0,0 +1,132 @@
+import datetime
+from dateutil import parser
+
+def compute_multipliers(candidate: dict, jd: dict) -> dict[str, float]:
+    """Computes the availability and location multipliers for a candidate, supporting nested schemas."""
+    signals = candidate.get("redrob_signals") or {}
+    profile = candidate.get("profile") or {}
+    
+    # --- Availability Multiplier ---
+    availability_mult = 1.0
+    
+    # open_to_work_flag
+    open_to_work = signals.get("open_to_work_flag")
+    if open_to_work is None:
+        open_to_work = candidate.get("open_to_work_flag")
+    if open_to_work is True:
+        availability_mult += 0.10
+        
+    # last_active_date
+    last_active_str = signals.get("last_active_date") or candidate.get("last_active_date")
+    days_inactive = None
+    if last_active_str:
+        try:
+            last_active = parser.parse(last_active_str)
+            if last_active.tzinfo is not None:
+                today = datetime.datetime.now(datetime.timezone.utc)
+            else:
+                today = datetime.datetime.now()
+            days_inactive = (today - last_active).days
+        except Exception:
+            pass
+            
+    if days_inactive is not None:
+        if days_inactive <= 14:
+            availability_mult += 0.10
+            if days_inactive <= 7:
+                availability_mult += 0.05  # stacks
+        if days_inactive > 90:
+            availability_mult -= 0.25
+            
+    # recruiter_response_rate
+    response_rate = signals.get("recruiter_response_rate")
+    if response_rate is None:
+        response_rate = candidate.get("recruiter_response_rate", 0.0)
+    if float(response_rate) >= 0.70:
+        availability_mult += 0.05
+        
+    # offer_acceptance_rate
+    acceptance_rate = signals.get("offer_acceptance_rate")
+    if acceptance_rate is None:
+        acceptance_rate = candidate.get("offer_acceptance_rate")
+    if acceptance_rate is not None and acceptance_rate != -1:
+        if float(acceptance_rate) >= 0.80:
+            availability_mult += 0.05
+            
+    # avg_response_time_hours
+    avg_resp_time = signals.get("avg_response_time_hours")
+    if avg_resp_time is None:
+        avg_resp_time = candidate.get("avg_response_time_hours")
+    if avg_resp_time is not None and float(avg_resp_time) > 72:
+        availability_mult -= 0.05
+        
+    # notice_period_days
+    notice_period = signals.get("notice_period_days")
+    if notice_period is None:
+        notice_period = candidate.get("notice_period_days")
+    if notice_period is not None and int(notice_period) > 90:
+        availability_mult -= 0.10
+        
+    # interview_completion_rate
+    completion_rate = signals.get("interview_completion_rate")
+    if completion_rate is None:
+        completion_rate = candidate.get("interview_completion_rate", 0.0)
+    if float(completion_rate) < 0.50:
+        availability_mult -= 0.15
+        
+    # expected_salary_range_inr_lpa.min
+    salary_range = signals.get("expected_salary_range_inr_lpa") or candidate.get("expected_salary_range_inr_lpa") or {}
+    salary_min = 0.0
+    if isinstance(salary_range, dict):
+        salary_min = salary_range.get("min") or 0.0
+    elif isinstance(salary_range, (int, float)):
+        salary_min = salary_range
+        
+    budget_max = jd.get("budget_max_inr_lpa") or 40
+    if salary_min > budget_max:
+        availability_mult -= 0.20
+        
+    # Clamp availability_mult to [0.50, 1.25]
+    availability_mult = max(0.50, min(availability_mult, 1.25))
+    
+    # --- Location Multiplier ---
+    location_mult = 1.0
+    
+    cand_loc = str(profile.get("location") or candidate.get("location") or "").lower().strip()
+    jd_locs = {str(loc).lower().strip() for loc in (jd.get("preferred_locations") or []) if loc}
+    
+    if cand_loc in jd_locs:
+        location_mult += 0.05
+    else:
+        willing_to_relocate = signals.get("willing_to_relocate")
+        if willing_to_relocate is None:
+            willing_to_relocate = candidate.get("willing_to_relocate", True)
+        if willing_to_relocate is False:
+            location_mult -= 0.05
+            
+    # Clamp location_mult to [0.70, 1.05]
+    location_mult = max(0.70, min(location_mult, 1.05))
+    
+    return {
+        "availability_mult": round(availability_mult, 4),
+        "location_mult": round(location_mult, 4)
+    }
+
+def apply_multipliers(scored_results: list[dict], jd: dict) -> list[dict]:
+    """Applies availability and location multipliers to the scored results."""
+    updated_results = []
+    for res in scored_results:
+        mults = compute_multipliers(res["candidate"], jd)
+        av_mult = mults["availability_mult"]
+        loc_mult = mults["location_mult"]
+        
+        raw_score = res["raw_score"]
+        final_score = round(min(raw_score * av_mult * loc_mult, 1.0), 4)
+        
+        updated_res = res.copy()
+        updated_res["availability_mult"] = av_mult
+        updated_res["location_mult"] = loc_mult
+        updated_res["final_score"] = final_score
+        updated_results.append(updated_res)
+        
+    return updated_results

src/hard_filter.py

@@ -0,0 +1,132 @@
+import datetime
+from dateutil import parser
+from sklearn.metrics.pairwise import cosine_similarity
+
+def is_killed(candidate: dict, jd: dict, tfidf) -> tuple[bool, str]:
+    """Applies the 8 hard filtering rules in order, short-circuiting on first match.
+    
+    Supports both nested (actual dataset) and flat (test fixture) candidate structures.
+    """
+    signals = candidate.get("redrob_signals") or {}
+    profile = candidate.get("profile") or {}
+    
+    # 1. Profile completeness score < 30
+    profile_completeness = signals.get("profile_completeness_score") or candidate.get("profile_completeness_score", 0)
+    if profile_completeness < 30:
+        return True, f"profile_completeness_score {profile_completeness} < 30"
+        
+    # 2. Verified email == False
+    verified_email = signals.get("verified_email")
+    if verified_email is None:
+        verified_email = candidate.get("verified_email", False)
+    if verified_email is False:
+        return True, "verified_email == False"
+        
+    # 3. Interview completion rate < 0.20
+    interview_completion_rate = signals.get("interview_completion_rate")
+    if interview_completion_rate is None:
+        interview_completion_rate = candidate.get("interview_completion_rate", 0.0)
+    if interview_completion_rate < 0.20:
+        return True, f"interview_completion_rate {interview_completion_rate} < 0.20"
+        
+    # 4. (today - last_active_date).days > 180
+    last_active_date_str = signals.get("last_active_date") or candidate.get("last_active_date")
+    if not last_active_date_str:
+        return True, "Missing last_active_date"
+    try:
+        last_active_date = parser.parse(last_active_date_str)
+        if last_active_date.tzinfo is not None:
+            today = datetime.datetime.now(datetime.timezone.utc)
+        else:
+            today = datetime.datetime.now()
+        days_inactive = (today - last_active_date).days
+        if days_inactive > 180:
+            return True, f"Inactive for {days_inactive} days (> 180 days)"
+    except Exception as e:
+        return True, f"Failed to parse last_active_date: {str(e)}"
+        
+    # 5. Open to work flag == False
+    open_to_work = signals.get("open_to_work_flag")
+    if open_to_work is None:
+        open_to_work = candidate.get("open_to_work_flag", False)
+    if open_to_work is False:
+        return True, "open_to_work_flag == False"
+        
+    # 6. Zero overlap between candidate's industries and jd.target_industries using Python set intersection
+    cand_industries = set()
+    # Check profile industry
+    prof_ind = profile.get("current_industry")
+    if prof_ind:
+        cand_industries.add(prof_ind.strip().lower())
+    # Check career history industries
+    history = candidate.get("career_history") or candidate.get("experience") or candidate.get("work_experience") or []
+    for job in history:
+        if isinstance(job, dict) and job.get("industry"):
+            cand_industries.add(job["industry"].strip().lower())
+    # Fallback to top-level industries list if present (like in mock fixtures)
+    for ind in (candidate.get("industries") or []):
+        if ind:
+            cand_industries.add(ind.strip().lower())
+            
+    jd_industries = {ind.strip().lower() for ind in (jd.get("target_industries") or []) if ind}
+    if not cand_industries.intersection(jd_industries):
+        return True, f"Zero industry overlap (Candidate: {cand_industries}, JD: {jd_industries})"
+        
+    # 7. TF-IDF cosine similarity between current_title and jd.title < 0.05
+    cand_title = profile.get("current_title") or candidate.get("current_title", "")
+    jd_title = jd.get("title", "")
+    if not cand_title or not jd_title:
+        sim = 0.0
+    else:
+        try:
+            cand_tfidf = tfidf.transform([cand_title])
+            jd_tfidf = tfidf.transform([jd_title])
+            sim = cosine_similarity(cand_tfidf, jd_tfidf)[0][0]
+        except Exception:
+            sim = 0.0
+    if sim < 0.05:
+        return True, f"Title similarity {sim:.4f} < 0.05 (Candidate: '{cand_title}', JD: '{jd_title}')"
+        
+    # 8. Honeypot check: github_activity_score == -1 AND empty skill_assessment_scores AND endorsements_received == 0 AND connection_count < 5
+    github_score = signals.get("github_activity_score")
+    if github_score is None:
+        github_score = candidate.get("github_activity_score", 0)
+        
+    skill_assessments = signals.get("skill_assessment_scores")
+    if skill_assessments is None:
+        skill_assessments = candidate.get("skill_assessment_scores") or {}
+        
+    endorsements = signals.get("endorsements_received")
+    if endorsements is None:
+        endorsements = candidate.get("endorsements_received", 0)
+        
+    connections = signals.get("connection_count")
+    if connections is None:
+        connections = candidate.get("connection_count", 0)
+    
+    if (github_score == -1 and 
+        not skill_assessments and 
+        endorsements == 0 and 
+        connections < 5):
+        return True, f"Honeypot detected (GitHub: {github_score}, Assessments: {skill_assessments}, Endorsements: {endorsements}, Connections: {connections})"
+        
+    return False, ""
+
+def apply_hard_filter(candidates: list[dict], jd: dict, tfidf) -> tuple[list[dict], list[dict]]:
+    """Applies the hard filter rules to a list of candidates.
+    
+    Returns (survivors_list, killed_list).
+    """
+    survivors = []
+    killed = []
+    for cand in candidates:
+        killed_flag, reason = is_killed(cand, jd, tfidf)
+        if killed_flag:
+            killed.append({
+                "candidate_id": cand.get("candidate_id") or cand.get("id"),
+                "candidate": cand,
+                "reason": reason
+            })
+        else:
+            survivors.append(cand)
+    return survivors, killed

src/output.py

@@ -0,0 +1,111 @@
+import polars as pl
+
+def generate_reasoning(result: dict, jd: dict) -> str:
+    """Generates a structured reasoning string for a candidate's fit, supporting nested schemas."""
+    candidate = result["candidate"]
+    profile = candidate.get("profile") or {}
+    signals = candidate.get("redrob_signals") or {}
+    
+    name = profile.get("anonymized_name") or profile.get("name") or candidate.get("name") or "Unknown"
+    yoe = profile.get("years_of_experience") or profile.get("yoe") or candidate.get("years_of_experience") or candidate.get("yoe") or 0
+    
+    # Get current company/title
+    history = candidate.get("career_history") or candidate.get("experience") or candidate.get("work_experience") or []
+    company = profile.get("current_company") or "Unknown"
+    title = profile.get("current_title") or candidate.get("current_title") or ""
+    
+    if history and isinstance(history, list) and isinstance(history[0], dict):
+        if not title:
+            title = history[0].get("title") or ""
+        if company == "Unknown":
+            company = history[0].get("company") or history[0].get("company_name") or "Unknown"
+            
+    if not title:
+        title = "Candidate"
+        
+    must_have_coverage = result.get("must_have_coverage", 0.0)
+    
+    # Extract top 3 skill names
+    cand_skills = candidate.get("skills") or []
+    def skill_key(s):
+        if isinstance(s, dict):
+            return (s.get("endorsements") or 0) + (s.get("duration_months") or 0)
+        return 0
+    sorted_skills = sorted(cand_skills, key=skill_key, reverse=True)
+    skill_names = []
+    for s in sorted_skills[:3]:
+        if isinstance(s, dict):
+            skill_names.append(s.get("name", ""))
+        elif isinstance(s, str):
+            skill_names.append(s)
+    top_skills_str = ", ".join(filter(None, skill_names)) or "None"
+    
+    open_to_work = bool(signals.get("open_to_work_flag") if signals.get("open_to_work_flag") is not None else candidate.get("open_to_work_flag", False))
+    notice_period_days = signals.get("notice_period_days") if signals.get("notice_period_days") is not None else candidate.get("notice_period_days", 0)
+    github_activity_score = signals.get("github_activity_score") if signals.get("github_activity_score") is not None else candidate.get("github_activity_score", 0)
+    
+    final_score = result["final_score"]
+    A = result["A"]
+    B = result["B"]
+    C = result["C"]
+    
+    return (
+        f"{name} | {yoe}y exp | {title} @ {company} | "
+        f"Skill match: {must_have_coverage:.0%} must-haves covered | "
+        f"Top skills: {top_skills_str} | Open to work: {open_to_work} | "
+        f"Notice: {notice_period_days}d | GitHub: {github_activity_score} | "
+        f"Score: {final_score:.4f} (A={A:.3f} B={B:.3f} C={C:.3f})"
+    )
+
+def write_submission(results: list[dict], jd: dict, out_path: str = 'submission.csv') -> pl.DataFrame:
+    """Converts results to a Polars DataFrame, ranks the top 100, and writes them to a CSV."""
+    rows = []
+    for res in results:
+        cand = res["candidate"]
+        signals = cand.get("redrob_signals") or {}
+        reasoning = generate_reasoning(res, jd)
+        
+        rows.append({
+            "candidate_id": res["candidate_id"],
+            "final_score": res["final_score"],
+            "raw_score": res["raw_score"],
+            "availability_mult": res["availability_mult"],
+            "location_mult": res["location_mult"],
+            "reasoning": reasoning,
+            "profile_completeness_score": signals.get("profile_completeness_score") or cand.get("profile_completeness_score", 0),
+            "saved_by_recruiters_30d": signals.get("saved_by_recruiters_30d") or cand.get("saved_by_recruiters_30d", 0),
+            "component_scores": f"A={res['A']:.3f}, B={res['B']:.3f}, C={res['C']:.3f}"
+        })
+        
+    df = pl.DataFrame(rows)
+    
+    # Sort by final_score, then profile_completeness_score, then saved_by_recruiters_30d (all descending)
+    df_sorted = df.sort(
+        by=["final_score", "profile_completeness_score", "saved_by_recruiters_30d"],
+        descending=[True, True, True]
+    )
+    
+    # Take top 100
+    df_top100 = df_sorted.head(100)
+    
+    # Add rank column (1-indexed)
+    df_top100 = df_top100.with_columns(
+        pl.int_range(1, df_top100.height + 1).alias("rank")
+    )
+    
+    # Select and reorder
+    df_final = df_top100.select([
+        "rank",
+        "candidate_id",
+        "final_score",
+        "raw_score",
+        "component_scores",
+        "availability_mult",
+        "location_mult",
+        "reasoning"
+    ])
+    
+    df_final.write_csv(out_path)
+    print(f"Successfully wrote top {df_final.height} candidates to {out_path}")
+    
+    return df_final

src/pipeline.py

@@ -0,0 +1,112 @@
+import os
+import json
+import pickle
+import time
+from pathlib import Path
+
+from src.hard_filter import apply_hard_filter
+from src.raw_score import compute_raw_scores
+from src.availability import apply_multipliers
+from src.output import write_submission
+
+def run_pipeline(candidates_path: str, jd_path: str, output_path: str):
+    """Executes the complete candidate vetting, scoring, and ranking pipeline."""
+    start_time = time.perf_counter()
+    
+    print("=" * 60)
+    print("STARTING VETTLY CANDIDATE SCORING AND FILTERING PIPELINE")
+    print("=" * 60)
+    
+    # Resolve paths
+    cand_path = Path(candidates_path)
+    job_desc_path = Path(jd_path)
+    out_path = Path(output_path)
+    
+    # Fallback to check if file exists, if not check common folders
+    if not cand_path.exists():
+        for alt in [Path("data").joinpath(cand_path.name), Path("[PUB] India_runs_data_and_ai_challenge").joinpath("India_runs_data_and_ai_challenge").joinpath(cand_path.name)]:
+            if alt.exists():
+                cand_path = alt
+                break
+                
+    if not job_desc_path.exists():
+        for alt in [Path("data").joinpath(job_desc_path.name), Path("[PUB] India_runs_data_and_ai_challenge").joinpath("India_runs_data_and_ai_challenge").joinpath(job_desc_path.name)]:
+            if alt.exists():
+                job_desc_path = alt
+                break
+                
+    print(f"Candidates Path: {cand_path.resolve()}")
+    print(f"Job Description Path: {job_desc_path.resolve()}")
+    print(f"Output Path: {out_path.resolve()}")
+    
+    # 1. Load candidates and job description
+    if not cand_path.exists():
+        raise FileNotFoundError(f"Candidates file not found at: {cand_path}")
+    if not job_desc_path.exists():
+        raise FileNotFoundError(f"Job description file not found at: {job_desc_path}")
+        
+    print("Loading datasets...")
+    if str(cand_path).endswith(".jsonl"):
+        with open(cand_path, "r", encoding="utf-8") as f:
+            candidates = [json.loads(line) for line in f]
+    else:
+        with open(cand_path, "r", encoding="utf-8") as f:
+            candidates = json.load(f)
+            
+    with open(job_desc_path, "r", encoding="utf-8") as f:
+        jd = json.load(f)
+        
+    print(f"Loaded {len(candidates)} candidates.")
+    
+    # 2. Load TF-IDF vectorizer
+    # TF-IDF resides in data/precomputed/tfidf.pkl relative to candidates directory
+    precomputed_dir = cand_path.parent / "precomputed"
+    tfidf_path = precomputed_dir / "tfidf.pkl"
+    
+    if not tfidf_path.exists():
+        # Try alternate location
+        script_dir = Path(__file__).resolve().parent
+        tfidf_path = script_dir.parent / "data" / "precomputed" / "tfidf.pkl"
+        
+    print(f"Loading TF-IDF Vectorizer from: {tfidf_path}")
+    with open(tfidf_path, "rb") as f:
+        tfidf = pickle.load(f)
+        
+    # 3. Run Hard Filter (Stage 1)
+    print("\n--- STAGE 1: Applying Hard Filters ---")
+    survivors, killed = apply_hard_filter(candidates, jd, tfidf)
+    print(f"Killed: {len(killed)} candidates.")
+    print(f"Surviving: {len(survivors)} candidates.")
+    
+    if not survivors:
+        print("WARNING: No candidates survived the hard filters! Pipeline exiting early.")
+        return
+        
+    # 4. Run Raw Scoring (Stage 2A, 2B, 2C in parallel)
+    print("\n--- STAGE 2: Computing Raw Scores ---")
+    raw_scored = compute_raw_scores(survivors, jd)
+    
+    # 5. Run Behavioral Multipliers (Stage 3)
+    print("\n--- STAGE 3: Applying Behavioral Multipliers ---")
+    final_scored = apply_multipliers(raw_scored, jd)
+    
+    # 6. Run Output Generation (Stage 4)
+    print("\n--- STAGE 4: Generating Output ---")
+    df_final = write_submission(final_scored, jd, str(out_path))
+    
+    elapsed_time = time.perf_counter() - start_time
+    print("=" * 60)
+    print(f"PIPELINE EXECUTED IN {elapsed_time:.2f} SECONDS")
+    print("=" * 60)
+    
+    # Print the top 10 rows from the output DataFrame
+    print("\nTOP 10 CANDIDATES:")
+    print(df_final.head(10))
+
+if __name__ == "__main__":
+    # Default paths assuming execution from workspace root
+    default_candidates = "data/candidates.json"
+    default_jd = "data/job_description.json"
+    default_output = "submission.csv"
+    
+    run_pipeline(default_candidates, default_jd, default_output)

src/precompute.py

@@ -0,0 +1,154 @@
+import os
+import json
+import pickle
+import numpy as np
+import faiss
+from sentence_transformers import SentenceTransformer
+from sklearn.feature_extraction.text import TfidfVectorizer
+from tqdm import tqdm
+
+def build_candidate_text(cand: dict) -> str:
+    """Build a consolidated text string from candidate fields, supporting nested schemas."""
+    profile = cand.get("profile") or {}
+    
+    current_title = profile.get("current_title") or cand.get("current_title") or ""
+    headline = profile.get("headline") or cand.get("headline") or ""
+    
+    # Skill names
+    skills = cand.get("skills") or []
+    skill_names = []
+    for s in skills:
+        if isinstance(s, dict):
+            skill_names.append(s.get("name") or "")
+        elif isinstance(s, str):
+            skill_names.append(s)
+    skills_joined = " ".join(filter(None, skill_names))
+    
+    # Career history
+    history = cand.get("career_history") or cand.get("experience") or cand.get("work_experience") or []
+    history_parts = []
+    if isinstance(history, list):
+        for job in history:
+            if isinstance(job, dict):
+                title = job.get("title") or ""
+                description = job.get("description") or ""
+                title_capped = title[:500]
+                description_capped = description[:500]
+                if title_capped:
+                    history_parts.append(title_capped)
+                if description_capped:
+                    history_parts.append(description_capped)
+                    
+    history_joined = " ".join(history_parts)
+    
+    # Combine parts
+    parts = [current_title, headline, skills_joined, history_joined]
+    cleaned_parts = [p.strip() for p in parts if p and p.strip()]
+    return " ".join(cleaned_parts)
+
+def build_jd_text(jd: dict) -> str:
+    """Build a consolidated text string from job description fields."""
+    parts = []
+    if jd.get("title"):
+        parts.append(jd["title"])
+    if jd.get("must_have_skills"):
+        parts.append(" ".join(jd["must_have_skills"]))
+    if jd.get("nice_to_have_skills"):
+        parts.append(" ".join(jd["nice_to_have_skills"]))
+    if jd.get("keywords"):
+        parts.append(" ".join(jd["keywords"]))
+    if jd.get("description"):
+        parts.append(jd["description"])
+    cleaned_parts = [p.strip() for p in parts if p and p.strip()]
+    return " ".join(cleaned_parts)
+
+def precompute(candidates_path=None, jd_path=None, precomputed_dir=None):
+    # Resolve paths
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    project_root = os.path.dirname(script_dir)
+    
+    if not precomputed_dir:
+        precomputed_dir = os.path.join(project_root, "data", "precomputed")
+    os.makedirs(precomputed_dir, exist_ok=True)
+    
+    if not candidates_path:
+        candidates_path = os.path.join(project_root, "data", "candidates.json")
+        # Try JSON Lines fallback
+        if not os.path.exists(candidates_path) and os.path.exists(candidates_path + "l"):
+            candidates_path += "l"
+            
+    if not jd_path:
+        jd_path = os.path.join(project_root, "data", "job_description.json")
+    
+    print(f"Loading job description from {jd_path}...")
+    with open(jd_path, "r", encoding="utf-8") as f:
+        jd_data = json.load(f)
+    jd_text = build_jd_text(jd_data)
+    
+    print(f"Loading candidates from {candidates_path}...")
+    if not os.path.exists(candidates_path):
+        raise FileNotFoundError(f"Candidates file not found at {candidates_path}")
+        
+    if candidates_path.endswith(".jsonl"):
+        with open(candidates_path, "r", encoding="utf-8") as f:
+            candidates = [json.loads(line) for line in f]
+    else:
+        with open(candidates_path, "r", encoding="utf-8") as f:
+            candidates = json.load(f)
+        
+    print(f"Processing {len(candidates)} candidates...")
+    candidate_texts = []
+    candidate_ids = []
+    for cand in tqdm(candidates, desc="Building candidate texts"):
+        cand_id = cand.get("candidate_id") or cand.get("id") or ""
+        candidate_ids.append(str(cand_id))
+        candidate_texts.append(build_candidate_text(cand))
+        
+    print("Loading SentenceTransformer model 'all-MiniLM-L6-v2'...")
+    model = SentenceTransformer("all-MiniLM-L6-v2")
+    
+    print("Embedding job description...")
+    jd_vec = model.encode(jd_text, normalize_embeddings=True).astype("float32")
+    
+    print("Embedding candidates...")
+    cand_vecs = model.encode(
+        candidate_texts,
+        batch_size=256,
+        show_progress_bar=True,
+        normalize_embeddings=True
+    ).astype("float32")
+    
+    # FAISS index
+    print("Building FAISS index...")
+    dimension = 384
+    index = faiss.IndexFlatIP(dimension)
+    index.add(cand_vecs)
+    
+    # TF-IDF Vectorizer
+    print("Fitting TF-IDF Vectorizer...")
+    tfidf = TfidfVectorizer(max_features=30000, ngram_range=(1, 2))
+    tfidf.fit(candidate_texts)
+    
+    # Save outputs
+    jd_vec_path = os.path.join(precomputed_dir, "jd_vec.npy")
+    cand_vecs_path = os.path.join(precomputed_dir, "cand_vecs.npy")
+    cand_ids_path = os.path.join(precomputed_dir, "cand_ids.json")
+    faiss_index_path = os.path.join(precomputed_dir, "faiss.index")
+    tfidf_pkl_path = os.path.join(precomputed_dir, "tfidf.pkl")
+    
+    print("Saving precomputed outputs...")
+    np.save(jd_vec_path, jd_vec)
+    np.save(cand_vecs_path, cand_vecs)
+    
+    with open(cand_ids_path, "w", encoding="utf-8") as f:
+        json.dump(candidate_ids, f, indent=2)
+        
+    faiss.write_index(index, faiss_index_path)
+    
+    with open(tfidf_pkl_path, "wb") as f:
+        pickle.dump(tfidf, f)
+        
+    print("Precomputation completed successfully.")
+
+if __name__ == "__main__":
+    precompute()

src/raw_score.py

@@ -0,0 +1,81 @@
+import os
+import pickle
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from tqdm import tqdm
+
+from src.score_career import compute_A, compute_keyword_max
+from src.score_skills import compute_B
+from src.score_embed import load_artifacts, get_C_map
+
+def candidate_worker(candidate: dict, jd: dict, tfidf, keyword_max: float) -> tuple[dict, dict, dict]:
+    """Module-level worker function to score a single candidate (picklable for ProcessPoolExecutor)."""
+    A_res = compute_A(candidate, jd, tfidf, keyword_max)
+    B_res = compute_B(candidate, jd)
+    return candidate, A_res, B_res
+
+def compute_raw_scores(survivors: list[dict], jd: dict) -> list[dict]:
+    """Hub module that coordinates Stage 2A, 2B, and 2C to compute raw scores in parallel."""
+    # Resolve paths
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    vettly_dir = os.path.dirname(script_dir)
+    precomputed_dir = os.path.join(vettly_dir, "data", "precomputed")
+    
+    tfidf_pkl_path = os.path.join(precomputed_dir, "tfidf.pkl")
+    
+    print("Loading TF-IDF vectorizer...")
+    with open(tfidf_pkl_path, "rb") as f:
+        tfidf = pickle.load(f)
+        
+    print("Loading embedding artifacts...")
+    jd_vec, cand_vecs, cand_ids = load_artifacts(precomputed_dir)
+    
+    print("Computing embedding similarities (Stage 2C)...")
+    C_map = get_C_map(jd_vec, cand_vecs, cand_ids)
+    
+    print("Computing pool-wide keyword density maximum...")
+    keyword_max = compute_keyword_max(survivors, jd, tfidf)
+    
+    results = []
+    
+    # Process in parallel
+    print(f"Parallelizing scoring for {len(survivors)} candidates across CPU cores...")
+    max_workers = os.cpu_count() or 4
+    
+    with ProcessPoolExecutor(max_workers=max_workers) as executor:
+        futures = {
+            executor.submit(candidate_worker, cand, jd, tfidf, keyword_max): cand
+            for cand in survivors
+        }
+        
+        for future in tqdm(as_completed(futures), total=len(futures), desc="Scoring candidates"):
+            try:
+                candidate, A_res, B_res = future.result()
+                cand_id = str(candidate.get("id") or candidate.get("candidate_id") or "")
+                
+                # Get Stage 2C embedding score
+                C = C_map.get(cand_id, 0.0)
+                
+                # Assemble raw score: 0.40*A + 0.35*B + 0.25*C
+                A = A_res["A"]
+                B = B_res["B"]
+                raw_score = round(0.40 * A + 0.35 * B + 0.25 * C, 4)
+                
+                results.append({
+                    "candidate_id": cand_id,
+                    "candidate": candidate,
+                    "A": A,
+                    "title_sim": A_res["title_sim"],
+                    "industry_match": A_res["industry_match"],
+                    "prod_keyword_density": A_res["prod_keyword_density"],
+                    "yoe_score": A_res["yoe_score"],
+                    "B": B,
+                    "must_have_coverage": B_res["must_have_coverage"],
+                    "nice_coverage": B_res["nice_coverage"],
+                    "cert_bonus": B_res["cert_bonus"],
+                    "C": C,
+                    "raw_score": raw_score
+                })
+            except Exception as e:
+                print(f"Error scoring candidate: {e}")
+                
+    return results

src/run_fast_pipeline.py

@@ -0,0 +1,156 @@
+import os
+import json
+import time
+import numpy as np
+import faiss
+import polars as pl
+from pathlib import Path
+from sentence_transformers import SentenceTransformer
+from sklearn.feature_extraction.text import TfidfVectorizer
+
+from src.hard_filter import apply_hard_filter, is_killed
+from src.score_career import compute_A, compute_keyword_max
+from src.score_skills import compute_B
+from src.score_embed import compute_C_all
+from src.availability import apply_multipliers
+from src.output import write_submission, generate_reasoning
+from src.precompute import build_candidate_text, build_jd_text
+
+def run_fast_pipeline():
+    start_time = time.perf_counter()
+    
+    print("=" * 60)
+    print("RUNNING HIGH-PERFORMANCE PIPELINE ON 100,000 CANDIDATES")
+    print("=" * 60)
+    
+    # Paths
+    jd_path = Path("data/job_description.json")
+    candidates_path = Path("data/candidates.jsonl")
+    output_path = Path("submission.csv")
+    
+    if not jd_path.exists():
+        raise FileNotFoundError(f"Job description not found at {jd_path}")
+    if not candidates_path.exists():
+        raise FileNotFoundError(f"Candidates dataset not found at {candidates_path}")
+        
+    # 1. Load Job Description
+    print("Loading Job Description...")
+    with open(jd_path, "r", encoding="utf-8") as f:
+        jd = json.load(f)
+        
+    # ── PASS 1: Stream titles only to fit TF-IDF (low memory) ──
+    print("Pass 1: Streaming 100,000 candidate titles for TF-IDF fit...")
+    titles = []
+    with open(candidates_path, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            cand = json.loads(line)
+            title = cand.get("profile", {}).get("current_title") or cand.get("current_title") or ""
+            titles.append(title)
+            
+    total_count = len(titles)
+    print(f"  Collected {total_count} titles.")
+    
+    # 2. Fit TF-IDF on all candidate titles for the hard filter
+    print("Fitting TF-IDF Vectorizer on all candidate titles...")
+    tfidf = TfidfVectorizer(max_features=30000, ngram_range=(1, 2))
+    tfidf.fit(titles)
+    del titles  # free memory immediately
+    
+    # ── PASS 2: Stream again — hard-filter inline and collect survivors ──
+    print("\n--- STAGE 1: Applying Hard Filters (streaming pass 2) ---")
+    survivors = []
+    killed_count = 0
+    with open(candidates_path, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            cand = json.loads(line)
+            killed_flag, reason = is_killed(cand, jd, tfidf)
+            if killed_flag:
+                killed_count += 1
+            else:
+                survivors.append(cand)
+                
+    print(f"Killed:    {killed_count} candidates.")
+    print(f"Surviving: {len(survivors)} candidates.")
+    
+    if not survivors:
+        print("No candidates survived. Exiting.")
+        return
+        
+    # 4. Generate Embeddings for ONLY the survivors + JD (Massive speedup!)
+    print("\n--- STAGE 2: Embedding Survivors & Job Description ---")
+    print("Loading SentenceTransformer model 'all-MiniLM-L6-v2'...")
+    model = SentenceTransformer("all-MiniLM-L6-v2")
+    
+    print("Embedding Job Description...")
+    jd_text = build_jd_text(jd)
+    jd_vec = model.encode(jd_text, normalize_embeddings=True).astype("float32")
+    
+    print(f"Embedding {len(survivors)} surviving candidates...")
+    survivor_texts = [build_candidate_text(s) for s in survivors]
+    cand_vecs = model.encode(survivor_texts, batch_size=256, normalize_embeddings=True).astype("float32")
+    
+    # 5. Compute Stage 2C Embedding Similarity
+    print("Computing embedding similarities (Stage 2C)...")
+    C_scores = compute_C_all(jd_vec, cand_vecs)
+    C_map = {str(s.get("candidate_id") or s.get("id")): float(score) for s, score in zip(survivors, C_scores)}
+    
+    # 6. Fit TF-IDF on survivor texts for career scoring
+    print("Fitting TF-IDF Vectorizer on survivor texts...")
+    tfidf_surv = TfidfVectorizer(max_features=30000, ngram_range=(1, 2))
+    tfidf_surv.fit(survivor_texts)
+    
+    # 7. Compute Stage 2A & 2B scores
+    print("Computing career and skill scores...")
+    keyword_max = compute_keyword_max(survivors, jd, tfidf_surv)
+    
+    raw_scored = []
+    for cand in survivors:
+        cand_id = str(cand.get("candidate_id") or cand.get("id"))
+        A_res = compute_A(cand, jd, tfidf_surv, keyword_max)
+        B_res = compute_B(cand, jd)
+        C = C_map.get(cand_id, 0.0)
+        
+        A = A_res["A"]
+        B = B_res["B"]
+        raw_score = round(0.40 * A + 0.35 * B + 0.25 * C, 4)
+        
+        raw_scored.append({
+            "candidate_id": cand_id,
+            "candidate": cand,
+            "A": A,
+            "title_sim": A_res["title_sim"],
+            "industry_match": A_res["industry_match"],
+            "prod_keyword_density": A_res["prod_keyword_density"],
+            "yoe_score": A_res["yoe_score"],
+            "B": B,
+            "must_have_coverage": B_res["must_have_coverage"],
+            "nice_coverage": B_res["nice_coverage"],
+            "cert_bonus": B_res["cert_bonus"],
+            "C": C,
+            "raw_score": raw_score
+        })
+        
+    # 8. Apply Behavioral Multipliers (Stage 3)
+    print("\n--- STAGE 3: Applying Behavioral Multipliers ---")
+    final_scored = apply_multipliers(raw_scored, jd)
+    
+    # 9. Output ranked results (Stage 4)
+    print("\n--- STAGE 4: Generating Output ---")
+    df_final = write_submission(final_scored, jd, str(output_path))
+    
+    elapsed_time = time.perf_counter() - start_time
+    print("=" * 60)
+    print(f"PIPELINE EXECUTED IN {elapsed_time:.2f} SECONDS")
+    print("=" * 60)
+    
+    print("\nTOP 10 CANDIDATES:")
+    print(df_final.head(10))
+
+if __name__ == "__main__":
+    run_fast_pipeline()

src/score_career.py

@@ -0,0 +1,119 @@
+import datetime
+import numpy as np
+from dateutil import parser
+from sklearn.metrics.pairwise import cosine_similarity
+
+def compute_job_age(job: dict) -> float:
+    """Calculate the age of a job in years from today."""
+    is_current = job.get("is_current")
+    end_date_str = job.get("end_date")
+    if is_current or not end_date_str or str(end_date_str).lower().strip() in ["present", "current", "none", "null"]:
+        return 0.0
+        
+    try:
+        end_date = parser.parse(str(end_date_str))
+        if end_date.tzinfo is not None:
+            today = datetime.datetime.now(datetime.timezone.utc)
+        else:
+            today = datetime.datetime.now()
+        days = (today - end_date).days
+        return max(0.0, days / 365.25)
+    except Exception:
+        return 0.0
+
+def compute_raw_keyword_score(candidate: dict, jd: dict) -> float:
+    """Compute the raw, recency-decayed keyword score for a candidate, supporting nested career_history."""
+    keywords = jd.get("keywords") or []
+    history = candidate.get("career_history") or candidate.get("experience") or candidate.get("work_experience") or []
+    if not keywords or not history:
+        return 0.0
+        
+    total_score = 0.0
+    for job in history:
+        if not isinstance(job, dict):
+            continue
+        job_title = job.get("title") or ""
+        job_desc = job.get("description") or ""
+        job_text = f"{job_title} {job_desc}".lower()
+        
+        # Count keyword occurrences
+        kw_count = sum(job_text.count(str(kw).lower()) for kw in keywords)
+        
+        age = compute_job_age(job)
+        decay = np.exp(-0.15 * age)
+        total_score += kw_count * decay
+        
+    return float(total_score)
+
+def compute_keyword_max(candidates: list[dict], jd: dict, tfidf=None) -> float:
+    """Scans the candidate pool and returns the maximum raw keyword score."""
+    max_val = 0.0
+    for cand in candidates:
+        score = compute_raw_keyword_score(cand, jd)
+        if score > max_val:
+            max_val = score
+    # Return at least 1.0 to avoid division by zero
+    return max(max_val, 1.0)
+
+def compute_A(candidate: dict, jd: dict, tfidf, keyword_max: float) -> dict:
+    """Computes the career fit score (A) for a candidate, supporting nested and flat schemas."""
+    profile = candidate.get("profile") or {}
+    
+    # 1. Title Similarity
+    cand_title = profile.get("current_title") or candidate.get("current_title", "")
+    jd_title = jd.get("title", "")
+    if not cand_title or not jd_title:
+        title_sim = 0.0
+    else:
+        try:
+            cand_tfidf = tfidf.transform([cand_title])
+            jd_tfidf = tfidf.transform([jd_title])
+            title_sim = float(cosine_similarity(cand_tfidf, jd_tfidf)[0][0])
+        except Exception:
+            title_sim = 0.0
+            
+    # 2. Industry Match (Jaccard similarity)
+    cand_industries = set()
+    # Check profile industry
+    prof_ind = profile.get("current_industry")
+    if prof_ind:
+        cand_industries.add(prof_ind.strip().lower())
+    # Check career history industries
+    history = candidate.get("career_history") or candidate.get("experience") or candidate.get("work_experience") or []
+    for job in history:
+        if isinstance(job, dict) and job.get("industry"):
+            cand_industries.add(job["industry"].strip().lower())
+    # Fallback to top-level industries list if present
+    for ind in (candidate.get("industries") or []):
+        if ind:
+            cand_industries.add(ind.strip().lower())
+            
+    jd_industries = {ind.strip().lower() for ind in (jd.get("target_industries") or []) if ind}
+    if not jd_industries:
+        industry_match = 0.0
+    else:
+        union = cand_industries.union(jd_industries)
+        intersection = cand_industries.intersection(jd_industries)
+        industry_match = len(intersection) / len(union) if union else 0.0
+        
+    # 3. Keyword Density
+    raw_kw = compute_raw_keyword_score(candidate, jd)
+    prod_keyword_density = min(raw_kw / keyword_max, 1.0)
+    
+    # 4. YoE Score
+    yoe = profile.get("years_of_experience") or profile.get("yoe") or candidate.get("years_of_experience") or candidate.get("yoe") or 0.0
+    min_yoe = jd.get("min_yoe") or 5
+    if min_yoe <= 0:
+        yoe_score = 1.0
+    else:
+        yoe_score = min(float(yoe) / float(min_yoe), 1.0)
+        
+    A = 0.35 * title_sim + 0.25 * industry_match + 0.25 * prod_keyword_density + 0.15 * yoe_score
+    
+    return {
+        "A": round(A, 4),
+        "title_sim": round(title_sim, 4),
+        "industry_match": round(industry_match, 4),
+        "prod_keyword_density": round(prod_keyword_density, 4),
+        "yoe_score": round(yoe_score, 4)
+    }

src/score_embed.py

@@ -0,0 +1,34 @@
+import os
+import json
+import numpy as np
+
+def load_artifacts(precomputed_dir: str) -> tuple[np.ndarray, np.ndarray, list]:
+    """Loads the precomputed artifacts: jd_vec.npy, cand_vecs.npy, and cand_ids.json."""
+    jd_vec_path = os.path.join(precomputed_dir, "jd_vec.npy")
+    cand_vecs_path = os.path.join(precomputed_dir, "cand_vecs.npy")
+    cand_ids_path = os.path.join(precomputed_dir, "cand_ids.json")
+    
+    jd_vec = np.load(jd_vec_path)
+    cand_vecs = np.load(cand_vecs_path)
+    
+    with open(cand_ids_path, "r", encoding="utf-8") as f:
+        cand_ids = json.load(f)
+        
+    return jd_vec, cand_vecs, cand_ids
+
+def compute_C_all(jd_vec: np.ndarray, cand_vecs: np.ndarray) -> np.ndarray:
+    """Computes semantic similarity for all candidates in a single vectorized matrix multiplication."""
+    # Ensure 2D shapes for multiplication
+    if len(jd_vec.shape) == 1:
+        jd_vec = jd_vec.reshape(1, -1)
+    if len(cand_vecs.shape) == 1:
+        cand_vecs = cand_vecs.reshape(1, -1)
+        
+    scores = (cand_vecs @ jd_vec.T).squeeze()
+    scores = np.clip(scores, 0.0, 1.0)
+    return np.atleast_1d(scores)
+
+def get_C_map(jd_vec: np.ndarray, cand_vecs: np.ndarray, cand_ids: list) -> dict[str, float]:
+    """Computes similarity and returns a dictionary mapping candidate IDs to embedding scores."""
+    scores = compute_C_all(jd_vec, cand_vecs)
+    return {str(cid): float(score) for cid, score in zip(cand_ids, scores)}

src/score_skills.py

@@ -0,0 +1,138 @@
+from rapidfuzz import fuzz
+
+def fuzzy_match_skill(cand_skill_name: str, jd_skills: list[str]) -> str | None:
+    """Finds the best matching JD skill for a candidate's skill using a threshold of 85.
+    
+    Returns the JD skill name if matched, else None.
+    """
+    if not cand_skill_name or not jd_skills:
+        return None
+    best_ratio = 0
+    best_skill = None
+    for jd_s in jd_skills:
+        ratio = fuzz.token_sort_ratio(cand_skill_name.lower().strip(), jd_s.lower().strip())
+        if ratio > best_ratio:
+            best_ratio = ratio
+            best_skill = jd_s
+    if best_ratio >= 85:
+        return best_skill
+    return None
+
+def skill_trust(skill_obj, assess_scores: dict, jd_skill: str) -> float:
+    """Calculates the trust score for a matched skill.
+    
+    Formula: trust = prof_weight * (0.35 + 0.25 * endorse_w + 0.25 * assess_w + 0.15 * duration_w)
+    """
+    if isinstance(skill_obj, str):
+        name = skill_obj
+        proficiency = "intermediate"
+        endorsements = 0
+        duration_months = 0
+    elif isinstance(skill_obj, dict):
+        name = skill_obj.get("name") or ""
+        proficiency = skill_obj.get("proficiency") or "intermediate"
+        endorsements = skill_obj.get("endorsements") or 0
+        duration_months = skill_obj.get("duration_months") or 0
+    else:
+        return 0.0
+        
+    # 1. Proficiency Weight
+    prof = str(proficiency).lower().strip()
+    if prof == "beginner":
+        prof_weight = 0.40
+    elif prof == "intermediate":
+        prof_weight = 0.70
+    elif prof == "advanced":
+        prof_weight = 0.90
+    elif prof in ["expert", "master"]:
+        prof_weight = 1.00
+    else:
+        prof_weight = 0.70  # default
+        
+    # 2. Endorsement Weight
+    endorse_w = min(float(endorsements) / 20.0, 1.0)
+    
+    # 3. Assessment Weight
+    assess_val = 0.0
+    if isinstance(assess_scores, dict):
+        # Try exact key lookup or fuzzy key lookup in the assessment scores
+        # First try exact case-insensitive match on jd_skill or name
+        assess_val = assess_scores.get(jd_skill) or assess_scores.get(name)
+        if assess_val is None:
+            # Try case-insensitive lookup
+            lower_scores = {k.lower(): v for k, v in assess_scores.items()}
+            assess_val = lower_scores.get(jd_skill.lower()) or lower_scores.get(name.lower()) or 0.0
+    assess_w = float(assess_val) / 100.0
+    
+    # 4. Duration Weight
+    duration_w = min(float(duration_months) / 24.0, 1.0)
+    
+    trust = prof_weight * (0.35 + 0.25 * endorse_w + 0.25 * assess_w + 0.15 * duration_w)
+    return float(trust)
+
+def compute_B(candidate: dict, jd: dict) -> dict:
+    """Computes the skill trust score (B) for a candidate, supporting nested and flat structures."""
+    must_have_skills = jd.get("must_have_skills") or []
+    nice_to_have_skills = jd.get("nice_to_have_skills") or []
+    
+    must_trust = {s: 0.0 for s in must_have_skills}
+    nice_trust = {s: 0.0 for s in nice_to_have_skills}
+    
+    cand_skills = candidate.get("skills") or []
+    
+    # Try looking in nested redrob_signals for assessment scores first
+    signals = candidate.get("redrob_signals") or {}
+    assess_scores = signals.get("skill_assessment_scores")
+    if assess_scores is None:
+        assess_scores = candidate.get("skill_assessment_scores") or {}
+    
+    # Calculate trust for matching skills
+    for s_obj in cand_skills:
+        s_name = s_obj if isinstance(s_obj, str) else s_obj.get("name", "")
+        if not s_name:
+            continue
+            
+        matched_must = fuzzy_match_skill(s_name, must_have_skills)
+        if matched_must:
+            t = skill_trust(s_obj, assess_scores, matched_must)
+            must_trust[matched_must] = max(must_trust[matched_must], t)
+            
+        matched_nice = fuzzy_match_skill(s_name, nice_to_have_skills)
+        if matched_nice:
+            t = skill_trust(s_obj, assess_scores, matched_nice)
+            nice_trust[matched_nice] = max(nice_trust[matched_nice], t)
+            
+    must_cov = sum(must_trust.values()) / len(must_have_skills) if must_have_skills else 0.0
+    nice_cov = sum(nice_trust.values()) / len(nice_to_have_skills) if nice_to_have_skills else 0.0
+    
+    # 5. Certification Bonus
+    certs = candidate.get("certifications") or candidate.get("certs") or []
+    all_jd_skills = must_have_skills + nice_to_have_skills
+    cert_matches = 0
+    for cert in certs:
+        cert_name = cert if isinstance(cert, str) else cert.get("name", "")
+        if not cert_name:
+            continue
+            
+        # Check if cert fuzzy matches any JD skill or contains it as a substring
+        matched = False
+        for jd_s in all_jd_skills:
+            if fuzz.token_sort_ratio(cert_name.lower().strip(), jd_s.lower().strip()) >= 85:
+                matched = True
+                break
+            if jd_s.lower().strip() in cert_name.lower():
+                matched = True
+                break
+        if matched:
+            cert_matches += 1
+            
+    cert_bonus = min(cert_matches * 0.05, 0.15)
+    
+    B = min(0.75 * must_cov + 0.25 * nice_cov + cert_bonus, 1.0)
+    
+    return {
+        "B": round(B, 4),
+        "must_have_coverage": round(must_cov, 4),
+        "nice_coverage": round(nice_cov, 4),
+        "cert_bonus": round(cert_bonus, 4)
+    }