ai-exchange/scripts/run_readonly_live_analysis_smoke.py

from __future__ import annotations

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


DEFAULT_BASE_URL = "http://192.168.3.90:8000"
DEFAULT_TIMEOUT_SECONDS = 20.0
DEFAULT_ATTEMPTS = 2
DEFAULT_SLEEP_SECONDS = 2.0


ENDPOINTS = (
    {
        "name": "health",
        "method": "GET",
        "path": "/health",
        "expected_statuses": (200,),
    },
    {
        "name": "runtime_status",
        "method": "GET",
        "path": "/trading/runtime-status",
        "expected_statuses": (200,),
    },
    {
        "name": "launch_readiness",
        "method": "GET",
        "path": "/trading/launch-readiness?instrument_id={instrument_id}&timeframe={timeframe}",
        "expected_statuses": (200,),
        "payload_check": "readonly_launch_ready",
    },
    {
        "name": "review_packet",
        "method": "GET",
        "path": "/trading/review-packet?instrument_id={instrument_id}",
        "expected_statuses": (200,),
        "payload_check": "review_packet_available",
    },
    {
        "name": "trader_mode_payload",
        "method": "GET",
        "path": "/trading/review-packet?instrument_id={instrument_id}",
        "expected_statuses": (200,),
        "payload_check": "trader_mode_payload_available",
    },
    {
        "name": "shadow_paper_validation",
        "method": "GET",
        "path": "/trading/shadow-paper-validation?instrument_id={instrument_id}&timeframe={timeframe}",
        "expected_statuses": (200,),
        "payload_check": "shadow_paper_surface_available",
    },
    {
        "name": "weekly_review",
        "method": "GET",
        "path": "/trading/research/weekly-review?instrument_id={instrument_id}",
        "expected_statuses": (200,),
        "payload_check": "weekly_review_entrypoint_available",
    },
    {
        "name": "ops_duty_maintenance",
        "method": "GET",
        "path": "/trading/maintenance-summary?instrument_id={instrument_id}&alert_limit=20&visual_limit=20&incident_limit=20",
        "expected_statuses": (200,),
        "payload_check": "ops_duty_maintenance_available",
    },
    {
        "name": "ops_duty_unattended_harness",
        "method": "GET",
        "path": "/trading/unattended-harness?instrument_id={instrument_id}",
        "expected_statuses": (200,),
        "payload_check": "ops_duty_harness_available",
    },
    {
        "name": "chart_bridge_candles",
        "method": "GET",
        "path": "/candles?instrument_id={instrument_id}&timeframe={timeframe}&limit=3",
        "expected_statuses": (200,),
        "payload_check": "chart_bridge_candles_available",
    },
    {
        "name": "candidates",
        "method": "GET",
        "path": "/trading/candidates?instrument_id={instrument_id}&limit=5",
        "expected_statuses": (200,),
        "payload_check": "candidates_surface_available",
    },
    {
        "name": "static_app",
        "method": "GET",
        "path": "/static/app.js",
        "expected_statuses": (200,),
        "payload_check": "readonly_boundary_copy_present",
        "decode_text": True,
    },
    {
        "name": "trader_mode_page",
        "method": "GET",
        "path": "/trader-mode",
        "expected_statuses": (200,),
        "payload_check": "trader_mode_page_present",
        "decode_text": True,
    },
    {
        "name": "static_trader_mode",
        "method": "GET",
        "path": "/static/trader_mode.js",
        "expected_statuses": (200,),
        "payload_check": "trader_mode_js_present",
        "decode_text": True,
    },
)


