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()