ai-exchange/scripts/run_analysis_decision_monitor.py

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request


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


DEFAULT_BASE_URL = "http://192.168.3.90:8000"
DEFAULT_TIMEOUT_SECONDS = 30


def main(env=None) -> int:
    env = os.environ if env is None else env
    base_url = str(env.get("ANALYSIS_DECISION_MONITOR_BASE_URL") or DEFAULT_BASE_URL)
    instrument_id = int(env.get("ANALYSIS_DECISION_MONITOR_INSTRUMENT_ID", "1"))
    timeframe = str(env.get("ANALYSIS_DECISION_MONITOR_TIMEFRAME") or "1m")
    timeout_seconds = float(env.get("ANALYSIS_DECISION_MONITOR_TIMEOUT_SECONDS", str(DEFAULT_TIMEOUT_SECONDS)))
    run_p6_tick = _get_bool(env, "ANALYSIS_DECISION_MONITOR_RUN_P6_TICK", True)
    output_json = _get_bool(env, "ANALYSIS_DECISION_MONITOR_OUTPUT_JSON", False)

    try:
        payload = run_monitor(
            base_url=base_url,
            instrument_id=instrument_id,
            timeframe=timeframe,
            timeout_seconds=timeout_seconds,
            run_p6_tick=run_p6_tick,
        )
    except Exception as exc:
        payload = {
            "passed": False,
            "generated_ts": datetime.now(timezone.utc).isoformat(),
            "instrument_id": instrument_id,
            "timeframe": timeframe,
            "base_url": base_url.rstrip("/"),
            "blockers": ["analysis_decision_monitor_exception"],
            "warnings": [],
            "error_type": exc.__class__.__name__,
            "error": str(exc),
        }

    if output_json:
        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
    else:
        print(f"Analysis/decision monitor passed: {payload.get('passed')}")
        print(f"Blockers: {', '.join(payload.get('blockers') or []) or 'none'}")
        print(f"Warnings: {', '.join(payload.get('warnings') or []) or 'none'}")
    return 0 if payload.get("passed") else 2


def run_monitor(
    *,
    base_url: str,
    instrument_id: int,
    timeframe: str,
    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
    run_p6_tick: bool = True,
) -> dict:
    base = base_url.rstrip("/")
    payloads = {
        "health": _http_get_json(f"{base}/health", timeout_seconds=timeout_seconds),
        "launch_readiness": _http_get_json(
            f"{base}/trading/launch-readiness?{urllib.parse.urlencode({'instrument_id': instrument_id, 'timeframe': timeframe})}",
            timeout_seconds=timeout_seconds,
        ),
        "broker_state": _http_get_json(
            f"{base}/trading/broker-state?{urllib.parse.urlencode({'instrument_id': instrument_id})}",
            timeout_seconds=timeout_seconds,
        ),
        "runtime_status": _http_get_json(f"{base}/trading/runtime-status", timeout_seconds=timeout_seconds),
        "decision_packet": _http_get_json(
            f"{base}/trading/decision-packet?{urllib.parse.urlencode({'instrument_id': instrument_id, 'timeframe': timeframe})}",
            timeout_seconds=timeout_seconds,
        ),
        "review_packet": _http_get_json(
            f"{base}/trading/review-packet?{urllib.parse.urlencode({'instrument_id': instrument_id})}",
            timeout_seconds=timeout_seconds,
        ),
    }
    if run_p6_tick:
        payloads["p6_governance_tick"] = _http_post_json(
            f"{base}/trading/research-council/governance-tick",
            {
                "instrument_id": instrument_id,
                "limit": 5,
                "stale_after_hours": 24,
                "auto_run_stale_cycle": False,
                "force_cycle": False,
                "timeframe": timeframe,
            },
            timeout_seconds=timeout_seconds,
        )
    return evaluate_analysis_decision_monitor_payloads(
        payloads=payloads,
        base_url=base,
        instrument_id=instrument_id,
        timeframe=timeframe,
    )


