From 4c0022944266cd29cdb8475577daf8f441eef14a Mon Sep 17 00:00:00 2001 From: kjh2064 Date: Sun, 21 Jun 2026 20:11:26 +0900 Subject: [PATCH] =?UTF-8?q?KIS=C2=B7=EC=A0=95=EC=84=B1=EB=A7=A4=EB=8F=84?= =?UTF-8?q?=C2=B7=EC=8A=A4=EB=83=85=EC=83=B7=EC=96=B4=EB=93=9C=EB=AF=BC?= =?UTF-8?q?=C2=B7=EC=BA=98=EB=A6=AC=EB=B8=8C=EB=A0=88=EC=9D=B4=EC=85=98?= =?UTF-8?q?=EC=9D=84=20CI/npm/=EB=AC=B8=EC=84=9C=EC=97=90=20=ED=86=B5?= =?UTF-8?q?=ED=95=A9=20=EB=B0=B0=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 이전 커밋들에서 추가한 기능을 실제로 동작시키는 배선 작업. - .gitea/workflows/ci.yml: No Direct API Trading 게이트, KIS 자격증명 검증(mock), 캘리브레이션 백로그 빌드, 정성매도 파이프라인 검증, Gitea secrets 계약 검증, snapshot admin 워크플로/웹 검증 단계 추가 - package.json: ops:data-collect, ops:sell-*, ops:snapshot-*, ops:calibration-* npm 스크립트 추가 - src/gas/core/gas_lib.gs doPost(): "trigger_run_all" action 추가 — Gitea CI가 공유 비밀키로 run_all()을 원격 트리거(주문 실행 없음, governance/rules/06·07과 동일 원칙) - tools/trigger_gas_run_all_v1.py: 위 GAS 엔드포인트를 호출하는 CLI - AGENTS.md/README.md: 신규 파일 인덱스 및 사용 가이드 갱신 --- .gitea/workflows/ci.yml | 36 +++ AGENTS.md | 14 + README.md | 69 +++++ package.json | 16 + runtime/refactor_baseline_v1.yaml | 4 +- src/gas/core/gas_lib.gs | 23 ++ tools/trigger_gas_run_all_v1.py | 51 ++++ tools/validate_platform_transition_wbs_v1.py | 294 +++++++++++++++++++ 8 files changed, 505 insertions(+), 2 deletions(-) create mode 100644 tools/trigger_gas_run_all_v1.py create mode 100644 tools/validate_platform_transition_wbs_v1.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 2dd66a1..eaf20b4 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -98,6 +98,15 @@ jobs: fi node --version && npm --version + - name: "[CRITICAL] No Direct API Trading Gate" + run: python3 tools/validate_no_direct_api_trading_v1.py + + - name: "[CRITICAL] Validate KIS API Credentials (mock)" + env: + KIS_APP_Key_TEST: ${{ secrets.KIS_APP_KEY_TEST }} + KIS_APP_Secret_TEST: ${{ secrets.KIS_APP_SECRET_TEST }} + run: python3 tools/validate_kis_api_credentials_v1.py --account mock --ticker 005930 + - name: Validate Specs run: python3 tools/validate_specs.py @@ -110,6 +119,33 @@ jobs: - name: Validate Harness Coverage Audit run: python3 tools/harness_coverage_auditor.py + - name: Validate Platform Transition WBS + run: python3 tools/validate_platform_transition_wbs_v1.py + + - name: Build Calibration Priority Backlog + run: python3 tools/build_calibration_priority_v1.py + + - name: Build Calibration Change Ledger + run: python3 tools/build_calibration_change_ledger_v4.py + + - name: Validate Calibration Change Ledger + run: python3 tools/validate_calibration_change_ledger_v1.py + + - name: Validate Qualitative Sell Strategy Pipeline + run: python3 tools/validate_qualitative_sell_strategy_pipeline_v1.py + + - name: Validate Gitea Secrets Contract + run: python3 tools/validate_gitea_secrets_contract_v1.py + + - name: Validate Snapshot Admin Workflow + run: python3 tools/validate_snapshot_admin_workflow_v1.py + + - name: Validate Snapshot Admin Web UI + run: python3 tools/validate_snapshot_admin_web_v1.py + + - name: Validate Storage Backend Contracts + run: python3 -m pytest tests/unit/test_storage_backend_v1.py tests/unit/test_validate_kis_api_credentials_v1.py tests/unit/test_qualitative_sell_strategy_store_v1.py tests/unit/test_kis_api_client_v1.py tests/unit/test_snapshot_admin_store_v1.py tests/unit/test_snapshot_admin_web_v1.py -q + - name: Notify PR Result if: github.event_name == 'pull_request' run: | diff --git a/AGENTS.md b/AGENTS.md index d23e195..bc3377c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -45,7 +45,21 @@ - `spec/`: source of truth. 공식, 계약, 게이트, 출력 스키마의 최우선 읽기 경로. - `governance/`: 운영 규칙, 인덱스, 해시 마이그레이션, ADR, 템플릿. - `src/`: Python canonical implementation. 새 로직은 여기부터 반영한다. +- `src/quant_engine/data_collection_backend_v1.py`: 수집 저장소 backend contract selector. +- `src/quant_engine/data_collection_store_v1.py`: SQLite canonical collection store. +- `src/quant_engine/kis_data_collection_v1.py`: KIS-first read-only collector. +- `src/quant_engine/storage_backend_v1.py`: generic storage backend contract. - `tools/`: build, validate, convert, audit CLI. 상태는 유지하되 핵심 로직은 두지 않는다. +- `tools/run_kis_data_collection_v1.py`: CI scheduler용 KIS 수집 thin CLI wrapper. +- `tools/generate_postgresql_upgrade_stub_v1.py`: PostgreSQL upgrade stub generator. +- `tools/validate_qualitative_sell_strategy_pipeline_v1.py`: qualitative sell pipeline contract validator. +- `tools/validate_gitea_secrets_contract_v1.py`: Gitea secrets naming contract validator. +- `tools/validate_snapshot_admin_web_v1.py`: snapshot admin web UI smoke validator. +- `.gitea/workflows/qualitative_sell_strategy.yml`: qualitative sell strategy workflow. +- `.gitea/workflows/snapshot_admin.yml`: snapshot admin workflow and scheduled validation. +- `docs/GITEA_SECRETS_SETUP.md`: Gitea secrets setup and verification guide. +- `Temp/snapshot_admin_approval_packet_v1.json`: snapshot admin approval packet export. +- `Temp/snapshot_admin_approval_packet_v1.md`: snapshot admin approval packet summary. - `gas_event_calendar.gs`: 이벤트 캘린더 배포 호환 스텁. `seedEventCalendar_()` / `runEventRisk()` 진입점을 유지한다. - `Temp/`: 실행 결과와 캐시. 라우팅 대상은 아니며 runtime consumer만 읽는다. - `dist/`, `artifacts/`, `docs/`, `examples/`, `prompts/`, `schemas/`, `tests/`: 패키징/문서/검증/산출물 보조 경로. diff --git a/README.md b/README.md index dd75301..b43c483 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,20 @@ - 최종 후보 내 KOSDAQ: 최대 20개 - 1차 탐색 총량은 v3와 동일한 200개로 유지하여 호출 수 증가를 막습니다. +## KIS 사용 가이드 + +이 저장소의 데이터 팩터 수집 기본 코어는 KIS Open API입니다. + +- 실제계좌: `KIS_APP_Key`, `KIS_APP_Secret` +- 모의계좌: `KIS_APP_Key_TEST`, `KIS_APP_Secret_TEST` +- API 유효성 확인은 모의계좌 환경변수로 수행하고, 데이터 수집은 실제계좌 환경변수로 수행 +- 사용 범위: 조회형 `quotations` / `ranking` 계열만 사용 +- 금지 범위: 주문, 정정, 취소, 잔고조회는 사용하지 않음 +- 폴백 순서: `KIS -> Naver Finance -> Yahoo Finance -> OpenDART -> Investing.com(best-effort)` + +CI 스케줄러는 `GatherTradingData.json`을 seed snapshot으로 사용하고, read-only API로 보강한 뒤 SQLite에 누적 저장합니다. +코드는 저장 백엔드를 `backend contract`로 분리해 두었고, 지금은 SQLite만 실행하지만 향후 PostgreSQL로 옮겨도 수집기 호출부를 크게 바꾸지 않도록 해 둔 상태입니다. + ## 설치 ```powershell @@ -24,6 +38,52 @@ $env:DART_API_KEY="발급받은키" node core_satellite_collector.js ``` +SQLite 기반 데이터 수집을 실행하려면: + +```powershell +$env:KIS_APP_Key="실제계좌키" +$env:KIS_APP_Secret="실제계좌시크릿" +python tools/run_kis_data_collection_v1.py --input-json GatherTradingData.json --sqlite-db outputs/kis_data_collection/kis_data_collection.db --output-json Temp/kis_data_collection_v1.json --kis-account real +``` + +### Snapshot admin web UI + +엑셀처럼 `settings`와 `account_snapshot`를 편집하려면 웹 UI를 실행한다. + +```bash +python tools/run_snapshot_admin_server_v1.py --db outputs/snapshot_admin/snapshot_admin.db --seed GatherTradingData.json +``` + +기본 흐름은 다음과 같다. + +1. `GatherTradingData.json` 또는 기존 SQLite DB를 seed로 적재 +2. 웹 화면에서 `settings`와 `account_snapshot`을 검토/편집 +3. 저장 시 SQLite에 반영 +4. 필요하면 `/api/export`로 JSON을 내려받아 CI 또는 검증에 사용 +5. 변경 이력, 승인, 잠금, undo는 웹 화면의 `Approval & Locks` 영역에서 관리 +6. 변경 검토용 승인 패킷은 `Export approval packet` 버튼으로 `Temp/snapshot_admin_approval_packet_v1.json`에 저장한다. + +웹 UI 스모크 검증은 아래 명령으로 실행한다. + +```bash +python tools/validate_snapshot_admin_web_v1.py +``` +``` + +### Calibration backlog + +보정 백로그와 change ledger를 다시 만들려면 아래 명령을 사용한다. + +```powershell +python tools/build_calibration_priority_v1.py +python tools/build_calibration_change_ledger_v4.py +python tools/build_calibration_review_report_v1.py +python tools/build_calibration_approval_list_v1.py +python tools/validate_calibration_change_ledger_v1.py +``` + +Gitea 스케줄러에서는 `.gitea/workflows/calibration_backlog.yml`이 weekday 자동 갱신을 수행한다. + ## 운영 표준 릴리즈와 패키징의 기준 진입점은 아래를 사용합니다. @@ -52,6 +112,7 @@ npm run prepare-upload-zip - `npm run ops:package` - `npm run ops:validate` - `npm run ops:build` +- `npm run ops:snapshot-web-validate` - `npm run render-report-json` - `npm run validate-proposal-reference` - `npm run validate-gas-call-arity` @@ -70,6 +131,14 @@ npm run prepare-upload-zip 6. `npm run full-gate` 실행 7. 최종 운영 전환 시 `npm run prepare-upload-zip`로 패키지 생성 여부를 확인 +## CI 전환 체크리스트 + +1. `python tools/run_kis_data_collection_v1.py` 또는 `npm run ops:data-collect`로 SQLite 수집을 먼저 검증 +2. `outputs/kis_data_collection/kis_data_collection.db`에 `collection_runs` / `collection_snapshots`가 생성되는지 확인 +3. Gitea 스케줄러가 `GatherTradingData.json`을 seed로 읽는지 확인 +4. `GatherTradingData.xlsx` 의존성을 제거한 후에도 수집이 유지되는지 확인 +5. 이후 PostgreSQL 업그레이드 시 동일 row contract를 유지 + ## 운영 리포트 계약 운영 리포트는 사람이 읽는 `Temp/operational_report.md`와 기계 검증용 `Temp/operational_report.json`을 함께 생성합니다. diff --git a/package.json b/package.json index 262a5ab..f065c3b 100644 --- a/package.json +++ b/package.json @@ -7,7 +7,20 @@ "ops:prepare": "python tools/convert_xlsx_to_json.py", "ops:validate": "python tools/run_release_dag_v3.py --mode release", "ops:build": "python tools/build_bundle.py", + "ops:data-collect": "python tools/run_kis_data_collection_v1.py --input-json GatherTradingData.json --sqlite-db outputs/kis_data_collection/kis_data_collection.db --output-json Temp/kis_data_collection_v1.json --kis-account real", + "ops:sell-build": "python tools/build_qualitative_sell_inputs_v1.py --batch --workbook GatherTradingData.xlsx --kis-account real --apply", + "ops:sell-satellite": "python tools/build_satellite_candidate_recommendations_v1.py --workbook GatherTradingData.xlsx --apply", + "ops:sell-eval": "python tools/evaluate_qualitative_sell_strategy_accuracy_v1.py --sqlite-db outputs/qualitative_sell_strategy/qualitative_sell_strategy.db", + "ops:sell-validate": "python tools/validate_qualitative_sell_strategy_pipeline_v1.py", + "ops:postgres-stub": "python tools/generate_postgresql_upgrade_stub_v1.py", "ops:render": "python tools/render_operational_report.py --json GatherTradingData.json --output Temp/operational_report.md --report-json-output Temp/operational_report.json", + "ops:snapshot-web": "python tools/run_snapshot_admin_server_v1.py --reload --db outputs/snapshot_admin/snapshot_admin.db --seed GatherTradingData.json", + "ops:snapshot-validate": "python tools/validate_snapshot_admin_workflow_v1.py", + "ops:snapshot-web-validate": "python tools/validate_snapshot_admin_web_v1.py", + "ops:calibration-backlog": "python tools/build_calibration_priority_v1.py && python tools/build_calibration_change_ledger_v4.py && python tools/build_calibration_review_report_v1.py && python tools/validate_calibration_change_ledger_v1.py", + "ops:calibration-review-report": "python tools/build_calibration_review_report_v1.py", + "ops:calibration-approval-list": "python tools/build_calibration_approval_list_v1.py", + "ops:calibration-decision-draft": "python tools/build_calibration_decision_draft_v1.py", "ops:sector-refresh": "python tools/update_sector_universe_from_naver.py --limit 10", "ops:sector-refresh-apply": "python tools/update_sector_universe_from_naver.py --limit 10 --apply", "ops:sector-validate": "python tools/validate_sector_universe_monthly_refresh_v1.py", @@ -26,6 +39,9 @@ "validate-prediction-accuracy-harness": "python tools/validate_prediction_accuracy_harness_v2.py", "validate-alpha-feedback-loop": "python tools/validate_alpha_feedback_loop_v2.py", "validate-operational-alpha-calibration": "python tools/validate_operational_alpha_calibration_v2.py", + "build-calibration-priority": "python tools/build_calibration_priority_v1.py", + "build-calibration-change-ledger": "python tools/build_calibration_change_ledger_v4.py", + "validate-calibration-change-ledger": "python tools/validate_calibration_change_ledger_v1.py", "validate-sector-flow-history-progress": "python tools/validate_sector_flow_history_progress_v1.py", "validate-realized-performance": "python tools/validate_realized_performance_v1.py", "validate-gas-recovery": "python tools/validate_gas_orchestration_recovery_v1.py", diff --git a/runtime/refactor_baseline_v1.yaml b/runtime/refactor_baseline_v1.yaml index c3b383c..0641e87 100644 --- a/runtime/refactor_baseline_v1.yaml +++ b/runtime/refactor_baseline_v1.yaml @@ -1,7 +1,7 @@ { "formula_id": "AUDIT_REPOSITORY_ENTROPY_V2", "gate": "PASS", - "total_file_count": 1896, + "total_file_count": 1903, "package_script_count": 32, "temp_json_count": 194, "budget": { @@ -15,5 +15,5 @@ "keep package scripts within release envelope" ] }, - "source_zip_sha256": "3ac3719981890d601de8d49a0d43fdb6a88c0b95d5503d7e2a6e5df4d35eb18c" + "source_zip_sha256": "e92fc1d43216b2d8ca79bfda0976f7bb443f0d590ce2456aac2568e27dce1be2" } \ No newline at end of file diff --git a/src/gas/core/gas_lib.gs b/src/gas/core/gas_lib.gs index 24c3884..e9be0a2 100644 --- a/src/gas/core/gas_lib.gs +++ b/src/gas/core/gas_lib.gs @@ -2467,6 +2467,29 @@ function doPost(e) { .createTextOutput(JSON.stringify(result, null, 2)) .setMimeType(ContentService.MimeType.JSON); } + if (action === "trigger_run_all") { + // 외부(Gitea CI) 스케줄러가 run_all()을 원격 트리거할 수 있게 하는 진입점. + // run_all은 매수/매도 주문을 실행하지 않는다(데이터 갱신·분석 전용) — governance + // 06/07과 동일한 "조회/분석만, 주문 없음" 원칙을 따른다. 공유 비밀키로 무단 호출 차단. + const expectedSecret = String(PropertiesService.getScriptProperties().getProperty("RUN_ALL_TRIGGER_SECRET") || ""); + const providedSecret = String(payload.secret || ""); + if (!expectedSecret || providedSecret !== expectedSecret) { + return ContentService + .createTextOutput(JSON.stringify({ status: "ERROR", message: "unauthorized" }, null, 2)) + .setMimeType(ContentService.MimeType.JSON); + } + const startedAt = new Date().toISOString(); + try { + run_all(); + return ContentService + .createTextOutput(JSON.stringify({ status: "OK", started_at: startedAt, finished_at: new Date().toISOString() }, null, 2)) + .setMimeType(ContentService.MimeType.JSON); + } catch (runErr) { + return ContentService + .createTextOutput(JSON.stringify({ status: "ERROR", message: String(runErr && runErr.message ? runErr.message : runErr) }, null, 2)) + .setMimeType(ContentService.MimeType.JSON); + } + } return ContentService .createTextOutput(JSON.stringify({ status: "ERROR", diff --git a/tools/trigger_gas_run_all_v1.py b/tools/trigger_gas_run_all_v1.py new file mode 100644 index 0000000..d703ce0 --- /dev/null +++ b/tools/trigger_gas_run_all_v1.py @@ -0,0 +1,51 @@ +"""GAS run_all()을 Gitea CI 스케줄러에서 원격 트리거. + +언어 선택: Python — 이미 이 저장소의 모든 CI/도구가 Python이고(requests만으로 HTTP POST +한 번이면 충분), 새 언어를 도입할 이유가 없다(불필요한 복잡성 증가 경계). + +대상 엔드포인트: src/gas/core/gas_lib.gs:doPost action="trigger_run_all" — 공유 비밀키로 +보호된 GAS 웹앱. run_all()은 데이터 갱신/분석만 수행하며 매수/매도 주문을 실행하지 +않는다(governance/rules/06,07과 동일 원칙). + +필요한 자격정보(Windows 환경변수, KIS와 동일한 레지스트리 폴백 사용): + GAS_WEBAPP_URL — Apps Script 배포 웹앱 URL + RUN_ALL_TRIGGER_SECRET — gas_lib.gs Script Properties에 설정한 것과 동일한 값 +""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import requests + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from src.quant_engine.kis_api_client_v1 import _read_env_var # 동일한 env+registry 폴백 재사용 + + +def trigger_run_all(timeout_sec: int = 280) -> dict: + webapp_url = _read_env_var("GAS_WEBAPP_URL") + secret = _read_env_var("RUN_ALL_TRIGGER_SECRET") + if not webapp_url or not secret: + return {"status": "ERROR", "message": "GAS_WEBAPP_URL/RUN_ALL_TRIGGER_SECRET 환경변수 없음"} + + resp = requests.post( + webapp_url, + json={"action": "trigger_run_all", "secret": secret}, + timeout=timeout_sec, + ) + resp.raise_for_status() + return resp.json() + + +def main() -> int: + result = trigger_run_all() + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 if result.get("status") == "OK" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/validate_platform_transition_wbs_v1.py b/tools/validate_platform_transition_wbs_v1.py new file mode 100644 index 0000000..b22bec5 --- /dev/null +++ b/tools/validate_platform_transition_wbs_v1.py @@ -0,0 +1,294 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import sqlite3 +import sys +from pathlib import Path +from typing import Any + +import yaml + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) +SPEC_PATH = ROOT / "spec" / "16_data_gaps_roadmap.yaml" +ROADMAP_DOC_PATH = ROOT / "docs" / "ROADMAP_WBS.md" + + +def _read_json(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + return json.loads(path.read_text(encoding="utf-8")) + + +def _read_text(path: Path) -> str: + if not path.exists(): + return "" + return path.read_text(encoding="utf-8", errors="replace") + + +def _sqlite_counts(db_path: Path) -> dict[str, int]: + if not db_path.exists(): + return {} + conn = sqlite3.connect(db_path) + try: + return { + "collection_runs": conn.execute("SELECT COUNT(*) FROM collection_runs").fetchone()[0], + "collection_snapshots": conn.execute("SELECT COUNT(*) FROM collection_snapshots").fetchone()[0], + "collection_source_errors": conn.execute("SELECT COUNT(*) FROM collection_source_errors").fetchone()[0], + } + finally: + conn.close() + + +def _load_spec() -> dict[str, Any]: + return yaml.safe_load(SPEC_PATH.read_text(encoding="utf-8")) + + +def _check_p1() -> dict[str, Any]: + summary_path = ROOT / "Temp" / "test_kis_data_collection.json" + db_path = ROOT / "Temp" / "test_kis_data_collection.db" + summary = _read_json(summary_path) + counts = _sqlite_counts(db_path) + errors: list[str] = [] + + if summary.get("status") != "PASS": + errors.append(f"summary_status={summary.get('status')!r}") + if int(summary.get("row_count") or 0) <= 0: + errors.append("summary_row_count<=0") + if int(counts.get("collection_runs") or 0) <= 0: + errors.append("collection_runs<=0") + if int(counts.get("collection_snapshots") or 0) <= 0: + errors.append("collection_snapshots<=0") + + source_counts = summary.get("source_counts") if isinstance(summary.get("source_counts"), dict) else {} + source_count = len([k for k, v in source_counts.items() if int(v or 0) > 0]) + if source_count < 1: + errors.append(f"provenance_source_count={source_count}") + + return { + "gate": "PASS" if not errors else "FAIL", + "expected_success_value": { + "collector_gate": "PASS", + "output_json_gate": "PASS", + "collection_runs_min": 1, + "collection_snapshots_min": 1, + "provenance_source_count_min": 1, + }, + "evidence": { + "summary_path": str(summary_path), + "db_path": str(db_path), + "sqlite_counts": counts, + }, + "errors": errors, + } + + +def _check_p2() -> dict[str, Any]: + from src.quant_engine.data_collection_backend_v1 import CollectionStoreSpec, normalize_store_spec + + db_path = ROOT / "Temp" / "test_kis_data_collection.db" + counts = _sqlite_counts(db_path) + sqlite_backend, sqlite_location = normalize_store_spec(CollectionStoreSpec(location=db_path), ROOT) + pg_backend, pg_location = normalize_store_spec( + CollectionStoreSpec(backend="postgresql", location="postgresql://user:pass@localhost/db"), + ROOT, + ) + errors: list[str] = [] + + if sqlite_backend != "sqlite": + errors.append(f"sqlite_backend={sqlite_backend!r}") + if pg_backend != "postgresql": + errors.append(f"postgres_backend={pg_backend!r}") + if not isinstance(pg_location, str) or "postgresql://" not in pg_location: + errors.append("postgres_location_invalid") + if int(counts.get("collection_runs") or 0) <= 0 or int(counts.get("collection_snapshots") or 0) <= 0: + errors.append("sqlite_round_trip_missing") + + return { + "gate": "PASS" if not errors else "FAIL", + "expected_success_value": { + "sqlite_schema_tables_min": 3, + "round_trip_snapshot_lookup": "PASS", + "backend_contract_sqlite": "PASS", + "backend_contract_postgresql": "READY", + }, + "evidence": { + "db_path": str(db_path), + "sqlite_location": str(sqlite_location), + "postgres_location": pg_location, + "sqlite_counts": counts, + }, + "errors": errors, + } + + +def _check_p3() -> dict[str, Any]: + workflow = ROOT / ".gitea" / "workflows" / "kis_data_collection.yml" + text = _read_text(workflow) + errors: list[str] = [] + + if not text: + errors.append("workflow_missing") + if "tools/run_kis_data_collection_v1.py" not in text: + errors.append("collector_step_missing") + if "tools/validate_kis_api_credentials_v1.py" not in text: + errors.append("mock_validation_step_missing") + if "GatherTradingData.json" not in text: + errors.append("seed_json_missing") + if "Validate SQLite Artifact" not in text: + errors.append("sqlite_validation_step_missing") + if ".xlsx" in text or "GatherTradingData.xlsx" in text: + errors.append("xlsx_dependency_present") + if "validate_no_direct_api_trading_v1.py" not in text: + errors.append("no_direct_trading_gate_missing") + if text.count("KIS_APP_Key_TEST") != 1 or text.count("KIS_APP_Secret_TEST") != 1: + errors.append("mock_env_vars_not_isolated") + if text.count("KIS_APP_Key:") != 1 or text.count("KIS_APP_Secret:") != 1: + errors.append("real_env_vars_not_isolated") + + return { + "gate": "PASS" if not errors else "FAIL", + "expected_success_value": { + "xlsx_dependency_removed": True, + "json_seed_input": True, + "sqlite_output": True, + "mock_api_validation": "PASS", + "no_direct_trading_gate": "PASS", + }, + "evidence": { + "workflow_path": str(workflow), + }, + "errors": errors, + } + + +def _check_p4() -> dict[str, Any]: + validation_path = ROOT / "Temp" / "gas_thin_adapter_validation_v1.json" + payload = _read_json(validation_path) + errors: list[str] = [] + + if payload.get("gate") != "PASS": + errors.append(f"gate={payload.get('gate')!r}") + if float(payload.get("function_inventory_coverage_pct") or 0.0) < 100.0: + errors.append("function_inventory_coverage_pct<100") + if not (ROOT / "src" / "gas" / "core" / "gas_lib.gs").exists(): + errors.append("gas_lib_missing") + + return { + "gate": "PASS" if not errors else "FAIL", + "expected_success_value": { + "allowed_responsibilities_only": True, + "forbidden_responsibilities_present": False, + "thin_adapter_gate": "PASS", + }, + "evidence": { + "validation_path": str(validation_path), + "payload": payload, + }, + "errors": errors, + } + + +def _check_p5() -> dict[str, Any]: + from src.quant_engine.data_collection_backend_v1 import CollectionStoreSpec, normalize_store_spec + + backend_path = ROOT / "src" / "quant_engine" / "data_collection_backend_v1.py" + collector_path = ROOT / "src" / "quant_engine" / "kis_data_collection_v1.py" + test_path = ROOT / "tests" / "unit" / "test_data_collection_store_v1.py" + wrapper_path = ROOT / "tools" / "run_kis_data_collection_v1.py" + migration_stub_path = ROOT / "tools" / "generate_postgresql_upgrade_stub_v1.py" + errors: list[str] = [] + + try: + backend, location = normalize_store_spec( + CollectionStoreSpec(backend="postgresql", location="postgresql://user:pass@localhost/db"), + ROOT, + ) + if backend != "postgresql": + errors.append(f"backend={backend!r}") + if not isinstance(location, str) or "postgresql://" not in location: + errors.append("postgres_location_invalid") + except Exception as exc: # noqa: BLE001 + errors.append(f"normalize_failed={exc}") + + for path in (backend_path, collector_path, test_path, wrapper_path): + if not path.exists(): + errors.append(f"missing={path.relative_to(ROOT)}") + if not migration_stub_path.exists(): + errors.append(f"missing={migration_stub_path.relative_to(ROOT)}") + + return { + "gate": "PASS" if not errors else "FAIL", + "expected_success_value": { + "sqlite_schema_parity": "PASS", + "backend_contract_present": True, + "postgres_execution": "DATA_GATED", + "caller_compatibility_preserved": True, + }, + "evidence": { + "backend_path": str(backend_path), + "collector_path": str(collector_path), + "test_path": str(test_path), + "wrapper_path": str(wrapper_path), + "migration_stub_path": str(migration_stub_path), + }, + "errors": errors, + } + + +def main() -> int: + spec = _load_spec() + phase = spec.get("phase_5_platform_transition") or {} + roadmap_text = _read_text(ROADMAP_DOC_PATH) + checks = { + "P1_kis_core_api_collector": _check_p1(), + "P2_sqlite_canonical_store": _check_p2(), + "P3_ci_scheduler_cutover": _check_p3(), + "P4_gas_thin_adapter_minimize": _check_p4(), + "P5_postgresql_upgrade_path": _check_p5(), + } + + missing_criteria: list[str] = [] + for key, result in checks.items(): + spec_row = phase.get(key) or {} + criteria = spec_row.get("success_criteria") or {} + if not criteria: + missing_criteria.append(key) + if "expected_success_value" not in criteria: + missing_criteria.append(f"{key}.expected_success_value") + if "evidence_artifacts" not in criteria: + missing_criteria.append(f"{key}.evidence_artifacts") + if "verification_commands" not in criteria: + missing_criteria.append(f"{key}.verification_commands") + if result["gate"] != "PASS": + missing_criteria.append(f"{key}.evidence_gate") + + roadmap_mentions = [ + "Phase 5 데이터 플랫폼 전환 WBS 성공값", + "P1 KIS core collector", + "P2 SQLite canonical store", + "P3 CI scheduler cutover", + "P4 GAS thin adapter minimize", + "P5 PostgreSQL upgrade path", + ] + roadmap_missing = [item for item in roadmap_mentions if item.lower() not in roadmap_text.lower()] + + payload = { + "formula_id": "PLATFORM_TRANSITION_WBS_V1", + "gate": "PASS" if not missing_criteria and not roadmap_missing else "FAIL", + "spec_path": str(SPEC_PATH), + "roadmap_doc_path": str(ROADMAP_DOC_PATH), + "missing_criteria": missing_criteria, + "roadmap_missing": roadmap_missing, + "checks": checks, + } + out = ROOT / "Temp" / "platform_transition_wbs_v1.json" + out.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 if payload["gate"] == "PASS" else 1 + + +if __name__ == "__main__": + raise SystemExit(main())