kjh2064/QuantEngineByItz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 9 Wiki Activity

chore: remove 34 orphaned one-off scripts from tools/

Browse Source 
Systematic scan (not manual per-file review): for every *.py under
tools/ and src/quant_engine/, searched the entire tracked repo for
its filename stem outside itself (code, docs, spec, CI workflows,
tests). 81 of 619 files had zero references anywhere. Deleting all
81 in one pass was judged too risky - some could be intentional
manual-run utilities that just don't get named by other files - so
this commit only removes the two highest-confidence tiers, 34 files:

- Superseded DB init/fix/load scripts, most with an explicit "_v2"/
  "_correct"/"_properly" sibling that replaced them (e.g.
  load_from_xlsx.py -> load_from_xlsx_correct.py,
  initialize_database.py -> initialize_database_v2.py). Verified via
  `git log` that both members of each pair were last touched the same
  day (2026-06-23) and never again - abandoned mid-iteration, not
  live tooling.
- Ad-hoc test/debug scripts sitting in tools/ instead of tests/
  (test_api_components.py, test_build_ui_state*.py, diagnose_api_error.py,
  verify_*.py, validate_json_conversion.py) - one-off interactive
  debugging session artifacts.

Re-ran the full-repo reference search after deletion (zero hits) and
the quick-mode release DAG + WBS validator (both still pass) to confirm
nothing references these files. The other 47 orphan candidates (WBS-
ticket-tagged one-offs and misc build_*/validate_* scripts) are left
in place pending further review - some may be intentional manual
utilities the automated reference scan can't distinguish from dead code.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
김재현
2026-07-30 14:12:09 +09:00
parent fa0da0b159
commit 977167a0db
34 changed files with 0 additions and 4343 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/archive_legacy_databases.py
-206
View File
@@ -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/")
tools/check_schema.py
-27
View File
@@ -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()
tools/diagnose_api_error.py
-95
View File
@@ -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()
tools/fix_account_snapshot_schema.py
-118
View File
@@ -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()
tools/fix_account_snapshot_v2.py
-110
View File
@@ -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()
tools/fix_settings_schema.py
-89
View File
@@ -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()
tools/fix_settings_schema_v2.py
-83
View File
@@ -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()
tools/init_performance_tables.py
-87
View File
@@ -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()
tools/initialize_database.py
-134
View File
@@ -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[완료] 데이터베이스 초기화 완료!")
tools/initialize_database_v2.py
-154
View File
@@ -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 초기화 완료!")
tools/initialize_snapshot_admin_db.py
-153
View File
@@ -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()
tools/load_all_trading_data.py
-183
View File
@@ -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")
tools/load_complete_trading_data.py
-189
View File
@@ -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[완료] 완전 로드 완료")
tools/load_from_xlsx.py
-153
View File
@@ -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[완료]")
tools/load_from_xlsx_correct.py
-212
View File
@@ -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 로드 완료")
tools/load_kis_sample_data_v1.py
-306
View File
@@ -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'])}")
tools/load_settings_config.py
-74
View File
@@ -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()
tools/load_settings_properly.py
-78
View File
@@ -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()
tools/normalize_formula_registry_v2.py
-48
View File
@@ -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())
tools/refactor_database_structure.py
-196
View File
@@ -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()
tools/test_api_components.py
-62
View File
@@ -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[완료]")
tools/test_api_settings_save.py
-57
View File
@@ -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[완료]")
tools/test_build_ui_state.py
-131
View File
@@ -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[완료]")
tools/test_build_ui_state_direct.py
-58
View File
@@ -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[완료]")
tools/test_build_ui_state_simple.py
-93
View File
@@ -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()
tools/test_import.py
-12
View File
@@ -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()
tools/test_remaining_apis.py
-79
View File
@@ -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[완료]")
tools/test_ui_completeness.py
-490
View File
@@ -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())
tools/test_ui_with_details.py
-99
View File
@@ -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 html>", "DOCTYPE 존재"),
("<html", "HTML 태그"),
("MudBlazor", "MudBlazor 포함"),
("mud-layout", "mud-layout 클래스"),
("mud-appbar", "mud-appbar 클래스"),
("Dashboard", "Dashboard 텍스트"),
]
for check_str, desc in checks:
found = check_str in html
status = "✓" if found else "✗"
print(f" {status} {desc}")
# 실제 DOM 덤프 (처음 2000자)
print("\n[3] HTML 미리보기 (처음 1000자):")
print("-" * 70)
print(html[:1000])
print("-" * 70)
# 요소 선택자 테스트
print("\n[4] DOM 요소 테스트...")
selectors = {
"h1": "h1",
"h2": "h2",
"h3": "h3",
"h4": "h4",
"h5": "h5",
"h6": "h6",
"MudText (p)": "p",
"MudCard (div.mud-card)": "div.mud-card",
"MudButton (button)": "button",
"MudLayout (div.mud-layout)": "div.mud-layout",
"MudAppBar (header)": "header",
}
for name, selector in selectors.items():
count = await page.locator(selector).count()
print(f" {name}: {count}개")
# 페이지 제목
print(f"\n[5] 페이지 제목: {await page.title()}")
# 콘솔 메시지
if console_messages:
print(f"\n[6] 콘솔 메시지:")
for msg in console_messages:
print(f" {msg}")
# 스크린샷
await page.screenshot(path="C:\\Temp\\data_feed\\screenshot_ui.png")
print("\n✓ 스크린샷 저장: screenshot_ui.png")
except Exception as e:
print(f"✗ 오류: {e}")
import traceback
traceback.print_exc()
finally:
await browser.close()
asyncio.run(test_ui())
tools/update_db_paths.py
-117
View File
@@ -1,117 +0,0 @@
#!/usr/bin/env python3
"""
데이터베이스 경로 자동 업데이트
레거시 경로 (outputs/) → 정규 경로 (src/quant_engine/)
"""
import re
from pathlib import Path
# 업데이트 규칙
REPLACEMENTS = [
# KIS data collection DB
(r'ROOT\s*\/\s*"outputs"\s*\/\s*"kis_data_collection"',
'ROOT / "src" / "quant_engine"'),
(r'"outputs"\s*\/\s*"kis_data_collection"\s*\/\s*"kis_data_collection\.db"',
'"src" / "quant_engine" / "kis_data_collection.db"'),
(r'ROOT\s*\/\s*"outputs"\s*\/\s*"snapshot_admin"\s*\/\s*"snapshot_admin\.db"',
'ROOT / "src" / "quant_engine" / "snapshot_admin.db"'),
# snapshot_admin DB
(r'"outputs"\s*\/\s*"snapshot_admin"\s*\/\s*"snapshot_admin\.db"',
'"src" / "quant_engine" / "snapshot_admin.db"'),
]
FILES_TO_UPDATE = [
"src/quant_engine/kis_api_client_v1.py",
"src/quant_engine/kis_data_collection_v1.py",
"src/quant_engine/snapshot_admin_server_v1.py",
"src/quant_engine/snapshot_admin_store_v1.py",
"tools/run_snapshot_admin_server_v1.py",
]
def update_file(file_path: Path) -> 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()
tools/validate_json_conversion.py
-59
View File
@@ -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))
tools/verify_data_load.py
-55
View File
@@ -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()
tools/verify_sheet_to_table_sync.py
-232
View File
@@ -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[완료] 시트-테이블 동기화 검증 완료")
tools/verify_table_coverage.py
-104
View File
@@ -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()