7c5a88dd3f
- rank.py: get_weights(profile) — when the profile reads non-tech (sales/finance/HR/ops/support/…), tilt the weights to role(0.16)/industry(0.12)/semantic(0.12), drop skill(0.18). Tech weights unchanged. (The competency-skill-backfill was tried + REVERTED — uniform per-role competencies added no discrimination and Phase 2's real-cosine f_semantic already supplies the non-tech signal.) - curate.py: non-tech instruction — role/responsibility fit is the spine; a missing skill list is NOT a negative; a strong role+industry+responsibility+seniority match is a genuine ~80, not ~65. - test_contracts.py: allowlist the new keyword-handled non-tech vocab (active boards keyword-fold; numeric codes deferred since the active feed/recall paths don't use them). Regression (120 pairs): MAE 18.4 → 16.1 (all from non-tech: sales 17.1→13.0, ops 16.3→11.7), recall holds 0.92, tech unchanged. 31 tests pass.
313 lines
14 KiB
Python
313 lines
14 KiB
Python
"""Coverage-aware, FLOOR-FREE white-box ranker (Phase 1 — ENGINE_DESIGN §3, floor-free per §3.3).
|
||
|
||
The cheap cold-start ranker — one of the three rankers fused in Phase 3, NOT "the engine."
|
||
Hard rule (the one the old engine broke): **every factor returns a real 0–1 score OR `None`**;
|
||
an absent factor leaves the weighted average (its weight redistributes) — never a constant floor.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
|
||
from app.engine import skills as S
|
||
from app.engine.board_adapters import coerce as C
|
||
from app.engine.schema import Growth, MatchDim, MatchResult
|
||
|
||
# Expert-prior weights (§3.4 — to calibrate, not optima). semantic = keyword placeholder → embedding in §3.
|
||
WEIGHTS = {"skill": 0.35, "experience": 0.25, "location": 0.15, "salary": 0.10, "industry": 0.05,
|
||
"role": 0.05, "semantic": 0.10}
|
||
FACTOR_LABELS = {"skill": "Skill match", "experience": "Experience", "location": "Location fit",
|
||
"salary": "Salary fit", "industry": "Industry", "role": "Role fit", "semantic": "Role relevance"}
|
||
# job seniority_level → (min, max) years band
|
||
BAND_YEARS = {"junior": (0, 3), "mid": (3, 8), "senior": (8, 13), "lead": (12, 99)}
|
||
SELECT = {"safe": 0.85, "balanced": 0.70, "reach": 0.55}
|
||
FLOOR, CAP = 10, 25
|
||
_STOP = {"the", "and", "for", "with", "you", "our", "your", "are", "this", "that", "will", "job", "role"}
|
||
|
||
|
||
@dataclass
|
||
class UserProfile:
|
||
title: str | None = None
|
||
skills: set[str] = field(default_factory=set)
|
||
years: int | None = None
|
||
cities: list[str] = field(default_factory=list)
|
||
countries: list[str] = field(default_factory=list)
|
||
onsite_only: bool = False
|
||
has_loc_pref: bool = False
|
||
industries: list[str] = field(default_factory=list)
|
||
roles: list[str] = field(default_factory=list)
|
||
comp_lpa: float | None = None # target minimum, in lakhs
|
||
deal_breakers: list[str] = field(default_factory=list)
|
||
stretch: str = "balanced"
|
||
profile_tokens: set[str] = field(default_factory=set)
|
||
|
||
|
||
def _comp_min_lpa(prefs: dict) -> float | None:
|
||
nums = re.findall(r"\d+", str(prefs.get("targetComp") or ""))
|
||
return float(nums[0]) if nums else None
|
||
|
||
|
||
def profile_from(prefs: dict, user_context: dict | None = None) -> UserProfile:
|
||
locs = C.parsed_locations(prefs)
|
||
cities = [lp["city"].lower() for lp in locs if lp.get("city")]
|
||
countries = [lp["country"].lower() for lp in locs if lp.get("country")]
|
||
wm = [w.lower() for w in (prefs.get("workMode") or [])]
|
||
title = C.title_of(prefs) or None
|
||
skills = S.canon((user_context or {}).get("skills")) # absent for non-tech → semantic (cosine) carries it
|
||
toks = {t for t in re.findall(r"[a-z0-9+#]+", f"{title or ''} {' '.join(skills)}".lower()) if len(t) > 2}
|
||
return UserProfile(
|
||
title=title, skills=skills, years=C.years_of(prefs),
|
||
cities=cities, countries=countries,
|
||
onsite_only=("on-site" in wm or "onsite" in wm),
|
||
has_loc_pref=bool(cities or countries),
|
||
industries=[i for i in (prefs.get("industry") or []) if i and i != "Any"],
|
||
roles=[r for r in (prefs.get("role") or []) if r],
|
||
comp_lpa=_comp_min_lpa(prefs),
|
||
deal_breakers=prefs.get("dealBreakers") or [],
|
||
stretch=(prefs.get("stretch") or "balanced"),
|
||
profile_tokens=toks,
|
||
)
|
||
|
||
|
||
# ── job accessors ──
|
||
def _jskills(job): return (job.get("details") or {}).get("skills") or job.get("required_skills") or []
|
||
def _jindustry(job): return (job.get("details") or {}).get("industry") or job.get("industry")
|
||
def _jrole(job): return (job.get("details") or {}).get("role_category")
|
||
|
||
|
||
# ── factors: each returns float in [0,1] OR None (absent → weight redistributes) ──
|
||
def f_skill(p, job):
|
||
o = S.overlap(p.skills, _jskills(job))
|
||
return o["score"] if o else None
|
||
|
||
|
||
def f_experience(p, job):
|
||
band = BAND_YEARS.get((job.get("seniority_level") or "").lower())
|
||
if band is None or p.years is None:
|
||
return None
|
||
lo, hi = band
|
||
if p.years < lo:
|
||
return max(0.2, 1.0 - (lo - p.years) / max(lo, 1) * 1.2)
|
||
if p.years > hi:
|
||
return max(0.4, 1.0 - (p.years - hi) / max(p.years, 1) * 0.9)
|
||
return 1.0
|
||
|
||
|
||
def f_location(p, job):
|
||
if (job.get("location_mode") or "").lower() == "remote":
|
||
return 0.0 if p.onsite_only else 1.0
|
||
if not p.has_loc_pref:
|
||
return None # "Anywhere"/open → absent, NOT a 60 floor
|
||
jl = f"{job.get('location_city') or ''} {job.get('location_country') or ''}".lower().strip()
|
||
if not jl:
|
||
return None
|
||
if any(c and c in jl for c in p.cities):
|
||
return 1.0 # exact city / metro
|
||
if any(c and c in jl for c in p.countries):
|
||
return 0.6 # same country, different city
|
||
return 0.0 # different country → hard-filtered out
|
||
|
||
|
||
def f_salary(p, job):
|
||
jl = job.get("salary_lpa")
|
||
if jl is None or p.comp_lpa is None:
|
||
return None
|
||
if jl >= p.comp_lpa:
|
||
return 1.0
|
||
gap = (p.comp_lpa - jl) / max(p.comp_lpa, 1)
|
||
return max(0.3, 1.0 - gap)
|
||
|
||
|
||
def _cat_match(wants, value):
|
||
if not wants or not value:
|
||
return None
|
||
v = str(value).lower()
|
||
return 1.0 if any(w.lower() in v or v in w.lower() for w in wants) else 0.3
|
||
|
||
|
||
def f_industry(p, job): return _cat_match(p.industries, _jindustry(job))
|
||
def f_role(p, job): return _cat_match(p.roles, _jrole(job))
|
||
|
||
|
||
def f_semantic(p, job):
|
||
# Real embedding cosine when the sift threaded it on (the true semantic signal); else token overlap.
|
||
c = job.get("_vibe_cosine")
|
||
if c is not None:
|
||
return float(c)
|
||
if not p.profile_tokens:
|
||
return None
|
||
blob = f"{job.get('title') or ''} {(job.get('details') or {}).get('description') or ''}".lower()
|
||
jt = {t for t in re.findall(r"[a-z0-9+#]+", blob) if len(t) > 2 and t not in _STOP}
|
||
if not jt:
|
||
return None
|
||
return len(p.profile_tokens & jt) / len(p.profile_tokens)
|
||
|
||
|
||
FACTORS = {"skill": f_skill, "experience": f_experience, "location": f_location, "salary": f_salary,
|
||
"industry": f_industry, "role": f_role, "semantic": f_semantic}
|
||
|
||
# Non-tech identity = role + industry + meaning, NOT a skill checklist (which boards rarely list).
|
||
# When the profile reads non-tech, tilt the weights there; skill drops, semantic/role/industry rise.
|
||
NON_TECH_FAMILIES = {"sales", "marketing", "finance", "accounting", "hr", "human resource", "operations",
|
||
"ops", "customer support", "customer service", "bpo", "legal", "supply chain", "admin",
|
||
"business development"}
|
||
NONTECH_WEIGHTS = {"skill": 0.18, "experience": 0.22, "location": 0.14, "salary": 0.06,
|
||
"industry": 0.12, "role": 0.16, "semantic": 0.12}
|
||
|
||
|
||
def _is_nontech(p: UserProfile) -> bool:
|
||
blob = " ".join([*(r.lower() for r in p.roles), (p.title or "").lower()])
|
||
return any(fam in blob for fam in NON_TECH_FAMILIES)
|
||
|
||
|
||
def get_weights(p: UserProfile) -> dict:
|
||
return NONTECH_WEIGHTS if _is_nontech(p) else WEIGHTS
|
||
|
||
|
||
def utility(p: UserProfile, job: dict):
|
||
w = get_weights(p)
|
||
factors = {k: fn(p, job) for k, fn in FACTORS.items()}
|
||
num = sum(w[k] * v for k, v in factors.items() if v is not None)
|
||
den = sum(w[k] for k, v in factors.items() if v is not None)
|
||
U = (num / den) if den else None # NO present signals → None (filtered out)
|
||
return U, factors
|
||
|
||
|
||
def passes_filter(p: UserProfile, job: dict) -> bool:
|
||
loc = f_location(p, job)
|
||
if loc == 0.0: # remote-for-onsite-only OR explicit no-fit
|
||
return False
|
||
band = BAND_YEARS.get((job.get("seniority_level") or "").lower())
|
||
if band and p.years is not None and p.stretch != "reach":
|
||
lo, hi = band
|
||
if p.years + (3 if p.stretch == "balanced" else 0) < lo: # job needs much more seniority
|
||
return False
|
||
if p.years - 8 > hi: # wildly over-qualified
|
||
return False
|
||
return True
|
||
|
||
|
||
def _level(score: int) -> str:
|
||
return "Strong" if score >= 72 else "Solid" if score >= 52 else "Light"
|
||
|
||
|
||
# per-dimension one-line notes (the Gemini report-card voice). Templated → Opus replaces later.
|
||
_DIM_NOTES = {
|
||
"skill": {"Strong": "Covers most of the role's must-have skills", "Solid": "Covers several must-haves — a few gaps", "Light": "Light on the listed skills"},
|
||
"experience": {"Strong": "Your level sits right in their band", "Solid": "Just outside their experience band", "Light": "Off their experience band"},
|
||
"location": {"Strong": "Right in your target location", "Solid": "Same country, different city", "Light": "Off your preferred locations"},
|
||
"salary": {"Strong": "Meets your target pay", "Solid": "Near your target pay", "Light": "Below your target pay"},
|
||
"industry": {"Strong": "In your target industry", "Solid": "An adjacent industry", "Light": "A different industry"},
|
||
"role": {"Strong": "Matches your target role area", "Solid": "An adjacent role area", "Light": "A different role area"},
|
||
"semantic": {"Strong": "Reads very close to your search", "Solid": "Reads close to your search", "Light": "Loosely related to your search"},
|
||
}
|
||
|
||
|
||
def _breakdown(factors) -> list[MatchDim]:
|
||
dims = []
|
||
for k, v in factors.items():
|
||
if v is None:
|
||
continue
|
||
sc = round(v * 100)
|
||
lvl = _level(sc)
|
||
dims.append(MatchDim(name=FACTOR_LABELS[k], score=sc, level=lvl, note=_DIM_NOTES.get(k, {}).get(lvl)))
|
||
dims.sort(key=lambda d: -d.score)
|
||
return dims[:4]
|
||
|
||
|
||
def _archetype(fit: str, score: int) -> str:
|
||
if fit == "fit":
|
||
return "The Strong Match" if score >= 85 else "The Solid Fit"
|
||
return "The Stretch Worth Taking" if score >= 68 else "The Long Shot"
|
||
|
||
|
||
def _one_line(dims: list[MatchDim]) -> str:
|
||
strong = [d.name.lower() for d in dims if d.level == "Strong"][:2]
|
||
light = [d.name.lower() for d in dims if d.level == "Light"][:1]
|
||
if strong and light:
|
||
return f"Your {' and '.join(strong)} land cleanly — {light[0]} is the gap to bridge."
|
||
if strong:
|
||
return f"Your {' and '.join(strong)} make this a clean match."
|
||
if light:
|
||
return f"A reach — {light[0]} is the stretch, but the rest lines up."
|
||
return "A balanced match worth a closer look."
|
||
|
||
|
||
def _coach(dims: list[MatchDim]) -> str:
|
||
strong = next((d.name.lower() for d in dims if d.level == "Strong"), None)
|
||
weak = min((d for d in dims if d.level == "Light"), key=lambda d: d.score, default=None)
|
||
if weak and strong:
|
||
return f"Lead with your {strong}; shore up {weak.name.lower()} before you apply and this reads as a clean fit."
|
||
if weak:
|
||
return f"Worth a look — close the {weak.name.lower()} gap to strengthen your case."
|
||
return "A strong fit — apply with confidence."
|
||
|
||
|
||
def _fit(p: UserProfile, U: float, factors) -> str:
|
||
sk = factors.get("skill")
|
||
if p.stretch == "reach":
|
||
return "stretch" if U < 0.8 else "fit"
|
||
strong = (sk is None or sk >= 0.5) and U >= 0.7
|
||
return "fit" if strong else "stretch"
|
||
|
||
|
||
def _reason(job, dims: list[MatchDim], fit: str) -> str:
|
||
board = (job.get("note") or "").replace("Live from ", "") or "the boards"
|
||
top = [d.name.lower() for d in dims if d.level in ("Strong", "Solid")][:2]
|
||
lead = " · ".join(top) if top else ("a reach worth a look" if fit == "stretch" else "a solid fit")
|
||
return f"{lead.capitalize()} — fresh from {board}"
|
||
|
||
|
||
def _growth(dims: list[MatchDim], score: int) -> Growth | None:
|
||
light = [d for d in dims if d.level == "Light"]
|
||
if not light:
|
||
return None
|
||
w = min(light, key=lambda d: d.score)
|
||
return Growth(text=f"Close the {w.name.lower()} gap to lift this match", **{"from": score, "to": min(99, score + 8)})
|
||
|
||
|
||
def _match(p: UserProfile, job: dict, U: float, factors) -> MatchResult:
|
||
score = round(U * 100)
|
||
dims = _breakdown(factors)
|
||
fit = _fit(p, U, factors)
|
||
return MatchResult(
|
||
score=score, fit=fit, archetype=_archetype(fit, score), one_line=_one_line(dims),
|
||
reason=_reason(job, dims, fit), breakdown=dims, growth=_growth(dims, score), coach_note=_coach(dims),
|
||
proofReady=(factors.get("skill") is not None and factors["skill"] >= 0.72),
|
||
factors={k: (round(v, 3) if v is not None else None) for k, v in factors.items()},
|
||
)
|
||
|
||
|
||
def rank(prefs: dict, user_context: dict | None, jobs: list[dict]) -> dict:
|
||
"""Score ALL jobs (no filter, no cut). The bouncer is fired — Apify already guarantees city/title
|
||
precision, so location/seniority are *scoring* signals, not gates. This is the cheap SIFTER pass:
|
||
it ranks everything; `sift.sift()` then blends it with the embedding vibe to pick the top ~18 the
|
||
curator (Opus) reads. The final cut is Opus's call (or `select()` in the fallback path)."""
|
||
p = profile_from(prefs, user_context)
|
||
scored = []
|
||
for job in jobs:
|
||
U, factors = utility(p, job)
|
||
if U is None: # no scorable factor at all → can't rank it
|
||
continue
|
||
scored.append((U, factors, job))
|
||
scored.sort(key=lambda x: -x[0])
|
||
|
||
out = []
|
||
for U, factors, job in scored:
|
||
m = _match(p, job, U, factors)
|
||
job["match"] = m.as_dict()
|
||
job["matchScore"] = m.score
|
||
out.append(job)
|
||
return {"opportunities": out, "ranked": len(out), "scanned": len(jobs)}
|
||
|
||
|
||
def select(jobs: list[dict], stretch: str = "balanced", floor: int = FLOOR, cap: int = CAP) -> list[dict]:
|
||
"""Final shortlist cut — applied AFTER fusion (Phase 3), NEVER on the cheap white-box alone.
|
||
Cuts by the (fused) `matchScore` against the stretch threshold, with a floor (never empty) and
|
||
cap (~25). `jobs` must already be ranked best-first."""
|
||
bar = round(SELECT.get(stretch, 0.70) * 100)
|
||
passing = [j for j in jobs if j.get("matchScore", 0) >= bar]
|
||
if len(passing) < floor:
|
||
passing = jobs[:floor]
|
||
return passing[:cap]
|