Deploy Lighthouse Streamlit app (docker space)

Dockerfile

@@ -0,0 +1,18 @@
+# Lighthouse — self-contained sandbox image (also used by the HuggingFace Docker Space).
+# Runs the Streamlit demo on CPU. For the full 100K rank, use rank.py (see README).
+FROM python:3.10-slim
+
+WORKDIR /app
+
+# CPU-only torch first (kept out of the CUDA default to keep the image small)
+RUN pip install --no-cache-dir torch==2.5.1 --index-url https://download.pytorch.org/whl/cpu
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+# HuggingFace Spaces expects the app on 7860
+ENV PORT=7860
+EXPOSE 7860
+CMD ["streamlit", "run", "app/app.py", "--server.port=7860", "--server.address=0.0.0.0"]

README.md

@@ -1,10 +1,20 @@
----
-title: Lighthouse
-emoji: 📉
-colorFrom: red
-colorTo: purple
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: Lighthouse Candidate Ranker
+emoji: 🔦
+colorFrom: indigo
+colorTo: yellow
+sdk: docker
+app_port: 7860
+pinned: false
+license: mit
+---
+
+# 🔦 Lighthouse - candidate ranker sandbox
+
+Live demo of the Lighthouse recruiter-grade candidate ranker (Redrob India Runs challenge).
+Upload up to 100 candidates (or use the preloaded 100-candidate sample) and Lighthouse ranks
+them end-to-end on CPU - five-component fit score, JD hard-negative gates, behavioral modifier,
+honeypot zeroing, grounded reasoning - then download the ranked CSV.
+
+Uploaded candidates are encoded on the fly with BAAI/bge-small-en-v1.5. Full code:
+https://github.com/jacklachan/lighthouse-indiaruns

app/app.py

@@ -0,0 +1,132 @@
+"""Lighthouse — HuggingFace Spaces / Streamlit sandbox.
+
+Accepts up to 100 candidates (preloaded sample or uploaded JSONL), runs the full
+Lighthouse ranker end-to-end on CPU, and shows the ranked table with scores +
+grounded reasoning. Candidates uploaded here are not precomputed, so the small
+bge-small encoder runs at request time (a few seconds for <=100 rows) — this is
+the demo path; the official 100K run uses fully precomputed embeddings.
+
+Run locally:  streamlit run app/app.py
+"""
+import json
+import os
+import sys
+
+import numpy as np
+import pandas as pd
+import streamlit as st
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from lighthouse import loader, ranker, reasoning  # noqa: E402
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+ART = os.path.join(ROOT, "artifacts")
+SAMPLE = os.path.join(HERE, "sample_candidates.jsonl")
+
+st.set_page_config(page_title="Lighthouse — Candidate Ranker", page_icon="🔦", layout="wide")
+
+
+@st.cache_resource
+def _facets_and_rubric():
+    rubric = json.load(open(os.path.join(ART, "jd_rubric.json"), encoding="utf-8"))
+    facet_emb = np.load(os.path.join(ART, "jd_facet_emb.npy"))
+    return rubric, facet_emb
+
+
+def _build_art(rubric, facet_emb):
+    """Empty precomputed set -> every candidate is encoded on the fly."""
+    dim = facet_emb.shape[1]
+    return {"rubric": rubric, "ids": [], "id_to_row": {},
+            "cand_emb": np.zeros((0, dim), dtype=np.float32), "facet_emb": facet_emb}
+
+
+def _parse_jsonl(text: str):
+    raws = []
+    for line in text.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            raws.append(json.loads(line))
+        except json.JSONDecodeError:
+            pass
+    return raws
+
+
+st.title("🔦 Lighthouse — recruiter-grade candidate ranker")
+st.caption("Keyword filters surface the loudest profiles. Lighthouse surfaces the right ones — "
+           "and ignores the fakes that fool keyword filters.")
+
+with st.sidebar:
+    st.header("Input")
+    mode = st.radio("Candidate source", ["Preloaded sample (100)", "Upload JSONL (≤100)"])
+    st.markdown("---")
+    st.markdown("**JD:** Senior AI Engineer @ Redrob AI — production embeddings/retrieval, "
+                "ranking-eval, product (not services), 6–8 yrs ideal, Noida/Pune/India.")
+    st.markdown("Ranking runs on **CPU, no hosted LLM**. The five-component score is gated by "
+                "JD hard-negatives and a behavioral modifier; honeypots are zeroed.")
+
+rubric, facet_emb = _facets_and_rubric()
+
+raws = []
+if mode.startswith("Preloaded"):
+    if os.path.exists(SAMPLE):
+        raws = _parse_jsonl(open(SAMPLE, encoding="utf-8").read())
+        st.info(f"Loaded {len(raws)} preloaded sample candidates.")
+    else:
+        st.error("sample_candidates.jsonl not found.")
+else:
+    up = st.file_uploader("Upload candidates JSONL (one JSON candidate per line)", type=["jsonl", "json"])
+    if up is not None:
+        raws = _parse_jsonl(up.read().decode("utf-8"))
+        st.info(f"Parsed {len(raws)} candidates.")
+
+raws = raws[:100]
+
+if raws and st.button("🔦 Rank candidates", type="primary"):
+    with st.spinner(f"Encoding + scoring {len(raws)} candidates on CPU ..."):
+        art = _build_art(rubric, facet_emb)
+        records = ranker.score_all(raws, art)
+        mx = max((r["final_score"] for r in records), default=0.0)
+        if mx > 0:
+            for r in records:
+                r["final_score"] = round(r["final_score"] / mx, 6)
+        top = ranker.rank_records(records, top=len(records))
+        raw_by_id = {loader.candidate_id(r): r for r in raws}
+        rows = []
+        for rec in top:
+            raw = raw_by_id[rec["candidate_id"]]
+            p = loader.get_profile(raw)
+            rows.append({
+                "rank": rec["rank"],
+                "candidate_id": rec["candidate_id"],
+                "score": rec["final_score"],
+                "title": loader._s(p, "current_title"),
+                "country": loader._s(p, "country"),
+                "yrs": loader._f(p, "years_of_experience"),
+                "honeypot": "⚠️" if rec["honeypot"] else "",
+                "reasoning": reasoning.generate(raw, rubric, rec),
+            })
+    df = pd.DataFrame(rows)
+    n_hp = sum(1 for r in top if r["honeypot"])
+    c1, c2, c3 = st.columns(3)
+    c1.metric("Candidates ranked", len(rows))
+    c2.metric("Honeypots flagged", n_hp)
+    c3.metric("Top score", f"{rows[0]['score']:.3f}" if rows else "—")
+    st.dataframe(df, use_container_width=True, hide_index=True,
+                 column_config={"score": st.column_config.NumberColumn(format="%.4f")})
+
+    with st.expander("Inspect the top candidate's component breakdown"):
+        best = top[0]
+        st.json({"candidate_id": best["candidate_id"], "components": best["components"],
+                 "base": best["base"], "gate_mult": best["gate_mult"],
+                 "gate_reasons": best["gate_reasons"], "behavior_mult": best["behavior_mult"],
+                 "honeypot": best["honeypot"]})
+
+    st.download_button("⬇️ Download ranking CSV",
+                       df[["candidate_id", "rank", "score", "reasoning"]].to_csv(index=False),
+                       file_name="submission.csv", mime="text/csv")
+elif not raws:
+    st.warning("Choose the preloaded sample or upload a JSONL to begin.")

artifacts/jd_facet_emb.npy

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a6b3d7b669928bc60e95057d092ba8bdac5c462b11f42d02a085cb314ed58e42
+size 15488

artifacts/jd_rubric.json

