precompute.py

#!/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()