def evaluate_analysis_decision_monitor_payloads(
    *,
    payloads: dict,
    base_url: str = DEFAULT_BASE_URL,
    instrument_id: int = 1,
    timeframe: str = "1m",
) -> dict:
    checks: list[dict] = []

    health = dict(payloads.get("health") or {})
    launch = dict(payloads.get("launch_readiness") or {})
    broker = dict(payloads.get("broker_state") or {})
    runtime = dict(payloads.get("runtime_status") or {})
    decision = dict(payloads.get("decision_packet") or {})
    review = dict(payloads.get("review_packet") or {})
    p6_tick = payloads.get("p6_governance_tick")
    p6_tick = dict(p6_tick or {}) if p6_tick is not None else None

    launch_warnings = list(launch.get("warning_reasons") or [])
    launch_next_actions = list(launch.get("next_actions") or [])
    market_state = _first_market_state(runtime) or dict(launch.get("market_state") or {})
    stream_code = _path(market_state, "stream_state", "code")
    freshness_code = _path(market_state, "freshness", "code")
    gap_code = _path(market_state, "data_gap_state", "code")

    _add_check(checks, "health_ok", health.get("app") == "ai-ict", detail=f"app={health.get('app')}")
    _add_check(
        checks,
        "launch_no_blockers",
        not list(launch.get("blocking_reasons") or []),
        detail=f"blocking_reasons={launch.get('blocking_reasons') or []}",
    )
    _add_check(checks, "launch_analysis_only_ready", bool(launch.get("analysis_only_ready")))
    _add_check(checks, "launch_restricted_decision_support_ready", bool(launch.get("restricted_decision_support_ready")))
    _add_check(checks, "operator_soft_rollout_ready", bool(launch.get("operator_soft_rollout_ready")))
    _add_check(
        checks,
        "real_trade_sample_gate_deferred",
        launch.get("real_trade_sample_gate_required") is False
        and launch.get("real_trade_sample_gate_status") == "deferred_analysis_decision_only",
        detail=(
            f"required={launch.get('real_trade_sample_gate_required')}, "
            f"status={launch.get('real_trade_sample_gate_status')}"
        ),
    )
    _add_check(
        checks,
        "real_trade_sample_not_current_action",
        "real_trade_sample_verified" not in launch_warnings
        and "wait_for_and_verify_real_fill_sample" not in launch_next_actions,
        detail=f"warnings={launch_warnings}, next_actions={launch_next_actions}",
    )
    _add_check(checks, "broker_state_ok", broker.get("status") == "broker_state_ok", detail=f"status={broker.get('status')}")
    _add_check(
        checks,
        "runtime_health_ok",
        _path(runtime, "runtime_health", "level") == "ok"
        and not list(runtime.get("current_blockers") or [])
        and not bool(_path(runtime, "degraded_mode_state", "is_degraded")),
        detail=(
            f"health={_path(runtime, 'runtime_health', 'level')}, "
            f"blockers={runtime.get('current_blockers') or []}, "
            f"degraded={_path(runtime, 'degraded_mode_state', 'is_degraded')}"
        ),
    )
    _add_check(checks, "market_stream_connected", stream_code == "market_stream_connected", detail=f"stream={stream_code}")
    _add_check(checks, "market_data_fresh", freshness_code == "market_data_fresh", detail=f"freshness={freshness_code}")
    _add_check(checks, "market_gap_clear", gap_code == "market_data_gap_clear", detail=f"gap={gap_code}")
    _add_check(
        checks,
        "decision_restricted_support",
        decision.get("action_recommendation") == "restricted_decision_support",
        detail=f"action={decision.get('action_recommendation')}",
    )
    _add_check(
        checks,
        "decision_observe_only",
        decision.get("operator_mode") == "observe_only",
        detail=f"operator_mode={decision.get('operator_mode')}",
    )
    _add_check(
        checks,
        "decision_execution_not_allowed",
        decision.get("execution_allowed") in {None, False}
        and "execution_assist_not_allowed" in set(decision.get("why_allowed") or []),
        detail=f"execution_allowed={decision.get('execution_allowed')}, why_allowed={decision.get('why_allowed') or []}",
    )
    _add_check(checks, "review_packet_present", bool(review.get("decision_packet")))
    _add_check(checks, "review_trust_score_present", bool(review.get("trust_score")))
    _add_check(checks, "review_research_summary_present", bool(review.get("research_council_summary")))
    if p6_tick is not None:
        _add_check(
            checks,
            "p6_governance_tick_noop",
            p6_tick.get("status") == "no_op_healthy"
            and not bool(p6_tick.get("requires_cycle_run"))
            and not bool(p6_tick.get("requires_bounded_slice")),
            detail=(
                f"status={p6_tick.get('status')}, "
                f"requires_cycle_run={p6_tick.get('requires_cycle_run')}, "
                f"requires_bounded_slice={p6_tick.get('requires_bounded_slice')}"
            ),
        )

    blockers = [item["code"] for item in checks if not item["passed"] and item["severity"] == "blocker"]
    warnings = [item["code"] for item in checks if not item["passed"] and item["severity"] == "warning"]
    return {
        "passed": not blockers,
        "generated_ts": datetime.now(timezone.utc).isoformat(),
        "base_url": base_url.rstrip("/"),
        "instrument_id": instrument_id,
        "timeframe": timeframe,
        "phase": "analysis_decision_only",
        "blockers": blockers,
        "warnings": warnings,
        "checks": checks,
        "observed": {
            "broker_status": broker.get("status"),
            "broker_fills": len(list(broker.get("fills") or [])),
            "runtime_health": _path(runtime, "runtime_health", "level"),
            "market_stream": stream_code,
            "market_freshness": freshness_code,
            "market_gap": gap_code,
            "launch_mode": launch.get("operator_soft_rollout_mode"),
            "real_trade_sample_gate_required": launch.get("real_trade_sample_gate_required"),
            "real_trade_sample_gate_status": launch.get("real_trade_sample_gate_status"),
            "decision_action": decision.get("action_recommendation"),
            "decision_operator_mode": decision.get("operator_mode"),
            "decision_trust_classification": decision.get("trust_classification"),
            "p6_governance_tick_status": p6_tick.get("status") if p6_tick else None,
        },
    }


def _http_get_json(url: str, *, timeout_seconds: float) -> dict:
    with urllib.request.urlopen(url, timeout=timeout_seconds) as response:
        return json.loads(response.read().decode("utf-8"))


def _http_post_json(url: str, payload: dict, *, timeout_seconds: float) -> dict:
    body = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=body,
        method="POST",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
        return json.loads(response.read().decode("utf-8"))


def _add_check(checks: list[dict], code: str, passed: bool, *, detail: str = "", severity: str = "blocker") -> None:
    checks.append({"code": code, "passed": bool(passed), "severity": severity, "detail": detail})


def _first_market_state(runtime: dict) -> dict:
    latest = dict(runtime.get("latest_market_state_by_symbol") or {})
    for value in latest.values():
        if isinstance(value, dict):
            return value
    return {}


def _path(payload: dict, *parts: str):
    current = payload
    for part in parts:
        if not isinstance(current, dict):
            return None
        current = current.get(part)
    return current


def _get_bool(env, key: str, default: bool) -> bool:
    value = env.get(key)
    if value is None or value == "":
        return default
    return str(value).strip().lower() in {"1", "true", "yes", "on"}


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