""" lib/feature_registry.py — V6 Dynamic Feature Registry Plugin-based feature system. Every feature is a registered object, not a hardcoded function call. This enables: - Clean ablation (remove a feature by name) - Feature importance tracking - Easy addition of new features - Intent-driven weight modulation Each feature: - Has a name, group, and description - Has a default weight (used in heuristic scoring) - Has an extract function (takes CanonicalProfile) - Has intent_modulation: how hiring intent changes its weight - Can depend on other features (for interaction features) """ from __future__ import annotations from dataclasses import dataclass, field from typing import Callable, Any from lib.candidate_profile import CanonicalProfile @dataclass class FeatureSpec: """Specification for a single feature.""" name: str group: str # JD_Fit, Impact, Ownership, etc. description: str default_weight: float # used in heuristic composite extract: Callable # fn(CanonicalProfile) -> float intent_modulation: dict[str, float] = field(default_factory=dict) # Example: {"founding": 1.5, "senior_ic": 1.0} # Means: if intent.ownership_expectation is "founding", multiply weight by 1.5 depends_on: list[str] = field(default_factory=list) is_interaction: bool = False # True for features that combine other features class FeatureRegistry: """ Dynamic feature registry. Register features, extract all, compute weighted composites, and run ablation studies. """ def __init__(self): self._features: dict[str, FeatureSpec] = {} self._extracted: dict[str, float] = {} def register(self, spec: FeatureSpec) -> "FeatureRegistry": """Register a feature.""" self._features[spec.name] = spec return self def extract_all(self, profile: CanonicalProfile) -> dict[str, float]: """Extract all registered features from a canonical profile.""" self._extracted = {} for name, spec in self._features.items(): if spec.is_interaction and spec.depends_on: # Interaction features depend on other features deps = {d: self._extracted.get(d, 0) for d in spec.depends_on} try: self._extracted[name] = spec.extract(profile, **deps) except TypeError: self._extracted[name] = spec.extract(profile) else: try: self._extracted[name] = spec.extract(profile) except Exception: self._extracted[name] = 0.0 return self._extracted def get(self, name: str) -> float: """Get extracted feature value.""" return self._extracted.get(name, 0.0) def get_all(self) -> dict[str, float]: """Get all extracted feature values.""" return dict(self._extracted) def get_group(self, group: str) -> dict[str, float]: """Get all features in a group.""" return {name: val for name, spec in self._features.items() if spec.group == group and name in self._extracted for name, val in [(n, self._extracted[n])]} if self._extracted.get(n, None) is not None} def get_names(self) -> list[str]: """Get all registered feature names.""" return list(self._features.keys()) def get_groups(self) -> list[str]: """Get all unique group names.""" return list(set(spec.group for spec in self._features.values())) def modulate_weights(self, intent: Any) -> dict[str, float]: """ Adjust feature weights based on hiring intent. Returns a dict of name -> adjusted_weight. """ from lib.hiring_intent import HiringIntent if not isinstance(intent, HiringIntent): return {name: spec.default_weight for name, spec in self._features.items()} adjusted = {} for name, spec in self._features.items(): w = spec.default_weight for intent_key, multiplier in spec.intent_modulation.items(): # Check if this intent signal matches if intent_key == "ownership_high" and intent.ownership_expectation >= 0.8: w *= multiplier elif intent_key == "shipping_fast" and intent.shipping_culture == "scrappy": w *= multiplier elif intent_key == "independence_high" and intent.independence >= 0.7: w *= multiplier elif intent_key == "specialist" and intent.depth_requirement == "specialist": w *= multiplier elif intent_key == "startup" and intent.team_context in ("founding", "early"): w *= multiplier elif intent_key == "mentorship" and intent.mentorship: w *= multiplier elif intent_key == "scale" and intent.scale_requirement == "large_scale": w *= multiplier elif intent_key == "production" and intent.primary_need == "production_systems": w *= multiplier adjusted[name] = w return adjusted def ablate(self, feature_name: str, profile: CanonicalProfile, composite_fn: Callable) -> dict: """ Run ablation: remove one feature and measure composite change. Returns: {"feature": name, "full_score": x, "ablated_score": y, "delta": d} """ # Full extraction full_features = self.extract_all(profile) full_score = composite_fn(full_features) # Ablated: set the feature to 0 (its default for missing) ablated_features = dict(full_features) ablated_features[feature_name] = 0.0 ablated_score = composite_fn(ablated_features) return { "feature": feature_name, "full_score": full_score, "ablated_score": ablated_score, "delta": full_score - ablated_score, } def full_ablation(self, profile: CanonicalProfile, composite_fn: Callable) -> list[dict]: """Run ablation for ALL features. Returns sorted by impact.""" results = [] for name in self._features: result = self.ablate(name, profile, composite_fn) results.append(result) results.sort(key=lambda x: abs(x["delta"]), reverse=True) return results # --------------------------------------------------------------------------- # Global registry instance + registration helpers # --------------------------------------------------------------------------- _registry: FeatureRegistry | None = None def get_registry() -> FeatureRegistry: """Get the global feature registry.""" global _registry if _registry is None: _registry = FeatureRegistry() _register_all_features(_registry) return _registry def _register_all_features(reg: FeatureRegistry) -> None: """Register all V6 features.""" # Import here to avoid circular imports from lib import features as feat # --- G1: JD Fit --- reg.register(FeatureSpec( name="skill_coverage", group="JD_Fit", description="Fraction of required JD skills in career context", default_weight=1.0, extract=lambda p: _safe_call(feat.skill_coverage, p._raw)[0], intent_modulation={"specialist": 1.3, "production": 1.1}, )) reg.register(FeatureSpec( name="preferred_coverage", group="JD_Fit", description="Fraction of preferred JD skills in career context", default_weight=0.7, extract=lambda p: _safe_call(feat.preferred_coverage, p._raw)[0], intent_modulation={"specialist": 1.2}, )) reg.register(FeatureSpec( name="domain_specialization", group="JD_Fit", description="Depth in JD's primary domain", default_weight=0.9, extract=lambda p: _safe_call(feat.domain_specialization, p._raw)[0], intent_modulation={"specialist": 1.4}, )) reg.register(FeatureSpec( name="skill_trust_avg", group="JD_Fit", description="Weighted avg trust signal for JD skills", default_weight=0.8, extract=lambda p: _safe_call(feat.skill_trust_avg, p._raw)[0], )) reg.register(FeatureSpec( name="title_relevance", group="JD_Fit", description="Title match to JD role", default_weight=0.9, extract=lambda p: _safe_call(feat.title_relevance, p._raw)[0], )) reg.register(FeatureSpec( name="seniority", group="JD_Fit", description="Seniority level from title", default_weight=0.7, extract=lambda p: _safe_call(feat.seniority_feature, p._raw)[0], intent_modulation={"startup": 0.9, "ownership_high": 1.1}, )) reg.register(FeatureSpec( name="jd_skill_count", group="JD_Fit", description="Raw count of JD skills found, normalized", default_weight=0.6, extract=lambda p: _safe_call(feat.jd_skill_count, p._raw, text=schema.unified_text_blob(p._raw))[0], )) # --- G2: Impact & Ownership --- reg.register(FeatureSpec( name="ownership_hierarchy", group="Impact_Ownership", description="Hierarchical ownership scoring", default_weight=1.0, extract=lambda p: p.ownership_level, intent_modulation={"ownership_high": 1.5, "startup": 1.3, "independence_high": 1.2}, )) reg.register(FeatureSpec( name="impact_magnitude", group="Impact_Ownership", description="Strength of best quantified impact", default_weight=1.0, extract=lambda p: p.best_impact_strength, intent_modulation={"production": 1.2, "scale": 1.1}, )) reg.register(FeatureSpec( name="impact_signals", group="Impact_Ownership", description="Non-quantified impact language", default_weight=0.8, extract=lambda p: _safe_call(feat.impact_signals, p._raw)[0], )) reg.register(FeatureSpec( name="evidence_strength", group="Impact_Ownership", description="Evidence density and quality", default_weight=0.9, extract=lambda p: p.evidence_summary.get("top3_avg", 0) / 18.0, intent_modulation={"production": 1.1}, )) # --- G3: Production & Scale --- reg.register(FeatureSpec( name="production_strength", group="Production_Scale", description="Production deployment evidence", default_weight=0.9, extract=lambda p: p.production_readiness, intent_modulation={"production": 1.4, "shipping_fast": 1.3}, )) reg.register(FeatureSpec( name="production_diversity", group="Production_Scale", description="Variety of production signals", default_weight=0.7, extract=lambda p: _safe_call(feat.production_diversity, p._raw)[0], intent_modulation={"production": 1.2}, )) reg.register(FeatureSpec( name="scale_evidence", group="Production_Scale", description="Evidence of system scale", default_weight=0.8, extract=lambda p: _safe_call(feat.scale_evidence, p._raw)[0], intent_modulation={"scale": 1.3}, )) # --- G4: Experience & Career --- reg.register(FeatureSpec( name="yoe_band_score", group="Experience_Career", description="YoE match to JD range", default_weight=0.8, extract=lambda p: _safe_call(feat.yoe_band_score, p._raw)[0], )) reg.register(FeatureSpec( name="career_depth_ratio", group="Experience_Career", description="Fraction of career in domain", default_weight=0.9, extract=lambda p: p.yoe_domain_ratio, intent_modulation={"specialist": 1.3}, )) reg.register(FeatureSpec( name="pre_llm_months", group="Experience_Career", description="Pre-2022 IR experience, normalized", default_weight=0.8, extract=lambda p: p.pre_llm_months / 36.0, )) reg.register(FeatureSpec( name="career_trajectory", group="Experience_Career", description="Career progression quality", default_weight=0.7, extract=lambda p: p.career_trajectory, intent_modulation={"startup": 1.1}, )) reg.register(FeatureSpec( name="company_quality", group="Experience_Career", description="Current company tier", default_weight=0.7, extract=lambda p: _safe_call(feat.company_quality_feature, p._raw), intent_modulation={"startup": 0.8}, )) reg.register(FeatureSpec( name="company_quality_avg", group="Experience_Career", description="Average company quality across career", default_weight=0.6, extract=lambda p: p.company_tier_avg, )) reg.register(FeatureSpec( name="career_stability", group="Experience_Career", description="Average tenure length", default_weight=0.6, extract=lambda p: _safe_call(feat.career_stability, p._raw)[0], )) reg.register(FeatureSpec( name="promotion_velocity", group="Experience_Career", description="Speed of promotions", default_weight=0.5, extract=lambda p: _safe_call(feat.promotion_velocity, p._raw)[0], intent_modulation={"startup": 0.8}, )) # --- G5: Retrieval & Evaluation --- reg.register(FeatureSpec( name="retrieval_depth", group="Retrieval_Eval", description="Retrieval system sophistication", default_weight=0.8, extract=lambda p: _safe_call(feat.retrieval_depth, p._raw)[0], intent_modulation={"specialist": 1.2, "production": 1.1}, )) reg.register(FeatureSpec( name="evaluation_experience", group="Retrieval_Eval", description="Evaluation framework experience", default_weight=0.8, extract=lambda p: _safe_call(feat.evaluation_experience, p._raw)[0], intent_modulation={"production": 1.2}, )) reg.register(FeatureSpec( name="system_design_evidence", group="Retrieval_Eval", description="System design work", default_weight=0.6, extract=lambda p: _safe_call(feat.system_design_evidence, p._raw)[0], )) # --- G6: Behavioural --- reg.register(FeatureSpec( name="recency", group="Behavioural", description="How recently active", default_weight=0.6, extract=lambda p: p.recency, )) reg.register(FeatureSpec( name="responsiveness", group="Behavioural", description="Response rate and speed", default_weight=0.6, extract=lambda p: p.responsiveness, )) reg.register(FeatureSpec( name="market_demand", group="Behavioural", description="Recruiter interest signals", default_weight=0.4, extract=lambda p: p.market_demand, )) reg.register(FeatureSpec( name="github_activity", group="Behavioural", description="Code activity signal", default_weight=0.4, extract=lambda p: p.github_activity, )) reg.register(FeatureSpec( name="availability_score", group="Behavioural", description="Open to work + notice period", default_weight=0.5, extract=lambda p: p.availability, )) reg.register(FeatureSpec( name="interview_completion", group="Behavioural", description="Interview follow-through rate", default_weight=0.4, extract=lambda p: p.interview_completion, )) reg.register(FeatureSpec( name="platform_trust", group="Behavioural", description="Verification + completeness signals", default_weight=0.4, extract=lambda p: p.platform_trust, )) # --- G7: Resume Quality --- reg.register(FeatureSpec( name="quantified_outcomes", group="Resume_Quality", description="Number of quantified achievements", default_weight=0.7, extract=lambda p: p.quantified_outcomes, intent_modulation={"production": 1.1}, )) reg.register(FeatureSpec( name="truthiness", group="Resume_Quality", description="Cross-validation of skill claims", default_weight=0.7, extract=lambda p: p.truthiness, )) reg.register(FeatureSpec( name="keyword_stuffing_risk", group="Resume_Quality", description="Risk of keyword stuffing", default_weight=0.6, extract=lambda p: p.keyword_stuffing_risk, )) reg.register(FeatureSpec( name="profile_completeness", group="Resume_Quality", description="Profile completeness score", default_weight=0.4, extract=lambda p: p.profile_completeness, )) # --- G8: Safety --- reg.register(FeatureSpec( name="disqualifier_penalty", group="Safety", description="Multi-factor disqualification penalty", default_weight=1.0, extract=lambda p: p.disqualifier_penalty, )) reg.register(FeatureSpec( name="is_honeypot", group="Safety", description="Synthetic profile flag", default_weight=1.0, extract=lambda p: 1.0 if p.is_honeypot else 0.0, )) # --- G9: Location --- reg.register(FeatureSpec( name="location_score", group="Location", description="Location match to JD", default_weight=0.5, extract=lambda p: _safe_call(feat.location_score, p._raw)[0], )) # --- G10: Career Narrative (NEW in V6) --- reg.register(FeatureSpec( name="career_coherence", group="Narrative", description="How coherent the career story is", default_weight=0.6, extract=lambda p: p.career_coherence, intent_modulation={"startup": 1.2, "ownership_high": 1.1}, )) # --- G11: Interaction Features (NEW in V6) --- reg.register(FeatureSpec( name="ownership_x_production", group="Interaction", description="High ownership + production evidence = strong signal", default_weight=0.8, extract=lambda p, ownership=0, production=0: ownership * production, depends_on=["ownership_hierarchy", "production_strength"], is_interaction=True, intent_modulation={"production": 1.3, "ownership_high": 1.3}, )) reg.register(FeatureSpec( name="skill_x_yoe", group="Interaction", description="Skill coverage weighted by experience level", default_weight=0.7, extract=lambda p, skill_coverage=0, yoe_band=0: skill_coverage * (0.5 + 0.5 * yoe_band), depends_on=["skill_coverage", "yoe_band_score"], is_interaction=True, intent_modulation={"specialist": 1.2}, )) reg.register(FeatureSpec( name="impact_x_domain", group="Interaction", description="Impact in the JD's specific domain", default_weight=0.7, extract=lambda p, impact=0, depth=0: impact * (0.3 + 0.7 * depth), depends_on=["impact_magnitude", "career_depth_ratio"], is_interaction=True, intent_modulation={"specialist": 1.3, "production": 1.1}, )) reg.register(FeatureSpec( name="trajectory_x_company", group="Interaction", description="Improving career at improving companies", default_weight=0.5, extract=lambda p, trajectory=0, company_avg=0: trajectory * company_avg, depends_on=["career_trajectory", "company_quality_avg"], is_interaction=True, )) def _safe_call(fn, *args, **kwargs): """Call a feature extraction function safely, returning (value, {}).""" try: result = fn(*args, **kwargs) if isinstance(result, tuple): return result return (result, {}) except Exception: return (0.0, {})