KIS·정성매도·스냅샷어드민·캘리브레이션을 CI/npm/문서에 통합 배선

이전 커밋들에서 추가한 기능을 실제로 동작시키는 배선 작업.

- .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: 신규 파일 인덱스 및 사용 가이드 갱신
This commit is contained in:
2026-06-21 20:11:26 +09:00
parent 670ab8e15a
commit 4c00229442
8 changed files with 505 additions and 2 deletions
+36
View File
@@ -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: |
+14
View File
@@ -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/`: 패키징/문서/검증/산출물 보조 경로.
+69
View File
@@ -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`을 함께 생성합니다.
+16
View File
@@ -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",
+2 -2
View File
@@ -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"
}
+23
View File
@@ -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",
+51
View File
@@ -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())
@@ -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())