root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat: add rectification event decision contract v2

Browse Source
This commit is contained in:
Jesse_Chen
2026-08-15 00:56:06 +08:00
parent 21fce9513c
commit 83fef19779
24 changed files with 3637 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/rectification/api_service.py
+63 -14
View File
@@ -4,13 +4,21 @@ from typing import Any
from uuid import NAMESPACE_URL, uuid5
from scripts.rectification.candidate_feature_service import build_candidate_feature_snapshot
from scripts.rectification.contracts import RectificationRequest
from scripts.rectification.contracts import EVENT_CONTRACT_VERSION, RectificationRequest
from scripts.rectification.decision_policy import (
EXECUTION_LEDGER_VERSION,
POLICY_VERSION,
build_candidate_decisions,
build_decision_receipt,
build_execution_ledger,
)
from scripts.rectification.diagnostics_service import run_diagnostics
from scripts.rectification.scoring_service import (
ALGORITHM_VERSION,
build_event_contribution_matrix,
calculation_spec,
score_from_matrix,
scoreable_request,
sha256,
)
@@ -18,25 +26,36 @@ from scripts.rectification.scoring_service import (
def candidate_features(request: RectificationRequest) -> dict[str, Any]:
spec = calculation_spec(request)
spec_hash = sha256(spec)
scoring_request = scoreable_request(request)
return {
"algorithm_version": ALGORITHM_VERSION,
"event_contract_version": EVENT_CONTRACT_VERSION,
"decision_policy_version": POLICY_VERSION,
"calculation_spec": spec,
"calculation_spec_hash": spec_hash,
"candidate_feature_snapshot": build_candidate_feature_snapshot(request, spec_hash),
"candidate_feature_snapshot": build_candidate_feature_snapshot(scoring_request, spec_hash),
"can_confirm_exact_minute": False,
}
def score_candidates(request: RectificationRequest) -> dict[str, Any]:
built = build_event_contribution_matrix(request)
rows = score_from_matrix(request, built)
scoring_request = scoreable_request(request)
built = build_event_contribution_matrix(scoring_request)
rows = score_from_matrix(scoring_request, built)
spec = calculation_spec(request)
spec_hash = sha256(spec)
diagnostics = run_diagnostics(request, rows, built)
diagnostic_values = run_diagnostics(scoring_request, rows, built)
fingerprint = sha256(request)
result_id = str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{fingerprint}"))
candidate_decisions = build_candidate_decisions(rows, result_id=result_id)
decision_receipt = build_decision_receipt(request, candidate_decisions, built, diagnostic_values)
execution_ledger = build_execution_ledger(request, built, diagnostic_values, candidate_decisions)
representative = candidate_decisions[0] if candidate_decisions else None
return {
"result_id": str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{fingerprint}")),
"result_id": result_id,
"algorithm_version": ALGORITHM_VERSION,
"event_contract_version": EVENT_CONTRACT_VERSION,
"decision_policy_version": POLICY_VERSION,
"calculation_spec": spec,
"calculation_spec_hash": spec_hash,
"candidate_scores": [{
@@ -45,17 +64,32 @@ def score_candidates(request: RectificationRequest) -> dict[str, Any]:
"supporting_event_ids": [item["event_id"] for item in row["evidence"] if item["points"] > 0],
"conflicting_event_ids": [item["event_id"] for item in row["evidence"] if item["points"] < 0],
} for row in rows],
"candidate_decisions": candidate_decisions,
"candidate_decision_receipt": decision_receipt,
"decision_receipt": decision_receipt,
"execution_ledger_version": EXECUTION_LEDGER_VERSION,
"execution_ledger": execution_ledger,
"event_contribution_matrix": built["matrix"],
"candidate_feature_snapshot": build_candidate_feature_snapshot(request, spec_hash, built.get("static_contexts")),
"diagnostics": diagnostics,
"candidate_feature_snapshot": build_candidate_feature_snapshot(
scoring_request, spec_hash, built.get("static_contexts")
),
"diagnostics": diagnostic_values,
"robustness": {
"neighbor_support_minutes": diagnostics["neighbor_support_minutes"],
"leave_one_out_retention_rate": diagnostics["leave_one_event_out_retention_rate"],
"leave_one_domain_out_retention_rate": diagnostics["leave_one_domain_out_retention_rate"],
"date_sensitivity_retention_rate": diagnostics["date_sensitivity_retention_rate"],
"neighbor_support_minutes": diagnostic_values.get("neighbor_support_minutes", 0),
"leave_one_out_retention_rate": diagnostic_values.get("leave_one_event_out_retention_rate", 0),
"leave_one_domain_out_retention_rate": diagnostic_values.get("leave_one_domain_out_retention_rate", 0),
"date_sensitivity_retention_rate": diagnostic_values.get("date_sensitivity_retention_rate", 0),
},
"missing_layers": built["missing_layers"],
"can_confirm_exact_minute": False,
"display_allowed": decision_receipt["display_allowed"],
"selection_allowed": decision_receipt["selection_allowed"],
"acceptance_allowed": decision_receipt["acceptance_allowed"],
"confirmation_allowed": decision_receipt["confirmation_allowed"],
"representative_candidate_id": representative["candidate_id"] if representative else None,
"representative_time": representative["time"] if representative else None,
"overall_confidence": decision_receipt["overall_confidence"],
"margin_percent": decision_receipt["margin_percent"],
"can_confirm_exact_minute": decision_receipt["confirmation_allowed"],
}
@@ -64,8 +98,23 @@ def diagnostics(request: RectificationRequest) -> dict[str, Any]:
return {
"result_id": scored["result_id"],
"algorithm_version": scored["algorithm_version"],
"event_contract_version": scored["event_contract_version"],
"decision_policy_version": scored["decision_policy_version"],
"calculation_spec_hash": scored["calculation_spec_hash"],
"candidate_decisions": scored["candidate_decisions"],
"candidate_decision_receipt": scored["candidate_decision_receipt"],
"decision_receipt": scored["decision_receipt"],
"execution_ledger_version": scored["execution_ledger_version"],
"execution_ledger": scored["execution_ledger"],
"diagnostics": scored["diagnostics"],
"missing_layers": scored["missing_layers"],
"can_confirm_exact_minute": False,
"display_allowed": scored["display_allowed"],
"selection_allowed": scored["selection_allowed"],
"acceptance_allowed": scored["acceptance_allowed"],
"confirmation_allowed": scored["confirmation_allowed"],
"representative_candidate_id": scored["representative_candidate_id"],
"representative_time": scored["representative_time"],
"overall_confidence": scored["overall_confidence"],
"margin_percent": scored["margin_percent"],
"can_confirm_exact_minute": scored["confirmation_allowed"],
}
scripts/rectification/contracts.py
+55 -10
View File
@@ -8,19 +8,41 @@ from typing import Any, Literal, NotRequired, TypedDict, cast
DatePrecision = Literal["day", "month", "quarter", "year", "range"]
EVENT_CONTRACT_VERSION = "rectification-event-contract-v2"
EVENT_KINDS: dict[str, frozenset[str]] = {
"education": frozenset({
"education_start", "education_completion", "education_interruption", "education_change",
"education_milestone", # v1 compatibility
}),
"career": frozenset({
"career_entry", "career_change", "promotion", "career_pressure", "career_exit", "business_start",
}),
"relationship": frozenset({
"relationship_start", "relationship_commitment", "relationship_separation", "relationship_end",
"relationship_change", # v1 compatibility
}),
"relocation": frozenset({"relocation", "foreign_move", "return", "home_change"}),
"finance": frozenset({"finance_gain", "finance_loss", "income_change", "asset_change", "finance_change"}),
"health": frozenset({"self_health_event", "pressure_period"}),
"health_pressure": frozenset({"self_health_event", "pressure_period"}), # v1 domain compatibility
"family": frozenset({"family_event"}),
"other": frozenset({"other"}),
}
BACKGROUND_EVENT_KINDS = frozenset({"family_event", "other"})
SCOREABLE_EVENT_KINDS: dict[str, frozenset[str]] = {
"education": frozenset({"education_milestone"}),
"relocation": frozenset({"relocation"}),
"relationship": frozenset({"relationship_start", "relationship_change"}),
"career": frozenset({"career_change"}),
"finance": frozenset({"finance_change"}),
"health_pressure": frozenset({"self_health_event"}),
domain: frozenset(kind for kind in kinds if kind not in BACKGROUND_EVENT_KINDS)
for domain, kinds in EVENT_KINDS.items()
if any(kind not in BACKGROUND_EVENT_KINDS for kind in kinds)
}
DATE_PRECISIONS = frozenset({"day", "month", "quarter", "year", "range"})
_BIRTH_TIME_SOURCES = frozenset({"hospital_record", "family_exact", "approximate", "period_only", "unknown", "legacy_import"})
_LOCAL_TIME_STATUSES = frozenset({"resolved", "not_provided", "ambiguous", "nonexistent"})
_REQUEST_PROVENANCE_FIELDS = frozenset({"birth_time_source", "timezone_id", "timezone_source", "local_time_status"})
_EVENT_PROVENANCE_FIELDS = frozenset({"date_source", "date_reliability", "date_corroboration", "date_conflict_status"})
_EVENT_PROVENANCE_FIELDS = frozenset({
"date_source", "date_reliability", "date_corroboration", "date_conflict_status",
"source_turn_id", "subject",
})
_REQUEST_FIELDS = frozenset({"birth_date", "start_time", "end_time", "lat", "lon", "tz", "events"}) | _REQUEST_PROVENANCE_FIELDS
_EVENT_FIELDS = frozenset({"id", "domain", "event_kind", "date_start", "date_end", "precision", "summary"}) | _EVENT_PROVENANCE_FIELDS
_CLOCK = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d\Z")
@@ -38,6 +60,8 @@ class LifeEvent(TypedDict):
date_reliability: NotRequired[str | None]
date_corroboration: NotRequired[str | None]
date_conflict_status: NotRequired[str | None]
source_turn_id: NotRequired[str | None]
subject: NotRequired[Literal["self", "family", "other"]]
class RectificationRequest(TypedDict):
@@ -57,6 +81,10 @@ class RectificationRequest(TypedDict):
JsonObject = dict[str, Any]
def is_scoreable_event(event: LifeEvent) -> bool:
return event["event_kind"] not in BACKGROUND_EVENT_KINDS
def _bounded_number(body: dict[str, Any], name: str, minimum: float, maximum: float) -> float:
value = body.get(name)
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)):
@@ -123,9 +151,9 @@ def normalize_rectification_request(body: Any, *, today: date | None = None) ->
except (ValueError, AttributeError) as exc:
raise ValueError(f"events[{index}].id must be a UUID") from exc
domain, event_kind = raw_event.get("domain"), raw_event.get("event_kind")
if domain not in SCOREABLE_EVENT_KINDS:
raise ValueError(f"events[{index}].domain is not scoreable")
if event_kind not in SCOREABLE_EVENT_KINDS[cast(str, domain)]:
if domain not in EVENT_KINDS:
raise ValueError(f"events[{index}].domain is invalid")
if event_kind not in EVENT_KINDS[cast(str, domain)]:
raise ValueError(f"events[{index}].event_kind does not match domain")
precision = raw_event.get("precision")
if precision not in DATE_PRECISIONS:
@@ -139,6 +167,13 @@ def normalize_rectification_request(body: Any, *, today: date | None = None) ->
summary = raw_event.get("summary", "")
if not isinstance(summary, str) or len(summary) > 1_000:
raise ValueError(f"events[{index}].summary must be a string up to 1000 characters")
subject = raw_event.get("subject")
if subject is None:
subject = "family" if domain == "family" else "other" if domain == "other" else "self"
if subject not in {"self", "family", "other"}:
raise ValueError(f"events[{index}].subject is invalid")
if event_kind not in BACKGROUND_EVENT_KINDS and subject != "self":
raise ValueError(f"events[{index}].subject must be self for scoreable events")
cleaned_event: dict[str, Any] = {
"id": event_id,
"domain": cast(str, domain),
@@ -147,11 +182,21 @@ def normalize_rectification_request(body: Any, *, today: date | None = None) ->
"date_end": end_day.isoformat(),
"precision": cast(DatePrecision, precision),
"summary": summary.strip(),
"subject": cast(Literal["self", "family", "other"], subject),
}
_copy_nullable_text(raw_event, cleaned_event, "date_source", f"events[{index}].date_source", 120)
_copy_nullable_text(raw_event, cleaned_event, "date_reliability", f"events[{index}].date_reliability", 120)
_copy_nullable_text(raw_event, cleaned_event, "date_corroboration", f"events[{index}].date_corroboration", 1_000)
_copy_nullable_text(raw_event, cleaned_event, "date_conflict_status", f"events[{index}].date_conflict_status", 120)
if "source_turn_id" in raw_event:
source_turn_id = raw_event.get("source_turn_id")
if source_turn_id is None:
cleaned_event["source_turn_id"] = None
else:
try:
cleaned_event["source_turn_id"] = str(uuid.UUID(str(source_turn_id)))
except (ValueError, AttributeError) as exc:
raise ValueError(f"events[{index}].source_turn_id must be null or a UUID") from exc
cleaned_events.append(cast(LifeEvent, cleaned_event))
cleaned_request: dict[str, Any] = {
scripts/rectification/decision_policy.py
+313
View File
@@ -0,0 +1,313 @@
from __future__ import annotations
from collections.abc import Sequence
from decimal import ROUND_FLOOR, ROUND_HALF_UP, Decimal
from typing import Any
from uuid import NAMESPACE_URL, uuid5
from scripts.active_rectification_events import CandidateScoreRow
from scripts.rectification.contracts import (
EVENT_CONTRACT_VERSION,
RectificationRequest,
is_scoreable_event,
)
from scripts.rectification.scoring_service import precision_weight
POLICY_VERSION = "rectification-candidate-policy-v2"
RECEIPT_VERSION = "candidate-decision-receipt-v2"
EXECUTION_LEDGER_VERSION = "rectification-execution-ledger-v2"
SCORE_QUANTUM = Decimal("0.0001")
TIE_ABSOLUTE_TOLERANCE = Decimal("0.0001")
MIN_ACCEPTANCE_EVENTS = 3
MIN_ACCEPTANCE_DOMAINS = 2
MIN_DATE_QUALITY_MEAN = Decimal("0.65")
MIN_DIAGNOSTIC_RETENTION = Decimal("0.75")
MIN_ACCEPTANCE_MARGIN_PERCENT = Decimal("10")
_BAD_DATE_RELIABILITY = frozenset({"low", "uncertain", "unreliable"})
_CLEAR_DATE_CONFLICT = frozenset({"", "none", "resolved", "no_conflict"})
def _decimal(value: Any, default: str = "0") -> Decimal:
if isinstance(value, bool):
return Decimal(default)
try:
return Decimal(str(value))
except Exception:
return Decimal(default)
def _quantized_score(row: CandidateScoreRow) -> Decimal:
return _decimal(row.get("score")).quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)
def _relative_support(scores: Sequence[Decimal]) -> list[int]:
if not scores:
return []
weights = [max(score, Decimal(0)) for score in scores]
total = sum(weights, Decimal(0))
if total == 0:
base, remainder = divmod(100, len(scores))
return [base + (1 if index < remainder else 0) for index in range(len(scores))]
exact = [weight * Decimal(100) / total for weight in weights]
floors = [int(value.to_integral_value(rounding=ROUND_FLOOR)) for value in exact]
remaining = 100 - sum(floors)
order = sorted(
range(len(scores)),
key=lambda index: (-(exact[index] - Decimal(floors[index])), index),
)
for index in order[:remaining]:
floors[index] += 1
return floors
def build_candidate_decisions(
rows: Sequence[CandidateScoreRow],
*,
result_id: str,
) -> list[dict[str, Any]]:
ranked = sorted(rows, key=lambda row: (-_quantized_score(row), row["time"]))
public_rows = ranked[:3]
supports = _relative_support([_quantized_score(row) for row in public_rows])
all_scores = [_quantized_score(row) for row in ranked]
decisions = []
for index, row in enumerate(public_rows):
score = _quantized_score(row)
tied_minute_count = sum(
abs(score - other) <= TIE_ABSOLUTE_TOLERANCE
for other in all_scores
)
decisions.append({
"candidate_id": str(uuid5(NAMESPACE_URL, f"{POLICY_VERSION}:{result_id}:{row['time']}")),
"rank": index + 1,
"time": row["time"],
"relative_support": supports[index],
"tied_minute_count": tied_minute_count,
})
return decisions
def _gate(passed: bool, **details: Any) -> dict[str, Any]:
return {"passed": passed, **details}
def _date_quality(events: Sequence[dict[str, Any]]) -> dict[str, Any]:
weights = [_decimal(precision_weight(str(event["precision"]))) for event in events]
total = sum(weights, Decimal(0))
mean = total / Decimal(len(weights)) if weights else Decimal(0)
low_reliability = sorted(
event["id"]
for event in events
if str(event.get("date_reliability") or "").strip().lower() in _BAD_DATE_RELIABILITY
)
unresolved_conflicts = sorted(
event["id"]
for event in events
if str(event.get("date_conflict_status") or "").strip().lower() not in _CLEAR_DATE_CONFLICT
)
passed = bool(events) and mean >= MIN_DATE_QUALITY_MEAN and not low_reliability and not unresolved_conflicts
return _gate(
passed,
precision_weight_total=float(total),
precision_weight_mean=float(mean.quantize(SCORE_QUANTUM, rounding=ROUND_HALF_UP)),
minimum_precision_weight_mean=float(MIN_DATE_QUALITY_MEAN),
low_reliability_event_ids=low_reliability,
unresolved_conflict_event_ids=unresolved_conflicts,
)
def _diagnostic_quality(diagnostics: dict[str, Any]) -> dict[str, Any]:
retention_names = (
"leave_one_event_out_retention_rate",
"leave_one_domain_out_retention_rate",
"date_sensitivity_retention_rate",
)
retentions = {name: _decimal(diagnostics.get(name)) for name in retention_names}
margin = _decimal(diagnostics.get("primary_secondary_margin_percent"))
passed = (
all(value >= MIN_DIAGNOSTIC_RETENTION for value in retentions.values())
and margin >= MIN_ACCEPTANCE_MARGIN_PERCENT
)
return _gate(
passed,
minimum_retention=float(MIN_DIAGNOSTIC_RETENTION),
minimum_margin_percent=float(MIN_ACCEPTANCE_MARGIN_PERCENT),
margin_percent=float(margin),
**{name: float(value) for name, value in retentions.items()},
)
def build_decision_receipt(
request: RectificationRequest,
candidate_decisions: Sequence[dict[str, Any]],
built: dict[str, Any],
diagnostics: dict[str, Any],
) -> dict[str, Any]:
scoreable_events = [event for event in request["events"] if is_scoreable_event(event)]
domains = sorted({event["domain"] for event in scoreable_events})
candidate_presence = _gate(bool(candidate_decisions), candidate_count=len(candidate_decisions))
event_quality = _gate(
len(scoreable_events) >= MIN_ACCEPTANCE_EVENTS,
scoreable_event_count=len(scoreable_events),
minimum=MIN_ACCEPTANCE_EVENTS,
)
domain_diversity = _gate(
len(domains) >= MIN_ACCEPTANCE_DOMAINS,
scoreable_domain_count=len(domains),
minimum=MIN_ACCEPTANCE_DOMAINS,
domains=domains,
)
date_quality = _date_quality(scoreable_events)
top_tied_count = candidate_decisions[0]["tied_minute_count"] if candidate_decisions else 0
unique_top = _gate(top_tied_count == 1, tied_minute_count=top_tied_count)
diagnostic_quality = _diagnostic_quality(diagnostics)
required_layers = _gate(
not built.get("missing_layers"),
missing_layers=sorted(built.get("missing_layers") or []),
)
acceptance_allowed = all((
candidate_presence["passed"],
event_quality["passed"],
domain_diversity["passed"],
date_quality["passed"],
unique_top["passed"],
diagnostic_quality["passed"],
))
margin = _decimal(diagnostics.get("primary_secondary_margin_percent"))
if acceptance_allowed and margin >= Decimal("20"):
overall_confidence = "high"
elif acceptance_allowed:
overall_confidence = "medium"
else:
overall_confidence = "low"
acceptance_reasons: list[str] = []
for passed, reason in (
(candidate_presence["passed"], "no_candidates"),
(event_quality["passed"], "insufficient_events"),
(domain_diversity["passed"], "insufficient_domain_diversity"),
(date_quality["passed"], "low_date_quality"),
(unique_top["passed"], "tied_top_score"),
(diagnostic_quality["passed"], "insufficient_diagnostic_stability"),
):
if not passed:
acceptance_reasons.append(reason)
confirmation_reasons = []
if not required_layers["passed"]:
confirmation_reasons.append("missing_mandatory_layers")
confirmation_reasons.extend(("engine_exact_confirmation_not_granted", "external_validation_not_passed"))
reasons = [*acceptance_reasons, *confirmation_reasons]
representative = candidate_decisions[0] if candidate_decisions else None
exact_confirmation = {
"passed": False,
"fail_closed": True,
"engine_granted": False,
"external_validation_status": "not_evaluated",
"required_scoreable_events": 4,
"required_scoreable_domains": 3,
"reason": "engine_and_external_validation_must_explicitly_pass",
}
receipt = {
"receipt_version": RECEIPT_VERSION,
"contract_version": "v2",
"event_contract_version": EVENT_CONTRACT_VERSION,
"policy_version": POLICY_VERSION,
"decision_policy_version": POLICY_VERSION,
"display_allowed": bool(candidate_decisions),
"selection_allowed": acceptance_allowed,
"acceptance_allowed": acceptance_allowed,
"confirmation_allowed": False,
"accept_allowed": acceptance_allowed,
"confirm_allowed": False,
"representative_candidate_id": representative["candidate_id"] if representative else None,
"representative_time": representative["time"] if representative else None,
"overall_confidence": overall_confidence,
"margin_percent": float(margin),
"reasons": reasons,
"acceptance_reasons": acceptance_reasons,
"confirmation_reasons": confirmation_reasons,
"tie_policy": {
"score_quantum": float(SCORE_QUANTUM),
"absolute_tolerance": float(TIE_ABSOLUTE_TOLERANCE),
"rounding": "ROUND_HALF_UP",
},
"gates": {
"candidate_presence": candidate_presence,
"event_quality": event_quality,
"domain_diversity": domain_diversity,
"date_quality": date_quality,
"unique_top": unique_top,
"diagnostic_quality": diagnostic_quality,
"required_layers": required_layers,
"exact_confirmation": exact_confirmation,
},
}
return receipt
def build_execution_ledger(
request: RectificationRequest,
built: dict[str, Any],
diagnostics: dict[str, Any],
candidate_decisions: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
matrix = built.get("matrix") or {}
date_sensitivity = {
item.get("event_id"): item
for item in built.get("date_sensitivity") or []
if isinstance(item, dict)
}
entries: list[dict[str, Any]] = []
all_layers: set[str] = set()
for event in request["events"]:
candidates = matrix.get(event["id"], {})
layers = sorted({
layer
for contribution in candidates.values()
for layer in contribution.get("technique_layers", [])
})
all_layers.update(layers)
sensitivity = date_sensitivity.get(event["id"], {})
scoreable = is_scoreable_event(event)
entries.append({
"ledger_version": EXECUTION_LEDGER_VERSION,
"stage": "event_scoring",
"status": "executed" if scoreable and candidates else "not_executed" if scoreable else "retained_not_scored",
"source": "python-engine",
"event_id": event["id"],
"domain": event["domain"],
"event_kind": event["event_kind"],
"date_precision": event["precision"],
"precision_weight": precision_weight(event["precision"]),
"sampled_date_count": len(sensitivity.get("sample_dates") or []),
"candidate_count": len(candidates),
"technique_layers": layers,
})
for layer in sorted(all_layers):
entries.append({
"ledger_version": EXECUTION_LEDGER_VERSION,
"stage": "technique_layer",
"method": layer,
"status": "executed",
"source": "python-engine",
})
entries.extend((
{
"ledger_version": EXECUTION_LEDGER_VERSION,
"stage": "candidate_ranking",
"status": "executed" if candidate_decisions else "not_executed",
"source": "python-decision-policy",
"candidate_count": len(candidate_decisions),
},
{
"ledger_version": EXECUTION_LEDGER_VERSION,
"stage": "diagnostics",
"status": "executed" if diagnostics else "not_executed",
"source": "python-engine",
"metrics": sorted(diagnostics),
},
))
return entries
scripts/rectification/scoring_service.py
+107 -20
View File
@@ -3,16 +3,54 @@ from __future__ import annotations
import hashlib
import json
from collections import defaultdict
from collections.abc import Callable, Sequence
from datetime import date, timedelta
from functools import lru_cache
from typing import Any, Callable, Sequence
from typing import Any
from scripts.active_rectification_event_engine import compute_candidate_static_contexts, compute_event_candidate_rows
from scripts.active_rectification_events import CandidateScoreRow
from scripts.rectification.contracts import LifeEvent, RectificationRequest
from scripts.rectification.contracts import LifeEvent, RectificationRequest, is_scoreable_event
ALGORITHM_VERSION = "rectification-v5-matrix-scoring-2"
INPUT_CONTRACT_VERSION = "rectification-calculation-spec-v4"
PRECISION_WEIGHTS = {
"day": 1.0,
"month": 0.8,
"quarter": 0.65,
"year": 0.5,
"range": 0.35,
}
_ENGINE_KIND_BY_NATIVE_KIND: dict[str, tuple[str, str]] = {
"education_start": ("education", "education_milestone"),
"education_completion": ("education", "education_milestone"),
"education_interruption": ("education", "education_milestone"),
"education_change": ("education", "education_milestone"),
"education_milestone": ("education", "education_milestone"),
"career_entry": ("career", "career_change"),
"career_change": ("career", "career_change"),
"promotion": ("career", "career_change"),
"career_pressure": ("career", "career_change"),
"career_exit": ("career", "career_change"),
"business_start": ("career", "career_change"),
"relationship_start": ("relationship", "relationship_start"),
"relationship_commitment": ("relationship", "relationship_start"),
"relationship_separation": ("relationship", "relationship_change"),
"relationship_end": ("relationship", "relationship_change"),
"relationship_change": ("relationship", "relationship_change"),
"relocation": ("relocation", "relocation"),
"foreign_move": ("relocation", "relocation"),
"return": ("relocation", "relocation"),
"home_change": ("relocation", "relocation"),
"finance_gain": ("finance", "finance_change"),
"finance_loss": ("finance", "finance_change"),
"income_change": ("finance", "finance_change"),
"asset_change": ("finance", "finance_change"),
"finance_change": ("finance", "finance_change"),
"self_health_event": ("health_pressure", "self_health_event"),
"pressure_period": ("health_pressure", "self_health_event"),
}
def _parse(value: str) -> date:
@@ -58,6 +96,7 @@ def sample_event_dates(event: LifeEvent) -> list[str]:
def _legacy_request(request: RectificationRequest, event: LifeEvent, sampled_date: str) -> dict[str, Any]:
engine_domain, _ = _ENGINE_KIND_BY_NATIVE_KIND[event["event_kind"]]
return {
"birth_date": request["birth_date"],
"start_time": request["start_time"],
@@ -66,8 +105,8 @@ def _legacy_request(request: RectificationRequest, event: LifeEvent, sampled_dat
"lon": request["lon"],
"tz": request["tz"],
"events": [{
"id": event["id"], "domain": event["domain"],
"event_kind": event.get("event_kind", event["domain"]),
"id": event["id"], "domain": engine_domain,
"event_kind": event["event_kind"],
"date": sampled_date, "precision": "day", "summary": event.get("summary", ""),
}],
}
@@ -82,57 +121,103 @@ def _cached_rows(serialized: str) -> tuple[CandidateScoreRow, ...]:
return tuple(compute_event_candidate_rows(json.loads(serialized)))
_RELATIONSHIP_SUPPORT_RULES = (
_SUPPORT_RULES = (
"functional_benefic_auxiliary",
"arudha_auxiliary",
"ashtakavarga_target_house_support_auxiliary",
"shadbala_sthana_drik_naisargika_support_auxiliary",
"controlled_transit_jupiter_domain_house",
)
_RELATIONSHIP_CHANGE_RULES = (
_PRESSURE_RULES = (
"functional_malefic_auxiliary",
"ashtakavarga_target_house_pressure_auxiliary",
"shadbala_sthana_drik_naisargika_pressure_auxiliary",
"controlled_transit_saturn_domain_house",
)
_KIND_SEMANTICS: dict[str, tuple[int, float]] = {
"education_start": (1, 1.0),
"education_completion": (1, 1.2),
"education_interruption": (-1, 1.0),
"education_change": (0, 1.0),
"career_entry": (1, 1.0),
"career_change": (0, 1.0),
"promotion": (1, 1.2),
"career_pressure": (-1, 1.0),
"career_exit": (-1, 1.2),
"business_start": (1, 1.2),
"relationship_start": (1, 1.0),
"relationship_commitment": (1, 1.2),
"relationship_separation": (-1, 1.0),
"relationship_end": (-1, 1.2),
"relationship_change": (-1, 1.0),
"relocation": (0, 1.0),
"foreign_move": (1, 1.0),
"return": (1, 0.8),
"home_change": (0, 1.0),
"finance_gain": (1, 1.0),
"finance_loss": (-1, 1.0),
"income_change": (0, 1.0),
"asset_change": (0, 1.0),
"self_health_event": (-1, 1.0),
"pressure_period": (-1, 1.2),
}
def _relationship_kind_factor(event_kind: str, rule_ids: Sequence[str]) -> float:
if event_kind not in {"relationship_start", "relationship_change"}:
return 1.0
support = sum(any(rule.endswith(marker) for marker in _RELATIONSHIP_SUPPORT_RULES) for rule in rule_ids)
change = sum(any(rule.endswith(marker) for marker in _RELATIONSHIP_CHANGE_RULES) for rule in rule_ids)
direction = support - change if event_kind == "relationship_start" else change - support
return max(0.8, min(1.2, 1 + 0.08 * direction))
def precision_weight(precision: str) -> float:
return PRECISION_WEIGHTS[precision]
def _event_kind_factor(event_kind: str, rule_ids: Sequence[str]) -> float:
direction, intensity = _KIND_SEMANTICS.get(event_kind, (0, 1.0))
support = sum(any(rule.endswith(marker) for marker in _SUPPORT_RULES) for rule in rule_ids)
pressure = sum(any(rule.endswith(marker) for marker in _PRESSURE_RULES) for rule in rule_ids)
semantic_signal = direction * (support - pressure) * intensity
return max(0.8, min(1.2, 1 + 0.08 * semantic_signal))
def _kind_adjusted_evidence(event: LifeEvent, evidence: dict[str, Any]) -> dict[str, Any]:
if event["domain"] != "relationship":
return evidence
event_kind = event["event_kind"]
rules = list(evidence["rule_ids"])
direction, _ = _KIND_SEMANTICS.get(event_kind, (0, 1.0))
semantic_label = "support" if direction > 0 else "pressure" if direction < 0 else "change"
return {
**evidence,
"rule_ids": [*rules, f"event_kind_profile:{event_kind}"],
"points": round(float(evidence["points"]) * _relationship_kind_factor(event_kind, rules), 4),
"rule_ids": [*rules, f"event_kind_profile:{event_kind}:{semantic_label}"],
"points": round(
float(evidence["points"])
* _event_kind_factor(event_kind, rules)
* precision_weight(event["precision"]),
4,
),
}
def scoreable_request(request: RectificationRequest) -> RectificationRequest:
return {**request, "events": [event for event in request["events"] if is_scoreable_event(event)]}
def build_event_contribution_matrix(
request: RectificationRequest,
row_provider: Callable[[dict[str, Any]], Sequence[CandidateScoreRow]] | None = None,
) -> dict[str, Any]:
static_contexts = None if row_provider is not None else compute_candidate_static_contexts(request)
scoring_request = scoreable_request(request)
if not scoring_request["events"]:
return {
"candidate_times": [], "matrix": {}, "date_sensitivity": [],
"missing_layers": [], "static_contexts": None,
}
static_contexts = None if row_provider is not None else compute_candidate_static_contexts(scoring_request)
provider = row_provider or (lambda value: compute_event_candidate_rows(value, static_contexts=static_contexts))
matrix: dict[str, dict[str, dict[str, Any]]] = defaultdict(dict)
missing_layers: set[str] = set()
date_sensitivity: list[dict[str, Any]] = []
candidate_grid: list[str] | None = None
for event in request["events"]:
for event in scoring_request["events"]:
samples = sample_event_dates(event)
sample_rows = []
for sampled in samples:
rows = list(provider(_legacy_request(request, event, sampled)))
rows = list(provider(_legacy_request(scoring_request, event, sampled)))
sample_rows.append([
{**row, "score": adjusted["points"], "evidence": [adjusted]}
for row in rows
@@ -184,6 +269,8 @@ def score_from_matrix(request: RectificationRequest, built: dict[str, Any]) -> l
for candidate_time in built["candidate_times"]:
evidence = []
for event in request["events"]:
if not is_scoreable_event(event):
continue
contribution = built["matrix"][event["id"]][candidate_time]
evidence.append({
"event_id": event["id"], "domain": event["domain"], "candidate_time": candidate_time,