#!/usr/bin/env python3 """Regression tests for reusable historical event backtest reporting.""" from __future__ import annotations from scripts import historical_event_backtest as backtest def _payload(events: list[dict]) -> dict: return { "subject": { "year": 1955, "month": 2, "day": 24, "hour": 19, "minute": 15, "lat": 37.7749, "lon": -122.4194, "tz": 8.0, "node_mode": "mean", }, "events": events, } def _strict_packet( route: str, *, verdict: str, dominant_label: str | None, score: int, blocked: bool = False, confidence_cap: str = "medium-high", missing_evidence: list[str] | None = None, official_level: str = "primary", source_priority_mode: str = "vedastro_official_primary", ) -> dict: return { "routing": {"question_type": route}, "strict_workflow": { "question_type": route, "blocked": blocked, "confidence_cap": confidence_cap, "missing_evidence": missing_evidence or [], "present_evidence": { "vedastro_official_snapshot": { "level": official_level, "source": "vedastro_official", "status": "partial" if official_level == "primary" else "fallback", }, "source_priority": { "mode": source_priority_mode, "priority": [ "vedastro_official_snapshot", "local_supplemental_modules", "local_engine_fallback_when_official_blocked", ], }, }, "event_judgement": { "event_family": route, "verdict": verdict, "dominant_label": dominant_label, "score": score, "primary_drivers": ["vimshottari_current", "narayana_current"], "secondary_context": ["functional_benefic_malefic_used"], }, "technique_audit": [ { "technique": "VedAstro Official Full Snapshot", "status": "used" if official_level == "primary" else "blocked", "role": "primary_raw_evidence", } ], "life_event_graph": { "version": "life_event_graph_v1", "route": route, "dominant_label": dominant_label, "verdict": verdict, "event_nodes": [], }, }, } def test_backtest_reports_strong_hit_with_official_snapshot_priority(monkeypatch) -> None: def fake_strict_workflow(**kwargs): assert kwargs["transit_date"] == "2019-12-15" return _strict_packet( "career", verdict="high_probability_window", dominant_label="career_status", score=84, ) monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow) report = backtest.build_report( _payload( [ { "id": "career_turn_2019", "date": "2019-12-15", "domain": "career", "expected_label": "career_status", "summary": "事业逐渐好转", } ] ) ) assert report["scope"] == "historical_event_backtest" assert report["summary"]["total_events"] == 1 assert report["summary"]["strong_hits"] == 1 assert report["summary"]["official_primary_events"] == 1 event = report["events"][0] assert event["route"] == "career" assert event["result_class"] == "strong_hit" assert event["official_snapshot"]["level"] == "primary" assert event["evidence"]["source_priority_mode"] == "vedastro_official_primary" assert event["matched_expected_label"] is True def test_backtest_marks_blocked_and_unsupported_domains(monkeypatch) -> None: def fake_strict_workflow(**kwargs): return _strict_packet( "relationship", verdict="insufficient_evidence", dominant_label=None, score=28, blocked=True, confidence_cap="low", missing_evidence=["d9_navamsa", "upapada_lagna"], official_level="fallback", source_priority_mode="local_fallback_only", ) monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow) report = backtest.build_report( _payload( [ { "id": "marriage_probe", "date": "2026-06-09", "domain": "marriage", "expected_label": "legal_marriage", }, { "id": "move_1999", "date": "1999-08-01", "domain": "move", "summary": "搬家", }, ] ) ) assert report["summary"]["blocked_events"] == 1 assert report["summary"]["unsupported_domain_events"] == 1 blocked, unsupported = report["events"] assert blocked["result_class"] == "blocked" assert blocked["boundary"]["reason"] == "strict_workflow_blocked" assert unsupported["result_class"] == "unsupported_domain" assert unsupported["boundary"]["reason"] == "route_not_yet_implemented_for_event_backtest" def test_backtest_distinguishes_weak_hit_and_miss(monkeypatch) -> None: def fake_strict_workflow(**kwargs): if kwargs["transit_date"] == "2025-02-28": return _strict_packet( "finance", verdict="moderate_probability_window", dominant_label="income_growth", score=66, ) return _strict_packet( "career", verdict="insufficient_evidence", dominant_label=None, score=22, confidence_cap="low", ) monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow) report = backtest.build_report( _payload( [ { "id": "project_end_cashflow", "date": "2025-02-28", "domain": "wealth", "expected_label": "public_wealth_status", "summary": "项目结束但收到小额定金", }, { "id": "career_false_start", "date": "2026-01-10", "domain": "career", "expected_label": "project_manifestation", "summary": "短期项目未成", }, ] ) ) assert report["summary"]["weak_hits"] == 1 assert report["summary"]["misses"] == 1 first, second = report["events"] assert first["result_class"] == "weak_hit" assert first["matched_expected_label"] is False assert first["boundary"]["reason"] == "label_mismatch_under_supported_route" assert second["result_class"] == "miss" assert second["boundary"]["reason"] == "insufficient_evidence" def test_backtest_carries_conflicts_and_blocked_items_from_strict_contract(monkeypatch) -> None: def fake_strict_workflow(**kwargs): packet = _strict_packet( "career", verdict="high_probability_window", dominant_label="career_status", score=84, ) packet["strict_workflow"]["blocked_items"] = ["official_event_radar_partial"] packet["strict_workflow"]["conflicts"] = [{"type": "official_local_dasha_conflict"}] return packet monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow) report = backtest.build_report( _payload( [ { "id": "career_turn_2019", "date": "2019-12-15", "domain": "career", "expected_label": "career_status", "summary": "事业逐渐好转", } ] ) ) assert report["events"][0]["evidence"]["blocked_items"] == ["official_event_radar_partial"] assert report["events"][0]["evidence"]["conflicts"] == [{"type": "official_local_dasha_conflict"}] def test_backtest_carries_top_reader_contract_summary_from_strict_contract(monkeypatch) -> None: def fake_strict_workflow(**kwargs): packet = _strict_packet( "career", verdict="high_probability_window", dominant_label="career_status", score=84, ) packet["strict_workflow"]["adjudication_stages"] = { "promise": {"status": "present"}, "activation": { "status": "present", "required_timing_systems": ["Vimshottari", "Narayana"], }, } packet["strict_workflow"]["multi_reference_reading_summary"] = { "root_frame": {"signal": "career_promise"}, "modifier_frame": {"functional_benefic_malefic": {"used": True}}, } packet["strict_workflow"]["main_conflicts"] = [{"type": "official_local_dasha_conflict"}] return packet monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow) report = backtest.build_report( _payload( [ { "id": "career_turn_2019", "date": "2019-12-15", "domain": "career", "expected_label": "career_status", "summary": "事业逐渐好转", } ] ) ) event = report["events"][0] assert event["evidence"]["adjudication_stages"]["activation"]["required_timing_systems"] == [ "Vimshottari", "Narayana", ] assert event["evidence"]["multi_reference_reading_summary"]["root_frame"]["signal"] == "career_promise" assert event["evidence"]["main_conflicts"] == [{"type": "official_local_dasha_conflict"}]