@@ -0,0 +1,179 @@
+{
+  "_meta": {
+    "title": "Senior AI Engineer — Founding Team @ Redrob AI",
+    "authored_by": "Claude (offline, reading job_description.docx)",
+    "authored_at": "2026-06-06",
+    "reference_date": "2026-06-06",
+    "note": "This rubric is a STATIC, committed artifact. rank.py reads it at rank-time; no LLM/API call is made during ranking. It encodes the gap between what the JD literally says and what it means, per the JD's own 'how to read between the lines' and 'final note for hackathon participants' sections."
+  },
+
+  "experience": {
+    "band_min": 5.0,
+    "band_max": 9.0,
+    "ideal_min": 6.0,
+    "ideal_max": 8.0,
+    "rationale": "JD: '5-9 years ... a range, not a requirement', ideal '6-8 years total, of which 4-5 in applied ML/AI at product companies'. Soft curve: peak inside ideal, gentle taper across the band, steeper outside. Strong signals can rescue out-of-band candidates, so this is a score component, not a hard gate.",
+    "soft": true
+  },
+
+  "facets": [
+    "Production experience with embeddings-based retrieval systems (sentence-transformers, OpenAI embeddings, BGE, E5) deployed to real users, handling embedding drift and retrieval-quality regression.",
+    "Production experience with vector databases or hybrid search infrastructure: Pinecone, Weaviate, Qdrant, Milvus, FAISS, OpenSearch, Elasticsearch.",
+    "Designed and ran evaluation frameworks for ranking systems: NDCG, MRR, MAP, offline-to-online correlation, A/B test interpretation.",
+    "Built and shipped an end-to-end ranking, search, or recommendation system to real users at meaningful scale.",
+    "Hybrid retrieval and re-ranking: dense vs sparse retrieval, learning-to-rank, LLM-based re-ranking, relevance tuning.",
+    "Applied machine learning at a product company: feature engineering through deployment, owning a model in production.",
+    "Information retrieval and NLP: semantic search, query understanding, text embeddings, retrieval relevance.",
+    "Strong Python engineering with attention to code quality.",
+    "Learning-to-rank models with XGBoost, LightGBM, or neural rankers over click-through and human-judgement labels.",
+    "Scrappy product-engineering: shipping a working ranker fast and iterating from real user feedback, comfortable owning the intelligence/matching layer of a product."
+  ],
+
+  "must_haves": [
+    {"key": "embeddings_retrieval", "label": "Production embeddings-based retrieval", "keywords": ["embedding", "embeddings", "sentence transformer", "sentence-transformers", "bge", "e5", "dense retrieval", "vector search", "semantic search", "openai embeddings"]},
+    {"key": "vector_search_infra", "label": "Vector DB / hybrid search infra", "keywords": ["pinecone", "weaviate", "qdrant", "milvus", "faiss", "opensearch", "elasticsearch", "vector database", "hybrid search", "ann index", "hnsw"]},
+    {"key": "ranking_eval", "label": "Ranking evaluation frameworks", "keywords": ["ndcg", "mrr", "map", "mean average precision", "offline-online", "offline to online", "a/b test", "ab test", "relevance label", "click-through", "ctr", "evaluation framework", "learning to rank", "learning-to-rank"]},
+    {"key": "shipped_ranking_system", "label": "Shipped a ranking/search/recsys system", "keywords": ["ranking", "search", "recommendation", "recommender", "recsys", "retrieval", "relevance", "personalization", "discovery feed", "matching"]},
+    {"key": "strong_python", "label": "Strong Python", "keywords": ["python", "pytorch", "tensorflow", "scikit-learn", "numpy", "pandas"]}
+  ],
+
+  "nice_to_haves": [
+    {"key": "llm_finetuning", "label": "LLM fine-tuning (LoRA/QLoRA/PEFT)", "keywords": ["lora", "qlora", "peft", "fine-tun", "fine tun", "instruction tuning", "rlhf"]},
+    {"key": "ltr_models", "label": "Learning-to-rank models", "keywords": ["xgboost", "lightgbm", "learning to rank", "learning-to-rank", "lambdamart", "ranknet"]},
+    {"key": "hrtech", "label": "HR-tech / recruiting / marketplace", "keywords": ["recruit", "hr-tech", "hrtech", "talent", "marketplace", "two-sided"]},
+    {"key": "distributed_inference", "label": "Distributed systems / large-scale inference", "keywords": ["distributed", "large-scale inference", "low latency", "serving", "throughput", "scale"]},
+    {"key": "open_source", "label": "Open-source contributions", "keywords": ["open source", "open-source", "github", "maintainer", "contributor", "published", "paper", "talk"]}
+  ],
+
+  "role_taxonomy": {
+    "positive_title_terms": [
+      "machine learning", "ml engineer", "ml scientist", "ai engineer", "applied scientist",
+      "applied ml", "data scientist", "research engineer", "nlp", "search engineer",
+      "ranking", "recommendation", "recsys", "deep learning", "software engineer",
+      "backend engineer", "data engineer", "mlops", "platform engineer", "staff engineer",
+      "research scientist", "information retrieval", "relevance engineer", "full stack"
+    ],
+    "strong_positive_title_terms": [
+      "machine learning", "ml engineer", "ai engineer", "applied scientist", "applied ml",
+      "nlp", "search engineer", "ranking", "recommendation", "recsys", "research engineer",
+      "data scientist", "relevance engineer", "information retrieval"
+    ],
+    "negative_title_terms": [
+      "marketing", "hr ", "human resources", "accountant", "accounting", "sales",
+      "operations manager", "graphic", "designer", "civil engineer", "mechanical engineer",
+      "content writer", "customer support", "recruiter", "business analyst",
+      "project manager", "program manager", "qa engineer", "quality assurance",
+      "frontend engineer", "mobile developer", "ui ", "ux "
+    ],
+    "rationale": "role_coherence is the decisive signal against keyword-stuffers. The JD's own example: 'A candidate who has all the AI keywords listed as skills but whose title is Marketing Manager is not a fit, no matter how perfect their skill list looks.' Title alone is noisy, so role_coherence blends taxonomy match of current+historical titles with semantic_fit of the career text."
+  },
+
+  "career_evidence_terms": [
+    "ranking", "learning to rank", "learning-to-rank", "search", "semantic search",
+    "recommendation", "recommender", "recsys", "retrieval", "information retrieval",
+    "relevance", "embedding", "embeddings", "vector", "personalization",
+    "discovery feed", "click-through", "offline-online", "a/b test", "ndcg",
+    "feature engineering", "production", "shipped", "deployed", "at scale", "real users"
+  ],
+
+  "hard_negatives": [
+    {
+      "key": "services_only",
+      "label": "Services/consulting-only career (no product company)",
+      "companies": ["tcs", "tata consultancy", "infosys", "wipro", "accenture", "cognizant", "capgemini", "tech mahindra", "hcl", "mindtree", "ltimindtree", "mphasis", "igate", "syntel", "l&t infotech", "larsen", "dxc", "ibm global services"],
+      "penalty": 0.45,
+      "rule": "Apply when EVERY role's company is a services/consulting firm. JD: 'People who have only worked at consulting firms ... in their entire career.' If currently at one but has prior product-company experience, do NOT penalize (JD says that's fine).",
+      "rationale": "Multiplicative penalty, not a zero — a services-only candidate is a weak fit but not impossible/fraudulent."
+    },
+    {
+      "key": "location_visa",
+      "label": "Outside India and unwilling to relocate (no visa sponsorship)",
+      "penalty": 0.30,
+      "soft_penalty": 0.80,
+      "rule": "country != India AND willing_to_relocate == false -> penalty (0.30). country != India AND willing_to_relocate == true -> soft_penalty (0.80, visa/logistics risk). country == India -> no penalty (JD welcomes Hyderabad/Pune/Mumbai/Delhi-NCR and is flexible within India).",
+      "rationale": "JD: 'Outside India: case-by-case, but we don't sponsor work visas.' A Toronto-based candidate who won't relocate (e.g. CAND_0000001) is effectively unhireable here."
+    },
+    {
+      "key": "research_only",
+      "label": "Pure research, no production deployment",
+      "positive_terms": ["research", "phd", "postdoc", "thesis", "publication", "academic", "professor", "lab", "paper"],
+      "production_terms": ["production", "shipped", "deployed", "real users", "at scale", "serving", "product", "launched", "a/b"],
+      "penalty": 0.55,
+      "rule": "Apply when research signals are strong across the career AND production signals are essentially absent. JD: 'pure research environments ... without any production deployment — we will not move forward.'"
+    },
+    {
+      "key": "cv_speech_only",
+      "label": "CV/speech/robotics primary, no NLP/IR",
+      "domain_terms": ["computer vision", "image classification", "object detection", "speech recognition", "tts", "asr", "robotics", "gans", "image segmentation", "ocr", "pose estimation"],
+      "nlp_ir_terms": ["nlp", "natural language", "information retrieval", "retrieval", "search", "ranking", "embedding", "recommendation", "text", "language model"],
+      "penalty": 0.60,
+      "rule": "Apply when CV/speech/robotics dominate skills/career AND NLP/IR signals are weak/absent. JD: 'primary expertise is computer vision, speech, or robotics without significant NLP/IR exposure ... you'd be re-learning fundamentals here.'"
+    },
+    {
+      "key": "langchain_only_recent",
+      "label": "AI experience only recent LangChain+OpenAI wrappers",
+      "wrapper_terms": ["langchain", "llamaindex", "llama-index", "openai api", "prompt engineering", "chatgpt", "gpt wrapper", "autogen"],
+      "depth_terms": ["embedding", "retrieval", "ranking", "recommendation", "feature engineering", "xgboost", "lightgbm", "model training", "mlops", "information retrieval", "pre-llm", "classical ml"],
+      "penalty": 0.65,
+      "rule": "Apply when AI signals are recent (<= ~18 months) and consist of LangChain/OpenAI-wrapper terms WITHOUT substantial pre-LLM-era ML depth. JD: '\"AI experience\" consists primarily of recent (under 12 months) projects using LangChain to call OpenAI ... unless you can demonstrate substantial pre-LLM-era ML production experience.'"
+    },
+    {
+      "key": "title_chaser",
+      "label": "Title-chaser (job-hops ~every 1.5 yrs for title bumps)",
+      "max_avg_tenure_months": 20,
+      "min_roles": 4,
+      "penalty": 0.80,
+      "rule": "Apply when 4+ roles with average completed tenure < 20 months AND titles escalate (Senior->Staff->Principal pattern). JD: 'optimizing for Senior -> Staff -> Principal titles by switching companies every 1.5 years ... we need someone who plans to be here for 3+ years.' Mild multiplicative penalty."
+    },
+    {
+      "key": "non_technical_role",
+      "label": "Non-engineering current role (keyword-stuffer bait)",
+      "penalty": 0.25,
+      "rule": "Apply when current_title and the whole career trajectory are clearly non-engineering (marketing/HR/accounting/sales/ops/design/civil/mechanical/content/support) regardless of how many AI skills are listed. This is the primary keyword-stuffer trap. Largely overlaps role_coherence; kept as an explicit gate so the reasoning can name it.",
+      "rationale": "Strong penalty (not zero): an Accountant listing 9 AI skills is the canonical trap candidate the sample_submission wrongly ranks #1."
+    }
+  ],
+
+  "behavioral": {
+    "reference_date": "2026-06-06",
+    "active_recent_days": 60,
+    "active_stale_days": 180,
+    "good_response_rate": 0.5,
+    "weak_response_rate": 0.15,
+    "good_interview_completion": 0.6,
+    "notice_preferred_days": 30,
+    "notice_acceptable_days": 90,
+    "modifier_floor": 0.80,
+    "modifier_ceiling": 1.12,
+    "rationale": "Behavioral signals MODIFY (multiplicatively) a fit score; they never drive it. 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. Down-weight them appropriately.' Reward recent activity, high recruiter_response_rate, open_to_work, verified contact, healthy interview_completion_rate; penalize stale/unreachable. Capped so behavior can lift a weak base only modestly (ceiling 1.12) but can meaningfully sink an unreachable one (floor 0.80). Sentinels (-1 github, -1 offer, empty assessments) are COMMON (>60% of pool) and are treated as neutral, never penalized."
+  },
+
+  "ai_core_skills": [
+    "nlp", "embeddings", "information retrieval", "sentence transformers", "hugging face transformers",
+    "faiss", "pinecone", "milvus", "weaviate", "qdrant", "machine learning", "deep learning",
+    "fine-tuning llms", "lora", "rag", "vector search", "recommendation systems", "learning to rank",
+    "xgboost", "lightgbm", "pytorch", "tensorflow", "scikit-learn", "mlops", "mlflow",
+    "feature engineering", "llms", "transformers", "semantic search", "retrieval", "speech recognition",
+    "image classification", "gans", "tts", "statistical modeling"
+  ],
+  "jd_relevant_skills": [
+    "nlp", "embeddings", "information retrieval", "sentence transformers", "hugging face transformers",
+    "faiss", "pinecone", "milvus", "weaviate", "qdrant", "machine learning", "vector search",
+    "recommendation systems", "learning to rank", "xgboost", "lightgbm", "mlops", "mlflow",
+    "feature engineering", "semantic search", "retrieval", "fine-tuning llms", "rag", "transformers"
+  ],
+  "off_target_skills": [
+    "speech recognition", "tts", "asr", "image classification", "object detection", "gans",
+    "ocr", "robotics", "photoshop", "tailwind", "react", "angular", "javascript", "typescript",
+    "marketing", "seo", "project management"
+  ],
+
+  "component_weights": {
+    "_comment": "Base weighted sum of the five components, tuned on the Claude-authored eval set to maximise NDCG@10. role_coherence + career_evidence dominate because they are what separate true fits from keyword-stuffers.",
+    "semantic_fit": 0.22,
+    "role_coherence": 0.26,
+    "career_evidence": 0.24,
+    "experience_fit": 0.10,
+    "trust_skills": 0.18
+  }
+}

