返回 last30days-skill
freshness.py
根目录 / skills / last30days / scripts / lib / freshness.py
1 """Deterministic, source-grounded act-time freshness verification."""
2
3 from __future__ import annotations
4
5 import hashlib
6 import re
7 from collections import Counter, defaultdict
8 from dataclasses import dataclass
9 from datetime import datetime, timezone
10 from typing import Any, Callable
11
12 from . import github, grounding, health, polymarket, schema, stocktwits
13
14
15 @dataclass(frozen=True)
16 class Claim:
17 """A conservative, machine-verifiable claim extracted from one source item."""
18
19 claim_id: str
20 candidate_id: str
21 text: str
22 source: str
23 source_item_id: str
24 source_url: str
25 source_timestamp: str | None
26 datum_kind: str
27 datum_key: str
28 original_value: Any
29
30
31 @dataclass(frozen=True)
32 class RefetchedDatum:
33 value: Any
34 url: str
35 timestamp: str | None = None
36 values: dict[str, Any] | None = None
37
38
39 Refetcher = Callable[[schema.SourceItem | None, str], RefetchedDatum | dict[str, Any] | Any]
40
41 _STATUS_PATTERN = re.compile(
42 r"\b(?P<subject>[A-Z][A-Za-z0-9&.'’/+_-]*(?:\s+[A-Z0-9][A-Za-z0-9&.'’/+_-]*){0,5})"
43 r"\s+(?:is|was|remains|became|has been)\s+"
44 r"(?P<status>open|closed|active|inactive|available|unavailable|"
45 r"approved|rejected|launched|discontinued|online|offline)\b"
46 )
47 _OPPOSITE_STATUS = {
48 "open": "closed",
49 "closed": "open",
50 "active": "inactive",
51 "inactive": "active",
52 "available": "unavailable",
53 "unavailable": "available",
54 "approved": "rejected",
55 "rejected": "approved",
56 "launched": "discontinued",
57 "discontinued": "launched",
58 "online": "offline",
59 "offline": "online",
60 }
61 _REFETCHABLE_SOURCES = frozenset({"polymarket", "github", "stocktwits"})
62 _USABLE_SOURCE_STATES = frozenset({health.OK, schema.PARTIAL})
63
64
65 def _now() -> str:
66 return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
67
68
69 def _claim_id(candidate_id: str, kind: str, key: str) -> str:
70 digest = hashlib.sha256(f"{candidate_id}\0{kind}\0{key}".encode()).hexdigest()[:12]
71 return f"claim-{digest}"
72
73
74 def _claim(
75 grounded: grounding.GroundedClaimText,
76 kind: str,
77 key: str,
78 value: Any,
79 text: str,
80 ) -> Claim:
81 item = grounded.item
82 return Claim(
83 claim_id=_claim_id(grounded.candidate_id, kind, key),
84 candidate_id=grounded.candidate_id,
85 text=text,
86 source=item.source,
87 source_item_id=item.item_id,
88 source_url=item.url,
89 source_timestamp=item.published_at,
90 datum_kind=kind,
91 datum_key=key,
92 original_value=value,
93 )
94
95
96 def extract_claims(report: schema.Report) -> list[Claim]:
97 """Extract only structured numerics/dates and tightly shaped status claims."""
98 claims: list[Claim] = []
99 item_level_repos: set[str] = set()
100 for grounded in grounding.claim_source_map(report).values():
101 item = grounded.item
102 if item.source == "polymarket":
103 outcome_pairs = item.metadata.get("outcome_prices") or []
104 outcome_counts = Counter(
105 str(pair[0]).strip().casefold()
106 for pair in outcome_pairs
107 if isinstance(pair, (list, tuple)) and len(pair) == 2
108 )
109 seen_outcomes: dict[str, int] = defaultdict(int)
110 for pair in outcome_pairs:
111 if not isinstance(pair, (list, tuple)) or len(pair) != 2:
112 continue
113 name, value = pair
114 if not isinstance(value, (int, float)) or isinstance(value, bool):
115 continue
116 key = str(name).strip()
117 if not key:
118 continue
119 normalized_key = key.casefold()
120 occurrence = seen_outcomes[normalized_key]
121 seen_outcomes[normalized_key] += 1
122 datum_key = (
123 f"{key}\x1f{occurrence}"
124 if outcome_counts[normalized_key] > 1
125 else key
126 )
127 claims.append(
128 _claim(
129 grounded,
130 "polymarket_probability",
131 datum_key,
132 float(value),
133 f"{item.title}: {key} is {float(value) * 100:g}%",
134 )
135 )
136 end_date = item.metadata.get("end_date")
137 if isinstance(end_date, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", end_date):
138 claims.append(
139 _claim(
140 grounded,
141 "polymarket_end_date",
142 "end_date",
143 end_date,
144 f"{item.title} closes {end_date}",
145 )
146 )
147 elif item.source == "github":
148 stars = item.engagement.get("stars")
149 repo = _github_repo(item)
150 if repo and isinstance(stars, (int, float)) and not isinstance(stars, bool):
151 item_level_repos.add((grounded.candidate_id, repo.casefold()))
152 claims.append(
153 _claim(
154 grounded,
155 "github_stars",
156 "stars",
157 int(stars),
158 f"{repo} has {int(stars):,} GitHub stars",
159 )
160 )
161 elif item.source == "stocktwits":
162 aggregate = item.metadata.get("sentiment_aggregate") or {}
163 pct = aggregate.get("pct_bullish") if isinstance(aggregate, dict) else None
164 symbol = str(item.metadata.get("symbol") or item.container or "").strip()
165 if symbol and isinstance(pct, (int, float)) and not isinstance(pct, bool):
166 claims.append(
167 _claim(
168 grounded,
169 "stocktwits_bullish_pct",
170 "pct_bullish",
171 float(pct),
172 f"StockTwits ${symbol} tagged sentiment is {float(pct):g}% bullish",
173 )
174 )
175
176 # Status assertions are accepted only when a short, explicit subject +
177 # copula + status occurs in the exact candidate text tied above.
178 status_text = " ".join(part for part in (grounded.title, grounded.summary) if part)
179 match = _STATUS_PATTERN.search(status_text)
180 if match:
181 subject = match.group("subject").strip()
182 status = match.group("status").lower()
183 claims.append(
184 _claim(
185 grounded,
186 "status_assertion",
187 subject.lower(),
188 status,
189 match.group(0),
190 )
191 )
192
193 claims.extend(_candidate_star_claims(report, item_level_repos))
194 return claims
195
196
197 def _candidate_star_claims(
198 report: schema.Report,
199 item_level_repos: set[tuple[str, str]],
200 ) -> list[Claim]:
201 """Emit star claims from candidate enrichment metadata.
202
203 Star enrichment attaches ``metadata["github_stars"]`` (repo -> stars)
204 after reranking, so these facts never appear on item-level engagement -
205 typically the candidate's primary item is a non-GitHub source. Each repo
206 becomes one repo-keyed claim unless the same candidate already claimed it
207 at item level; a different candidate's item-level claim never suppresses
208 this candidate's own verdict (and its inline freshness flag).
209 """
210 claims: list[Claim] = []
211 candidates_by_id = {
212 candidate.candidate_id: candidate for candidate in report.ranked_candidates
213 }
214 for grounded in grounding.claim_source_map(report).values():
215 candidate = candidates_by_id.get(grounded.candidate_id)
216 if candidate is None:
217 continue
218 stars_map = candidate.metadata.get("github_stars")
219 if not isinstance(stars_map, dict):
220 continue
221 for repo, stars in sorted(stars_map.items()):
222 if not isinstance(repo, str) or not re.fullmatch(r"[^/\s]+/[^/\s]+", repo):
223 continue
224 if isinstance(stars, bool) or not isinstance(stars, (int, float)):
225 continue
226 if (grounded.candidate_id, repo.casefold()) in item_level_repos:
227 continue
228 item = grounded.item
229 claims.append(
230 Claim(
231 claim_id=_claim_id(grounded.candidate_id, "github_stars", repo),
232 candidate_id=grounded.candidate_id,
233 text=f"{repo} has {int(stars):,} GitHub stars",
234 source="github",
235 source_item_id=item.item_id,
236 source_url=f"https://github.com/{repo}",
237 source_timestamp=item.published_at,
238 datum_kind="github_stars",
239 datum_key=repo,
240 original_value=int(stars),
241 )
242 )
243 return claims
244
245
246 def _github_repo(item: schema.SourceItem) -> str | None:
247 if item.container and re.fullmatch(r"[^/\s]+/[^/\s]+", item.container):
248 return item.container
249 match = re.match(r"https?://github\.com/([^/]+/[^/#?]+)", item.url)
250 return match.group(1).removesuffix(".git") if match else None
251
252
253 def _default_refetchers() -> dict[str, Refetcher]:
254 return {
255 "polymarket": polymarket.refetch_datum,
256 "github": github.refetch_datum,
257 "stocktwits": stocktwits.refetch_datum,
258 }
259
260
261 def _coerce_refetched(value: RefetchedDatum | dict[str, Any] | Any, fallback_url: str) -> RefetchedDatum:
262 if isinstance(value, RefetchedDatum):
263 return value
264 if isinstance(value, dict) and "value" in value:
265 return RefetchedDatum(
266 value=value["value"],
267 url=str(value.get("url") or fallback_url),
268 timestamp=value.get("timestamp"),
269 values=value.get("values") if isinstance(value.get("values"), dict) else None,
270 )
271 return RefetchedDatum(value=value, url=fallback_url)
272
273
274 def _format_verdict_value(kind: str, value: Any) -> str:
275 """Format a verdict value the way the matching claim text renders it."""
276 if kind == "polymarket_probability":
277 try:
278 return f"{float(value) * 100:g}%"
279 except (TypeError, ValueError):
280 return str(value)
281 if kind == "stocktwits_bullish_pct":
282 try:
283 return f"{float(value):g}%"
284 except (TypeError, ValueError):
285 return str(value)
286 if isinstance(value, bool):
287 return str(value)
288 if isinstance(value, int):
289 return f"{value:,}"
290 if isinstance(value, float):
291 return f"{value:g}"
292 return str(value)
293
294
295 def _values_match(claim: Claim, current: Any) -> bool:
296 if claim.datum_kind == "polymarket_probability":
297 try:
298 return abs(float(claim.original_value) - float(current)) < 0.005
299 except (TypeError, ValueError):
300 return False
301 if isinstance(claim.original_value, (int, float)) and isinstance(current, (int, float)):
302 return float(claim.original_value) == float(current)
303 return claim.original_value == current
304
305
306 def _newer_status_contradiction(
307 report: schema.Report,
308 claim: Claim,
309 ) -> schema.SourceItem | None:
310 opposite = _OPPOSITE_STATUS.get(str(claim.original_value))
311 if not opposite:
312 return None
313 subject_tokens = [
314 token.lower()
315 for token in re.findall(r"[A-Za-z0-9]+", claim.datum_key)
316 if len(token) >= 3
317 ]
318 if not subject_tokens:
319 return None
320 candidates = [
321 item
322 for items in report.items_by_source.values()
323 for item in items
324 if (item.source, item.item_id) != (claim.source, claim.source_item_id)
325 and item.published_at
326 and (not claim.source_timestamp or item.published_at > claim.source_timestamp)
327 ]
328 candidates.sort(key=lambda item: item.published_at or "", reverse=True)
329 for item in candidates:
330 text = f"{item.title} {item.snippet} {item.body}"
331 for match in _STATUS_PATTERN.finditer(text):
332 asserted_subject = [
333 token.lower()
334 for token in re.findall(r"[A-Za-z0-9]+", match.group("subject"))
335 if len(token) >= 3
336 ]
337 if asserted_subject == subject_tokens and match.group("status").lower() == opposite:
338 return item
339 return None
340
341
342 def _point_refetch_key(item: schema.SourceItem, claim: Claim) -> tuple[str, ...]:
343 """Identify the source snapshot shared by claims in one verification pass."""
344 if claim.source == "polymarket":
345 key = item.metadata.get("event_id") or item.url
346 elif claim.source == "stocktwits":
347 window = item.metadata.get("freshness_window") or {}
348 return tuple(
349 str(value or "").strip().casefold()
350 for value in (
351 claim.source,
352 item.metadata.get("symbol") or item.container or item.url,
353 window.get("depth"),
354 window.get("from_date"),
355 window.get("to_date"),
356 )
357 )
358 elif claim.source == "github":
359 key = _github_repo(item) or item.url
360 else:
361 key = item.item_id
362 return claim.source, str(key).strip().casefold()
363
364
365 def _point_verdict(
366 claim: Claim,
367 checked_at: str,
368 refreshed: RefetchedDatum,
369 ) -> schema.FreshnessVerdict:
370 """Build the current/stale verdict for a successfully re-fetched datum."""
371 matches = _values_match(claim, refreshed.value)
372 return schema.FreshnessVerdict(
373 claim_id=claim.claim_id,
374 candidate_id=claim.candidate_id,
375 claim=claim.text,
376 source=claim.source,
377 source_item_id=claim.source_item_id,
378 verdict="current" if matches else "stale",
379 checked_at=checked_at,
380 source_url=claim.source_url,
381 source_timestamp=claim.source_timestamp,
382 evidence_url=refreshed.url,
383 evidence_timestamp=refreshed.timestamp or checked_at,
384 original_value=claim.original_value,
385 current_value=refreshed.value,
386 detail=None if matches else (
387 "moved: "
388 f"{_format_verdict_value(claim.datum_kind, claim.original_value)}"
389 " -> "
390 f"{_format_verdict_value(claim.datum_kind, refreshed.value)}"
391 ),
392 )
393
394
395 def _unsupported(
396 claim: Claim,
397 checked_at: str,
398 detail: str,
399 ) -> schema.FreshnessVerdict:
400 return schema.FreshnessVerdict(
401 claim_id=claim.claim_id,
402 candidate_id=claim.candidate_id,
403 claim=claim.text,
404 source=claim.source,
405 source_item_id=claim.source_item_id,
406 verdict="unsupported",
407 checked_at=checked_at,
408 source_url=claim.source_url,
409 source_timestamp=claim.source_timestamp,
410 # No fresh evidence was obtained; the original source stays on
411 # source_url/source_timestamp and the evidence fields stay empty.
412 evidence_url="",
413 evidence_timestamp=None,
414 original_value=claim.original_value,
415 detail=detail,
416 )
417
418
419 def verify_report(
420 report: schema.Report,
421 *,
422 refetchers: dict[str, Refetcher] | None = None,
423 allow_network: bool = True,
424 checked_at: str | None = None,
425 ) -> list[schema.FreshnessVerdict]:
426 """Attach and return deterministic freshness verdicts for ``report``."""
427 checked = checked_at or _now()
428 dispatch = _default_refetchers() if refetchers is None else refetchers
429 items = {
430 (item.source, item.item_id): item
431 for source_items in report.items_by_source.values()
432 for item in source_items
433 }
434 for candidate in report.ranked_candidates:
435 for item in candidate.source_items:
436 items.setdefault((item.source, item.item_id), item)
437
438 verdicts: list[schema.FreshnessVerdict] = []
439 point_cache: dict[tuple[str, ...], tuple[str, RefetchedDatum]] = {}
440 point_errors: dict[tuple[str, ...], str] = {}
441 for claim in extract_claims(report):
442 if claim.datum_kind == "status_assertion":
443 contradiction = _newer_status_contradiction(report, claim)
444 if contradiction:
445 verdicts.append(
446 schema.FreshnessVerdict(
447 claim_id=claim.claim_id,
448 candidate_id=claim.candidate_id,
449 claim=claim.text,
450 source=claim.source,
451 source_item_id=claim.source_item_id,
452 verdict="contradicted",
453 checked_at=checked,
454 source_url=claim.source_url,
455 source_timestamp=claim.source_timestamp,
456 evidence_url=contradiction.url,
457 evidence_timestamp=contradiction.published_at,
458 original_value=claim.original_value,
459 current_value=_OPPOSITE_STATUS.get(str(claim.original_value)),
460 detail=f"Newer {contradiction.source} item disagrees",
461 )
462 )
463 else:
464 verdicts.append(
465 _unsupported(
466 claim,
467 checked,
468 "Status could not be positively re-derived from a current source",
469 )
470 )
471 continue
472
473 if claim.datum_kind == "github_stars" and claim.datum_key != "stars":
474 # Candidate-enrichment star claim: the repo slug in datum_key is
475 # the refetch subject. The datum came from post-rerank enrichment,
476 # not the github search source, so it bypasses the grounding-item
477 # lookup and the per-source outcome gate.
478 refetcher = dispatch.get("github")
479 if refetcher is None:
480 verdicts.append(
481 _unsupported(claim, checked, "No point-refetch verifier is registered")
482 )
483 continue
484 if not allow_network:
485 verdicts.append(
486 _unsupported(claim, checked, "Network verification is disabled for this run")
487 )
488 continue
489 cache_key = ("github", claim.datum_key.strip().casefold())
490 if cache_key in point_errors:
491 verdicts.append(_unsupported(claim, checked, point_errors[cache_key]))
492 continue
493 try:
494 cached = point_cache.get(cache_key)
495 if cached:
496 # Any snapshot for this repo is the star count, whether an
497 # item-level claim ("stars") or a repo-keyed one fetched it.
498 refreshed = cached[1]
499 else:
500 refreshed = _coerce_refetched(
501 refetcher(None, claim.datum_key), claim.source_url
502 )
503 point_cache[cache_key] = (claim.datum_key, refreshed)
504 verdicts.append(_point_verdict(claim, checked, refreshed))
505 except Exception as exc: # verifier failures degrade to a typed verdict
506 detail = f"Re-check failed: {exc}"
507 point_errors[cache_key] = detail
508 verdicts.append(_unsupported(claim, checked, detail))
509 continue
510
511 item = items.get((claim.source, claim.source_item_id))
512 outcome = report.source_status.get(claim.source)
513 if item is None:
514 verdicts.append(_unsupported(claim, checked, "Grounding source item is unavailable"))
515 continue
516 if outcome and outcome.state not in _USABLE_SOURCE_STATES:
517 verdicts.append(
518 _unsupported(
519 claim,
520 checked,
521 f"Source status is {outcome.state}; the datum could not be re-checked",
522 )
523 )
524 continue
525 refetcher = dispatch.get(claim.source)
526 if claim.source not in _REFETCHABLE_SOURCES or refetcher is None:
527 verdicts.append(_unsupported(claim, checked, "No point-refetch verifier is registered"))
528 continue
529 if not allow_network:
530 verdicts.append(_unsupported(claim, checked, "Network verification is disabled for this run"))
531 continue
532 cache_key = _point_refetch_key(item, claim)
533 if cache_key in point_errors:
534 verdicts.append(_unsupported(claim, checked, point_errors[cache_key]))
535 continue
536 try:
537 cached = point_cache.get(cache_key)
538 if cached and cached[0] == claim.datum_key:
539 refreshed = cached[1]
540 elif cached and cached[1].values and claim.datum_key in cached[1].values:
541 refreshed = RefetchedDatum(
542 value=cached[1].values[claim.datum_key],
543 url=cached[1].url,
544 timestamp=cached[1].timestamp,
545 values=cached[1].values,
546 )
547 elif cached:
548 verdicts.append(
549 _unsupported(
550 claim,
551 checked,
552 "Re-fetched snapshot did not include this datum",
553 )
554 )
555 continue
556 else:
557 refreshed = _coerce_refetched(refetcher(item, claim.datum_key), claim.source_url)
558 point_cache[cache_key] = (claim.datum_key, refreshed)
559 verdicts.append(_point_verdict(claim, checked, refreshed))
560 except Exception as exc: # verifier failures degrade to a typed verdict
561 detail = f"Re-check failed: {exc}"
562 point_errors[cache_key] = detail
563 verdicts.append(_unsupported(claim, checked, detail))
564
565 report.freshness_verdicts = verdicts
566 return verdicts
567
567 lines PYTHON