169 lines
8.5 KiB
Python
169 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline, deterministic pilot benchmark for Phase 16.
|
|
|
|
The benchmark never performs network I/O. It evaluates the Phase 6/9/8/10 pure
|
|
functions against a versioned synthetic fixture and writes JSON suitable for CI
|
|
or trend collection.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import statistics
|
|
import sys
|
|
import time
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
API = ROOT / "apps" / "api"
|
|
FIXTURE_PATH = API / "fixtures" / "phase16.json"
|
|
DEFAULT_REPORT = ROOT / "docs" / "benchmarks" / "phase16.latest.json"
|
|
if str(API) not in sys.path:
|
|
sys.path.insert(0, str(API))
|
|
|
|
from app.contact_extractor import extract_contacts
|
|
from app.domain import match_businesses, normalize_business
|
|
from app.scoring import DEFAULT_RULES, evaluate_score
|
|
from app.website_scanner import classify_website
|
|
|
|
ACCEPTANCE_THRESHOLDS = {
|
|
"normalization_accuracy": 1.0,
|
|
"matching_precision": 0.90,
|
|
"matching_recall": 0.90,
|
|
"contact_precision": 0.85,
|
|
"contact_recall": 0.85,
|
|
"website_accuracy": 0.90,
|
|
"score_reproducible": True,
|
|
"tenant_leakage": 0,
|
|
}
|
|
|
|
|
|
def _pr(predicted: set, expected: set) -> dict:
|
|
tp = len(predicted & expected)
|
|
fp = len(predicted - expected)
|
|
fn = len(expected - predicted)
|
|
return {"true_positive": tp, "false_positive": fp, "false_negative": fn,
|
|
"precision": round(tp / (tp + fp), 4) if tp + fp else (1.0 if not expected else 0.0),
|
|
"recall": round(tp / (tp + fn), 4) if tp + fn else 1.0}
|
|
|
|
|
|
def _timed(callable_, iterations: int = 1000) -> dict:
|
|
samples = []
|
|
for _ in range(iterations):
|
|
start = time.perf_counter_ns()
|
|
callable_()
|
|
samples.append((time.perf_counter_ns() - start) / 1_000_000)
|
|
ordered = sorted(samples)
|
|
return {"iterations": iterations, "median_ms": round(statistics.median(samples), 4),
|
|
"p95_ms": round(ordered[max(0, int(iterations * 0.95) - 1)], 4),
|
|
"max_ms": round(max(samples), 4)}
|
|
|
|
|
|
def run_benchmark(fixtures: dict | None = None, *, measure_latency: bool = True) -> dict:
|
|
data = fixtures or json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
|
|
normalization_correct = 0
|
|
for case in data.get("normalization", []):
|
|
actual = normalize_business(case["input"])
|
|
fields_ok = all(actual.get(field) == expected for field, expected in case["expected"].items())
|
|
normalization_correct += int(fields_ok)
|
|
normalization_total = len(data.get("normalization", []))
|
|
|
|
matching_predicted, matching_expected = set(), set()
|
|
matching_false_positive = 0
|
|
matching_cases = []
|
|
for case in data["matching"]:
|
|
expected = set(case["expected_ids"])
|
|
predicted = {item["id"] for item in match_businesses(case["source"], case["candidates"], threshold=case["threshold"])}
|
|
matching_predicted |= predicted
|
|
matching_expected |= expected
|
|
matching_false_positive += len(predicted - expected)
|
|
matching_cases.append({"name": case["name"], "predicted_ids": sorted(predicted), "expected_ids": sorted(expected)})
|
|
matching = _pr(matching_predicted, matching_expected)
|
|
matching["cases"] = matching_cases
|
|
matching["false_positive_cases"] = matching_false_positive
|
|
|
|
contact_predicted, contact_expected = set(), set()
|
|
contact_cases = []
|
|
for case in data["contacts"]:
|
|
predicted = {(item["kind"], item["value"]) for item in extract_contacts(case["html"], case["source_url"])}
|
|
expected = {tuple(item) for item in case["expected"]}
|
|
contact_predicted |= predicted
|
|
contact_expected |= expected
|
|
contact_cases.append({"name": case["name"], "predicted_count": len(predicted), "expected_count": len(expected), "false_positives": sorted([list(x) for x in predicted - expected])})
|
|
contacts = _pr(contact_predicted, contact_expected)
|
|
contacts["cases"] = contact_cases
|
|
|
|
website_correct = sum(classify_website(c["status"], c["url"], c["body"], error=c.get("error")) == c["expected"] for c in data["websites"])
|
|
website_total = len(data["websites"])
|
|
website = {"correct": website_correct, "total": website_total, "accuracy": round(website_correct / website_total, 4) if website_total else 1.0}
|
|
|
|
score_case = data["scoring"][0]
|
|
first_score = evaluate_score(score_case["signals"], DEFAULT_RULES)
|
|
score_reproducible = all(evaluate_score(c["signals"], DEFAULT_RULES) == evaluate_score(c["signals"], DEFAULT_RULES) for c in data["scoring"])
|
|
scoring = {"reproducible": score_reproducible, "fixture_scores": [{"name": c["name"], "score": evaluate_score(c["signals"], DEFAULT_RULES)["score"], "eligible": evaluate_score(c["signals"], DEFAULT_RULES)["eligible"]} for c in data["scoring"]]}
|
|
if measure_latency:
|
|
scoring["latency"] = _timed(lambda: evaluate_score(score_case["signals"], DEFAULT_RULES))
|
|
|
|
@lru_cache(maxsize=128)
|
|
def cached_score(payload: str) -> dict:
|
|
return evaluate_score(json.loads(payload), DEFAULT_RULES)
|
|
cache_payload = json.dumps(score_case["signals"], sort_keys=True, separators=(",", ":"))
|
|
cached_score.cache_clear()
|
|
if measure_latency:
|
|
cold = _timed(lambda: (cached_score.cache_clear(), cached_score(cache_payload)), 100)
|
|
warm = _timed(lambda: cached_score(cache_payload), 1000)
|
|
cache = {"enabled": True, "cold": cold, "warm": warm, "speedup": round(cold["median_ms"] / warm["median_ms"], 4) if warm["median_ms"] else None}
|
|
else:
|
|
cache = {"enabled": True, "timing_omitted": True}
|
|
|
|
tenant_a = set(data["tenant_isolation"]["tenant_a"]["business_ids"])
|
|
tenant_b = set(data["tenant_isolation"]["tenant_b"]["business_ids"])
|
|
# This is the same allow-list operation required before a tenant query.
|
|
visible_to_a = [ident for ident in sorted(tenant_a | tenant_b) if ident in tenant_a]
|
|
leakage = len(set(visible_to_a) & tenant_b)
|
|
tenant = {"tenant_a_visible_ids": visible_to_a, "tenant_b_ids": sorted(tenant_b), "leakage": leakage, "isolated": leakage == 0}
|
|
|
|
latency = {}
|
|
if measure_latency:
|
|
latency = {
|
|
"normalization": _timed(lambda: normalize_business(data["normalization"][0]["input"])),
|
|
"matching": _timed(lambda: match_businesses(data["matching"][0]["source"], data["matching"][0]["candidates"], threshold=data["matching"][0]["threshold"])),
|
|
"contact_extraction": _timed(lambda: extract_contacts(data["contacts"][0]["html"], data["contacts"][0]["source_url"])),
|
|
"website_classification": _timed(lambda: classify_website(data["websites"][0]["status"], data["websites"][0]["url"], data["websites"][0]["body"])),
|
|
}
|
|
|
|
results = {
|
|
"normalization": {"correct": normalization_correct, "total": normalization_total, "accuracy": round(normalization_correct / normalization_total, 4) if normalization_total else 1.0},
|
|
"matching": matching, "contacts": contacts, "website_classification": website,
|
|
"scoring": scoring, "cache": cache, "tenant_isolation": tenant, "latency": latency,
|
|
}
|
|
checks = {
|
|
"normalization_accuracy": results["normalization"]["accuracy"] >= ACCEPTANCE_THRESHOLDS["normalization_accuracy"],
|
|
"matching_precision": matching["precision"] >= ACCEPTANCE_THRESHOLDS["matching_precision"],
|
|
"matching_recall": matching["recall"] >= ACCEPTANCE_THRESHOLDS["matching_recall"],
|
|
"contact_precision": contacts["precision"] >= ACCEPTANCE_THRESHOLDS["contact_precision"],
|
|
"contact_recall": contacts["recall"] >= ACCEPTANCE_THRESHOLDS["contact_recall"],
|
|
"website_accuracy": website["accuracy"] >= ACCEPTANCE_THRESHOLDS["website_accuracy"],
|
|
"score_reproducible": scoring["reproducible"], "tenant_leakage": leakage == 0,
|
|
}
|
|
return {"benchmark": "phase16", "version": 1, "fixture": str(FIXTURE_PATH.relative_to(ROOT)),
|
|
"offline": True, "limitations": data["limitations"], "acceptance_thresholds": ACCEPTANCE_THRESHOLDS,
|
|
"results": results, "checks": checks, "passed": all(checks.values())}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--output", type=Path, default=DEFAULT_REPORT)
|
|
parser.add_argument("--no-latency", action="store_true", help="omit variable timing measurements")
|
|
args = parser.parse_args()
|
|
report = run_benchmark(measure_latency=not args.no_latency)
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
return 0 if report["passed"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|