lighthouse/__init__.py

@@ -0,0 +1,10 @@
+"""Lighthouse — a recruiter-grade, reasoning-first candidate ranker.
+
+Keyword filters surface the loudest profiles. Lighthouse surfaces the right
+ones — and ignores the fakes that fool keyword filters.
+"""
+
+__version__ = "1.0.0"
+
+# Single source of truth for determinism across the whole pipeline.
+SEED = 1729

lighthouse/features.py

@@ -0,0 +1,232 @@
+"""Deterministic per-candidate feature extraction.
+
+These are the building blocks of the five scoring components that do NOT require
+embeddings (role-taxonomy coherence, career evidence, experience fit, skill
+trust). `semantic_fit` is computed in `scoring.py` from precomputed embeddings.
+
+Every function is pure and deterministic. All text matching is lowercase
+substring matching against term lists in the JD rubric, so the logic is fully
+explainable and traceable back to the JD.
+"""
+from __future__ import annotations
+
+import math
+from datetime import date
+from typing import Dict, List, Optional
+
+from . import loader
+
+PROFICIENCY_WEIGHT = {"beginner": 0.25, "intermediate": 0.5, "advanced": 0.8, "expert": 1.0}
+
+
+# ---------------------------------------------------------------------------
+# text helpers
+# ---------------------------------------------------------------------------
+
+def _norm(s: str) -> str:
+    return (s or "").lower()
+
+
+def count_hits(text: str, terms: List[str]) -> int:
+    t = _norm(text)
+    return sum(1 for term in terms if term in t)
+
+
+def any_hit(text: str, terms: List[str]) -> bool:
+    t = _norm(text)
+    return any(term in t for term in terms)
+
+
+# ---------------------------------------------------------------------------
+# title taxonomy
+# ---------------------------------------------------------------------------
+
+def classify_title(title: str, rubric: dict) -> str:
+    """Return 'strong' | 'positive' | 'negative' | 'neutral' for a job title."""
+    t = _norm(title)
+    if not t:
+        return "neutral"
+    tax = rubric["role_taxonomy"]
+    if any(term in t for term in tax["strong_positive_title_terms"]):
+        return "strong"
+    if any(term in t for term in tax["negative_title_terms"]):
+        return "negative"
+    if any(term in t for term in tax["positive_title_terms"]):
+        return "positive"
+    return "neutral"
+
+
+_TITLE_SCORE = {"strong": 1.0, "positive": 0.7, "neutral": 0.4, "negative": 0.05}
+
+
+def role_coherence_taxonomy(raw: dict, rubric: dict) -> float:
+    """Coherence of titles with an AI/ML/IR/SWE-ranking trajectory, in [0,1].
+
+    Blends the current title (weighted heavily) with the fraction of historical
+    roles that are positive. This is the taxonomy half of `role_coherence`;
+    `scoring.py` blends it with semantic fit.
+    """
+    p = loader.get_profile(raw)
+    cur = classify_title(loader._s(p, "current_title"), rubric)
+    cur_score = _TITLE_SCORE[cur]
+
+    career = loader.get_career(raw)
+    if career:
+        hist_scores = [_TITLE_SCORE[classify_title(h["title"], rubric)] for h in career]
+        hist = sum(hist_scores) / len(hist_scores)
+    else:
+        hist = cur_score
+    return round(0.6 * cur_score + 0.4 * hist, 4)
+
+
+# ---------------------------------------------------------------------------
+# services / product company detection
+# ---------------------------------------------------------------------------
+
+def is_services_company(name: str, rubric: dict) -> bool:
+    n = _norm(name)
+    return any(c in n for c in rubric["hard_negatives"][0]["companies"])
+
+
+def services_fraction(raw: dict, rubric: dict) -> float:
+    career = loader.get_career(raw)
+    if not career:
+        return 0.0
+    s = sum(1 for h in career if is_services_company(h["company"], rubric))
+    return s / len(career)
+
+
+# ---------------------------------------------------------------------------
+# career evidence
+# ---------------------------------------------------------------------------
+
+def career_evidence(raw: dict, rubric: dict) -> float:
+    """Did they actually BUILD ranking/search/recsys/retrieval at product cos? [0,1]
+
+    Per role: term density over the description x a product-vs-services weight.
+    Aggregate emphasises the single best role (JD: 'shipped at least one
+    end-to-end ranking/search/recommendation system'), with a mean term so a
+    consistently-strong career also benefits.
+    """
+    career = loader.get_career(raw)
+    if not career:
+        return 0.0
+    terms = rubric["career_evidence_terms"]
+    role_ev = []
+    for h in career:
+        hits = count_hits(h["description"] + " " + h["title"], terms)
+        density = min(hits / 4.0, 1.0)
+        if is_services_company(h["company"], rubric):
+            product_w = 0.35
+        elif h["company"]:
+            product_w = 1.0
+        else:
+            product_w = 0.85
+        role_ev.append(density * product_w)
+    return round(0.65 * max(role_ev) + 0.35 * (sum(role_ev) / len(role_ev)), 4)
+
+
+# ---------------------------------------------------------------------------
+# experience fit (soft curve)
+# ---------------------------------------------------------------------------
+
+def experience_fit(raw: dict, rubric: dict) -> float:
+    """Soft curve peaking inside ideal [6,8], in-band [5,9], gentle taper. [0,1]"""
+    yoe = loader._f(loader.get_profile(raw), "years_of_experience")
+    e = rubric["experience"]
+    lo, hi = e["ideal_min"], e["ideal_max"]
+    blo, bhi = e["band_min"], e["band_max"]
+    if lo <= yoe <= hi:
+        return 1.0
+    if blo <= yoe < lo:
+        return round(0.85 + 0.15 * (yoe - blo) / (lo - blo), 4)
+    if hi < yoe <= bhi:
+        return round(0.85 + 0.15 * (bhi - yoe) / (bhi - hi), 4)
+    # outside band: decay (don't zero — strong signals can rescue)
+    dist = (blo - yoe) if yoe < blo else (yoe - bhi)
+    return round(max(0.15, 0.7 * math.exp(-dist / 3.0)), 4)
+
+
+# ---------------------------------------------------------------------------
+# skill trust
+# ---------------------------------------------------------------------------
+
+def trust_skills(raw: dict, rubric: dict) -> float:
+    """Skills weighted by proficiency x duration x endorsements x assessment,
+    counting only JD-relevant skills. Saturating. [0,1]
+
+    This is what makes keyword-stuffing structurally worthless: an 'expert'
+    skill with 0 months, no endorsements and a low Redrob assessment contributes
+    almost nothing. Missing assessment is NEUTRAL (76% of pool has none).
+    """
+    skills = loader.get_skills(raw)
+    if not skills:
+        return 0.0
+    sig = loader.get_signals(raw)
+    assess = sig.get("skill_assessment_scores") or {}
+    # build a case-insensitive assessment lookup
+    assess_lc = {str(k).lower(): v for k, v in assess.items() if isinstance(v, (int, float))}
+
+    relevant = set(rubric["jd_relevant_skills"])
+    off = set(rubric["off_target_skills"])
+
+    trust_sum = 0.0
+    for sk in skills:
+        name = _norm(sk["name"])
+        if name in relevant:
+            rel_w = 1.0
+        elif name in off:
+            rel_w = 0.15
+        else:
+            rel_w = 0.4
+        if rel_w < 0.5:
+            continue  # only JD-relevant skills build trust
+
+        prof = PROFICIENCY_WEIGHT.get(sk["proficiency"], 0.4)
+        dur_f = min(sk["duration_months"] / 24.0, 1.0)
+        dur_w = 0.3 + 0.7 * dur_f
+        end_f = min(sk["endorsements"] / 20.0, 1.0)
+        end_w = 0.7 + 0.3 * end_f
+        if name in assess_lc:
+            assess_w = assess_lc[name] / 100.0
+        else:
+            assess_w = 0.65  # neutral when no assessment taken
+
+        trust_sum += prof * dur_w * end_w * assess_w * rel_w
+
+    # saturate: ~3 well-backed relevant skills -> ~0.78
+    return round(1.0 - math.exp(-trust_sum / 2.0), 4)
+
+
+# ---------------------------------------------------------------------------
+# tenure stats (for title-chaser gate + reasoning)
+# ---------------------------------------------------------------------------
+
+def tenure_stats(raw: dict) -> Dict[str, float]:
+    career = loader.get_career(raw)
+    completed = [h["duration_months"] for h in career if not h["is_current"] and h["duration_months"] > 0]
+    n_roles = len(career)
+    avg_tenure = (sum(completed) / len(completed)) if completed else 0.0
+    return {"n_roles": n_roles, "avg_tenure_months": round(avg_tenure, 1)}
+
+
+# ---------------------------------------------------------------------------
+# bundle: all non-embedding features for one candidate
+# ---------------------------------------------------------------------------
+
+def extract_features(raw: dict, rubric: dict) -> dict:
+    """All deterministic (non-embedding) features for one candidate."""
+    p = loader.get_profile(raw)
+    feats = {
+        "candidate_id": loader.candidate_id(raw),
+        "yoe": loader._f(p, "years_of_experience"),
+        "current_title": loader._s(p, "current_title"),
+        "country": loader._s(p, "country"),
+        "role_coherence_tax": role_coherence_taxonomy(raw, rubric),
+        "career_evidence": career_evidence(raw, rubric),
+        "experience_fit": experience_fit(raw, rubric),
+        "trust_skills": trust_skills(raw, rubric),
+        "services_fraction": services_fraction(raw, rubric),
+    }
+    feats.update(tenure_stats(raw))
+    return feats