def run_readonly_live_analysis_smoke(
    *,
    base_url: str = DEFAULT_BASE_URL,
    instrument_id: int = 1,
    timeframe: str = "1m",
    attempts: int = DEFAULT_ATTEMPTS,
    sleep_seconds: float = DEFAULT_SLEEP_SECONDS,
    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
    opener: Optional[Callable[[urllib.request.Request, float], object]] = None,
    sleeper: Optional[Callable[[float], None]] = None,
) -> dict:
    opener = opener or _urlopen_no_proxy
    sleeper = sleeper or time.sleep
    endpoint_results = []
    blockers = []
    warnings = []
    passed_checks = []
    launch_payload = {}

    for endpoint in ENDPOINTS:
        result = _call_endpoint(
            endpoint=endpoint,
            base_url=base_url,
            instrument_id=instrument_id,
            timeframe=timeframe,
            attempts=max(attempts, 1),
            sleep_seconds=max(sleep_seconds, 0.0),
            timeout_seconds=max(timeout_seconds, 0.1),
            opener=opener,
            sleeper=sleeper,
        )
        endpoint_results.append(result)
        if result["name"] == "launch_readiness":
            launch_payload = dict(result.get("payload_excerpt") or {})
        if result["status"] == "passed":
            passed_checks.append(result["name"])
        elif result["status"] == "warning":
            warnings.append(result["name"])
        else:
            blockers.append(result["name"])

    launch_summary = dict(launch_payload.get("live_analysis_assistant_summary") or {})
    ready = bool(launch_payload.get("live_analysis_assistant_ready") or launch_summary.get("ready_for_readonly_live_analysis"))
    must_not_trade = bool(launch_summary.get("must_not_trade", True))
    blocked_reason = blockers[0] if blockers else ""
    return {
        "status": "blocked" if blockers else "passed",
        "mode": "readonly_live_analysis_smoke",
        "generated_ts": datetime.now(timezone.utc).isoformat(),
        "base_url": base_url.rstrip("/"),
        "instrument_id": instrument_id,
        "timeframe": timeframe,
        "ready_for_readonly_live_analysis": ready,
        "must_not_trade": must_not_trade,
        "real_trade_required": False,
        "trade_permission_required": False,
        "submit_cancel_or_live_execution_called": False,
        "blocked_reason": blocked_reason,
        "warnings": warnings,
        "blockers": blockers,
        "passed_checks": passed_checks,
        "endpoints": endpoint_results,
    }


def _call_endpoint(
    *,
    endpoint: dict,
    base_url: str,
    instrument_id: int,
    timeframe: str,
    attempts: int,
    sleep_seconds: float,
    timeout_seconds: float,
    opener,
    sleeper,
) -> dict:
    name = endpoint["name"]
    method = endpoint["method"]
    path = endpoint["path"].format(
        instrument_id=urllib.parse.quote(str(instrument_id)),
        timeframe=urllib.parse.quote(str(timeframe or "1m")),
    )
    url = base_url.rstrip("/") + path
    request = urllib.request.Request(url, headers={"Accept": "application/json"}, method=method)
    expected_statuses = tuple(endpoint.get("expected_statuses") or (200,))
    last_error = None

    for attempt in range(1, attempts + 1):
        try:
            with opener(request, timeout_seconds) as response:
                raw_body = response.read()
            payload = _decode_payload(raw_body, decode_text=bool(endpoint.get("decode_text")))
            status_code = int(getattr(response, "status", 0) or 0)
            return _classify_response(
                endpoint=endpoint,
                path=path,
                status_code=status_code,
                expected_statuses=expected_statuses,
                body_size=len(raw_body),
                payload=payload,
                attempt=attempt,
            )
        except urllib.error.HTTPError as exc:
            raw_body = exc.read()
            payload = _decode_payload(raw_body, decode_text=bool(endpoint.get("decode_text")))
            return _classify_response(
                endpoint=endpoint,
                path=path,
                status_code=int(exc.code),
                expected_statuses=expected_statuses,
                body_size=len(raw_body),
                payload=payload,
                attempt=attempt,
            )
        except Exception as exc:
            last_error = exc
            if attempt < attempts:
                sleeper(sleep_seconds)

    return {
        "name": name,
        "method": method,
        "path": path,
        "status": "blocked",
        "reason": "endpoint_unreachable",
        "attempts": attempts,
        "error": str(last_error),
    }


