from __future__ import annotations import argparse import json from datetime import datetime from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_JSON = ROOT / "GatherTradingData.json" DEFAULT_REQUEST = ROOT / "Temp" / "request_result.txt" DEFAULT_OUT_TXT = ROOT / "Temp" / "request_result_adoption.txt" DEFAULT_OUT_JSON = ROOT / "Temp" / "request_result_adoption_v1.json" def _load_json(path: Path) -> dict[str, Any]: if not path.exists(): return {} try: obj = json.loads(path.read_text(encoding="utf-8")) except Exception: return {} return obj if isinstance(obj, dict) else {} def _parse_jsonish(value: Any) -> Any: if isinstance(value, (dict, list)): return value if isinstance(value, str) and value.strip(): try: return json.loads(value) except Exception: return value return value def _to_json_string_if_needed(original: Any, value: Any) -> Any: if isinstance(original, str): return json.dumps(value, ensure_ascii=False) return value def _rows(v: Any) -> list[dict[str, Any]]: p = _parse_jsonish(v) if isinstance(p, list): return [x for x in p if isinstance(x, dict)] return [] def _obj(v: Any) -> dict[str, Any]: p = _parse_jsonish(v) return p if isinstance(p, dict) else {} def _latest_snapshot_captured_at_iso(rows: list[dict[str, Any]]) -> str | None: latest_dt: datetime | None = None latest_iso: str | None = None for row in rows: if not isinstance(row, dict): continue for key in ("captured_at", "last_updated"): raw = str(row.get(key) or "").strip() if not raw: continue try: dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) except Exception: continue if dt.tzinfo is None: dt = dt.astimezone() if latest_dt is None or dt > latest_dt: latest_dt = dt latest_iso = raw return latest_iso def _build_predictive_dialectic_bridge(h: dict[str, Any]) -> dict[str, Any]: src = _rows(h.get("predictive_alpha_json")) if not src: return { "formula_id": "PREDICTIVE_ALPHA_DIALECTIC_ENGINE_V1_BRIDGE", "status": "DATA_MISSING", "source": "predictive_alpha_json", "rows": [], } out: list[dict[str, Any]] = [] for row in src: thesis = float(row.get("thesis_score") or 0.0) anti = float(row.get("antithesis_score") or 0.0) verdict = str(row.get("synthesis_verdict") or "") allowed = verdict not in {"EXIT_SIGNAL", "TRIM_SIGNAL"} out.append( { "ticker": str(row.get("ticker") or ""), "name": str(row.get("name") or ""), "thesis_score": thesis, "antithesis_score": anti, "synthesis_verdict": verdict, "action_allowed": allowed, "direction_confidence": row.get("direction_confidence"), "prediction_confidence_pct": row.get("prediction_confidence_pct"), "source_formula_id": str(row.get("formula_id") or ""), "bridge_rule": "B1_PREDICTIVE_ALPHA_JSON_TO_DIALECTIC", } ) return { "formula_id": "PREDICTIVE_ALPHA_DIALECTIC_ENGINE_V1_BRIDGE", "status": "BRIDGED", "source": "predictive_alpha_json", "rows": out, } def _build_dynamic_value_preservation_bridge(h: dict[str, Any]) -> dict[str, Any]: src = _rows(h.get("sell_value_preservation_json")) if not src: return { "formula_id": "DYNAMIC_VALUE_PRESERVATION_SELL_V3_BRIDGE", "status": "DATA_MISSING", "source": "sell_value_preservation_json", "rows": [], } out: list[dict[str, Any]] = [] for row in src: imm = row.get("immediate_qty") reb = row.get("rebound_wait_qty") imm_n = float(imm) if isinstance(imm, (int, float)) else 0.0 reb_n = float(reb) if isinstance(reb, (int, float)) else 0.0 total = imm_n + reb_n if total > 0: imm_ratio = round(imm_n / total, 4) reb_ratio = round(reb_n / total, 4) elif imm_n > 0: imm_ratio = 1.0 reb_ratio = 0.0 else: imm_ratio = 0.0 reb_ratio = 0.0 out.append( { "ticker": str(row.get("ticker") or ""), "name": str(row.get("name") or ""), "immediate_sell_ratio": imm_ratio, "rebound_wait_ratio": reb_ratio, "trailing_ratchet_active": bool(row.get("auto_trailing_stop")), "auto_trailing_stop": row.get("auto_trailing_stop"), "sell_value_preservation_state": str(row.get("sell_value_preservation_state") or ""), "reason_codes": row.get("reason_codes") if isinstance(row.get("reason_codes"), list) else [], "source_formula_id": str(row.get("formula_id") or ""), "bridge_rule": "B2_SELL_VALUE_PRESERVATION_TO_DYNAMIC_VALUE", } ) return { "formula_id": "DYNAMIC_VALUE_PRESERVATION_SELL_V3_BRIDGE", "status": "BRIDGED", "source": "sell_value_preservation_json", "rows": out, } def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--json", default=str(DEFAULT_JSON)) ap.add_argument("--request", default=str(DEFAULT_REQUEST)) ap.add_argument("--out-txt", default=str(DEFAULT_OUT_TXT)) ap.add_argument("--out-json", default=str(DEFAULT_OUT_JSON)) args = ap.parse_args() json_path = Path(args.json) req_path = Path(args.request) out_txt = Path(args.out_txt) out_json = Path(args.out_json) if not json_path.is_absolute(): json_path = ROOT / json_path if not req_path.is_absolute(): req_path = ROOT / req_path if not out_txt.is_absolute(): out_txt = ROOT / out_txt if not out_json.is_absolute(): out_json = ROOT / out_json payload = _load_json(json_path) if not payload: print("REQUEST_RESULT_ADOPTION_FAIL: invalid json payload") return 1 request_text = req_path.read_text(encoding="utf-8") if req_path.exists() else "" data = payload.get("data") if isinstance(payload.get("data"), dict) else {} hctx = data.get("_harness_context") if isinstance(data.get("_harness_context"), dict) else {} hapex = payload.get("hApex") if isinstance(payload.get("hApex"), dict) else {} merged = dict(hctx) merged.update(hapex) fresh_captured_at = _latest_snapshot_captured_at_iso(data.get("account_snapshot", []) if isinstance(data.get("account_snapshot"), list) else []) if fresh_captured_at: hctx["captured_at"] = fresh_captured_at hapex["captured_at"] = fresh_captured_at merged["captured_at"] = fresh_captured_at artifacts = { "gas_engine_upgrade_v7.gs": (ROOT / "gas_engine_upgrade_v7.gs").exists(), "tools/apply_engine_upgrade_v7.py": (ROOT / "tools" / "apply_engine_upgrade_v7.py").exists(), "spec/strategy/fundamental_quality_v2.yaml": (ROOT / "spec" / "strategy" / "fundamental_quality_v2.yaml").exists(), "spec/strategy/predictive_alpha_dialectic_v1.yaml": (ROOT / "spec" / "strategy" / "predictive_alpha_dialectic_v1.yaml").exists(), "spec/strategy/horizon_allocation_v1.yaml": (ROOT / "spec" / "strategy" / "horizon_allocation_v1.yaml").exists(), "spec/exit/dynamic_value_preservation_sell_v3.yaml": (ROOT / "spec" / "exit" / "dynamic_value_preservation_sell_v3.yaml").exists(), "spec/routing_trace_v2.yaml": (ROOT / "spec" / "routing_trace_v2.yaml").exists(), } adopted: list[dict[str, Any]] = [] for key, exists in artifacts.items(): adopted.append( { "item": key, "decision": "ADOPT" if exists else "REJECT", "reason": "artifact_exists" if exists else "artifact_missing", } ) bridge_applied: list[str] = [] if merged.get("predictive_alpha_dialectic_json") in (None, "", [], {}): merged["predictive_alpha_dialectic_json"] = _build_predictive_dialectic_bridge(merged) bridge_applied.append("predictive_alpha_dialectic_json") if merged.get("dynamic_value_preservation_json") in (None, "", [], {}): merged["dynamic_value_preservation_json"] = _build_dynamic_value_preservation_bridge(merged) bridge_applied.append("dynamic_value_preservation_json") # write back preserving json-string contract where needed hctx["predictive_alpha_dialectic_json"] = _to_json_string_if_needed(hctx.get("predictive_alpha_dialectic_json"), merged.get("predictive_alpha_dialectic_json")) hctx["dynamic_value_preservation_json"] = _to_json_string_if_needed(hctx.get("dynamic_value_preservation_json"), merged.get("dynamic_value_preservation_json")) hapex["predictive_alpha_dialectic_json"] = _to_json_string_if_needed(hapex.get("predictive_alpha_dialectic_json"), merged.get("predictive_alpha_dialectic_json")) hapex["dynamic_value_preservation_json"] = _to_json_string_if_needed(hapex.get("dynamic_value_preservation_json"), merged.get("dynamic_value_preservation_json")) data["_harness_context"] = hctx payload["data"] = data payload["hApex"] = hapex json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") summary = { "formula_id": "REQUEST_RESULT_ADOPTION_V1", "request_path": str(req_path), "request_loaded": bool(request_text.strip()), "artifact_adoption": adopted, "bridge_applied": bridge_applied, "post_keys": { "predictive_alpha_dialectic_json": merged.get("predictive_alpha_dialectic_json") is not None, "dynamic_value_preservation_json": merged.get("dynamic_value_preservation_json") is not None, }, } out_json.parent.mkdir(parents=True, exist_ok=True) out_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") lines = [ "# REQUEST_RESULT 채택/보완 결과", "", f"- formula_id: {summary['formula_id']}", f"- request_loaded: {summary['request_loaded']}", f"- bridge_applied: {', '.join(bridge_applied) if bridge_applied else 'none'}", "", "## 채택 판정", ] for row in adopted: lines.append(f"- {row['item']}: {row['decision']} ({row['reason']})") lines.extend( [ "", "## 근본 원인/정석 조치", "- 원인: 제안 아티팩트는 존재하나 실행 컨텍스트에 일부 키(predictive_alpha_dialectic_json, dynamic_value_preservation_json) 누락.", "- 조치: 기존 산출키(predictive_alpha_json, sell_value_preservation_json)에서 결정론적 브리지 적용.", "- 통제: 브리지 레코드에 source_formula_id/bridge_rule를 기록해 임의 수치 생성 금지.", ] ) out_txt.write_text("\n".join(lines) + "\n", encoding="utf-8") print(json.dumps(summary, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())