""" GAS_THIN_ADAPTER_POLICY_V1 — Phase 3: thin_adapter annotation spec/39_gas_thin_adapter_policy.yaml 참조. 각 GAS forbidden 함수의 첫 번째 실행 라인 직전에 // THIN_ADAPTER: delegated to Python — : 한 줄 주석을 삽입한다. 기능 코드는 변경하지 않는다 (additive-only). 이 주석은 Phase 4 검증 도구가 "이 함수는 이전 대상으로 등록됨"을 확인하는 마커가 된다. """ from __future__ import annotations import json import re import sys from pathlib import Path ROOT = Path(__file__).parent.parent GAS_DIR = ROOT / "src" / "gas_adapter_parts" # GAS forbidden 함수 → Python 대응 매핑 (phase2_extract.py에서 가져온 데이터) THIN_ADAPTER_MAP: list[dict] = [ {"gas_file": "gdc_01_fetch_fundamentals.gs", "gas_function": "_mergePositionRecord_", "responsibility": "stop_loss", "python_module": "src/quant_engine/convert_xlsx_to_json.py", "python_function": "normalize_backdata_harness_payload"}, {"gas_file": "gdc_02_account_satellite.gs", "gas_function": "_addTickerRoute_", "responsibility": "unknown", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "calc_semiconductor_cluster"}, {"gas_file": "gdf_01_price_metrics.gs", "gas_function": "calcApexTradePlan_", "responsibility": "sizing/normalize", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "compute_position_size"}, {"gas_file": "gdf_02_harness_assembly.gs", "gas_function": "assembleHarnessCoreLayers_", "responsibility": "sizing", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "main"}, {"gas_file": "gdf_02_harness_assembly.gs", "gas_function": "applyApexCashPreservationSuite_", "responsibility": "decision", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "cash_recovery"}, {"gas_file": "gdf_02_harness_assembly.gs", "gas_function": "applyApexFeedbackSignalSuite_", "responsibility": "decision", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "compute_final_decision"}, {"gas_file": "gdf_02_harness_assembly.gs", "gas_function": "applyProposal54BuyBlockLocks_", "responsibility": "decision", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "main"}, {"gas_file": "gdf_02_harness_assembly.gs", "gas_function": "calcStopBreachAlert_", "responsibility": "stop_loss", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "calc_stop_breach_alerts"}, {"gas_file": "gdf_02_harness_assembly.gs", "gas_function": "calcAbsoluteRiskStopV1_", "responsibility": "stop_loss", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "compute_stop_price_core"}, {"gas_file": "gdf_02_harness_assembly.gs", "gas_function": "calcTpTriggerAlert_", "responsibility": "take_profit", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "compute_tp_validity"}, {"gas_file": "gdf_03_portfolio_gates.gs", "gas_function": "calcTpQuantityLadder_", "responsibility": "sizing/take_profit", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "compute_position_size"}, {"gas_file": "gdf_03_portfolio_gates.gs", "gas_function": "scoreSellCandidate_", "responsibility": "decision", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "check_sanity"}, {"gas_file": "gdf_03_portfolio_gates.gs", "gas_function": "calcPrices_", "responsibility": "stop_loss/take_profit", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "compute_stop_price_core"}, {"gas_file": "gdf_03_portfolio_gates.gs", "gas_function": "runRouteFlow_", "responsibility": "stop_loss", "python_module": "tools/gas_thin_adapter_stubs_v1.py", "python_function": "stub_run_route_flow"}, {"gas_file": "gdf_03_portfolio_gates.gs", "gas_function": "buildOrderBlueprint_", "responsibility": "stop_loss/take_profit", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "main (order_blueprint_json)"}, {"gas_file": "gdf_03_portfolio_gates.gs", "gas_function": "calcDistributionRiskRow_", "responsibility": "risk_score", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "calc_distribution_detector_per_ticker"}, {"gas_file": "gdf_04_execution_quality.gs", "gas_function": "calcProfitPreservationRow_", "responsibility": "stop_loss", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "trailing_stop_v2"}, {"gas_file": "gdf_04_execution_quality.gs", "gas_function": "calcSmartCashRaiseV2_", "responsibility": "stop_loss", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "cash_recovery"}, {"gas_file": "gdf_04_execution_quality.gs", "gas_function": "calcApexExecutionHarness_", "responsibility": "sizing/decision", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "main"}, {"gas_file": "gdf_04_execution_quality.gs", "gas_function": "calcCashPreservationSellEngineV2_", "responsibility": "sizing", "python_module": "src/quant_engine/inject_computed_harness.py", "python_function": "cash_recovery"}, {"gas_file": "gdf_04_execution_quality.gs", "gas_function": "calcExportGate_", "responsibility": "unknown", "python_module": "tools/gas_thin_adapter_stubs_v1.py", "python_function": "stub_calc_export_gate"}, {"gas_file": "gdf_04_execution_quality.gs", "gas_function": "buildWatchLedger_", "responsibility": "stop_loss/take_profit", "python_module": "tools/gas_thin_adapter_stubs_v1.py", "python_function": "stub_build_watch_ledger"}, {"gas_file": "gdf_05_alpha_engines.gs", "gas_function": "buildShadowLedger_", "responsibility": "stop_loss/sizing/take_profit", "python_module": "src/quant_engine/compute_formula_outputs.py", "python_function": "check_sell_price_sanity"}, ] MARKER_PREFIX = "// THIN_ADAPTER:" def _build_annotation(entry: dict) -> str: return ( f" {MARKER_PREFIX} [{entry['responsibility']}] delegated to Python " f"— {entry['python_module']}:{entry['python_function']}" ) def _find_function_body_start(lines: list[str], func_name: str) -> int | None: """함수 선언 다음 줄 ({이 시작되는 줄 이후 첫 번째 실행 코드 라인 인덱스)를 반환한다.""" # function 선언 패턴: function funcName(... { pattern = re.compile( r"^(?:function\s+)" + re.escape(func_name) + r"\s*\(" ) for i, line in enumerate(lines): if pattern.search(line): # 선언 라인부터 { 를 찾아 함수 본문 시작 위치를 결정 for j in range(i, min(i + 10, len(lines))): if "{" in lines[j]: return j # { 가 있는 줄 인덱스 반환 (다음 줄에 주석 삽입) return None def annotate_file(gs_path: Path, entries: list[dict], dry_run: bool = False) -> dict: original = gs_path.read_text(encoding="utf-8") lines = original.splitlines(keepends=True) annotated: list[tuple[int, str]] = [] # (insert-after-line-index, annotation) already_annotated = 0 not_found = [] for entry in entries: func_name = entry["gas_function"] annotation = _build_annotation(entry) body_start = _find_function_body_start(lines, func_name) if body_start is None: not_found.append(func_name) continue # 이미 마커가 있으면 건너뜀 next_few = "".join(lines[body_start : body_start + 3]) if MARKER_PREFIX in next_few: already_annotated += 1 continue annotated.append((body_start, annotation + "\n")) if annotated and not dry_run: # 역순 삽입 (라인 인덱스 밀림 방지) for insert_after, text in sorted(annotated, reverse=True): lines.insert(insert_after + 1, text) gs_path.write_text("".join(lines), encoding="utf-8") return { "file": gs_path.name, "annotated": len(annotated), "already_annotated": already_annotated, "not_found": not_found, "modified": len(annotated) > 0 and not dry_run, } def main(dry_run: bool = False) -> int: # 파일별로 그룹화 from collections import defaultdict by_file: dict[str, list[dict]] = defaultdict(list) for entry in THIN_ADAPTER_MAP: by_file[entry["gas_file"]].append(entry) total_annotated = 0 results = [] for fname, entries in by_file.items(): gs_path = GAS_DIR / fname if not gs_path.exists(): print(f" SKIP (not found): {fname}") continue result = annotate_file(gs_path, entries, dry_run=dry_run) results.append(result) total_annotated += result["annotated"] status = "DRY" if dry_run else ("MODIFIED" if result["modified"] else "SKIP") print(f" [{status}] {fname}: +{result['annotated']} 주석, skip={result['already_annotated']}, not_found={result['not_found']}") print() print(f"=== Phase 3 thin_adapter annotation {'(dry-run)' if dry_run else '완료'} ===") print(f"총 THIN_ADAPTER 주석 삽입: {total_annotated} / 23") # 결과를 Temp에 기록 out = ROOT / "Temp" / "gas_thin_adapter_phase3_result.json" out.parent.mkdir(exist_ok=True) out.write_text(json.dumps({ "phase": "thin_adapter", "dry_run": dry_run, "total_annotated": total_annotated, "results": results, }, ensure_ascii=False, indent=2), encoding="utf-8") print(f"결과 저장: {out}") return 0 if __name__ == "__main__": dry = "--dry-run" in sys.argv sys.exit(main(dry_run=dry))