lighthouse/gates.py

@@ -0,0 +1,146 @@
+"""Hard-negative gates — multiplicative penalties derived from the JD rubric.
+
+Each gate encodes one disqualifier the JD *explicitly* names. Gates are
+multiplicative (not hard zeros — that is reserved for honeypots), so a strong
+candidate with one soft concern is dampened, not annihilated, and the reasoning
+generator can name the concern honestly.
+
+Every gate returns at most one (key, multiplier, reason). `apply_gates` returns
+the product of multipliers and the list of fired reasons.
+"""
+from __future__ import annotations
+
+from datetime import date
+from typing import List, Tuple
+
+from . import features, loader
+
+_MIN_DATE = date(1900, 1, 1)
+
+
+def _career_text(raw: dict) -> str:
+    parts = [loader._s(loader.get_profile(raw), "summary")]
+    for h in loader.get_career(raw):
+        parts.append(h["title"])
+        parts.append(h["description"])
+    parts += [s["name"] for s in loader.get_skills(raw)]
+    return " ".join(parts).lower()
+
+
+def _gate(rubric: dict, key: str) -> dict:
+    return next(g for g in rubric["hard_negatives"] if g["key"] == key)
+
+
+def gate_services_only(raw: dict, rubric: dict, text: str) -> Tuple[float, str]:
+    g = _gate(rubric, "services_only")
+    if features.services_fraction(raw, rubric) >= 0.999 and loader.get_career(raw):
+        cur = loader._s(loader.get_profile(raw), "current_company")
+        return g["penalty"], f"entire career at services/consulting firms (e.g. {cur})"
+    return 1.0, ""
+
+
+def gate_location_visa(raw: dict, rubric: dict, text: str) -> Tuple[float, str]:
+    g = _gate(rubric, "location_visa")
+    p = loader.get_profile(raw)
+    sig = loader.get_signals(raw)
+    country = loader._s(p, "country").strip().lower()
+    if country and country != "india":
+        relocate = bool(sig.get("willing_to_relocate"))
+        loc = loader._s(p, "location") or country.title()
+        if not relocate:
+            return g["penalty"], f"based in {loc} ({country.title()}) and not willing to relocate; no visa sponsorship"
+        return g["soft_penalty"], f"based outside India ({loc}); relocation willing but visa/logistics risk"
+    return 1.0, ""
+
+
+def gate_research_only(raw: dict, rubric: dict, text: str) -> Tuple[float, str]:
+    g = _gate(rubric, "research_only")
+    research = features.count_hits(text, g["positive_terms"])
+    production = features.count_hits(text, g["production_terms"])
+    if research >= 2 and production == 0:
+        return g["penalty"], "research-heavy profile with no evident production deployment"
+    return 1.0, ""
+
+
+def gate_cv_speech_only(raw: dict, rubric: dict, text: str) -> Tuple[float, str]:
+    g = _gate(rubric, "cv_speech_only")
+    domain = features.count_hits(text, g["domain_terms"])
+    nlp_ir = features.count_hits(text, g["nlp_ir_terms"])
+    if domain >= 2 and nlp_ir == 0:
+        return g["penalty"], "primary expertise in computer vision/speech/robotics with no NLP/IR signal"
+    return 1.0, ""
+
+
+def gate_langchain_only_recent(raw: dict, rubric: dict, text: str) -> Tuple[float, str]:
+    g = _gate(rubric, "langchain_only_recent")
+    wrapper = features.count_hits(text, g["wrapper_terms"])
+    depth = features.count_hits(text, g["depth_terms"])
+    if wrapper >= 2 and depth == 0:
+        return g["penalty"], "AI experience appears limited to recent LLM-wrapper tooling without classical ML/retrieval depth"
+    return 1.0, ""
+
+
+def _seniority_level(title: str) -> int:
+    t = title.lower()
+    if any(k in t for k in ("principal", "staff", "director", "vp", "head of", "distinguished")):
+        return 3
+    if "lead" in t or "manager" in t:
+        return 2
+    if "senior" in t or "sr." in t or "sr " in t:
+        return 1
+    return 0
+
+
+def gate_title_chaser(raw: dict, rubric: dict, text: str) -> Tuple[float, str]:
+    """Fire only on genuine title-chasing: short tenures AND an escalating
+    seniority ladder. Lateral moves at short tenure (common for early-career ML
+    engineers, e.g. RecSys -> Search -> NLP) are NOT title-chasing.
+    """
+    g = _gate(rubric, "title_chaser")
+    stats = features.tenure_stats(raw)
+    if not (stats["n_roles"] >= g["min_roles"] and 0 < stats["avg_tenure_months"] < g["max_avg_tenure_months"]):
+        return 1.0, ""
+    # titles in chronological (oldest-first) order
+    career = sorted(loader.get_career(raw), key=lambda h: (h["start_date"] or _MIN_DATE))
+    levels = [_seniority_level(h["title"]) for h in career]
+    escalated = levels and (max(levels) - min(levels) >= 2) and levels[-1] >= levels[0] and levels[-1] >= 2
+    if escalated:
+        return g["penalty"], f"escalating titles while job-hopping every ~{stats['avg_tenure_months']:.0f} months across {stats['n_roles']} roles (title-chaser pattern)"
+    return 1.0, ""
+
+
+def gate_non_technical_role(raw: dict, rubric: dict, text: str) -> Tuple[float, str]:
+    g = _gate(rubric, "non_technical_role")
+    p = loader.get_profile(raw)
+    cur_class = features.classify_title(loader._s(p, "current_title"), rubric)
+    if cur_class == "negative":
+        # whole trajectory non-technical? (no strong AI role anywhere)
+        career = loader.get_career(raw)
+        has_strong = any(features.classify_title(h["title"], rubric) == "strong" for h in career)
+        if not has_strong:
+            return g["penalty"], f"current role '{loader._s(p,'current_title')}' is non-engineering with no AI/ML role in career history"
+    return 1.0, ""
+
+
+GATES = [
+    gate_non_technical_role,
+    gate_services_only,
+    gate_location_visa,
+    gate_research_only,
+    gate_cv_speech_only,
+    gate_langchain_only_recent,
+    gate_title_chaser,
+]
+
+
+def apply_gates(raw: dict, rubric: dict) -> Tuple[float, List[str]]:
+    """Return (combined_multiplier, [fired reasons])."""
+    text = _career_text(raw)
+    mult = 1.0
+    reasons: List[str] = []
+    for fn in GATES:
+        m, reason = fn(raw, rubric, text)
+        if m < 1.0:
+            mult *= m
+            reasons.append(reason)
+    return mult, reasons