def _classify_response(
    *,
    endpoint: dict,
    path: str,
    status_code: int,
    expected_statuses: tuple,
    body_size: int,
    payload,
    attempt: int,
) -> dict:
    result = {
        "name": endpoint["name"],
        "method": endpoint["method"],
        "path": path,
        "status_code": status_code,
        "body_size": body_size,
        "attempt": attempt,
    }
    if status_code not in expected_statuses:
        result.update(
            {
                "status": "blocked",
                "reason": "unexpected_http_status",
                "expected_statuses": list(expected_statuses),
                "payload_excerpt": _payload_excerpt(payload),
            }
        )
        return result
    payload_check = str(endpoint.get("payload_check") or "")
    if payload_check:
        check_result = _run_payload_check(payload_check=payload_check, payload=payload)
        result.update(check_result)
        result["payload_excerpt"] = _payload_excerpt(payload)
        return result
    result["status"] = "passed"
    result["payload_excerpt"] = _payload_excerpt(payload)
    return result


def _run_payload_check(*, payload_check: str, payload) -> dict:
    if payload_check == "readonly_launch_ready":
        return _check_readonly_launch_ready(dict(payload or {}))
    if payload_check == "review_packet_available":
        payload_dict = dict(payload or {})
        decision_packet = dict(payload_dict.get("decision_packet") or {})
        if (
            payload_dict.get("status")
            or payload_dict.get("action_recommendation")
            or decision_packet.get("status")
            or decision_packet.get("action_recommendation")
        ):
            return {"status": "passed"}
        return {"status": "blocked", "reason": "review_packet_missing_status"}
    if payload_check == "shadow_paper_surface_available":
        payload_dict = dict(payload or {})
        if payload_dict.get("status") and payload_dict.get("execution_boundary") == "real_live_still_blocked":
            return {"status": "passed"}
        return {"status": "blocked", "reason": "shadow_paper_surface_unavailable"}
    if payload_check == "trader_mode_payload_available":
        payload_dict = dict(payload or {})
        trader = dict((payload_dict.get("decision_assistant_summary") or {}).get("trader_mode_summary") or {})
        missing = [
            key
            for key in [
                "status_label",
                "primary_conclusion",
                "focus_quality_explanation",
                "plain_language_reasons",
                "recovery_actions",
                "keyboard_shortcuts",
                "profile_template",
                "multi_instrument_queue_template",
                "chart_context_bridge",
                "post_market_review_draft",
                "ops_duty_panel",
                "freshness_summary",
                "r2_evidence_window",
                "primary_question_trade_action",
            ]
            if key not in trader
        ]
        if missing:
            return {"status": "blocked", "reason": "trader_mode_payload_missing", "missing_fields": missing}
        if trader.get("primary_question_trade_action") != "not_a_buy_sell_instruction":
            return {"status": "blocked", "reason": "trader_mode_boundary_missing"}
        if trader.get("execution_allowed") or trader.get("real_live_ready"):
            return {"status": "blocked", "reason": "trader_mode_execution_boundary_not_readonly"}
        if trader.get("status_label") not in {"Review Now", "Observe Only", "Do Not Use"}:
            return {"status": "blocked", "reason": "trader_mode_status_label_not_safe"}
        return {"status": "passed"}
    if payload_check == "candidates_surface_available":
        if isinstance(payload, dict) and isinstance(payload.get("items"), list):
            return {"status": "passed"}
        return {"status": "blocked", "reason": "candidate_surface_unavailable"}
    if payload_check == "weekly_review_entrypoint_available":
        payload_dict = dict(payload or {})
        if "entry_count" in payload_dict and isinstance(payload_dict.get("next_week_actions"), list):
            return {"status": "passed"}
        return {"status": "blocked", "reason": "weekly_review_entrypoint_unavailable"}
    if payload_check == "ops_duty_maintenance_available":
        payload_dict = dict(payload or {})
        if payload_dict.get("status") and isinstance(payload_dict.get("blockers", []), list):
            return {"status": "passed"}
        return {"status": "blocked", "reason": "ops_duty_maintenance_unavailable"}
    if payload_check == "ops_duty_harness_available":
        payload_dict = dict(payload or {})
        if payload_dict.get("status"):
            return {"status": "passed"}
        return {"status": "blocked", "reason": "ops_duty_harness_unavailable"}
    if payload_check == "chart_bridge_candles_available":
        payload_dict = dict(payload or {})
        if isinstance(payload_dict.get("items"), list):
            return {"status": "passed"}
        return {"status": "blocked", "reason": "chart_bridge_candles_unavailable"}
    if payload_check == "readonly_boundary_copy_present":
        text = str(payload or "")
        missing = [
            item
            for item in [
                "Read-only Live Analysis",
                "No order submission",
                "Not a buy/sell instruction",
                "Execution remains disabled",
                "ready_for_readonly_live_analysis",
                "must_not_trade",
            ]
            if item not in text
        ]
        if not missing:
            return {"status": "passed"}
        return {"status": "blocked", "reason": "readonly_boundary_copy_missing", "missing_copy": missing}
    if payload_check == "trader_mode_page_present":
        text = str(payload or "")
        missing = [
            item
            for item in [
                "Trader Mode",
                "Now",
                "Wait",
                "ICT Fast Read",
                "Market Last Updated",
                "Review Packet Updated",
                "Fast Queue",
                "Chart Bridge",
                "Duty Panel",
                "Profile",
                "Sessions",
                "Setups",
                "Recovery",
                "No Trade",
                "Need More Evidence",
                "After-Close Draft",
                "R2 Evidence Window",
            ]
            if item not in text
        ]
        if not missing:
            return {"status": "passed"}
        return {"status": "blocked", "reason": "trader_mode_page_missing_copy", "missing_copy": missing}
    if payload_check == "trader_mode_js_present":
        text = str(payload or "")
        missing = [
            item
            for item in [
                "/trading/review-packet",
                "/trading/shadow-paper/observe",
                "/trading/operator-notes",
                "Do Not Use",
                "Review Now",
                "Observe Only",
                "TRADER_SIGNATURE_KEY",
                "TRADER_PROFILE_STORE_KEY",
                "ETH London",
                "SOL",
                "DEFAULT_QUEUE_INSTRUMENTS",
                "/trading/research/weekly-review",
                "/trading/maintenance-summary",
                "/trading/unattended-harness",
                "/candles",
                "/trading/backtests",
                "Historical Edge",
                "buildHistoricalEdgeSummary",
                "post_market_review_draft",
                "chart_context_bridge",
                "ops_duty_panel",
                "aria-keyshortcuts",
                "pause_or_resume_auto_refresh",
            ]
            if item not in text
        ]
        if not missing:
            return {"status": "passed"}
        return {"status": "blocked", "reason": "trader_mode_js_missing_copy", "missing_copy": missing}
    return {"status": "passed"}


