lib/features.py

"""
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"}