#!/usr/bin/env python3 """Tune fusion weights + post-fusion modifiers; verify with live RankingPipeline.""" from __future__ import annotations import argparse import json import logging import subprocess import sys from datetime import datetime, timezone from pathlib import Path import optuna PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT_ROOT)) from app.artifact_names import FUSION_PARAMS, MODIFIER_PARAMS, TUNING_REPORT_MODIFIERS # noqa: E402 from app.career_recall import load_career_scores # noqa: E402 from app.config import get_settings # noqa: E402 from app.fusion import ( # noqa: E402 build_fusion_cache, evaluate_full_ranking, settings_with_fusion_params, ) from app.feature_store import load_silver_tiers # noqa: E402 from app.fusion import rank_from_fusion_cache # noqa: E402 from app.jd_requirements import load_or_build_jd_requirements # noqa: E402 from app.logging_setup import configure_logging # noqa: E402 from app.modifier_params import ModifierParams, save_modifier_params # noqa: E402 from app.pipeline import RankingPipeline # noqa: E402 from app.rank_data import load_candidates_lookup, load_feature_lookup # noqa: E402 logger = logging.getLogger(__name__) POOL_SIZES = (500, 600, 700, 800) CAP_SIZES = (28, 30, 32, 35) CLONE_CAPS = (6, 8, 10, 12) def _git_commit() -> str | None: try: out = subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=PROJECT_ROOT, stderr=subprocess.DEVNULL, text=True, ) return out.strip() except (subprocess.CalledProcessError, FileNotFoundError): return None def verify_live_pipeline( settings, artifacts_dir: Path, candidates_path: Path, silver: dict[str, int], candidate_lookup: dict, ) -> dict: """Run live RankingPipeline and evaluate same metrics as cache path.""" settings = settings_with_fusion_params(settings, artifacts_dir) pipeline = RankingPipeline(settings, artifacts_dir) results = pipeline.rank(candidates_path, top_k=100) ranked_ids = [r["candidate_id"] for r in results] return evaluate_full_ranking( ranked_ids, silver, candidate_lookup, reference_date=settings.reference_date, ) def tune_modifiers( settings, artifacts_dir: Path, candidates_path: Path, *, n_trials: int = 120, max_rerank_pool: int = 800, verify_live: bool = True, ) -> dict: silver = load_silver_tiers(artifacts_dir / "labels_silver.parquet") jd = load_or_build_jd_requirements( settings.job_description_path, artifacts_dir / "jd_requirements.json", settings=settings, ) feature_lookup = load_feature_lookup(artifacts_dir / "candidate_features.parquet") candidate_lookup = load_candidates_lookup(candidates_path) career_scores = load_career_scores(str(artifacts_dir)) cache = build_fusion_cache(settings, artifacts_dir, candidates_path, max_rerank_pool=max_rerank_pool) def objective(trial: optuna.Trial) -> float: ce_w = trial.suggest_float("rerank_ce_weight", 0.30, 0.55) ltr_w = trial.suggest_float("rerank_ltr_weight", 0.45, 0.52) rrf_w = 1.0 - ce_w - ltr_w if rrf_w < 0.05 or rrf_w > 0.25: raise optuna.TrialPruned() mod = ModifierParams( tier5_boost=trial.suggest_float("tier5_boost", 1.02, 1.12), low_avail_mult=trial.suggest_float("low_avail_mult", 0.25, 0.45), low_rr_steep_mult=trial.suggest_float("low_rr_steep_mult", 0.45, 0.65), clone_tier5_mult=trial.suggest_float("clone_tier5_mult", 0.88, 0.98), research_with_prod_mult=trial.suggest_float("research_with_prod_mult", 0.85, 0.98), plain_language_career_boost=trial.suggest_float("plain_language_career_boost", 1.02, 1.12), plain_language_career_threshold=trial.suggest_float("plain_language_career_threshold", 0.50, 0.65), stretched_scientist_mult=trial.suggest_float("stretched_scientist_mult", 0.55, 0.75), cv_speech_mult=trial.suggest_float("cv_speech_mult", 0.30, 0.50), primary_hub_boost=trial.suggest_float("primary_hub_boost", 1.00, 1.10), preferred_hub_boost=trial.suggest_float("preferred_hub_boost", 1.00, 1.06), clone_top30_max=trial.suggest_categorical("clone_top30_max", list(CLONE_CAPS)), top_availability_cap=trial.suggest_categorical("top_availability_cap", list(CAP_SIZES)), ) pool_size = trial.suggest_categorical("rerank_pool_size", list(POOL_SIZES)) career_w = trial.suggest_float("career_rrf_weight", 0.50, 0.80) trial_settings = settings.model_copy(update={"career_rrf_weight": career_w}) ranked = rank_from_fusion_cache( cache, jd=jd, candidate_lookup=candidate_lookup, feature_lookup=feature_lookup, reference_date=trial_settings.reference_date, rerank_pool_size=pool_size, top_k=100, ce_weight=ce_w, ltr_weight=ltr_w, rrf_weight=rrf_w, modifier_params=mod, career_scores=career_scores, ) metrics = evaluate_full_ranking( ranked, silver, candidate_lookup, reference_date=trial_settings.reference_date, ) if metrics["honeypots_in_top_10"] > 0 or metrics["low_avail_top_30"] > 0: raise optuna.TrialPruned() trial.set_user_attr("ndcg_at_10", metrics["ndcg_at_10"]) trial.set_user_attr("tier5_in_top_10", metrics["tier5_in_top_10"]) trial.set_user_attr("tier5_in_top_30", metrics["tier5_in_top_30"]) return ( metrics["ndcg_at_10"] + 0.02 * metrics["tier5_in_top_10"] + 0.008 * metrics["tier5_in_top_30"] + 0.003 * metrics["tier5_in_top_100"] - 0.01 * metrics.get("template_clones_top_30", 0) ) study = optuna.create_study(direction="maximize") study.optimize(objective, n_trials=n_trials, show_progress_bar=False) best = study.best_params best["rerank_rrf_weight"] = round(1.0 - best["rerank_ce_weight"] - best["rerank_ltr_weight"], 4) mod = ModifierParams( tier5_boost=best["tier5_boost"], low_avail_mult=best["low_avail_mult"], low_rr_steep_mult=best["low_rr_steep_mult"], clone_tier5_mult=best["clone_tier5_mult"], research_with_prod_mult=best["research_with_prod_mult"], plain_language_career_boost=best["plain_language_career_boost"], plain_language_career_threshold=best["plain_language_career_threshold"], stretched_scientist_mult=best["stretched_scientist_mult"], cv_speech_mult=best["cv_speech_mult"], primary_hub_boost=best["primary_hub_boost"], preferred_hub_boost=best["preferred_hub_boost"], clone_top30_max=best["clone_top30_max"], top_availability_cap=best["top_availability_cap"], ) cache_metrics = evaluate_full_ranking( rank_from_fusion_cache( cache, jd=jd, candidate_lookup=candidate_lookup, feature_lookup=feature_lookup, reference_date=settings.reference_date, rerank_pool_size=best["rerank_pool_size"], top_k=100, ce_weight=best["rerank_ce_weight"], ltr_weight=best["rerank_ltr_weight"], rrf_weight=best["rerank_rrf_weight"], modifier_params=mod, career_scores=career_scores, ), silver, candidate_lookup, reference_date=settings.reference_date, ) fusion_payload = { "rerank_ce_weight": best["rerank_ce_weight"], "rerank_ltr_weight": best["rerank_ltr_weight"], "rerank_rrf_weight": best["rerank_rrf_weight"], "rerank_pool_size": best["rerank_pool_size"], "career_rrf_weight": best["career_rrf_weight"], "proxy_objective": study.best_value, "cache_ndcg_at_10": cache_metrics["ndcg_at_10"], "tier5_in_top_10": cache_metrics["tier5_in_top_10"], "tier5_in_top_30": cache_metrics["tier5_in_top_30"], "tier5_in_top_100": cache_metrics["tier5_in_top_100"], "low_avail_top_30": cache_metrics["low_avail_top_30"], "n_trials": n_trials, "tuned_with_modifiers": True, } save_modifier_params(artifacts_dir / MODIFIER_PARAMS, mod) (artifacts_dir / FUSION_PARAMS).write_text(json.dumps(fusion_payload, indent=2), encoding="utf-8") live_metrics: dict | None = None if verify_live: settings = settings.model_copy(update={"career_rrf_weight": best["career_rrf_weight"]}) live_metrics = verify_live_pipeline(settings, artifacts_dir, candidates_path, silver, candidate_lookup) fusion_payload["live_ndcg_at_10"] = live_metrics["ndcg_at_10"] fusion_payload["live_tier5_in_top_10"] = live_metrics["tier5_in_top_10"] fusion_payload["live_tier5_in_top_30"] = live_metrics["tier5_in_top_30"] fusion_payload["live_tier5_in_top_100"] = live_metrics["tier5_in_top_100"] (artifacts_dir / FUSION_PARAMS).write_text(json.dumps(fusion_payload, indent=2), encoding="utf-8") save_modifier_params( artifacts_dir / MODIFIER_PARAMS, mod, extra={ "live_ndcg_at_10": live_metrics["ndcg_at_10"], "live_tier5_in_top_10": live_metrics["tier5_in_top_10"], "live_tier5_in_top_30": live_metrics["tier5_in_top_30"], "live_tier5_in_top_100": live_metrics["tier5_in_top_100"], }, ) top10_ids = ( [r["candidate_id"] for r in RankingPipeline(settings_with_fusion_params(settings, artifacts_dir), artifacts_dir).rank(candidates_path, top_k=10)] if verify_live else rank_from_fusion_cache( cache, jd=jd, candidate_lookup=candidate_lookup, feature_lookup=feature_lookup, reference_date=settings.reference_date, rerank_pool_size=best["rerank_pool_size"], top_k=10, ce_weight=best["rerank_ce_weight"], ltr_weight=best["rerank_ltr_weight"], rrf_weight=best["rerank_rrf_weight"], modifier_params=mod, career_scores=career_scores, ) ) report = { "timestamp": datetime.now(timezone.utc).isoformat(), "git_commit": _git_commit(), "best_params": {**best, **{k: v for k, v in fusion_payload.items() if k.startswith("rerank")}}, "cache_metrics": cache_metrics, "live_metrics": live_metrics, "top10_fingerprint": top10_ids, } (artifacts_dir / TUNING_REPORT_MODIFIERS).write_text(json.dumps(report, indent=2), encoding="utf-8") logger.info( "Cache: NDCG@10=%.4f tier5@30=%.0f | Live: %s", cache_metrics["ndcg_at_10"], cache_metrics["tier5_in_top_30"], f"NDCG@10={live_metrics['ndcg_at_10']:.4f}" if live_metrics else "skipped", ) return fusion_payload def main() -> int: parser = argparse.ArgumentParser(description="Tune fusion + modifier params (full ranking path).") parser.add_argument("--artifacts", type=Path, default=None) parser.add_argument("--candidates", type=Path, default=None) parser.add_argument("--trials", type=int, default=120) parser.add_argument("--max-pool", type=int, default=800) parser.add_argument( "--verify-live", action="store_true", default=True, help="Re-run live RankingPipeline after tuning (default: on).", ) parser.add_argument("--no-verify-live", action="store_true", help="Skip live pipeline verification.") args = parser.parse_args() settings = get_settings() configure_logging(settings.log_level) artifacts_dir = (args.artifacts or settings.artifacts_dir).resolve() candidates_path = (args.candidates or settings.candidates_path).resolve() if not (artifacts_dir / "ltr_model.lgb").exists(): logger.error("Train LTR first: python scripts/train_ltr.py") return 1 tune_modifiers( settings, artifacts_dir, candidates_path, n_trials=args.trials, max_rerank_pool=args.max_pool, verify_live=not args.no_verify_live, ) return 0 if __name__ == "__main__": raise SystemExit(main())