Add career VedAstro radar

scripts/career_vedastro_radar.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""VedAstro-assisted career timing radar.
+
+External VedAstro signals are secondary evidence only. They do not change
+local scores, dominant labels, or final career/prashna adjudication by
+themselves.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+import vedastro_service_adapter
+
+
+def build_career_radar_packet(case: dict[str, Any], *, start_date: str, end_date: str, case_id: str = "user_chart") -> dict[str, Any]:
+    result = vedastro_service_adapter.run_range_scan_for_case(
+        case,
+        "career",
+        start_date,
+        end_date,
+        case_id=case_id,
+    )
+    policy = result.get("adjudicator_policy") if isinstance(result.get("adjudicator_policy"), dict) else {}
+    can_change_score = bool(policy.get("can_change_score", False))
+    status = result.get("status", "blocked")
+    return {
+        "scope": "career_vedastro_radar",
+        "status": "ok" if status == "ok" else "blocked",
+        "blocked_reason": None if status == "ok" else result.get("reason") or status,
+        "domain": "career",
+        "adjudicator_use": "secondary_evidence_only",
+        "can_change_score": can_change_score,
+        "can_set_final_verdict": False,
+        "start_date": start_date,
+        "end_date": end_date,
+        "vedastro_range_scan_result": result,
+        "technique_audit_row": {
+            "technique": "VedAstro Career Range Scan",
+            "used": status == "ok",
+            "status": status,
+            "role": "external_secondary_evidence",
+            "confidence_effect": "raises_attention_only_not_final_score" if status == "ok" else "blocked_no_effect",
+        },
+    }
+
+
+def _load_case(path: Path) -> dict[str, Any]:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--case-json", type=Path, required=True)
+    parser.add_argument("--start-date", required=True)
+    parser.add_argument("--end-date", required=True)
+    parser.add_argument("--case-id", default="user_chart")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    packet = build_career_radar_packet(
+        _load_case(args.case_json),
+        start_date=args.start_date,
+        end_date=args.end_date,
+        case_id=args.case_id,
+    )
+    text = json.dumps(packet, ensure_ascii=False, indent=2, sort_keys=True)
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(text + "\n", encoding="utf-8")
+    print(text)
+    return 0 if packet["status"] == "ok" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_career_vedastro_radar.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+import career_vedastro_radar  # noqa: E402
+
+
+def test_career_radar_wraps_vedastro_range_scan_as_secondary_evidence(monkeypatch) -> None:
+    def fake_run(case, domain, start_date, end_date, case_id="user_chart"):
+        return {
+            "status": "ok",
+            "domain": domain,
+            "operation": "range_scan",
+            "evidence_ledger": [{"event_id": "CareerExpansionWindow", "date": "2026-09-01"}],
+            "adjudicator_policy": {"can_change_score": False},
+        }
+
+    monkeypatch.setattr(career_vedastro_radar.vedastro_service_adapter, "run_range_scan_for_case", fake_run)
+
+    packet = career_vedastro_radar.build_career_radar_packet(
+        {"year": 1990, "month": 1, "day": 1, "hour": 12, "minute": 0, "lat": 36.4, "lon": 114.2, "tz": 8},
+        start_date="2026-07-16",
+        end_date="2026-12-31",
+    )
+
+    assert packet["status"] == "ok"
+    assert packet["domain"] == "career"
+    assert packet["adjudicator_use"] == "secondary_evidence_only"
+    assert packet["can_change_score"] is False
+    assert packet["vedastro_range_scan_result"]["evidence_ledger"][0]["event_id"] == "CareerExpansionWindow"
+
+
+def test_career_radar_preserves_blocked_boundary(monkeypatch) -> None:
+    def fake_run(case, domain, start_date, end_date, case_id="user_chart"):
+        return {"status": "blocked", "reason": "official_endpoint_not_configured", "evidence_ledger": []}
+
+    monkeypatch.setattr(career_vedastro_radar.vedastro_service_adapter, "run_range_scan_for_case", fake_run)
+
+    packet = career_vedastro_radar.build_career_radar_packet(
+        {"year": 1990, "month": 1, "day": 1, "hour": 12, "minute": 0, "lat": 36.4, "lon": 114.2, "tz": 8},
+        start_date="2026-07-16",
+        end_date="2026-12-31",
+    )
+
+    assert packet["status"] == "blocked"
+    assert packet["can_change_score"] is False
+    assert packet["blocked_reason"] == "official_endpoint_not_configured"