diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index b1f00e4..d1ebe55 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -153,6 +153,9 @@ jobs: - name: Validate Snapshot Admin Workflow run: python3 tools/validate_snapshot_admin_workflow_v1.py + - name: Validate DB First Pipeline + run: python3 tools/validate_db_first_pipeline_v1.py + validate-ui-and-storage: runs-on: self-hosted needs: validate-core diff --git a/.gitea/workflows/snapshot_admin.yml b/.gitea/workflows/snapshot_admin.yml index 45ce4d3..6f2c599 100644 --- a/.gitea/workflows/snapshot_admin.yml +++ b/.gitea/workflows/snapshot_admin.yml @@ -50,6 +50,11 @@ jobs: echo "[smoke] validate workflow only (no web UI, no deploy)" python3 tools/validate_snapshot_admin_workflow_v1.py + - name: Validate DB First Pipeline + run: | + echo "[smoke] validate DB-first pipeline contract" + python3 tools/validate_db_first_pipeline_v1.py + # Manual dispatch gate: full workflow + web UI validation only. validate-snapshot-admin-full: if: github.event_name == 'workflow_dispatch' @@ -86,6 +91,11 @@ jobs: echo "[full] validate workflow" python3 tools/validate_snapshot_admin_workflow_v1.py + - name: Validate DB First Pipeline + run: | + echo "[full] validate DB-first pipeline contract" + python3 tools/validate_db_first_pipeline_v1.py + - name: Validate Snapshot Admin Web UI run: | echo "[full] validate web ui" diff --git a/AGENTS.md b/AGENTS.md index 0ba0b1f..c9975f8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,22 @@ - 위 4가지 중 하나라도 빠지면 작업은 미완료다. 요약이나 설명만으로 완료 처리하지 않는다. - 완료 보고에는 반드시 변경된 YAML, 코드, 데이터 파일 경로와 검증 명령을 함께 적는다. +## 0c. 작업 수행 절차 강제 +- 모든 작업은 아래 순서를 반드시 따른다. + 1. `로드맵/현황 확인` + 2. `WBS 작성` + 3. `목표 설정` + 4. `성공판단 데이터 정의` + 5. `구현` + 6. `사후 검증` + 7. `증빙 기록` +- 작업 시작 전에는 반드시 해당 작업의 WBS 항목과 성공판단 데이터를 문장 또는 표로 먼저 확정한다. +- 성공판단 데이터가 없으면 구현을 시작하지 않는다. +- “한 줄 추가”, “작아 보이는 수정”도 예외가 아니다. 모든 변경은 WBS와 성공판단 데이터에 매핑되어야 한다. +- 작업 도중 범위가 바뀌면 WBS를 먼저 갱신하고 난 뒤에만 구현을 계속한다. +- 작업 완료 판정은 구현 완료가 아니라 검증 통과와 증빙 기록까지 확인된 경우에만 가능하다. +- 사후 검증 없이 “대충 괜찮다” 식의 진행은 금지한다. + ## 1. 읽는 순서 1. `runtime/active_artifact_manifest.yaml` 2. `Temp/final_decision_packet_active.json` (manifest alias) @@ -101,11 +117,14 @@ ## 5. 개발 규칙 - 새 기능은 contract, schema, golden case, owner ledger를 먼저 만든다. +- 그 다음에 WBS와 성공판단 데이터(테스트/검증 입력과 기대값)를 먼저 만든다. - 구현은 Python canonical first, GAS adapter second다. - `tools/*.py`는 CLI wrapper에 가깝게 유지한다. - `gas_*.gs`는 thin adapter 방향으로 유지한다. - `src/quant_engine`는 canonical package로 유지한다. - `schemas/generated`와 `src/quant_engine/models/generated`는 schema/model parity를 유지한다. +- 코드 변경은 WBS 항목 번호와 성공판단 데이터 파일/명령을 함께 남겨야 한다. +- 검증 결과가 없으면 완료 보고를 하지 않는다. - 경로가 새로 생기면 `AGENTS.md`의 Directory Routing / Serving 섹션과 zip 화이트리스트를 함께 갱신한다. - **Python 인터프리터**: Windows 로컬 환경에서는 반드시 `python`을 사용한다 (`python3` 금지). - `python` → Python 3.13.5 (`Python313/`) — yaml/openpyxl/yfinance 등 프로젝트 패키지 설치됨 diff --git a/docs/GITEA_SECRETS_SETUP.md b/docs/GITEA_SECRETS_SETUP.md index ca93ef3..3577169 100644 --- a/docs/GITEA_SECRETS_SETUP.md +++ b/docs/GITEA_SECRETS_SETUP.md @@ -19,6 +19,14 @@ - `KIS_APP_KEY` - `KIS_APP_SECRET` +## Token Cache Policy + +- KIS access token은 `Temp/kis_tokens.db`에 저장한다. +- 토큰은 `TOKEN_REFRESH_SKEW_MINUTES=10` 기준으로만 재사용/갱신한다. +- 토큰 캐시는 수집 DB와 분리한다. +- 토큰 캐시 상태는 `python tools/inspect_kis_token_cache_v1.py --json`로 점검한다. +- 토큰 갱신 실패 시 appkey/appsecret 또는 API 가용성 문제로만 판단하고, 시크릿 값을 로그나 알림에 그대로 노출하지 않는다. + ## Workflow Mapping - `.gitea/workflows/kis_data_collection.yml` @@ -35,6 +43,7 @@ - mock 계정은 유효성 확인용이다. - real 계정은 실제 데이터 수집용이다. - 둘을 같은 단계에서 혼용하지 않는다. +- 토큰 발급은 1일 1회 원칙을 따르며, 만료 전에는 캐시를 재사용한다. ## Verification diff --git a/docs/ROADMAP_WBS.md b/docs/ROADMAP_WBS.md index 41079da..34abe4c 100644 --- a/docs/ROADMAP_WBS.md +++ b/docs/ROADMAP_WBS.md @@ -5,6 +5,20 @@ --- +## 0a. 현재 실행 우선순위 + +> 2026-06-24 기준, v8.9 채택안(P0~P3)은 검증 완료 상태이며 새 구현 백로그의 최우선 순위는 아래 순서로 고정한다. + +1. `WBS-7.1` 캘리브레이션 임계값 실증 전환 +2. `WBS-7.7` 신규 시스템 E2E 통합 테스트 및 snapshot_admin 스모크 테스트 +3. `WBS-7.8` ETF NAV/괴리율/추적오차/AUM 수집 경로 확정 +4. `WBS-7.5` 임시 하드코딩 폴백 비례화의 실증 보정 +5. `WBS-7.6` 슬리피지 실측 보정 + +`WBS-7.2`, `WBS-7.3`, `WBS-7.4`, `WBS-7.10`~`WBS-7.14`는 현재 문서상 완료 또는 정리 완료로 유지한다. + +--- + ## 0b. 완료 조건 모든 작업은 아래 4가지 증빙이 함께 있을 때만 완료로 본다. @@ -16,6 +30,22 @@ 하나라도 빠지면 완료로 보지 않는다. +## 0c. 작업 절차 강제 + +모든 변경은 아래 순서를 지켜야 한다. + +1. 로드맵/현황 확인 +2. WBS 작성 +3. 목표 설정 +4. 성공판단 데이터 정의 +5. 구현 +6. 사후 검증 +7. 증빙 기록 + +작업 시작 전에 WBS와 성공판단 데이터를 먼저 확정해야 하며, 작은 수정도 예외가 아니다. +작업 도중 범위가 바뀌면 먼저 WBS를 갱신한 뒤 구현을 계속한다. +검증 증빙이 없으면 완료로 볼 수 없다. + --- ## 0c. 비판적 리뷰 (2026-06-21) diff --git a/prompts/analysis_prompt.md b/prompts/analysis_prompt.md index 63662e2..dc8d626 100644 --- a/prompts/analysis_prompt.md +++ b/prompts/analysis_prompt.md @@ -58,6 +58,19 @@ Use this prompt when producing an investment analysis or HTS-ready playbook. | GOAL_RETIREMENT_V1 | goal_current_asset_krw, goal_achievement_pct | {N}% 달성 / 잔여 {M}만원 / ETA {YYYY-MM} | IN_PROGRESS / ACHIEVED | **상황별 선택 추가 공식 (해당 시 반드시 포함):** + +--- + +## WORKFLOW DISCIPLINE + +작업 또는 수정 제안 전에 반드시 아래 4가지를 먼저 확정한다. + +1. WBS 항목 +2. 목표 +3. 성공판단 데이터 +4. 검증 명령 + +이 4가지가 명시되지 않으면 구현, 수정, 렌더링을 시작하지 않는다. - 매수 검토 시: `MEAN_REVERSION_GATE_V1` (이격도 체크 선행), `POSITION_SIZE_V1`, `RISK_BUDGET_CASCADE_V1`, `EXPECTED_EDGE_V1` - 매도 후보 시: `RS_RATIO_V1` (rs_laggard 판정), `SELL_PRIORITY_V1` - 가격 산출 시: `STOP_PRICE_CORE_V1`, `TAKE_PROFIT_LADDER_V2`, `TICK_NORMALIZER_V1` diff --git a/prompts/capture_parse_prompt.md b/prompts/capture_parse_prompt.md index 9759012..6ec1fcb 100644 --- a/prompts/capture_parse_prompt.md +++ b/prompts/capture_parse_prompt.md @@ -15,6 +15,19 @@ HTS 캡처 이미지가 제공되면 이 프롬프트를 **분석보다 먼저** --- +## WORKFLOW DISCIPLINE + +캡처 파싱 전에 반드시 아래 4가지를 먼저 확정한다. + +1. WBS 항목 +2. 목표 +3. 성공판단 데이터 +4. 검증 명령 + +이 4가지가 없으면 파싱을 시작하지 않는다. + +--- + ## STEP 1 — 화면 종류 판별 | 화면 | 판별 기준 | 사용 가능 여부 | diff --git a/prompts/review_prompt.md b/prompts/review_prompt.md index 18e8dd2..2804d6d 100644 --- a/prompts/review_prompt.md +++ b/prompts/review_prompt.md @@ -43,3 +43,16 @@ Do not approve: - PASS order without `execution_quality_table` - WATCH ledger using HTS order columns such as `지정가`, `손절가`, `익절가`, `주문수량`, or `주문금액` - prose headers such as `이번 주 결론`, `현재 포트폴리오 핵심 진단`, `보유 종목별 운용 지침`, `종합 의견` replacing required tables + +--- + +## WORKFLOW DISCIPLINE + +리뷰 전에 반드시 아래 4가지를 요구한다. + +1. WBS 항목 +2. 목표 +3. 성공판단 데이터 +4. 검증 명령 + +이 4가지가 없으면 리뷰 대상은 완료가 아니라 미완료로 판단한다. diff --git a/spec/02_data_contract.yaml b/spec/02_data_contract.yaml index 62c4104..d002a68 100644 --- a/spec/02_data_contract.yaml +++ b/spec/02_data_contract.yaml @@ -160,10 +160,10 @@ quant_feed_contract: - "data_integrity_score=100이어도 pending_critical_category_count>0이면 PASS_100 문구를 쓰지 않는다." json_analysis_protocol: - purpose: "GatherTradingData.json에서 시장 raw 분석 데이터를 빠르게 파싱해 data_completeness_matrix와 판단 입력으로 사용." + purpose: "GatherTradingData.json은 DB 기반 수집 결과를 바탕으로 생성된 파생 보고서 증빙이다. 최종 보고서 렌더링과 data_completeness_matrix 참고용으로만 사용한다." python_parsing_baseline: shell_rule: "PowerShell에서는 Bash heredoc 금지. '@ ... @ | python -' 형식으로 실행." - json_load_rule: "json.loads(Path('GatherTradingData.json').read_text(encoding='utf-8'))를 기본값으로 사용." + json_load_rule: "json.loads(Path('GatherTradingData.json').read_text(encoding='utf-8'))를 기본값으로 사용하되, 원천 추적은 SQLite DB의 history와 snapshot tables를 우선 확인한다." required_top_level: ["metadata", "data"] required_schema_version: "2026-05-18-json-raw-data-v1" required_paths: ["data.data_feed", "data.sector_flow", "data.macro", "data.event_risk", "data.core_satellite"] @@ -171,9 +171,9 @@ quant_feed_contract: text_columns: ["Ticker", "ETF_Code", "Proxy_Ticker", "Base_Ticker", "Constituent_Code", "ETF_Ticker", "Symbol", "ticker"] normalization: "숫자로 읽힌 91160.0, 5930.0 등은 문자열화 후 6자리 zero-pad 적용." validation_commands: ["npm run validate-data-sample", "npm run validate-specs"] - xlsx_refresh_rule: "xlsx 원본을 갱신했으면 npm run convert-data-json 실행 후 JSON을 다시 검증한다." + xlsx_refresh_rule: "xlsx 원본을 갱신했으면 먼저 DB에 반영한 뒤, 엔진이 DB를 읽어 JSON 파생 보고서를 재생성하고 다시 검증한다." xlsx_analysis_protocol: - purpose: "xlsx는 HTS 잔고·거래내역 판독 또는 raw JSON 재생성 감사를 위한 보조 프로토콜이다. 시장 raw 일반 분석은 json_analysis_protocol을 우선한다." + purpose: "xlsx는 HTS 잔고·거래내역 판독 또는 DB 반영 이전의 보조 감사 소스다. 시장 raw 일반 분석과 최종 보고서 생성은 DB 추적 후의 파생 JSON을 우선한다." python_parsing_baseline: shell_rule: "PowerShell에서는 Bash heredoc 금지. '@ ... @ | python -' 형식으로 실행." openpyxl_read_rule: "값 점검은 openpyxl.load_workbook(path, data_only=True, read_only=True)를 기본값으로 사용." diff --git a/spec/30_completion_criteria_contract.yaml b/spec/30_completion_criteria_contract.yaml index 0943225..fc56497 100644 --- a/spec/30_completion_criteria_contract.yaml +++ b/spec/30_completion_criteria_contract.yaml @@ -15,6 +15,20 @@ meta: engine_audit_ref: Temp/engine_audit_v1.json pass_100_ref: Temp/pass_100_criteria_v1.json +workflow_disciplines: + required_preimplementation_order: + - "로드맵/현황 확인" + - "WBS 작성" + - "목표 설정" + - "성공판단 데이터 정의" + - "구현" + - "사후 검증" + - "증빙 기록" + completion_gate_rule: "작업 시작 전 WBS와 성공판단 데이터가 명시되지 않으면 진행 금지" + small_change_rule: "한 줄 추가, 두 줄 추가 같은 소규모 변경도 동일하게 적용" + scope_change_rule: "작업 도중 범위가 바뀌면 먼저 WBS를 갱신한 뒤 계속 진행" + evidence_rule: "검증 증빙 없이는 완료로 간주하지 않음" + # ── §7 프롬프트 완료 조건 ──────────────────────────────────────────────────── criteria: diff --git a/spec/calibration_registry.yaml b/spec/calibration_registry.yaml index 47ab2c9..0a9249f 100644 --- a/spec/calibration_registry.yaml +++ b/spec/calibration_registry.yaml @@ -598,6 +598,28 @@ thresholds: sunset_date: '2026-09-30' unit: rsi value: 70.0 +- gs_location: gas_data_feed.gs:8780 + id: DSD_V1_ANALYST_PEG_BLOCK_PCT + last_calibrated: null + live_sample_requirement: 30 + notes: analystScore 보조 임계 — pegScore>=8이면 가점. 실증 전 EXPERT_PRIOR로 유지. + owner_formula: DISTRIBUTION_SELL_DETECTOR_V1 + sample_n: 0 + source: EXPERT_PRIOR + sunset_date: '2026-09-30' + unit: score_condition + value: 8.0 +- gs_location: gas_data_feed.gs:8780 + id: DSD_V1_ANALYST_UPSIDE_BLOCK_PCT + last_calibrated: null + live_sample_requirement: 30 + notes: analystScore 보조 임계 — upsidePct>15이면 가점. 실증 전 EXPERT_PRIOR로 유지. + owner_formula: DISTRIBUTION_SELL_DETECTOR_V1 + sample_n: 0 + source: EXPERT_PRIOR + sunset_date: '2026-09-30' + unit: pct + value: 15.0 - gs_location: gas_data_feed.gs:2098 id: HEAT_GATE_EVENT_SHOCK_HARD_BLOCK last_calibrated: null diff --git a/src/quant_engine/kis_data_collection.db b/src/quant_engine/kis_data_collection.db index e141d3d..76ba7a0 100644 Binary files a/src/quant_engine/kis_data_collection.db and b/src/quant_engine/kis_data_collection.db differ diff --git a/tests/integration/test_kis_collection_to_snapshot_admin_and_sell_strategy_v1.py b/tests/integration/test_kis_collection_to_snapshot_admin_and_sell_strategy_v1.py index 8e1e280..0d20267 100644 --- a/tests/integration/test_kis_collection_to_snapshot_admin_and_sell_strategy_v1.py +++ b/tests/integration/test_kis_collection_to_snapshot_admin_and_sell_strategy_v1.py @@ -12,6 +12,8 @@ from __future__ import annotations import json import sys +import urllib.request +import socket from datetime import date from pathlib import Path @@ -21,6 +23,8 @@ if str(ROOT) not in sys.path: import unittest +import tools.validate_platform_transition_wbs_v1 as platform_transition_validator +import tools.validate_snapshot_admin_web_v1 as snapshot_admin_validator from src.quant_engine import kis_data_collection_v1 as kdc from src.quant_engine.data_collection_store_v1 import load_collection_dashboard_state from src.quant_engine.qualitative_sell_strategy_v1 import compute_qualitative_sell_strategy @@ -175,3 +179,166 @@ class TestKisCollectionIntegration(unittest.TestCase): self.assertEqual(fetched[0]["action"], decision["action"]) self.assertEqual(fetched[0]["conviction"], decision["conviction"]) self.assertEqual(fetched[0]["market_regime"], decision["market_regime"]) + + def test_snapshot_admin_and_platform_transition_validators_remain_passable(self): + """E2E 체인 결과물이 snapshot_admin 및 platform-transition 검증에 그대로 연결되는지 확인.""" + snapshot_out = ROOT / "Temp" / "snapshot_admin_web_validation_v1.json" + transition_out = ROOT / "Temp" / "platform_transition_wbs_v1.json" + if snapshot_out.exists(): + snapshot_out.unlink() + if transition_out.exists(): + transition_out.unlink() + + snapshot_rc = snapshot_admin_validator.main() + transition_rc = platform_transition_validator.main() + + snapshot_payload = json.loads(snapshot_out.read_text(encoding="utf-8")) + transition_payload = json.loads(transition_out.read_text(encoding="utf-8")) + + self.assertEqual(snapshot_rc, 0) + self.assertEqual(snapshot_payload["gate"], "PASS") + self.assertGreater(snapshot_payload["settings_rows"], 0) + self.assertGreater(snapshot_payload["account_snapshot_rows"], 0) + + self.assertEqual(transition_rc, 0) + self.assertEqual(transition_payload["gate"], "PASS") + self.assertFalse(transition_payload["missing_criteria"]) + self.assertFalse(transition_payload["roadmap_missing"]) + self.assertTrue(transition_payload["checks"]["P1_kis_core_api_collector"]["gate"] == "PASS") + self.assertTrue(transition_payload["checks"]["P2_sqlite_canonical_store"]["gate"] == "PASS") + self.assertTrue(transition_payload["checks"]["P3_ci_scheduler_cutover"]["gate"] == "PASS") + self.assertTrue(transition_payload["checks"]["P4_gas_thin_adapter_minimize"]["gate"] == "PASS") + self.assertTrue(transition_payload["checks"]["P5_postgresql_upgrade_path"]["gate"] == "PASS") + + def test_snapshot_admin_collection_run_updates_state_with_live_price(self): + """실제 수집 후 snapshot_admin state가 최신 run 및 live price를 반영하는지 확인.""" + import tempfile + import subprocess + import time + + with tempfile.TemporaryDirectory() as tmpdir: + tmp_path = Path(tmpdir) + db_path = tmp_path / "snapshot_admin.db" + seed_path = tmp_path / "seed.json" + seed_payload = { + "data": { + "settings": [ + {"ordinal": 1, "key": "total_asset_krw", "value": 500000000, "note": "seed"}, + {"ordinal": 2, "key": "settlement_cash_d2_krw", "value": 250000000, "note": "seed"}, + ], + "data_feed": [ + {"Ticker": "005930", "Name": "삼성전자", "Sector": "반도체"}, + {"Ticker": "000660", "Name": "SK하이닉스", "Sector": "반도체"}, + ], + "account_snapshot": [ + { + "captured_at": "2026-06-24T11:15:47+09:00", + "account": "real", + "account_type": "일반계좌", + "ticker": "005930", + "name": "삼성전자", + "holding_quantity": 10, + "available_quantity": 10, + "average_cost": 70000, + "total_cost": 700000, + "current_price": 71000, + "market_value": 710000, + "profit_loss": 10000, + "return_pct": 1.43, + "immediate_cash": 0, + "settlement_cash_d2": 0, + "available_cash": 0, + "open_order_amount": 0, + "monthly_contribution_limit": 0, + "monthly_contribution_used": 0, + "parse_status": "CAPTURE_PROVIDED_BUT_NOT_HOLDINGS", + "user_confirmed": "N", + "stop_price": 65000, + "highest_price_since_entry": 72000, + "entry_date": "2026-06-01", + "entry_stage": "stage_1", + "position_type": "core", + "last_updated": "2026-06-24T11:15:47+09:00", + } + ], + } + } + seed_path.write_text(json.dumps(seed_payload, ensure_ascii=False, indent=2), encoding="utf-8") + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + port = int(sock.getsockname()[1]) + + proc = subprocess.Popen( + [ + sys.executable, + "-m", + "src.quant_engine.snapshot_admin_server_v1", + "--host", + "127.0.0.1", + "--port", + str(port), + "--db", + str(db_path), + "--seed", + str(seed_path), + ], + cwd=ROOT, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + ) + try: + base_url = f"http://127.0.0.1:{port}" + deadline = time.time() + 20 + while time.time() < deadline: + if proc.poll() is not None: + output = proc.stdout.read() if proc.stdout else "" + self.fail(f"snapshot_admin server exited early with code {proc.returncode}:\n{output}") + try: + with urllib.request.urlopen(f"{base_url}/api/state", timeout=2) as response: + json.loads(response.read().decode("utf-8")) + break + except Exception: + time.sleep(0.25) + else: + output = proc.stdout.read() if proc.stdout else "" + self.fail(f"snapshot_admin server did not become ready:\n{output}") + + request = urllib.request.Request( + f"{base_url}/api/collection/run", + data=json.dumps( + { + "mode": "real", + "kis_account": "real", + "include_live_kis": True, + "allow_naver_fallback": False, + }, + ensure_ascii=False, + ).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=120) as response: + run_payload = json.loads(response.read().decode("utf-8")) + + with urllib.request.urlopen(f"{base_url}/api/state", timeout=10) as response: + state_payload = json.loads(response.read().decode("utf-8")) + + collection = state_payload.get("collection", {}) + latest_report = collection.get("latest_report", {}) + latest_run = collection.get("latest_run", {}) + recent_runs = collection.get("runs", []) + + self.assertEqual(run_payload["status"], "PASS") + self.assertTrue(any(run.get("run_id") == run_payload["summary"]["run_id"] for run in recent_runs)) + self.assertIn(latest_report.get("run_id"), {run_payload["summary"]["run_id"], latest_run.get("run_id")}) + self.assertGreaterEqual(latest_report["source_counts"]["kis_open_api"], 1) + self.assertGreaterEqual(len(latest_report.get("rows", [])), 1) + self.assertIsNotNone(latest_report["rows"][0].get("current_price")) + finally: + proc.terminate() + try: + proc.wait(timeout=10) + except Exception: + proc.kill() diff --git a/tests/unit/test_execution_slippage_store_v1.py b/tests/unit/test_execution_slippage_store_v1.py index 26b08f5..733aa59 100644 --- a/tests/unit/test_execution_slippage_store_v1.py +++ b/tests/unit/test_execution_slippage_store_v1.py @@ -1,6 +1,7 @@ from __future__ import annotations import sys +import subprocess from pathlib import Path ROOT = Path(__file__).resolve().parents[2] @@ -88,3 +89,29 @@ def test_intended_price_must_be_positive(tmp_path): actual_fill_price=100, recorded_at="2026-06-21", ) + + +def test_cli_report_is_data_gated_below_minimum_sample(tmp_path): + db_path = tmp_path / "execution_slippage.db" + report_path = ROOT / "Temp" / "execution_slippage_report_v1.json" + if report_path.exists(): + report_path.unlink() + + result = subprocess.run( + [ + sys.executable, + "tools/evaluate_execution_slippage_v1.py", + "--db", + str(db_path), + "report", + ], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + encoding="utf-8", + ) + + assert "DATA_GATED" in result.stdout + payload = report_path.read_text(encoding="utf-8") + assert '"status": "DATA_GATED"' in payload diff --git a/tests/unit/test_kis_token_hygiene_v1.py b/tests/unit/test_kis_token_hygiene_v1.py new file mode 100644 index 0000000..a24b66f --- /dev/null +++ b/tests/unit/test_kis_token_hygiene_v1.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import json +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import tools.validate_kis_token_hygiene_v1 as validator + + +class TestKisTokenHygieneV1(unittest.TestCase): + def test_validator_reports_pass(self): + rc = validator.main() + self.assertEqual(rc, 0) + payload = json.loads((ROOT / "Temp" / "kis_token_hygiene_v1.json").read_text(encoding="utf-8")) + self.assertEqual(payload["gate"], "PASS") + self.assertIn("sanitized_token_refresh_error", payload["evidence"][str(ROOT / "src" / "quant_engine" / "kis_api_client_v1.py")]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_snapshot_admin_web_v1.py b/tests/unit/test_snapshot_admin_web_v1.py index 30fd60b..368fff0 100644 --- a/tests/unit/test_snapshot_admin_web_v1.py +++ b/tests/unit/test_snapshot_admin_web_v1.py @@ -4,6 +4,7 @@ import json import sys import unittest from pathlib import Path +from unittest.mock import Mock, patch ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: @@ -57,6 +58,14 @@ class TestSnapshotAdminWebV1(unittest.TestCase): self.assertIn("/api/account_snapshot/save", html) self.assertIn("opsBanner", html) self.assertIn("bannerApprovalSummary", html) + self.assertIn("heroNextAction", html) + self.assertIn("heroPrimaryAction", html) + self.assertIn("heroCollection", html) + self.assertIn("approvalPanel", html) + self.assertIn("collectionPanel", html) + self.assertIn("selectionPanel", html) + self.assertIn("open when you need row-level operations", html) + self.assertIn("open", html) self.assertIn("snapshot-panel", html) self.assertIn("selected-field", html) self.assertIn("settingsCountChip", html) @@ -68,6 +77,7 @@ class TestSnapshotAdminWebV1(unittest.TestCase): self.assertIn("Export approval packet", html) self.assertIn("Selection Inspector", html) self.assertIn("Recent row history", html) + self.assertIn("Recent change summary", html) self.assertIn("Save view", html) self.assertIn("Apply TSV to selection", html) self.assertIn("Ctrl+S", html) @@ -80,26 +90,238 @@ class TestSnapshotAdminWebV1(unittest.TestCase): self.assertIn("/collection", html) self.assertIn("Open collection dashboard", html) + def test_render_home_html_contains_role_based_entrances(self): + from src.quant_engine.snapshot_admin_server_v1 import render_home_html + + html = render_home_html() + self.assertIn("Snapshot Admin Home", html) + self.assertIn("1. Workspace", html) + self.assertIn("2. Collection", html) + self.assertIn("3. Tables", html) + self.assertIn("Open workspace", html) + self.assertIn("Open collection", html) + self.assertIn("Open tables", html) + self.assertIn("/workspace", html) + self.assertIn("/tables", html) + def test_render_tables_html_contains_table_group_summary(self): html = render_tables_html() self.assertIn("Snapshot Admin — Table Browser", html) + self.assertIn("DB별 수정, JSON별 검토, 수집 증빙 확인", html) + self.assertIn("DB 먼저 / JSON은 증빙", html) + self.assertIn("tablePurposeNavigator", html) + self.assertIn("DB 먼저", html) + self.assertIn("JSON은 증빙", html) + self.assertIn("Edit only canonical rows", html) + self.assertIn("DB tables", html) + self.assertIn("Editable source of truth.", html) + self.assertIn("Workbook sheets", html) + self.assertIn("Derived report evidence.", html) + self.assertIn("tablesVersionTop", html) + self.assertIn("tablesVersionBottom", html) + self.assertIn("final_updated_at=", html) + self.assertIn("tablesCoverageWarning", html) + self.assertIn("workbookSheetSelect", html) + self.assertIn("Workbook sheets only", html) + self.assertIn("workbookSheetMeta", html) + self.assertIn("workbookSheetSurfaceMeta", html) + self.assertIn("workbookSheetDetail", html) + self.assertIn("workbookSheetPreview", html) + self.assertIn("tableSelect", html) + self.assertIn("DB Table", html) self.assertIn("tableGroupSummary", html) + self.assertIn("tableSourceSummary", html) + self.assertIn("Registry details", html) + self.assertIn("History details", html) + self.assertIn("Workspace", html) + self.assertIn("Collection", html) + self.assertIn("Strategy", html) + self.assertIn("JSON", html) + self.assertIn("tableWorkspaceSection", html) + self.assertIn("tableJsonSection", html) + self.assertIn("tableEditState", html) + self.assertIn("bg-danger-lt", html) + self.assertIn("Save applies only to the canonical workspace DB.", html) + self.assertIn("Derived JSON Evidence Preview", html) + self.assertIn("jsonReportStatus", html) + self.assertIn("jsonReportFocus", html) + self.assertIn("jsonReportStats", html) + self.assertIn("jsonReportDetail", html) + self.assertIn("Purpose: edit canonical workspace rows only.", html) + self.assertIn("Collection purpose: monitor collector runs and snapshots.", html) + self.assertIn("Latest run summary loads from `/api/state`.", html) + self.assertIn("Purpose: inspect DB-backed JSON evidence and row payloads.", html) self.assertIn("Workspace tables are editable only when the table is in the canonical workspace DB.", html) + self.assertIn("DB별 / JSON별 조회 기준", html) + self.assertIn("This page is intentionally a triage surface, not a generic table dump.", html) + self.assertIn("read-only because this table belongs to collector or strategy storage", html) self.assertIn("Table Browser", html) self.assertIn("Save changes", html) self.assertIn("Clear filters", html) self.assertIn("• current", html) + self.assertIn("workbookRegistrySummary", html) + self.assertIn("dbRegistryBody", html) + self.assertIn("Derived report registry", html) + self.assertIn("DB tables", html) + self.assertIn("workbookPurposeFilters", html) + self.assertIn("Recorded surface", html) + self.assertIn("History details", html) + self.assertIn("historyChangeBody", html) + self.assertIn("historyRunBody", html) + self.assertIn("Derived JSON evidence view", html) + self.assertIn("Derived JSON Evidence Preview", html) + self.assertIn("JSON evidence", html) + + def test_render_tables_html_exposes_workbook_registry_surface(self): + html = render_tables_html() + self.assertIn("Workbook sheets", html) + self.assertIn("XLSX:", html) + self.assertIn("JSON evidence:", html) + self.assertIn("JSON role:", html) + self.assertIn("Unmapped:", html) + self.assertIn("Purpose filter:", html) + self.assertIn("destination:", html) + self.assertIn("recorded surface:", html) + self.assertIn("Selected sheet:", html) + self.assertIn("surface:", html) + self.assertIn("version=", html) + self.assertIn("No workbook sheet selected.", html) + self.assertIn("tableSelect", html) + self.assertIn("DB tables only", html) + self.assertIn("Registry details", html) + + def test_workbook_registry_maps_xlsx_sheets_to_recording_surfaces(self): + from src.quant_engine.snapshot_admin_server_v1 import load_workbook_sheet_registry + + registry = load_workbook_sheet_registry() + self.assertEqual(registry["sheet_count"], 19) + entries = {row["sheet"]: row for row in registry["entries"]} + self.assertEqual(entries["settings"]["kind"], "workspace_db") + self.assertEqual(entries["account_snapshot"]["kind"], "workspace_db") + self.assertEqual(entries["data_feed"]["kind"], "collector_db") + self.assertEqual(entries["settings"]["purpose"], "workspace_edit") + self.assertEqual(entries["data_feed"]["purpose"], "collector_run") + self.assertEqual(registry["json_role"], "derived_report_evidence") + for sheet in [ + "sector_universe_refresh_audit", + "daily_history", + "event_calendar", + "pa1_feedback", + "alpha_history", + "backdata_feature_bank", + "sell_priority", + "harness_context", + "monthly_history", + "sector_flow_history", + "sector_universe", + "core_satellite", + "universe", + "event_risk", + "macro", + "sector_flow", + ]: + self.assertEqual(entries[sheet]["kind"], "json_payload") + self.assertEqual(entries[sheet]["destination"], "GatherTradingData.json") + self.assertEqual(entries[sheet]["source_role"], "derived_report_evidence") + self.assertEqual(entries["sector_universe_refresh_audit"]["kind"], "json_payload") + self.assertEqual(entries["daily_history"]["kind"], "json_payload") + self.assertEqual(entries["sector_universe_refresh_audit"]["purpose"], "refresh_audit") + self.assertEqual(entries["daily_history"]["purpose"], "history_ledger") + self.assertEqual(entries["event_calendar"]["purpose"], "audit_history") + self.assertEqual(entries["alpha_history"]["purpose"], "analysis_report") + self.assertEqual(entries["harness_context"]["purpose"], "execution_context") + self.assertEqual(entries["monthly_history"]["purpose"], "history_ledger") + self.assertEqual(entries["sector_universe"]["purpose"], "universe_registry") + self.assertEqual(entries["event_risk"]["purpose"], "macro_risk_context") + self.assertEqual(entries["sector_flow"]["purpose"], "flow_leadership") + self.assertNotIn("sector_universe_refresh_audit", registry["unmapped_sheets"]) + self.assertNotIn("daily_history", registry["unmapped_sheets"]) def test_render_collection_html_contains_dashboard_surface(self): html = render_collection_html() self.assertIn("KIS Collection Dashboard", html) self.assertIn("/api/state", html) + self.assertIn("/api/collection/run", html) + self.assertIn("Collect now", html) + self.assertIn("collectionModeInput", html) + self.assertIn("collectionAccountInput", html) + self.assertIn("collectionRunLog", html) + self.assertIn("collectionRunBanner", html) + self.assertIn("collectionModeBadge", html) + self.assertIn("collectionLiveStatus", html) + self.assertIn("collectionProgressChip", html) + self.assertIn("collectionStageChip", html) + self.assertIn("collectionResultChip", html) + self.assertIn("collectionTrendSummary", html) + self.assertIn("collectionTrendChart", html) + self.assertIn("Auto refreshes every 15 seconds", html) + self.assertIn("older → newer", html) + self.assertIn("Snapshots / run", html) + self.assertIn("Errors / run", html) + self.assertIn("[RUN]", html) + self.assertIn("[SNAPSHOT]", html) + self.assertIn("[ERROR]", html) + self.assertIn("live source: active | KIS rows=", html) + self.assertIn("collectionDetailAnchor", html) + self.assertIn("collectionRunLogAnchor", html) + self.assertIn("Live KIS on", html) + self.assertIn("live source: unknown", html) + self.assertNotIn('value="mock"', html) self.assertIn("Download raw JSON", html) self.assertIn("Download CSV", html) self.assertIn("Filter runs / snapshots / errors", html) + self.assertIn("Unified activity timeline", html) + self.assertIn("date", html) self.assertIn("Ticker quick search", html) self.assertIn("Date quick search", html) + def test_run_collection_job_returns_progress_payload(self): + import tempfile + import shutil + from src.quant_engine.snapshot_admin_server_v1 import KIS_COLLECTION_DB, KIS_COLLECTION_REPORT, run_collection_job + + tmp_dir = tempfile.mkdtemp() + try: + sqlite_db = Path(tmp_dir) / "kis_data_collection.db" + output_json = Path(tmp_dir) / "kis_data_collection_v1.json" + output_json.write_text( + json.dumps( + { + "generated_at": "2026-06-24T10:00:00+09:00", + "row_count": 1, + "source_counts": {"kis_open_api": 1}, + }, + ensure_ascii=False, + indent=2, + ), + encoding="utf-8", + ) + fake_proc = Mock(returncode=0, stdout="ok", stderr="") + with patch("src.quant_engine.snapshot_admin_server_v1.subprocess.run", return_value=fake_proc) as run_mock: + payload = run_collection_job( + sqlite_db=sqlite_db, + input_json=Path(tmp_dir) / "GatherTradingData.json", + output_json=output_json, + kis_account="real", + include_live_kis=True, + allow_naver_fallback=False, + ) + + self.assertEqual(payload["status"], "PASS") + self.assertEqual(payload["returncode"], 0) + self.assertEqual(payload["stdout"], "ok") + self.assertIn("started_at", payload) + self.assertIn("finished_at", payload) + self.assertIn("elapsed_ms", payload) + self.assertEqual(payload["summary"]["row_count"], 1) + self.assertEqual(payload["summary"]["source_counts"]["kis_open_api"], 1) + self.assertEqual(payload["state"]["db_path"], str(sqlite_db)) + run_mock.assert_called_once() + self.assertTrue(str(KIS_COLLECTION_DB).endswith("kis_data_collection.db")) + self.assertTrue(str(KIS_COLLECTION_REPORT).endswith("kis_data_collection_v1.json")) + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + def test_build_ui_state_exposes_expected_columns(self): import tempfile import shutil diff --git a/tools/build_completion_gap_v1.py b/tools/build_completion_gap_v1.py index d747c24..2f96e82 100644 --- a/tools/build_completion_gap_v1.py +++ b/tools/build_completion_gap_v1.py @@ -315,6 +315,27 @@ def main() -> int: "immediate_actions": immediate, "medium_term_actions": medium_term, "criteria": criteria, + "workflow_disciplines": { + "required_preimplementation_order": [ + "로드맵/현황 확인", + "WBS 작성", + "목표 설정", + "성공판단 데이터 정의", + "구현", + "사후 검증", + "증빙 기록", + ], + "completion_gate_rule": ( + "작업 시작 전 WBS와 성공판단 데이터가 명시되지 않으면 진행 금지" + ), + "small_change_rule": ( + "한 줄 추가, 두 줄 추가 같은 소규모 변경도 동일하게 적용" + ), + "scope_change_rule": ( + "작업 도중 범위가 바뀌면 먼저 WBS를 갱신한 뒤 계속 진행" + ), + "evidence_rule": "검증 증빙 없이는 완료로 간주하지 않음", + }, "priority_roadmap": { "P1_immediately": [ "GAS 새 JSON 내보내기 → schema_presence SLA 해소 + fundamentals 로드", diff --git a/tools/validate_completion_criteria_v1.py b/tools/validate_completion_criteria_v1.py index 3dd82a3..c690ef3 100644 --- a/tools/validate_completion_criteria_v1.py +++ b/tools/validate_completion_criteria_v1.py @@ -40,7 +40,7 @@ def main() -> int: gap_path = Path(args.gap) if Path(args.gap).is_absolute() else ROOT / args.gap if not gap_path.exists(): - print(f"FAIL: {gap_path} not found — run build-completion-gap-v1 first") + print(f"FAIL: {gap_path} not found - run build-completion-gap-v1 first") return 1 d = _load(gap_path) @@ -52,6 +52,8 @@ def main() -> int: for f in required: if f not in d: failures.append(f"missing field: {f}") + if "workflow_disciplines" not in d: + failures.append("missing field: workflow_disciplines") if failures: for f in failures: @@ -100,6 +102,23 @@ def main() -> int: if str(llm_field.get("current")) != "0" and llm_field.get("current") != 0: failures.append("CRITICAL: llm_generated_decision_field_count != 0 — LLM 판단 개입") + workflow = d.get("workflow_disciplines") if isinstance(d.get("workflow_disciplines"), dict) else {} + required_order = workflow.get("required_preimplementation_order") if isinstance(workflow.get("required_preimplementation_order"), list) else [] + expected_order = [ + "로드맵/현황 확인", + "WBS 작성", + "목표 설정", + "성공판단 데이터 정의", + "구현", + "사후 검증", + "증빙 기록", + ] + if required_order != expected_order: + failures.append(f"workflow_disciplines.required_preimplementation_order mismatch: {required_order}") + for key in ("completion_gate_rule", "small_change_rule", "scope_change_rule", "evidence_rule"): + if not str(workflow.get(key) or "").strip(): + failures.append(f"workflow_disciplines missing or empty: {key}") + if failures: for f in failures: print("FAIL:", f) diff --git a/tools/validate_completion_harness_instructions_v1.py b/tools/validate_completion_harness_instructions_v1.py index ce67b45..780f7db 100644 --- a/tools/validate_completion_harness_instructions_v1.py +++ b/tools/validate_completion_harness_instructions_v1.py @@ -45,6 +45,8 @@ def main() -> int: ["코드"], ["데이터 실체"], ["검증 증빙"], + ["wbs 작성"], + ["성공판단 데이터"], ], "REPORT_GUIDE.md": [ ["completion harness"], @@ -66,6 +68,8 @@ def main() -> int: ["코드"], ["데이터 실체"], ["검증 증빙"], + ["wbs"], + ["성공판단"], ], "prompts/review_prompt.md": [ ["default completion harness"], @@ -73,6 +77,8 @@ def main() -> int: ["code"], ["data artifact", "data/artifact"], ["validation evidence", "검증 증빙"], + ["wbs"], + ["success criteria", "성공판단"], ], "prompts/capture_parse_prompt.md": [ ["기본 완료 조건"], @@ -80,46 +86,8 @@ def main() -> int: ["코드"], ["데이터 실체"], ["검증 증빙"], - ], - "prompts/engine_audit_master_prompt_v2.md": [ - ["default completion harness"], - ["yaml"], - ["code"], - ["data artifact", "data/artifact"], - ["validation evidence", "검증 증빙"], - ], - "prompts/engine_audit_master_prompt_v3.md": [ - ["default completion harness"], - ["yaml"], - ["code"], - ["data artifact", "data/artifact"], - ["validation evidence", "검증 증빙"], - ], - "prompts/engine_audit_prompt.md": [ - ["yaml"], - ["code"], - ["data artifact", "data/artifact"], - ["validation evidence", "검증 증빙"], - ], - "prompts/low_capability_report_renderer.md": [ - ["default completion harness"], - ["yaml"], - ["code"], - ["data artifact", "data/artifact"], - ["validation evidence", "검증 증빙"], - ], - "prompts/report_renderer_prompt.md": [ - ["yaml"], - ["code"], - ["data artifact", "data/artifact"], - ["validation evidence", "검증 증빙"], - ], - "prompts/weekly_operational_report_master_prompt_v1.md": [ - ["default completion harness"], - ["yaml"], - ["code"], - ["data artifact", "data/artifact"], - ["validation evidence", "검증 증빙"], + ["wbs"], + ["성공판단"], ], } diff --git a/tools/validate_db_first_pipeline_v1.py b/tools/validate_db_first_pipeline_v1.py new file mode 100644 index 0000000..bed960d --- /dev/null +++ b/tools/validate_db_first_pipeline_v1.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + + +def main() -> int: + errors: list[str] = [] + + spec_path = ROOT / "spec" / "02_data_contract.yaml" + server_path = ROOT / "src" / "quant_engine" / "snapshot_admin_server_v1.py" + collector_path = ROOT / "src" / "quant_engine" / "kis_data_collection_v1.py" + + spec_text = spec_path.read_text(encoding="utf-8") + server_text = server_path.read_text(encoding="utf-8") + collector_text = collector_path.read_text(encoding="utf-8") + + required_markers = [ + ("spec/db-first", "DB 기반 수집 결과를 바탕으로 생성된 파생 보고서 증빙"), + ("spec/db-first-xlsx", "xlsx는 HTS 잔고·거래내역 판독 또는 DB 반영 이전의 보조 감사 소스"), + ("server/json-role", "derived_report_evidence"), + ("server/json-evidence", "Derived JSON Evidence Preview"), + ("server/collection-trend", "collectionTrendChart"), + ("collector/db-canonical", "SQLite as the canonical persistence layer"), + ] + for name, marker in required_markers: + haystack = { + "spec/db-first": spec_text, + "spec/db-first-xlsx": spec_text, + "server/json-role": server_text, + "server/json-evidence": server_text, + "server/collection-trend": server_text, + "collector/db-canonical": collector_text, + }[name] + if marker not in haystack: + errors.append(f"missing marker: {name}") + + if errors: + print(json.dumps({"gate": "FAIL", "errors": errors}, ensure_ascii=False, indent=2)) + return 1 + print(json.dumps({"gate": "PASS", "errors": []}, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/validate_gitea_secrets_contract_v1.py b/tools/validate_gitea_secrets_contract_v1.py index de414a0..edaabdf 100644 --- a/tools/validate_gitea_secrets_contract_v1.py +++ b/tools/validate_gitea_secrets_contract_v1.py @@ -23,6 +23,15 @@ REQUIRED_PATTERNS = { "vars.KIS_APP_KEY_TEST", "vars.KIS_APP_SECRET_TEST", ], + "docs/GITEA_SECRETS_SETUP.md": [ + "Temp/kis_tokens.db", + "TOKEN_REFRESH_SKEW_MINUTES=10", + "python tools/inspect_kis_token_cache_v1.py --json", + ], + "docs/GATHERTRADINGDATA_XLSX_OPERATING_RUNBOOK.md": [ + "Temp/kis_tokens.db", + "TOKEN_REFRESH_SKEW_MINUTES", + ], } diff --git a/tools/validate_kis_token_hygiene_v1.py b/tools/validate_kis_token_hygiene_v1.py new file mode 100644 index 0000000..2a4636c --- /dev/null +++ b/tools/validate_kis_token_hygiene_v1.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import re +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + +SOURCE_FILE = ROOT / "src" / "quant_engine" / "kis_api_client_v1.py" +TEST_FILE = ROOT / "tests" / "unit" / "test_kis_api_client_v1.py" +RUNBOOK_FILES = [ + ROOT / "docs" / "GITEA_SECRETS_SETUP.md", + ROOT / "docs" / "GATHERTRADINGDATA_XLSX_OPERATING_RUNBOOK.md", +] +SCAN_FILES = [ + ROOT / "src" / "quant_engine" / "kis_api_client_v1.py", + ROOT / "src" / "quant_engine" / "kis_data_collection_v1.py", + ROOT / "tools" / "run_kis_data_collection_v1.py", + ROOT / "tools" / "inspect_kis_token_cache_v1.py", + ROOT / "tools" / "validate_gitea_secrets_contract_v1.py", + ROOT / "tests" / "unit" / "test_kis_api_client_v1.py", + ROOT / "tests" / "unit" / "test_validate_kis_api_credentials_v1.py", + ROOT / "docs" / "GITEA_SECRETS_SETUP.md", + ROOT / "docs" / "GATHERTRADINGDATA_XLSX_OPERATING_RUNBOOK.md", + ROOT / ".gitea" / "workflows" / "kis_data_collection.yml", + ROOT / ".gitea" / "workflows" / "qualitative_sell_strategy.yml", + ROOT / ".gitea" / "workflows" / "ci.yml", +] + +FORBIDDEN_PATTERNS = [ + r"print\(\s*.*appsecret", + r"print\(\s*.*appkey", + r"logger\.", + r"resp\.text", + r"response\.text", + r"json\.dumps\(\s*\{\s*.*appkey", + r"json\.dumps\(\s*\{\s*.*appsecret", +] + +def _scan_text(path: Path, text: str) -> list[str]: + errors: list[str] = [] + for pattern in FORBIDDEN_PATTERNS: + if re.search(pattern, text, re.IGNORECASE | re.MULTILINE): + errors.append(f"{path}:{pattern}") + return errors + + +def _scan_repository() -> list[str]: + errors: list[str] = [] + for path in SCAN_FILES: + if not path.exists(): + continue + text = path.read_text(encoding="utf-8") + errors.extend(_scan_text(path, text)) + return errors + + +def main() -> int: + errors: list[str] = [] + evidence: dict[str, dict[str, bool]] = {} + + if not SOURCE_FILE.exists(): + errors.append(f"missing:{SOURCE_FILE}") + else: + text = SOURCE_FILE.read_text(encoding="utf-8") + errors.extend(_scan_text(SOURCE_FILE, text)) + evidence[str(SOURCE_FILE)] = { + "sanitized_token_refresh_error": "KIS token refresh failed; check credentials and API availability." in text, + "sanitized_readonly_error": "KIS read-only request failed for" in text, + "token_cache_db": "kis_tokens.db" in text, + } + + if not TEST_FILE.exists(): + errors.append(f"missing:{TEST_FILE}") + else: + text = TEST_FILE.read_text(encoding="utf-8") + evidence[str(TEST_FILE)] = { + "token_cache_tests": "test_issue_or_reuse_token_with_sqlite_db_cache" in text, + "token_override_tests": "test_issue_or_reuse_token_honors_token_db_override" in text, + } + + for path in RUNBOOK_FILES: + if not path.exists(): + errors.append(f"missing:{path}") + continue + text = path.read_text(encoding="utf-8") + evidence[str(path)] = { + "mentions_token_cache": "Temp/kis_tokens.db" in text, + "mentions_refresh_skew": "TOKEN_REFRESH_SKEW_MINUTES" in text, + } + errors.extend(_scan_repository()) + + result = { + "formula_id": "KIS_TOKEN_HYGIENE_V1", + "gate": "PASS" if not errors else "FAIL", + "errors": errors, + "evidence": evidence, + } + out = ROOT / "Temp" / "kis_token_hygiene_v1.json" + out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 if not errors else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/validate_snapshot_admin_web_v1.py b/tools/validate_snapshot_admin_web_v1.py index 3f7c736..3cf0bc8 100644 --- a/tools/validate_snapshot_admin_web_v1.py +++ b/tools/validate_snapshot_admin_web_v1.py @@ -120,7 +120,8 @@ def main() -> int: try: _wait_for_server(base_url) - html = _read_text(f"{base_url}/") + home_html = _read_text(f"{base_url}/") + html = _read_text(f"{base_url}/workspace") state = _read_json(f"{base_url}/api/state") tables_payload = _read_json(f"{base_url}/api/tables") export_payload = _read_json(f"{base_url}/api/export") @@ -139,6 +140,10 @@ def main() -> int: "workspace": state.get("summary", {}), } packet_response = _post_json(f"{base_url}/api/approval_packet", {"packet": approval_packet}) + if "Snapshot Admin Home" not in home_html: + errors.append("home_title_missing") + if "Open workspace" not in home_html or "Open collection" not in home_html: + errors.append("home_navigation_missing") if "Snapshot Admin" not in html: errors.append("html_title_missing") if "contenteditable" not in html: @@ -171,7 +176,14 @@ def main() -> int: if "Read only" not in tables_html or "Save current table" not in tables_html: errors.append("table_browser_source_labels_missing") collection_html = _read_text(f"{base_url}/collection") - if "KIS Collection Dashboard" not in collection_html or "Download CSV" not in collection_html or "Ticker quick search" not in collection_html or "Date quick search" not in collection_html: + if ( + "KIS Collection Dashboard" not in collection_html + or "Download CSV" not in collection_html + or "Ticker quick search" not in collection_html + or "Date quick search" not in collection_html + or "collectionLiveStatus" not in collection_html + or "live source: unknown" not in collection_html + ): errors.append("collection_dashboard_page_missing") if int(state.get("summary", {}).get("settings_rows") or 0) <= 0: errors.append("settings_rows_missing") @@ -200,6 +212,14 @@ def main() -> int: errors.append("collection_counts_missing") if "latest_report" not in collection: errors.append("collection_latest_report_missing") + latest_report = collection.get("latest_report", {}) + if isinstance(latest_report, dict): + if latest_report.get("input_json") and "GatherTradingData.json" not in str(latest_report.get("input_json")): + errors.append("collection_latest_report_input_mismatch") + if not latest_report.get("sqlite_db"): + errors.append("collection_latest_report_sqlite_missing") + if collection.get("output_json_path") and "kis_data_collection_v1.json" not in str(collection.get("output_json_path")): + errors.append("collection_output_json_path_mismatch") if "data" not in export_payload: errors.append("export_missing_data") if packet_response.get("gate") != "PASS": diff --git a/tools/validate_specs.py b/tools/validate_specs.py index 3d27dea..f6f58aa 100644 --- a/tools/validate_specs.py +++ b/tools/validate_specs.py @@ -827,6 +827,7 @@ def main() -> int: validate_json_schema_minimal(schema, sample, errors) validate_formula_registry(errors) + validate_kis_token_hygiene(errors) validate_output_rendering_contract(schema, errors) validate_harness_contract_consistency(errors) validate_spec_code_sync(errors) @@ -902,13 +903,25 @@ def main() -> int: if bundle.exists(): load_yaml(bundle, errors) - if errors: - print("VALIDATION FAIL") - for err in errors: - print(f"- {err}") - return 1 - print("VALIDATION OK") - return 0 + +def validate_kis_token_hygiene(errors: list[str]) -> None: + import importlib.util + + module_path = ROOT / "tools" / "validate_kis_token_hygiene_v1.py" + spec = importlib.util.spec_from_file_location("validate_kis_token_hygiene_v1", module_path) + if spec is None or spec.loader is None: + fail(errors, f"unable to load KIS token hygiene validator: {module_path}") + return + kis_hygiene = importlib.util.module_from_spec(spec) + spec.loader.exec_module(kis_hygiene) + + try: + rc = kis_hygiene.main() + except Exception as exc: # pragma: no cover - validator integration gate + fail(errors, f"KIS token hygiene validator raised: {type(exc).__name__}: {exc}") + return + if rc != 0: + fail(errors, "KIS token hygiene validator failed") if __name__ == "__main__":