"""Versioned deterministic joiner resource-demand forecasting. This module is deliberately datastore-free. Collectors provide anonymous candidate/cohort keys, eligible peer holdings, and confirmed supply. The kernel then produces reproducible 30/60/90-day demand and gaps; an LLM may explain the result later but cannot change quantities, confidence, or actions. """ from __future__ import annotations from dataclasses import dataclass from datetime import date, timedelta from typing import Literal FORECAST_VERSION = "demand-forecast-v1" FORECAST_HORIZONS = (30, 60, 90) MIN_SAMPLE_SIZE = 5 MIN_PREVALENCE_BPS = 6000 Confidence = Literal["insufficient", "low", "medium", "high"] SampleTier = Literal["none", "role", "department_role", "exact"] ScopeMode = Literal["global", "location"] ActionKind = Literal["none", "review", "purchase", "configure"] @dataclass(frozen=True, order=True) class ResourceKey: kind: str key: str scope_mode: ScopeMode @dataclass(frozen=True) class Candidate: """One future joiner; ``ref`` is an anonymous local evidence reference.""" ref: str hire_date: date location_key: str department_key: str job_key: str employment_status: str = "prehire" cancelled: bool = False anomalous: bool = False @dataclass(frozen=True) class Peer: """One current comparison person and their effective resource holdings.""" ref: str location_key: str department_key: str job_key: str resources: frozenset[ResourceKey] employment_status: str = "active" anomalous: bool = False @dataclass(frozen=True) class SupplyArrival: available_on: date quantity: int @dataclass(frozen=True) class SupplyPosition: resource: ResourceKey scope_key: str available: int reserved: int = 0 arrivals: tuple[SupplyArrival, ...] = () lead_time_days: int = 0 unlimited: bool = False # Anonymous candidate refs whose unit is already included in ``reserved``. committed_candidate_refs: tuple[str, ...] = () @dataclass(frozen=True) class InferredResource: resource: ResourceKey prevalence_bps: int holder_count: int expected_quantity: int @dataclass(frozen=True) class CohortForecast: location_key: str department_key: str job_key: str joiner_count: int sample_tier: SampleTier sample_size: int confidence: Confidence candidate_refs: tuple[str, ...] resources: tuple[InferredResource, ...] @dataclass(frozen=True) class ResourceForecast: resource: ResourceKey scope_key: str demand: int supply: int gap: int affected_joiners: int day1_at_risk_count: int unlimited_supply: bool confidence: Confidence action_kind: ActionKind lead_time_risk: bool @dataclass(frozen=True) class HorizonForecast: horizon_days: int cutoff_date: date joiner_count: int insufficient_sample_joiner_count: int day1_at_risk_count: int cohorts: tuple[CohortForecast, ...] resources: tuple[ResourceForecast, ...] @dataclass(frozen=True) class DemandForecast: forecast_version: str as_of_date: date horizons: tuple[HorizonForecast, ...] @dataclass class _AggregateDemand: demand: int confidences: list[Confidence] # One expected unit per anonymous joiner, ordered later by hire date. events: list[tuple[date, str]] def _clean(value: str, field: str) -> str: cleaned = value.strip().casefold() if not cleaned: raise ValueError(f"{field} must not be empty") if len(cleaned) > 128: raise ValueError(f"{field} must be at most 128 characters") return cleaned def _resource(resource: ResourceKey) -> ResourceKey: kind = _clean(resource.kind, "resource kind") key = _clean(resource.key, "resource key") if resource.scope_mode not in ("global", "location"): raise ValueError("resource scope_mode must be global or location") return ResourceKey(kind=kind, key=key, scope_mode=resource.scope_mode) def _confidence_rank(value: Confidence) -> int: return {"insufficient": 0, "low": 1, "medium": 2, "high": 3}[value] def _sample_for( cohort: tuple[str, str, str], peers: tuple[Peer, ...], ) -> tuple[SampleTier, Confidence, tuple[Peer, ...]]: location, department, job = cohort eligible = tuple( row for row in peers if row.employment_status == "active" and not row.anomalous and row.job_key == job ) exact = tuple( row for row in eligible if row.department_key == department and row.location_key == location ) if len(exact) >= MIN_SAMPLE_SIZE: return "exact", "high", exact department_role = tuple(row for row in eligible if row.department_key == department) if len(department_role) >= MIN_SAMPLE_SIZE: return "department_role", "medium", department_role if len(eligible) >= MIN_SAMPLE_SIZE: return "role", "low", eligible return "none", "insufficient", () def _normalise_inputs( candidates: tuple[Candidate, ...], peers: tuple[Peer, ...], supply: tuple[SupplyPosition, ...], ) -> tuple[tuple[Candidate, ...], tuple[Peer, ...], dict[tuple[ResourceKey, str], SupplyPosition]]: candidate_refs: set[str] = set() clean_candidates: list[Candidate] = [] for candidate_row in candidates: ref = _clean(candidate_row.ref, "candidate ref") if ref in candidate_refs: raise ValueError(f"duplicate candidate ref: {ref}") candidate_refs.add(ref) clean_candidates.append(Candidate( ref=ref, hire_date=candidate_row.hire_date, location_key=_clean(candidate_row.location_key, "candidate location"), department_key=_clean(candidate_row.department_key, "candidate department"), job_key=_clean(candidate_row.job_key, "candidate job"), employment_status=candidate_row.employment_status.strip().casefold(), cancelled=candidate_row.cancelled, anomalous=candidate_row.anomalous, )) peer_refs: set[str] = set() clean_peers: list[Peer] = [] for peer_row in peers: ref = _clean(peer_row.ref, "peer ref") if ref in peer_refs: raise ValueError(f"duplicate peer ref: {ref}") peer_refs.add(ref) clean_peers.append(Peer( ref=ref, location_key=_clean(peer_row.location_key, "peer location"), department_key=_clean(peer_row.department_key, "peer department"), job_key=_clean(peer_row.job_key, "peer job"), resources=frozenset(_resource(item) for item in peer_row.resources), employment_status=peer_row.employment_status.strip().casefold(), anomalous=peer_row.anomalous, )) supply_by_key: dict[tuple[ResourceKey, str], SupplyPosition] = {} for supply_row in supply: resource = _resource(supply_row.resource) scope_key = ( "global" if resource.scope_mode == "global" else _clean(supply_row.scope_key, "supply scope") ) if min(supply_row.available, supply_row.reserved, supply_row.lead_time_days) < 0: raise ValueError("supply quantities and lead time must be non-negative") arrivals = tuple(sorted(supply_row.arrivals, key=lambda item: item.available_on)) if any(item.quantity < 0 for item in arrivals): raise ValueError("arrival quantity must be non-negative") committed_refs = tuple(sorted( _clean(value, "committed candidate ref") for value in supply_row.committed_candidate_refs )) if len(committed_refs) != len(set(committed_refs)): raise ValueError("duplicate committed candidate ref") unknown_refs = set(committed_refs) - candidate_refs if unknown_refs: raise ValueError("committed candidate ref is not in forecast candidates") if len(committed_refs) > supply_row.reserved: raise ValueError("committed candidates cannot exceed reserved supply") key = (resource, scope_key) if key in supply_by_key: raise ValueError(f"duplicate supply position: {key}") supply_by_key[key] = SupplyPosition( resource=resource, scope_key=scope_key, available=supply_row.available, reserved=supply_row.reserved, arrivals=arrivals, lead_time_days=supply_row.lead_time_days, unlimited=supply_row.unlimited, committed_candidate_refs=committed_refs, ) return tuple(clean_candidates), tuple(clean_peers), supply_by_key def build_demand_forecast( *, as_of_date: date, candidates: tuple[Candidate, ...], peers: tuple[Peer, ...], supply: tuple[SupplyPosition, ...], ) -> DemandForecast: """Build cumulative 30/60/90-day forecasts with portfolio-level supply. Cancelled, anomalous, left, and already-started candidates are excluded. Low-confidence role-only samples can describe demand but never emit a purchase/configure recommendation. Fewer than five peers emits no inferred resources at all, preventing a sparse sample from becoming a strong recommendation. """ candidates, peers, supply_by_key = _normalise_inputs(candidates, peers, supply) eligible = tuple( row for row in candidates if not row.cancelled and not row.anomalous and row.employment_status in ("prehire", "active") and as_of_date < row.hire_date <= as_of_date + timedelta(days=90) ) horizons: list[HorizonForecast] = [] for horizon_days in FORECAST_HORIZONS: cutoff = as_of_date + timedelta(days=horizon_days) horizon_candidates = tuple(row for row in eligible if row.hire_date <= cutoff) grouped: dict[tuple[str, str, str], list[Candidate]] = {} for row in horizon_candidates: grouped.setdefault( (row.location_key, row.department_key, row.job_key), [], ).append(row) cohorts: list[CohortForecast] = [] aggregate: dict[tuple[ResourceKey, str], _AggregateDemand] = {} insufficient = 0 for cohort_key in sorted(grouped): rows = tuple(sorted(grouped[cohort_key], key=lambda row: (row.hire_date, row.ref))) tier, confidence, sample = _sample_for(cohort_key, peers) inferred: list[InferredResource] = [] if not sample: insufficient += len(rows) else: resource_keys = {item for peer in sample for item in peer.resources} row_refs = {row.ref for row in rows} for (supply_resource, supply_scope), supply_position in supply_by_key.items(): scope_matches = ( supply_resource.scope_mode == "global" or supply_scope == cohort_key[0] ) if ( scope_matches and row_refs.intersection(supply_position.committed_candidate_refs) ): resource_keys.add(supply_resource) for resource in sorted(resource_keys): holders = sum(resource in peer.resources for peer in sample) prevalence_bps = holders * 10_000 // len(sample) scope_key = ( "global" if resource.scope_mode == "global" else cohort_key[0] ) cohort_position = supply_by_key.get((resource, scope_key)) committed = ( set(cohort_position.committed_candidate_refs) if cohort_position else set() ) committed_rows = [row for row in rows if row.ref in committed] if prevalence_bps < MIN_PREVALENCE_BPS and not committed_rows: continue expected = (len(rows) * holders + len(sample) - 1) // len(sample) selected_rows = committed_rows + [ row for row in rows if row.ref not in committed ] expected = max(expected, len(committed_rows)) selected_rows = selected_rows[:expected] inferred.append(InferredResource( resource=resource, prevalence_bps=prevalence_bps, holder_count=holders, expected_quantity=expected, )) bucket = aggregate.setdefault( (resource, scope_key), _AggregateDemand(0, [], []), ) bucket.demand += expected bucket.confidences.append(confidence) bucket.events.extend( (row.hire_date, row.ref) for row in selected_rows ) cohorts.append(CohortForecast( location_key=cohort_key[0], department_key=cohort_key[1], job_key=cohort_key[2], joiner_count=len(rows), sample_tier=tier, sample_size=len(sample), confidence=confidence, candidate_refs=tuple(row.ref for row in rows), resources=tuple(inferred), )) resource_rows: list[ResourceForecast] = [] at_risk_refs: set[str] = set() for key in sorted(aggregate, key=lambda item: (item[0], item[1])): resource, scope_key = key values = aggregate[key] demand = values.demand resource_position = supply_by_key.get(key) supply_count = 0 lead_time_days = 0 arrivals: tuple[SupplyArrival, ...] = () if resource_position is not None: supply_count = max( 0, resource_position.available - resource_position.reserved, ) supply_count += sum( item.quantity for item in resource_position.arrivals if item.available_on <= cutoff ) lead_time_days = resource_position.lead_time_days arrivals = resource_position.arrivals committed_refs = ( set(resource_position.committed_candidate_refs) & {ref for _, ref in values.events} if resource_position else set() ) supply_count += len(committed_refs) unlimited_supply = bool(resource_position and resource_position.unlimited) if unlimited_supply: supply_count = demand gap = max(0, demand - supply_count) confidence = min(values.confidences, key=_confidence_rank) if gap == 0: action: ActionKind = "none" elif confidence == "low": action = "review" elif resource.kind == "hardware": action = "purchase" else: action = "configure" events = sorted(values.events) earliest_days = (events[0][0] - as_of_date).days if events else horizon_days lead_time_risk = gap > 0 and lead_time_days >= earliest_days affected = min(len(events), gap) # Day-1 readiness is temporal: an arrival on day 45 may close a # 60-day capacity gap but cannot make a day-10 joiner ready. available_now = max( 0, ( resource_position.available - resource_position.reserved if resource_position else 0 ), ) arrival_index = 0 resource_at_risk: set[str] = set() for hire_date, candidate_ref in events: if unlimited_supply: continue if candidate_ref in committed_refs: continue while ( arrival_index < len(arrivals) and arrivals[arrival_index].available_on <= hire_date ): available_now += arrivals[arrival_index].quantity arrival_index += 1 if available_now: available_now -= 1 else: resource_at_risk.add(candidate_ref) at_risk_refs.add(candidate_ref) resource_rows.append(ResourceForecast( resource=resource, scope_key=scope_key, demand=demand, supply=supply_count, gap=gap, affected_joiners=affected, day1_at_risk_count=len(resource_at_risk), unlimited_supply=unlimited_supply, confidence=confidence, action_kind=action, lead_time_risk=lead_time_risk, )) horizons.append(HorizonForecast( horizon_days=horizon_days, cutoff_date=cutoff, joiner_count=len(horizon_candidates), insufficient_sample_joiner_count=insufficient, day1_at_risk_count=min(len(horizon_candidates), len(at_risk_refs)), cohorts=tuple(cohorts), resources=tuple(resource_rows), )) return DemandForecast( forecast_version=FORECAST_VERSION, as_of_date=as_of_date, horizons=tuple(horizons), ) __all__ = [ "Candidate", "CohortForecast", "DemandForecast", "FORECAST_HORIZONS", "FORECAST_VERSION", "HorizonForecast", "InferredResource", "MIN_SAMPLE_SIZE", "Peer", "ResourceForecast", "ResourceKey", "SupplyArrival", "SupplyPosition", "build_demand_forecast", ]