#!/usr/bin/env python3 """ rank.py — V7 Winning-Level Intelligent Recruitment Engine TIMED STEP (CPU-only, no network, <5 minutes, <=16GB RAM). V7 Pipeline: 1. Auto-run precompute if artifact missing 2. Load Parquet features 3. Compute intent-aware elite composite (V7 scoring) 4. Apply safety & consistency layers 5. Take top 100 6. Generate evidence-graph-powered reasoning 7. Write CSV 8. Validate CSV format V7 changes: - Fixed precompute.py integration (v6.1 was missing this file) - Fixed scoring.py tiebreaker bug - NDCG@10-optimized weights - V7 new features (assessment, endorsement, education, cross-validation) - Built-in submission validation - Candidates loaded from JSONL (one JSON per line, no CSV header) - Reproducible single-command interface """ from __future__ import annotations import argparse import csv import json import subprocess import sys import time from pathlib import Path import numpy as np import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent)) from lib import scoring, reasoning TOP_N = 100 def load_jsonl(path: str) -> list[dict]: """Load candidates from JSONL file (one JSON object per line).""" import gzip candidates = [] opener = gzip.open if path.endswith(".gz") else open with opener(path, "rt", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue # Skip header lines if line.startswith("candidate_id,") or line.startswith("#"): continue try: candidates.append(json.loads(line)) except json.JSONDecodeError: # Might be CSV with JSON in first column if "," in line: first_field = line.split(",", 1)[0] if first_field.startswith("{"): try: candidates.append(json.loads(first_field)) except json.JSONDecodeError: pass return candidates def validate_output(csv_path: str) -> list[str]: """Validate submission CSV against challenge rules.""" errors = [] path = Path(csv_path) if path.suffix.lower() != ".csv": errors.append(f"File extension must be .csv, got {path.suffix}") try: with open(path, "r", encoding="utf-8", newline="") as f: reader = csv.reader(f) header = next(reader, None) if header != ["candidate_id", "rank", "score", "reasoning"]: errors.append(f"Header mismatch: {header}") data_rows = [] for row in reader: if any(cell.strip() for cell in row): data_rows.append(row) except Exception as e: errors.append(f"Read error: {e}") return errors n = len(data_rows) if n != 100: errors.append(f"Expected 100 data rows, found {n}") seen_ids = set() seen_ranks = set() by_rank = [] for i, cells in enumerate(data_rows): row_num = i + 2 if len(cells) != 4: errors.append(f"Row {row_num}: expected 4 columns, got {len(cells)}") continue cid, rank_s, score_s, reasoning = cells if not cid or not cid.startswith("CAND_"): errors.append(f"Row {row_num}: invalid candidate_id '{cid}'") elif cid in seen_ids: errors.append(f"Row {row_num}: duplicate candidate_id '{cid}'") else: seen_ids.add(cid) try: rank = int(rank_s) if not 1 <= rank <= 100: errors.append(f"Row {row_num}: rank {rank} out of 1-100 range") elif rank in seen_ranks: errors.append(f"Row {row_num}: duplicate rank {rank}") else: seen_ranks.add(rank) except ValueError: errors.append(f"Row {row_num}: rank '{rank_s}' is not an integer") rank = None try: score = float(score_s) except ValueError: errors.append(f"Row {row_num}: score '{score_s}' is not a float") score = None if rank is not None and score is not None and cid: by_rank.append((rank, score, cid)) # Check monotonicity by_rank.sort() for i in range(len(by_rank) - 1): r1, s1, _ = by_rank[i] r2, s2, _ = by_rank[i + 1] if s1 < s2: errors.append(f"Score not non-increasing: rank {r1} ({s1}) < rank {r2} ({s2})") if s1 == s2: # Tie-break: candidate_id ascending _, _, c1 = by_rank[i] _, _, c2 = by_rank[i + 1] if c1 > c2: errors.append(f"Equal scores at {r1}/{r2}: tie-break needs ascending ID ({c1} > {c2})") missing_ranks = set(range(1, 101)) - seen_ranks if missing_ranks: errors.append(f"Missing ranks: {sorted(missing_ranks)}") return errors def main(): ap = argparse.ArgumentParser(description="V7 Winning-Level Candidate Ranker") ap.add_argument("--candidates", required=True, help="Path to candidates JSONL file") ap.add_argument("--out", required=True, help="Path to output submission.csv") ap.add_argument("--artifacts", default="./artifacts/features.parquet", help="Path to precomputed features parquet") ap.add_argument("--skip-validation", action="store_true", help="Skip submission format validation") args = ap.parse_args() t0 = time.time() candidates_path = Path(args.candidates).resolve() # If no precomputed artifact, run precompute first if not Path(args.artifacts).exists(): print(f"[rank] artifact {args.artifacts} not found -- running precompute.py first") if not candidates_path.exists(): sys.exit(f"[rank] ERROR: candidates file not found: {candidates_path}") precompute_py = Path(__file__).resolve().parent / "precompute.py" subprocess.run( [sys.executable, str(precompute_py), "--candidates", str(candidates_path), "--out", args.artifacts], check=True, ) # Load precomputed features df = pd.read_parquet(args.artifacts) print(f"[rank] loaded {len(df)} precomputed feature rows") # Print pipeline info try: from lib.hiring_intent import get_intent intent = get_intent() print(f"[rank] V7 Hiring Intent: {intent.philosophy} | " f"ownership={intent.ownership_expectation:.2f} | " f"need={intent.primary_need} | team={intent.team_context}") except Exception as e: print(f"[rank] Warning: hiring intent failed: {e}") # Compute intent-aware scores elite = scoring.elite_score_vec(df) final = scoring.final_score_vec(df) df = df.assign(elite_score=elite.values, raw_score=final.values) top_n = min(TOP_N, len(df)) if len(df) < TOP_N: print(f"[rank] WARNING: pool has only {len(df)} candidates (<{TOP_N}).") # Sort by score, then by candidate_id for deterministic tie-breaking top = df.sort_values("raw_score", ascending=False).head(top_n).copy() # Apply score stretch sigmoid for confident-looking distribution raw = top["raw_score"].values.astype(float) raw_min, raw_max = raw.min(), raw.max() if raw_max > raw_min: norm = (raw - raw_min) / (raw_max - raw_min) stretched = 1.0 / (1.0 + np.exp(-10.0 * (norm - 0.5))) final_scores = 0.52 + 0.47 * stretched else: final_scores = np.full(top_n, 0.75) top["score"] = final_scores top["score"] = top["score"].round(6) top = top.sort_values(["score", "candidate_id"], ascending=[False, True]).reset_index(drop=True) top["rank"] = top.index + 1 honeypots_in_top = int(top["is_honeypot"].sum()) print(f"[rank] honeypots in top {top_n}: {honeypots_in_top} " f"({'OK' if top_n == 0 or honeypots_in_top / top_n <= 0.10 else 'FAIL >10%'})") # Score distribution scores = top["score"].values print(f"[rank] score distribution: top1={scores[0]:.6f} " f"top10_avg={scores[:min(10, len(scores))].mean():.6f} " f"top50_avg={scores[:min(50, len(scores))].mean():.6f} " f"bottom={scores[-1]:.6f}") print(f"[rank] unique scores: {len(set(scores))}/{len(scores)}") # Check monotonicity non_mono = sum(1 for i in range(len(scores) - 1) if scores[i] < scores[i + 1]) print(f"[rank] monotonicity violations: {non_mono}") # Generate reasoning print(f"[rank] generating V7 evidence-graph reasoning for {top_n} candidates") t_reason = time.time() out_rows = [] for _, row in top.iterrows(): cid = row["candidate_id"] candidate = json.loads(row["_candidate_json"]) disq_reasons = json.loads(row["_disq_reasons"]) beh_evidence = json.loads(row["_behaviour_evidence"]) narrative_suspicious = json.loads(row.get("narrative_suspicious", "[]")) # All features needed by the reasoning engine feat = { "current_title": row["current_title"], "current_company": row["current_company"], "years_of_experience": row["years_of_experience"], "skill_coverage": float(row.get("skill_coverage", 0)), "impact_magnitude": float(row.get("impact_magnitude", 0)), "ownership_hierarchy": float(row.get("ownership_hierarchy", 0)), "evaluation_experience": float(row.get("evaluation_experience", 0)), "pre_llm_months": float(row.get("pre_llm_months", 0)), "evidence_strength": float(row.get("evidence_strength", 0)), "notice_period_days": beh_evidence.get("notice_period_days", 45), "days_since_active": beh_evidence.get("days_since_active", 90), "recruiter_response_rate": beh_evidence.get("recruiter_response_rate", 0.3), "disqualifier_reasons": disq_reasons, # V6.1 features for reasoning "tier5_signature": float(row.get("tier5_signature", 0)), "behavioral_twin_penalty": float(row.get("behavioral_twin_penalty", 1.0)), "langchain_only_penalty": float(row.get("langchain_only_penalty", 1.0)), "closed_source_penalty": float(row.get("closed_source_penalty", 1.0)), "pre_llm_x_ownership": float(row.get("pre_llm_x_ownership", 0)), "salary_compatibility": float(row.get("salary_compatibility", 0.7)), "is_honeypot": bool(row.get("is_honeypot", False)), # V7 features for reasoning "assessment_signal": float(row.get("assessment_signal", 0.5)), "endorsement_signal": float(row.get("endorsement_signal", 0.5)), "education_tier": float(row.get("education_tier", 0.5)), "cross_validation": float(row.get("cross_validation", 0.5)), # Raw data for reasoning "_candidate_json": row.get("_candidate_json", "{}"), "_beh_twin_evidence": row.get("_beh_twin_evidence", "{}"), "_langchain_evidence": row.get("_langchain_evidence", "{}"), "_closed_source_evidence": row.get("_closed_source_evidence", "{}"), "_salary_evidence": row.get("_salary_evidence", "{}"), "_assessment_evidence": row.get("_assessment_evidence", "{}"), "_endorsement_evidence": row.get("_endorsement_evidence", "{}"), "_education_evidence": row.get("_education_evidence", "{}"), } text = reasoning.generate(cid, candidate, feat, narrative_suspicious) out_rows.append({ "candidate_id": cid, "rank": int(row["rank"]), "score": f'{row["score"]:.6f}', "reasoning": text, }) print(f"[rank] reasoning generated in {time.time() - t_reason:.1f}s") # Write CSV out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["candidate_id", "rank", "score", "reasoning"]) writer.writeheader() writer.writerows(out_rows) print(f"[rank] wrote {len(out_rows)} rows -> {out_path}") # Validate output if not args.skip_validation: print(f"[rank] validating output...") errors = validate_output(str(out_path)) if errors: print(f"[rank] VALIDATION FAILED ({len(errors)} issue(s)):") for e in errors: print(f" - {e}") sys.exit(1) else: print(f"[rank] validation PASSED") total_time = time.time() - t0 print(f"[rank] total ranking time: {total_time:.2f}s") if total_time > 300: print(f"[rank] WARNING: exceeded 5-minute budget ({total_time:.1f}s)") if __name__ == "__main__": main()