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 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, is_scoreable_event ALGORITHM_VERSION = "rectification-v5-matrix-scoring-5" 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"), "family_event": ("family", "family_event"), "appearance_note": ("appearance", "appearance_note"), "birthmark_or_scar": ("appearance", "birthmark_or_scar"), } def _parse(value: str) -> date: return date.fromisoformat(value) def _iso(value: date) -> str: return value.isoformat() def _month_end(value: date) -> date: next_month = value.replace(day=28) + timedelta(days=4) return next_month - timedelta(days=next_month.day) def _even_dates(start: date, end: date, count: int) -> list[date]: if count <= 1 or start == end: return [start] span = (end - start).days return sorted({start + timedelta(days=round(span * index / (count - 1))) for index in range(count)}) def sample_event_dates(event: LifeEvent) -> list[str]: start, end = _parse(event["date_start"]), _parse(event["date_end"]) precision = event["precision"] if start > end: raise ValueError("invalid_event_date_range") if precision == "day" or start == end: return [_iso(start)] if precision == "month": middle = start.replace(day=min(15, _month_end(start).day)) return sorted({_iso(start), _iso(middle), _iso(end)}) if precision == "quarter": values: list[date] = [] cursor = start.replace(day=15) while cursor <= end and len(values) < 3: values.append(cursor) cursor = (cursor.replace(day=28) + timedelta(days=4)).replace(day=15) return [_iso(item) for item in values] or [_iso(start)] if precision == "year": return [_iso(start.replace(month=month, day=15)) for month in range(1, 13)] return [_iso(item) for item in _even_dates(start, end, 12)] 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"], "end_time": request["end_time"], "lat": request["lat"], "lon": request["lon"], "tz": request["tz"], "events": [{ "id": event["id"], "domain": engine_domain, "event_kind": event["event_kind"], "date": sampled_date, "precision": "day", "summary": event.get("summary", ""), }], } def _canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) @lru_cache(maxsize=4096) def _cached_rows(serialized: str) -> tuple[CandidateScoreRow, ...]: return tuple(compute_event_candidate_rows(json.loads(serialized))) _SUPPORT_RULES = ( "functional_benefic_auxiliary", "arudha_auxiliary", "ashtakavarga_target_house_support_auxiliary", "shadbala_sthana_drik_naisargika_support_auxiliary", "controlled_transit_jupiter_domain_house", ) _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), "family_event": (0, 1.0), "appearance_note": (0, 0.8), "birthmark_or_scar": (-1, 0.8), } def precision_weight(precision: str) -> float: return PRECISION_WEIGHTS[precision] def public_technique_layers(domain: str, rule_ids: Sequence[str]) -> list[str]: """Public methods actually computed for this event. Career always lists D1-10 and D10.""" layers = { rule.split(":", 1)[0] for rule in rule_ids if not rule.startswith(("event_kind:", "event_kind_profile:")) } if domain == "career": layers.update({"d1-rashi", "d10-dashamsa"}) elif domain == "family": layers.update({"d1-rashi", "d12-dwadashamsha", "d7-saptamsha", "d3-drekkana"}) elif domain == "education": layers.update({"d1-rashi", "d24-chaturvimshamsha", "d5-panchamsha"}) elif domain == "relocation": layers.update({"d1-rashi", "d4-chaturthamsha"}) elif domain == "finance": layers.update({"d1-rashi", "d2-hora", "d11-labhamsha"}) elif domain == "health_pressure": layers.update({"d1-rashi", "d30-trimshamsha"}) elif domain in {"appearance", "marks"}: layers.add("d1-rashi") return sorted(layers) 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]: 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}:{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]: 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 scoring_request["events"]: samples = sample_event_dates(event) sample_rows = [] for sampled in samples: rows = list(provider(_legacy_request(scoring_request, event, sampled))) sample_rows.append([ {**row, "score": adjusted["points"], "evidence": [adjusted]} for row in rows for adjusted in [_kind_adjusted_evidence(event, row["evidence"][0])] ]) grids = [[row["time"] for row in rows] for rows in sample_rows] if any(grid != grids[0] for grid in grids[1:]) or (candidate_grid is not None and grids[0] != candidate_grid): raise ValueError("candidate_grid_mismatch") candidate_grid = grids[0] winners = [] for rows in sample_rows: winners.append(max(rows, key=lambda row: row["score"])["time"]) missing_layers.update(layer for row in rows for layer in row["missing_layers"]) for index, candidate_time in enumerate(candidate_grid): evidences = [rows[index]["evidence"][0] for rows in sample_rows] points = [float(item["points"]) for item in evidences] matrix[event["id"]][candidate_time] = { "points": round(sum(points) / len(points), 4), "rule_ids": sorted({rule for item in evidences for rule in item["rule_ids"]}), "technique_layers": public_technique_layers( event["domain"], [rule for item in evidences for rule in item["rule_ids"]], ), } winner = max(set(winners), key=winners.count) mean = sum(matrix[event["id"]][time]["points"] for time in candidate_grid) / len(candidate_grid) variance = sum((matrix[event["id"]][time]["points"] - mean) ** 2 for time in candidate_grid) / len(candidate_grid) date_sensitivity.append({ "event_id": event["id"], "declared_date_range": {"start": event["date_start"], "end": event["date_end"], "precision": event["precision"]}, "sample_dates": samples, "winner_retention_rate": winners.count(winner) / len(winners), "score_variance": round(variance, 6), "sample_winners": winners, }) return { "candidate_times": candidate_grid or [], "matrix": dict(matrix), "date_sensitivity": date_sensitivity, "missing_layers": sorted(missing_layers), "static_contexts": static_contexts, } def score_from_matrix(request: RectificationRequest, built: dict[str, Any]) -> list[CandidateScoreRow]: rows: list[CandidateScoreRow] = [] 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, "rule_ids": contribution["rule_ids"], "points": contribution["points"], }) rows.append({ "time": candidate_time, "score": round(sum(item["points"] for item in evidence), 4), "evidence": evidence, "missing_layers": built["missing_layers"], }) return rows def calculation_spec(request: RectificationRequest) -> dict[str, Any]: def json_number(value: float) -> int | float: return int(value) if value.is_integer() else value spec = { "version": INPUT_CONTRACT_VERSION, "birthDate": request["birth_date"], "candidateRange": {"start": request["start_time"], "end": request["end_time"]}, "latitude": json_number(request["lat"]), "longitude": json_number(request["lon"]), "timezoneOffsetHours": json_number(request["tz"]), "ayanamsa": "lahiri", "nodeMode": "mean", "minuteStep": 1, } for source, target in ( ("birth_time_source", "birthTimeSource"), ("timezone_id", "timezoneId"), ("timezone_source", "timezoneSource"), ("local_time_status", "localTimeStatus"), ): if source in request: spec[target] = request[source] # type: ignore[literal-required] return spec def sha256(value: Any) -> str: return hashlib.sha256(_canonical(value).encode()).hexdigest()