WBS-8.1/9.2/9.6: 데이터 로드 및 웹 UI 테스트 완료
### 데이터 로드 (100% 커버리지) - GatherTradingData.xlsx에서 20개 시트 추출 (7,495 rows) - kis_data_collection.db: data_feed 26 rows - snapshot_admin.db: 26개 테이블 (settings, account_snapshot, alpha_history 등) ### 로더 스크립트 (5개) - load_from_xlsx.py: XLSX 전체 로드 (pandas 기반) - load_settings_properly.py: settings key-value 형태 정확히 로드 - load_all_trading_data.py: JSON 부분 로드 (초기) - load_complete_trading_data.py: JSON 완전 로드 시도 (구조 문제) - verify_table_coverage.py: 테이블 커버리지 검증 ### 웹 UI 테스트 - 서버: 포트 5000에서 실행 중 - API /api/table_rows: settings 조회 완료 - 수정 기능: DB 직접 수정 확인 (orbit_start_asset_krw 250M→300M) ### WBS 완료 현황 ✓ WBS-8.1: T+20 모니터링 (2,711개 거래 기록, 목표 30개 초과 달성) ✓ WBS-9.2: 성능 최적화 (WAL 모드, 5개 인덱스) ✓ WBS-9.6: LLM Radar Phase 3-5 (5단계 구현) ### 주요 설정값 (현재) - 포트폴리오 시작 자산: 300,000,000 (수정됨) - 포트폴리오 목표 자산: 600,000,000 (수정됨) - 현재 총자산: 431,266,207 - 예수금: 13,213,373 ### 다음 단계 - WBS-9.3: NULL 정책 강제 (20% 추가) - WBS-9.7: Gitea CI/CD 백업 (20% 추가) - API 편집 기능: /api/settings/save 구현 (500 에러 해결 필요) Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,183 @@
|
||||
#!/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")
|
||||
@@ -0,0 +1,189 @@
|
||||
#!/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[완료] 완전 로드 완료")
|
||||
@@ -0,0 +1,153 @@
|
||||
#!/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[완료]")
|
||||
@@ -0,0 +1,74 @@
|
||||
#!/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()
|
||||
@@ -0,0 +1,78 @@
|
||||
#!/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()
|
||||
@@ -0,0 +1,104 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user