""" validate_harness_context.py 목적: 1. JSON data._harness_context 또는 하네스 단독 JSON의 최소 계약 충족 여부를 검증한다. 2. STRICT_HARNESS에서 필수 lock/key 누락, JSON 직렬화 오류, 가격/수량 타입 오류를 조기 차단한다. """ from __future__ import annotations import json import re import sys from pathlib import Path from typing import Any _FORMULA_ID_HARNESS_CONTEXT_VALIDATOR_V2 = "HARNESS_CONTEXT_VALIDATOR_V2" _FORMULA_ID_MACRO_REGIME_ALIGNMENT_GATE_V2 = "MACRO_REGIME_ALIGNMENT_GATE_V2" def _crc32_v1(s: str) -> int: """I3: CRC32_V1 체크섬 — GAS computeStringChecksum_ 와 동일 알고리즘.""" total = 0 for ch in s: total = (total + ord(ch)) & 0xFFFFFFFF return total REQUIRED_SCALARS = [ "harness_version", "captured_at", "request_route", "route_reason_code", "bundle_selected", "prompt_entrypoint", "json_validation_status", "capture_required", "cash_ledger_basis", "intraday_lock", "snapshot_execution_gate", "immediate_cash_krw", "settlement_cash_d2_krw", "open_order_amount_krw", "buy_power_krw", "cash_floor_status", "snapshot_execution_reason", "total_heat_pct", "heat_gate_status", "heat_gate_threshold_pct", "sell_priority_lock", "quantities_lock", "prices_lock", "decision_lock", "backdata_learning_lock", "blueprint_row_count", "blueprint_checksum", "blueprint_hash_algo", # E1: M4 — GOAL_RETIREMENT_V1 목표 자산 추적 "goal_asset_krw", "goal_current_asset_krw", "goal_achievement_pct", "goal_remaining_krw", "goal_eta_label", "goal_status", # G1: CASH_SHORTFALL_V1 — GAS 결정론적 현금 부족액 계산 "cash_current_pct_d2", "cash_target_pct", "cash_shortfall_min_krw", "cash_shortfall_target_krw", # I3: CHECKSUM_V2 — source_manifest + decision_trace 무결성 "source_manifest_checksum", "decision_trace_checksum", "checksum_hash_algo", # M1: DRAWDOWN_GUARD_V1 "drawdown_guard_state", "drawdown_buy_scale", # M2: PORTFOLIO_BETA_GATE_V1 "portfolio_beta_gate", # M5: SECTOR_CONCENTRATION_LIMIT_V1 "sector_concentration_gate", # N1: POSITION_SIZE_REGIME_SCALE_V1 "regime_size_scale", # N5: REGIME_CASH_UPLIFT_V1 "regime_cash_uplift_min_pct", # O1: SINGLE_POSITION_WEIGHT_CAP_V1 "single_position_weight_gate", # O2: SEMICONDUCTOR_CLUSTER_GATE_V1 "semiconductor_cluster_gate", # O3: PORTFOLIO_DRAWDOWN_GATE_V1 "portfolio_drawdown_gate", # O4: WIN_LOSS_STREAK_GUARD_V1 "win_loss_streak_state", "win_loss_streak_buy_scale", # O5: POSITION_COUNT_LIMIT_V1 "position_count_gate", "position_count", # P1: STOP_BREACH_ALERT_V1 "stop_breach_gate", # P2: TP_TRIGGER_ALERT_V1 "tp_trigger_gate", # P3: HEAT_CONCENTRATION_ALERT_V1 "heat_concentration_gate", # P4: REGIME_TRANSITION_ALERT_V1 "regime_transition_type", # P5: PORTFOLIO_HEALTH_SCORE_V1 "portfolio_health_label", "portfolio_health_score", # [2026-05-20_HARNESS_V5] V5 결정론적 체크섬 + 포트폴리오 라우터 스칼라 "input_snapshot_checksum", "rendered_output_checksum", "non_deterministic_flag", "smart_cash_raise_route", # [2026-05-20_HARNESS_V5] V5 게이트 잠금 스칼라 "breakout_quality_gate_lock", "anti_whipsaw_gate_lock", "follow_through_lock", ] REQUIRED_NONEMPTY_STRINGS = [ "request_route", "bundle_selected", "prompt_entrypoint", "json_validation_status", "cash_ledger_basis", ] GOAL_ASSET_KRW_CONST = 500_000_000 VALID_GOAL_STATUSES = {"ACHIEVED", "IN_PROGRESS"} _ETA_LABEL_PATTERN = re.compile(r"^\d{4}-\d{2}$") REQUIRED_COLLECTIONS = [ "source_manifest_json", "allowed_actions", "blocked_actions", "account_snapshot_freshness_json", "sell_candidates_json", "sell_quantities_json", "buy_qty_inputs_json", "prices_json", "decisions_json", "decision_trace_json", "order_blueprint_json", "p4_intraday_allowed_actions", "regime_trim_guidance_json", # H7: M1 결정론적 감축비율 "secular_leader_gate_json", # H7: H3 주도주 게이트 결과 "backdata_feature_bank_json", "trim_plan_to_min_cash_json", # G2: TRIM_PLAN_MIN_CASH_V1 "regime_adjusted_sell_priority_json", # K3: 국면·섹터 연계 H2 동적 우선순위 "sector_rotation_momentum_json", # L1: 섹터 로테이션 모멘텀 추적 "portfolio_beta_gate_json", # M2: 포트폴리오 베타 게이트 "tp_quantity_ladder_json", # M3: 분할 익절 수량 "event_risk_json", # M4: 이벤트 리스크 홀드 "sector_concentration_json", # M5: 섹터 편중도 한도 "stop_adequacy_json", # N3: 손절가 적정성 검증 "holding_stale_json", # N4: 장기 보유 재검토 플래그 "single_position_weight_json", # O1: 개별 종목 비중 상한 "semiconductor_cluster_json", # O2: 반도체 클러스터 게이트 "stop_breach_alert_json", # P1: 손절가 이탈 경보 "portfolio_health_blocked_json", # P5: 건전성 게이트 상세 # [2026-05-20_HARNESS_V5] V5 게이트 JSON 컬렉션 "breakout_quality_gate_json", # H6: BREAKOUT_QUALITY_GATE_V2 "anti_whipsaw_gate_json", # H7: ANTI_WHIPSAW_HOLD_GATE_V1 "smart_cash_raise_json", # H8: SMART_CASH_RAISE_V2 "follow_through_json", # Gate 4b: FOLLOW_THROUGH_CONFIRM_V1 "benchmark_relative_timeseries_json", # BRT1: BENCHMARK_RELATIVE_TIMESERIES_V1 "index_relative_health_json", # BRT1b: INDEX_RELATIVE_HEALTH_GATE_V1 "saqg_json", # BRT2: SATELLITE_ALPHA_QUALITY_GATE_V1 "cash_creation_purpose_lock_json", # BRT3: CASH_CREATION_PURPOSE_LOCK_V1 "sapg_json", # BRT4: SATELLITE_AGGREGATE_PNL_GATE_V1 "alpha_feedback_json", # AFL1: ALPHA_FEEDBACK_LOOP_V1 ] VALID_SEMICONDUCTOR_CLUSTER_GATES = { # 기존 상태 (SEMICONDUCTOR_CLUSTER_GATE_V1) "PASS", "CLUSTER_BLOCK", "CLUSTER_HOLD_ONLY", "CLUSTER_OPEN", # 신규 상태 (MARKET_WEIGHT_AWARE_CLUSTER_GATE_V1, 2026-05-30) "CLUSTER_OVERWEIGHT_WARN", "CLUSTER_OVERWEIGHT_TRIM", # GAS 구버전 상태 — 하위호환 "CLUSTER_WARN", } VALID_SINGLE_POSITION_WEIGHT_GATES = { "PASS", "OVERWEIGHT_TRIM", } VALID_PROFIT_LOCK_STAGES = { # spec 기준 명칭 (B06 정정 2026-05-30) "NORMAL", "BREAKEVEN_RATCHET", "PROFIT_LOCK_10", "PROFIT_LOCK_20", "PROFIT_LOCK_30", "APEX_TRAILING", "APEX_SUPER", # 구버전 명칭 — 하위호환 (이전 데이터 재실행 대응) "PROFIT_LOCK_STAGE_10", "PROFIT_LOCK_STAGE_20", "PROFIT_LOCK_STAGE_30", "PROFIT_LOCK_STAGE_50", } VALID_TP1_STATES = { "PENDING", "TP1_ALREADY_TRIGGERED", "DEFERRED_SECULAR_LEADER", "DEFERRED_SECULAR_LEADER_OVERHEAT_PENDING", "TRAILING_STOP_PRIORITY_SECULAR_LEADER", "NO_POSITION", "INACTIVE", "UNKNOWN_NO_CLOSE", } APEX_LOCK_TO_COLLECTIONS = { "alpha_lead_lock": ("alpha_lead_json", "follow_through_json"), "follow_through_lock": ("follow_through_json",), "distribution_lock": ("distribution_risk_json",), "profit_preservation_lock": ("profit_preservation_json",), "smart_cash_raise_lock": ( "cash_raise_plan_json", "rebound_sell_trigger_json", "smart_sell_quantities_json", ), "execution_quality_lock": ("execution_quality_json", "limit_price_policy_json"), "backdata_learning_lock": ("backdata_feature_bank_json",), # [2026-05-20_HARNESS_V5] V5 게이트 잠금 "breakout_quality_gate_lock": ("breakout_quality_gate_json",), "anti_whipsaw_gate_lock": ("anti_whipsaw_gate_json",), "follow_through_lock": ("follow_through_json",), } VALID_BUY_PERMISSION_STATES = {"ALLOW_PILOT", "ALLOW_ADD_ON", "WATCH", "BLOCKED"} VALID_DISTRIBUTION_STATES = {"PASS", "TRIM_REVIEW", "BLOCK_BUY", "DATA_MISSING"} VALID_EXECUTION_STYLES = { "URGENT_LIQUIDITY_TRIM", "OVERSOLD_REBOUND_SELL", "DISTRIBUTION_EXIT", "PROFIT_PROTECT_TRIM", } VALID_BACKDATA_SOURCE_ORIGIN = { "GAS_AUTO", "PERFORMANCE_FALLBACK", "MANUAL_CORRECTION", } # [2026-05-20_HARNESS_V5] V5 게이트 열거형 상수 VALID_BREAKOUT_QUALITY_GATE_STATES = {"BLOCKED_LATE_CHASE", "WATCH_COOLING_OFF", "PILOT_ALLOWED"} VALID_ANTI_WHIPSAW_GATE_STATES = { "WHIPSAW_SUSPECTED", "INCONCLUSIVE", "CONFIRMED_SELL", "WHIPSAW_CONFIRMED", "WHIPSAW_WEAKENING", "WHIPSAW_AUTO_RELEASED", } VALID_SMART_CASH_RAISE_ROUTES = {"ROUTE_A", "ROUTE_B", "ROUTE_C", "ROUTE_D", "NO_ACTION"} VALID_FOLLOW_THROUGH_RESULTS = { "BUY_PILOT_ALLOWED", "WATCH_FOLLOW_THROUGH_PENDING", "WATCH_RESET_REQUIRED", "WATCH_TOO_LATE", "WATCH_NO_BREAKOUT_TRACKED", } VALID_PROPOSAL_EXECUTION_STATUSES = {"PROPOSAL_ONLY", "EXECUTION_WAIT", "EXECUTION_READY"} VALID_BRT_VERDICTS = {"LEADER", "MARKET", "LAGGARD", "BROKEN", "UNKNOWN"} VALID_INDEX_RELATIVE_HEALTH_STATES = {"HEALTHY", "OVER_EXTENDED", "UNDERPERFORMING", "DECOUPLED", "INSUFFICIENT_DATA"} VALID_SAQG_STATES = {"ELIGIBLE", "WATCHLIST_ONLY", "EXCLUDED", "EXEMPT"} VALID_SAPG_STATUSES = {"PASS", "SAPG_ALERT", "SAPG_CRITICAL", "INSUFFICIENT_DATA"} def parse_bool(value: Any) -> bool | None: if isinstance(value, bool): return value if isinstance(value, int): if value == 1: return True if value == 0: return False if isinstance(value, str): normalized = value.strip().lower() if normalized in {"true", "y", "yes", "1"}: return True if normalized in {"false", "n", "no", "0"}: return False return None def _has_any_key(payload: dict[str, Any], *keys: str) -> bool: return any(key in payload for key in keys) def _get_first(payload: dict[str, Any], *keys: str) -> Any: for key in keys: if key in payload: return payload.get(key) return None def load_harness(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if isinstance(payload, dict) and "data" in payload and isinstance(payload["data"], dict): maybe = payload["data"].get("_harness_context") if isinstance(maybe, dict): return maybe return payload def parse_jsonish(value: Any, key: str, errors: list[str]) -> Any: if isinstance(value, (list, dict)): return value if isinstance(value, str): text = value.strip() if not text: errors.append(f"{key}: empty string") return None try: return json.loads(text) except json.JSONDecodeError as exc: errors.append(f"{key}: invalid JSON ({exc.msg})") return None errors.append(f"{key}: unsupported type {type(value).__name__}") return None def is_int_or_none(value: Any) -> bool: return value is None or isinstance(value, int) def to_number(value: Any) -> float | None: if isinstance(value, (int, float)): return float(value) if isinstance(value, str): text = value.strip() if not text: return None try: return float(text) except ValueError: return None return None def validate_prices(prices: Any, errors: list[str]) -> None: if not isinstance(prices, list): errors.append("prices_json: must be a list") return for idx, row in enumerate(prices): if not isinstance(row, dict): errors.append(f"prices_json[{idx}]: must be an object") continue for field in ("stop_price", "tp1_price", "tp2_price"): if field in row and not is_int_or_none(row.get(field)): errors.append(f"prices_json[{idx}].{field}: must be integer or null") # H7: profit_lock_stage 열거형 검증 if "profit_lock_stage" in row: stage = row["profit_lock_stage"] if stage not in VALID_PROFIT_LOCK_STAGES: errors.append( f"prices_json[{idx}].profit_lock_stage: unknown value {stage!r}" ) # H7: tp1_state 열거형 검증 (알려진 값으로만 제한) if "tp1_state" in row: tp1_state = row["tp1_state"] if tp1_state is not None and tp1_state not in VALID_TP1_STATES: errors.append( f"prices_json[{idx}].tp1_state: unknown value {tp1_state!r}" ) # H7: secular_leader_gate_active boolean 타입 검증 if "secular_leader_gate_active" in row: gate_active = row["secular_leader_gate_active"] if not isinstance(gate_active, bool): errors.append( f"prices_json[{idx}].secular_leader_gate_active: must be boolean, got {type(gate_active).__name__}" ) def validate_decisions(decisions: Any, traces: Any, errors: list[str]) -> None: if not isinstance(decisions, list): errors.append("decisions_json: must be a list") return action_by_ticker: dict[str, set] = {} for idx, row in enumerate(decisions): if not isinstance(row, dict): errors.append(f"decisions_json[{idx}]: must be an object") continue ticker = row.get("ticker") action = row.get("final_action") if not ticker: errors.append(f"decisions_json[{idx}].ticker: missing") if not action: errors.append(f"decisions_json[{idx}].final_action: missing") if ticker: action_by_ticker.setdefault(str(ticker), set()).add(action) if isinstance(traces, list): for idx, row in enumerate(traces): if not isinstance(row, dict): errors.append(f"decision_trace_json[{idx}]: must be an object") continue ticker = str(row.get("ticker") or "") selected_action = row.get("selected_action") if ticker and ticker in action_by_ticker and selected_action not in (None, *action_by_ticker[ticker]): errors.append( f"decision_trace_json[{idx}].selected_action: " f"{selected_action} not in decisions_json[{ticker}]={sorted(action_by_ticker[ticker])}" ) def compute_blueprint_checksum(blueprint: list[dict[str, Any]]) -> int: s = "" for row in blueprint: s += ( f"{row.get('ticker', '')}|" f"{row.get('order_type', '')}|" f"{row.get('quantity', '') if row.get('quantity') is not None else ''}|" f"{row.get('limit_price_krw', '') if row.get('limit_price_krw') is not None else ''}|" f"{row.get('validation_status', '')};" ) total = 0 for ch in s: total = (total + ord(ch)) & 0xFFFFFFFF return total def validate_blueprint(blueprint: Any, harness: dict[str, Any], errors: list[str]) -> None: if not isinstance(blueprint, list): errors.append("order_blueprint_json: must be a list") return required_fields = { "account", "ticker", "name", "order_type", "mode", "limit_price_krw", "quantity", "stop_price_krw", "stop_quantity", "take_profit_price_krw", "take_profit_quantity", "validation_status", } for idx, row in enumerate(blueprint): if not isinstance(row, dict): errors.append(f"order_blueprint_json[{idx}]: must be an object") continue missing = sorted(required_fields - set(row)) if missing: errors.append(f"order_blueprint_json[{idx}] missing fields: {missing}") row_count = harness.get("blueprint_row_count") if isinstance(row_count, str) and row_count.isdigit(): row_count = int(row_count) if row_count != len(blueprint): errors.append(f"blueprint_row_count mismatch: stored={row_count}, actual={len(blueprint)}") algo = harness.get("blueprint_hash_algo") if algo != "CRC32_V1": errors.append(f"blueprint_hash_algo must be CRC32_V1, found={algo!r}") checksum = harness.get("blueprint_checksum") if isinstance(checksum, str) and checksum.isdigit(): checksum = int(checksum) computed = compute_blueprint_checksum(blueprint) if checksum != computed: errors.append(f"blueprint_checksum mismatch: stored={checksum}, computed={computed}") def validate_proposal_reference(payload: Any, errors: list[str]) -> None: if not isinstance(payload, list): errors.append("proposal_reference_json: must be a list") return required_fields = { "account", "ticker", "name", "proposal_type", "proposed_limit_price_krw", "proposed_price_basis", "proposed_quantity", "proposed_quantity_basis", "stop1_price_krw", "stop1_quantity", "stop2_price_krw", "stop2_quantity", "stop3_price_krw", "stop3_quantity", "tp1_price_krw", "tp1_quantity", "tp2_price_krw", "tp2_quantity", "tp3_price_krw", "tp3_quantity", "execution_status", "block_reason", } for idx, row in enumerate(payload): if not isinstance(row, dict): errors.append(f"proposal_reference_json[{idx}]: must be an object") continue missing = sorted(required_fields - set(row)) if missing: errors.append(f"proposal_reference_json[{idx}] missing fields: {missing}") for field in ( "proposed_limit_price_krw", "proposed_quantity", "stop1_price_krw", "stop1_quantity", "stop2_price_krw", "stop2_quantity", "stop3_price_krw", "stop3_quantity", "tp1_price_krw", "tp1_quantity", "tp2_price_krw", "tp2_quantity", "tp3_price_krw", "tp3_quantity", ): value = row.get(field) if value is not None and not isinstance(value, int): errors.append(f"proposal_reference_json[{idx}].{field}: must be integer or null") status = row.get("execution_status") if status is not None and status not in VALID_PROPOSAL_EXECUTION_STATUSES: errors.append( f"proposal_reference_json[{idx}].execution_status invalid: {status!r}" ) def validate_quantities(payload: Any, key: str, errors: list[str]) -> None: if not isinstance(payload, list): errors.append(f"{key}: must be a list") return for idx, row in enumerate(payload): if not isinstance(row, dict): errors.append(f"{key}[{idx}]: must be an object") continue for field, value in row.items(): if field.endswith("_qty") or field == "final_qty": if not is_int_or_none(value) and value != "CAPTURE_REQUIRED" and value != "NO_BUY_QUANTITY": errors.append(f"{key}[{idx}].{field}: invalid quantity value {value!r}") def validate_goal_tracking(harness: dict[str, Any], errors: list[str]) -> None: """E1: GOAL_RETIREMENT_V1 출력 필드 타입·범위 검증.""" goal_krw = harness.get("goal_asset_krw") if goal_krw is not None: g = to_number(goal_krw) if g != GOAL_ASSET_KRW_CONST: errors.append( f"goal_asset_krw must be {GOAL_ASSET_KRW_CONST}, found={goal_krw!r}" ) achieve = to_number(harness.get("goal_achievement_pct")) if achieve is not None and achieve < 0: errors.append(f"goal_achievement_pct must be >= 0, found={achieve}") remain = to_number(harness.get("goal_remaining_krw")) if remain is not None and remain < 0: errors.append(f"goal_remaining_krw must be >= 0, found={remain}") status = harness.get("goal_status") if status is not None and status not in VALID_GOAL_STATUSES: errors.append( f"goal_status must be one of {sorted(VALID_GOAL_STATUSES)}, found={status!r}" ) eta_label = harness.get("goal_eta_label") if eta_label is not None and eta_label not in ("ACHIEVED", "DATA_MISSING"): if not _ETA_LABEL_PATTERN.match(str(eta_label)): errors.append( f"goal_eta_label must be ACHIEVED | DATA_MISSING | YYYY-MM, found={eta_label!r}" ) eta_months = harness.get("goal_eta_months") if eta_months is not None and not isinstance(eta_months, (int, float)): errors.append( f"goal_eta_months must be integer or null, found type={type(eta_months).__name__}" ) def validate_cash_shortfall(harness: dict[str, Any], errors: list[str]) -> None: """G1: CASH_SHORTFALL_V1 필드 타입·범위·부호 검증.""" for key in ("cash_current_pct_d2", "cash_target_pct"): val = to_number(harness.get(key)) if val is not None and (val < 0 or val > 100): errors.append(f"{key} must be in [0, 100], found={val}") for key in ("cash_shortfall_min_krw", "cash_shortfall_target_krw"): val = to_number(harness.get(key)) if val is not None and val < 0: errors.append(f"{key} must be >= 0, found={val}") # D+2 현금비중이 목표보다 높으면 shortfall은 0이어야 함 pct_d2 = to_number(harness.get("cash_current_pct_d2")) target_pct = to_number(harness.get("cash_target_pct")) shortfall_tgt = to_number(harness.get("cash_shortfall_target_krw")) if pct_d2 is not None and target_pct is not None and shortfall_tgt is not None: if pct_d2 >= target_pct and shortfall_tgt != 0: errors.append( f"cash_shortfall_target_krw must be 0 when cash_current_pct_d2({pct_d2}) >= cash_target_pct({target_pct}), found={shortfall_tgt}" ) def validate_trim_plan(harness: dict[str, Any], errors: list[str]) -> None: """G2: TRIM_PLAN_MIN_CASH_V1 배열 구조 검증.""" raw = harness.get("trim_plan_to_min_cash_json") if raw is None: return plan = parse_jsonish(raw, "trim_plan_to_min_cash_json", errors) if not isinstance(plan, list): return required_fields = ("rank", "ticker", "tier", "estimated_sell_krw", "accumulated_krw", "covers_shortfall") for idx, row in enumerate(plan): if not isinstance(row, dict): errors.append(f"trim_plan_to_min_cash_json[{idx}]: must be an object") continue for field in required_fields: if field not in row: errors.append(f"trim_plan_to_min_cash_json[{idx}]: missing field {field!r}") est = row.get("estimated_sell_krw") accum = row.get("accumulated_krw") if isinstance(est, (int, float)) and est < 0: errors.append(f"trim_plan_to_min_cash_json[{idx}].estimated_sell_krw must be >= 0") if isinstance(accum, (int, float)) and accum < 0: errors.append(f"trim_plan_to_min_cash_json[{idx}].accumulated_krw must be >= 0") def validate_external_context(harness: dict[str, Any], errors: list[str]) -> None: """I5: external_context_json 외부 데이터 격리 구조 검증 (존재할 경우만).""" raw = harness.get("external_context_json") if raw is None: return items = parse_jsonish(raw, "external_context_json", errors) if not isinstance(items, list): return required = ("source_name", "fetched_at", "symbol", "value", "used_for") valid_used_for = {"CONTEXT_ONLY", "VALIDATION_ONLY"} for idx, item in enumerate(items): if not isinstance(item, dict): errors.append(f"external_context_json[{idx}]: must be an object") continue for field in required: if field not in item: errors.append(f"external_context_json[{idx}]: missing required field {field!r}") used_for = item.get("used_for") if used_for is not None and used_for not in valid_used_for: errors.append( f"external_context_json[{idx}].used_for must be CONTEXT_ONLY|VALIDATION_ONLY, found={used_for!r}" ) def validate_snapshot_gate(harness: dict[str, Any], errors: list[str]) -> None: """장중 스냅샷 신선도에 따라 실행 게이트가 잠기는지 검증한다.""" gate = str(harness.get("snapshot_execution_gate") or "").strip() if gate not in {"ALLOW_EXECUTION", "REVIEW_ONLY", "BLOCK_EXECUTION"}: errors.append(f"snapshot_execution_gate invalid: {gate!r}") return freshness = parse_jsonish(harness.get("account_snapshot_freshness_json"), "account_snapshot_freshness_json", errors) if isinstance(freshness, dict): fresh = freshness.get("fresh") if fresh is False and gate not in {"REVIEW_ONLY", "ALLOW_EXECUTION"}: errors.append("account_snapshot_freshness_json.fresh=false requires REVIEW_ONLY or ALLOW_EXECUTION") if fresh is True and gate != "ALLOW_EXECUTION": errors.append("account_snapshot_freshness_json.fresh=true requires snapshot_execution_gate=ALLOW_EXECUTION") if gate == "BLOCK_EXECUTION": blocked = harness.get("blocked_actions") if isinstance(blocked, list): blocked_set = set(map(str, blocked)) if not {"BUY", "STAGED_BUY"}.issubset(blocked_set): errors.append("snapshot_execution_gate=BLOCK_EXECUTION requires BUY and STAGED_BUY in blocked_actions") def _parse_optional_list(harness: dict[str, Any], key: str, errors: list[str]) -> list[Any] | None: if key not in harness: return None parsed = parse_jsonish(harness.get(key), key, errors) if not isinstance(parsed, list): errors.append(f"{key}: must be a list when provided") return None return parsed def validate_apex_upgrade(harness: dict[str, Any], errors: list[str]) -> None: """APEX_V1 optional upgrade: lock이 켜진 경우에만 신규 하네스 payload를 엄격 검증.""" for lock_key, required_payloads in APEX_LOCK_TO_COLLECTIONS.items(): lock_value = parse_bool(harness.get(lock_key)) if lock_value is None and lock_key in harness: errors.append(f"{lock_key}: must be boolean or boolean-like string") if lock_value is True: for payload_key in required_payloads: if payload_key not in harness: errors.append(f"{lock_key}=true but missing {payload_key}") alpha_rows = _parse_optional_list(harness, "alpha_lead_json", errors) if alpha_rows is not None: for idx, row in enumerate(alpha_rows): if not isinstance(row, dict): errors.append(f"alpha_lead_json[{idx}]: must be an object") continue score = to_number(row.get("alpha_lead_score")) if score is not None and not (0 <= score <= 100): errors.append(f"alpha_lead_json[{idx}].alpha_lead_score must be in [0,100]") tranche = to_number(row.get("allowed_tranche_pct")) if tranche is not None and tranche > 30: errors.append(f"alpha_lead_json[{idx}].allowed_tranche_pct must be <= 30") distribution_rows = _parse_optional_list(harness, "distribution_risk_json", errors) if distribution_rows is not None: for idx, row in enumerate(distribution_rows): if not isinstance(row, dict): errors.append(f"distribution_risk_json[{idx}]: must be an object") continue score = to_number(row.get("distribution_risk_score")) if score is not None and not (0 <= score <= 100): errors.append(f"distribution_risk_json[{idx}].distribution_risk_score must be in [0,100]") state = row.get("anti_distribution_state") if state is not None and state not in VALID_DISTRIBUTION_STATES: errors.append( f"distribution_risk_json[{idx}].anti_distribution_state invalid: {state!r}" ) _VALID_TRANCHE_PHASES = { "WAIT_PILOT_SETUP", "TRANCHE_1_PILOT", "TRANCHE_2_ADD_ON", "TRANCHE_3_PULLBACK_ADD", "HOLD_CURRENT", } buy_permission_rows = _parse_optional_list(harness, "buy_permission_json", errors) if buy_permission_rows is not None: for idx, row in enumerate(buy_permission_rows): if not isinstance(row, dict): errors.append(f"buy_permission_json[{idx}]: must be an object") continue state = row.get("buy_permission_state") if state not in VALID_BUY_PERMISSION_STATES: errors.append(f"buy_permission_json[{idx}].buy_permission_state invalid: {state!r}") tranche = to_number(row.get("max_tranche_pct")) if state == "ALLOW_PILOT" and tranche is not None and tranche > 30: errors.append(f"buy_permission_json[{idx}].max_tranche_pct must be <= 30 for ALLOW_PILOT") # K1: 트랜치 페이즈 검증 t_phase = row.get("tranche_phase") if t_phase is not None and t_phase not in _VALID_TRANCHE_PHASES: errors.append(f"buy_permission_json[{idx}].tranche_phase invalid: {t_phase!r}") cur_t = to_number(row.get("current_tranche_allowed_pct")) if t_phase in ("WAIT_PILOT_SETUP", "HOLD_CURRENT") and cur_t is not None and cur_t > 0: errors.append( f"buy_permission_json[{idx}]: {t_phase} must have current_tranche_allowed_pct=0" ) cash_raise_rows = _parse_optional_list(harness, "cash_raise_plan_json", errors) if cash_raise_rows is not None: for idx, row in enumerate(cash_raise_rows): if not isinstance(row, dict): errors.append(f"cash_raise_plan_json[{idx}]: must be an object") continue style = row.get("execution_style") if style is not None and style not in VALID_EXECUTION_STYLES: errors.append(f"cash_raise_plan_json[{idx}].execution_style invalid: {style!r}") immediate = row.get("immediate_qty") rebound = row.get("rebound_wait_qty") for field, value in (("immediate_qty", immediate), ("rebound_wait_qty", rebound)): if value is not None and not isinstance(value, int): errors.append(f"cash_raise_plan_json[{idx}].{field}: must be integer or null") # K2: OVERSOLD_REBOUND_SELL은 emergency_full_sell=true 예외 허용 emergency = row.get("emergency_full_sell", False) if style == "OVERSOLD_REBOUND_SELL" and rebound in (None, 0) and not emergency: errors.append( f"cash_raise_plan_json[{idx}]: OVERSOLD_REBOUND_SELL requires rebound_wait_qty > 0 " f"unless emergency_full_sell=true" ) smart_sell_rows = _parse_optional_list(harness, "smart_sell_quantities_json", errors) if smart_sell_rows is not None: for idx, row in enumerate(smart_sell_rows): if not isinstance(row, dict): errors.append(f"smart_sell_quantities_json[{idx}]: must be an object") continue # staged_total_qty (K2 업데이트 후 필드명) for field in ("immediate_sell_qty", "staged_total_qty", "rebound_wait_qty"): value = row.get(field) if value is not None and not isinstance(value, int): errors.append(f"smart_sell_quantities_json[{idx}].{field}: must be integer or null") execution_rows = _parse_optional_list(harness, "execution_quality_json", errors) if execution_rows is not None: for idx, row in enumerate(execution_rows): if not isinstance(row, dict): errors.append(f"execution_quality_json[{idx}]: must be an object") continue hts_allowed = row.get("hts_allowed") if hts_allowed is not None and not isinstance(hts_allowed, bool): errors.append(f"execution_quality_json[{idx}].hts_allowed: must be boolean") backdata_rows = _parse_optional_list(harness, "backdata_feature_bank_json", errors) if backdata_rows is not None: for idx, row in enumerate(backdata_rows): if not isinstance(row, dict): errors.append(f"backdata_feature_bank_json[{idx}]: must be an object") continue origin = row.get("Source_Origin") if origin is not None and origin not in VALID_BACKDATA_SOURCE_ORIGIN: errors.append( f"backdata_feature_bank_json[{idx}].Source_Origin invalid: {origin!r}" ) if row.get("Entry_Stage") is not None and row.get("Entry_Stage") not in {"stage_1", "stage_2", "stage_3", "PERFORMANCE_FALLBACK"}: errors.append( f"backdata_feature_bank_json[{idx}].Entry_Stage invalid: {row.get('Entry_Stage')!r}" ) # K3: 국면·섹터 연계 H2 동적 우선순위 검증 regime_adj_rows = _parse_optional_list(harness, "regime_adjusted_sell_priority_json", errors) if regime_adj_rows is not None: seen_final_ranks: set[int] = set() for idx, row in enumerate(regime_adj_rows): if not isinstance(row, dict): errors.append(f"regime_adjusted_sell_priority_json[{idx}]: must be an object") continue adj = to_number(row.get("regime_priority_adjustment")) if adj is not None and not (-5 <= adj <= 0): errors.append( f"regime_adjusted_sell_priority_json[{idx}].regime_priority_adjustment " f"must be in [-5,0]: got {adj}" ) final_rank = row.get("final_regime_rank") if final_rank is not None: if not isinstance(final_rank, int) or final_rank < 1: errors.append( f"regime_adjusted_sell_priority_json[{idx}].final_regime_rank " f"must be positive integer" ) elif final_rank in seen_final_ranks: errors.append( f"regime_adjusted_sell_priority_json: duplicate final_regime_rank={final_rank}" ) else: seen_final_ranks.add(final_rank) def validate_v5_gates(harness: dict[str, Any], errors: list[str]) -> None: """[2026-05-20_HARNESS_V5] H6/H7/H8/Gate-4b 구조·열거형·불변식 검증.""" # H6: BREAKOUT_QUALITY_GATE_V2 bq_rows = _parse_optional_list(harness, "breakout_quality_gate_json", errors) if bq_rows is not None: for idx, row in enumerate(bq_rows): if not isinstance(row, dict): errors.append(f"breakout_quality_gate_json[{idx}]: must be an object") continue score = to_number(row.get("breakout_quality_score")) if score is not None and not (0 <= score <= 100): errors.append( f"breakout_quality_gate_json[{idx}].breakout_quality_score must be in [0,100], got {score}" ) gate = row.get("breakout_quality_gate") if gate is not None and gate not in VALID_BREAKOUT_QUALITY_GATE_STATES: errors.append( f"breakout_quality_gate_json[{idx}].breakout_quality_gate invalid: {gate!r}" ) # H7: ANTI_WHIPSAW_HOLD_GATE_V1 aw_rows = _parse_optional_list(harness, "anti_whipsaw_gate_json", errors) if aw_rows is not None: for idx, row in enumerate(aw_rows): if not isinstance(row, dict): errors.append(f"anti_whipsaw_gate_json[{idx}]: must be an object") continue score = to_number(row.get("anti_whipsaw_score")) if score is not None and not (-50 <= score <= 100): errors.append( f"anti_whipsaw_gate_json[{idx}].anti_whipsaw_score must be in [-50,100], got {score}" ) gate = row.get("anti_whipsaw_gate") if gate is not None and gate not in VALID_ANTI_WHIPSAW_GATE_STATES: errors.append( f"anti_whipsaw_gate_json[{idx}].anti_whipsaw_gate invalid: {gate!r}" ) hold_days = row.get("anti_whipsaw_hold_days") if hold_days is not None and not isinstance(hold_days, int): errors.append( f"anti_whipsaw_gate_json[{idx}].anti_whipsaw_hold_days must be integer" ) if gate == "WHIPSAW_SUSPECTED" and isinstance(hold_days, int) and hold_days < 1: errors.append( f"anti_whipsaw_gate_json[{idx}]: WHIPSAW_SUSPECTED requires anti_whipsaw_hold_days >= 1" ) # H8: SMART_CASH_RAISE_V2 scr_rows = _parse_optional_list(harness, "smart_cash_raise_json", errors) if scr_rows is not None: for idx, row in enumerate(scr_rows): if not isinstance(row, dict): errors.append(f"smart_cash_raise_json[{idx}]: must be an object") continue route = row.get("smart_cash_raise_route") if route is not None and route not in VALID_SMART_CASH_RAISE_ROUTES: errors.append( f"smart_cash_raise_json[{idx}].smart_cash_raise_route invalid: {route!r}" ) rebound_pct = to_number(row.get("rebound_wait_pct")) if route == "ROUTE_B" and rebound_pct is not None and rebound_pct != 50: errors.append( f"smart_cash_raise_json[{idx}]: ROUTE_B must have rebound_wait_pct=50, got {rebound_pct}" ) if route == "ROUTE_D": emergency = row.get("emergency_full_sell") is True stop_gate = row.get("stop_breach_gate") if not emergency and stop_gate != "BREACH": errors.append( f"smart_cash_raise_json[{idx}]: ROUTE_D requires emergency_full_sell=true or stop_breach_gate='BREACH'" ) # Gate 4b: FOLLOW_THROUGH_CONFIRM_V1 follow_through_key = "follow_through_json" if "follow_through_json" in harness else "follow_through_confirm_json" ftd_rows = _parse_optional_list(harness, follow_through_key, errors) if ftd_rows is not None: for idx, row in enumerate(ftd_rows): if not isinstance(row, dict): errors.append(f"{follow_through_key}[{idx}]: must be an object") continue result = row.get("follow_through_result") if result is not None and result not in VALID_FOLLOW_THROUGH_RESULTS: errors.append( f"{follow_through_key}[{idx}].follow_through_result invalid: {result!r}" ) days = row.get("days_since_breakout") if days is not None and not isinstance(days, (int, float)): errors.append( f"{follow_through_key}[{idx}].days_since_breakout must be numeric or null" ) # 불변식: WATCH_FOLLOW_THROUGH_PENDING은 day 0에서만 발생 if result == "WATCH_FOLLOW_THROUGH_PENDING" and isinstance(days, (int, float)) and int(days) != 0: errors.append( f"{follow_through_key}[{idx}]: WATCH_FOLLOW_THROUGH_PENDING requires days_since_breakout=0" ) # 불변식: WATCH_TOO_LATE는 day > 7에서만 발생 if result == "WATCH_TOO_LATE" and isinstance(days, (int, float)) and int(days) <= 7: errors.append( f"{follow_through_key}[{idx}]: WATCH_TOO_LATE requires days_since_breakout > 7" ) # 포트폴리오 레벨 스칼라 라우트 검증 port_route = harness.get("smart_cash_raise_route") if port_route is not None and port_route not in VALID_SMART_CASH_RAISE_ROUTES: errors.append(f"smart_cash_raise_route (portfolio-level) invalid: {port_route!r}") def validate_brt_harness(harness: dict[str, Any], errors: list[str]) -> None: """[2026-05-21_BRT_HARNESS_V1] BRT/SAQG/CCPL/SAPG 구조·열거형 검증.""" brt_rows = parse_jsonish(harness.get("benchmark_relative_timeseries_json"), "benchmark_relative_timeseries_json", errors) if not isinstance(brt_rows, list): errors.append("benchmark_relative_timeseries_json: must be a list") else: for idx, row in enumerate(brt_rows): if not isinstance(row, dict): errors.append(f"benchmark_relative_timeseries_json[{idx}]: must be an object") continue verdict = row.get("brt_verdict") if verdict not in VALID_BRT_VERDICTS: errors.append(f"benchmark_relative_timeseries_json[{idx}].brt_verdict invalid: {verdict!r}") if not row.get("formula_id"): errors.append(f"benchmark_relative_timeseries_json[{idx}].formula_id missing") index_rows = parse_jsonish(harness.get("index_relative_health_json"), "index_relative_health_json", errors) if not isinstance(index_rows, list): errors.append("index_relative_health_json: must be a list") else: for idx, row in enumerate(index_rows): if not isinstance(row, dict): errors.append(f"index_relative_health_json[{idx}]: must be an object") continue state = row.get("relative_health_state") if state not in VALID_INDEX_RELATIVE_HEALTH_STATES: errors.append(f"index_relative_health_json[{idx}].relative_health_state invalid: {state!r}") if not row.get("formula_id"): errors.append(f"index_relative_health_json[{idx}].formula_id missing") saqg_rows = parse_jsonish(harness.get("saqg_json"), "saqg_json", errors) if not isinstance(saqg_rows, list): errors.append("saqg_json: must be a list") else: for idx, row in enumerate(saqg_rows): if not isinstance(row, dict): errors.append(f"saqg_json[{idx}]: must be an object") continue state = row.get("saqg_v1") if state not in VALID_SAQG_STATES: errors.append(f"saqg_json[{idx}].saqg_v1 invalid: {state!r}") if state in {"EXCLUDED", "WATCHLIST_ONLY"} and row.get("hts_allowed") is True: errors.append(f"saqg_json[{idx}].hts_allowed must be false for {state}") cash_lock_rows = parse_jsonish(harness.get("cash_creation_purpose_lock_json"), "cash_creation_purpose_lock_json", errors) if not isinstance(cash_lock_rows, list): errors.append("cash_creation_purpose_lock_json: must be a list") else: for idx, row in enumerate(cash_lock_rows): if not isinstance(row, dict): errors.append(f"cash_creation_purpose_lock_json[{idx}]: must be an object") continue validity = row.get("sell_reason_validity") if validity not in {"VALID_SELL_REASON", "INVALID_SELL_REASON"}: errors.append(f"cash_creation_purpose_lock_json[{idx}].sell_reason_validity invalid: {validity!r}") sapg = parse_jsonish(harness.get("sapg_json"), "sapg_json", errors) if not isinstance(sapg, dict): errors.append("sapg_json: must be an object") else: status = sapg.get("sapg_status") if status not in VALID_SAPG_STATUSES: errors.append(f"sapg_json.sapg_status invalid: {status!r}") def _validate_v5_checksums(harness: dict[str, Any], errors: list[str]) -> None: """[2026-05-20_HARNESS_V5] V5 체크섬 불변식 검증. rendered_output_checksum == blueprint_checksum (legacy alias: rendered_report_checksum) non_deterministic_flag 는 GAS emit 시 항상 false; true면 데이터 소스 변경 경보. """ rc_stored = _get_first(harness, "rendered_output_checksum", "rendered_report_checksum") bp_stored = harness.get("blueprint_checksum") if rc_stored is not None and bp_stored is not None: def _to_int(v: Any) -> int | None: if isinstance(v, int): return v if isinstance(v, str) and v.isdigit(): return int(v) return None rc_num = _to_int(rc_stored) bp_num = _to_int(bp_stored) if rc_num is not None and bp_num is not None and rc_num != bp_num: errors.append( f"rendered_output_checksum must equal blueprint_checksum: " f"rendered={rc_num}, blueprint={bp_num}" ) nd_flag = parse_bool(harness.get("non_deterministic_flag")) if nd_flag is True: errors.append( "non_deterministic_flag=true: 동일 입력 재호출 시 체크섬 불일치 — " "GAS 재실행 또는 데이터 소스 변경 여부를 확인하세요." ) elif nd_flag is None and "non_deterministic_flag" in harness: errors.append("non_deterministic_flag: must be boolean or boolean-like string") def validate_context(harness: dict[str, Any]) -> list[str]: errors: list[str] = [] for key in REQUIRED_SCALARS: if key == "rendered_output_checksum": if not _has_any_key(harness, "rendered_output_checksum", "rendered_report_checksum"): errors.append("missing scalar key: rendered_output_checksum") continue if key not in harness: errors.append(f"missing scalar key: {key}") for key in REQUIRED_COLLECTIONS: if key == "follow_through_json": if not _has_any_key(harness, "follow_through_json", "follow_through_confirm_json"): errors.append("missing collection key: follow_through_json") continue if key not in harness: errors.append(f"missing collection key: {key}") for key in REQUIRED_NONEMPTY_STRINGS: value = harness.get(key) if not isinstance(value, str) or not value.strip(): errors.append(f"{key}: must be non-empty string") allowed = harness.get("allowed_actions") blocked = harness.get("blocked_actions") if not isinstance(allowed, list): errors.append("allowed_actions: must be a list") if not isinstance(blocked, list): errors.append("blocked_actions: must be a list") if isinstance(allowed, list) and isinstance(blocked, list): overlap = sorted(set(map(str, allowed)) & set(map(str, blocked))) if overlap: errors.append(f"allowed_actions and blocked_actions overlap: {overlap}") for lock_key in ("sell_priority_lock", "quantities_lock", "prices_lock", "decision_lock", "backdata_learning_lock"): if parse_bool(harness.get(lock_key)) is None: errors.append(f"{lock_key}: must be boolean or boolean-like string") if "proposal_reference_lock" in harness and parse_bool(harness.get("proposal_reference_lock")) is None: errors.append("proposal_reference_lock: must be boolean or boolean-like string") if harness.get("cash_ledger_basis") != "D2_ONLY": errors.append(f"cash_ledger_basis must be 'D2_ONLY', found={harness.get('cash_ledger_basis')!r}") sell_candidates = parse_jsonish(harness.get("sell_candidates_json"), "sell_candidates_json", errors) source_manifest = parse_jsonish(harness.get("source_manifest_json"), "source_manifest_json", errors) sell_quantities = parse_jsonish(harness.get("sell_quantities_json"), "sell_quantities_json", errors) buy_quantities = parse_jsonish(harness.get("buy_qty_inputs_json"), "buy_qty_inputs_json", errors) prices = parse_jsonish(harness.get("prices_json"), "prices_json", errors) decisions = parse_jsonish(harness.get("decisions_json"), "decisions_json", errors) traces = parse_jsonish(harness.get("decision_trace_json"), "decision_trace_json", errors) blueprint = parse_jsonish(harness.get("order_blueprint_json"), "order_blueprint_json", errors) proposal_reference = None if "proposal_reference_json" in harness: proposal_reference = parse_jsonish(harness.get("proposal_reference_json"), "proposal_reference_json", errors) p4_allowed = parse_jsonish(harness.get("p4_intraday_allowed_actions"), "p4_intraday_allowed_actions", errors) regime_trim = parse_jsonish(harness.get("regime_trim_guidance_json"), "regime_trim_guidance_json", errors) secular_gate = parse_jsonish(harness.get("secular_leader_gate_json"), "secular_leader_gate_json", errors) if parse_bool(harness.get("sell_priority_lock")) is True and not isinstance(sell_candidates, list): errors.append("sell_priority_lock=true but sell_candidates_json is not a list") if source_manifest is not None and not isinstance(source_manifest, list): errors.append("source_manifest_json: must be a list") if parse_bool(harness.get("quantities_lock")) is True: validate_quantities(sell_quantities, "sell_quantities_json", errors) validate_quantities(buy_quantities, "buy_qty_inputs_json", errors) if parse_bool(harness.get("prices_lock")) is True: validate_prices(prices, errors) if parse_bool(harness.get("decision_lock")) is True: validate_decisions(decisions, traces, errors) validate_blueprint(blueprint, harness, errors) if parse_bool(harness.get("proposal_reference_lock")) is True and proposal_reference is not None: validate_proposal_reference(proposal_reference, errors) # H7: regime_trim_guidance_json 구조 검증 (M1 결정론적 감축비율) if regime_trim is not None: for field in ("phase", "new_buy_gate"): if not isinstance(regime_trim, dict) or field not in regime_trim: errors.append(f"regime_trim_guidance_json: missing field {field!r}") break # H7: secular_leader_gate_json 구조 검증 (H3 주도주 게이트) if secular_gate is not None and isinstance(secular_gate, dict): for ticker, gate_info in secular_gate.items(): if not isinstance(gate_info, dict): errors.append(f"secular_leader_gate_json[{ticker}]: must be an object") continue if "active" not in gate_info: errors.append(f"secular_leader_gate_json[{ticker}]: missing 'active' field") if "status" not in gate_info: errors.append(f"secular_leader_gate_json[{ticker}]: missing 'status' field") settlement_cash = to_number(harness.get("settlement_cash_d2_krw")) open_order_amount = to_number(harness.get("open_order_amount_krw")) buy_power = to_number(harness.get("buy_power_krw")) if settlement_cash is not None and buy_power is not None: expected_buy_power = settlement_cash - (open_order_amount or 0.0) if abs(expected_buy_power - buy_power) > 0.5: errors.append( f"buy_power_krw mismatch: stored={buy_power}, expected={expected_buy_power} " "(must equal settlement_cash_d2_krw - open_order_amount_krw)" ) intraday_lock = parse_bool(harness.get("intraday_lock")) if intraday_lock is True: if not isinstance(p4_allowed, list): errors.append("intraday_lock=true but p4_intraday_allowed_actions is not a list") elif isinstance(decisions, list): allowed_set = {str(v) for v in p4_allowed} for idx, row in enumerate(decisions): if not isinstance(row, dict): continue action = str(row.get("final_action") or "") if action and action not in allowed_set: errors.append(f"decisions_json[{idx}].final_action not allowed under intraday_lock: {action}") # Hard-lock: decision_lock=true 이면 decisions_json / decision_trace_json 최소 1행 이상 필수 if parse_bool(harness.get("decision_lock")) is True: if not isinstance(decisions, list) or len(decisions) == 0: errors.append("decision_lock=true but decisions_json is empty") if not isinstance(traces, list) or len(traces) == 0: errors.append("decision_lock=true but decision_trace_json is empty") # Hard-lock: prices_lock/quantities_lock=true 이면 prices/sell_qty/buy_qty 비어있으면 실패 if parse_bool(harness.get("prices_lock")) is True: if not isinstance(prices, list) or len(prices) == 0: errors.append("prices_lock=true but prices_json is empty") if parse_bool(harness.get("quantities_lock")) is True: if not isinstance(sell_quantities, list) or len(sell_quantities) == 0: errors.append("quantities_lock=true but sell_quantities_json is empty") if not isinstance(buy_quantities, list): errors.append("quantities_lock=true but buy_qty_inputs_json is not a list") # E1: M4 — GOAL_RETIREMENT_V1 목표 자산 추적 검증 validate_goal_tracking(harness, errors) # G1: CASH_SHORTFALL_V1 — 현금 부족액 잠금 검증 validate_cash_shortfall(harness, errors) # G2: TRIM_PLAN_MIN_CASH_V1 — 현금 회복 TRIM 계획 구조 검증 validate_trim_plan(harness, errors) # I5: external_context_json — 외부 데이터 격리 구조 검증 (존재할 경우) validate_external_context(harness, errors) # S1: account_snapshot 장중 신선도 / 실행 게이트 검증 validate_snapshot_gate(harness, errors) # APEX_V1 — 선행 알파·설거지 차단·현금확보 실행품질 optional lock 검증 validate_apex_upgrade(harness, errors) # I3: CHECKSUM_V2 — source_manifest / decision_trace 체크섬 재산출 비교 _validate_checksums(harness, errors) # [2026-05-30] MARKET_WEIGHT_AWARE_CLUSTER_GATE_V1 / LEADER_POSITION_WEIGHT_CAP_V1 enum 검증 sc_gate = harness.get("semiconductor_cluster_gate") if sc_gate is not None and sc_gate not in VALID_SEMICONDUCTOR_CLUSTER_GATES: errors.append(f"semiconductor_cluster_gate: unknown value {sc_gate!r} — " f"허용값: {sorted(VALID_SEMICONDUCTOR_CLUSTER_GATES)}") sp_gate = harness.get("single_position_weight_gate") if sp_gate is not None and sp_gate not in VALID_SINGLE_POSITION_WEIGHT_GATES: errors.append(f"single_position_weight_gate: unknown value {sp_gate!r} — " f"허용값: {sorted(VALID_SINGLE_POSITION_WEIGHT_GATES)}") # [2026-05-20_HARNESS_V5] H6/H7/H8/Gate-4b 게이트 구조·열거형·불변식 검증 validate_v5_gates(harness, errors) # [2026-05-21_BRT_HARNESS_V1] BRT/SAQG/CCPL/SAPG 구조·열거형 검증 validate_brt_harness(harness, errors) # [2026-05-20_HARNESS_V5] V5 체크섬 불변식 검증 (rendered_report == blueprint, non_deterministic) _validate_v5_checksums(harness, errors) return errors def _validate_checksums(harness: dict[str, Any], errors: list[str]) -> None: """I3: CHECKSUM_V2 — source_manifest_checksum / decision_trace_checksum 재산출 검증.""" algo = harness.get("checksum_hash_algo") if algo not in (None, "CRC32_V1"): errors.append(f"checksum_hash_algo must be CRC32_V1, found={algo!r}") return # source_manifest_checksum sm_raw = harness.get("source_manifest_json") sm_stored = harness.get("source_manifest_checksum") if sm_raw is not None and sm_stored is not None: sm_str = sm_raw if isinstance(sm_raw, str) else json.dumps(sm_raw, ensure_ascii=False, separators=(",", ":")) sm_computed = _crc32_v1(sm_str) stored_num = int(sm_stored) if isinstance(sm_stored, str) and sm_stored.isdigit() else sm_stored if isinstance(stored_num, (int, float)) and int(stored_num) != sm_computed: errors.append( f"source_manifest_checksum mismatch: stored={stored_num}, computed={sm_computed}" ) # decision_trace_checksum dt_raw = harness.get("decision_trace_json") dt_stored = harness.get("decision_trace_checksum") if dt_raw is not None and dt_stored is not None: dt_str = dt_raw if isinstance(dt_raw, str) else json.dumps(dt_raw, ensure_ascii=False, separators=(",", ":")) dt_computed = _crc32_v1(dt_str) stored_num = int(dt_stored) if isinstance(dt_stored, str) and dt_stored.isdigit() else dt_stored if isinstance(stored_num, (int, float)) and int(stored_num) != dt_computed: errors.append( f"decision_trace_checksum mismatch: stored={stored_num}, computed={dt_computed}" ) def main() -> int: if len(sys.argv) != 2: print("Usage: python tools/validate_harness_context.py ") return 1 path = Path(sys.argv[1]) harness = load_harness(path) errors = validate_context(harness) if errors: print("HARNESS CONTEXT FAIL") for err in errors: print(f"- {err}") return 1 print("HARNESS CONTEXT OK") return 0 if __name__ == "__main__": sys.exit(main())