def _check_readonly_launch_ready(payload: dict) -> dict:
    summary = dict(payload.get("live_analysis_assistant_summary") or {})
    ready = bool(payload.get("live_analysis_assistant_ready") or summary.get("ready_for_readonly_live_analysis"))
    must_not_trade = bool(summary.get("must_not_trade", True))
    execution_assist_ready = bool(payload.get("execution_assist_ready"))
    real_live_ready = bool(payload.get("real_live_ready"))
    if not ready:
        return {
            "status": "blocked",
            "reason": "readonly_live_analysis_not_ready",
            "blocking_reasons": list(payload.get("blocking_reasons") or []),
        }
    if not must_not_trade:
        return {"status": "blocked", "reason": "must_not_trade_boundary_missing"}
    if execution_assist_ready or real_live_ready:
        return {
            "status": "blocked",
            "reason": "execution_boundary_not_readonly",
            "execution_assist_ready": execution_assist_ready,
            "real_live_ready": real_live_ready,
        }
    return {"status": "passed"}


def _decode_payload(raw_body: bytes, *, decode_text: bool):
    if not raw_body:
        return "" if decode_text else None
    if decode_text:
        return raw_body.decode("utf-8", errors="replace")
    try:
        return json.loads(raw_body.decode("utf-8"))
    except Exception:
        return raw_body[:512].decode("utf-8", errors="replace")