lighthouse/honeypot.py

@@ -0,0 +1,94 @@
+"""Honeypot / anomaly detection — explainable, rule-based.
+
+The dataset seeds ~80 honeypots: "subtly impossible" profiles (tenure longer
+than the company has existed; 'expert' in many skills with 0 months used; total
+skill-months far exceeding career length; dates that don't add up). The spec
+forces them to relevance tier 0 and disqualifies submissions with >10% of them
+in the top 100. `rank.py` zeroes any candidate flagged here.
+
+Design notes grounded in EDA over the real 100K:
+  * The naive "sum(skill_months) > years_of_experience*12" check fires on 63% of
+    the pool (skills are used concurrently), so it is USELESS as written. We only
+    flag a *single* skill whose duration exceeds the whole career — that is
+    genuinely impossible, not just overlapping.
+  * Every rule returns a human-readable reason so flags are auditable and feed
+    the reasoning generator.
+
+Reference "now" is the JD reference date (2026-06-06).
+"""
+from __future__ import annotations
+
+from datetime import date
+from typing import List, Tuple
+
+from . import loader
+
+REFERENCE_DATE = date(2026, 6, 6)
+
+
+def _months_between(d0: date, d1: date) -> int:
+    return (d1.year - d0.year) * 12 + (d1.month - d0.month)
+
+
+def detect(raw: dict) -> Tuple[bool, List[str]]:
+    """Return (is_honeypot, reasons). Any single hard-impossible rule flags."""
+    reasons: List[str] = []
+    p = loader.get_profile(raw)
+    yoe = loader._f(p, "years_of_experience")
+    career = loader.get_career(raw)
+    skills = loader.get_skills(raw)
+    edu = loader.get_education(raw)
+
+    # --- 1. claimed expertise with zero usage ---
+    expert_zero = [s["name"] for s in skills
+                   if s["proficiency"] in ("advanced", "expert") and s["duration_months"] == 0]
+    if len(expert_zero) >= 3:
+        reasons.append(
+            f"{len(expert_zero)} skills claimed advanced/expert with 0 months used "
+            f"(e.g. {', '.join(expert_zero[:3])})")
+
+    # NOTE: a tempting rule — "a skill used more months than the whole career" —
+    # was REMOVED after EDA: skill durations routinely exceed current-job YOE
+    # (skills are learned before/outside a role), so that rule flagged 9% of the
+    # pool. Genuine impossibility lives in the date math and expertise claims,
+    # not in skill-vs-YOE.
+
+    # --- 2 & 3. per-role date integrity ---
+    total_role_months = 0
+    for h in career:
+        sd, ed = h["start_date"], h["end_date"]
+        dm = h["duration_months"]
+        total_role_months += dm
+        if sd and sd > REFERENCE_DATE:
+            reasons.append(f"role at {h['company']} starts in the future ({sd.isoformat()})")
+        if sd and ed and ed < sd:
+            reasons.append(f"role at {h['company']} ends before it starts")
+        if sd and ed:
+            span = _months_between(sd, ed)
+            if span >= 0 and abs(span - dm) > 9:
+                reasons.append(
+                    f"role at {h['company']}: stated {dm}mo vs {span}mo implied by dates")
+
+    # --- 5. total tenure overflows the career length ---
+    if yoe > 0 and total_role_months > yoe * 12 * 1.6 + 18:
+        reasons.append(
+            f"career roles sum to {total_role_months}mo, impossible for {yoe:.1f} yrs experience")
+
+    # --- 6. tenure longer than a young company could exist (proxy) ---
+    # A single role of >15 yrs is implausible in this pool's company ages.
+    for h in career:
+        if h["duration_months"] > 200:
+            reasons.append(f"implausible {h['duration_months']}mo tenure at {h['company']}")
+            break
+
+    # --- 7. education dates invalid ---
+    for e in edu:
+        if e["start_year"] and e["end_year"] and e["end_year"] < e["start_year"]:
+            reasons.append(
+                f"education '{e['degree']}' ends ({e['end_year']}) before it starts ({e['start_year']})")
+
+    return (len(reasons) > 0, reasons)
+
+
+def is_honeypot(raw: dict) -> bool:
+    return detect(raw)[0]

lighthouse/loader.py

