jesse-ux
cf972f4040
Offline v4 holdout probe. Last round's width=window result was the no-elimination metric; production still-valid ranges after six answers are 15/33/56 minutes. Adjacent merge never fires under step-2 radii. W1/W2 match baseline; W3 is uncertain after one coverage squeeze. No production clustering or scoring defaults changed.
374 lines
13 KiB
Python
374 lines
13 KiB
Python
"""Pure helpers for the cluster-width research probe. No production defaults."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from collections.abc import Iterable, Sequence
|
|
from typing import Any
|
|
|
|
from scripts.rectification.candidate_contrast import (
|
|
SIGNATURE_LAYERS,
|
|
cluster_contexts_by_signature,
|
|
context_time,
|
|
layer_value,
|
|
)
|
|
|
|
SEPARATION_LEAD = 8
|
|
DEFAULT_MAX_CLUSTERS = 64
|
|
NARROW_WIDTH = 5
|
|
|
|
|
|
def clock(value: str) -> int:
|
|
stamp = str(value)[:5]
|
|
return int(stamp[:2]) * 60 + int(stamp[3:5])
|
|
|
|
|
|
def hhmm(value: object) -> str | None:
|
|
text = str(value or "")[:5]
|
|
return text if len(text) == 5 and text[2] == ":" else None
|
|
|
|
|
|
def range_width(times: Sequence[str]) -> int | None:
|
|
clocks = sorted(clock(item) for item in times if hhmm(item))
|
|
if not clocks:
|
|
return None
|
|
return clocks[-1] - clocks[0] + 1
|
|
|
|
|
|
def range_edges(times: Sequence[str]) -> tuple[str, str] | None:
|
|
stamps = [str(item)[:5] for item in times if hhmm(item)]
|
|
if not stamps:
|
|
return None
|
|
ordered = sorted(stamps, key=clock)
|
|
return ordered[0], ordered[-1]
|
|
|
|
|
|
def shannon_entropy(weights: Iterable[float]) -> float:
|
|
values = [max(float(item), 0.0) for item in weights]
|
|
total = sum(values)
|
|
if total <= 0:
|
|
return 0.0
|
|
entropy = 0.0
|
|
for value in values:
|
|
if value <= 0:
|
|
continue
|
|
share = value / total
|
|
entropy -= share * math.log(share, 2)
|
|
return round(entropy, 6)
|
|
|
|
|
|
def peak_score(cluster: dict[str, Any], by_time: dict[str, dict[str, Any]]) -> float:
|
|
scores = [
|
|
float(by_time[time].get("score") or 0)
|
|
for time in cluster.get("times") or []
|
|
if time in by_time
|
|
]
|
|
return max(scores) if scores else 0.0
|
|
|
|
|
|
def layer_change_counts(contexts: Sequence[dict[str, Any]]) -> dict[str, dict[str, int]]:
|
|
rows: dict[str, dict[str, int]] = {}
|
|
for layer in SIGNATURE_LAYERS:
|
|
values = [layer_value(context, layer) for context in contexts]
|
|
present = [item for item in values if item is not None]
|
|
unique = set(present)
|
|
changes = sum(1 for left, right in zip(present, present[1:]) if left != right)
|
|
rows[layer] = {
|
|
"unique": len(unique),
|
|
"changes": changes,
|
|
"constant": int(len(unique) <= 1),
|
|
}
|
|
return rows
|
|
|
|
|
|
def cluster_of(true_time: str, clusters: Sequence[dict[str, Any]]) -> dict[str, Any] | None:
|
|
stamp = str(true_time)[:5]
|
|
for cluster in clusters:
|
|
times = [str(item)[:5] for item in cluster.get("times") or []]
|
|
if stamp in times:
|
|
return cluster
|
|
return None
|
|
|
|
|
|
def truth_cluster_independent(
|
|
true_time: str,
|
|
raw_clusters: Sequence[dict[str, Any]],
|
|
merged_clusters: Sequence[dict[str, Any]],
|
|
) -> bool:
|
|
raw = cluster_of(true_time, raw_clusters)
|
|
merged = cluster_of(true_time, merged_clusters)
|
|
if raw is None or merged is None:
|
|
return False
|
|
return set(raw.get("times") or []) == set(merged.get("times") or [])
|
|
|
|
|
|
def merge_adjacent_traced(
|
|
clusters: Sequence[dict[str, Any]],
|
|
by_time: dict[str, dict[str, Any]],
|
|
max_clusters: int | None = DEFAULT_MAX_CLUSTERS,
|
|
max_peak_delta: float | None = None,
|
|
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
|
"""Keep every minute. Merge adjacent weak clusters when over the cap.
|
|
|
|
If ``max_peak_delta`` is set, pairs whose peak-score gap exceeds that
|
|
threshold are not merged, even if that leaves more than ``max_clusters``.
|
|
``max_clusters is None`` means never merge.
|
|
"""
|
|
work = [
|
|
{
|
|
**cluster,
|
|
"times": list(cluster.get("times") or []),
|
|
"contexts": list(cluster.get("contexts") or []),
|
|
}
|
|
for cluster in clusters
|
|
if cluster.get("times")
|
|
]
|
|
trace: list[dict[str, Any]] = []
|
|
cap = len(work) if max_clusters is None else max(int(max_clusters), 1)
|
|
while len(work) > cap:
|
|
best_index = None
|
|
best_key: tuple[float, float, int] | None = None
|
|
for index in range(len(work) - 1):
|
|
left = peak_score(work[index], by_time)
|
|
right = peak_score(work[index + 1], by_time)
|
|
delta = abs(left - right)
|
|
if max_peak_delta is not None and delta > max_peak_delta + 1e-12:
|
|
continue
|
|
key = (min(left, right), left + right, index)
|
|
if best_key is None or key < best_key:
|
|
best_key = key
|
|
best_index = index
|
|
if best_index is None:
|
|
break
|
|
left = work[best_index]
|
|
right = work[best_index + 1]
|
|
left_peak = peak_score(left, by_time)
|
|
right_peak = peak_score(right, by_time)
|
|
merged_times = sorted(set(left["times"] + right["times"]), key=clock)
|
|
trace.append({
|
|
"left_key": left.get("signature_key"),
|
|
"right_key": right.get("signature_key"),
|
|
"left_peak": round(left_peak, 4),
|
|
"right_peak": round(right_peak, 4),
|
|
"delta": round(abs(left_peak - right_peak), 4),
|
|
"min_peak": round(min(left_peak, right_peak), 4),
|
|
"merged_size": len(merged_times),
|
|
})
|
|
work[best_index] = {
|
|
"signature": left.get("signature"),
|
|
"signature_key": f"{left.get('signature_key')}+{right.get('signature_key')}",
|
|
"contexts": list(left.get("contexts") or []) + list(right.get("contexts") or []),
|
|
"times": merged_times,
|
|
"representative_time": merged_times[len(merged_times) // 2],
|
|
"representative": left.get("representative") or right.get("representative"),
|
|
}
|
|
del work[best_index + 1]
|
|
return work, trace
|
|
|
|
|
|
def public_from_clusters(
|
|
clusters: Sequence[dict[str, Any]],
|
|
rows: Sequence[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
by_time = {stamp: row for row in rows if (stamp := hhmm(row.get("time")))}
|
|
representatives: list[dict[str, Any]] = []
|
|
for cluster in clusters:
|
|
members = [by_time[time] for time in cluster.get("times") or [] if time in by_time]
|
|
if not members:
|
|
continue
|
|
best = max(members, key=lambda row: (float(row.get("score") or 0), str(row.get("time"))))
|
|
times = [time for time in cluster.get("times") or [] if time in by_time]
|
|
representatives.append({
|
|
**best,
|
|
"cluster_times": times,
|
|
"signature_key": cluster.get("signature_key"),
|
|
})
|
|
representatives.sort(key=lambda row: (-float(row.get("score") or 0), str(row.get("time"))))
|
|
return representatives
|
|
|
|
|
|
def raw_signature_clusters(contexts: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return cluster_contexts_by_signature(contexts)
|
|
|
|
|
|
def still_valid_public(
|
|
public: Sequence[dict[str, Any]],
|
|
scores: dict[str, float],
|
|
eliminated: set[str] | None = None,
|
|
lead: float | None = SEPARATION_LEAD,
|
|
) -> list[dict[str, Any]]:
|
|
dropped = eliminated or set()
|
|
active = [
|
|
row for row in public
|
|
if hhmm(row.get("time")) not in dropped
|
|
]
|
|
if not active:
|
|
return []
|
|
peak = max(float(scores.get(str(row.get("time"))[:5], row.get("score") or 0)) for row in active)
|
|
if lead is None:
|
|
return list(active)
|
|
return [
|
|
row for row in active
|
|
if peak - float(scores.get(str(row.get("time"))[:5], row.get("score") or 0)) < lead
|
|
]
|
|
|
|
|
|
def delivery_from_public(public: Sequence[dict[str, Any]]) -> dict[str, Any]:
|
|
times: list[str] = []
|
|
seen: set[str] = set()
|
|
for row in public:
|
|
members = row.get("cluster_times") or [str(row.get("time"))[:5]]
|
|
for time in members:
|
|
stamp = str(time)[:5]
|
|
if hhmm(stamp) and stamp not in seen:
|
|
seen.add(stamp)
|
|
times.append(stamp)
|
|
edges = range_edges(times)
|
|
return {
|
|
"times": times,
|
|
"width": range_width(times),
|
|
"start": None if edges is None else edges[0],
|
|
"end": None if edges is None else edges[1],
|
|
"cluster_count": len(public),
|
|
}
|
|
|
|
|
|
def minute_mass_from_public(
|
|
public: Sequence[dict[str, Any]],
|
|
scores: dict[str, float],
|
|
) -> dict[str, float]:
|
|
mass: dict[str, float] = {}
|
|
for row in public:
|
|
stamp = str(row.get("time"))[:5]
|
|
weight = max(float(scores.get(stamp, row.get("score") or 0)), 0.0)
|
|
for time in row.get("cluster_times") or [stamp]:
|
|
member = str(time)[:5]
|
|
if hhmm(member):
|
|
mass[member] = max(mass.get(member, 0.0), weight)
|
|
return mass
|
|
|
|
|
|
def smallest_mass_interval(
|
|
mass: dict[str, float],
|
|
coverage: float,
|
|
) -> dict[str, Any]:
|
|
"""Smallest contiguous clock span whose non-negative mass share >= coverage."""
|
|
items = sorted(
|
|
((time, weight) for time, weight in mass.items() if hhmm(time) and weight > 0),
|
|
key=lambda item: clock(item[0]),
|
|
)
|
|
empty = {
|
|
"times": [],
|
|
"width": None,
|
|
"start": None,
|
|
"end": None,
|
|
"mass_share": 0.0,
|
|
"covered": False,
|
|
}
|
|
if not items:
|
|
return empty
|
|
total = sum(weight for _time, weight in items)
|
|
if total <= 0:
|
|
return empty
|
|
need = total * float(coverage)
|
|
n = len(items)
|
|
prefix = [0.0]
|
|
for _time, weight in items:
|
|
prefix.append(prefix[-1] + weight)
|
|
best_key: tuple[int, float, int] | None = None
|
|
best_span: tuple[int, int] | None = None
|
|
for left in range(n):
|
|
for right in range(left, n):
|
|
share = prefix[right + 1] - prefix[left]
|
|
if share + 1e-12 < need:
|
|
continue
|
|
width = clock(items[right][0]) - clock(items[left][0]) + 1
|
|
key = (width, -share, left)
|
|
if best_key is None or key < best_key:
|
|
best_key = key
|
|
best_span = (left, right)
|
|
if best_span is None:
|
|
times = [time for time, _weight in items]
|
|
edges = range_edges(times)
|
|
return {
|
|
"times": times,
|
|
"width": range_width(times),
|
|
"start": None if edges is None else edges[0],
|
|
"end": None if edges is None else edges[1],
|
|
"mass_share": 1.0,
|
|
"covered": True,
|
|
}
|
|
left, right = best_span
|
|
times = [items[index][0] for index in range(left, right + 1)]
|
|
share = (prefix[right + 1] - prefix[left]) / total
|
|
edges = range_edges(times)
|
|
return {
|
|
"times": times,
|
|
"width": range_width(times),
|
|
"start": None if edges is None else edges[0],
|
|
"end": None if edges is None else edges[1],
|
|
"mass_share": round(share, 6),
|
|
"covered": True,
|
|
}
|
|
|
|
|
|
def time_in_delivery(true_time: str, times: Sequence[str]) -> bool:
|
|
stamp = str(true_time)[:5]
|
|
if stamp in {str(item)[:5] for item in times}:
|
|
return True
|
|
clocks = [clock(item) for item in times if hhmm(item)]
|
|
if not clocks:
|
|
return False
|
|
point = clock(stamp)
|
|
return min(clocks) <= point <= max(clocks)
|
|
|
|
|
|
def top1_from_public(public: Sequence[dict[str, Any]], true_time: str) -> bool:
|
|
if not public:
|
|
return False
|
|
scores = [float(row.get("score") or 0) for row in public]
|
|
best = max(scores)
|
|
leaders = [row for row in public if abs(float(row.get("score") or 0) - best) <= 1e-9]
|
|
stamp = str(true_time)[:5]
|
|
for row in leaders:
|
|
members = [str(item)[:5] for item in (row.get("cluster_times") or [str(row.get("time"))[:5]])]
|
|
if stamp in members or str(row.get("time"))[:5] == stamp:
|
|
return True
|
|
return False
|
|
|
|
|
|
def tied_first(public: Sequence[dict[str, Any]]) -> bool:
|
|
if len(public) < 2:
|
|
return False
|
|
scores = sorted({round(float(row.get("score") or 0), 4) for row in public}, reverse=True)
|
|
if len(scores) < 2:
|
|
return True
|
|
return abs(scores[0] - scores[1]) <= 1e-9
|
|
|
|
|
|
def metrics_bundle(
|
|
*,
|
|
public: Sequence[dict[str, Any]],
|
|
true_time: str,
|
|
window_times: Sequence[str],
|
|
delivery_times: Sequence[str],
|
|
delivery_width: int | None,
|
|
independent: bool,
|
|
entropy_scores: Sequence[float] | None = None,
|
|
) -> dict[str, Any]:
|
|
width = delivery_width if delivery_width is not None else range_width(list(delivery_times))
|
|
covered = time_in_delivery(true_time, delivery_times)
|
|
scores = entropy_scores if entropy_scores is not None else [float(row.get("score") or 0) for row in public]
|
|
return {
|
|
"top1_hit": top1_from_public(public, true_time),
|
|
"coverage": covered,
|
|
"width": width,
|
|
"tie": tied_first(public),
|
|
"entropy": shannon_entropy(max(score, 0.0) for score in scores),
|
|
"truth_squeezed": not covered,
|
|
"too_narrow": width is not None and width <= NARROW_WIDTH and covered,
|
|
"public_count": len(public),
|
|
"independent": independent,
|
|
"window_width": range_width(list(window_times)),
|
|
}
|