def _payload_excerpt(payload):
    if isinstance(payload, dict):
        keys = [
            "status",
            "action_recommendation",
            "scheduler_state",
            "current_blockers",
            "live_analysis_assistant_ready",
            "execution_assist_ready",
            "real_live_ready",
            "product_operating_mode",
            "readonly_account_context_status",
            "blocking_reasons",
            "warning_reasons",
            "live_analysis_assistant_summary",
            "readonly_live_health_summary",
            "entry_count",
            "next_week_actions",
        ]
        excerpt = {key: payload.get(key) for key in keys if key in payload}
        if "decision_packet" in payload and isinstance(payload.get("decision_packet"), dict):
            packet = dict(payload.get("decision_packet") or {})
            excerpt["decision_packet"] = {
                "status": packet.get("status"),
                "action_recommendation": packet.get("action_recommendation"),
                "candidate_id": packet.get("candidate_id"),
            }
        if "items" in payload and isinstance(payload.get("items"), list):
            excerpt["items_count"] = len(payload.get("items") or [])
        return excerpt or {key: payload.get(key) for key in sorted(payload.keys())[:8]}
    if isinstance(payload, str):
        return payload[:512]
    return payload


def _urlopen_no_proxy(request: urllib.request.Request, timeout_seconds: float):
    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
    return opener.open(request, timeout=timeout_seconds)


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"}


def _parse_args(argv: Optional[list[str]], env) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run read-only live analysis launch smoke against ai-exchange API.")
    parser.add_argument("--base-url", default=env.get("READONLY_LIVE_ANALYSIS_SMOKE_BASE_URL", DEFAULT_BASE_URL))
    parser.add_argument("--instrument-id", type=int, default=int(env.get("READONLY_LIVE_ANALYSIS_SMOKE_INSTRUMENT_ID", "1")))
    parser.add_argument("--timeframe", default=env.get("READONLY_LIVE_ANALYSIS_SMOKE_TIMEFRAME", "1m"))
    parser.add_argument("--attempts", type=int, default=int(env.get("READONLY_LIVE_ANALYSIS_SMOKE_ATTEMPTS", str(DEFAULT_ATTEMPTS))))
    parser.add_argument("--sleep-seconds", type=float, default=float(env.get("READONLY_LIVE_ANALYSIS_SMOKE_SLEEP_SECONDS", str(DEFAULT_SLEEP_SECONDS))))
    parser.add_argument("--timeout-seconds", type=float, default=float(env.get("READONLY_LIVE_ANALYSIS_SMOKE_TIMEOUT_SECONDS", str(DEFAULT_TIMEOUT_SECONDS))))
    parser.add_argument("--output-json", action="store_true", default=_get_bool(env, "READONLY_LIVE_ANALYSIS_SMOKE_OUTPUT_JSON", False))
    parser.add_argument("--artifact-path", default=env.get("READONLY_LIVE_ANALYSIS_SMOKE_ARTIFACT_PATH", ""))
    return parser.parse_args(argv)


def main(argv: Optional[list[str]] = None, env=None) -> int:
    env = os.environ if env is None else env
    args = _parse_args(argv, env)
    payload = run_readonly_live_analysis_smoke(
        base_url=args.base_url,
        instrument_id=args.instrument_id,
        timeframe=args.timeframe,
        attempts=args.attempts,
        sleep_seconds=args.sleep_seconds,
        timeout_seconds=args.timeout_seconds,
    )
    if args.artifact_path:
        Path(args.artifact_path).write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    if args.output_json:
        print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
    else:
        print(f"Read-only live analysis smoke status: {payload['status']}")
        print(f"Base URL: {payload['base_url']}")
        print(f"ready_for_readonly_live_analysis: {payload['ready_for_readonly_live_analysis']}")
        print(f"must_not_trade: {payload['must_not_trade']}")
        print(f"Blocked reason: {payload['blocked_reason'] or 'none'}")
        print(f"Warnings: {', '.join(payload['warnings']) or 'none'}")
        print(f"Passed checks: {', '.join(payload['passed_checks']) or 'none'}")
    return 0 if payload["status"] == "passed" else 2


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