123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
# default_policy.py
|
|
# Pressure-gated policy: every admit must carry at least one statistically-scarce attribute
|
|
# until quotas are back on track. Designed to prevent over-admitting "easy" attributes while
|
|
# 'creative' and 'berlin_local' lag.
|
|
#
|
|
# Use policy_reason() in your UI to display decision explanations.
|
|
|
|
from __future__ import annotations
|
|
import sys
|
|
from typing import Dict
|
|
|
|
DEBUG = False
|
|
_last_reason: str | None = None
|
|
|
|
def _set_reason(msg: str) -> None:
|
|
global _last_reason
|
|
_last_reason = msg
|
|
if DEBUG:
|
|
print(msg, file=sys.stderr, flush=True)
|
|
|
|
def policy_reason() -> str | None:
|
|
return _last_reason
|
|
|
|
# --- Tunables ---
|
|
PRESSURE_SLACK = 0.0 # >0 softens the must-have gate slightly (e.g., 0.5)
|
|
PACING_START = 20 # ignore pacing/pressure noise for the first N admits
|
|
FEASIBILITY_SLACK = 0 # extra slack for the feasibility guard
|
|
|
|
# If runner doesn't pass base frequencies, use conservative fallbacks
|
|
DEFAULT_P = {
|
|
"creative": 0.06,
|
|
"berlin_local": 0.40,
|
|
"techno_lover": 0.60,
|
|
"well_connected": 0.45,
|
|
}
|
|
|
|
ORDER = ("creative", "berlin_local", "techno_lover", "well_connected")
|
|
|
|
def _need(mins: Dict[str,int], tallies: Dict[str,int]) -> Dict[str,int]:
|
|
return {a: max(0, int(mins.get(a, 0)) - int(tallies.get(a, 0))) for a in mins}
|
|
|
|
def _p(p: Dict[str,float] | None, a: str) -> float:
|
|
v = (p or {}).get(a, DEFAULT_P.get(a, 0.0))
|
|
try:
|
|
return max(0.0, min(1.0, float(v)))
|
|
except Exception:
|
|
return DEFAULT_P.get(a, 0.0)
|
|
|
|
def decide(attributes: Dict[str, bool],
|
|
tallies: Dict[str, int],
|
|
mins: Dict[str, int],
|
|
admitted_count: int,
|
|
venue_cap: int,
|
|
p: Dict[str, float] | None = None,
|
|
**_) -> bool:
|
|
|
|
has = lambda a: bool(attributes.get(a, False))
|
|
|
|
S = venue_cap - admitted_count
|
|
need = _need(mins, tallies)
|
|
total_need = sum(need.values())
|
|
|
|
# 0) All mins satisfied -> admit freely
|
|
if total_need == 0:
|
|
_set_reason("Admit (all minimums satisfied)")
|
|
return True
|
|
|
|
# Ensure candidate helps at least one unmet attribute
|
|
helps = [a for a in ORDER if need.get(a, 0) > 0 and has(a)]
|
|
if not helps:
|
|
_set_reason("Reject (no contribution to unmet attributes)")
|
|
return False
|
|
|
|
# Early game: admit any helper while counts are tiny to avoid noise
|
|
if admitted_count < PACING_START:
|
|
_set_reason(f"Admit (early game; helps {', '.join(helps)})")
|
|
return True
|
|
|
|
# 1) Compute base probs and PRESSURE = shortfall vs expected future supply
|
|
P = {a: _p(p, a) for a in ORDER}
|
|
S_after = max(0, S - 1)
|
|
|
|
pressure: Dict[str, float] = {}
|
|
for a in ORDER:
|
|
# remaining need after counting this candidate (if they have 'a')
|
|
rem_after = max(0, need.get(a, 0) - (1 if has(a) else 0))
|
|
# expected future supply after this admit
|
|
exp_after = S_after * P[a]
|
|
pressure[a] = rem_after - exp_after # >0 means we're short in expectation
|
|
|
|
# 2) Build MUST set: attributes whose pressure exceeds slack (i.e., statistically scarce)
|
|
must = [a for a in ORDER if pressure[a] > PRESSURE_SLACK]
|
|
|
|
if must:
|
|
# Require the candidate to carry at least one MUST attribute
|
|
if not any(has(a) for a in must):
|
|
_set_reason("Reject (must carry a scarce attribute: " + ", ".join(must) + ")")
|
|
return False
|
|
|
|
# 3) Feasibility: after admitting, do we still have enough slots to finish?
|
|
total_need_after = sum(
|
|
max(0, need.get(a, 0) - (1 if has(a) else 0)) for a in ORDER
|
|
)
|
|
if S_after + FEASIBILITY_SLACK < total_need_after:
|
|
_set_reason("Reject (feasibility would break after admitting)")
|
|
return False
|
|
|
|
# 4) Passed gates -> admit, with crisp reason
|
|
if must:
|
|
# Preferential message by priority
|
|
for a in ("creative", "berlin_local"):
|
|
if a in must and has(a):
|
|
_set_reason(f"Admit (scarce: {a}; pressure {pressure[a]:.1f})")
|
|
return True
|
|
# else any must-have present
|
|
found = [a for a in must if has(a)]
|
|
_set_reason("Admit (meets scarce requirement: " + ", ".join(found) + ")")
|
|
return True
|
|
else:
|
|
_set_reason(f"Admit (on pace; helps {', '.join(helps)})")
|
|
return True
|
|
|