RCA
Elite Member
- Joined
- Jun 27, 2017
- Professional Status
- Certified General Appraiser
- State
- California
3) Prototype Matching Engine
3.1 Python (runnable, minimal external deps)
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Dict, Optional
import math
import time
import hashlib
from collections import deque, defaultdict
# --- Data Models --------------------------------------------------------------
@dataclass
class SLA:
target_days: int
latest_delivery_utc: str
@dataclass
class Filters:
geo_radius_miles: int
min_experience_hours: int
licenses_required: List[str]
form: str
inspection: str
@dataclass
class Order:
order_id: str
county_fips: str
product_form: str # e.g., "1004"
inspection: str # "full" | "desktop" | "exterior"
property_type: str # "SFR" | "CONDO" | "2-4"
complexity_tier: str # "standard" | "complex"
approvals_required: List[str] # e.g., ["FHA:false","VA:false"]
sla: SLA
filters: Filters
@dataclass
class Appraiser:
appraiser_id: str
licenses: List[str]
counties: List[str]
forms: List[str]
inspections: List[str]
property_types: List[str]
experience_hours: int
approvals: Dict[str, bool] # {"FHA": False, "VA": False, ...}
quality_index: float # 0..1 (1 best)
sla_overage_rate_90d: float # 0..1 (0 best)
recent_assignments_share_90d: float # 0..1 (lower is favored)
unavailable: bool = False
conflicts: bool = False
# --- Matching Engine ----------------------------------------------------------
class MatchingEngine:
def __init__(self):
self.assignment_log: List[Dict] = []
self.recent_winners: deque[str] = deque(maxlen=1000)
self.rotation_weight: Dict[str, float] = defaultdict(float)
def eligible(self, order: Order, a: Appraiser) -> bool:
if a.unavailable or a.conflicts:
return False
if order.filters.min_experience_hours and a.experience_hours < order.filters.min_experience_hours:
return False
if not set(order.filters.licenses_required).issubset(set(a.licenses)):
return False
if order.county_fips not in a.counties:
return False
if order.product_form not in a.forms:
return False
if order.inspection not in a.inspections:
return False
if order.property_type not in a.property_types:
return False
# approvals (FHA/VA) required?
for k in ["FHA", "VA"]:
req = any(f"{k}:true" == x.lower() for x in order.approvals_required)
if req and not a.approvals.get(k, False):
return False
return True
def score(self, order: Order, a: Appraiser) -> float:
# Base: inverse recent share + inverse local rotation weight
base = 1.0 / (1.0 + max(a.recent_assignments_share_90d, 1e-6))
base += 1.0 / (1.0 + max(self.rotation_weight[a.appraiser_id], 0.0))
# Competency depth heuristic by complexity
comp = min(1.0, a.experience_hours / 4000.0) # saturate ~4k hrs
if order.complexity_tier == "complex":
comp *= 1.5
# Quality stabilizer: small cap to avoid favoritism
q_adj = min(0.15, (a.quality_index - 0.8)) # only reward above 0.8, capped
# SLA penalty: cap impact to avoid backdoor favoritism
sla_pen = min(0.2, a.sla_overage_rate_90d * 0.5)
return base + comp + q_adj - sla_pen
def match(self, order: Order, pool: List[Appraiser]) -> Dict:
elig = [a for a in pool if self.eligible(order, a)]
if not elig:
return {"status": "no_eligible", "eligible_pool_size": 0}
scored = [(self.score(order, a), a) for a in elig]
scored.sort(key=lambda t: t[0], reverse=True)
winner_score, winner = scored[0]
# update rotation memory
self.recent_winners.append(winner.appraiser_id)
self.rotation_weight[winner.appraiser_id] += 1.0
for aid in list(self.rotation_weight.keys()):
self.rotation_weight[aid] *= 0.98 # decay
audit_payload = f"{order.order_id}|{sorted([a.appraiser_id for a in elig])}|{winner.appraiser_id}|{winner_score:.5f}"
audit_hash = hashlib.sha256(audit_payload.encode()).hexdigest()
decision_card = {
"filters": {
"licenses_required": order.filters.licenses_required,
"min_experience_hours": order.filters.min_experience_hours,
"form": order.product_form,
"inspection": order.inspection
},
"eligible_pool_size": len(elig),
"top_candidates": [
{"appraiser_id": a.appraiser_id, "score": round(s, 4)}
for (s, a) in scored[:5]
],
"selection_rule": "rotation_with_capped_quality_adjustments_v1",
"audit_hash": audit_hash
}
record = {
"timestamp": int(time.time()),
"order_id": order.order_id,
"winner": winner.appraiser_id,
"decision_card": decision_card
}
self.assignment_log.append(record)
return {"status": "matched", **record}
# --- Example Usage ------------------------------------------------------------
if __name__ == "__main__":
engine = MatchingEngine()
pool = [
Appraiser("APR1", ["CA_CERTIFIED_RESIDENTIAL"], ["06081","06085"], ["1004","1073"], ["full","desktop"], ["SFR","CONDO"], 5200, {"FHA":True,"VA":False}, 0.92, 0.05, 0.02),
Appraiser("APR2", ["CA_CERTIFIED_RESIDENTIAL"], ["06081"], ["1004"], ["full"], ["SFR"], 2100, {"FHA":False,"VA":False}, 0.88, 0.02, 0.03),
Appraiser("APR3", ["CA_CERTIFIED_RESIDENTIAL"], ["06081","06075"], ["1004","2055"], ["full","exterior"], ["SFR","2-4"], 3800, {"FHA":True,"VA":True}, 0.96, 0.10, 0.01),
]
order = Order(
order_id="ORD-001",
county_fips="06081",
product_form="1004",
inspection="full",
property_type="SFR",
complexity_tier="standard",
approvals_required=["FHA:false","VA:false"],
sla=SLA(target_days=7, latest_delivery_utc="2025-10-15T00:00:00Z"),
filters=Filters(
geo_radius_miles=25,
min_experience_hours=2000,
licenses_required=["CA_CERTIFIED_RESIDENTIAL"],
form="1004",
inspection="full"
)
)
result = engine.match(order, pool)
print(result)
What this does:
- Filters by license, geography, form, inspection, property type, approvals, experience, and availability/conflicts.
- Scores with a rotation-friendly base (punishes recent over-assignment), complexity-sensitive competency boost, capped quality bonus, and capped SLA penalty.
- Produces a compact decision_card with the top candidates and an audit_hash to anchor logs.
