diff --git a/spec/61_dotnet_postgresql_json_cutover.yaml b/spec/61_dotnet_postgresql_json_cutover.yaml
index b8dfa881..5b966aa0 100644
--- a/spec/61_dotnet_postgresql_json_cutover.yaml
+++ b/spec/61_dotnet_postgresql_json_cutover.yaml
@@ -8,8 +8,8 @@ canonical_runtime:
migration: src/dotnet/QuantEngine.Infrastructure/Migrations/V5__Add_Normalized_Learning_History.sql
output: Temp/kis_dotnet_collection_v1.json
legacy_policy:
- python_collector: migration_only
- sqlite_store: migration_only
+ python_collector: forbidden
+ sqlite_store: forbidden
xlsx_runtime_input: forbidden
gates:
- dotnet_collector_registered
diff --git a/src/quant_engine/data_collection_store_v1.py b/src/quant_engine/data_collection_store_v1.py
deleted file mode 100644
index 7363d1ac..00000000
--- a/src/quant_engine/data_collection_store_v1.py
+++ /dev/null
@@ -1,464 +0,0 @@
-"""SQLite store for platform-transition data collection outputs.
-
-This store is intentionally small and backend-agnostic enough to be upgraded to
-PostgreSQL later without changing the row contract. The canonical payload is the
-normalized factor row plus provenance metadata.
-"""
-from __future__ import annotations
-
-import json
-import sqlite3
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Iterable
-
-
-SCHEMA = """
-PRAGMA journal_mode=WAL;
-
-CREATE TABLE IF NOT EXISTS collection_runs (
- run_id TEXT PRIMARY KEY,
- collector_name TEXT NOT NULL,
- started_at TEXT NOT NULL,
- finished_at TEXT,
- status TEXT NOT NULL,
- input_source TEXT,
- output_json_path TEXT,
- output_db_path TEXT,
- notes TEXT,
- created_at TEXT DEFAULT (datetime('now'))
-);
-
-CREATE TABLE IF NOT EXISTS collection_snapshots (
- run_id TEXT NOT NULL,
- dataset_name TEXT NOT NULL,
- ticker TEXT NOT NULL,
- name TEXT,
- sector TEXT,
- as_of_date TEXT,
- source_priority TEXT,
- source_status TEXT,
- payload_json TEXT NOT NULL,
- provenance_json TEXT NOT NULL,
- created_at TEXT DEFAULT (datetime('now')),
- PRIMARY KEY (run_id, dataset_name, ticker)
-);
-
-CREATE TABLE IF NOT EXISTS collection_source_errors (
- run_id TEXT NOT NULL,
- ticker TEXT,
- source_name TEXT NOT NULL,
- error_kind TEXT NOT NULL,
- error_message TEXT NOT NULL,
- payload_json TEXT,
- created_at TEXT DEFAULT (datetime('now'))
-);
-
-CREATE INDEX IF NOT EXISTS idx_collection_snapshots_ticker_time
- ON collection_snapshots(ticker, created_at DESC);
-
-CREATE INDEX IF NOT EXISTS idx_collection_source_errors_run
- ON collection_source_errors(run_id, source_name);
-"""
-
-
-@dataclass(frozen=True)
-class CollectionRun:
- run_id: str
- collector_name: str
- started_at: str
- status: str
- input_source: str | None = None
- output_json_path: str | None = None
- output_db_path: str | None = None
- notes: str | None = None
-
-
-# SQLite와 PostgreSQL 연결을 동적으로 감지하여 연결 인스턴스를 리턴하는 헬퍼
-def _get_connection(db_target: Path | str) -> Any:
- db_str = str(db_target)
- if db_str.startswith("postgresql://") or db_str.startswith("postgres://"):
- try:
- import psycopg2
- from psycopg2.extras import RealDictCursor
- conn = psycopg2.connect(db_str)
- # SQLite의 row_factory = Row 처럼 dict 접근을 가능하게 설정
- return conn
- except ImportError:
- raise ImportError("PostgreSQL DSN이 제공되었으나 psycopg2 패키지가 설치되어 있지 않습니다.")
- else:
- return sqlite3.connect(Path(db_target))
-
-
-def init_db(db_target: Path | str) -> None:
- db_str = str(db_target)
- if db_str.startswith("postgresql://") or db_str.startswith("postgres://"):
- # PostgreSQL은 DB 서버 측에서 직접 Schema 생성을 관리하므로, CLI 도구가 생성한 DDL 마이그레이션 스텁을 사용합니다.
- # 런타임 수집 중 자동 DDL 실행은 락 이슈 예방을 위해 스킵하고 트랜잭션 연결만 보장합니다.
- conn = _get_connection(db_target)
- conn.close()
- return
-
- db_path = Path(db_target)
- db_path.parent.mkdir(parents=True, exist_ok=True)
- conn = sqlite3.connect(db_path)
- try:
- conn.executescript(SCHEMA)
- conn.commit()
- finally:
- conn.close()
-
-
-def upsert_collection_run(db_target: Path | str, run: CollectionRun, finished_at: str | None = None) -> None:
- init_db(db_target)
- conn = _get_connection(db_target)
- db_str = str(db_target)
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- try:
- # SQLite와 PostgreSQL 쿼리 바인딩 플레이스홀더 분기 (? vs %s)
- param_char = "%s" if is_pg else "?"
- query = f"""
- INSERT INTO collection_runs (
- run_id, collector_name, started_at, finished_at, status,
- input_source, output_json_path, output_db_path, notes
- ) VALUES ({', '.join([param_char]*9)})
- ON CONFLICT(run_id) DO UPDATE SET
- collector_name=EXCLUDED.collector_name,
- started_at=EXCLUDED.started_at,
- finished_at=EXCLUDED.finished_at,
- status=EXCLUDED.status,
- input_source=EXCLUDED.input_source,
- output_json_path=EXCLUDED.output_json_path,
- output_db_path=EXCLUDED.output_db_path,
- notes=EXCLUDED.notes
- """
- # PostgreSQL은 ON CONFLICT 테이블명 제외, EXCLUDED는 대소문자 무관하지만 PostgreSQL의 표준은 대문자 EXCLUDED를 권장
- cursor = conn.cursor()
- cursor.execute(
- query,
- (
- run.run_id,
- run.collector_name,
- run.started_at,
- finished_at,
- run.status,
- run.input_source,
- run.output_json_path,
- run.output_db_path,
- run.notes,
- ),
- )
- conn.commit()
- finally:
- conn.close()
-
-
-def upsert_collection_snapshot(
- db_target: Path | str,
- *,
- run_id: str,
- dataset_name: str,
- ticker: str,
- name: str | None,
- sector: str | None,
- as_of_date: str | None,
- source_priority: str,
- source_status: str,
- payload: dict[str, Any],
- provenance: dict[str, Any],
-) -> None:
- init_db(db_target)
- conn = _get_connection(db_target)
- db_str = str(db_target)
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- try:
- param_char = "%s" if is_pg else "?"
- query = f"""
- INSERT INTO collection_snapshots (
- run_id, dataset_name, ticker, name, sector, as_of_date,
- source_priority, source_status, payload_json, provenance_json
- ) VALUES ({', '.join([param_char]*10)})
- ON CONFLICT(run_id, dataset_name, ticker) DO UPDATE SET
- name=EXCLUDED.name,
- sector=EXCLUDED.sector,
- as_of_date=EXCLUDED.as_of_date,
- source_priority=EXCLUDED.source_priority,
- source_status=EXCLUDED.source_status,
- payload_json=EXCLUDED.payload_json,
- provenance_json=EXCLUDED.provenance_json
- """
- cursor = conn.cursor()
- cursor.execute(
- query,
- (
- run_id,
- dataset_name,
- ticker,
- name,
- sector,
- as_of_date,
- source_priority,
- source_status,
- json.dumps(payload, ensure_ascii=False, default=str),
- json.dumps(provenance, ensure_ascii=False, default=str),
- ),
- )
- conn.commit()
- finally:
- conn.close()
-
-
-def append_collection_error(
- db_target: Path | str,
- *,
- run_id: str,
- source_name: str,
- error_kind: str,
- error_message: str,
- ticker: str | None = None,
- payload: dict[str, Any] | None = None,
-) -> None:
- init_db(db_target)
- conn = _get_connection(db_target)
- db_str = str(db_target)
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- try:
- param_char = "%s" if is_pg else "?"
- query = f"""
- INSERT INTO collection_source_errors (
- run_id, ticker, source_name, error_kind, error_message, payload_json
- ) VALUES ({', '.join([param_char]*6)})
- """
- cursor = conn.cursor()
- cursor.execute(
- query,
- (
- run_id,
- ticker,
- source_name,
- error_kind,
- error_message,
- json.dumps(payload or {}, ensure_ascii=False, default=str),
- ),
- )
- conn.commit()
- finally:
- conn.close()
-
-
-def fetch_latest_snapshots(db_target: Path | str, ticker: str, dataset_name: str | None = None) -> list[dict[str, Any]]:
- db_str = str(db_target)
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- if not is_pg and not Path(db_target).exists():
- return []
-
- conn = _get_connection(db_target)
- if not is_pg:
- conn.row_factory = sqlite3.Row
- try:
- param_char = "%s" if is_pg else "?"
- cursor = conn.cursor()
- if dataset_name:
- cursor.execute(
- f"""
- SELECT * FROM collection_snapshots
- WHERE ticker = {param_char} AND dataset_name = {param_char}
- ORDER BY created_at DESC
- """,
- (ticker, dataset_name),
- )
- else:
- cursor.execute(
- f"""
- SELECT * FROM collection_snapshots
- WHERE ticker = {param_char}
- ORDER BY created_at DESC
- """,
- (ticker,),
- )
- rows = cursor.fetchall()
- return [dict(row) for row in rows]
- finally:
- conn.close()
-
-
-def iter_recent_snapshots(db_target: Path | str, limit: int = 50) -> Iterable[dict[str, Any]]:
- db_str = str(db_target)
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- if not is_pg and not Path(db_target).exists():
- return []
-
- conn = _get_connection(db_target)
- if not is_pg:
- conn.row_factory = sqlite3.Row
- try:
- param_char = "%s" if is_pg else "?"
- cursor = conn.cursor()
- cursor.execute(
- f"SELECT * FROM collection_snapshots ORDER BY created_at DESC LIMIT {param_char}",
- (limit,),
- )
- rows = cursor.fetchall()
- return [dict(row) for row in rows]
- finally:
- conn.close()
-
-
-def load_collection_runs(db_target: Path | str, limit: int = 20) -> list[dict[str, Any]]:
- db_str = str(db_target)
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- if not is_pg and not Path(db_target).exists():
- return []
-
- conn = _get_connection(db_target)
- if not is_pg:
- conn.row_factory = sqlite3.Row
- try:
- param_char = "%s" if is_pg else "?"
- cursor = conn.cursor()
- cursor.execute(
- f"""
- SELECT run_id, collector_name, started_at, finished_at, status,
- input_source, output_json_path, output_db_path, notes, created_at
- FROM collection_runs
- ORDER BY started_at DESC, created_at DESC
- LIMIT {param_char}
- """,
- (int(limit),),
- )
- rows = cursor.fetchall()
- return [dict(row) for row in rows]
- finally:
- conn.close()
-
-
-def load_collection_errors(db_target: Path | str, limit: int = 20) -> list[dict[str, Any]]:
- db_str = str(db_target)
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- if not is_pg and not Path(db_target).exists():
- return []
-
- conn = _get_connection(db_target)
- if not is_pg:
- conn.row_factory = sqlite3.Row
- try:
- param_char = "%s" if is_pg else "?"
- cursor = conn.cursor()
- cursor.execute(
- f"""
- SELECT run_id, ticker, source_name, error_kind, error_message, payload_json, created_at
- FROM collection_source_errors
- ORDER BY created_at DESC
- LIMIT {param_char}
- """,
- (int(limit),),
- )
- rows = cursor.fetchall()
- return [dict(row) for row in rows]
- finally:
- conn.close()
-
-
-def load_collection_dashboard_state(
- db_target: Path | str | None = None,
- output_json_path: Path | str | None = None,
- *,
- limit: int = 8,
-) -> dict[str, Any]:
- db_str = str(db_target or "")
- is_pg = db_str.startswith("postgresql://") or db_str.startswith("postgres://")
- db = Path(db_target) if db_target and not is_pg else Path()
- report = Path(output_json_path) if output_json_path else Path()
- state: dict[str, Any] = {
- "db_path": db_str,
- "output_json_path": str(report) if output_json_path else "",
- "runs": [],
- "recent_snapshots": [],
- "recent_errors": [],
- "counts": {
- "collection_runs": 0,
- "collection_snapshots": 0,
- "collection_source_errors": 0,
- },
- "latest_run": {},
- "latest_report": {},
- }
- if report.exists():
- try:
- state["latest_report"] = json.loads(report.read_text(encoding="utf-8"))
- except Exception:
- state["latest_report"] = {}
-
- if not is_pg and (not db_target or not db.exists()):
- return state
-
- conn = _get_connection(db_target)
- if not is_pg:
- conn.row_factory = sqlite3.Row
- try:
- cursor = conn.cursor()
- state["counts"] = {
- "collection_runs": cursor.execute("SELECT COUNT(*) FROM collection_runs").fetchone()[0] if not is_pg else cursor.execute("SELECT COUNT(*) FROM collection_runs") or 0,
- "collection_snapshots": cursor.execute("SELECT COUNT(*) FROM collection_snapshots").fetchone()[0] if not is_pg else cursor.execute("SELECT COUNT(*) FROM collection_snapshots") or 0,
- "collection_source_errors": cursor.execute("SELECT COUNT(*) FROM collection_source_errors").fetchone()[0] if not is_pg else cursor.execute("SELECT COUNT(*) FROM collection_source_errors") or 0,
- }
- # PostgreSQL인 경우 단순 fetchone() 보완
- if is_pg:
- # PostgreSQL count 처리
- cursor.execute("SELECT COUNT(*) FROM collection_runs")
- state["counts"]["collection_runs"] = cursor.fetchone()[0]
- cursor.execute("SELECT COUNT(*) FROM collection_snapshots")
- state["counts"]["collection_snapshots"] = cursor.fetchone()[0]
- cursor.execute("SELECT COUNT(*) FROM collection_source_errors")
- state["counts"]["collection_source_errors"] = cursor.fetchone()[0]
-
- cursor.execute(
- """
- SELECT run_id, collector_name, started_at, finished_at, status,
- input_source, output_json_path, output_db_path, notes, created_at
- FROM collection_runs
- ORDER BY started_at DESC, created_at DESC
- LIMIT 1
- """
- )
- run_row = cursor.fetchone()
- state["latest_run"] = dict(run_row) if run_row is not None else {}
-
- param_char = "%s" if is_pg else "?"
- cursor.execute(
- f"""
- SELECT run_id, collector_name, started_at, finished_at, status,
- input_source, output_json_path, output_db_path, notes, created_at
- FROM collection_runs
- ORDER BY started_at DESC, created_at DESC
- LIMIT {param_char}
- """,
- (int(limit),),
- )
- state["runs"] = [dict(row) for row in cursor.fetchall()]
-
- cursor.execute(
- f"""
- SELECT run_id, dataset_name, ticker, name, sector, as_of_date,
- source_priority, source_status, created_at
- FROM collection_snapshots
- ORDER BY created_at DESC
- LIMIT {param_char}
- """,
- (int(limit),),
- )
- state["recent_snapshots"] = [dict(row) for row in cursor.fetchall()]
-
- cursor.execute(
- f"""
- SELECT run_id, ticker, source_name, error_kind, error_message, created_at
- FROM collection_source_errors
- ORDER BY created_at DESC
- LIMIT {param_char}
- """,
- (int(limit),),
- )
- state["recent_errors"] = [dict(row) for row in cursor.fetchall()]
- finally:
- conn.close()
- return state
diff --git a/src/quant_engine/kis_data_collection.db b/src/quant_engine/kis_data_collection.db
deleted file mode 100644
index add2cae4..00000000
Binary files a/src/quant_engine/kis_data_collection.db and /dev/null differ
diff --git a/src/quant_engine/snapshot_admin.db b/src/quant_engine/snapshot_admin.db
deleted file mode 100644
index 45e68d12..00000000
Binary files a/src/quant_engine/snapshot_admin.db and /dev/null differ
diff --git a/src/quant_engine/snapshot_admin_server_v1.py b/src/quant_engine/snapshot_admin_server_v1.py
deleted file mode 100644
index d2b397db..00000000
--- a/src/quant_engine/snapshot_admin_server_v1.py
+++ /dev/null
@@ -1,5072 +0,0 @@
-from __future__ import annotations
-
-import argparse
-import datetime as dt
-import json
-import sqlite3
-import subprocess
-import time
-import sys
-from http import HTTPStatus
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
-from hashlib import sha256
-from typing import Any
-from urllib.parse import urlparse, parse_qs
-
-ROOT = Path(__file__).resolve().parents[2]
-SNAPSHOT_ADMIN_VERSION = "snapshot-admin-web-v6"
-KIS_COLLECTION_DB = ROOT / "src" / "quant_engine" / "kis_data_collection.db"
-KIS_COLLECTION_REPORT = ROOT / "Temp" / "kis_data_collection_v1.json"
-WORKBOOK_JSON = ROOT / "GatherTradingData.json"
-WORKBOOK_XLSX = ROOT / "GatherTradingData.xlsx"
-QUALITATIVE_SELL_DB = ROOT / "outputs" / "qualitative_sell_strategy" / "qualitative_sell_strategy.db"
-
-# WBS-7.9 부속 — 테이블별 그리드 조회(Tabler). 화이트리스트에 없는 테이블명은
-# SQL에 절대 보간되지 않는다(요청 테이블명을 그대로 SELECT 문에 넣지 않고
-# 아래 레지스트리 키와 정확히 일치할 때만 허용).
-WORKSPACE_BROWSABLE_TABLES = (
- "settings",
- "account_snapshot",
- "workspace_change_log",
- "workspace_approval_v2",
- "workspace_lock",
- "workspace_meta",
-)
-COLLECTION_BROWSABLE_TABLES = (
- "collection_runs",
- "collection_snapshots",
- "collection_source_errors",
-)
-QUALITATIVE_SELL_BROWSABLE_TABLES = (
- "sell_strategy_results",
- "satellite_recommendations",
-)
-
-# Editable tables configurations (WBS requirement 2)
-EDITABLE_TABLES = {
- "settings",
- "account_snapshot",
-}
-
-
-def _resolve_table_db(table: str, workspace_db_path: Path) -> Path | None:
- if table in WORKSPACE_BROWSABLE_TABLES:
- return Path(workspace_db_path)
- if table in COLLECTION_BROWSABLE_TABLES:
- return KIS_COLLECTION_DB
- if table in QUALITATIVE_SELL_BROWSABLE_TABLES:
- return QUALITATIVE_SELL_DB
- return None
-
-
-def list_browsable_tables(workspace_db_path: Path) -> list[dict[str, Any]]:
- tables: list[dict[str, Any]] = []
- for table in (
- *WORKSPACE_BROWSABLE_TABLES,
- *COLLECTION_BROWSABLE_TABLES,
- *QUALITATIVE_SELL_BROWSABLE_TABLES,
- ):
- db_path = _resolve_table_db(table, workspace_db_path)
- exists = bool(db_path and db_path.exists())
- row_count = 0
- if exists:
- try:
- with sqlite3.connect(db_path) as conn:
- row_count = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] # noqa: S608 - table is whitelist-checked above
- except sqlite3.OperationalError:
- exists = False
- tables.append({
- "table": table,
- "db": str(db_path) if db_path else "",
- "exists": exists,
- "row_count": row_count,
- "editable": table in EDITABLE_TABLES,
- })
- tables.sort(key=lambda item: (
- 0 if item["table"] == "account_snapshot" else 1 if item["table"] == "settings" else 2,
- 0 if item["row_count"] else 1,
- item["table"],
- ))
- return tables
-
-
-def fetch_table_rows(
- table: str,
- workspace_db_path: Path,
- *,
- limit: int = 50,
- offset: int = 0,
- filter_text: str = "",
- column_filters: dict[str, str] | None = None,
-) -> dict[str, Any]:
- db_path = _resolve_table_db(table, workspace_db_path)
- if db_path is None:
- raise ValueError(f"unknown or non-browsable table: {table}")
- if not db_path.exists():
- return {"table": table, "db": str(db_path), "columns": [], "rows": [], "total": 0, "limit": limit, "offset": offset, "editable": table in EDITABLE_TABLES}
- with sqlite3.connect(db_path) as conn:
- conn.row_factory = sqlite3.Row
- cursor = conn.execute(f"SELECT rowid as _rowid, * FROM {table} ORDER BY rowid DESC",) # noqa: S608 - whitelisted table name
- all_rows = [dict(row) for row in cursor.fetchall()]
- columns = [description[0] for description in cursor.description] if cursor.description else []
- cleaned_filter_text = str(filter_text or "").strip().lower()
- normalized_column_filters = {str(key): str(value).strip().lower() for key, value in (column_filters or {}).items() if str(value).strip()}
-
- def _match_row(row: dict[str, Any]) -> bool:
- display_row = {k: v for k, v in row.items() if not str(k).startswith("_")}
- haystack = json.dumps(display_row, ensure_ascii=False, default=str).lower()
- if cleaned_filter_text and cleaned_filter_text not in haystack:
- return False
- for key, needle in normalized_column_filters.items():
- cell = str(display_row.get(key, "") or "").lower()
- if needle not in cell:
- return False
- return True
-
- filtered_rows = [row for row in all_rows if _match_row(row)]
- total = len(filtered_rows)
- rows = filtered_rows[offset: offset + limit]
- return {"table": table, "db": str(db_path), "columns": columns, "rows": rows, "total": total, "limit": limit, "offset": offset, "editable": table in EDITABLE_TABLES, "filter_text": cleaned_filter_text, "column_filters": normalized_column_filters}
-
-
-def fetch_domain_rows(domain: str, workspace_db_path: Path) -> dict[str, Any]:
- if domain == "settings":
- rows = load_settings_rows(workspace_db_path)
- return {"domain": domain, "db": str(workspace_db_path), "columns": ["ordinal", "key", "value", "note", "updated_at"], "rows": rows}
- if domain == "account_snapshot":
- rows = load_account_snapshot_rows(workspace_db_path)
- return {
- "domain": domain,
- "db": str(workspace_db_path),
- "columns": list(ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS),
- "rows": rows,
- }
- raise ValueError(f"unknown editable domain: {domain}")
-SNAPSHOT_ADMIN_VERSION_FILES = (
- ROOT / "src" / "quant_engine" / "snapshot_admin_server_v1.py",
- ROOT / "src" / "quant_engine" / "snapshot_admin_store_v1.py",
- ROOT / "src" / "quant_engine" / "data_collection_store_v1.py",
- ROOT / "tools" / "run_snapshot_admin_server_v1.py",
- ROOT / "tools" / "validate_snapshot_admin_web_v1.py",
- ROOT / "tests" / "unit" / "test_snapshot_admin_web_v1.py",
- ROOT / "package.json",
-)
-
-from .snapshot_admin_store_v1 import (
- ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS,
- DEFAULT_DB,
- DEFAULT_SEED_JSON,
- export_payload,
- clear_lock,
- import_seed_json,
- is_locked,
- load_account_snapshot_rows,
- load_approval_for_domain,
- load_approval_rows,
- load_change_log_rows,
- load_locks,
- load_settings_rows,
- normalize_db_path,
- now_kst_iso,
- open_connection,
- parse_account_snapshot_tsv,
- parse_scalar,
- record_change_log,
- validate_account_snapshot_rows,
- validate_settings_rows,
- build_validation_suggestions,
- build_safe_autofix_actions,
- apply_safe_autofix_action,
- lock_conflicts_for_rows,
- set_approval,
- set_lock,
- replace_account_snapshot,
- replace_settings,
- undo_last_change,
- summarize_workspace,
-)
-from .data_collection_store_v1 import load_collection_dashboard_state
-
-
-def _strip_internal_fields(row: dict[str, Any]) -> dict[str, Any]:
- return {key: value for key, value in row.items() if not key.startswith("_")}
-
-
-def _snapshot_columns_from_rows(rows: list[dict[str, Any]]) -> list[str]:
- columns = list(ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS)
- extras = sorted(
- {
- key
- for row in rows
- for key in row.keys()
- if not key.startswith("_") and key not in ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS
- }
- )
- for key in extras:
- if key not in columns:
- columns.append(key)
- return columns
-
-
-def _write_json(path: Path, payload: dict[str, Any]) -> Path:
- path.parent.mkdir(parents=True, exist_ok=True)
- path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
- return path
-
-
-def _render_approval_packet_md(packet: dict[str, Any]) -> str:
- pending = packet.get("pending_targets") if isinstance(packet.get("pending_targets"), list) else []
- summary = packet.get("summary") if isinstance(packet.get("summary"), dict) else {}
- lines = [
- "# Snapshot Admin Approval Packet",
- "",
- "## Summary",
- "",
- f"- settings_changed: {summary.get('settings_changed', 0)}",
- f"- account_snapshot_changed: {summary.get('account_snapshot_changed', 0)}",
- f"- pending_target_count: {summary.get('pending_target_count', 0)}",
- "",
- "## Pending Targets",
- "",
- ]
- if pending:
- for item in pending[:100]:
- if not isinstance(item, dict):
- continue
- lines.append(f"- {item.get('domain', '')}:{item.get('target_ref', '')} ({item.get('change_type', '')})")
- else:
- lines.append("_none_")
- return "\n".join(lines)
-
-
-def write_approval_packet_artifacts(packet: dict[str, Any]) -> dict[str, str]:
- json_path = ROOT / "Temp" / "snapshot_admin_approval_packet_v1.json"
- md_path = ROOT / "Temp" / "snapshot_admin_approval_packet_v1.md"
- _write_json(json_path, packet)
- md_path.parent.mkdir(parents=True, exist_ok=True)
- md_path.write_text(_render_approval_packet_md(packet), encoding="utf-8")
- return {"json_path": str(json_path), "md_path": str(md_path)}
-
-
-def _git_info() -> dict[str, Any]:
- try:
- commit = subprocess.check_output(
- ["git", "rev-parse", "--short", "HEAD"],
- cwd=str(ROOT),
- text=True,
- stderr=subprocess.DEVNULL,
- ).strip()
- status = subprocess.check_output(
- ["git", "status", "--porcelain"],
- cwd=str(ROOT),
- text=True,
- stderr=subprocess.DEVNULL,
- )
- return {
- "commit": commit,
- "dirty": bool(status.strip()),
- "tree_state": "DIRTY" if status.strip() else "CLEAN",
- }
- except Exception:
- return {
- "commit": "",
- "dirty": False,
- "tree_state": "UNKNOWN",
- }
-
-
-def _source_fingerprint() -> dict[str, Any]:
- digest = sha256()
- latest_mtime = 0.0
- for path in SNAPSHOT_ADMIN_VERSION_FILES:
- if not path.exists():
- continue
- try:
- data = path.read_bytes()
- digest.update(path.as_posix().encode("utf-8"))
- digest.update(b"\0")
- digest.update(data)
- latest_mtime = max(latest_mtime, path.stat().st_mtime)
- except OSError:
- continue
- latest_updated_at = ""
- if latest_mtime:
- latest_updated_at = dt.datetime.fromtimestamp(latest_mtime, tz=dt.timezone.utc).astimezone(
- dt.timezone(dt.timedelta(hours=9))
- ).isoformat()
- return {
- "fingerprint": digest.hexdigest()[:16],
- "latest_mtime": latest_mtime,
- "latest_updated_at": latest_updated_at,
- }
-
-
-def _approval_entry_from_conn(conn, domain: str, target_ref: str = "*") -> dict[str, Any] | None:
- ensure_schema(conn)
- row = conn.execute(
- f"""
- SELECT domain, target_ref, status, approved_by, approved_at, note, updated_at
- FROM {APPROVAL_TABLE}
- WHERE domain = ? AND target_ref = ?
- LIMIT 1
- """,
- (domain, target_ref or "*"),
- ).fetchone()
- return dict(row) if row is not None else None
-
-
-def _lock_entry_from_conn(conn, domain: str, target_ref: str = "*") -> dict[str, Any] | None:
- ensure_schema(conn)
- row = conn.execute(
- f"""
- SELECT domain, target_ref, locked_by, reason, locked_at
- FROM {LOCK_TABLE}
- WHERE domain = ? AND target_ref = ?
- LIMIT 1
- """,
- (domain, target_ref or "*"),
- ).fetchone()
- return dict(row) if row is not None else None
-
-
-def build_ui_state(db_path: Path | str | None = None) -> dict[str, Any]:
- summary = summarize_workspace(db_path)
- settings_rows = load_settings_rows(db_path)
- account_rows = [_strip_internal_fields(row) for row in load_account_snapshot_rows(db_path)]
- settings_errors = validate_settings_rows(settings_rows)
- snapshot_errors = validate_account_snapshot_rows(account_rows)
- suggestions = build_validation_suggestions(settings_rows, account_rows)
- autofix_actions = build_safe_autofix_actions(settings_rows, account_rows)
- try:
- collection = load_collection_dashboard_state(KIS_COLLECTION_DB, KIS_COLLECTION_REPORT)
- except Exception:
- collection = {}
- workbook_registry = load_workbook_sheet_registry()
- return {
- "version": {
- "app": SNAPSHOT_ADMIN_VERSION,
- "git": _git_info(),
- "source": _source_fingerprint(),
- },
- "summary": summary,
- "approval_rows": load_approval_rows(db_path),
- "approval_settings": load_approval_for_domain(db_path, "settings"),
- "approval_account_snapshot": load_approval_for_domain(db_path, "account_snapshot"),
- "locks": load_locks(db_path),
- "recent_changes": load_change_log_rows(db_path, limit=12),
- "history_counts": {
- "changes": len(load_change_log_rows(db_path, limit=200)),
- "approvals": len(load_approval_rows(db_path)),
- "locks": len(load_locks(db_path)),
- },
- "settings_rows": settings_rows,
- "account_snapshot_rows": account_rows,
- "account_snapshot_columns": _snapshot_columns_from_rows(account_rows),
- "validation": {
- "settings": settings_errors,
- "account_snapshot": snapshot_errors,
- "suggestions": suggestions,
- },
- "autofix_actions": autofix_actions,
- "collection": collection,
- "workbook_registry": workbook_registry,
- "generated_at": now_kst_iso(),
- }
-
-
-def load_workbook_sheet_registry() -> dict[str, Any]:
- try:
- payload = json.loads(WORKBOOK_JSON.read_text(encoding="utf-8"))
- except Exception:
- payload = {}
- metadata = payload.get("metadata") if isinstance(payload, dict) else {}
- sheets = metadata.get("sheets_included") if isinstance(metadata, dict) else []
- sheet_headers = metadata.get("sheet_headers") if isinstance(metadata, dict) else {}
- data = payload.get("data") if isinstance(payload, dict) else {}
- if not isinstance(sheets, list):
- sheets = []
- if not isinstance(data, dict):
- data = {}
- entries: list[dict[str, Any]] = []
- for sheet in sheets:
- sheet_name = str(sheet)
- source_role = "derived_report_evidence"
- if sheet_name in {"settings", "account_snapshot"}:
- destination = "snapshot_admin.db"
- kind = "workspace_db"
- purpose = "workspace_edit"
- source_role = "canonical_db"
- elif sheet_name == "data_feed":
- destination = "kis_data_collection.db"
- kind = "collector_db"
- purpose = "collector_run"
- source_role = "collector_db"
- elif sheet_name 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"}:
- destination = "GatherTradingData.json"
- if sheet_name in {"sector_universe_refresh_audit"}:
- purpose = "refresh_audit"
- elif sheet_name in {"daily_history"}:
- purpose = "history_ledger"
- elif sheet_name in {"event_calendar", "pa1_feedback"}:
- purpose = "audit_history"
- elif sheet_name in {"alpha_history", "backdata_feature_bank", "sell_priority"}:
- purpose = "analysis_report"
- elif sheet_name in {"harness_context"}:
- purpose = "execution_context"
- elif sheet_name in {"monthly_history", "sector_flow_history"}:
- purpose = "history_ledger"
- elif sheet_name in {"sector_universe", "core_satellite", "universe"}:
- purpose = "universe_registry"
- elif sheet_name in {"event_risk", "macro"}:
- purpose = "macro_risk_context"
- elif sheet_name in {"sector_flow"}:
- purpose = "flow_leadership"
- else:
- purpose = "json_payload"
- kind = "json_payload"
- else:
- destination = "unknown"
- kind = "unmapped"
- purpose = "unmapped"
- header_meta = sheet_headers.get(sheet_name) if isinstance(sheet_headers, dict) else {}
- entries.append(
- {
- "sheet": sheet_name,
- "destination": destination,
- "kind": kind,
- "purpose": purpose,
- "source_role": source_role,
- "has_json_payload": sheet_name in data,
- "row_count": _sheet_payload_row_count(data.get(sheet_name), header_meta),
- }
- )
- explicit_target_sheets = [
- "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",
- "data_feed",
- ]
- existing_sheets = {item["sheet"] for item in entries}
- for sheet_name in explicit_target_sheets:
- if sheet_name in existing_sheets:
- continue
- if sheet_name == "data_feed":
- destination = "kis_data_collection.db"
- kind = "collector_db"
- purpose = "collector_run"
- source_role = "collector_db"
- else:
- destination = "GatherTradingData.json"
- kind = "json_payload"
- source_role = "derived_report_evidence"
- if sheet_name == "sector_universe_refresh_audit":
- purpose = "refresh_audit"
- elif sheet_name == "daily_history":
- purpose = "history_ledger"
- elif sheet_name in {"event_calendar", "pa1_feedback"}:
- purpose = "audit_history"
- elif sheet_name in {"alpha_history", "backdata_feature_bank", "sell_priority"}:
- purpose = "analysis_report"
- elif sheet_name == "harness_context":
- purpose = "execution_context"
- elif sheet_name in {"monthly_history", "sector_flow_history"}:
- purpose = "history_ledger"
- elif sheet_name in {"sector_universe", "core_satellite", "universe"}:
- purpose = "universe_registry"
- elif sheet_name in {"event_risk", "macro"}:
- purpose = "macro_risk_context"
- elif sheet_name == "sector_flow":
- purpose = "flow_leadership"
- else:
- purpose = "json_payload"
- entries.append(
- {
- "sheet": sheet_name,
- "destination": destination,
- "kind": kind,
- "purpose": purpose,
- "source_role": source_role,
- "has_json_payload": kind == "json_payload",
- "row_count": 0,
- }
- )
- return {
- "xlsx_path": str(WORKBOOK_XLSX),
- "json_path": str(WORKBOOK_JSON),
- "json_role": "derived_report_evidence",
- "sheet_count": len(entries),
- "entries": entries,
- "unmapped_sheets": [item["sheet"] for item in entries if item["kind"] == "unmapped"],
- }
-
-
-def _sheet_payload_row_count(value: Any, header_meta: Any | None = None) -> int | None:
- if isinstance(header_meta, dict) and isinstance(header_meta.get("row_count"), int):
- return int(header_meta["row_count"])
- if isinstance(value, list):
- return len(value)
- if isinstance(value, dict):
- return len(value)
- if value is None:
- return 0
- return None
-
-
-def run_collection_job(
- *,
- sqlite_db: Path | None = None,
- input_json: Path | None = None,
- output_json: Path | None = None,
- allow_naver_fallback: bool = False,
- include_live_kis: bool = False,
- kis_account: str = "real",
-) -> dict[str, Any]:
- sqlite_db = sqlite_db or KIS_COLLECTION_DB
- input_json = input_json or (ROOT / "GatherTradingData.json")
- output_json = output_json or KIS_COLLECTION_REPORT
- cmd = [
- sys.executable,
- str(ROOT / "tools" / "run_kis_data_collection_v1.py"),
- "--input-json",
- str(input_json),
- "--sqlite-db",
- str(sqlite_db),
- "--output-json",
- str(output_json),
- "--kis-account",
- kis_account,
- ]
- if allow_naver_fallback:
- cmd.append("--allow-naver-fallback")
- if not include_live_kis:
- cmd.append("--no-live-kis")
-
- started_at = now_kst_iso()
- started = time.perf_counter()
- proc = subprocess.run(
- cmd,
- cwd=str(ROOT),
- capture_output=True,
- text=True,
- encoding="utf-8",
- )
- finished_at = now_kst_iso()
- elapsed_ms = round((time.perf_counter() - started) * 1000.0, 1)
- summary = {}
- if output_json.exists():
- try:
- loaded = json.loads(output_json.read_text(encoding="utf-8"))
- summary = loaded if isinstance(loaded, dict) else {}
- except Exception:
- summary = {}
- state = {}
- try:
- state = load_collection_dashboard_state(sqlite_db, output_json)
- except Exception:
- state = {}
- return {
- "status": "PASS" if proc.returncode == 0 else "FAIL",
- "started_at": started_at,
- "finished_at": finished_at,
- "elapsed_ms": elapsed_ms,
- "command": cmd,
- "returncode": proc.returncode,
- "stdout": proc.stdout,
- "stderr": proc.stderr,
- "summary": summary,
- "state": state,
- }
-
-
-def _json_response(handler: BaseHTTPRequestHandler, status: int, payload: Any) -> None:
- body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
- handler.send_response(status)
- handler.send_header("Content-Type", "application/json; charset=utf-8")
- handler.send_header("Content-Length", str(len(body)))
- handler.end_headers()
- handler.wfile.write(body)
-
-
-def _text_response(handler: BaseHTTPRequestHandler, status: int, text: str, content_type: str = "text/plain; charset=utf-8") -> None:
- body = text.encode("utf-8")
- handler.send_response(status)
- handler.send_header("Content-Type", content_type)
- handler.send_header("Content-Length", str(len(body)))
- handler.end_headers()
- handler.wfile.write(body)
-
-
-def _json_response(handler: BaseHTTPRequestHandler, status: int, payload: Any) -> None:
- body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
- handler.send_response(status)
- handler.send_header("Content-Type", "application/json; charset=utf-8")
- handler.send_header("Content-Length", str(len(body)))
- handler.end_headers()
- handler.wfile.write(body)
-
-
-def _text_response(handler: BaseHTTPRequestHandler, status: int, text: str, content_type: str = "text/plain; charset=utf-8") -> None:
- body = text.encode("utf-8")
- handler.send_response(status)
- handler.send_header("Content-Type", content_type)
- handler.send_header("Content-Length", str(len(body)))
- handler.end_headers()
- handler.wfile.write(body)
-
-
-def _read_json_body(handler: BaseHTTPRequestHandler) -> dict[str, Any]:
- length = int(handler.headers.get("Content-Length") or "0")
- raw = handler.rfile.read(length).decode("utf-8") if length else "{}"
- payload = json.loads(raw or "{}")
- if not isinstance(payload, dict):
- raise ValueError("JSON body must be an object")
- return payload
-
-
-def render_home_html() -> str:
- return """
-
-
-
-
- Snapshot Admin Home
-
-
-
-
-
- Snapshot Admin Home
- 이 화면은 운영 판단용입니다. 편집, 승인, 수집이 모두 한 곳에 섞여 있으면 목적이 흐려지므로, 먼저 해야 할 일만 보여줍니다.
-
-
-
1. Workspace
-
데이터를 고치고 저장
-
settings와 account_snapshot 편집, TSV 적용, 검증 확인은 workspace에서 처리합니다.
-
-
-
2. Collection
-
수집 실행과 결과 확인
-
수집 버튼, 진행상태, 결과 로그는 collection 화면에 둡니다.
-
-
-
3. Tables
-
DB와 JSON 증빙을 분리 검토
-
DB별 조회, JSON별 조회, 시계열 이력은 tables에서 확인합니다.
-
-
-
-
-
-
-
-
-"""
-
-
-def render_inspector_html() -> str:
- return """
-
-
-
-
- Snapshot Inspector
-
-
-
-
-
- Snapshot Inspector
- Row-level operations move here so the workspace stays focused on editing.
-
- Select a domain and target_ref, then load.
-
-
-
-
-
-
-"""
-
-
-def render_index_html() -> str:
- return """
-
-
-
-
- Snapshot Admin
-
-
-
-
-
-
-
-
-
Workspace
-
- Reload
- Seed from GatherTradingData.json
- Download JSON
- Undo settings
- Undo snapshot
- Apply safe autofix
-
-
- Loading...
-
-
-
-
Validation
-
-
Suggestions
-
-
-
-
-
-
-
-
-
- Approval & Locks
-
- open when you need to approve or lock rows
-
-
-
-
- Lock settings
- Unlock settings
- Lock snapshot
- Unlock snapshot
- Approve settings
- Approve snapshot
-
-
-
-
settings approval
-
snapshot approval
-
-
-
-
-
-
- Lock target
- Unlock target
- Approve pending
- Refresh diff
- Export approval packet
-
-
-
-
Recent change log
-
-
- Clear filter
-
-
-
Timeline
-
-
-
-
-
-
-
-
-
- KIS Collection
-
- open when you want to run or inspect collector output
-
-
-
-
- Refresh collection
- Copy status
-
-
-
-
collection: loading...
-
-
-
Collection trend
-
-
-
-
- Clear filter
-
-
-
-
Recent collector runs
-
-
Recent collector snapshots
-
-
Recent collector errors
-
-
Collection detail
-
-
-
-
-
-
-
-
- Selection Inspector
-
- open when you need row-level operations
-
-
-
-
- Focus row
- Copy target
- Approve selected
- Lock selected
- Unlock selected
-
-
-
-
No row selected.
-
-
Recent row history
-
-
-
-
Batch paste
-
-
- settings
- account_snapshot
-
- Apply TSV to selection
-
-
-
Tip: clipboard paste still works directly in the grid. This panel is for multi-row batch edit against the selected row.
-
Shortcuts: `Ctrl+S` save current domain, `Ctrl+Enter` save current domain, `Delete` remove selected row.
-
-
-
-
-
-
-
-
Settings 0 rows
-
- Add row
- Save settings
-
-
-
-
-
-
- asc
- desc
-
- Save view
-
- Delete view
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
Account Snapshot 0 rows
-
- Add row
- Save snapshot
-
-
-
- Paste TSV below and replace all rows
- Canonical column order follows spec/15_account_snapshot_contract.yaml
-
-
-
-
Account snapshot editing surface
-
This panel is intentionally separated from settings so row selection, field edits, and save approval stay visually dominant.
-
-
-
-
- asc
- desc
-
- Save view
-
- Delete view
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- Import TSV (replace)
- Insert blank template
-
-
-
-
-
-
-
-
-
-
-"""
-
-
-def render_collection_html() -> str:
- return """
-
-
-
-
- KIS Collection Dashboard
-
-
-
-
- KIS Collection Dashboard
- Separate read-only view for KIS collection run, snapshots, errors, and raw JSON evidence.
-
-
-
-
-
-
Execution / Status / Result
-
collection: loading...
-
run: idle
-
progress: idle
-
stage: idle
-
result: pending
-
-
-
-
-
-
-
-
-
-
Real mode is selected by default. Offline replay is available only when explicitly enabled.
-
live source: unknown
-
Auto refreshes every 15 seconds while this page is open.
-
-
-
-
-
- Clear filter
-
-
Recent collector runs
-
-
Recent collector snapshots
-
-
Recent collector errors
-
-
Unified activity timeline
-
-
-
-
Collection detail
-
-
Run log
-
-
-
-
-
-
-
-
-
-"""
-
-
-def render_tables_html() -> str:
- return """
-
-
-
-
- Snapshot Admin — Table Browser
-
-
-
-
-
-
-
Snapshot Admin — Table Browser
-
-
-
-
-
-
-
-
-
-
-
DB 먼저 / JSON은 증빙
-
DB별 수정, JSON별 검토, 수집 증빙 확인
-
Workspace rows are editable only in the canonical DB. Collection and strategy views are read-only proof surfaces.
-
-
-
-
- 1. Workspace
- 2. Collection
- 3. Strategy
- 4. JSON evidence
-
-
- DB 먼저
- JSON은 증빙
- Edit only canonical rows
-
-
-
-
-
DB tables
-
Editable source of truth. Only canonical workspace rows are mutated here.
-
-
-
-
-
Workbook sheets
-
Derived report evidence. These sheets summarize DB-backed outputs and run history.
-
-
-
-
version: loading...
-
- Checking whether the table combo covers the target workbook sheets...
-
-
- Workbook sheets only
-
- sheet: none
- surface: none
-
-
-
-
-
-
Save applies only to the canonical workspace DB. This selector is for DB tables only; workbook sheets are handled above.
-
-
- Workspace
- Collection
- Strategy
- JSON
-
-
- Registry details
-
-
-
-
- Open only when you need the table/sheet mapping.
-
-
-
-
-
-
-
-
- Table
- DB
- Rows
- Edit
-
-
-
-
-
-
-
-
-
-
-
-
- DB가 원천이고 JSON은 DB 기반 파생 보고서 증빙이다. 이 화면은 그 관계만 보여준다.
-
-
-
-
-
-
- Sheet
- Destination
- Kind
- Source Role
- Purpose
- Rows
- Recorded surface
-
-
-
-
-
-
-
-
-
-
- History details
- Recent workspace change log and collector run history.
-
-
-
-
-
- When Domain Action Target
-
-
-
-
-
-
-
-
-
- Run Status Started Finished
-
-
-
-
-
-
-
-
- Collection purpose: monitor collector runs and snapshots.
- Latest run summary loads from `/api/state`.
-
-
- DB별 / JSON별 조회 기준을 먼저 확인한 뒤 수정하세요.
- This page is intentionally a triage surface, not a generic table dump.
- Workspace tables are editable only when the table is in the canonical workspace DB.
- Collection and strategy tables are read-only by design unless the backing store explicitly supports editing.
-
-
-
- Workspace and DB view
- Purpose: edit canonical workspace rows only.
-
-
- Table browser ready
- 0 rows
- filter=none
- page=0
-
-
If a table shows "no rows", the current filter or selected table has no visible records.
-
-
-
-
- Derived JSON evidence view
- Purpose: inspect DB-backed JSON evidence and row payloads.
-
-
-
-
-
-
-
Select a workbook sheet to inspect its registry detail.
-
No workbook sheet selected.
-
-
Click a row to inspect the derived JSON payload.
-
No JSON row selected.
-
version: loading...
-
-
-
-
-
-
-
-
-
-
-
-"""
-
-
-class SnapshotAdminHandler(BaseHTTPRequestHandler):
- db_path: Path = DEFAULT_DB
- seed_json_path: Path = DEFAULT_SEED_JSON
-
- def log_message(self, format: str, *args: Any) -> None: # noqa: A003
- return
-
- def _handle_exception(self, exc: Exception) -> None:
- _json_response(self, HTTPStatus.INTERNAL_SERVER_ERROR, {"detail": str(exc)})
-
- def do_GET(self) -> None: # noqa: N802
- parsed = urlparse(self.path)
- if parsed.path == "/":
- _text_response(self, HTTPStatus.OK, render_home_html(), "text/html; charset=utf-8")
- return
- if parsed.path == "/workspace":
- _text_response(self, HTTPStatus.OK, render_index_html(), "text/html; charset=utf-8")
- return
- if parsed.path == "/collection":
- _text_response(self, HTTPStatus.OK, render_collection_html(), "text/html; charset=utf-8")
- return
- if parsed.path == "/tables":
- _text_response(self, HTTPStatus.OK, render_tables_html(), "text/html; charset=utf-8")
- return
- if parsed.path == "/api/tables":
- _json_response(self, HTTPStatus.OK, {"tables": list_browsable_tables(self.db_path)})
- return
- if parsed.path == "/api/table_rows":
- query = parse_qs(parsed.query)
- table = (query.get("table") or [""])[0]
- try:
- limit = int((query.get("limit") or ["50"])[0])
- offset = int((query.get("offset") or ["0"])[0])
- except ValueError:
- _json_response(self, HTTPStatus.BAD_REQUEST, {"detail": "limit/offset must be integers"})
- return
- limit = min(max(limit, 1), 500)
- offset = max(offset, 0)
- filter_text = (query.get("filter") or [""])[0]
- column_filters: dict[str, str] = {}
- for key, values in query.items():
- if key.startswith("filter_") and values:
- column_filters[key.removeprefix("filter_")] = values[0]
- try:
- payload = fetch_table_rows(
- table,
- self.db_path,
- limit=limit,
- offset=offset,
- filter_text=filter_text,
- column_filters=column_filters,
- )
- except ValueError as exc:
- _json_response(self, HTTPStatus.BAD_REQUEST, {"detail": str(exc)})
- return
- _json_response(self, HTTPStatus.OK, payload)
- return
- if parsed.path == "/api/state":
- _json_response(self, HTTPStatus.OK, build_ui_state(self.db_path))
- return
- if parsed.path == "/api/collection/run":
- _json_response(self, HTTPStatus.METHOD_NOT_ALLOWED, {"detail": "POST required"})
- return
- if parsed.path == "/api/history":
- _json_response(
- self,
- HTTPStatus.OK,
- {
- "settings": load_change_log_rows(self.db_path, limit=25),
- "approvals": load_approval_rows(self.db_path),
- "locks": load_locks(self.db_path),
- },
- )
- return
- if parsed.path == "/api/export":
- _text_response(
- self,
- HTTPStatus.OK,
- json.dumps(export_payload(self.db_path), ensure_ascii=False, indent=2),
- "application/json; charset=utf-8",
- )
- return
- if parsed.path == "/favicon.ico":
- _text_response(self, HTTPStatus.NO_CONTENT, "")
- return
- _json_response(self, HTTPStatus.NOT_FOUND, {"detail": "not found"})
-
- def do_POST(self) -> None: # noqa: N802
- parsed = urlparse(self.path)
- try:
- if parsed.path == "/api/bootstrap":
- summary = import_seed_json(self.db_path, self.seed_json_path)
- _json_response(self, HTTPStatus.OK, summary)
- return
- if parsed.path == "/api/collection/run":
- content_type = str(self.headers.get("Content-Type") or "").lower()
- if "application/json" in content_type:
- payload = _read_json_body(self)
- else:
- length = int(self.headers.get("Content-Length") or 0)
- raw = self.rfile.read(length).decode("utf-8") if length > 0 else ""
- form_payload = {key: values[0] for key, values in parse_qs(raw).items()}
- payload = {
- "kis_account": form_payload.get("kis_account") or "real",
- "include_live_kis": str(form_payload.get("include_live_kis") or "true").lower() == "true",
- "allow_naver_fallback": str(form_payload.get("allow_naver_fallback") or "false").lower() == "true",
- }
- result = run_collection_job(
- sqlite_db=KIS_COLLECTION_DB,
- input_json=ROOT / "GatherTradingData.json",
- output_json=KIS_COLLECTION_REPORT,
- allow_naver_fallback=bool(payload.get("allow_naver_fallback")),
- include_live_kis=bool(payload.get("include_live_kis")),
- kis_account=str(payload.get("kis_account") or "real"),
- )
- _json_response(self, HTTPStatus.OK, result)
- return
- payload = _read_json_body(self)
- if parsed.path == "/api/settings/save":
- if is_locked(self.db_path, "settings"):
- raise ValueError("settings are locked")
- rows = payload.get("rows")
- if not isinstance(rows, list):
- raise ValueError("rows must be a list")
- normalized_rows = []
- for idx, row in enumerate(rows, start=1):
- if not isinstance(row, dict):
- continue
- key = str(row.get("key") or "").strip()
- if not key:
- continue
- normalized_rows.append(
- {
- "ordinal": idx,
- "key": key,
- "value": row.get("value", ""),
- "note": str(row.get("note") or ""),
- }
- )
- conflicts = lock_conflicts_for_rows(self.db_path, "settings", normalized_rows)
- if conflicts:
- refs = ", ".join(sorted({str(item.get("target_ref") or "") for item in conflicts if item.get("target_ref")}))
- raise ValueError(f"settings lock conflict: {refs}")
- with open_connection(self.db_path) as conn:
- replace_settings(conn, normalized_rows)
- _json_response(self, HTTPStatus.OK, summarize_workspace(self.db_path))
- return
- if parsed.path == "/api/account_snapshot/save":
- if is_locked(self.db_path, "account_snapshot"):
- raise ValueError("account_snapshot is locked")
- rows = payload.get("rows")
- if not isinstance(rows, list):
- raise ValueError("rows must be a list")
- normalized_rows: list[dict[str, Any]] = []
- for idx, row in enumerate(rows, start=1):
- if not isinstance(row, dict):
- continue
- candidate = {key: value for key, value in row.items() if not key.startswith("_")}
- candidate["ordinal"] = idx
- normalized_rows.append(candidate)
- conflicts = lock_conflicts_for_rows(self.db_path, "account_snapshot", normalized_rows)
- if conflicts:
- refs = ", ".join(sorted({str(item.get("target_ref") or "") for item in conflicts if item.get("target_ref")}))
- raise ValueError(f"account_snapshot lock conflict: {refs}")
- with open_connection(self.db_path) as conn:
- replace_account_snapshot(conn, normalized_rows)
- _json_response(self, HTTPStatus.OK, summarize_workspace(self.db_path))
- return
- if parsed.path == "/api/account_snapshot/import_tsv":
- if is_locked(self.db_path, "account_snapshot"):
- raise ValueError("account_snapshot is locked")
- tsv_text = str(payload.get("tsv") or "")
- rows = parse_account_snapshot_tsv(tsv_text)
- with open_connection(self.db_path) as conn:
- replace_account_snapshot(conn, rows)
- _json_response(self, HTTPStatus.OK, summarize_workspace(self.db_path))
- return
- if parsed.path == "/api/table/save":
- table = str(payload.get("table") or "").strip()
- rows = payload.get("rows")
- if table not in EDITABLE_TABLES:
- raise ValueError(f"table not editable: {table}")
- if not isinstance(rows, list):
- raise ValueError("rows must be a list")
- db_path = _resolve_table_db(table, self.db_path)
- if not db_path:
- raise ValueError(f"database not found for table: {table}")
- with open_connection(db_path) as conn:
- conn.execute("BEGIN TRANSACTION")
- try:
- conn.execute(f"DELETE FROM {table}") # noqa: S608 - Whitelisted table name
- if rows:
- first_row = rows[0]
- columns = [k for k in first_row.keys() if not k.startswith("_")]
- if "rowid" in columns:
- columns.remove("rowid")
- if "_rowid" in columns:
- columns.remove("_rowid")
- placeholders = ", ".join(["?"] * len(columns))
- col_list = ", ".join(columns)
- insert_sql = f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})" # noqa: S608 - Whitelisted table name
- for row in rows:
- values = [row.get(col) for col in columns]
- conn.execute(insert_sql, values)
- conn.commit()
- except Exception as e:
- conn.rollback()
- raise e
- _json_response(self, HTTPStatus.OK, {"status": "SUCCESS", "table": table, "row_count": len(rows)})
- return
- if parsed.path == "/api/approval_packet":
- packet = payload.get("packet")
- if not isinstance(packet, dict):
- raise ValueError("packet must be an object")
- artifacts = write_approval_packet_artifacts(packet)
- response = {
- "gate": "PASS",
- "packet_path": artifacts["json_path"],
- "md_path": artifacts["md_path"],
- "formula_id": packet.get("formula_id", "SNAPSHOT_ADMIN_APPROVAL_PACKET_V1"),
- }
- _json_response(self, HTTPStatus.OK, response)
- return
- if parsed.path == "/api/approve":
- domain = str(payload.get("domain") or "")
- if domain not in {"settings", "account_snapshot"}:
- raise ValueError("domain must be settings or account_snapshot")
- target_ref = str(payload.get("target_ref") or "*")
- with open_connection(self.db_path) as conn:
- before = _approval_entry_from_conn(conn, domain, target_ref)
- set_approval(conn, domain, "APPROVED", target_ref=target_ref, approved_by="ui", note="manual approval")
- after = _approval_entry_from_conn(conn, domain, target_ref)
- record_change_log(
- conn,
- domain=domain,
- action="approve",
- target_ref=target_ref,
- before_json=before,
- after_json=after,
- actor="ui",
- note="manual approval",
- )
- conn.commit()
- _json_response(self, HTTPStatus.OK, {"domain": domain, "target_ref": target_ref, "status": "APPROVED"})
- return
- if parsed.path == "/api/lock":
- domain = str(payload.get("domain") or "")
- target_ref = str(payload.get("target_ref") or "*")
- if domain not in {"settings", "account_snapshot"}:
- raise ValueError("domain must be settings or account_snapshot")
- with open_connection(self.db_path) as conn:
- before = _lock_entry_from_conn(conn, domain, target_ref)
- set_lock(conn, domain, target_ref, locked_by="ui", reason="manual lock")
- after = _lock_entry_from_conn(conn, domain, target_ref)
- record_change_log(
- conn,
- domain=domain,
- action="lock",
- target_ref=target_ref,
- before_json=before,
- after_json=after,
- actor="ui",
- note="manual lock",
- )
- conn.commit()
- _json_response(self, HTTPStatus.OK, {"domain": domain, "target_ref": target_ref, "status": "LOCKED"})
- return
- if parsed.path == "/api/unlock":
- domain = str(payload.get("domain") or "")
- target_ref = str(payload.get("target_ref") or "*")
- if domain not in {"settings", "account_snapshot"}:
- raise ValueError("domain must be settings or account_snapshot")
- with open_connection(self.db_path) as conn:
- before = _lock_entry_from_conn(conn, domain, target_ref)
- clear_lock(conn, domain, target_ref)
- after = _lock_entry_from_conn(conn, domain, target_ref)
- record_change_log(
- conn,
- domain=domain,
- action="unlock",
- target_ref=target_ref,
- before_json=before,
- after_json=after,
- actor="ui",
- note="manual unlock",
- )
- conn.commit()
- _json_response(self, HTTPStatus.OK, {"domain": domain, "target_ref": target_ref, "status": "UNLOCKED"})
- return
- if parsed.path == "/api/undo":
- domain = str(payload.get("domain") or "")
- if domain not in {"settings", "account_snapshot"}:
- raise ValueError("domain must be settings or account_snapshot")
- if is_locked(self.db_path, domain):
- raise ValueError(f"{domain} is locked")
- with open_connection(self.db_path) as conn:
- result = undo_last_change(conn, domain, actor="ui")
- _json_response(self, HTTPStatus.OK, result if result else {"domain": domain, "status": "UNDONE"})
- return
- if parsed.path == "/api/autofix":
- action_id = str(payload.get("action_id") or "")
- if not action_id:
- raise ValueError("action_id required")
- with open_connection(self.db_path) as conn:
- result = apply_safe_autofix_action(conn, action_id, actor="ui")
- _json_response(self, HTTPStatus.OK, result)
- return
- _json_response(self, HTTPStatus.NOT_FOUND, {"detail": "not found"})
- except Exception as exc: # noqa: BLE001
- self._handle_exception(exc)
-
-
-def serve(host: str, port: int, db_path: Path | str | None = None, seed_json_path: Path | str | None = None, bootstrap: bool = True) -> None:
- db = normalize_db_path(db_path)
- seed = Path(seed_json_path) if seed_json_path else DEFAULT_SEED_JSON
- if bootstrap and seed.exists():
- with open_connection(db) as conn:
- from .snapshot_admin_store_v1 import ensure_schema
-
- ensure_schema(conn)
- if summarize_workspace(db)["settings_rows"] == 0 and summarize_workspace(db)["account_snapshot_rows"] == 0:
- import_seed_json(db, seed)
- SnapshotAdminHandler.db_path = db
- SnapshotAdminHandler.seed_json_path = seed
- server = ThreadingHTTPServer((host, port), SnapshotAdminHandler)
- print(f"Snapshot Admin listening on http://{host}:{port}")
- print(f"SQLite DB: {db}")
- print(f"Seed JSON: {seed}")
- try:
- server.serve_forever()
- except KeyboardInterrupt:
- pass
- finally:
- server.server_close()
-
-
-def main() -> int:
- parser = argparse.ArgumentParser(description="Run the snapshot admin web server.")
- parser.add_argument("--host", default="127.0.0.1")
- parser.add_argument("--port", type=int, default=8787)
- parser.add_argument("--db", type=Path, default=DEFAULT_DB)
- parser.add_argument("--seed", type=Path, default=DEFAULT_SEED_JSON)
- parser.add_argument("--no-bootstrap", action="store_true")
- args = parser.parse_args()
- serve(args.host, args.port, args.db, args.seed, bootstrap=not args.no_bootstrap)
- return 0
-
-
-if __name__ == "__main__":
- raise SystemExit(main())
diff --git a/src/quant_engine/snapshot_admin_store_v1.py b/src/quant_engine/snapshot_admin_store_v1.py
deleted file mode 100644
index 2dddc3a6..00000000
--- a/src/quant_engine/snapshot_admin_store_v1.py
+++ /dev/null
@@ -1,1033 +0,0 @@
-from __future__ import annotations
-
-import json
-import re
-import sqlite3
-from datetime import datetime, timedelta, timezone
-from functools import lru_cache
-from pathlib import Path
-from typing import Any
-
-import yaml
-
-
-ROOT = Path(__file__).resolve().parents[2]
-DEFAULT_DB = ROOT / "src" / "quant_engine" / "snapshot_admin.db"
-DEFAULT_SEED_JSON = ROOT / "GatherTradingData.json"
-KST = timezone(timedelta(hours=9))
-
-SETTINGS_TABLE = "settings"
-SNAPSHOT_TABLE = "account_snapshot"
-CHANGE_LOG_TABLE = "workspace_change_log"
-APPROVAL_TABLE = "workspace_approval_v2"
-LOCK_TABLE = "workspace_lock"
-
-ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS = [
- "captured_at",
- "account",
- "account_type",
- "ticker",
- "name",
- "holding_quantity",
- "available_quantity",
- "average_cost",
- "total_cost",
- "current_price",
- "market_value",
- "profit_loss",
- "return_pct",
- "immediate_cash",
- "settlement_cash_d2",
- "available_cash",
- "open_order_amount",
- "monthly_contribution_limit",
- "monthly_contribution_used",
- "parse_status",
- "user_confirmed",
- "stop_price",
- "highest_price_since_entry",
- "entry_date",
- "entry_stage",
- "position_type",
- "last_updated",
-]
-
-ALLOWED_PARSE_STATUS = {
- "CAPTURE_READ_OK",
- "CAPTURE_READ_FAILED",
- "CAPTURE_PROVIDED_BUT_NOT_HOLDINGS",
- "NOT_PROVIDED",
-}
-
-SETTINGS_SPEC_PATH = ROOT / "spec" / "18_settings_contract.yaml"
-ACCOUNT_SNAPSHOT_SPEC_PATH = ROOT / "spec" / "15_account_snapshot_contract.yaml"
-
-
-def now_kst_iso() -> str:
- return datetime.now(tz=KST).isoformat(timespec="seconds")
-
-
-def parse_scalar(value: str) -> Any:
- text = value.strip()
- if text == "":
- return ""
- if text.lower() in {"null", "none"}:
- return None
- if text.lower() in {"true", "false"}:
- return text.lower() == "true"
- try:
- return json.loads(text)
- except Exception:
- return text
-
-
-def _json_dump(value: Any) -> str:
- return json.dumps(value, ensure_ascii=False)
-
-
-def _json_load(text: str) -> Any:
- try:
- return json.loads(text)
- except Exception:
- return text
-
-
-def normalize_db_path(db_path: Path | str | None = None) -> Path:
- path = Path(db_path) if db_path else DEFAULT_DB
- path.parent.mkdir(parents=True, exist_ok=True)
- return path
-
-
-def open_connection(db_path: Path | str | None = None) -> sqlite3.Connection:
- conn = sqlite3.connect(normalize_db_path(db_path))
- conn.row_factory = sqlite3.Row
- conn.execute("PRAGMA foreign_keys = ON")
- conn.execute("PRAGMA journal_mode = WAL")
- return conn
-
-
-def ensure_schema(conn: sqlite3.Connection) -> None:
- conn.execute(
- f"""
- CREATE TABLE IF NOT EXISTS {SETTINGS_TABLE} (
- ordinal INTEGER NOT NULL,
- key TEXT PRIMARY KEY,
- value_json TEXT NOT NULL,
- note TEXT NOT NULL DEFAULT '',
- updated_at TEXT NOT NULL
- )
- """
- )
- conn.execute(
- f"""
- CREATE TABLE IF NOT EXISTS {SNAPSHOT_TABLE} (
- ordinal INTEGER NOT NULL,
- row_json TEXT NOT NULL,
- captured_at TEXT NOT NULL DEFAULT '',
- account TEXT NOT NULL DEFAULT '',
- account_type TEXT NOT NULL DEFAULT '',
- ticker TEXT NOT NULL DEFAULT '',
- name TEXT NOT NULL DEFAULT '',
- parse_status TEXT NOT NULL DEFAULT '',
- user_confirmed TEXT NOT NULL DEFAULT '',
- updated_at TEXT NOT NULL
- )
- """
- )
- conn.execute(
- f"CREATE INDEX IF NOT EXISTS idx_{SNAPSHOT_TABLE}_captured_at ON {SNAPSHOT_TABLE}(captured_at)"
- )
- conn.execute(
- f"CREATE INDEX IF NOT EXISTS idx_{SNAPSHOT_TABLE}_ticker ON {SNAPSHOT_TABLE}(ticker)"
- )
- conn.execute(
- "CREATE TABLE IF NOT EXISTS workspace_meta (key TEXT PRIMARY KEY, value_json TEXT NOT NULL)"
- )
- conn.execute(
- f"""
- CREATE TABLE IF NOT EXISTS {CHANGE_LOG_TABLE} (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- domain TEXT NOT NULL,
- action TEXT NOT NULL,
- target_ref TEXT NOT NULL DEFAULT '',
- actor TEXT NOT NULL DEFAULT 'system',
- note TEXT NOT NULL DEFAULT '',
- before_json TEXT NOT NULL DEFAULT 'null',
- after_json TEXT NOT NULL DEFAULT 'null',
- created_at TEXT NOT NULL
- )
- """
- )
- conn.execute(
- f"""
- CREATE TABLE IF NOT EXISTS {APPROVAL_TABLE} (
- domain TEXT NOT NULL,
- target_ref TEXT NOT NULL DEFAULT '*',
- status TEXT NOT NULL,
- approved_by TEXT NOT NULL DEFAULT '',
- approved_at TEXT NOT NULL DEFAULT '',
- note TEXT NOT NULL DEFAULT '',
- updated_at TEXT NOT NULL,
- PRIMARY KEY (domain, target_ref)
- )
- """
- )
- conn.execute(
- f"""
- CREATE TABLE IF NOT EXISTS {LOCK_TABLE} (
- domain TEXT NOT NULL,
- target_ref TEXT NOT NULL DEFAULT '',
- locked_by TEXT NOT NULL DEFAULT '',
- reason TEXT NOT NULL DEFAULT '',
- locked_at TEXT NOT NULL,
- PRIMARY KEY (domain, target_ref)
- )
- """
- )
- conn.commit()
-
-
-def _normalize_settings_rows(settings: Any) -> list[dict[str, Any]]:
- if isinstance(settings, list):
- rows: list[dict[str, Any]] = []
- for idx, item in enumerate(settings, start=1):
- if isinstance(item, dict) and "key" in item:
- rows.append(
- {
- "ordinal": int(item.get("ordinal") or idx),
- "key": str(item.get("key") or ""),
- "value": item.get("value", ""),
- "note": str(item.get("note") or ""),
- }
- )
- return rows
- if isinstance(settings, dict):
- rows = []
- for idx, (key, value) in enumerate(settings.items(), start=1):
- rows.append({"ordinal": idx, "key": str(key), "value": value, "note": ""})
- return rows
- return []
-
-
-def _normalize_snapshot_rows(rows: Any) -> list[dict[str, Any]]:
- if not isinstance(rows, list):
- return []
- normalized: list[dict[str, Any]] = []
- for idx, item in enumerate(rows, start=1):
- if isinstance(item, dict):
- row = dict(item)
- row.setdefault("ordinal", idx)
- normalized.append(row)
- return normalized
-
-
-def seed_payload_from_json(json_path: Path | str) -> dict[str, Any]:
- payload = json.loads(Path(json_path).read_text(encoding="utf-8"))
- data = payload.get("data") if isinstance(payload, dict) else None
- if not isinstance(data, dict):
- data = payload if isinstance(payload, dict) else {}
- settings = _normalize_settings_rows(data.get("settings"))
- account_snapshot = _normalize_snapshot_rows(data.get("account_snapshot"))
- return {
- "meta": payload.get("meta") if isinstance(payload, dict) else {},
- "settings": settings,
- "account_snapshot": account_snapshot,
- }
-
-
-def replace_settings(conn: sqlite3.Connection, rows: list[dict[str, Any]]) -> None:
- ensure_schema(conn)
- errors = validate_settings_rows(rows)
- if errors:
- raise ValueError("; ".join(errors))
- old_rows = load_settings_rows_from_conn(conn)
- conn.execute(f"DELETE FROM {SETTINGS_TABLE}")
- for idx, row in enumerate(rows, start=1):
- key = str(row.get("key") or "").strip()
- if not key:
- continue
- conn.execute(
- f"""
- INSERT INTO {SETTINGS_TABLE} (ordinal, key, value_json, note, updated_at)
- VALUES (?, ?, ?, ?, ?)
- """,
- (
- int(row.get("ordinal") or idx),
- key,
- _json_dump(row.get("value", "")),
- str(row.get("note") or ""),
- now_kst_iso(),
- ),
- )
- record_change_log(
- conn,
- domain=SETTINGS_TABLE,
- action="replace",
- before_json=old_rows,
- after_json=rows,
- target_ref="*",
- note="settings replace",
- )
- set_approval(conn, SETTINGS_TABLE, "PENDING", note="settings updated")
- conn.commit()
-
-
-def replace_account_snapshot(conn: sqlite3.Connection, rows: list[dict[str, Any]]) -> None:
- ensure_schema(conn)
- errors = validate_account_snapshot_rows(rows)
- if errors:
- raise ValueError("; ".join(errors))
- old_rows = load_account_snapshot_rows_from_conn(conn)
- conn.execute(f"DELETE FROM {SNAPSHOT_TABLE}")
- for idx, row in enumerate(rows, start=1):
- normalized = dict(row)
- ordinal = int(normalized.pop("ordinal", idx) or idx)
- captured_at = str(normalized.get("captured_at") or "")
- account = str(normalized.get("account") or "")
- account_type = str(normalized.get("account_type") or "")
- ticker = str(normalized.get("ticker") or "")
- name = str(normalized.get("name") or "")
- parse_status = str(normalized.get("parse_status") or "")
- user_confirmed = str(normalized.get("user_confirmed") or "")
- conn.execute(
- f"""
- INSERT INTO {SNAPSHOT_TABLE} (
- ordinal, row_json, captured_at, account, account_type, ticker, name,
- parse_status, user_confirmed, updated_at
- )
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
- """,
- (
- ordinal,
- _json_dump(normalized),
- captured_at,
- account,
- account_type,
- ticker,
- name,
- parse_status,
- user_confirmed,
- now_kst_iso(),
- ),
- )
- record_change_log(
- conn,
- domain=SNAPSHOT_TABLE,
- action="replace",
- before_json=old_rows,
- after_json=rows,
- target_ref="*",
- note="account_snapshot replace",
- )
- set_approval(conn, SNAPSHOT_TABLE, "PENDING", note="account_snapshot updated")
- conn.commit()
-
-
-def import_seed_json(db_path: Path | str | None, json_path: Path | str) -> dict[str, Any]:
- payload = seed_payload_from_json(json_path)
- with open_connection(db_path) as conn:
- replace_settings(conn, payload["settings"])
- replace_account_snapshot(conn, payload["account_snapshot"])
- conn.execute(
- "INSERT OR REPLACE INTO workspace_meta(key, value_json) VALUES (?, ?)",
- ("seed_json_path", _json_dump(str(Path(json_path).resolve()))),
- )
- conn.execute(
- "INSERT OR REPLACE INTO workspace_meta(key, value_json) VALUES (?, ?)",
- ("seeded_at", _json_dump(now_kst_iso())),
- )
- conn.commit()
- return summarize_workspace(db_path)
-
-
-def load_settings_rows(db_path: Path | str | None = None) -> list[dict[str, Any]]:
- with open_connection(db_path) as conn:
- return load_settings_rows_from_conn(conn)
-
-
-def load_settings_rows_from_conn(conn: sqlite3.Connection) -> list[dict[str, Any]]:
- ensure_schema(conn)
- rows = conn.execute(
- f"SELECT ordinal, key, value_json, note, updated_at FROM {SETTINGS_TABLE} ORDER BY ordinal ASC, key ASC"
- ).fetchall()
- return [
- {
- "ordinal": int(row["ordinal"]),
- "key": row["key"],
- "value": _json_load(row["value_json"]),
- "note": row["note"],
- "updated_at": row["updated_at"],
- }
- for row in rows
- ]
-
-
-def load_account_snapshot_rows(db_path: Path | str | None = None) -> list[dict[str, Any]]:
- with open_connection(db_path) as conn:
- return load_account_snapshot_rows_from_conn(conn)
-
-
-def load_account_snapshot_rows_from_conn(conn: sqlite3.Connection) -> list[dict[str, Any]]:
- ensure_schema(conn)
- rows = conn.execute(
- f"""
- SELECT ordinal, row_json, captured_at, account, account_type, ticker, name,
- parse_status, user_confirmed, updated_at
- FROM {SNAPSHOT_TABLE}
- ORDER BY ordinal ASC
- """
- ).fetchall()
- loaded: list[dict[str, Any]] = []
- for row in rows:
- payload = _json_load(row["row_json"])
- item = payload if isinstance(payload, dict) else {}
- item.setdefault("captured_at", row["captured_at"])
- item.setdefault("account", row["account"])
- item.setdefault("account_type", row["account_type"])
- item.setdefault("ticker", row["ticker"])
- item.setdefault("name", row["name"])
- item.setdefault("parse_status", row["parse_status"])
- item.setdefault("user_confirmed", row["user_confirmed"])
- item["_ordinal"] = int(row["ordinal"])
- item["_updated_at"] = row["updated_at"]
- loaded.append(item)
- return loaded
-
-
-def export_payload(db_path: Path | str | None = None) -> dict[str, Any]:
- settings_rows = load_settings_rows(db_path)
- settings = {row["key"]: row["value"] for row in settings_rows}
- account_snapshot = load_account_snapshot_rows(db_path)
- return {
- "meta": {
- "generated_at": now_kst_iso(),
- "source_db": str(normalize_db_path(db_path)),
- },
- "data": {
- "settings": settings,
- "account_snapshot": account_snapshot,
- },
- }
-
-
-def write_export_json(db_path: Path | str | None, output_path: Path | str) -> Path:
- payload = export_payload(db_path)
- output = Path(output_path)
- output.parent.mkdir(parents=True, exist_ok=True)
- output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
- return output
-
-
-def load_meta(db_path: Path | str | None = None) -> dict[str, Any]:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- rows = conn.execute("SELECT key, value_json FROM workspace_meta ORDER BY key ASC").fetchall()
- return {row["key"]: _json_load(row["value_json"]) for row in rows}
-
-
-def record_change_log(
- conn: sqlite3.Connection,
- *,
- domain: str,
- action: str,
- before_json: Any,
- after_json: Any,
- target_ref: str = "",
- actor: str = "ui",
- note: str = "",
-) -> None:
- ensure_schema(conn)
- conn.execute(
- f"""
- INSERT INTO {CHANGE_LOG_TABLE} (
- domain, action, target_ref, actor, note, before_json, after_json, created_at
- )
- VALUES (?, ?, ?, ?, ?, ?, ?, ?)
- """,
- (
- domain,
- action,
- target_ref,
- actor,
- note,
- _json_dump(before_json),
- _json_dump(after_json),
- now_kst_iso(),
- ),
- )
-
-
-def set_approval(
- conn: sqlite3.Connection,
- domain: str,
- status: str,
- *,
- target_ref: str = "*",
- approved_by: str = "",
- note: str = "",
-) -> None:
- ensure_schema(conn)
- conn.execute(
- f"""
- INSERT INTO {APPROVAL_TABLE} (domain, target_ref, status, approved_by, approved_at, note, updated_at)
- VALUES (?, ?, ?, ?, ?, ?, ?)
- ON CONFLICT(domain, target_ref) DO UPDATE SET
- status=excluded.status,
- approved_by=excluded.approved_by,
- approved_at=excluded.approved_at,
- note=excluded.note,
- updated_at=excluded.updated_at
- """,
- (
- domain,
- target_ref or "*",
- status,
- approved_by,
- now_kst_iso() if status == "APPROVED" else "",
- note,
- now_kst_iso(),
- ),
- )
-
-
-def load_approval_rows(db_path: Path | str | None = None) -> list[dict[str, Any]]:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- rows = conn.execute(
- f"SELECT domain, target_ref, status, approved_by, approved_at, note, updated_at FROM {APPROVAL_TABLE} ORDER BY domain ASC, target_ref ASC"
- ).fetchall()
- return [dict(row) for row in rows]
-
-
-def load_approval_entry(db_path: Path | str | None, domain: str, target_ref: str = "*") -> dict[str, Any] | None:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- row = conn.execute(
- f"""
- SELECT domain, target_ref, status, approved_by, approved_at, note, updated_at
- FROM {APPROVAL_TABLE}
- WHERE domain = ? AND target_ref = ?
- LIMIT 1
- """,
- (domain, target_ref or "*"),
- ).fetchone()
- return dict(row) if row is not None else None
-
-
-def load_change_log_rows(db_path: Path | str | None = None, limit: int = 20) -> list[dict[str, Any]]:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- rows = conn.execute(
- f"""
- SELECT id, domain, action, target_ref, actor, note, before_json, after_json, created_at
- FROM {CHANGE_LOG_TABLE}
- ORDER BY id DESC
- LIMIT ?
- """,
- (int(limit),),
- ).fetchall()
- items = []
- for row in rows:
- items.append(
- {
- "id": int(row["id"]),
- "domain": row["domain"],
- "action": row["action"],
- "target_ref": row["target_ref"],
- "actor": row["actor"],
- "note": row["note"],
- "before_json": _json_load(row["before_json"]),
- "after_json": _json_load(row["after_json"]),
- "created_at": row["created_at"],
- }
- )
- return items
-
-
-def load_last_change_row(conn: sqlite3.Connection, domain: str) -> dict[str, Any] | None:
- ensure_schema(conn)
- row = conn.execute(
- f"""
- SELECT id, domain, action, target_ref, actor, note, before_json, after_json, created_at
- FROM {CHANGE_LOG_TABLE}
- WHERE domain = ?
- ORDER BY id DESC
- LIMIT 1
- """,
- (domain,),
- ).fetchone()
- if row is None:
- return None
- return {
- "id": int(row["id"]),
- "domain": row["domain"],
- "action": row["action"],
- "target_ref": row["target_ref"],
- "actor": row["actor"],
- "note": row["note"],
- "before_json": _json_load(row["before_json"]),
- "after_json": _json_load(row["after_json"]),
- "created_at": row["created_at"],
- }
-
-
-def set_lock(conn: sqlite3.Connection, domain: str, target_ref: str, *, locked_by: str, reason: str) -> None:
- ensure_schema(conn)
- conn.execute(
- f"""
- INSERT INTO {LOCK_TABLE} (domain, target_ref, locked_by, reason, locked_at)
- VALUES (?, ?, ?, ?, ?)
- ON CONFLICT(domain, target_ref) DO UPDATE SET
- locked_by=excluded.locked_by,
- reason=excluded.reason,
- locked_at=excluded.locked_at
- """,
- (domain, target_ref, locked_by, reason, now_kst_iso()),
- )
-
-
-def clear_lock(conn: sqlite3.Connection, domain: str, target_ref: str) -> None:
- ensure_schema(conn)
- conn.execute(
- f"DELETE FROM {LOCK_TABLE} WHERE domain = ? AND target_ref = ?",
- (domain, target_ref),
- )
-
-
-def load_locks(db_path: Path | str | None = None) -> list[dict[str, Any]]:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- rows = conn.execute(
- f"SELECT domain, target_ref, locked_by, reason, locked_at FROM {LOCK_TABLE} ORDER BY domain ASC, target_ref ASC"
- ).fetchall()
- return [dict(row) for row in rows]
-
-
-def load_lock_entry(db_path: Path | str | None, domain: str, target_ref: str = "*") -> dict[str, Any] | None:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- row = conn.execute(
- f"""
- SELECT domain, target_ref, locked_by, reason, locked_at
- FROM {LOCK_TABLE}
- WHERE domain = ? AND target_ref = ?
- LIMIT 1
- """,
- (domain, target_ref or "*"),
- ).fetchone()
- return dict(row) if row is not None else None
-
-
-def is_locked(db_path: Path | str | None, domain: str, target_ref: str = "*") -> bool:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- row = conn.execute(
- f"SELECT 1 FROM {LOCK_TABLE} WHERE domain = ? AND target_ref IN (?, '*') LIMIT 1",
- (domain, target_ref),
- ).fetchone()
- return row is not None
-
-
-def lock_conflicts_for_rows(
- db_path: Path | str | None,
- domain: str,
- rows: list[dict[str, Any]],
-) -> list[dict[str, Any]]:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- locks = conn.execute(
- f"SELECT domain, target_ref, locked_by, reason, locked_at FROM {LOCK_TABLE} WHERE domain = ? ORDER BY target_ref ASC",
- (domain,),
- ).fetchall()
- if not locks:
- return []
- row_refs: list[str] = []
- for idx, row in enumerate(rows, start=1):
- if domain == SETTINGS_TABLE:
- ref = str(row.get("_row_ref") or "").strip() or str(row.get("key") or "").strip()
- elif domain == SNAPSHOT_TABLE:
- ref = str(row.get("_row_ref") or "").strip()
- if not ref:
- ordinal = str(row.get("_ordinal") or row.get("ordinal") or idx).strip()
- ref = f"row:{ordinal}"
- else:
- ref = str(row.get("target_ref") or "").strip()
- if ref:
- row_refs.append(ref)
- if domain == SETTINGS_TABLE:
- key = str(row.get("key") or "").strip()
- if key:
- row_refs.append(key)
- if domain == SNAPSHOT_TABLE:
- ticker = str(row.get("ticker") or "").strip()
- if ticker:
- row_refs.append(ticker)
- conflicts: list[dict[str, Any]] = []
- for lock in locks:
- target_ref = str(lock["target_ref"] or "").strip()
- if target_ref == "*" or target_ref in row_refs:
- conflicts.append(dict(lock))
- return conflicts
-
-
-def undo_last_change(conn: sqlite3.Connection, domain: str, *, actor: str = "ui") -> dict[str, Any]:
- ensure_schema(conn)
- last = load_last_change_row(conn, domain)
- if not last:
- raise ValueError(f"no change log for domain={domain}")
- before_json = last.get("before_json")
- if domain == SETTINGS_TABLE:
- rows = before_json if isinstance(before_json, list) else []
- replace_settings(conn, rows)
- elif domain == SNAPSHOT_TABLE:
- rows = before_json if isinstance(before_json, list) else []
- replace_account_snapshot(conn, rows)
- else:
- raise ValueError(f"unsupported domain={domain}")
- record_change_log(
- conn,
- domain=domain,
- action="undo",
- before_json=last.get("after_json"),
- after_json=before_json,
- target_ref=last.get("target_ref", "*"),
- actor=actor,
- note=f"undo change #{last['id']}",
- )
- conn.commit()
- return load_last_change_row(conn, domain) or {}
-
-
-def load_approval_for_domain(db_path: Path | str | None, domain: str) -> dict[str, Any]:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- row = conn.execute(
- f"""
- SELECT domain, target_ref, status, approved_by, approved_at, note, updated_at
- FROM {APPROVAL_TABLE}
- WHERE domain = ? AND target_ref = '*'
- """,
- (domain,),
- ).fetchone()
- return (
- dict(row)
- if row
- else {"domain": domain, "target_ref": "*", "status": "MISSING", "approved_by": "", "approved_at": "", "note": "", "updated_at": ""}
- )
-
-
-def summarize_workspace(db_path: Path | str | None = None) -> dict[str, Any]:
- with open_connection(db_path) as conn:
- ensure_schema(conn)
- settings_count = conn.execute(f"SELECT COUNT(*) FROM {SETTINGS_TABLE}").fetchone()[0]
- snapshot_count = conn.execute(f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE}").fetchone()[0]
- latest_update = conn.execute(
- f"""
- SELECT MAX(latest_ts)
- FROM (
- SELECT updated_at as latest_ts FROM {SETTINGS_TABLE}
- UNION ALL
- SELECT captured_at FROM {SNAPSHOT_TABLE}
- )
- """
- ).fetchone()[0]
- table_rows = conn.execute(
- "SELECT name FROM sqlite_master WHERE type='table' AND name IN (?, ?, ?, ?, ?)",
- (SETTINGS_TABLE, SNAPSHOT_TABLE, CHANGE_LOG_TABLE, APPROVAL_TABLE, LOCK_TABLE),
- ).fetchall()
- tables = sorted(row[0] for row in table_rows)
- workspace_db = str(normalize_db_path(db_path))
- return {
- "db_path": workspace_db,
- "settings_rows": int(settings_count),
- "account_snapshot_rows": int(snapshot_count),
- "latest_update": latest_update or "",
- "tables": tables,
- "topology": {
- "mode": "single_workspace_sqlite",
- "workspace_db": workspace_db,
- "collector_db": str(ROOT / "src" / "quant_engine" / "kis_data_collection.db"),
- "settings_and_snapshot_share_db": True,
- "collector_separate_db": True,
- },
- "meta": load_meta(db_path),
- }
-
-
-def parse_account_snapshot_tsv(tsv_text: str) -> list[dict[str, Any]]:
- lines = [line.rstrip("\r") for line in tsv_text.splitlines() if line.strip() != ""]
- if not lines:
- return []
- rows: list[list[str]] = [line.split("\t") for line in lines]
- first_row = rows[0]
- if first_row == ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS:
- data_rows = rows[1:]
- elif set(first_row) >= {"captured_at", "account", "ticker"}:
- header = first_row
- data_rows = rows[1:]
- converted: list[dict[str, Any]] = []
- for idx, row in enumerate(data_rows, start=1):
- item: dict[str, Any] = {"ordinal": idx}
- for col_index, column in enumerate(header):
- value = row[col_index] if col_index < len(row) else ""
- item[column] = parse_scalar(value)
- converted.append(item)
- return converted
- else:
- data_rows = rows
- converted = []
- for idx, row in enumerate(data_rows, start=1):
- item: dict[str, Any] = {"ordinal": idx}
- for col_index, column in enumerate(ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS):
- value = row[col_index] if col_index < len(row) else ""
- item[column] = parse_scalar(value)
- converted.append(item)
- return converted
-
-
-def settings_rows_to_dict(rows: list[dict[str, Any]]) -> dict[str, Any]:
- result: dict[str, Any] = {}
- for row in rows:
- key = str(row.get("key") or "").strip()
- if key:
- result[key] = row.get("value", "")
- return result
-
-
-def _as_number(value: Any) -> float | None:
- if value is None:
- return None
- if isinstance(value, bool):
- return None
- if isinstance(value, (int, float)):
- return float(value)
- text = str(value).strip()
- if not text:
- return None
- try:
- return float(text)
- except Exception:
- return None
-
-
-def _as_int(value: Any) -> int | None:
- if value is None or value == "":
- return None
- if isinstance(value, bool):
- return None
- if isinstance(value, int):
- return value
- if isinstance(value, float):
- return int(value) if value.is_integer() else None
- try:
- text = str(value).strip().replace(",", "")
- if not text:
- return None
- parsed = float(text)
- return int(parsed) if parsed.is_integer() else None
- except Exception:
- return None
-
-
-@lru_cache(maxsize=1)
-def _load_settings_spec() -> dict[str, Any]:
- return yaml.safe_load(SETTINGS_SPEC_PATH.read_text(encoding="utf-8")) or {}
-
-
-@lru_cache(maxsize=1)
-def _load_account_snapshot_spec() -> dict[str, Any]:
- return yaml.safe_load(ACCOUNT_SNAPSHOT_SPEC_PATH.read_text(encoding="utf-8")) or {}
-
-
-def validate_settings_rows(rows: list[dict[str, Any]]) -> list[str]:
- errors: list[str] = []
- spec = _load_settings_spec().get("required_keys") or {}
- optional_spec = _load_settings_spec().get("optional_keys") or {}
- seen: set[str] = set()
- total_asset_found = False
- for idx, row in enumerate(rows, start=1):
- key = str(row.get("key") or "").strip()
- if not key:
- errors.append(f"settings row {idx}: missing key")
- continue
- if key in seen:
- errors.append(f"settings row {idx}: duplicate key {key}")
- seen.add(key)
- value = row.get("value", "")
- if key == "total_asset_krw":
- total_asset_found = True
- amount = _as_number(value)
- if amount is None or amount <= 0:
- errors.append("settings.total_asset_krw must be positive number")
- if key in {"weekly_target_cash_pct", "fc_budget_pct_override"}:
- pct = _as_number(value)
- if pct is None or pct < 0:
- errors.append(f"settings.{key} must be non-negative number")
- if key in spec and spec[key].get("type") == "string":
- if value is not None and not isinstance(value, str):
- errors.append(f"settings.{key} must be string")
- if key in optional_spec and optional_spec[key].get("format") == "YYYY-MM":
- text = str(value).strip()
- if text and not re.fullmatch(r"\d{4}-\d{2}(-.*)?", text):
- errors.append(f"settings.{key} must use YYYY-MM")
- if not total_asset_found:
- errors.append("settings.total_asset_krw is required")
- return errors
-
-
-def validate_account_snapshot_rows(rows: list[dict[str, Any]]) -> list[str]:
- errors: list[str] = []
- spec = _load_account_snapshot_spec().get("account_snapshot_contract") or {}
- canonical = spec.get("canonical_fields") or {}
- for idx, row in enumerate(rows, start=1):
- captured_at = str(row.get("captured_at") or "").strip()
- account = str(row.get("account") or "").strip()
- ticker = str(row.get("ticker") or "").strip()
- name = str(row.get("name") or "").strip()
- account_type = str(row.get("account_type") or "").strip()
- parse_status = str(row.get("parse_status") or "").strip()
- holding_quantity = _as_int(row.get("holding_quantity"))
- available_quantity = _as_int(row.get("available_quantity"))
- average_cost = _as_number(row.get("average_cost"))
- total_cost = _as_number(row.get("total_cost"))
- current_price = _as_number(row.get("current_price"))
- market_value = _as_number(row.get("market_value"))
- profit_loss = _as_number(row.get("profit_loss"))
- return_pct = _as_number(row.get("return_pct"))
- stop_price = _as_number(row.get("stop_price"))
- entry_stage = str(row.get("entry_stage") or "").strip()
- position_type = str(row.get("position_type") or "").strip()
- user_confirmed = str(row.get("user_confirmed") or "").strip().upper()
- if not captured_at:
- errors.append(f"account_snapshot row {idx}: captured_at required")
- if not account:
- errors.append(f"account_snapshot row {idx}: account required")
- if not account_type:
- errors.append(f"account_snapshot row {idx}: account_type required")
- if account_type and canonical.get("account_type", {}).get("allowed") and account_type not in canonical["account_type"]["allowed"]:
- errors.append(f"account_snapshot row {idx}: invalid account_type {account_type!r}")
- if not ticker and name != "예수금/D+2현금":
- errors.append(f"account_snapshot row {idx}: ticker required")
- if ticker and not re.fullmatch(r"(?:\d{6}|[A-Z0-9]{6}|[A-Z]{1,5})", ticker):
- errors.append(f"account_snapshot row {idx}: ticker must be 6 digits or an uppercase symbol")
- if not name:
- errors.append(f"account_snapshot row {idx}: name required")
- if parse_status not in ALLOWED_PARSE_STATUS:
- errors.append(f"account_snapshot row {idx}: invalid parse_status {parse_status!r}")
- if holding_quantity is not None and holding_quantity < 0:
- errors.append(f"account_snapshot row {idx}: holding_quantity must be >= 0")
- if available_quantity is not None and available_quantity < 0:
- errors.append(f"account_snapshot row {idx}: available_quantity must be >= 0")
- if average_cost is not None and average_cost < 0:
- errors.append(f"account_snapshot row {idx}: average_cost must be >= 0")
- if total_cost is not None and total_cost < 0:
- errors.append(f"account_snapshot row {idx}: total_cost must be >= 0")
- if current_price is not None and current_price < 0:
- errors.append(f"account_snapshot row {idx}: current_price must be >= 0")
- if market_value is not None and market_value < 0:
- errors.append(f"account_snapshot row {idx}: market_value must be >= 0")
- if profit_loss is not None and profit_loss != profit_loss:
- errors.append(f"account_snapshot row {idx}: profit_loss invalid")
- if return_pct is not None and abs(return_pct) > 1000:
- errors.append(f"account_snapshot row {idx}: return_pct out of range")
- if stop_price is not None and stop_price < 0:
- errors.append(f"account_snapshot row {idx}: stop_price must be >= 0")
- if user_confirmed and user_confirmed not in {"Y", "N"}:
- errors.append(f"account_snapshot row {idx}: user_confirmed must be Y or N")
- if parse_status == "CAPTURE_READ_OK" and user_confirmed != "Y":
- errors.append(f"account_snapshot row {idx}: CAPTURE_READ_OK rows require user_confirmed=Y")
- if entry_stage and canonical.get("entry_stage", {}).get("allowed") and entry_stage not in canonical["entry_stage"]["allowed"]:
- errors.append(f"account_snapshot row {idx}: invalid entry_stage {entry_stage!r}")
- if position_type and name != "예수금/D+2현금" and canonical.get("position_type", {}).get("allowed") and position_type not in canonical["position_type"]["allowed"]:
- errors.append(f"account_snapshot row {idx}: invalid position_type {position_type!r}")
- return errors
-
-
-def build_validation_suggestions(settings_rows: list[dict[str, Any]], snapshot_rows: list[dict[str, Any]]) -> list[str]:
- suggestions: list[str] = []
- settings_map = settings_rows_to_dict(settings_rows)
- snapshot_count = len(snapshot_rows)
- if "total_asset_krw" not in settings_map:
- suggestions.append("settings: add total_asset_krw from current investable asset total")
- if str(settings_map.get("weekly_target_cash_pct", "")).strip() == "":
- suggestions.append("settings: weekly_target_cash_pct can stay blank unless weekly rebalance is active")
- for row in snapshot_rows:
- if str(row.get("parse_status") or "").strip() == "CAPTURE_READ_OK" and str(row.get("user_confirmed") or "").strip().upper() != "Y":
- suggestions.append(
- f"account_snapshot {row.get('ticker') or row.get('name') or 'row'}: set user_confirmed=Y for CAPTURE_READ_OK"
- )
- account_type = str(row.get("account_type") or "").strip()
- if account_type and account_type not in {"일반계좌", "ISA", "연금저축"}:
- suggestions.append(
- f"account_snapshot {row.get('ticker') or row.get('name') or 'row'}: account_type should be one of 일반계좌/ISA/연금저축"
- )
- if str(row.get("entry_stage") or "").strip() and str(row.get("position_type") or "").strip() == "":
- suggestions.append(
- f"account_snapshot {row.get('ticker') or row.get('name') or 'row'}: consider setting position_type when entry_stage is present"
- )
- if not snapshot_rows:
- suggestions.append("account_snapshot: import TSV from HTS capture before saving snapshot")
- return suggestions[:20]
-
-
-def build_safe_autofix_actions(settings_rows: list[dict[str, Any]], snapshot_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
- actions: list[dict[str, Any]] = []
- if any(str(row.get("parse_status") or "").strip() == "CAPTURE_READ_OK" and str(row.get("user_confirmed") or "").strip().upper() != "Y" for row in snapshot_rows):
- actions.append(
- {
- "action_id": "confirm_captured_rows",
- "domain": "account_snapshot",
- "label": "Set user_confirmed=Y for CAPTURE_READ_OK rows",
- "description": "Safe autofix using the contract default confirmation flag.",
- }
- )
- if any(str(row.get("position_type") or "").strip() == "" and str(row.get("entry_stage") or "").strip() for row in snapshot_rows):
- actions.append(
- {
- "action_id": "default_position_type_satellite",
- "domain": "account_snapshot",
- "label": "Default blank position_type to satellite",
- "description": "Uses the contract default when position_type is missing.",
- }
- )
- if not any(str(row.get("key") or "").strip() == "total_asset_krw" for row in settings_rows):
- actions.append(
- {
- "action_id": "required_total_asset_missing",
- "domain": "settings",
- "label": "Settings total_asset_krw missing",
- "description": "Manual input required. No safe autofix.",
- }
- )
- return actions
-
-
-def apply_safe_autofix_action(
- conn: sqlite3.Connection,
- action_id: str,
- *,
- actor: str = "ui",
-) -> dict[str, Any]:
- ensure_schema(conn)
- snapshot_rows = load_account_snapshot_rows_from_conn(conn)
- if action_id == "confirm_captured_rows":
- updated = []
- for row in snapshot_rows:
- candidate = dict(row)
- if str(candidate.get("parse_status") or "").strip() == "CAPTURE_READ_OK" and str(candidate.get("user_confirmed") or "").strip().upper() != "Y":
- candidate["user_confirmed"] = "Y"
- updated.append(candidate)
- replace_account_snapshot(conn, updated)
- return {"domain": SNAPSHOT_TABLE, "status": "AUTOFIXED", "action_id": action_id}
- if action_id == "default_position_type_satellite":
- updated = []
- for row in snapshot_rows:
- candidate = dict(row)
- if str(candidate.get("entry_stage") or "").strip() and str(candidate.get("position_type") or "").strip() == "":
- candidate["position_type"] = "satellite"
- updated.append(candidate)
- replace_account_snapshot(conn, updated)
- return {"domain": SNAPSHOT_TABLE, "status": "AUTOFIXED", "action_id": action_id}
- if action_id == "required_total_asset_missing":
- return {"domain": SETTINGS_TABLE, "status": "MANUAL_REQUIRED", "action_id": action_id}
- raise ValueError(f"unknown action_id={action_id}")
diff --git a/tools/initialize_databases.py b/tools/initialize_databases.py
deleted file mode 100644
index 4f0c930e..00000000
--- a/tools/initialize_databases.py
+++ /dev/null
@@ -1,142 +0,0 @@
-#!/usr/bin/env python3
-"""
-데이터베이스 초기화 도구 (2개 DB)
-1. kis_data_collection.db - KIS API 데이터 수집
-2. snapshot_admin.db - 성능/포지션 관리
-"""
-
-import sqlite3
-from pathlib import Path
-from datetime import datetime
-
-DB1_PATH = "src/quant_engine/kis_data_collection.db"
-DB2_PATH = "src/quant_engine/snapshot_admin.db"
-
-def create_kis_db():
- """kis_data_collection.db: KIS API 데이터 스키마"""
- conn = sqlite3.connect(DB1_PATH)
- cursor = conn.cursor()
-
- cursor.execute("""
- CREATE TABLE IF NOT EXISTS data_feed (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- ticker TEXT NOT NULL,
- name TEXT,
- close_price REAL,
- entry_price REAL,
- quantity INTEGER,
- stop_price REAL,
- target_price REAL,
- entry_stage TEXT,
- account TEXT,
- entry_date TEXT,
- updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- velocity_1d REAL,
- velocity_5d REAL,
- ma20 REAL,
- atr20 REAL,
- rsi_14 REAL,
- volume INTEGER,
- avg_trade_value_5d REAL,
- sector TEXT,
- beta REAL,
- UNIQUE(ticker, entry_date)
- )
- """)
-
- cursor.execute("CREATE INDEX IF NOT EXISTS idx_data_feed_ticker ON data_feed(ticker)")
- cursor.execute("CREATE INDEX IF NOT EXISTS idx_data_feed_entry_date ON data_feed(entry_date)")
-
- conn.commit()
- conn.close()
- print(f"[OK] kis_data_collection.db: data_feed 테이블 생성")
-
-def create_snapshot_admin_db():
- """snapshot_admin.db: 성능/포지션 스키마"""
- conn = sqlite3.connect(DB2_PATH)
- cursor = conn.cursor()
-
- cursor.execute("""
- CREATE TABLE IF NOT EXISTS performance (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- ticker TEXT NOT NULL,
- name TEXT,
- entry_date TEXT NOT NULL,
- entry_price REAL NOT NULL,
- quantity INTEGER,
- stop_price REAL,
- target_price REAL,
- exit_date TEXT,
- current_price REAL,
- pnl_pct REAL,
- status TEXT,
- t20_milestone TEXT,
- entry_stage TEXT,
- account TEXT,
- recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- UNIQUE(ticker, entry_date)
- )
- """)
-
- cursor.execute("""
- CREATE TABLE IF NOT EXISTS positions (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- ticker TEXT NOT NULL UNIQUE,
- name TEXT,
- quantity INTEGER,
- entry_price REAL,
- current_price REAL,
- average_cost REAL,
- sector TEXT,
- weight_pct REAL,
- updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
- )
- """)
-
- cursor.execute("CREATE INDEX IF NOT EXISTS idx_performance_ticker ON performance(ticker)")
- cursor.execute("CREATE INDEX IF NOT EXISTS idx_performance_entry_date ON performance(entry_date)")
- cursor.execute("CREATE INDEX IF NOT EXISTS idx_positions_ticker ON positions(ticker)")
-
- conn.commit()
- conn.close()
- print(f"[OK] snapshot_admin.db: performance, positions 테이블 생성")
-
-def verify():
- """검증"""
- print("\n" + "="*80)
- print("데이터베이스 구조 확인")
- print("="*80)
-
- for db_path, db_name in [(DB1_PATH, "kis_data_collection.db"), (DB2_PATH, "snapshot_admin.db")]:
- conn = sqlite3.connect(db_path)
- cursor = conn.cursor()
-
- cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
- tables = [row[0] for row in cursor.fetchall()]
-
- file_size = Path(db_path).stat().st_size / 1024
- print(f"\n[{db_name}]")
- print(f" 크기: {file_size:.2f} KB")
- print(f" 테이블: {', '.join(tables)}")
-
- for table in tables:
- if table != 'sqlite_sequence':
- cursor.execute(f"SELECT COUNT(*) FROM {table}")
- row_count = cursor.fetchone()[0]
- cursor.execute(f"PRAGMA table_info({table})")
- col_count = len(cursor.fetchall())
- print(f" - {table}: {col_count}개 컬럼, {row_count}개 행")
-
- conn.close()
-
-if __name__ == "__main__":
- print("2개 데이터베이스 초기화 중...\n")
-
- create_kis_db()
- create_snapshot_admin_db()
-
- verify()
-
- print("\n[완료] 2개 DB 초기화 완료!")
- print(f" 1. kis_data_collection.db (KIS API)")
- print(f" 2. snapshot_admin.db (성능/포지션)")