Jyotisha/scripts/historical_event_backtest.py

#!/usr/bin/env python3
"""Reusable historical event backtest entrypoint built on strict workflow."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import mcp_server


SUPPORTED_DOMAINS = {
    "career": {
        "route": "career",
        "question": "请严格回测这条事业事件是否成立，并判断是职业状态、角色变化、升迁窗口还是项目兑现。",
    },
    "wealth": {
        "route": "finance",
        "question": "请严格回测这条财富事件是否成立，并判断更接近收入增长、到账、套现还是公众财富状态。",
    },
    "finance": {
        "route": "finance",
        "question": "请严格回测这条财富事件是否成立，并判断更接近收入增长、到账、套现还是公众财富状态。",
    },
    "marriage": {
        "route": "relationship",
        "question": "请严格回测这条婚恋事件是否成立，并判断是否达到正式关系或婚姻层。",
    },
    "relationship": {
        "route": "relationship",
        "question": "请严格回测这条婚恋事件是否成立，并判断是否达到正式关系或婚姻层。",
    },
}


def _route_for_domain(domain: str) -> dict[str, str] | None:
    return SUPPORTED_DOMAINS.get(str(domain).strip().lower())


def _load_payload(path: str) -> dict[str, Any]:
    with open(path, "r", encoding="utf-8") as handle:
        return json.load(handle)


def _event_result_class(
    verdict: str | None,
    blocked: bool,
    expected_label: str | None,
    actual_label: str | None,
) -> tuple[str, dict[str, Any]]:
    if blocked:
        return "blocked", {"reason": "strict_workflow_blocked"}
    if verdict == "high_probability_window" and actual_label and (
        not expected_label or expected_label == actual_label
    ):
        return "strong_hit", {"reason": "supported_route_and_label"}
    if verdict in {"high_probability_window", "moderate_probability_window"}:
        if expected_label and actual_label and expected_label != actual_label:
            return "weak_hit", {"reason": "label_mismatch_under_supported_route"}
        return "weak_hit", {"reason": "supported_route_without_exact_label"}
    if verdict == "weak_window_needs_confirmation":
        return "weak_hit", {"reason": "weak_window_needs_confirmation"}
    return "miss", {"reason": verdict or "insufficient_evidence"}


def _official_snapshot_summary(strict: dict[str, Any]) -> dict[str, Any]:
    present = strict.get("present_evidence") or {}
    official = present.get("vedastro_official_snapshot")
    if not isinstance(official, dict):
        return {"level": "missing", "status": "missing", "source": None}
    return {
        "level": official.get("level") or "missing",
        "status": official.get("status"),
        "source": official.get("source"),
    }


def _source_priority_mode(strict: dict[str, Any]) -> str | None:
    present = strict.get("present_evidence") or {}
    source_priority = present.get("source_priority")
    if not isinstance(source_priority, dict):
        return None
    return source_priority.get("mode")


def _run_supported_event(subject: dict[str, Any], event: dict[str, Any]) -> dict[str, Any]:
    route_info = _route_for_domain(event.get("domain", ""))
    if route_info is None:
        return {
            "id": event.get("id"),
            "date": event.get("date"),
            "domain": event.get("domain"),
            "route": None,
            "expected_label": event.get("expected_label"),
            "actual_label": None,
            "matched_expected_label": False,
            "result_class": "unsupported_domain",
            "boundary": {"reason": "route_not_yet_implemented_for_event_backtest"},
            "official_snapshot": {"level": "missing", "status": "missing", "source": None},
            "evidence": {"source_priority_mode": None, "confidence_cap": "unsupported"},
        }

    result = mcp_server.strict_workflow(
        question=route_info["question"],
        year=int(subject["year"]),
        month=int(subject["month"]),
        day=int(subject["day"]),
        hour=int(subject["hour"]),
        minute=int(subject["minute"]),
        lat=float(subject["lat"]),
        lon=float(subject["lon"]),
        tz=float(subject["tz"]),
        age=int(subject.get("age", 0)),
        transit_date=str(event["date"]),
        node_mode=str(subject.get("node_mode", "mean")),
    )
    strict = result.get("strict_workflow") if isinstance(result, dict) else {}
    if not isinstance(strict, dict):
        strict = {}

    judgement = strict.get("event_judgement") if isinstance(strict.get("event_judgement"), dict) else {}
    actual_label = judgement.get("dominant_label")
    expected_label = event.get("expected_label")
    verdict = judgement.get("verdict")
    blocked = bool(strict.get("blocked"))
    result_class, boundary = _event_result_class(verdict, blocked, expected_label, actual_label)

    return {
        "id": event.get("id"),
        "date": event.get("date"),
        "domain": event.get("domain"),
        "route": route_info["route"],
        "expected_label": expected_label,
        "actual_label": actual_label,
        "matched_expected_label": bool(expected_label and expected_label == actual_label),
        "result_class": result_class,
        "boundary": boundary,
        "official_snapshot": _official_snapshot_summary(strict),
        "evidence": {
            "verdict": verdict,
            "score": judgement.get("score"),
            "confidence_cap": strict.get("confidence_cap"),
            "missing_evidence": strict.get("missing_evidence") or [],
            "blocked_items": strict.get("blocked_items") or [],
            "conflicts": strict.get("conflicts") or [],
            "adjudication_stages": strict.get("adjudication_stages") or {},
            "multi_reference_reading_summary": strict.get("multi_reference_reading_summary") or {},
            "main_conflicts": strict.get("main_conflicts") or strict.get("conflicts") or [],
            "source_priority_mode": _source_priority_mode(strict),
            "primary_drivers": judgement.get("primary_drivers") or [],
            "secondary_context": judgement.get("secondary_context") or [],
            "technique_audit": strict.get("technique_audit") or [],
            "life_event_graph": strict.get("life_event_graph") or {},
        },
    }


def build_report(payload: dict[str, Any]) -> dict[str, Any]:
    subject = payload.get("subject") or {}
    events = payload.get("events") or []
    rows = [_run_supported_event(subject, event) for event in events]

    summary = {
        "total_events": len(rows),
        "strong_hits": sum(1 for row in rows if row["result_class"] == "strong_hit"),
        "weak_hits": sum(1 for row in rows if row["result_class"] == "weak_hit"),
        "misses": sum(1 for row in rows if row["result_class"] == "miss"),
        "blocked_events": sum(1 for row in rows if row["result_class"] == "blocked"),
        "unsupported_domain_events": sum(1 for row in rows if row["result_class"] == "unsupported_domain"),
        "official_primary_events": sum(
            1 for row in rows if row["official_snapshot"].get("level") == "primary"
        ),
    }

    return {
        "scope": "historical_event_backtest",
        "summary": summary,
        "boundary": (
            "This report measures whether current strict routes can support supplied historical events. "
            "Unsupported domains and blocked routes must not be overstated as validated predictive accuracy."
        ),
        "events": rows,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Run reusable historical-event backtest")
    parser.add_argument("--input", required=True, help="Path to local backtest payload JSON")
    args = parser.parse_args()

    report = build_report(_load_payload(args.input))
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())