@@ -0,0 +1,196 @@
+"""Schema-tolerant loading of Redrob candidate records.
+
+The challenge file is a 100K-line JSONL (~487 MB). We stream it line-by-line so
+we never hold more than one raw record in memory at a time during parsing.
+
+Every accessor here is defensive: optional fields may be missing, lists may be
+empty, and the dataset uses sentinels (`github_activity_score == -1`,
+`offer_acceptance_rate == -1`, empty `skill_assessment_scores`). Nothing in this
+module should ever raise on a well-formed-but-sparse profile.
+"""
+from __future__ import annotations
+
+import json
+from datetime import date, datetime
+from typing import Any, Dict, Iterator, List, Optional
+
+# ---------------------------------------------------------------------------
+# Low-level safe accessors
+# ---------------------------------------------------------------------------
+
+def _s(d: Optional[dict], key: str, default: str = "") -> str:
+    if not isinstance(d, dict):
+        return default
+    v = d.get(key, default)
+    return v if isinstance(v, str) else (default if v is None else str(v))
+
+
+def _f(d: Optional[dict], key: str, default: float = 0.0) -> float:
+    if not isinstance(d, dict):
+        return default
+    v = d.get(key, default)
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return default
+
+
+def _i(d: Optional[dict], key: str, default: int = 0) -> int:
+    if not isinstance(d, dict):
+        return default
+    v = d.get(key, default)
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return default
+
+
+def _b(d: Optional[dict], key: str, default: bool = False) -> bool:
+    if not isinstance(d, dict):
+        return default
+    v = d.get(key, default)
+    return bool(v) if isinstance(v, bool) else default
+
+
+def parse_date(s: Any) -> Optional[date]:
+    """Parse an ISO date string; return None for missing/invalid/null."""
+    if not s or not isinstance(s, str):
+        return None
+    try:
+        return datetime.strptime(s[:10], "%Y-%m-%d").date()
+    except (ValueError, TypeError):
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Streaming reader
+# ---------------------------------------------------------------------------
+
+def iter_raw(path: str) -> Iterator[dict]:
+    """Yield raw candidate dicts from a JSONL file, one line at a time."""
+    with open(path, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                yield json.loads(line)
+            except json.JSONDecodeError:
+                # Skip malformed lines rather than aborting the whole run.
+                continue
+
+
+def load_all(path: str) -> List[dict]:
+    """Load every raw candidate record into a list."""
+    return list(iter_raw(path))
+
+
+# ---------------------------------------------------------------------------
+# Normalisation
+# ---------------------------------------------------------------------------
+
+def get_skills(raw: dict) -> List[dict]:
+    skills = raw.get("skills") or []
+    out = []
+    for sk in skills:
+        if not isinstance(sk, dict):
+            continue
+        out.append({
+            "name": _s(sk, "name"),
+            "proficiency": _s(sk, "proficiency", "beginner").lower(),
+            "endorsements": _i(sk, "endorsements"),
+            "duration_months": _i(sk, "duration_months"),
+        })
+    return out
+
+
+def get_career(raw: dict) -> List[dict]:
+    hist = raw.get("career_history") or []
+    out = []
+    for h in hist:
+        if not isinstance(h, dict):
+            continue
+        out.append({
+            "company": _s(h, "company"),
+            "title": _s(h, "title"),
+            "start_date": parse_date(h.get("start_date")),
+            "end_date": parse_date(h.get("end_date")),
+            "duration_months": _i(h, "duration_months"),
+            "is_current": _b(h, "is_current"),
+            "industry": _s(h, "industry"),
+            "company_size": _s(h, "company_size"),
+            "description": _s(h, "description"),
+        })
+    return out
+
+
+def get_education(raw: dict) -> List[dict]:
+    edu = raw.get("education") or []
+    out = []
+    for e in edu:
+        if not isinstance(e, dict):
+            continue
+        out.append({
+            "institution": _s(e, "institution"),
+            "degree": _s(e, "degree"),
+            "field_of_study": _s(e, "field_of_study"),
+            "start_year": _i(e, "start_year"),
+            "end_year": _i(e, "end_year"),
+            "tier": _s(e, "tier", "unknown"),
+        })
+    return out
+
+
+def get_signals(raw: dict) -> dict:
+    """Return the redrob_signals object (raw dict, with safe fallback)."""
+    sig = raw.get("redrob_signals")
+    return sig if isinstance(sig, dict) else {}
+
+
+def get_profile(raw: dict) -> dict:
+    p = raw.get("profile")
+    return p if isinstance(p, dict) else {}
+
+
+def candidate_id(raw: dict) -> str:
+    return _s(raw, "candidate_id")
+
+
+# ---------------------------------------------------------------------------
+# Canonical text blob (used by both embeddings and BM25)
+# ---------------------------------------------------------------------------
+
+def build_text_blob(raw: dict) -> str:
+    """Build one clean text blob per candidate for semantic + lexical matching.
+
+    Order matters for readability but not for embeddings: headline + summary
+    first (densest signal), then each role's title/description, then skill names.
+    We deliberately repeat titles/descriptions verbatim — they are the strongest
+    evidence of what the person actually *did*.
+    """
+    p = get_profile(raw)
+    parts: List[str] = []
+
+    headline = _s(p, "headline")
+    if headline:
+        parts.append(headline)
+
+    summary = _s(p, "summary")
+    if summary:
+        parts.append(summary)
+
+    cur_title = _s(p, "current_title")
+    cur_co = _s(p, "current_company")
+    if cur_title:
+        parts.append(f"{cur_title} at {cur_co}".strip(" at"))
+
+    for h in get_career(raw):
+        seg = f"{h['title']} at {h['company']} ({h['industry']}). {h['description']}".strip()
+        if seg:
+            parts.append(seg)
+
+    skill_names = [sk["name"] for sk in get_skills(raw) if sk["name"]]
+    if skill_names:
+        parts.append("Skills: " + ", ".join(skill_names))
+
+    return "\n".join(parts).strip()

lighthouse/metrics.py

@@ -0,0 +1,65 @@
+"""Ranking metrics: NDCG@k, MAP, P@k — matching the challenge spec.
+
+The composite the organizers use:
+    composite = 0.50*NDCG@10 + 0.30*NDCG@50 + 0.15*MAP + 0.05*P@10
+
+Relevance tiers are 0-5. Conventions used here (documented in eval/results.md):
+  * NDCG uses exponential gain  (2^rel - 1) / log2(rank+1)  — the common form.
+  * "relevant" for P@k and MAP means tier >= 3 (the spec: 'P@10 counts tier-3+
+    as relevant'); we apply the same threshold to MAP for consistency.
+All functions take `rels`, the true relevance tiers in the system's ranked order.
+"""
+from __future__ import annotations
+
+import math
+from typing import List
+
+RELEVANT_TIER = 3
+
+
+def dcg_at_k(rels: List[float], k: int) -> float:
+    return sum((2 ** r - 1) / math.log2(i + 2) for i, r in enumerate(rels[:k]))
+
+
+def ndcg_at_k(rels: List[float], k: int) -> float:
+    ideal = sorted(rels, reverse=True)
+    idcg = dcg_at_k(ideal, k)
+    if idcg == 0:
+        return 0.0
+    return dcg_at_k(rels, k) / idcg
+
+
+def precision_at_k(rels: List[float], k: int, threshold: int = RELEVANT_TIER) -> float:
+    if k == 0:
+        return 0.0
+    topk = rels[:k]
+    return sum(1 for r in topk if r >= threshold) / k
+
+
+def average_precision(rels: List[float], threshold: int = RELEVANT_TIER) -> float:
+    """AP over the ranked list; R = total relevant in the list."""
+    n_rel = sum(1 for r in rels if r >= threshold)
+    if n_rel == 0:
+        return 0.0
+    hits = 0
+    score = 0.0
+    for i, r in enumerate(rels, start=1):
+        if r >= threshold:
+            hits += 1
+            score += hits / i
+    return score / n_rel
+
+
+def evaluate_ranking(rels: List[float]) -> dict:
+    """All metrics + the spec composite for one ranked list of true tiers."""
+    ndcg10 = ndcg_at_k(rels, 10)
+    ndcg50 = ndcg_at_k(rels, 50)
+    mapv = average_precision(rels)
+    p10 = precision_at_k(rels, 10)
+    composite = 0.50 * ndcg10 + 0.30 * ndcg50 + 0.15 * mapv + 0.05 * p10
+    return {
+        "NDCG@10": round(ndcg10, 4), "NDCG@50": round(ndcg50, 4),
+        "MAP": round(mapv, 4), "P@10": round(p10, 4),
+        "P@5": round(precision_at_k(rels, 5), 4),
+        "composite": round(composite, 4),
+    }

lighthouse/ranker.py

@@ -0,0 +1,94 @@
+"""Rank-time engine: load artifacts, score every candidate, sort, take top-N.
+
+Pure numpy/pandas/Python. No network, no model inference over the full pool
+(embeddings are precomputed). The only place a model may load is the small-sample
+fallback (`_embeddings_for`) used by the HuggingFace app when it is handed
+candidates that were never precomputed — never on the 100K official path.
+"""
+from __future__ import annotations
+
+import json
+import os
+from typing import Dict, List, Optional, Tuple
+
+import numpy as np
+
+from . import loader, scoring
+
+
+# ---------------------------------------------------------------------------
+# artifact loading
+# ---------------------------------------------------------------------------
+
+def load_artifacts(art_dir: str) -> dict:
+    rubric = json.load(open(os.path.join(art_dir, "jd_rubric.json"), encoding="utf-8"))
+    ids = json.load(open(os.path.join(art_dir, "candidate_ids.json"), encoding="utf-8"))
+    cand_emb = np.load(os.path.join(art_dir, "cand_emb.npy"))
+    facet_emb = np.load(os.path.join(art_dir, "jd_facet_emb.npy"))
+    id_to_row = {cid: i for i, cid in enumerate(ids)}
+    return {"rubric": rubric, "ids": ids, "cand_emb": cand_emb,
+            "facet_emb": facet_emb, "id_to_row": id_to_row}
+
+
+def _embeddings_for(raws: List[dict], art: dict, model_name: Optional[str]) -> np.ndarray:
+    """Return an embedding matrix aligned to `raws`, using precomputed vectors
+    where available and encoding any missing ones on the fly (small-sample only).
+    """
+    cand_emb = art["cand_emb"]
+    dim = cand_emb.shape[1]
+    out = np.zeros((len(raws), dim), dtype=np.float32)
+    missing_idx, missing_blobs = [], []
+    for i, raw in enumerate(raws):
+        row = art["id_to_row"].get(loader.candidate_id(raw))
+        if row is not None:
+            out[i] = cand_emb[row].astype(np.float32)
+        else:
+            missing_idx.append(i)
+            missing_blobs.append(loader.build_text_blob(raw))
+    if missing_blobs:
+        # small-sample fallback (e.g. HF app) — load the small encoder lazily
+        from sentence_transformers import SentenceTransformer
+        meta_path = os.path.join("artifacts", "precompute_meta.json")
+        mn = model_name or (json.load(open(meta_path))["model"]
+                            if os.path.exists(meta_path) else "BAAI/bge-small-en-v1.5")
+        model = SentenceTransformer(mn, device="cpu")
+        vecs = model.encode(missing_blobs, normalize_embeddings=True, convert_to_numpy=True)
+        for j, i in enumerate(missing_idx):
+            out[i] = vecs[j]
+    return out
+
+
+# ---------------------------------------------------------------------------
+# scoring all candidates
+# ---------------------------------------------------------------------------
+
+def score_all(raws: List[dict], art: dict, drop: str = None,
+              model_name: Optional[str] = None, use_gates: bool = True,
+              use_honeypot: bool = True, use_behavior: bool = True) -> List[dict]:
+    rubric = art["rubric"]
+    emb = _embeddings_for(raws, art, model_name)
+    sem_raw = scoring.raw_semantic_fit(emb, art["facet_emb"])
+    sem_norm = scoring.normalize_semantic(sem_raw)
+    records = []
+    for raw, sf in zip(raws, sem_norm):
+        records.append(scoring.score_candidate(
+            raw, rubric, float(sf), drop=drop, use_gates=use_gates,
+            use_honeypot=use_honeypot, use_behavior=use_behavior))
+    return records
+
+
+# ---------------------------------------------------------------------------
+# ranking + ordering (spec-compliant)
+# ---------------------------------------------------------------------------
+
+def rank_records(records: List[dict], top: int = 100) -> List[dict]:
+    """Sort by final_score desc, tie-break candidate_id ascending; take top-N.
+
+    Guarantees the validator's constraints: scores non-increasing by rank and
+    equal scores ordered by candidate_id ascending.
+    """
+    ordered = sorted(records, key=lambda r: (-r["final_score"], r["candidate_id"]))
+    top_recs = ordered[:top]
+    for i, r in enumerate(top_recs):
+        r["rank"] = i + 1
+    return top_recs

lighthouse/reasoning.py

@@ -0,0 +1,169 @@
+"""Deterministic, fact-grounded reasoning generator (NO LLM at rank-time).
+
+For each top-100 candidate we compose a 1-2 sentence justification that:
+  * cites only REAL field values (years, title, named skills, signal numbers),
+  * connects them to specific JD requirements,
+  * honestly names gaps (gate concerns, behavioral risks),
+  * varies by which factors dominate THIS candidate, and
+  * matches its rank band in tone (confident high, "filler" honesty low).
+
+Grounding guarantee: every skill or employer surfaced is pulled directly from
+the candidate's own `skills` / `career_history` / `profile`, never invented.
+`grounded_terms()` exposes exactly what was used so tests can verify it.
+"""
+from __future__ import annotations
+
+from datetime import date
+from typing import Dict, List, Tuple
+
+from . import features, loader
+
+REFERENCE_DATE = date(2026, 6, 6)
+
+# representative "build" words; only surfaced if present in the candidate's text
+_EVIDENCE_WORDS = [
+    ("ranking", "ranking"), ("learning to rank", "learning-to-rank"),
+    ("recommendation", "recommendation"), ("recommender", "recommendation"),
+    ("retrieval", "retrieval"), ("search", "search"), ("embedding", "embeddings"),
+    ("semantic search", "semantic search"), ("personalization", "personalization"),
+    ("information retrieval", "information retrieval"),
+]
+
+
+def _fmt_yoe(y: float) -> str:
+    return f"{y:.1f}".rstrip("0").rstrip(".") + " yrs"
+
+
+def extract_facts(raw: dict, rubric: dict, record: dict) -> dict:
+    """Pull grounded facts used to compose reasoning."""
+    p = loader.get_profile(raw)
+    sig = loader.get_signals(raw)
+    skills = loader.get_skills(raw)
+    career = loader.get_career(raw)
+
+    # JD-relevant skills the candidate actually has, strongest first
+    relevant = set(rubric["jd_relevant_skills"])
+    rel_present = [s for s in skills if s["name"].lower() in relevant]
+    rel_present.sort(key=lambda s: (features.PROFICIENCY_WEIGHT.get(s["proficiency"], 0.4)
+                                    * min(s["duration_months"] / 24.0, 1.0)), reverse=True)
+    top_skills = [s["name"] for s in rel_present[:3]]
+
+    # product (non-services) employers actually in the career history
+    product_cos = []
+    for h in career:
+        if h["company"] and not features.is_services_company(h["company"], rubric):
+            if h["company"] not in product_cos:
+                product_cos.append(h["company"])
+
+    # which "build" word is genuinely present in their text
+    text = " ".join([loader._s(p, "summary")] + [h["description"] + " " + h["title"] for h in career]).lower()
+    evidence_word = None
+    for needle, label in _EVIDENCE_WORDS:
+        if needle in text:
+            evidence_word = label
+            break
+
+    la = loader.parse_date(sig.get("last_active_date"))
+    days_active = (REFERENCE_DATE - la).days if la else None
+
+    return {
+        "yoe": loader._f(p, "years_of_experience"),
+        "title": loader._s(p, "current_title"),
+        "company": loader._s(p, "current_company"),
+        "country": loader._s(p, "country"),
+        "location": loader._s(p, "location"),
+        "top_skills": top_skills,
+        "product_cos": product_cos[:2],
+        "evidence_word": evidence_word,
+        "resp_rate": sig.get("recruiter_response_rate"),
+        "days_active": days_active,
+        "notice": sig.get("notice_period_days"),
+        "relocate": sig.get("willing_to_relocate"),
+        "n_relevant_skills": len(rel_present),
+    }
+
+
+def grounded_terms(raw: dict, rubric: dict, record: dict) -> Dict[str, List[str]]:
+    """The skills/companies the generator is allowed to mention (for tests)."""
+    f = extract_facts(raw, rubric, record)
+    return {"skills": f["top_skills"], "companies": f["product_cos"]}
+
+
+def _positive_clause(f: dict, comps: dict, band: str) -> str:
+    """Lead clause built from the candidate's strongest grounded evidence."""
+    title = f["title"] or "Engineer"
+    yoe = _fmt_yoe(f["yoe"])
+    lead = f"{title} with {yoe}"
+
+    # choose the dominant evidence to feature
+    bits = []
+    if f["evidence_word"] and comps["career_evidence"] >= 0.45:
+        cos = f" at {', '.join(f['product_cos'])}" if f["product_cos"] else " at product companies"
+        bits.append(f"built {f['evidence_word']} systems{cos}")
+    elif f["product_cos"] and comps["role_coherence"] >= 0.6:
+        bits.append(f"applied-ML track record at {', '.join(f['product_cos'])}")
+
+    if f["top_skills"]:
+        if len(f["top_skills"]) == 1:
+            bits.append(f"hands-on with {f['top_skills'][0]}")
+        else:
+            bits.append(f"hands-on with {', '.join(f['top_skills'][:2])}")
+
+    if not bits:
+        # nothing strong to feature — keep it honest
+        if comps["semantic_fit"] >= 0.5:
+            bits.append("profile semantically adjacent to the retrieval/ranking mandate")
+        else:
+            bits.append("limited direct evidence of production retrieval/ranking work")
+
+    body = "; ".join(bits)
+    if band == "top":
+        return f"{lead} — {body}."
+    if band == "mid":
+        return f"{lead}; {body}."
+    # capitalize only the first letter (preserve proper-noun casing in company/skill names)
+    body_cap = body[0].upper() + body[1:] if body else body
+    return f"{lead}. {body_cap}."
+
+
+def _concern_clause(f: dict, record: dict, band: str) -> str:
+    """Honest concern/gap clause. Prefers the most material concern."""
+    concerns = list(record.get("gate_reasons", []))
+    concerns += [c for c in record.get("behavior_facts", [])
+                 if any(k in c for k in ("low", "inactive", "notice", "last active"))]
+
+    if record.get("honeypot"):
+        return f" Flagged as anomalous: {record['honeypot_reasons'][0]}."
+
+    if concerns:
+        return " Concern: " + concerns[0] + "."
+
+    # no concern -> reinforce with a real positive behavioral signal
+    pos = []
+    if isinstance(f["resp_rate"], (int, float)) and f["resp_rate"] >= 0.5:
+        pos.append(f"{f['resp_rate']:.0%} recruiter response")
+    if f["days_active"] is not None and f["days_active"] <= 60:
+        pos.append(f"active {f['days_active']}d ago")
+    if f["country"].lower() == "india":
+        pos.append("India-based")
+    elif f["relocate"]:
+        pos.append("willing to relocate")
+    if pos:
+        s = ", ".join(pos)
+        return " " + s[0].upper() + s[1:] + "."
+    return ""
+
+
+def generate(raw: dict, rubric: dict, record: dict) -> str:
+    """Compose the 1-2 sentence grounded reasoning for one ranked candidate."""
+    rank = record.get("rank", 999)
+    band = "top" if rank <= 10 else ("mid" if rank <= 50 else "low")
+    f = extract_facts(raw, rubric, record)
+    comps = record["components"]
+
+    text = _positive_clause(f, comps, band) + _concern_clause(f, record, band)
+
+    if band == "low" and not record.get("gate_reasons") and not record.get("honeypot"):
+        text += " Adjacent fit included near the cutoff."
+
+    return " ".join(text.split()).strip()

lighthouse/scoring.py

@@ -0,0 +1,175 @@
+"""The Lighthouse scoring model — the core IP.
+
+Five explainable components in [0,1] are combined into a base score, then
+modified multiplicatively:
+
+    final = base_weighted_sum  ×  Π(hard-negative gates)  ×  behavioral_modifier
+    final = 0                  if honeypot/anomaly detected
+
+| component       | what it measures                                            | anti-trap role            |
+|-----------------|-------------------------------------------------------------|---------------------------|
+| semantic_fit    | cosine of candidate embedding vs JD facets                  | plain-language Tier-5s    |
+| role_coherence  | is the title/trajectory actually AI/ML/IR/SWE-ranking?      | decisive vs keyword-stuff |
+| career_evidence | did they BUILD ranking/search/recsys at product companies?  | rewards real builders     |
+| experience_fit  | soft curve peaking 6-8 yrs, in-band 5-9                     | matches the JD band       |
+| trust_skills    | skills weighted by proficiency×duration×endorse×assessment  | kills keyword stuffing    |
+
+`semantic_fit` is supplied pre-normalized (population percentile-scaled) by the
+caller; everything else is computed here from the rubric + raw fields.
+"""
+from __future__ import annotations
+
+from datetime import date
+from typing import Dict, List, Tuple
+
+import numpy as np
+
+from . import features, gates, honeypot, loader
+
+REFERENCE_DATE = date(2026, 6, 6)
+
+
+# ---------------------------------------------------------------------------
+# semantic fit (embedding side)
+# ---------------------------------------------------------------------------
+
+def raw_semantic_fit(cand_emb: np.ndarray, facet_emb: np.ndarray) -> np.ndarray:
+    """Per-candidate raw semantic fit = 0.6*max + 0.4*mean facet cosine.
+
+    Both matrices are L2-normalized, so the dot product is cosine. Returns a
+    1-D array aligned to cand_emb rows. (max rewards a strong single-facet
+    match; mean rewards broad alignment.)
+    """
+    cos = cand_emb.astype(np.float32) @ facet_emb.T          # (N, F)
+    return 0.6 * cos.max(axis=1) + 0.4 * cos.mean(axis=1)
+
+
+def normalize_semantic(raw: np.ndarray) -> np.ndarray:
+    """Percentile-clip raw semantic fit into [0,1] across the population."""
+    if len(raw) == 0:
+        return raw
+    lo = np.percentile(raw, 5)
+    hi = np.percentile(raw, 95)
+    if hi - lo < 1e-6:
+        return np.clip((raw - raw.min()) / (raw.ptp() + 1e-6), 0, 1)
+    return np.clip((raw - lo) / (hi - lo), 0.0, 1.0)
+
+
+# ---------------------------------------------------------------------------
+# behavioral modifier (multiplicative, clamped)
+# ---------------------------------------------------------------------------
+
+def behavioral_modifier(raw: dict, rubric: dict) -> Tuple[float, List[str]]:
+    """Reward reachable/active candidates, penalize stale/unreachable. [floor,ceiling].
+
+    Sentinels (-1 github, -1 offer, empty assessments) are NEUTRAL — only
+    present signals move the modifier. Behavior modifies; it never drives.
+    """
+    b = rubric["behavioral"]
+    sig = loader.get_signals(raw)
+    delta = 0.0
+    facts: List[str] = []
+
+    la = loader.parse_date(sig.get("last_active_date"))
+    if la:
+        days = (REFERENCE_DATE - la).days
+        if days <= b["active_recent_days"]:
+            delta += 0.05; facts.append(f"active {days}d ago")
+        elif days >= b["active_stale_days"]:
+            delta -= 0.15; facts.append(f"inactive {days}d")
+        elif days >= 120:
+            delta -= 0.07; facts.append(f"last active {days}d ago")
+
+    rr = sig.get("recruiter_response_rate")
+    if isinstance(rr, (int, float)):
+        if rr >= b["good_response_rate"]:
+            delta += 0.04; facts.append(f"{rr:.0%} recruiter response")
+        elif rr <= b["weak_response_rate"]:
+            delta -= 0.12; facts.append(f"low {rr:.0%} recruiter response")
+        elif rr <= 0.3:
+            delta -= 0.05
+
+    if sig.get("open_to_work_flag"):
+        delta += 0.02
+    if sig.get("verified_email") and sig.get("verified_phone"):
+        delta += 0.02
+
+    ic = sig.get("interview_completion_rate")
+    if isinstance(ic, (int, float)) and ic > 0:
+        if ic >= b["good_interview_completion"]:
+            delta += 0.02
+        elif ic < 0.3:
+            delta -= 0.06; facts.append(f"low {ic:.0%} interview completion")
+
+    notice = sig.get("notice_period_days")
+    if isinstance(notice, (int, float)):
+        if notice <= b["notice_preferred_days"]:
+            delta += 0.02
+        elif notice > b["notice_acceptable_days"]:
+            delta -= 0.05; facts.append(f"{int(notice)}-day notice")
+
+    mult = float(np.clip(1.0 + delta, b["modifier_floor"], b["modifier_ceiling"]))
+    return round(mult, 4), facts
+
+
+# ---------------------------------------------------------------------------
+# component assembly + final score
+# ---------------------------------------------------------------------------
+
+def components(raw: dict, rubric: dict, semantic_fit: float) -> Dict[str, float]:
+    tax = features.role_coherence_taxonomy(raw, rubric)
+    # role_coherence blends taxonomy (dominant) with semantic fit
+    role_coherence = round(0.7 * tax + 0.3 * semantic_fit, 4)
+    return {
+        "semantic_fit": round(float(semantic_fit), 4),
+        "role_coherence": role_coherence,
+        "career_evidence": features.career_evidence(raw, rubric),
+        "experience_fit": features.experience_fit(raw, rubric),
+        "trust_skills": features.trust_skills(raw, rubric),
+    }
+
+
+def base_score(comps: Dict[str, float], rubric: dict, drop: str = None) -> float:
+    """Weighted sum of components. `drop` ablates one component (re-normalising)."""
+    w = dict(rubric["component_weights"])
+    w.pop("_comment", None)
+    if drop and drop in w:
+        w.pop(drop)
+    total_w = sum(w.values())
+    s = sum(comps[k] * wv for k, wv in w.items())
+    return s / total_w if total_w else 0.0
+
+
+def score_candidate(raw: dict, rubric: dict, semantic_fit: float, drop: str = None,
+                    use_gates: bool = True, use_honeypot: bool = True,
+                    use_behavior: bool = True) -> dict:
+    """Full scoring record for one candidate (feeds ranking + reasoning).
+
+    The use_* flags exist for the ablation study in eval/ (e.g. measure NDCG
+    with the gates or honeypot filter switched off)."""
+    comps = components(raw, rubric, semantic_fit)
+    base = base_score(comps, rubric, drop=drop)
+
+    hp, hp_reasons = honeypot.detect(raw)
+    gate_mult, gate_reasons = gates.apply_gates(raw, rubric)
+    beh_mult, beh_facts = behavioral_modifier(raw, rubric)
+
+    eff_gate = gate_mult if use_gates else 1.0
+    eff_beh = beh_mult if use_behavior else 1.0
+    if hp and use_honeypot:
+        final = 0.0
+    else:
+        final = base * eff_gate * eff_beh
+
+    return {
+        "candidate_id": loader.candidate_id(raw),
+        "components": comps,
+        "base": round(base, 4),
+        "gate_mult": gate_mult,
+        "gate_reasons": gate_reasons,
+        "behavior_mult": beh_mult,
+        "behavior_facts": beh_facts,
+        "honeypot": hp,
+        "honeypot_reasons": hp_reasons,
+        "final_score": round(float(final), 6),
+    }

requirements.txt

@@ -0,0 +1,7 @@
+# HuggingFace Space (Streamlit) dependencies for the Lighthouse sandbox.
+streamlit==1.40.2
+sentence-transformers==3.3.1
+torch==2.5.1
+numpy==2.2.6
+pandas==2.3.3
+PyYAML==6.0.2