From 977167a0dbaab3d4c7d1eb05401a04bab59382a5 Mon Sep 17 00:00:00 2001 From: kjh2064 Date: Thu, 30 Jul 2026 14:12:09 +0900 Subject: [PATCH] chore: remove 34 orphaned one-off scripts from tools/ Systematic scan (not manual per-file review): for every *.py under tools/ and src/quant_engine/, searched the entire tracked repo for its filename stem outside itself (code, docs, spec, CI workflows, tests). 81 of 619 files had zero references anywhere. Deleting all 81 in one pass was judged too risky - some could be intentional manual-run utilities that just don't get named by other files - so this commit only removes the two highest-confidence tiers, 34 files: - Superseded DB init/fix/load scripts, most with an explicit "_v2"/ "_correct"/"_properly" sibling that replaced them (e.g. load_from_xlsx.py -> load_from_xlsx_correct.py, initialize_database.py -> initialize_database_v2.py). Verified via `git log` that both members of each pair were last touched the same day (2026-06-23) and never again - abandoned mid-iteration, not live tooling. - Ad-hoc test/debug scripts sitting in tools/ instead of tests/ (test_api_components.py, test_build_ui_state*.py, diagnose_api_error.py, verify_*.py, validate_json_conversion.py) - one-off interactive debugging session artifacts. Re-ran the full-repo reference search after deletion (zero hits) and the quick-mode release DAG + WBS validator (both still pass) to confirm nothing references these files. The other 47 orphan candidates (WBS- ticket-tagged one-offs and misc build_*/validate_* scripts) are left in place pending further review - some may be intentional manual utilities the automated reference scan can't distinguish from dead code. Co-Authored-By: Claude Sonnet 5 --- tools/archive_legacy_databases.py | 206 ----------- tools/check_schema.py | 27 -- tools/diagnose_api_error.py | 95 ----- tools/fix_account_snapshot_schema.py | 118 ------ tools/fix_account_snapshot_v2.py | 110 ------ tools/fix_settings_schema.py | 89 ----- tools/fix_settings_schema_v2.py | 83 ----- tools/init_performance_tables.py | 87 ----- tools/initialize_database.py | 134 ------- tools/initialize_database_v2.py | 154 -------- tools/initialize_snapshot_admin_db.py | 153 -------- tools/load_all_trading_data.py | 183 --------- tools/load_complete_trading_data.py | 189 ---------- tools/load_from_xlsx.py | 153 -------- tools/load_from_xlsx_correct.py | 212 ----------- tools/load_kis_sample_data_v1.py | 306 --------------- tools/load_settings_config.py | 74 ---- tools/load_settings_properly.py | 78 ---- tools/normalize_formula_registry_v2.py | 48 --- tools/refactor_database_structure.py | 196 ---------- tools/test_api_components.py | 62 ---- tools/test_api_settings_save.py | 57 --- tools/test_build_ui_state.py | 131 ------- tools/test_build_ui_state_direct.py | 58 --- tools/test_build_ui_state_simple.py | 93 ----- tools/test_import.py | 12 - tools/test_remaining_apis.py | 79 ---- tools/test_ui_completeness.py | 490 ------------------------- tools/test_ui_with_details.py | 99 ----- tools/update_db_paths.py | 117 ------ tools/validate_json_conversion.py | 59 --- tools/verify_data_load.py | 55 --- tools/verify_sheet_to_table_sync.py | 232 ------------ tools/verify_table_coverage.py | 104 ------ 34 files changed, 4343 deletions(-) delete mode 100644 tools/archive_legacy_databases.py delete mode 100644 tools/check_schema.py delete mode 100644 tools/diagnose_api_error.py delete mode 100644 tools/fix_account_snapshot_schema.py delete mode 100644 tools/fix_account_snapshot_v2.py delete mode 100644 tools/fix_settings_schema.py delete mode 100644 tools/fix_settings_schema_v2.py delete mode 100644 tools/init_performance_tables.py delete mode 100644 tools/initialize_database.py delete mode 100644 tools/initialize_database_v2.py delete mode 100644 tools/initialize_snapshot_admin_db.py delete mode 100644 tools/load_all_trading_data.py delete mode 100644 tools/load_complete_trading_data.py delete mode 100644 tools/load_from_xlsx.py delete mode 100644 tools/load_from_xlsx_correct.py delete mode 100644 tools/load_kis_sample_data_v1.py delete mode 100644 tools/load_settings_config.py delete mode 100644 tools/load_settings_properly.py delete mode 100644 tools/normalize_formula_registry_v2.py delete mode 100644 tools/refactor_database_structure.py delete mode 100644 tools/test_api_components.py delete mode 100644 tools/test_api_settings_save.py delete mode 100644 tools/test_build_ui_state.py delete mode 100644 tools/test_build_ui_state_direct.py delete mode 100644 tools/test_build_ui_state_simple.py delete mode 100644 tools/test_import.py delete mode 100644 tools/test_remaining_apis.py delete mode 100644 tools/test_ui_completeness.py delete mode 100644 tools/test_ui_with_details.py delete mode 100644 tools/update_db_paths.py delete mode 100644 tools/validate_json_conversion.py delete mode 100644 tools/verify_data_load.py delete mode 100644 tools/verify_sheet_to_table_sync.py delete mode 100644 tools/verify_table_coverage.py diff --git a/tools/archive_legacy_databases.py b/tools/archive_legacy_databases.py deleted file mode 100644 index 7804c989..00000000 --- a/tools/archive_legacy_databases.py +++ /dev/null @@ -1,206 +0,0 @@ -#!/usr/bin/env python3 -""" -Database archive helper (migration/archive only). - -This tool exists to copy legacy or transient DB files into archive_db/ and -generate a manifest. It is not an operational source-of-truth manager. -""" - -import shutil -import json -from pathlib import Path -from datetime import datetime -from typing import Dict, List - -class DatabaseArchiver: - """Legacy DB archive helper.""" - - def __init__(self): - self.root = Path(".") - self.archive_root = self.root / "archive_db" - self.timestamp = datetime.now().strftime("%Y-%m-%d") - self.results = { - "timestamp": datetime.now().isoformat(), - "archived": [], - "skipped": [], - "errors": [] - } - - def create_archive_structure(self) -> None: - """아카이브 디렉토리 구조 생성""" - dirs = [ - self.archive_root / f"{self.timestamp}_outputs_kis_data_collection", - self.archive_root / f"{self.timestamp}_outputs_snapshot_admin", - self.archive_root / f"{self.timestamp}_temp_test_files", - ] - - for d in dirs: - d.mkdir(parents=True, exist_ok=True) - print(f"[OK] Created: {d.relative_to(self.root)}") - - def archive_outputs_kis_data_collection(self) -> None: - """Archive legacy outputs/kis_data_collection/ contents.""" - src = self.root / "outputs" / "kis_data_collection" - if not src.exists(): - print(f"[SKIP] {src.relative_to(self.root)} not found") - self.results["skipped"].append(str(src.relative_to(self.root))) - return - - dest = self.archive_root / f"{self.timestamp}_outputs_kis_data_collection" / "kis_data_collection" - - try: - shutil.copytree(src, dest, dirs_exist_ok=True) - print(f"[OK] Archived: {src.relative_to(self.root)}") - self.results["archived"].append({ - "source": str(src.relative_to(self.root)), - "destination": str(dest.relative_to(self.root)), - "type": "directory", - "timestamp": self.timestamp - }) - except Exception as e: - print(f"[ERROR] Failed to archive {src}: {e}") - self.results["errors"].append(str(e)) - - def archive_outputs_snapshot_admin(self) -> None: - """Archive legacy outputs/snapshot_admin/ smoke*.db files.""" - src_dir = self.root / "outputs" / "snapshot_admin" - if not src_dir.exists(): - print(f"[SKIP] {src_dir.relative_to(self.root)} not found") - self.results["skipped"].append(str(src_dir.relative_to(self.root))) - return - - dest_dir = self.archive_root / f"{self.timestamp}_outputs_snapshot_admin" - - # smoke*.db 파일들 찾기 - smoke_files = list(src_dir.glob("smoke*.db")) - if not smoke_files: - print(f"[SKIP] No smoke*.db files in {src_dir.relative_to(self.root)}") - return - - for src_file in smoke_files: - try: - dest_file = dest_dir / src_file.name - shutil.copy2(src_file, dest_file) - print(f"[OK] Archived: {src_file.relative_to(self.root)}") - self.results["archived"].append({ - "source": str(src_file.relative_to(self.root)), - "destination": str(dest_file.relative_to(self.root)), - "type": "file", - "size_kb": src_file.stat().st_size / 1024, - "timestamp": self.timestamp - }) - except Exception as e: - print(f"[ERROR] Failed to archive {src_file}: {e}") - self.results["errors"].append(str(e)) - - def archive_temp_files(self) -> None: - """Archive transient Temp/ test DB files.""" - temp_dir = self.root / "Temp" - if not temp_dir.exists(): - print(f"[SKIP] {temp_dir.relative_to(self.root)} not found") - self.results["skipped"].append(str(temp_dir.relative_to(self.root))) - return - - dest_dir = self.archive_root / f"{self.timestamp}_temp_test_files" - - patterns = [ - "*_collection.db", - "*_admin*.db", - ] - - files_archived = 0 - for pattern in patterns: - for src_file in temp_dir.glob(pattern): - # snapshot_admin.db와 kis_data_collection.db 제외 (canonical 파일들) - if src_file.name in ["kis_data_collection.db", "snapshot_admin.db"]: - continue - - try: - dest_file = dest_dir / src_file.name - shutil.copy2(src_file, dest_file) - print(f"[OK] Archived: {src_file.relative_to(self.root)}") - self.results["archived"].append({ - "source": str(src_file.relative_to(self.root)), - "destination": str(dest_file.relative_to(self.root)), - "type": "file", - "size_kb": src_file.stat().st_size / 1024, - "timestamp": self.timestamp - }) - files_archived += 1 - except Exception as e: - print(f"[ERROR] Failed to archive {src_file}: {e}") - self.results["errors"].append(str(e)) - - if files_archived == 0: - print(f"[SKIP] No test DB files found in {temp_dir.relative_to(self.root)}") - - def create_manifest(self) -> None: - """Create archive manifest.json.""" - manifest = { - "archive_date": self.timestamp, - "created_at": datetime.now().isoformat(), - "archived_count": len(self.results["archived"]), - "skipped_count": len(self.results["skipped"]), - "error_count": len(self.results["errors"]), - "files": self.results["archived"], - "notes": [ - "These files were archived due to database consolidation.", - "Single source of truth is now: src/quant_engine/", - "To restore: use archive_db/{date}_*/ directories", - "Canonical files: kis_data_collection.db, snapshot_admin.db" - ] - } - - manifest_file = self.archive_root / "manifest.json" - with open(manifest_file, 'w', encoding='utf-8') as f: - json.dump(manifest, f, indent=2, ensure_ascii=False) - - print(f"\n[OK] Manifest created: {manifest_file.relative_to(self.root)}") - - def run(self) -> Dict: - """전체 실행""" - print("="*80) - print("Database Archiving Process") - print("="*80) - print(f"Archive date: {self.timestamp}\n") - - # 아카이브 디렉토리 구조 생성 - self.create_archive_structure() - - print("\n[Archiving files...]") - # 각 레거시 파일 아카이빙 - self.archive_outputs_kis_data_collection() - self.archive_outputs_snapshot_admin() - self.archive_temp_files() - - # manifest 생성 - self.create_manifest() - - # 요약 - print("\n" + "="*80) - print("Archive Summary") - print("="*80) - print(f"Archived: {len(self.results['archived'])} items") - print(f"Skipped: {len(self.results['skipped'])} items") - print(f"Errors: {len(self.results['errors'])} items") - print(f"\nArchive location: {self.archive_root.relative_to(self.root)}") - - if self.results['errors']: - print("\n[Errors encountered]") - for error in self.results['errors']: - print(f" - {error}") - - return self.results - -if __name__ == "__main__": - archiver = DatabaseArchiver() - results = archiver.run() - - print("\n" + "="*80) - print("[Next Steps]") - print("="*80) - print("1. Verify archive contents: git status") - print("2. Add archive to git: git add archive_db/") - print("3. Commit: git commit -m 'Archive legacy database files'") - print("4. Delete legacy files (after verification)") - print("5. Update code references to use src/quant_engine/") diff --git a/tools/check_schema.py b/tools/check_schema.py deleted file mode 100644 index 468c94bd..00000000 --- a/tools/check_schema.py +++ /dev/null @@ -1,27 +0,0 @@ -#!/usr/bin/env python3 -import sqlite3 -from pathlib import Path - -db_path = Path('src/quant_engine/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()] -print(f"전체 테이블: {tables}\n") - -# 각 테이블 스키마 -for table_name in ['account_snapshot', 'snapshot', 'settings', 'performance', 'positions']: - try: - cursor.execute(f"PRAGMA table_info({table_name})") - cols = cursor.fetchall() - if cols: - print(f"{table_name} 컬럼:") - for col in cols: - print(f" {col[1]} ({col[2]})") - print() - except: - pass - -conn.close() diff --git a/tools/diagnose_api_error.py b/tools/diagnose_api_error.py deleted file mode 100644 index e89d0413..00000000 --- a/tools/diagnose_api_error.py +++ /dev/null @@ -1,95 +0,0 @@ -#!/usr/bin/env python3 -""" -/api/settings/save 500 에러 진단 -replace_settings 함수 직접 테스트 -""" - -import sys -sys.path.insert(0, 'src/quant_engine') - -from snapshot_admin_store_v1 import ( - open_connection, - replace_settings, - load_settings_rows, - validate_settings_rows, -) -from pathlib import Path - -def diagnose_settings_save(): - """Settings 저장 함수 직접 테스트""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - - print("="*80) - print("Settings 저장 함수 진단") - print("="*80) - - # 테스트 데이터 - test_rows = [ - { - "ordinal": 5, - "key": "total_asset_krw", - "value": "450000000", - "note": "테스트 수정" - } - ] - - print("\n[1단계] 검증 테스트") - try: - errors = validate_settings_rows(test_rows) - if errors: - print(f" [FAIL] 검증 오류: {errors}") - return - print(f" [OK] 검증 통과") - except Exception as e: - print(f" [ERROR] 검증 함수 실패: {e}") - return - - print("\n[2단계] replace_settings 함수 테스트") - try: - with open_connection(db_path) as conn: - replace_settings(conn, test_rows) - print(f" [OK] replace_settings 성공") - except Exception as e: - print(f" [FAIL] replace_settings 오류") - print(f" 오류 타입: {type(e).__name__}") - print(f" 오류 메시지: {e}") - import traceback - traceback.print_exc() - return - - print("\n[3단계] 저장 결과 확인") - try: - with open_connection(db_path) as conn: - rows = load_settings_rows_from_conn(conn) - for row in rows: - if row['key'] == 'total_asset_krw': - print(f" [OK] {row['key']} = {row['value']}") - except Exception as e: - print(f" [ERROR] 조회 실패: {e}") - - print("\n[완료] 진단 끝") - -# Helper 함수 -def load_settings_rows_from_conn(conn): - """직접 로드""" - import sqlite3 - import json - - rows = conn.execute( - "SELECT ordinal, key, value_json, note, updated_at FROM settings ORDER BY ordinal ASC" - ).fetchall() - - return [ - { - "ordinal": int(row[0]), - "key": row[1], - "value": json.loads(row[2]), - "note": row[3], - "updated_at": row[4], - } - for row in rows - ] - -if __name__ == "__main__": - diagnose_settings_save() diff --git a/tools/fix_account_snapshot_schema.py b/tools/fix_account_snapshot_schema.py deleted file mode 100644 index 9542f608..00000000 --- a/tools/fix_account_snapshot_schema.py +++ /dev/null @@ -1,118 +0,0 @@ -#!/usr/bin/env python3 -""" -account_snapshot 테이블 스키마 수정 -last_updated -> updated_at -""" - -import sqlite3 -from pathlib import Path - -def fix_account_snapshot_schema(): - """account_snapshot 테이블 스키마 수정""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - cursor = conn.cursor() - - # 현재 컬럼 확인 - cursor.execute("PRAGMA table_info(account_snapshot)") - current_cols = {col[1]: col[2] for col in cursor.fetchall()} - print(f"현재 컬럼: {list(current_cols.keys())}") - - # 데이터 백업 - cursor.execute("SELECT * FROM account_snapshot LIMIT 1") - sample = cursor.fetchone() - print(f"\n샘플 행 컬럼: {sample.keys() if sample else 'NO DATA'}") - - # account_snapshot 데이터 백업 - cursor.execute("SELECT * FROM account_snapshot") - backup = cursor.fetchall() - print(f"백업 행 수: {len(backup)}") - - # 기존 테이블 삭제 - cursor.execute("DROP TABLE IF EXISTS account_snapshot") - - # 올바른 스키마로 생성 - cursor.execute(""" - CREATE TABLE account_snapshot ( - 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 - ) - """) - - cursor.execute("CREATE INDEX IF NOT EXISTS idx_account_snapshot_captured_at ON account_snapshot(captured_at)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_account_snapshot_ticker ON account_snapshot(ticker)") - - print("\n새 스키마 생성:") - print(" ordinal INTEGER NOT NULL") - print(" row_json TEXT NOT NULL") - print(" captured_at TEXT NOT NULL DEFAULT ''") - print(" account TEXT NOT NULL DEFAULT ''") - print(" account_type TEXT NOT NULL DEFAULT ''") - print(" ticker TEXT NOT NULL DEFAULT ''") - print(" name TEXT NOT NULL DEFAULT ''") - print(" parse_status TEXT NOT NULL DEFAULT ''") - print(" user_confirmed TEXT NOT NULL DEFAULT ''") - print(" updated_at TEXT NOT NULL") - - # 데이터 복원 - if backup: - print(f"\n데이터 복원 중: {len(backup)}개 행") - - for row_dict in backup: - # row_dict는 sqlite3.Row 타입 - values = [] - for col_name in [ - 'ordinal', 'row_json', 'captured_at', 'account', 'account_type', - 'ticker', 'name', 'parse_status', 'user_confirmed' - ]: - if col_name in row_dict.keys(): - values.append(row_dict[col_name]) - else: - values.append(None) - - # updated_at: last_updated 또는 captured_at 사용 - if 'last_updated' in row_dict.keys() and row_dict['last_updated']: - values.append(str(row_dict['last_updated'])) - elif 'captured_at' in row_dict.keys() and row_dict['captured_at']: - values.append(str(row_dict['captured_at'])) - else: - values.append('') - - cursor.execute(""" - INSERT INTO account_snapshot ( - ordinal, row_json, captured_at, account, account_type, ticker, name, - parse_status, user_confirmed, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, values) - - conn.commit() - print(f"[OK] {len(backup)}개 행 복원") - else: - conn.commit() - print("[OK] 테이블 생성 (데이터 없음)") - - # 검증 - cursor.execute("SELECT COUNT(*) FROM account_snapshot") - count = cursor.fetchone()[0] - print(f"\n검증: {count}개 행") - - cursor.execute("PRAGMA table_info(account_snapshot)") - new_cols = {col[1]: col[2] for col in cursor.fetchall()} - print(f"새 컬럼: {list(new_cols.keys())}") - - conn.close() - - print("\n[OK] account_snapshot 테이블 스키마 수정 완료") - -if __name__ == "__main__": - fix_account_snapshot_schema() diff --git a/tools/fix_account_snapshot_v2.py b/tools/fix_account_snapshot_v2.py deleted file mode 100644 index 78b8d259..00000000 --- a/tools/fix_account_snapshot_v2.py +++ /dev/null @@ -1,110 +0,0 @@ -#!/usr/bin/env python3 -""" -account_snapshot 테이블을 올바른 스키마로 마이그레이션 -현재: captured_at, account, ticker, ... (XLSX 스키마) -목표: ordinal (PK), row_json, captured_at, account, ... , updated_at -""" - -import sqlite3 -import json -from pathlib import Path -from datetime import datetime - -def fix_account_snapshot_v2(): - """account_snapshot 테이블 마이그레이션""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - cursor = conn.cursor() - - # 현재 데이터 백업 - cursor.execute("SELECT * FROM account_snapshot") - old_rows = cursor.fetchall() - print(f"현재 account_snapshot: {len(old_rows)}개 행") - - # 기존 컬럼 확인 - cursor.execute("PRAGMA table_info(account_snapshot)") - old_cols = {col[1]: col[2] for col in cursor.fetchall()} - print(f"현재 컬럼: {len(old_cols)}개") - - # 기존 테이블 백업 - cursor.execute("ALTER TABLE account_snapshot RENAME TO account_snapshot_old") - - # 올바른 스키마로 생성 - cursor.execute(""" - CREATE TABLE account_snapshot ( - ordinal INTEGER PRIMARY KEY, - 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 - ) - """) - - cursor.execute("CREATE INDEX IF NOT EXISTS idx_account_snapshot_captured_at ON account_snapshot(captured_at)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_account_snapshot_ticker ON account_snapshot(ticker)") - - print("\n새 스키마 생성:") - print(" ordinal INTEGER PRIMARY KEY") - print(" row_json TEXT NOT NULL") - print(" captured_at TEXT NOT NULL DEFAULT ''") - print(" ... (8개 핵심 컬럼) ...") - print(" updated_at TEXT NOT NULL") - - # 데이터 마이그레이션 - timestamp = datetime.now().isoformat() - - print(f"\n데이터 마이그레이션 중: {len(old_rows)}개 행") - - for ordinal, old_row in enumerate(old_rows, start=1): - # sqlite3.Row를 dict로 변환 - row_dict = dict(old_row) - - # 필요한 필드 추출 - captured_at = str(row_dict.get('captured_at') or '') - account = str(row_dict.get('account') or '') - account_type = str(row_dict.get('account_type') or '') - ticker = str(row_dict.get('ticker') or '') - name = str(row_dict.get('name') or '') - parse_status = str(row_dict.get('parse_status') or '') - user_confirmed = str(row_dict.get('user_confirmed') or '') - - # 전체 행을 row_json으로 저장 - row_json = json.dumps(row_dict, default=str, ensure_ascii=False) - - cursor.execute(""" - INSERT INTO account_snapshot ( - ordinal, row_json, captured_at, account, account_type, ticker, name, - parse_status, user_confirmed, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, ( - ordinal, row_json, captured_at, account, account_type, ticker, name, - parse_status, user_confirmed, timestamp - )) - - conn.commit() - - # 검증 - cursor.execute("SELECT COUNT(*) FROM account_snapshot") - count = cursor.fetchone()[0] - print(f"마이그레이션된 account_snapshot: {count}개 행") - - cursor.execute("PRAGMA table_info(account_snapshot)") - new_cols = {col[1]: col[2] for col in cursor.fetchall()} - print(f"새 컬럼: {list(new_cols.keys())}") - - # 이전 테이블 삭제 - cursor.execute("DROP TABLE account_snapshot_old") - - conn.close() - - print(f"\n[OK] account_snapshot 테이블 마이그레이션 완료") - -if __name__ == "__main__": - fix_account_snapshot_v2() diff --git a/tools/fix_settings_schema.py b/tools/fix_settings_schema.py deleted file mode 100644 index 2e470785..00000000 --- a/tools/fix_settings_schema.py +++ /dev/null @@ -1,89 +0,0 @@ -#!/usr/bin/env python3 -""" -settings 테이블 스키마 수정 -올바른 스키마: ordinal, key, value_json, note, updated_at -""" - -import sqlite3 -import json -from pathlib import Path -from datetime import datetime - -def fix_settings_schema(): - """settings 테이블 스키마 수정""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 현재 settings 데이터 백업 - cursor.execute("SELECT * FROM settings") - current_rows = cursor.fetchall() - - print(f"현재 settings: {len(current_rows)}개 행") - - # 현재 컬럼 확인 - cursor.execute("PRAGMA table_info(settings)") - current_cols = cursor.fetchall() - print(f"현재 컬럼: {[col[1] for col in current_cols]}") - - # 기존 테이블 삭제 - cursor.execute("DROP TABLE IF EXISTS settings") - - # 올바른 스키마로 생성 - cursor.execute(""" - CREATE TABLE settings ( - ordinal INTEGER PRIMARY KEY, - key TEXT NOT NULL, - value_json TEXT NOT NULL, - note TEXT DEFAULT '', - updated_at TEXT NOT NULL - ) - """) - - print("\n새 스키마 생성:") - print(" ordinal INTEGER PRIMARY KEY") - print(" key TEXT NOT NULL") - print(" value_json TEXT NOT NULL") - print(" note TEXT DEFAULT ''") - print(" updated_at TEXT NOT NULL") - - # 데이터 복원 - # 현재 컬럼: [key, value] - timestamp = datetime.now().isoformat() - - inserted = 0 - for ordinal, (key, value) in enumerate(current_rows, start=1): - # value를 JSON으로 변환 - try: - value_json = json.dumps(str(value), ensure_ascii=False) - except: - value_json = json.dumps("", ensure_ascii=False) - - cursor.execute( - """INSERT INTO settings (ordinal, key, value_json, note, updated_at) - VALUES (?, ?, ?, ?, ?)""", - (ordinal, key, value_json, "", timestamp) - ) - inserted += 1 - - conn.commit() - - # 검증 - cursor.execute("SELECT COUNT(*) FROM settings") - count = cursor.fetchone()[0] - print(f"\n복원된 settings: {count}개 행") - - # 샘플 확인 - cursor.execute("SELECT ordinal, key, value_json FROM settings LIMIT 3") - print("\n샘플 데이터:") - for ordinal, key, value_json in cursor.fetchall(): - value = json.loads(value_json) - print(f" {ordinal}. {key} = {value}") - - conn.close() - - print("\n[OK] settings 테이블 스키마 수정 완료") - -if __name__ == "__main__": - fix_settings_schema() diff --git a/tools/fix_settings_schema_v2.py b/tools/fix_settings_schema_v2.py deleted file mode 100644 index 41838b54..00000000 --- a/tools/fix_settings_schema_v2.py +++ /dev/null @@ -1,83 +0,0 @@ -#!/usr/bin/env python3 -""" -settings 테이블을 올바른 스키마로 수정 -현재: key, value -목표: ordinal (PK), key (NOT NULL), value_json (JSON), note, updated_at -""" - -import sqlite3 -import json -from pathlib import Path -from datetime import datetime - -def fix_settings_schema_v2(): - """settings 테이블 스키마 수정""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - cursor = conn.cursor() - - # 현재 데이터 백업 - cursor.execute("SELECT key, value FROM settings") - old_rows = cursor.fetchall() - print(f"현재 settings: {len(old_rows)}개 행") - - # 기존 테이블 삭제 - cursor.execute("DROP TABLE IF EXISTS settings") - - # 올바른 스키마로 생성 - cursor.execute(""" - CREATE TABLE settings ( - ordinal INTEGER PRIMARY KEY, - key TEXT NOT NULL, - value_json TEXT NOT NULL, - note TEXT DEFAULT '', - updated_at TEXT NOT NULL - ) - """) - - print("새 스키마 생성:") - print(" ordinal INTEGER PRIMARY KEY") - print(" key TEXT NOT NULL") - print(" value_json TEXT NOT NULL") - print(" note TEXT DEFAULT ''") - print(" updated_at TEXT NOT NULL") - - # 데이터 복원 - timestamp = datetime.now().isoformat() - - for ordinal, row in enumerate(old_rows, start=1): - key = row['key'] - value = row['value'] - - # value를 JSON으로 변환 - try: - value_json = json.dumps(str(value), ensure_ascii=False) - except: - value_json = json.dumps("", ensure_ascii=False) - - cursor.execute(""" - INSERT INTO settings (ordinal, key, value_json, note, updated_at) - VALUES (?, ?, ?, ?, ?) - """, (ordinal, key, value_json, "", timestamp)) - - conn.commit() - - # 검증 - cursor.execute("SELECT COUNT(*) FROM settings") - count = cursor.fetchone()[0] - print(f"\n복원된 settings: {count}개 행") - - cursor.execute("SELECT ordinal, key, value_json FROM settings LIMIT 3") - print("샘플 데이터:") - for ordinal, key, value_json in cursor.fetchall(): - value = json.loads(value_json) - print(f" {ordinal}. {key} = {value}") - - conn.close() - - print(f"\n[OK] settings 테이블 스키마 수정 완료") - -if __name__ == "__main__": - fix_settings_schema_v2() diff --git a/tools/init_performance_tables.py b/tools/init_performance_tables.py deleted file mode 100644 index da176e07..00000000 --- a/tools/init_performance_tables.py +++ /dev/null @@ -1,87 +0,0 @@ -#!/usr/bin/env python3 -""" -performance, positions 테이블 초기 데이터 생성 -T+20 모니터링 활성화 -""" - -import sqlite3 -from pathlib import Path -from datetime import datetime - -def init_performance_tables(): - """초기 성과 데이터 생성""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 1. performance 테이블에 샘플 거래 기록 추가 - sample_trades = [ - ('005930', 'Samsung Electronics', '2026-06-01', 70000, 100, 70000, 72000, 2.86, 'COMPLETED', 'T+20'), - ('000660', 'SK Hynix', '2026-06-05', 120000, 50, 120000, 121500, 1.25, 'ACTIVE', None), - ('035420', 'NAVER', '2026-06-10', 385000, 10, 385000, 390000, 1.30, 'COMPLETED', 'T+20'), - ] - - print(f"performance 초기화: {len(sample_trades)}개 거래 추가") - for ticker, name, entry_date, entry_price, qty, _, current_price, pnl_pct, status, t20_milestone in sample_trades: - cursor.execute(""" - INSERT INTO performance - (ticker, name, entry_date, entry_price, quantity, current_price, pnl_pct, status, t20_milestone) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, ( - ticker, - name, - entry_date, - entry_price, - qty, - current_price, - pnl_pct, - status, - t20_milestone - )) - - # 2. positions 테이블에 현재 포지션 추가 - sample_positions = [ - ('005930', 'Samsung Electronics', 100, 70000, 72000, 70500, 'IT'), - ('000660', 'SK Hynix', 50, 120000, 121500, 120750, 'IT'), - ('035420', 'NAVER', 10, 385000, 390000, 387500, 'Internet'), - ] - - print(f"positions 초기화: {len(sample_positions)}개 포지션 추가") - - today = datetime.now().isoformat() - for ticker, name, quantity, entry_price, current_price, avg_cost, sector in sample_positions: - cursor.execute(""" - INSERT INTO positions - (ticker, name, quantity, entry_price, current_price, average_cost, sector, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """, ( - ticker, - name, - quantity, - entry_price, - current_price, - avg_cost, - sector, - today - )) - - conn.commit() - - # 검증 - cursor.execute("SELECT COUNT(*) FROM performance") - perf_count = cursor.fetchone()[0] - - cursor.execute("SELECT COUNT(*) FROM positions") - pos_count = cursor.fetchone()[0] - - print(f"\n[결과]") - print(f" performance: {perf_count}개 행") - print(f" positions: {pos_count}개 행") - - conn.close() - - print(f"\n[OK] T+20 모니터링 초기화 완료") - -if __name__ == "__main__": - init_performance_tables() diff --git a/tools/initialize_database.py b/tools/initialize_database.py deleted file mode 100644 index 4df41eea..00000000 --- a/tools/initialize_database.py +++ /dev/null @@ -1,134 +0,0 @@ -#!/usr/bin/env python3 -""" -데이터베이스 초기화 도구 -data_feed, performance, positions 테이블 생성 -""" - -import sqlite3 -from pathlib import Path -from datetime import datetime - -DB_PATH = "src/quant_engine/data_feed.db" - -def create_tables(): - """필수 테이블 생성""" - conn = sqlite3.connect(DB_PATH) - cursor = conn.cursor() - - # data_feed 테이블 - 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) - ) - """) - - # performance 테이블 - 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) - ) - """) - - # positions 테이블 - 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_data_feed_ticker ON data_feed(ticker)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_data_feed_entry_date ON data_feed(entry_date)") - 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] 데이터베이스 초기화 완료: {DB_PATH}") - print(f" 생성된 테이블: data_feed, performance, positions") - -def verify_database(): - """데이터베이스 검증""" - 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()] - - print(f"\n[검증] 테이블 목록: {', '.join(tables)}") - print(f"[검증] 파일 크기: {Path(DB_PATH).stat().st_size / 1024:.2f} KB") - - # 각 테이블 구조 확인 - for table in tables: - cursor.execute(f"PRAGMA table_info({table})") - columns = cursor.fetchall() - print(f"\n{table} 컬럼:") - for col in columns: - print(f" - {col[1]} ({col[2]})") - - conn.close() - -if __name__ == "__main__": - # 기존 DB 백업 - db_file = Path(DB_PATH) - if db_file.exists(): - backup_path = f"{DB_PATH}.backup.{datetime.now().strftime('%Y%m%d_%H%M%S')}" - db_file.rename(backup_path) - print(f"[OK] 기존 데이터베이스 백업: {backup_path}") - - # 새 DB 생성 - create_tables() - - # 검증 - verify_database() - - print("\n[완료] 데이터베이스 초기화 완료!") diff --git a/tools/initialize_database_v2.py b/tools/initialize_database_v2.py deleted file mode 100644 index 91c5e19d..00000000 --- a/tools/initialize_database_v2.py +++ /dev/null @@ -1,154 +0,0 @@ -#!/usr/bin/env python3 -""" -데이터베이스 초기화 도구 (2개 DB 분리) -- test_kis_data_collection.db: data_feed 테이블 (KIS API 데이터) -- snapshot_admin_livecheck.db: performance, positions 테이블 (라이브 체크) -""" - -import sqlite3 -from pathlib import Path -from datetime import datetime - -DB1_PATH = "src/quant_engine/test_kis_data_collection.db" -DB2_PATH = "src/quant_engine/snapshot_admin_livecheck.db" - -def create_db1_schema(): - """DB1: KIS 데이터 수집 스키마""" - conn = sqlite3.connect(DB1_PATH) - cursor = conn.cursor() - - # data_feed 테이블 - 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] DB1 생성: {DB1_PATH}") - print(f" 테이블: data_feed (KIS API 데이터 수집용)") - -def create_db2_schema(): - """DB2: snapshot_admin 라이브 체크 스키마""" - conn = sqlite3.connect(DB2_PATH) - cursor = conn.cursor() - - # performance 테이블 - 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) - ) - """) - - # positions 테이블 - 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] DB2 생성: {DB2_PATH}") - print(f" 테이블: performance, positions (snapshot_admin 라이브용)") - -def verify_databases(): - """DB 검증""" - print("\n" + "="*80) - print("데이터베이스 검증") - print("="*80) - - for db_path in [DB1_PATH, DB2_PATH]: - 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[{Path(db_path).name}]") - print(f" 크기: {file_size:.2f} KB") - print(f" 테이블: {', '.join(tables)}") - - for table in tables: - cursor.execute(f"PRAGMA table_info({table})") - col_count = len(cursor.fetchall()) - print(f" - {table}: {col_count}개 컬럼") - - conn.close() - -if __name__ == "__main__": - # 기존 DB 백업 - for db_path in [DB1_PATH, DB2_PATH]: - db_file = Path(db_path) - if db_file.exists(): - backup_path = f"{db_path}.backup.{datetime.now().strftime('%Y%m%d_%H%M%S')}" - db_file.rename(backup_path) - print(f"[OK] 백업: {backup_path}") - - print("\n새 데이터베이스 생성 중...\n") - - # 새 DB 생성 - create_db1_schema() - create_db2_schema() - - # 검증 - verify_databases() - - print("\n[완료] 2개 DB 초기화 완료!") diff --git a/tools/initialize_snapshot_admin_db.py b/tools/initialize_snapshot_admin_db.py deleted file mode 100644 index 8923e1aa..00000000 --- a/tools/initialize_snapshot_admin_db.py +++ /dev/null @@ -1,153 +0,0 @@ -#!/usr/bin/env python3 -""" -snapshot_admin.db를 올바른 스키마와 XLSX 데이터로 초기화 -""" - -import sqlite3 -import json -import pandas as pd -from pathlib import Path -from datetime import datetime - -def initialize_snapshot_admin_db(): - """snapshot_admin.db 초기화""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - xlsx_file = Path('GatherTradingData.xlsx') - json_file = Path('GatherTradingData.json') - - print("="*80) - print("snapshot_admin.db 초기화") - print("="*80) - - # JSON 메타데이터 로드 - with open(json_file, encoding='utf-8') as f: - metadata = json.load(f).get('metadata', {}) - - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 1. settings 테이블 초기화 - print("\n[1] settings 테이블 초기화") - cursor.execute("DROP TABLE IF EXISTS settings") - cursor.execute(""" - CREATE TABLE settings ( - ordinal INTEGER PRIMARY KEY, - key TEXT NOT NULL, - value_json TEXT NOT NULL, - note TEXT DEFAULT '', - updated_at TEXT NOT NULL - ) - """) - - # XLSX에서 settings 데이터 로드 - df_settings = pd.read_excel(xlsx_file, sheet_name='settings', header=None) - # 처음 2개 컬럼만 사용 (key, value) - df_settings = df_settings.iloc[:, :2] - df_settings.columns = ['key', 'value'] - timestamp = datetime.now().isoformat() - - print(f" {len(df_settings)}개 설정 로드 중...") - for ordinal, (idx, row) in enumerate(df_settings.iterrows(), start=1): - key = str(row['key']) - value = str(row['value']) - value_json = json.dumps(value, ensure_ascii=False) - - cursor.execute(""" - INSERT INTO settings (ordinal, key, value_json, note, updated_at) - VALUES (?, ?, ?, ?, ?) - """, (ordinal, key, value_json, "", timestamp)) - - cursor.execute("SELECT COUNT(*) FROM settings") - count = cursor.fetchone()[0] - print(f" [OK] {count}개 설정 로드") - - # 2. account_snapshot 테이블 초기화 - print("\n[2] account_snapshot 테이블 초기화") - cursor.execute("DROP TABLE IF EXISTS account_snapshot") - cursor.execute(""" - CREATE TABLE account_snapshot ( - ordinal INTEGER PRIMARY KEY, - 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 - ) - """) - - # XLSX에서 account_snapshot 데이터 로드 - df_snapshot = pd.read_excel(xlsx_file, sheet_name='account_snapshot', header=1) - - print(f" {len(df_snapshot)}개 스냅샷 로드 중...") - for ordinal, (idx, row) in enumerate(df_snapshot.iterrows(), start=1): - row_dict = row.to_dict() - - # row_json으로 저장 - row_json = json.dumps(row_dict, default=str, ensure_ascii=False) - - # 핵심 필드 추출 - captured_at = str(row_dict.get('captured_at', '')) - account = str(row_dict.get('account', '')) - account_type = str(row_dict.get('account_type', '')) - ticker = str(row_dict.get('ticker', '')) - name = str(row_dict.get('name', '')) - parse_status = str(row_dict.get('parse_status', '')) - user_confirmed = str(row_dict.get('user_confirmed', '')) - - cursor.execute(""" - INSERT INTO account_snapshot ( - ordinal, row_json, captured_at, account, account_type, ticker, name, - parse_status, user_confirmed, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, ( - ordinal, row_json, captured_at, account, account_type, ticker, name, - parse_status, user_confirmed, timestamp - )) - - cursor.execute("SELECT COUNT(*) FROM account_snapshot") - count = cursor.fetchone()[0] - print(f" [OK] {count}개 스냅샷 로드") - - # 3. 인덱스 생성 - print("\n[3] 인덱스 생성") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_account_snapshot_captured_at ON account_snapshot(captured_at)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_account_snapshot_ticker ON account_snapshot(ticker)") - print(f" [OK] 인덱스 생성") - - conn.commit() - conn.close() - - # 검증 - print("\n[검증]") - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - cursor.execute("SELECT COUNT(*) FROM settings") - settings_count = cursor.fetchone()[0] - - cursor.execute("SELECT COUNT(*) FROM account_snapshot") - snapshot_count = cursor.fetchone()[0] - - print(f" settings: {settings_count}개") - print(f" account_snapshot: {snapshot_count}개") - - # 스키마 확인 - cursor.execute("PRAGMA table_info(settings)") - settings_cols = [col[1] for col in cursor.fetchall()] - print(f" settings 컬럼: {settings_cols}") - - cursor.execute("PRAGMA table_info(account_snapshot)") - snapshot_cols = [col[1] for col in cursor.fetchall()] - print(f" account_snapshot 컬럼: {snapshot_cols[:5]}... ({len(snapshot_cols)} 개)") - - conn.close() - - print("\n[OK] snapshot_admin.db 초기화 완료") - -if __name__ == "__main__": - initialize_snapshot_admin_db() diff --git a/tools/load_all_trading_data.py b/tools/load_all_trading_data.py deleted file mode 100644 index b4531a5c..00000000 --- a/tools/load_all_trading_data.py +++ /dev/null @@ -1,183 +0,0 @@ -#!/usr/bin/env python3 -""" -GatherTradingData.json 전체 데이터를 SQLite에 로드 -""" - -import json -import sqlite3 -from pathlib import Path -from datetime import datetime - -class GatherTradingDataLoader: - """전체 거래 데이터 로더""" - - def __init__(self): - self.json_file = Path('GatherTradingData.json') - self.kis_db = Path('src/quant_engine/kis_data_collection.db') - self.snapshot_db = Path('src/quant_engine/snapshot_admin.db') - self.results = { - "timestamp": datetime.now().isoformat(), - "tables_loaded": {}, - "errors": [] - } - - def load_json_data(self) -> tuple: - """JSON 로드 - (metadata, data) 반환""" - try: - with open(self.json_file, encoding='utf-8') as f: - full_data = json.load(f) - metadata = full_data.get('metadata', {}) - data = full_data.get('data', {}) - return metadata, data - except: - with open(self.json_file, encoding='euc-kr') as f: - full_data = json.load(f) - metadata = full_data.get('metadata', {}) - data = full_data.get('data', {}) - return metadata, data - - def infer_column_types(self, data: list) -> dict: - """컬럼 타입 추론""" - if not data: - return {} - - first_row = data[0] - types = {} - - for col, val in first_row.items(): - if val is None: - types[col] = "TEXT" - elif isinstance(val, bool): - types[col] = "INTEGER" - elif isinstance(val, int): - types[col] = "INTEGER" - elif isinstance(val, float): - types[col] = "REAL" - else: - types[col] = "TEXT" - - return types - - def create_and_load_table(self, db_path: Path, table_name: str, data: list) -> dict: - """테이블 생성 및 데이터 로드""" - if not data: - return {"table": table_name, "status": "EMPTY", "rows": 0} - - try: - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 컬럼 타입 추론 - column_types = self.infer_column_types(data) - columns = list(column_types.keys()) - - # CREATE TABLE - col_defs = ", ".join([f"{col} {column_types[col]}" for col in columns]) - cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute(f"CREATE TABLE {table_name} ({col_defs})") - - # INSERT 데이터 - placeholders = ", ".join(["?" for _ in columns]) - insert_sql = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})" - - for row in data: - values = [row.get(col) for col in columns] - cursor.execute(insert_sql, values) - - conn.commit() - conn.close() - - return { - "table": table_name, - "status": "SUCCESS", - "rows": len(data), - "columns": len(columns) - } - - except Exception as e: - return { - "table": table_name, - "status": "ERROR", - "error": str(e) - } - - def run(self) -> dict: - """전체 실행""" - print("="*80) - print("GatherTradingData.json 전체 로드") - print("="*80) - - # JSON 로드 - metadata, data = self.load_json_data() - sheets = metadata.get('sheets_included', []) - - print(f"\n[발견된 시트] {len(sheets)}개") - for sheet in sheets: - print(f" - {sheet}") - - # 각 시트를 테이블로 로드 - print("\n[로드 중...]") - for sheet_name in sheets: - sheet_data = data.get(sheet_name, []) - - if not sheet_data: - print(f" [{sheet_name}] SKIP (empty)") - continue - - # 타겟 DB 결정 - # kis_data_collection.db: data_feed만 - # snapshot_admin.db: settings, account_snapshot, 그 외 모든 것 - if sheet_name == 'data_feed': - db_path = self.kis_db - else: - db_path = self.snapshot_db - - # 테이블 생성 - result = self.create_and_load_table(db_path, sheet_name, sheet_data) - - if result['status'] == 'SUCCESS': - print(f" [{sheet_name}] OK ({result['rows']} rows, {result['columns']} cols)") - self.results["tables_loaded"][sheet_name] = result - else: - print(f" [{sheet_name}] FAIL: {result.get('error', 'unknown')}") - self.results["errors"].append(sheet_name) - - # 최종 검증 - print("\n[최종 검증]") - for db_name, db_path in [("kis_data_collection", self.kis_db), ("snapshot_admin", self.snapshot_db)]: - if not db_path.exists(): - continue - - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name != 'sqlite_sequence'") - tables = [row[0] for row in cursor.fetchall()] - conn.close() - - print(f" {db_name}.db: {len(tables)} 테이블") - for table in tables: - cursor = sqlite3.connect(db_path).cursor() - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - print(f" - {table}: {count} rows") - - self.results["summary"] = { - "total_sheets": len(sheets), - "loaded_sheets": len(self.results["tables_loaded"]), - "failed_sheets": len(self.results["errors"]), - "coverage_pct": (len(self.results["tables_loaded"]) / len(sheets) * 100) if sheets else 0 - } - - print(f"\n[결과]") - print(f" 커버리지: {self.results['summary']['coverage_pct']:.1f}%") - print(f" 로드됨: {self.results['summary']['loaded_sheets']}/{self.results['summary']['total_sheets']}") - - return self.results - -if __name__ == "__main__": - loader = GatherTradingDataLoader() - result = loader.run() - - print(f"\n[완료] GatherTradingData.json → DB 로드 완료") - print(f"파일 크기: kis_data_collection.db = {Path('src/quant_engine/kis_data_collection.db').stat().st_size/1024:.1f}KB") - print(f"파일 크기: snapshot_admin.db = {Path('src/quant_engine/snapshot_admin.db').stat().st_size/1024:.1f}KB") diff --git a/tools/load_complete_trading_data.py b/tools/load_complete_trading_data.py deleted file mode 100644 index 63321673..00000000 --- a/tools/load_complete_trading_data.py +++ /dev/null @@ -1,189 +0,0 @@ -#!/usr/bin/env python3 -""" -GatherTradingData.json 완전 로드 (모든 23개 시트) -""" - -import json -import sqlite3 -from pathlib import Path -from datetime import datetime - -class CompleteDataLoader: - """전체 거래 데이터 완전 로더""" - - def __init__(self): - self.json_file = Path('GatherTradingData.json') - self.kis_db = Path('src/quant_engine/kis_data_collection.db') - self.snapshot_db = Path('src/quant_engine/snapshot_admin.db') - self.results = { - "timestamp": datetime.now().isoformat(), - "tables_loaded": {}, - "errors": [] - } - - def load_json_data(self) -> tuple: - """JSON 로드""" - try: - with open(self.json_file, encoding='utf-8') as f: - full_data = json.load(f) - metadata = full_data.get('metadata', {}) - data = full_data.get('data', {}) - return metadata, data - except: - with open(self.json_file, encoding='euc-kr') as f: - full_data = json.load(f) - metadata = full_data.get('metadata', {}) - data = full_data.get('data', {}) - return metadata, data - - def infer_column_types(self, data: list) -> dict: - """컬럼 타입 추론""" - if not data: - return {} - - first_row = data[0] - types = {} - - for col, val in first_row.items(): - if val is None: - types[col] = "TEXT" - elif isinstance(val, bool): - types[col] = "INTEGER" - elif isinstance(val, int): - types[col] = "INTEGER" - elif isinstance(val, float): - types[col] = "REAL" - else: - types[col] = "TEXT" - - return types - - def create_and_load_table(self, db_path: Path, table_name: str, sheet_data: list) -> dict: - """테이블 생성 및 데이터 로드""" - if not sheet_data: - return {"table": table_name, "status": "SKIP", "rows": 0} - - try: - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 컬럼 타입 추론 - column_types = self.infer_column_types(sheet_data) - columns = list(column_types.keys()) - - # 기존 테이블 삭제 및 생성 - cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - col_defs = ", ".join([f"{col} {column_types[col]}" for col in columns]) - cursor.execute(f"CREATE TABLE {table_name} ({col_defs})") - - # INSERT 데이터 - placeholders = ", ".join(["?" for _ in columns]) - insert_sql = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})" - - for row in sheet_data: - values = [row.get(col) for col in columns] - cursor.execute(insert_sql, values) - - conn.commit() - conn.close() - - return { - "table": table_name, - "status": "SUCCESS", - "rows": len(sheet_data), - "columns": len(columns), - "db": str(db_path) - } - - except Exception as e: - return { - "table": table_name, - "status": "ERROR", - "error": str(e), - "db": str(db_path) - } - - def run(self) -> dict: - """전체 실행""" - print("="*80) - print("GatherTradingData.json 완전 로드 (모든 시트)") - print("="*80) - - # JSON 로드 - metadata, data = self.load_json_data() - - print(f"\n[JSON에서 발견된 시트] {len(data)}개") - for sheet_name in sorted(data.keys()): - print(f" - {sheet_name}: {len(data[sheet_name])} rows") - - # 각 시트를 테이블로 로드 - print("\n[로드 중...]") - - for sheet_name in sorted(data.keys()): - sheet_data = data[sheet_name] - - if not sheet_data: - print(f" [SKIP] {sheet_name} (empty)") - continue - - # 타겟 DB 결정 - # kis_data_collection.db: data_feed만 - # snapshot_admin.db: 나머지 모두 - if sheet_name == 'data_feed': - db_path = self.kis_db - else: - db_path = self.snapshot_db - - # 테이블 생성 - result = self.create_and_load_table(db_path, sheet_name, sheet_data) - - if result['status'] == 'SUCCESS': - print(f" [OK] {sheet_name}: {result['rows']} rows, {result['columns']} cols") - self.results["tables_loaded"][sheet_name] = result - elif result['status'] == 'SKIP': - print(f" [SKIP] {sheet_name}") - else: - print(f" [FAIL] {sheet_name}: {result.get('error', 'unknown')}") - self.results["errors"].append(sheet_name) - - # 최종 검증 - print("\n[최종 검증]") - for db_name, db_path in [("kis_data_collection", self.kis_db), ("snapshot_admin", self.snapshot_db)]: - if not db_path.exists(): - continue - - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name != 'sqlite_sequence' ORDER BY name") - tables = [row[0] for row in cursor.fetchall()] - conn.close() - - print(f" {db_name}.db: {len(tables)} 테이블") - - total_rows = 0 - for table in tables: - cursor = sqlite3.connect(db_path).cursor() - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - total_rows += count - - print(f" → 총 {total_rows:,} rows") - - self.results["summary"] = { - "total_sheets_in_json": len(data), - "loaded_sheets": len(self.results["tables_loaded"]), - "failed_sheets": len(self.results["errors"]), - "coverage_pct": (len(self.results["tables_loaded"]) / len(data) * 100) if data else 0 - } - - print(f"\n[결과]") - print(f" 로드: {self.results['summary']['loaded_sheets']}/{self.results['summary']['total_sheets_in_json']}") - print(f" 커버리지: {self.results['summary']['coverage_pct']:.1f}%") - - return self.results - -if __name__ == "__main__": - loader = CompleteDataLoader() - result = loader.run() - - print(f"\n[완료] 완전 로드 완료") diff --git a/tools/load_from_xlsx.py b/tools/load_from_xlsx.py deleted file mode 100644 index 060114e7..00000000 --- a/tools/load_from_xlsx.py +++ /dev/null @@ -1,153 +0,0 @@ -#!/usr/bin/env python3 -""" -GatherTradingData.xlsx에서 직접 추출해서 DB에 로드 -""" - -import sqlite3 -from pathlib import Path -from datetime import datetime -import pandas as pd - -class XLSXDataLoader: - """XLSX 직접 로더""" - - def __init__(self): - self.xlsx_file = Path('GatherTradingData.xlsx') - self.kis_db = Path('src/quant_engine/kis_data_collection.db') - self.snapshot_db = Path('src/quant_engine/snapshot_admin.db') - self.results = { - "timestamp": datetime.now().isoformat(), - "sheets_loaded": {}, - "errors": [] - } - - def load_excel_sheets(self) -> dict: - """Excel에서 모든 시트 로드""" - print("[로드 중] Excel 파일 읽기...") - - try: - # 모든 시트 이름 먼저 얻기 - excel_file = pd.ExcelFile(self.xlsx_file) - sheet_names = excel_file.sheet_names - - print(f"발견된 시트: {len(sheet_names)}개") - for i, sheet in enumerate(sheet_names, 1): - print(f" {i}. {sheet}") - - # 각 시트 로드 - sheets_data = {} - for sheet_name in sheet_names: - try: - df = pd.read_excel(self.xlsx_file, sheet_name=sheet_name) - sheets_data[sheet_name] = df - print(f" [OK] {sheet_name}: {len(df)} rows, {len(df.columns)} cols") - except Exception as e: - print(f" [FAIL] {sheet_name}: {str(e)[:50]}") - self.results["errors"].append(sheet_name) - - return sheets_data - - except Exception as e: - print(f"[ERROR] Excel 로드 실패: {e}") - return {} - - def load_to_database(self, sheets_data: dict) -> None: - """데이터를 DB에 로드""" - - print("\n[DB 로드 중...]") - - for sheet_name, df in sheets_data.items(): - if df.empty: - print(f" [SKIP] {sheet_name} (empty)") - continue - - # 타겟 DB 결정 - if sheet_name == 'data_feed': - db_path = self.kis_db - else: - db_path = self.snapshot_db - - try: - # NaN을 None으로 변환 - df = df.where(pd.notna(df), None) - - # DB에 로드 (기존 테이블 교체) - conn = sqlite3.connect(db_path) - df.to_sql(sheet_name, conn, if_exists='replace', index=False) - conn.close() - - print(f" [OK] {sheet_name}: {len(df)} rows loaded to {db_path.name}") - self.results["sheets_loaded"][sheet_name] = { - "rows": len(df), - "cols": len(df.columns), - "db": str(db_path) - } - - except Exception as e: - print(f" [FAIL] {sheet_name}: {str(e)[:80]}") - self.results["errors"].append(sheet_name) - - def verify_load(self) -> None: - """로드 검증""" - print("\n[검증 중...]") - - for db_name, db_path in [("kis_data_collection", self.kis_db), ("snapshot_admin", self.snapshot_db)]: - if not db_path.exists(): - continue - - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name != 'sqlite_sequence' ORDER BY name") - tables = [row[0] for row in cursor.fetchall()] - - total_rows = 0 - for table in tables: - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - total_rows += count - - print(f" {db_name}.db: {len(tables)} 테이블, {total_rows:,} rows") - conn.close() - - def run(self) -> dict: - """전체 실행""" - print("="*80) - print("GatherTradingData.xlsx 직접 로드") - print("="*80) - print() - - # Excel 로드 - sheets_data = self.load_excel_sheets() - - if not sheets_data: - print("[ERROR] 로드된 시트가 없습니다") - return self.results - - # DB 로드 - self.load_to_database(sheets_data) - - # 검증 - self.verify_load() - - self.results["summary"] = { - "total_sheets": len(sheets_data), - "loaded_sheets": len(self.results["sheets_loaded"]), - "failed_sheets": len(self.results["errors"]), - "coverage_pct": (len(self.results["sheets_loaded"]) / len(sheets_data) * 100) if sheets_data else 0 - } - - print("\n[결과 요약]") - print(f" 로드됨: {self.results['summary']['loaded_sheets']}/{self.results['summary']['total_sheets']}") - print(f" 커버리지: {self.results['summary']['coverage_pct']:.1f}%") - - if self.results["errors"]: - print(f" 실패: {', '.join(self.results['errors'][:5])}") - - return self.results - -if __name__ == "__main__": - loader = XLSXDataLoader() - result = loader.run() - - print("\n[완료]") diff --git a/tools/load_from_xlsx_correct.py b/tools/load_from_xlsx_correct.py deleted file mode 100644 index b162a057..00000000 --- a/tools/load_from_xlsx_correct.py +++ /dev/null @@ -1,212 +0,0 @@ -#!/usr/bin/env python3 -""" -GatherTradingData.xlsx 올바르게 로드 (metadata 기반 header 파라미터) - -JSON metadata의 header_row_1based를 사용해서 각 시트마다 올바른 header를 지정 -""" - -import json -import sqlite3 -from pathlib import Path -from datetime import datetime -import pandas as pd - -class CorrectXLSXLoader: - """메타데이터 기반 정확한 XLSX 로더""" - - def __init__(self): - self.json_file = Path('GatherTradingData.json') - self.xlsx_file = Path('GatherTradingData.xlsx') - self.kis_db = Path('src/quant_engine/kis_data_collection.db') - self.snapshot_db = Path('src/quant_engine/snapshot_admin.db') - self.results = { - "timestamp": datetime.now().isoformat(), - "sheets_loaded": {}, - "errors": [] - } - - def load_metadata(self) -> dict: - """JSON 메타데이터 로드""" - with open(self.json_file, encoding='utf-8') as f: - data = json.load(f) - return data.get('metadata', {}) - - def load_excel_sheets(self, metadata: dict) -> dict: - """Excel에서 올바른 header를 사용해서 모든 시트 로드 (account_snapshot 제외)""" - print("[로드 중] Excel 파일 읽기...") - - sheet_headers = metadata.get('sheet_headers', {}) - excel_file = pd.ExcelFile(self.xlsx_file) - sheet_names = excel_file.sheet_names - - print(f"발견된 시트: {len(sheet_names)}개") - - sheets_data = {} - for sheet_name in sheet_names: - # account_snapshot은 건너뛴다 (별도 처리) - if sheet_name == 'account_snapshot': - print(f" [SKIP] {sheet_name} (수동 처리)") - continue - - # metadata에서 header_row_1based 읽기 - header_info = sheet_headers.get(sheet_name, {}) - header_row_1based = header_info.get('header_row_1based', 1) - header_param = header_row_1based - 1 # pandas는 0-indexed - - try: - # settings 특수 처리: 헤더가 없음 (key-value 쌍) - if sheet_name == 'settings': - df = pd.read_excel(self.xlsx_file, sheet_name=sheet_name, header=None) - df.columns = ['key', 'value', 'note1', 'note2'] - df = df[['key', 'value']] # 필요한 컬럼만 - else: - df = pd.read_excel(self.xlsx_file, sheet_name=sheet_name, header=header_param) - - # NaN을 None으로 변환 - df = df.where(pd.notna(df), None) - - sheets_data[sheet_name] = df - print(f" [OK] {sheet_name}: {len(df)} rows, {len(df.columns)} cols (header={header_param})") - - except Exception as e: - print(f" [FAIL] {sheet_name}: {str(e)[:50]}") - self.results["errors"].append(sheet_name) - - return sheets_data - - def load_to_database(self, sheets_data: dict) -> None: - """데이터를 DB에 로드""" - print("\n[DB 로드 중...]") - - for sheet_name, df in sheets_data.items(): - if df.empty: - print(f" [SKIP] {sheet_name} (empty)") - continue - - # 타겟 DB 결정 - if sheet_name == 'data_feed': - db_path = self.kis_db - else: - db_path = self.snapshot_db - - try: - conn = sqlite3.connect(db_path) - - # account_snapshot은 특별하게 처리: 스키마를 보존하면서 데이터만 추가 - if sheet_name == 'account_snapshot': - self._load_account_snapshot(conn, df) - else: - df.to_sql(sheet_name, conn, if_exists='replace', index=False) - - conn.close() - - print(f" [OK] {sheet_name}: {len(df)} rows → {db_path.name}") - self.results["sheets_loaded"][sheet_name] = { - "rows": len(df), - "cols": len(df.columns), - "db": str(db_path) - } - - except Exception as e: - print(f" [FAIL] {sheet_name}: {str(e)[:80]}") - self.results["errors"].append(sheet_name) - - def _load_account_snapshot(self, conn: sqlite3.Connection, df: pd.DataFrame) -> None: - """account_snapshot 데이터를 올바른 스키마로 로드""" - import json - from datetime import datetime - - cursor = conn.cursor() - timestamp = datetime.now().isoformat() - - # 기존 데이터 삭제 (옵션: DELETE 또는 유지) - cursor.execute("DELETE FROM account_snapshot") - - for ordinal, row in enumerate(df.iterrows(), start=1): - idx, series = row - row_dict = series.to_dict() - - # row_json으로 저장 - row_json = json.dumps(row_dict, default=str, ensure_ascii=False) - - # 핵심 필드 추출 - captured_at = str(row_dict.get('captured_at', '')) - account = str(row_dict.get('account', '')) - account_type = str(row_dict.get('account_type', '')) - ticker = str(row_dict.get('ticker', '')) - name = str(row_dict.get('name', '')) - parse_status = str(row_dict.get('parse_status', '')) - user_confirmed = str(row_dict.get('user_confirmed', '')) - - cursor.execute(""" - INSERT INTO account_snapshot ( - ordinal, row_json, captured_at, account, account_type, ticker, name, - parse_status, user_confirmed, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, ( - ordinal, row_json, captured_at, account, account_type, ticker, name, - parse_status, user_confirmed, timestamp - )) - - conn.commit() - - def verify(self) -> None: - """로드 검증""" - print("\n[검증 중...]") - - for db_name, db_path in [("kis_data_collection", self.kis_db), ("snapshot_admin", self.snapshot_db)]: - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name != 'sqlite_sequence'") - tables = [row[0] for row in cursor.fetchall()] - - total_rows = 0 - for table in tables: - cursor.execute(f"SELECT COUNT(*) FROM {table}") - total_rows += cursor.fetchone()[0] - - print(f" {db_name}.db: {len(tables)} 테이블, {total_rows:,} rows") - conn.close() - - def run(self) -> dict: - """전체 실행""" - print("="*80) - print("GatherTradingData.xlsx 정확하게 로드 (메타데이터 기반)") - print("="*80) - print() - - # 메타데이터 로드 - metadata = self.load_metadata() - - # Excel 로드 - sheets_data = self.load_excel_sheets(metadata) - - if not sheets_data: - print("[ERROR] 로드된 시트가 없습니다") - return self.results - - # DB 로드 - self.load_to_database(sheets_data) - - # 검증 - self.verify() - - self.results["summary"] = { - "total_sheets": len(sheets_data), - "loaded_sheets": len(self.results["sheets_loaded"]), - "failed_sheets": len(self.results["errors"]), - "coverage_pct": (len(self.results["sheets_loaded"]) / len(sheets_data) * 100) if sheets_data else 0 - } - - print("\n[결과 요약]") - print(f" 로드됨: {self.results['summary']['loaded_sheets']}/{self.results['summary']['total_sheets']}") - print(f" 커버리지: {self.results['summary']['coverage_pct']:.1f}%") - - return self.results - -if __name__ == "__main__": - loader = CorrectXLSXLoader() - result = loader.run() - - print("\n[완료] 정확한 XLSX 로드 완료") diff --git a/tools/load_kis_sample_data_v1.py b/tools/load_kis_sample_data_v1.py deleted file mode 100644 index 29263e8a..00000000 --- a/tools/load_kis_sample_data_v1.py +++ /dev/null @@ -1,306 +0,0 @@ -#!/usr/bin/env python3 -""" -KIS 데이터 수집 DB 로드 도구 - -GatherTradingData.json의 data_feed 시트 데이터를 -kis_data_collection.db에 로드 -""" - -import json -import sqlite3 -from pathlib import Path -from datetime import datetime -from typing import Dict, List -import sys - -class KISSampleDataLoader: - """KIS 샘플 데이터 로더""" - - def __init__(self, json_file: str, db_path: str): - self.json_file = Path(json_file) - self.db_path = Path(db_path) - self.results = { - "timestamp": datetime.now().isoformat(), - "loaded_records": 0, - "errors": 0, - "tickers": set() - } - - def load_json_data(self) -> Dict: - """GatherTradingData.json 로드""" - if not self.json_file.exists(): - print(f"[ERROR] JSON file not found: {self.json_file}") - return {} - - try: - with open(self.json_file, encoding='utf-8') as f: - data = json.load(f) - return data - except UnicodeDecodeError: - # 다른 인코딩 시도 - with open(self.json_file, encoding='euc-kr') as f: - data = json.load(f) - return data - - def extract_data_feed_sheet(self, data: Dict) -> List[Dict]: - """data_feed 시트 추출""" - # GatherTradingData.json의 구조 확인 필요 - # 일반적으로 sheet별 데이터가 포함됨 - - # 샘플: data_feed 시트가 최상위 키일 수 있음 - if "data_feed" in data: - return data["data_feed"] - - # 또는 sheets 내에 있을 수 있음 - if "sheets" in data and "data_feed" in data["sheets"]: - return data["sheets"]["data_feed"] - - # 또는 data 내에 있을 수 있음 - if "data" in data and "data_feed" in data["data"]: - return data["data"]["data_feed"] - - print("[WARNING] data_feed sheet not found in JSON") - return [] - - def create_sample_data(self) -> List[Dict]: - """테스트용 샘플 데이터 생성""" - today = datetime.now().strftime("%Y-%m-%d") - - sample_records = [ - { - "ticker": "005930", - "name": "삼성전자", - "close_price": 70500.0, - "entry_price": 69000.0, - "quantity": 10, - "stop_price": 65000.0, - "target_price": 75000.0, - "entry_stage": "1st", - "account": "main", - "entry_date": today, - "velocity_1d": 2.17, - "velocity_5d": 1.85, - "ma20": 68500.0, - "atr20": 1500.0, - "rsi_14": 65.2, - "volume": 15000000, - "avg_trade_value_5d": 850000000000, - "sector": "반도체", - "beta": 1.2 - }, - { - "ticker": "000660", - "name": "SK하이닉스", - "close_price": 175000.0, - "entry_price": 170000.0, - "quantity": 5, - "stop_price": 162000.0, - "target_price": 190000.0, - "entry_stage": "1st", - "account": "main", - "entry_date": today, - "velocity_1d": 2.94, - "velocity_5d": 2.15, - "ma20": 172000.0, - "atr20": 3500.0, - "rsi_14": 72.1, - "volume": 8000000, - "avg_trade_value_5d": 1400000000000, - "sector": "반도체", - "beta": 1.35 - }, - { - "ticker": "035420", - "name": "NAVER", - "close_price": 435000.0, - "entry_price": 420000.0, - "quantity": 3, - "stop_price": 400000.0, - "target_price": 480000.0, - "entry_stage": "2nd", - "account": "main", - "entry_date": today, - "velocity_1d": 3.57, - "velocity_5d": 2.62, - "ma20": 425000.0, - "atr20": 8000.0, - "rsi_14": 78.5, - "volume": 2500000, - "avg_trade_value_5d": 1090000000000, - "sector": "IT", - "beta": 1.08 - }, - { - "ticker": "051910", - "name": "LG화학", - "close_price": 455000.0, - "entry_price": 440000.0, - "quantity": 2, - "stop_price": 415000.0, - "target_price": 500000.0, - "entry_stage": "2nd", - "account": "main", - "entry_date": today, - "velocity_1d": 3.41, - "velocity_5d": 2.27, - "ma20": 442000.0, - "atr20": 9000.0, - "rsi_14": 75.3, - "volume": 1800000, - "avg_trade_value_5d": 820000000000, - "sector": "화학", - "beta": 0.95 - }, - { - "ticker": "373220", - "name": "LG에너지솔루션", - "close_price": 385000.0, - "entry_price": 370000.0, - "quantity": 3, - "stop_price": 350000.0, - "target_price": 420000.0, - "entry_stage": "1st", - "account": "main", - "entry_date": today, - "velocity_1d": 4.05, - "velocity_5d": 3.15, - "ma20": 372000.0, - "atr20": 7500.0, - "rsi_14": 81.2, - "volume": 3500000, - "avg_trade_value_5d": 1350000000000, - "sector": "전지", - "beta": 1.42 - } - ] - - return sample_records - - def load_into_db(self, records: List[Dict]) -> int: - """DB에 레코드 로드""" - if not records: - print("[WARNING] No records to load") - return 0 - - conn = sqlite3.connect(self.db_path) - cursor = conn.cursor() - - loaded = 0 - errors = 0 - - for record in records: - try: - cursor.execute(""" - INSERT INTO data_feed ( - ticker, name, close_price, entry_price, quantity, - stop_price, target_price, entry_stage, account, - entry_date, velocity_1d, velocity_5d, ma20, atr20, - rsi_14, volume, avg_trade_value_5d, sector, beta - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, ( - record.get("ticker"), - record.get("name"), - record.get("close_price"), - record.get("entry_price"), - record.get("quantity"), - record.get("stop_price"), - record.get("target_price"), - record.get("entry_stage"), - record.get("account"), - record.get("entry_date"), - record.get("velocity_1d"), - record.get("velocity_5d"), - record.get("ma20"), - record.get("atr20"), - record.get("rsi_14"), - record.get("volume"), - record.get("avg_trade_value_5d"), - record.get("sector"), - record.get("beta") - )) - loaded += 1 - self.results["tickers"].add(record.get("ticker")) - - except Exception as e: - errors += 1 - print(f"[ERROR] Failed to load {record.get('ticker')}: {e}") - - conn.commit() - conn.close() - - self.results["loaded_records"] = loaded - self.results["errors"] = errors - - return loaded - - def verify_data(self) -> Dict: - """로드된 데이터 검증""" - conn = sqlite3.connect(self.db_path) - cursor = conn.cursor() - - cursor.execute("SELECT COUNT(*) FROM data_feed") - total = cursor.fetchone()[0] - - cursor.execute(""" - SELECT ticker, name, close_price, entry_date - FROM data_feed - ORDER BY entry_date DESC - LIMIT 5 - """) - samples = cursor.fetchall() - - conn.close() - - return { - "total_records": total, - "sample_records": [ - { - "ticker": s[0], - "name": s[1], - "close_price": s[2], - "entry_date": s[3] - } - for s in samples - ] - } - - def run(self) -> Dict: - """전체 실행""" - print("KIS Sample Data Loader") - print("="*80) - - # 항상 샘플 데이터 사용 - # (JSON 파싱은 나중에 별도 도구로 처리) - print("[OK] Using KIS sample data (real market snapshot)") - records = self.create_sample_data() - - print(f"[OK] {len(records)} records to load") - - # DB에 로드 - loaded = self.load_into_db(records) - print(f"[OK] Loaded {loaded} records") - - # 검증 - verification = self.verify_data() - print(f"\n[검증]") - print(f" 총 레코드: {verification['total_records']}") - print(f" 보유 종목:") - for sample in verification['sample_records']: - print(f" - {sample['ticker']} ({sample['name']}): {sample['close_price']} KRW @ {sample['entry_date']}") - - self.results["verification"] = verification - self.results["tickers"] = list(self.results["tickers"]) - - return self.results - -if __name__ == "__main__": - loader = KISSampleDataLoader( - json_file="GatherTradingData.json", - db_path="src/quant_engine/kis_data_collection.db" - ) - result = loader.run() - - print("\n" + "="*80) - print(f"[완료] {result['loaded_records']}개 레코드 로드") - print(f" 에러: {result['errors']}") - print(f" 종목 수: {len(result['tickers'])}") diff --git a/tools/load_settings_config.py b/tools/load_settings_config.py deleted file mode 100644 index 7f20d2da..00000000 --- a/tools/load_settings_config.py +++ /dev/null @@ -1,74 +0,0 @@ -#!/usr/bin/env python3 -""" -settings를 dict → list로 변환해서 로드 -""" - -import json -import sqlite3 -from pathlib import Path - -def load_settings_to_db(): - """settings를 DB에 로드""" - - # JSON에서 settings 로드 - with open('GatherTradingData.json', encoding='utf-8') as f: - data = json.load(f) - - settings_dict = data['data']['settings'] - print(f"settings 타입: {type(settings_dict)}") - print(f"settings 항목: {len(settings_dict)}") - - # dict → list로 변환 - settings_list = [] - for ordinal, (key, value) in enumerate(settings_dict.items(), start=1): - settings_list.append({ - "ordinal": ordinal, - "key": key, - "value": value, - "note": "" - }) - - print(f"\n변환된 settings list: {len(settings_list)} 행") - print(f"첫 항목: {settings_list[0]}") - - # DB에 로드 - db_path = Path('src/quant_engine/snapshot_admin.db') - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 기존 settings 삭제 - cursor.execute("DROP TABLE IF EXISTS settings") - - # 테이블 생성 - cursor.execute(""" - CREATE TABLE settings ( - ordinal INTEGER, - key TEXT, - value TEXT, - note TEXT - ) - """) - - # 데이터 삽입 - for row in settings_list: - cursor.execute( - "INSERT INTO settings (ordinal, key, value, note) VALUES (?, ?, ?, ?)", - (row['ordinal'], row['key'], row['value'], row['note']) - ) - - conn.commit() - - # 검증 - cursor.execute("SELECT COUNT(*) FROM settings") - count = cursor.fetchone()[0] - print(f"\n[OK] settings 로드 완료: {count} rows") - - # 샘플 보기 - cursor.execute("SELECT * FROM settings LIMIT 5") - for row in cursor.fetchall(): - print(f" {row}") - - conn.close() - -if __name__ == "__main__": - load_settings_to_db() diff --git a/tools/load_settings_properly.py b/tools/load_settings_properly.py deleted file mode 100644 index ded5c0fe..00000000 --- a/tools/load_settings_properly.py +++ /dev/null @@ -1,78 +0,0 @@ -#!/usr/bin/env python3 -""" -settings를 올바르게 로드 (key-value 구조) -""" - -import sqlite3 -from pathlib import Path -import pandas as pd - -def load_settings_correctly(): - """settings를 올바르게 로드""" - - # XLSX에서 settings 로드 (헤더 없이) - df = pd.read_excel('GatherTradingData.xlsx', sheet_name='settings', header=None) - - print("settings 원본 데이터:") - print(f" Shape: {df.shape}") - print(f" Row 0: {df.iloc[0, 0]} = {df.iloc[0, 1]}") - - # Column 0: key, Column 1: value, Column 2: note - settings_list = [] - for idx, row in df.iterrows(): - key = str(row[0]) if pd.notna(row[0]) else "" - value = str(row[1]) if pd.notna(row[1]) else "" - note = str(row[2]) if pd.notna(row[2]) else "" - - if key: - settings_list.append({ - "ordinal": idx + 1, - "key": key, - "value": value, - "note": note - }) - - print(f"\n변환된 settings: {len(settings_list)} 행") - print(f"첫 항목: {settings_list[0]}") - - # DB에 로드 - db_path = Path('src/quant_engine/snapshot_admin.db') - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 기존 설정 스키마와 맞추기 - # snapshot_admin_store_v1.py에서 기대하는 스키마 확인 - cursor.execute("DROP TABLE IF EXISTS settings") - cursor.execute(""" - CREATE TABLE settings ( - ordinal INTEGER PRIMARY KEY, - key TEXT, - value TEXT, - note TEXT - ) - """) - - # 데이터 삽입 - for row in settings_list: - cursor.execute( - "INSERT INTO settings (ordinal, key, value, note) VALUES (?, ?, ?, ?)", - (row['ordinal'], row['key'], row['value'], row['note']) - ) - - conn.commit() - - # 검증 - cursor.execute("SELECT COUNT(*) FROM settings") - count = cursor.fetchone()[0] - print(f"\n[OK] settings 로드 완료: {count} rows") - - # 샘플 출력 - print("\n샘플 데이터:") - cursor.execute("SELECT ordinal, key, value FROM settings LIMIT 5") - for ordinal, key, value in cursor.fetchall(): - print(f" {ordinal}. {key} = {value}") - - conn.close() - -if __name__ == "__main__": - load_settings_correctly() diff --git a/tools/normalize_formula_registry_v2.py b/tools/normalize_formula_registry_v2.py deleted file mode 100644 index 163b90dc..00000000 --- a/tools/normalize_formula_registry_v2.py +++ /dev/null @@ -1,48 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import yaml -from pathlib import Path - -ROOT = Path(__file__).resolve().parents[1] - -def main() -> int: - ap = argparse.ArgumentParser() - ap.add_argument("--in", dest="input_file", default="spec/13_formula_registry.yaml") - ap.add_argument("--out", dest="output_file", default="spec/03_formulas/formula_registry.normalized.yaml") - args = ap.parse_args() - - in_path = ROOT / args.input_file - out_path = ROOT / args.output_file - - if not in_path.exists(): - print(f"Input registry file not found: {in_path}") - return 1 - - try: - data = yaml.safe_load(in_path.read_text(encoding="utf-8")) - except Exception as e: - print(f"Error parsing input YAML: {e}") - return 1 - - # Simple validation and copying to normalized version - # Each formula should have owner, inputs, outputs, etc. - formulas = data.get("formula_registry", {}).get("formulas", {}) - - # Let's ensure the output structure is clean - normalized = { - "schema_version": "formula_registry.normalized.v2", - "description": "Normalized formula registry for QEDD development framework", - "formulas": formulas - } - - out_path.parent.mkdir(parents=True, exist_ok=True) - out_path.write_text(yaml.safe_dump(normalized, sort_keys=False, allow_unicode=True), encoding="utf-8") - - print(f"Successfully normalized registry: {out_path}") - return 0 - -if __name__ == "__main__": - import sys - sys.exit(main()) diff --git a/tools/refactor_database_structure.py b/tools/refactor_database_structure.py deleted file mode 100644 index 0f5fc3b5..00000000 --- a/tools/refactor_database_structure.py +++ /dev/null @@ -1,196 +0,0 @@ -#!/usr/bin/env python3 -""" -데이터베이스 구조 리팩토링 - -파편화된 DB 파일들을 정리하고 단일 canonical 위치(src/quant_engine/)를 기준으로 통합한다. -""" - -import shutil -from pathlib import Path -from datetime import datetime - -class DatabaseRefactorer: - """데이터베이스 구조 리팩토링""" - - def __init__(self): - self.root = Path(".") - self.canonical_dir = Path("src/quant_engine") - self.results = { - "timestamp": datetime.now().isoformat(), - "consolidated": [], - "moved": [], - "deleted": [], - "errors": [] - } - - def get_canonical_location(self, db_name: str) -> Path: - """DB의 정규 위치 반환""" - canonical_map = { - "kis_data_collection.db": self.canonical_dir / "kis_data_collection.db", - "snapshot_admin.db": self.canonical_dir / "snapshot_admin.db", - } - return canonical_map.get(db_name) - - def find_scattered_dbs(self) -> dict: - """파편화된 DB 파일 찾기""" - scattered = { - "outputs": [], - "temp": [], - "other": [] - } - - # outputs/ 검색 - outputs_dir = self.root / "outputs" - if outputs_dir.exists(): - for db in outputs_dir.rglob("*.db"): - scattered["outputs"].append(db) - - # Temp/ 검색 - temp_dir = self.root / "Temp" - if temp_dir.exists(): - for db in temp_dir.glob("*_collection.db"): - scattered["temp"].append(db) - for db in temp_dir.glob("*_admin*.db"): - scattered["temp"].append(db) - - return scattered - - def analyze(self) -> dict: - """분석""" - scattered = self.find_scattered_dbs() - - analysis = { - "canonical_location": str(self.canonical_dir), - "canonical_files": { - "kis_data_collection.db": (self.canonical_dir / "kis_data_collection.db").exists(), - "snapshot_admin.db": (self.canonical_dir / "snapshot_admin.db").exists(), - }, - "scattered_files": { - "outputs": [str(f.relative_to(self.root)) for f in scattered["outputs"]], - "temp": [str(f.relative_to(self.root)) for f in scattered["temp"]], - }, - "total_scattered": len(scattered["outputs"]) + len(scattered["temp"]) - } - - return analysis, scattered - - def consolidate(self, scattered: dict, dry_run: bool = True) -> dict: - """통합""" - action = "Would consolidate" if dry_run else "Consolidating" - - print(f"\n[Analysis]") - analysis, _ = self.analyze() - - print(f"Canonical location: {analysis['canonical_location']}") - print(f" kis_data_collection.db: {'EXISTS' if analysis['canonical_files']['kis_data_collection.db'] else 'MISSING'}") - print(f" snapshot_admin.db: {'EXISTS' if analysis['canonical_files']['snapshot_admin.db'] else 'MISSING'}") - - print(f"\nScattered files found:") - print(f" outputs/: {len(analysis['scattered_files']['outputs'])} files") - for f in analysis['scattered_files']['outputs'][:5]: - print(f" - {f}") - print(f" Temp/: {len(analysis['scattered_files']['temp'])} files") - for f in analysis['scattered_files']['temp']: - print(f" - {f}") - - print(f"\n[Recommendation]") - print(f"1. Keep canonical location: src/quant_engine/") - print(f" - kis_data_collection.db (KIS API 데이터)") - print(f" - snapshot_admin.db (성능/포지션)") - print(f"") - print(f"2. Archive old files: archive_db/ (2026-06-23)") - print(f" - outputs/kis_data_collection/*") - print(f" - outputs/snapshot_admin/smoke*.db") - print(f" - Temp/*_collection.db") - print(f" - Temp/*_admin*.db") - print(f"") - print(f"3. Delete: qualitative_sell_strategy.db (unrelated)") - - return analysis - - def create_consolidation_plan(self) -> str: - """통합 계획서 작성""" - analysis, _ = self.analyze() - - plan = f""" -# Database Consolidation Plan (2026-06-23) - -## Current State: FRAGMENTED -- Canonical: src/quant_engine/ (2 files) -- Scattered: outputs/ ({len(analysis['scattered_files']['outputs'])}) + Temp/ ({len(analysis['scattered_files']['temp'])}) -- Total: {analysis['total_scattered'] + 2} database files - -## Issue -1. kis_data_collection.db in 3 locations: - - src/quant_engine/ (CANONICAL) - - outputs/kis_data_collection/ - - Temp/test_kis_data_collection.db - -2. snapshot_admin.db in 4+ locations: - - src/quant_engine/ (CANONICAL) - - outputs/snapshot_admin/ - - Temp/snapshot_admin_*.db (multiple variants) - - outputs/qualitative_sell_strategy/ (unrelated) - -## Solution - -### Step 1: Verify Canonical Copies (src/quant_engine/) -- kis_data_collection.db: 5 records [OK] -- snapshot_admin.db: 0 records (initialized) [OK] - -### Step 2: Archive Scattered Files (archive_db/) -Create archive directory with timestamp: -``` -archive_db/ -├── 2026-06-23_outputs_kis_data_collection/ -├── 2026-06-23_outputs_snapshot_admin/ -├── 2026-06-23_temp_test_files/ -└── manifest.json (record what was archived) -``` - -### Step 3: Clean Obsolete References -- Remove imports from legacy non-canonical database paths -- Remove imports from archive/backup database paths -- Update any code expecting these paths - -### Step 4: Update Documentation -- Update all references to use: src/quant_engine/ -- Update deployment docs (Synology) -- Update CI/CD workflows - -## Benefits -- Single source of truth -- Easier backup/recovery -- Clear separation: live vs. archived -- Faster data access -- Simplified deployment - -## Files to Delete (After Archiving) -- archive only genuinely obsolete duplicate DBs -- keep canonical DBs in src/quant_engine/ -- keep Temp/ only for transient validation artifacts -""" - return plan - -def main(): - refactorer = DatabaseRefactorer() - - print("="*80) - print("Database Structure Refactoring Analysis") - print("="*80) - - analysis = refactorer.consolidate(None, dry_run=True) - - plan = refactorer.create_consolidation_plan() - print(plan) - - # 계획서 저장 - plan_file = Path("docs/archive/DATABASE_CONSOLIDATION_PLAN_2026_06_23.md") - plan_file.parent.mkdir(parents=True, exist_ok=True) - with open(plan_file, 'w', encoding='utf-8') as f: - f.write(plan) - - print(f"\n[Saved] Consolidation plan: {plan_file}") - -if __name__ == "__main__": - main() diff --git a/tools/test_api_components.py b/tools/test_api_components.py deleted file mode 100644 index 7281c2ec..00000000 --- a/tools/test_api_components.py +++ /dev/null @@ -1,62 +0,0 @@ -#!/usr/bin/env python3 -""" -API 핸들러의 각 컴포넌트 테스트 -""" - -import sys -sys.path.insert(0, 'src/quant_engine') - -from snapshot_admin_store_v1 import ( - is_locked, - lock_conflicts_for_rows, - summarize_workspace, - open_connection, -) -from pathlib import Path - -db_path = Path('src/quant_engine/snapshot_admin.db') - -print("="*80) -print("API 컴포넌트 테스트") -print("="*80) - -# 1. is_locked 테스트 -print("\n[1] is_locked 테스트") -try: - locked = is_locked(db_path, "settings") - print(f" [OK] is_locked result: {locked}") -except Exception as e: - print(f" [ERROR] {e}") - -# 2. lock_conflicts_for_rows 테스트 -print("\n[2] lock_conflicts_for_rows 테스트") -try: - test_rows = [ - { - "ordinal": 5, - "key": "total_asset_krw", - "value": "450000000", - "note": "test" - } - ] - - conflicts = lock_conflicts_for_rows(db_path, "settings", test_rows) - print(f" [OK] conflicts: {conflicts}") -except Exception as e: - print(f" [ERROR] {e}") - -# 3. summarize_workspace 테스트 -print("\n[3] summarize_workspace 테스트") -try: - import time - start = time.time() - summary = summarize_workspace(db_path) - elapsed = time.time() - start - print(f" [OK] summarize_workspace completed in {elapsed:.2f}s") - print(f" Keys: {list(summary.keys())[:5]}") -except Exception as e: - print(f" [ERROR] {e}") - import traceback - traceback.print_exc() - -print("\n[완료]") diff --git a/tools/test_api_settings_save.py b/tools/test_api_settings_save.py deleted file mode 100644 index 838a3f9b..00000000 --- a/tools/test_api_settings_save.py +++ /dev/null @@ -1,57 +0,0 @@ -#!/usr/bin/env python3 -""" -/api/settings/save 엔드포인트 테스트 -""" - -import requests -import json -import time - -BASE_URL = "http://localhost:5000" - -print("="*80) -print("/api/settings/save 엔드포인트 테스트") -print("="*80) - -# 데이터 준비 -test_data = { - "rows": [ - { - "ordinal": 5, - "key": "total_asset_krw", - "value": "500000000", - "note": "API 테스트 수정" - } - ] -} - -print(f"\n[요청] POST {BASE_URL}/api/settings/save") -print(f"[데이터] {json.dumps(test_data, ensure_ascii=False, indent=2)}") - -try: - start = time.time() - response = requests.post( - f"{BASE_URL}/api/settings/save", - json=test_data, - timeout=10 - ) - elapsed = time.time() - start - - print(f"\n[응답 시간] {elapsed:.2f}s") - print(f"[상태 코드] {response.status_code}") - - if response.status_code == 200: - result = response.json() - print(f"[결과] {json.dumps(result, ensure_ascii=False, indent=2)}") - print(f"\n[OK] /api/settings/save 성공") - else: - print(f"[오류 응답]") - print(f" 상태: {response.status_code}") - print(f" 본문: {response.text[:200]}") - print(f"\n[FAIL] /api/settings/save 실패") - -except Exception as e: - print(f"[ERROR] {e}") - print(f"\n[FAIL] 요청 실패") - -print("\n[완료]") diff --git a/tools/test_build_ui_state.py b/tools/test_build_ui_state.py deleted file mode 100644 index 1fc966e0..00000000 --- a/tools/test_build_ui_state.py +++ /dev/null @@ -1,131 +0,0 @@ -#!/usr/bin/env python3 -""" -build_ui_state 함수의 각 단계 테스트 -""" - -import sys -sys.path.insert(0, '.') - -# Import from package -import importlib.util -spec = importlib.util.spec_from_file_location( - "snapshot_admin_store_v1", - "src/quant_engine/snapshot_admin_store_v1.py" -) -store = importlib.util.module_from_spec(spec) -spec.loader.exec_module(store) - -from snapshot_admin_store_v1 import ( - summarize_workspace, - load_settings_rows, - load_account_snapshot_rows, - validate_settings_rows, - validate_account_snapshot_rows, - load_approval_rows, - load_locks, - load_change_log_rows, -) -from pathlib import Path -import time - -db_path = Path('src/quant_engine/snapshot_admin.db') - -print("="*80) -print("build_ui_state 함수 단계별 테스트") -print("="*80) - -# 1. summarize_workspace -print("\n[1] summarize_workspace") -try: - start = time.time() - result = summarize_workspace(db_path) - elapsed = time.time() - start - print(f" [OK] {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 2. load_settings_rows -print("\n[2] load_settings_rows") -try: - start = time.time() - result = load_settings_rows(db_path) - elapsed = time.time() - start - print(f" [OK] {len(result)} rows, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 3. load_account_snapshot_rows -print("\n[3] load_account_snapshot_rows") -try: - start = time.time() - result = load_account_snapshot_rows(db_path) - elapsed = time.time() - start - print(f" [OK] {len(result)} rows, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 4. validate_settings_rows -print("\n[4] validate_settings_rows") -try: - start = time.time() - settings = load_settings_rows(db_path) - result = validate_settings_rows(settings) - elapsed = time.time() - start - print(f" [OK] {len(result)} errors, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 5. validate_account_snapshot_rows -print("\n[5] validate_account_snapshot_rows") -try: - start = time.time() - snapshot = load_account_snapshot_rows(db_path) - result = validate_account_snapshot_rows(snapshot) - elapsed = time.time() - start - print(f" [OK] {len(result)} errors, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 6. load_approval_rows -print("\n[6] load_approval_rows") -try: - start = time.time() - result = load_approval_rows(db_path) - elapsed = time.time() - start - print(f" [OK] {len(result)} rows, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 7. load_locks -print("\n[7] load_locks") -try: - start = time.time() - result = load_locks(db_path) - elapsed = time.time() - start - print(f" [OK] {len(result)} rows, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 8. load_change_log_rows -print("\n[8] load_change_log_rows") -try: - start = time.time() - result = load_change_log_rows(db_path, limit=12) - elapsed = time.time() - start - print(f" [OK] {len(result)} rows, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - -# 9. Full build_ui_state (at the end) -print("\n[9] build_ui_state (FULL)") -try: - start = time.time() - result = build_ui_state(db_path) - elapsed = time.time() - start - print(f" [OK] keys={len(result)}, {elapsed:.2f}s") -except Exception as e: - print(f" [ERROR] {e}") - import traceback - traceback.print_exc() - -print("\n[완료]") diff --git a/tools/test_build_ui_state_direct.py b/tools/test_build_ui_state_direct.py deleted file mode 100644 index 3b3de9dc..00000000 --- a/tools/test_build_ui_state_direct.py +++ /dev/null @@ -1,58 +0,0 @@ -#!/usr/bin/env python3 -""" -build_ui_state 각 단계 직접 테스트 -""" - -import sys -import os -sys.path.insert(0, os.getcwd()) - -# src.quant_engine 패키지를 임포트할 수 있도록 -import importlib.util -spec = importlib.util.spec_from_file_location( - "snapshot_admin_store_v1", - "src/quant_engine/snapshot_admin_store_v1.py" -) -store = importlib.util.module_from_spec(spec) -sys.modules['snapshot_admin_store_v1'] = store -spec.loader.exec_module(store) - -from pathlib import Path -from datetime import datetime -import time - -db_path = Path('src/quant_engine/snapshot_admin.db') - -print("="*80) -print("build_ui_state 각 단계 테스트") -print("="*80) - -functions_to_test = [ - ('summarize_workspace', lambda: store.summarize_workspace(db_path)), - ('load_settings_rows', lambda: store.load_settings_rows(db_path)), - ('load_account_snapshot_rows', lambda: store.load_account_snapshot_rows(db_path)), - ('validate_settings_rows', lambda: store.validate_settings_rows(store.load_settings_rows(db_path))), - ('validate_account_snapshot_rows', lambda: store.validate_account_snapshot_rows(store.load_account_snapshot_rows(db_path))), - ('load_approval_rows', lambda: store.load_approval_rows(db_path)), - ('load_locks', lambda: store.load_locks(db_path)), - ('load_change_log_rows', lambda: store.load_change_log_rows(db_path, limit=12)), -] - -for name, func in functions_to_test: - try: - start = time.time() - result = func() - elapsed = time.time() - start - if isinstance(result, list): - print(f"[OK] {name}: {len(result)} 항목, {elapsed:.2f}s") - elif isinstance(result, dict): - print(f"[OK] {name}: {len(result)} 키, {elapsed:.2f}s") - else: - print(f"[OK] {name}: {type(result).__name__}, {elapsed:.2f}s") - except Exception as e: - print(f"[ERROR] {name}: {e}") - import traceback - traceback.print_exc() - break - -print("\n[완료]") diff --git a/tools/test_build_ui_state_simple.py b/tools/test_build_ui_state_simple.py deleted file mode 100644 index 8004eaa1..00000000 --- a/tools/test_build_ui_state_simple.py +++ /dev/null @@ -1,93 +0,0 @@ -#!/usr/bin/env python3 -""" -build_ui_state 함수의 각 단계 테스트 - 직접 호출 -""" - -import sys -sys.path.insert(0, '.') - -# Store 함수들을 직접 import -from pathlib import Path -import sqlite3 -import time -import json - -def test_each_function(): - """각 함수를 개별 테스트""" - - db_path = Path('src/quant_engine/snapshot_admin.db') - - print("="*80) - print("Database 함수 단계별 테스트") - print("="*80) - - # 1. 테이블 존재 확인 - print("\n[테이블 확인]") - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") - tables = [row[0] for row in cursor.fetchall()] - print(f" 테이블 수: {len(tables)}") - print(f" 주요 테이블: settings, account_snapshot, workspace_approval_v2") - - # 각 테이블의 행 수 - for table in ['settings', 'account_snapshot', 'workspace_approval_v2', 'workspace_change_log']: - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - print(f" {table}: {count} rows") - - # 2. settings SELECT 테스트 - print("\n[settings SELECT 테스트]") - try: - cursor.execute("SELECT ordinal, key, value_json, note, updated_at FROM settings LIMIT 1") - row = cursor.fetchone() - if row: - print(f" [OK] {len(row)} 컬럼: {row}") - else: - print(f" [OK] 데이터 없음") - except Exception as e: - print(f" [ERROR] {e}") - - # 3. account_snapshot SELECT 테스트 - print("\n[account_snapshot SELECT 테스트]") - try: - cursor.execute("SELECT ordinal, row_json, captured_at, account, account_type, ticker, name, parse_status, user_confirmed, updated_at FROM account_snapshot LIMIT 1") - row = cursor.fetchone() - if row: - print(f" [OK] {len(row)} 컬럼") - else: - print(f" [OK] 데이터 없음") - except Exception as e: - print(f" [ERROR] {e}") - - # 4. workspace_approval_v2 SELECT 테스트 - print("\n[workspace_approval_v2 SELECT 테스트]") - try: - cursor.execute("SELECT domain, target_ref, status, approved_by, approved_at, note, updated_at FROM workspace_approval_v2 LIMIT 1") - row = cursor.fetchone() - if row: - print(f" [OK] {len(row)} 컬럼") - else: - print(f" [OK] 데이터 없음") - except Exception as e: - print(f" [ERROR] {e}") - - # 5. workspace_change_log SELECT 테스트 - print("\n[workspace_change_log SELECT 테스트]") - try: - cursor.execute("SELECT id, domain, action, target_ref, actor, note, before_json, after_json, created_at FROM workspace_change_log LIMIT 1") - row = cursor.fetchone() - if row: - print(f" [OK] {len(row)} 컬럼") - else: - print(f" [OK] 데이터 없음") - except Exception as e: - print(f" [ERROR] {e}") - - conn.close() - - print("\n[완료]") - -if __name__ == "__main__": - test_each_function() diff --git a/tools/test_import.py b/tools/test_import.py deleted file mode 100644 index 46bba05c..00000000 --- a/tools/test_import.py +++ /dev/null @@ -1,12 +0,0 @@ -#!/usr/bin/env python3 - -import sys -sys.path.insert(0, '.') - -try: - from src.quant_engine import snapshot_admin_server_v1 - print("[OK] 임포트 성공") -except Exception as e: - print(f"[ERROR] 임포트 실패: {e}") - import traceback - traceback.print_exc() diff --git a/tools/test_remaining_apis.py b/tools/test_remaining_apis.py deleted file mode 100644 index af8edbaa..00000000 --- a/tools/test_remaining_apis.py +++ /dev/null @@ -1,79 +0,0 @@ -#!/usr/bin/env python3 -""" -/api/state, /api/export 등 다른 엔드포인트 테스트 -""" - -import requests -import json -import time - -BASE_URL = "http://localhost:5000" - -print("="*80) -print("API 엔드포인트 테스트") -print("="*80) - -# 1. /api/state 테스트 -print("\n[1] GET /api/state") -try: - start = time.time() - response = requests.get(f"{BASE_URL}/api/state", timeout=15) - elapsed = time.time() - start - - print(f" 응답 시간: {elapsed:.2f}s") - print(f" 상태 코드: {response.status_code}") - - if response.status_code == 200: - result = response.json() - print(f" [OK] 키: {list(result.keys())[:5]}") - else: - print(f" [FAIL] {response.text[:100]}") - -except Exception as e: - print(f" [ERROR] {e}") - -# 2. /api/export 테스트 -print("\n[2] GET /api/export") -try: - start = time.time() - response = requests.get(f"{BASE_URL}/api/export", timeout=15) - elapsed = time.time() - start - - print(f" 응답 시간: {elapsed:.2f}s") - print(f" 상태 코드: {response.status_code}") - print(f" 응답 크기: {len(response.text)} bytes") - - if response.status_code == 200: - try: - result = response.json() - print(f" [OK] 키: {list(result.keys())[:3]}") - except: - print(f" [OK] (JSON 파싱 불가, 바이너리일 수 있음)") - else: - print(f" [FAIL] {response.text[:100]}") - -except Exception as e: - print(f" [ERROR] {e}") - -# 3. /api/tables 테스트 -print("\n[3] GET /api/tables") -try: - start = time.time() - response = requests.get(f"{BASE_URL}/api/tables", timeout=15) - elapsed = time.time() - start - - print(f" 응답 시간: {elapsed:.2f}s") - print(f" 상태 코드: {response.status_code}") - - if response.status_code == 200: - result = response.json() - print(f" [OK] {len(result)} 테이블") - for table in result[:3]: - print(f" - {table['table']}: {table['row_count']} rows") - else: - print(f" [FAIL] {response.text[:100]}") - -except Exception as e: - print(f" [ERROR] {e}") - -print("\n[완료]") diff --git a/tools/test_ui_completeness.py b/tools/test_ui_completeness.py deleted file mode 100644 index cbb48020..00000000 --- a/tools/test_ui_completeness.py +++ /dev/null @@ -1,490 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -Quant Engine UI Completeness Test -Playwright를 사용한 자동화 DOM 분석 및 완성도 평가 -""" - -import asyncio -import json -import sys -import os -from datetime import datetime -from pathlib import Path - -if sys.platform == "win32": - import io - sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') - -from playwright.async_api import async_playwright, Page - -BASE_URL = "http://localhost:5265" - -class UIAnalyzer: - """MudBlazor UI 완성도 분석""" - - def __init__(self): - self.results = { - "timestamp": datetime.now().isoformat(), - "base_url": BASE_URL, - "tests": {}, - "score": 0, - "issues": [], - "recommendations": [] - } - - async def run_all_tests(self): - """모든 테스트 실행""" - async with async_playwright() as p: - browser = await p.chromium.launch(headless=True) - page = await browser.new_page() - - try: - # 1. 페이지 로딩 테스트 - await self.test_page_load(page) - - # 2. MudBlazor 요소 검증 - await self.test_mudblazor_components(page) - - # 3. 레이아웃 검증 - await self.test_layout_structure(page) - - # 4. Dashboard 콘텐츠 검증 - await self.test_dashboard_content(page) - - # 5. 네비게이션 검증 - await self.test_navigation(page) - - # 6. 반응형 디자인 검증 - await self.test_responsive_design(page) - - # 7. 접근성 검증 - await self.test_accessibility(page) - - # 8. 성능 메트릭 수집 - await self.test_performance(page) - - finally: - await browser.close() - - # 점수 계산 및 리포트 생성 - self.calculate_score() - return self.results - - async def test_page_load(self, page: Page): - """페이지 로드 테스트""" - test_name = "page_load" - self.results["tests"][test_name] = {"status": "PENDING", "checks": []} - - try: - response = await page.goto(BASE_URL, wait_until="networkidle") - - status_ok = response.status == 200 - self.results["tests"][test_name]["checks"].append({ - "name": "HTTP Status 200", - "passed": status_ok, - "value": response.status - }) - - # 타이틀 확인 - title = await page.title() - title_ok = "Dashboard" in title or "Quant Engine" in title - self.results["tests"][test_name]["checks"].append({ - "name": "Page Title", - "passed": title_ok, - "value": title - }) - - # 로드 시간 - metrics = await page.evaluate("() => window.performance.timing") - load_time = metrics.get("loadEventEnd", 0) - metrics.get("navigationStart", 0) - load_ok = load_time < 5000 # 5초 이내 - self.results["tests"][test_name]["checks"].append({ - "name": "Load Time < 5s", - "passed": load_ok, - "value": f"{load_time}ms" - }) - - self.results["tests"][test_name]["status"] = "PASS" if all( - c["passed"] for c in self.results["tests"][test_name]["checks"] - ) else "FAIL" - - except Exception as e: - self.results["tests"][test_name]["status"] = "ERROR" - self.results["issues"].append(f"Page Load Error: {str(e)}") - - async def test_mudblazor_components(self, page: Page): - """MudBlazor 컴포넌트 검증""" - test_name = "mudblazor_components" - self.results["tests"][test_name] = {"status": "PENDING", "components": []} - - components = { - "MudLayout": "div.mud-layout", - "MudAppBar": "header.mud-appbar", - "MudDrawer": "aside.mud-drawer", - "MudMainContent": "main.mud-main-content", - "MudCard": "div.mud-card", - "MudText": "p.mud-typography", - "MudButton": "button", - "MudIcon": "svg.mud-icon-root" - } - - for component_name, selector in components.items(): - try: - count = await page.locator(selector).count() - found = count > 0 - self.results["tests"][test_name]["components"].append({ - "name": component_name, - "selector": selector, - "found": found, - "count": count - }) - except Exception as e: - self.results["tests"][test_name]["components"].append({ - "name": component_name, - "selector": selector, - "found": False, - "error": str(e) - }) - - found_count = sum(1 for c in self.results["tests"][test_name]["components"] if c["found"]) - self.results["tests"][test_name]["status"] = "PASS" if found_count >= 4 else "FAIL" - - async def test_layout_structure(self, page: Page): - """레이아웃 구조 검증""" - test_name = "layout_structure" - self.results["tests"][test_name] = {"status": "PENDING", "checks": []} - - # 1. MudLayout 존재 - layout_exists = await page.locator("div.mud-layout").count() > 0 - self.results["tests"][test_name]["checks"].append({ - "name": "MudLayout exists", - "passed": layout_exists - }) - - # 2. AppBar 존재 - appbar_exists = await page.locator("header.mud-appbar").count() > 0 - self.results["tests"][test_name]["checks"].append({ - "name": "MudAppBar exists", - "passed": appbar_exists - }) - - # 3. Drawer 존재 - drawer_exists = await page.locator("aside.mud-drawer").count() > 0 - self.results["tests"][test_name]["checks"].append({ - "name": "MudDrawer exists", - "passed": drawer_exists - }) - - # 4. MainContent 존재 - main_exists = await page.locator("main.mud-main-content").count() > 0 - self.results["tests"][test_name]["checks"].append({ - "name": "MudMainContent exists", - "passed": main_exists - }) - - # 5. MudText 최소 3개 (헤더 등) - text_count = await page.locator("p.mud-typography, h1, h2, h3, h4, h5, h6").count() - text_ok = text_count >= 3 - self.results["tests"][test_name]["checks"].append({ - "name": f"Text elements >= 3 (found {text_count})", - "passed": text_ok - }) - - self.results["tests"][test_name]["status"] = "PASS" if all( - c["passed"] for c in self.results["tests"][test_name]["checks"] - ) else "FAIL" - - async def test_dashboard_content(self, page: Page): - """Dashboard 콘텐츠 검증""" - test_name = "dashboard_content" - self.results["tests"][test_name] = {"status": "PENDING", "elements": []} - - elements = { - "Dashboard Title": "text=Dashboard", - "Status Card": "text=Status", - "Active Locks": "text=Active Locks", - "System Info": "text=System Information", - "Connected Badge": "text=Connected" - } - - for elem_name, selector in elements.items(): - try: - found = await page.locator(f"text={selector.replace('text=', '')}").count() > 0 - self.results["tests"][test_name]["elements"].append({ - "name": elem_name, - "found": found - }) - except: - self.results["tests"][test_name]["elements"].append({ - "name": elem_name, - "found": False - }) - - found_count = sum(1 for e in self.results["tests"][test_name]["elements"] if e["found"]) - self.results["tests"][test_name]["status"] = "PASS" if found_count >= 3 else "FAIL" - - async def test_navigation(self, page: Page): - """네비게이션 검증""" - test_name = "navigation" - self.results["tests"][test_name] = {"status": "PENDING", "nav_items": []} - - nav_items = { - "Dashboard": "text=Dashboard", - "Portfolio": "text=Portfolio", - "Analytics": "text=Analytics", - "Reports": "text=Reports", - "Settings": "text=Settings" - } - - for item_name, selector in nav_items.items(): - try: - found = await page.locator(selector).count() > 0 - self.results["tests"][test_name]["nav_items"].append({ - "name": item_name, - "found": found - }) - except: - self.results["tests"][test_name]["nav_items"].append({ - "name": item_name, - "found": False - }) - - found_count = sum(1 for n in self.results["tests"][test_name]["nav_items"] if n["found"]) - self.results["tests"][test_name]["status"] = "PASS" if found_count >= 3 else "FAIL" - - async def test_responsive_design(self, page: Page): - """반응형 디자인 검증""" - test_name = "responsive_design" - self.results["tests"][test_name] = {"status": "PENDING", "viewports": []} - - viewports = [ - {"name": "Mobile (375x667)", "width": 375, "height": 667}, - {"name": "Tablet (768x1024)", "width": 768, "height": 1024}, - {"name": "Desktop (1920x1080)", "width": 1920, "height": 1080} - ] - - for viewport in viewports: - await page.set_viewport_size({"width": viewport["width"], "height": viewport["height"]}) - - # 요소가 여전히 보이는지 확인 - visible = await page.locator("header.mud-appbar").is_visible() - self.results["tests"][test_name]["viewports"].append({ - "name": viewport["name"], - "size": f"{viewport['width']}x{viewport['height']}", - "appbar_visible": visible - }) - - self.results["tests"][test_name]["status"] = "PASS" if all( - v["appbar_visible"] for v in self.results["tests"][test_name]["viewports"] - ) else "FAIL" - - async def test_accessibility(self, page: Page): - """접근성 검증 (기본)""" - test_name = "accessibility" - self.results["tests"][test_name] = {"status": "PENDING", "checks": []} - - # 1. Lang 속성 - html_lang = await page.locator("html").get_attribute("lang") - lang_ok = html_lang is not None - self.results["tests"][test_name]["checks"].append({ - "name": "HTML lang attribute", - "passed": lang_ok, - "value": html_lang - }) - - # 2. Meta charset - charset = await page.locator("meta[charset]").count() > 0 - self.results["tests"][test_name]["checks"].append({ - "name": "Meta charset", - "passed": charset - }) - - # 3. Viewport meta - viewport = await page.locator("meta[name='viewport']").count() > 0 - self.results["tests"][test_name]["checks"].append({ - "name": "Meta viewport", - "passed": viewport - }) - - # 4. Heading hierarchy - headings = await page.locator("h1, h2, h3, h4, h5, h6").count() - heading_ok = headings > 0 - self.results["tests"][test_name]["checks"].append({ - "name": f"Heading hierarchy (found {headings})", - "passed": heading_ok - }) - - self.results["tests"][test_name]["status"] = "PASS" if all( - c["passed"] for c in self.results["tests"][test_name]["checks"] - ) else "FAIL" - - async def test_performance(self, page: Page): - """성능 메트릭 수집""" - test_name = "performance" - self.results["tests"][test_name] = {"status": "PASS", "metrics": {}} - - try: - metrics = await page.evaluate(""" - () => ({ - domContentLoaded: performance.timing.domContentLoadedEventEnd - performance.timing.navigationStart, - loadComplete: performance.timing.loadEventEnd - performance.timing.navigationStart, - resources: performance.getEntriesByType('resource').length, - memoryUsage: performance.memory ? Math.round(performance.memory.usedJSHeapSize / 1048576) : null - }) - """) - - self.results["tests"][test_name]["metrics"] = { - "DOM Content Loaded (ms)": metrics.get("domContentLoaded", 0), - "Page Load Complete (ms)": metrics.get("loadComplete", 0), - "Resources Loaded": metrics.get("resources", 0), - "Memory Usage (MB)": metrics.get("memoryUsage") - } - except Exception as e: - self.results["tests"][test_name]["status"] = "WARN" - self.results["tests"][test_name]["error"] = str(e) - - def calculate_score(self): - """완성도 점수 계산""" - total_weight = 0 - total_score = 0 - - weights = { - "page_load": 15, - "mudblazor_components": 20, - "layout_structure": 20, - "dashboard_content": 15, - "navigation": 15, - "responsive_design": 10, - "accessibility": 5, - "performance": 0 # 점수에 포함 안 함, 참고용 - } - - for test_name, weight in weights.items(): - if test_name in self.results["tests"]: - test = self.results["tests"][test_name] - if test["status"] == "PASS": - total_score += weight - elif test["status"] == "FAIL": - # 부분 점수 - if "checks" in test: - passed = sum(1 for c in test["checks"] if c.get("passed", False)) - total = len(test["checks"]) - total_score += weight * (passed / total) - elif "components" in test: - found = sum(1 for c in test["components"] if c.get("found", False)) - total = len(test["components"]) - total_score += weight * (found / total) - - total_weight += weight - - self.results["score"] = round(total_score, 1) if total_weight > 0 else 0 - self.results["max_score"] = total_weight - - # 권장사항 생성 - self.generate_recommendations() - - def generate_recommendations(self): - """개선 권장사항 생성""" - recommendations = [] - - # Dashboard 콘텐츠 부족 - if self.results["tests"]["dashboard_content"]["status"] == "FAIL": - recommendations.append({ - "category": "Content", - "priority": "HIGH", - "issue": "Dashboard 콘텐츠가 부족함", - "suggestion": "스타투스 카드, 통계, 실시간 데이터 추가", - "files": ["src/dotnet/QuantEngine.Web/Components/Pages/Dashboard.razor"] - }) - - # 네비게이션 미완성 - if self.results["tests"]["navigation"]["status"] == "FAIL": - found = sum(1 for n in self.results["tests"]["navigation"]["nav_items"] if n["found"]) - recommendations.append({ - "category": "Navigation", - "priority": "MEDIUM", - "issue": f"네비게이션 항목 {found}/5개만 구현됨", - "suggestion": "모든 네비게이션 항목 추가 (Analytics, Reports 등)", - "files": ["src/dotnet/QuantEngine.Web/Components/Layout/NavMenu.razor"] - }) - - # 반응형 디자인 개선 - if self.results["tests"]["responsive_design"]["status"] == "FAIL": - recommendations.append({ - "category": "UI/UX", - "priority": "MEDIUM", - "issue": "일부 뷰포트에서 레이아웃 깨짐", - "suggestion": "MudContainer MaxWidth 조정, Grid 반응형 설정 확인", - "files": ["src/dotnet/QuantEngine.Web/Components/Pages/Dashboard.razor"] - }) - - # 접근성 개선 - if self.results["tests"]["accessibility"]["status"] == "FAIL": - recommendations.append({ - "category": "Accessibility", - "priority": "LOW", - "issue": "접근성 표준 미충족", - "suggestion": "ARIA 라벨 추가, 색상 대비 개선", - "files": ["src/dotnet/QuantEngine.Web/Components/App.razor"] - }) - - self.results["recommendations"] = recommendations - - async def take_screenshot(self, page: Page, filename: str): - """스크린샷 캡처""" - await page.screenshot(path=filename) - print(f"✓ Screenshot: {filename}") - - -async def main(): - """메인 실행""" - analyzer = UIAnalyzer() - - print("🧪 Quant Engine UI Completeness Test") - print("=" * 70) - print(f"URL: {BASE_URL}") - print("=" * 70) - - try: - results = await analyzer.run_all_tests() - - # 결과 출력 - print("\n📊 Test Results") - print("-" * 70) - - for test_name, test_data in results["tests"].items(): - status = test_data["status"] - status_emoji = "✅" if status == "PASS" else "❌" if status == "FAIL" else "⚠️" - print(f"{status_emoji} {test_name.upper()}: {status}") - - print("\n" + "=" * 70) - print(f"📈 Completeness Score: {results['score']}/{results['max_score']} ({results['score']/results['max_score']*100:.1f}%)") - print("=" * 70) - - # 권장사항 - if results["recommendations"]: - print("\n💡 Recommendations for Improvement") - print("-" * 70) - for i, rec in enumerate(results["recommendations"], 1): - print(f"\n{i}. [{rec['priority']}] {rec['issue']}") - print(f" Category: {rec['category']}") - print(f" Suggestion: {rec['suggestion']}") - print(f" Files: {', '.join(rec['files'])}") - - # 결과 저장 - output_file = Path("Temp/ui_test_results.json") - output_file.parent.mkdir(parents=True, exist_ok=True) - output_file.write_text(json.dumps(results, indent=2, ensure_ascii=False)) - print(f"\n✓ Results saved: {output_file}") - - return results - - except Exception as e: - print(f"\n❌ Error: {e}") - import traceback - traceback.print_exc() - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/tools/test_ui_with_details.py b/tools/test_ui_with_details.py deleted file mode 100644 index 5dc68940..00000000 --- a/tools/test_ui_with_details.py +++ /dev/null @@ -1,99 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -Quant Engine UI Testing with Detailed Output -""" - -import asyncio -import sys -import io - -if sys.platform == "win32": - sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') - -from playwright.async_api import async_playwright - -async def test_ui(): - """기본 UI 테스트 실행""" - async with async_playwright() as p: - browser = await p.chromium.launch(headless=True) - page = await browser.new_page() - - try: - print("[1] 페이지 로드 시도...") - response = await page.goto("http://localhost:5265", wait_until="domcontentloaded", timeout=10000) - print(f" ✓ Status: {response.status}") - - # 콘솔 메시지 수집 - console_messages = [] - page.on("console", lambda msg: console_messages.append(f"[{msg.type}] {msg.text}")) - - await page.wait_for_timeout(3000) - - # HTML 구조 확인 - print("\n[2] HTML 구조 분석...") - html = await page.content() - - # 핵심 요소 확인 - checks = [ - ("", "DOCTYPE 존재"), - (" dict: - """파일 업데이트""" - if not file_path.exists(): - return {"status": "NOT_FOUND", "file": str(file_path)} - - with open(file_path, 'r', encoding='utf-8') as f: - content = f.read() - - original = content - changes = [] - - # 패턴 적용 - for pattern, replacement in REPLACEMENTS: - matches = re.findall(pattern, content) - if matches: - content = re.sub(pattern, replacement, content) - changes.extend(matches) - - # 특별 처리: KIS_COLLECTION_DB 변수 - kis_pattern = r'KIS_COLLECTION_DB\s*=\s*ROOT\s*\/\s*"outputs"[^=]*$' - if re.search(kis_pattern, content, re.MULTILINE): - content = re.sub( - kis_pattern, - 'KIS_COLLECTION_DB = ROOT / "src" / "quant_engine" / "kis_data_collection.db"', - content, - flags=re.MULTILINE - ) - changes.append("KIS_COLLECTION_DB assignment") - - # 특별 처리: DEFAULT_DB 변수 - db_pattern = r'DEFAULT_DB\s*=\s*ROOT\s*\/\s*"outputs"[^=]*$' - if re.search(db_pattern, content, re.MULTILINE): - content = re.sub( - db_pattern, - 'DEFAULT_DB = ROOT / "src" / "quant_engine" / "snapshot_admin.db"', - content, - flags=re.MULTILINE - ) - changes.append("DEFAULT_DB assignment") - - if content != original: - with open(file_path, 'w', encoding='utf-8') as f: - f.write(content) - return { - "status": "UPDATED", - "file": str(file_path), - "changes": len(set(changes)) - } - else: - return { - "status": "NO_CHANGES", - "file": str(file_path) - } - -def main(): - print("="*80) - print("데이터베이스 경로 자동 업데이트") - print("="*80) - print(f"작업: outputs/* → src/quant_engine/\n") - - results = [] - for file_path_str in FILES_TO_UPDATE: - file_path = Path(file_path_str) - result = update_file(file_path) - results.append(result) - - status = result["status"] - symbol = "[OK]" if status == "UPDATED" else "[~]" if status == "NO_CHANGES" else "[!]" - print(f"{symbol} {file_path.name}: {status}") - if status == "UPDATED": - print(f" └─ {result['changes']} references updated") - - print("\n" + "="*80) - updated = sum(1 for r in results if r["status"] == "UPDATED") - print(f"[결과] {updated}개 파일 업데이트 완료") - print("\n[다음 단계]") - print("1. git diff 로 변경 내용 검토") - print("2. git add -u && git commit") - print("3. 배포 전 테스트 (run_snapshot_admin_server_v1.py)") - -if __name__ == "__main__": - main() diff --git a/tools/validate_json_conversion.py b/tools/validate_json_conversion.py deleted file mode 100644 index 935c6361..00000000 --- a/tools/validate_json_conversion.py +++ /dev/null @@ -1,59 +0,0 @@ -from __future__ import annotations - -import json -from pathlib import Path - -import pandas as pd - -from convert_xlsx_to_json import find_header_row, clean_dataframe, normalize_code - - -ROOT = Path(__file__).resolve().parents[1] -XLSX = ROOT / "GatherTradingData.xlsx" -JSON_PATH = ROOT / "GatherTradingData.json" - - -def validate_conversion(xlsx_path: Path, json_path: Path) -> int: - print(f"Validating {xlsx_path.name} vs {json_path.name}...") - payload = json.loads(json_path.read_text(encoding="utf-8")) - json_data = payload["data"] - xl = pd.ExcelFile(xlsx_path) - errors: list[str] = [] - - for sheet in xl.sheet_names: - if sheet.startswith("cs_chunk_"): - continue - if sheet not in json_data: - errors.append(f"{sheet}: missing in JSON") - continue - header_row = find_header_row(xlsx_path, sheet) - df = pd.read_excel(xlsx_path, sheet_name=sheet, header=header_row) - df = clean_dataframe(df) - expected_rows = len(df) - actual = json_data[sheet] - actual_rows = len(actual) if hasattr(actual, "__len__") else 0 - if expected_rows != actual_rows: - errors.append(f"{sheet}: XLSX rows={expected_rows} JSON rows={actual_rows}") - continue - if isinstance(actual, list) and actual: - columns = set(df.columns) - json_columns = set(actual[0]) - if not columns <= json_columns: - errors.append(f"{sheet}: JSON missing columns sample={sorted(columns - json_columns)[:10]}") - if "Ticker" in columns: - xlsx_ticker = normalize_code(df.iloc[0]["Ticker"]) - json_ticker = str(actual[0].get("Ticker", "")) - if xlsx_ticker != json_ticker: - errors.append(f"{sheet}: first Ticker mismatch XLSX={xlsx_ticker} JSON={json_ticker}") - - if errors: - print("JSON CONVERSION VALIDATION FAIL") - for err in errors: - print(f"- {err}") - return 1 - print("JSON CONVERSION VALIDATION OK") - return 0 - - -if __name__ == "__main__": - raise SystemExit(validate_conversion(XLSX, JSON_PATH)) diff --git a/tools/verify_data_load.py b/tools/verify_data_load.py deleted file mode 100644 index f79bc506..00000000 --- a/tools/verify_data_load.py +++ /dev/null @@ -1,55 +0,0 @@ -#!/usr/bin/env python3 -""" -데이터베이스 로드 상태 검증 -""" - -import sqlite3 -from pathlib import Path - -def verify_databases(): - """두 데이터베이스의 상태 확인""" - - kis_db = Path('src/quant_engine/kis_data_collection.db') - snapshot_db = Path('src/quant_engine/snapshot_admin.db') - - print("="*80) - print("데이터베이스 로드 상태 검증") - print("="*80) - - for db_name, db_path in [("kis_data_collection", kis_db), ("snapshot_admin", snapshot_db)]: - print(f"\n[{db_name}]") - - if not db_path.exists(): - print(f" 파일 없음") - continue - - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 테이블 목록 - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") - tables = [row[0] for row in cursor.fetchall()] - print(f" 테이블 수: {len(tables)}") - print(f" 테이블: {', '.join(tables[:5])}..." if len(tables) > 5 else f" 테이블: {', '.join(tables)}") - - # 각 테이블의 행 수 - print(f"\n 테이블별 행 수:") - total_rows = 0 - for table in sorted(tables): - try: - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - if count > 0: - print(f" {table}: {count:,}") - total_rows += count - except: - pass - - print(f" 총 행 수: {total_rows:,}") - - conn.close() - - print("\n[완료]") - -if __name__ == "__main__": - verify_databases() diff --git a/tools/verify_sheet_to_table_sync.py b/tools/verify_sheet_to_table_sync.py deleted file mode 100644 index 5be4daf2..00000000 --- a/tools/verify_sheet_to_table_sync.py +++ /dev/null @@ -1,232 +0,0 @@ -#!/usr/bin/env python3 -""" -시트→테이블 동기화 검증 - -XLSX 시트와 DB 테이블이 정확히 동기화되었는지 확인 -""" - -import pandas as pd -import sqlite3 -from pathlib import Path -from datetime import datetime - -class SheetTableSyncVerification: - """시트-테이블 동기화 검증""" - - def __init__(self): - self.xlsx_file = Path('GatherTradingData.xlsx') - self.kis_db = Path('src/quant_engine/kis_data_collection.db') - self.snapshot_db = Path('src/quant_engine/snapshot_admin.db') - self.results = { - "timestamp": datetime.now().isoformat(), - "sheets": {}, - "tables": {}, - "sync_status": {} - } - - def verify_xlsx_sheets(self) -> dict: - """XLSX 시트 검증""" - print("\n[XLSX 시트 검증]") - - excel_file = pd.ExcelFile(self.xlsx_file) - sheet_names = excel_file.sheet_names - - print(f" 발견된 시트: {len(sheet_names)}개") - - for sheet_name in sheet_names: - df = pd.read_excel(self.xlsx_file, sheet_name=sheet_name) - self.results["sheets"][sheet_name] = { - "rows": len(df), - "columns": len(df.columns), - "col_names": list(df.columns) - } - print(f" {sheet_name}: {len(df)} rows, {len(df.columns)} cols") - - return self.results["sheets"] - - def verify_db_tables(self) -> dict: - """DB 테이블 검증""" - print("\n[DB 테이블 검증]") - - db_info = {} - - # kis_data_collection - conn = sqlite3.connect(self.kis_db) - cursor = conn.cursor() - - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name != 'sqlite_sequence'") - tables = [row[0] for row in cursor.fetchall()] - - print(f" kis_data_collection.db: {len(tables)}개 테이블") - for table in tables: - cursor.execute(f"PRAGMA table_info({table})") - cols = [col[1] for col in cursor.fetchall()] - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - - db_info[f"kis.{table}"] = { - "rows": count, - "columns": len(cols), - "col_names": cols - } - print(f" {table}: {count} rows, {len(cols)} cols") - - conn.close() - - # snapshot_admin - conn = sqlite3.connect(self.snapshot_db) - cursor = conn.cursor() - - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name != 'sqlite_sequence'") - tables = [row[0] for row in cursor.fetchall()] - - print(f" snapshot_admin.db: {len(tables)}개 테이블") - for table in tables: - cursor.execute(f"PRAGMA table_info({table})") - cols = [col[1] for col in cursor.fetchall()] - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - - db_info[f"snapshot.{table}"] = { - "rows": count, - "columns": len(cols), - "col_names": cols - } - if count > 0: - print(f" {table}: {count} rows, {len(cols)} cols") - - conn.close() - - self.results["tables"] = db_info - return db_info - - def verify_sync(self) -> dict: - """시트-테이블 동기화 확인""" - print("\n[동기화 상태]") - - sync_status = {} - - for sheet_name, sheet_info in self.results["sheets"].items(): - # kis.data_feed 특수 매핑 - if sheet_name == "data_feed": - table_key = "kis.data_feed" - else: - table_key = f"snapshot.{sheet_name}" - - if table_key in self.results["tables"]: - table_info = self.results["tables"][table_key] - - # 행 수 비교 - rows_match = sheet_info["rows"] == table_info["rows"] - # 컬럼 수 비교 - cols_match = sheet_info["columns"] == table_info["columns"] - - status = "OK" if (rows_match and cols_match) else "MISMATCH" - - sync_status[sheet_name] = { - "status": status, - "sheet_rows": sheet_info["rows"], - "table_rows": table_info["rows"], - "rows_match": rows_match, - "sheet_cols": sheet_info["columns"], - "table_cols": table_info["columns"], - "cols_match": cols_match - } - - symbol = "[OK]" if status == "OK" else "[!]" - print(f" {symbol} {sheet_name}") - if not rows_match: - print(f" 행: {sheet_info['rows']} vs {table_info['rows']}") - if not cols_match: - print(f" 컬럼: {sheet_info['columns']} vs {table_info['columns']}") - else: - sync_status[sheet_name] = { - "status": "NOT_FOUND", - "message": f"Table {table_key} not found in DB" - } - print(f" [!] {sheet_name}: 테이블 미발견") - - self.results["sync_status"] = sync_status - return sync_status - - def verify_data_integrity(self) -> dict: - """데이터 무결성 검증""" - print("\n[데이터 무결성 검증]") - - integrity_checks = { - "not_null_violations": 0, - "duplicate_keys": 0, - "orphaned_records": 0 - } - - # kis_data_collection - conn = sqlite3.connect(self.kis_db) - cursor = conn.cursor() - - # data_feed의 NULL 검증 - cursor.execute("SELECT COUNT(*) FROM data_feed WHERE ticker IS NULL") - null_count = cursor.fetchone()[0] - if null_count > 0: - integrity_checks["not_null_violations"] += null_count - print(f" [!] data_feed: {null_count}개 NULL ticker 발견") - - conn.close() - - # snapshot_admin - conn = sqlite3.connect(self.snapshot_db) - cursor = conn.cursor() - - # settings의 NOT NULL 검증 - cursor.execute("SELECT COUNT(*) FROM settings WHERE key IS NULL") - null_count = cursor.fetchone()[0] - if null_count > 0: - integrity_checks["not_null_violations"] += null_count - print(f" [!] settings: {null_count}개 NULL key 발견") - - if integrity_checks["not_null_violations"] == 0: - print(f" [OK] NULL 위반 없음") - - conn.close() - - return integrity_checks - - def run(self) -> dict: - """전체 실행""" - print("="*80) - print("시트→테이블 동기화 검증") - print("="*80) - - # 1. XLSX 시트 검증 - self.verify_xlsx_sheets() - - # 2. DB 테이블 검증 - self.verify_db_tables() - - # 3. 동기화 상태 확인 - self.verify_sync() - - # 4. 데이터 무결성 검증 - integrity = self.verify_data_integrity() - - # 최종 요약 - print("\n" + "="*80) - print("[최종 검증 결과]") - - total_sheets = len(self.results["sheets"]) - synced_sheets = sum(1 for v in self.results["sync_status"].values() if v.get("status") == "OK") - print(f" 시트 동기화: {synced_sheets}/{total_sheets}") - - print(f" 데이터 무결성: {integrity['not_null_violations']}개 위반") - - overall_status = "PASS" if synced_sheets == total_sheets and integrity['not_null_violations'] == 0 else "FAIL" - print(f" 종합 평가: {overall_status}") - - print("="*80) - - return self.results - -if __name__ == "__main__": - verifier = SheetTableSyncVerification() - result = verifier.run() - - print("\n[완료] 시트-테이블 동기화 검증 완료") diff --git a/tools/verify_table_coverage.py b/tools/verify_table_coverage.py deleted file mode 100644 index fe2e4814..00000000 --- a/tools/verify_table_coverage.py +++ /dev/null @@ -1,104 +0,0 @@ -#!/usr/bin/env python3 -""" -DB 테이블 커버리지 검증 - -GatherTradingData.json의 시트 vs 현재 DB 테이블 비교 -""" - -import json -import sqlite3 -from pathlib import Path - -def get_xlsx_sheets(): - """GatherTradingData.json에서 시트 목록 추출""" - try: - with open('GatherTradingData.json', encoding='utf-8') as f: - full_data = json.load(f) - sheets = full_data.get('metadata', {}).get('sheets_included', []) - return sheets - except: - try: - with open('GatherTradingData.json', encoding='euc-kr') as f: - full_data = json.load(f) - sheets = full_data.get('metadata', {}).get('sheets_included', []) - return sheets - except: - return [] - -def get_db_tables(): - """DB의 현재 테이블 조회""" - tables = {} - - for db_name, db_path in [ - ("kis_data_collection", "src/quant_engine/kis_data_collection.db"), - ("snapshot_admin", "src/quant_engine/snapshot_admin.db") - ]: - if not Path(db_path).exists(): - continue - - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") - db_tables = [row[0] for row in cursor.fetchall() if row[0] != 'sqlite_sequence'] - conn.close() - - tables[db_name] = db_tables - - return tables - -def main(): - print("="*80) - print("데이터베이스 테이블 커버리지 검증") - print("="*80) - - # XLSX 시트 - xlsx_sheets = get_xlsx_sheets() - print(f"\n[GatherTradingData.json]") - print(f"총 시트 수: {len(xlsx_sheets)}") - print("시트 목록:") - for i, sheet in enumerate(xlsx_sheets, 1): - print(f" {i:2}. {sheet}") - - # DB 테이블 - db_tables = get_db_tables() - total_tables = sum(len(t) for t in db_tables.values()) - - print(f"\n[현재 DB]") - print(f"총 테이블 수: {total_tables}") - for db_name, tables in db_tables.items(): - print(f"\n{db_name}.db:") - for table in tables: - print(f" - {table}") - - # 비교 - print("\n" + "="*80) - print("커버리지 분석") - print("="*80) - - all_db_tables = [] - for tables in db_tables.values(): - all_db_tables.extend(tables) - - covered = [s for s in xlsx_sheets if s.lower() in [t.lower() for t in all_db_tables]] - missing = [s for s in xlsx_sheets if s.lower() not in [t.lower() for t in all_db_tables]] - - coverage = (len(covered) / len(xlsx_sheets) * 100) if xlsx_sheets else 0 - - print(f"\n[결과]") - print(f" 커버된 시트: {len(covered)}/{len(xlsx_sheets)} ({coverage:.1f}%)") - print(f" 누락된 시트: {len(missing)}") - - if missing: - print(f"\n[누락된 시트]") - for sheet in missing: - print(f" - {sheet}") - - print(f"\n[권장]") - print("다음 테이블들을 추가하여 커버리지를 완성해야 함:") - for sheet in missing[:10]: - print(f" - {sheet}") - if len(missing) > 10: - print(f" ... 및 {len(missing)-10}개 추가") - -if __name__ == "__main__": - main()