Files
QuantEngineByItz/verify_admin_db.py
T
kjh2064 05d9f8ed41 코드 참조 경로 업데이트 & WBS-8.1 모니터링 준비
경로 정규화 (outputs/ → src/quant_engine/):
✓ kis_api_client_v1.py: KIS 데이터 수집 경로
✓ kis_data_collection_v1.py: 기본 DB 인자
✓ snapshot_admin_server_v1.py: KIS_COLLECTION_DB
✓ snapshot_admin_store_v1.py: DEFAULT_DB + collector_db
✓ run_snapshot_admin_server_v1.py: --db 기본값

모니터링 도구 추가:
✓ verify_admin_db.py: 어드민 서버 & DB 검증
✓ setup_wbs81_monitoring.py: WBS-8.1 목표 추적 시스템
✓ update_db_paths.py: 자동화된 경로 업데이트

효과:
- 단일 소스 (src/quant_engine/)
- 배포 스크립트 단순화
- WBS-8.1: T+20 30건 모니터링 준비 완료
- 22일 남음 (목표: 2026-07-15)

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
2026-06-23 00:33:32 +09:00

52 lines
1.4 KiB
Python

#!/usr/bin/env python3
import sqlite3
from pathlib import Path
print("="*80)
print("어드민 서버 & DB 연결 검증")
print("="*80)
dbs = {
'kis_data_collection.db': 'src/quant_engine/kis_data_collection.db',
'snapshot_admin.db': 'src/quant_engine/snapshot_admin.db'
}
all_ok = True
for name, path in dbs.items():
if not Path(path).exists():
print(f'[FAIL] {name} not found')
all_ok = False
continue
try:
conn = sqlite3.connect(path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in cursor.fetchall()]
# 각 테이블 행 수
table_info = {}
for table in tables:
if table == 'sqlite_sequence':
continue
cursor.execute(f'SELECT COUNT(*) FROM {table}')
count = cursor.fetchone()[0]
table_info[table] = count
conn.close()
file_size = Path(path).stat().st_size / 1024
print(f'\n[OK] {name} ({file_size:.2f} KB)')
for table, count in sorted(table_info.items()):
print(f' └─ {table}: {count} records')
except Exception as e:
print(f'\n[FAIL] {name}: {e}')
all_ok = False
print("\n" + "="*80)
if all_ok:
print("[결과] [OK] 어드민 서버 & DB 모두 정상 접속")
else:
print("[결과] [FAIL] DB 연결 실패")