32 lines
1.5 KiB
Python
32 lines
1.5 KiB
Python
import sys, os
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)) + '/..')
|
|
from kis_trader.utils.env import get_env_from_db
|
|
|
|
key_ob_real = get_env_from_db("KIS_APP_KEY_OB_REAL", "")
|
|
key_ob_mock = get_env_from_db("KIS_APP_KEY_OB_MOCK", "")
|
|
print(f"KIS_APP_KEY_OB_REAL = {'SET(' + key_ob_real[:6] + '...)' if key_ob_real else 'EMPTY/NOT_SET'}")
|
|
print(f"KIS_APP_KEY_OB_MOCK = {'SET(' + key_ob_mock[:6] + '...)' if key_ob_mock else 'EMPTY/NOT_SET'}")
|
|
|
|
# 또한 ws_orderbook에 저장된 실제 데이터 확인 (KIS 호가가 어디에 쌓이는지)
|
|
from database import TradeDB
|
|
db = TradeDB()
|
|
try:
|
|
tables = [dict(r)['Tables_in_kis_quant_db'] for r in db.conn.execute("SHOW TABLES").fetchall()]
|
|
ob_tables = [t for t in tables if 'orderbook' in t.lower()]
|
|
print("\n[호가 테이블 현황]")
|
|
for t in ob_tables:
|
|
cnt = dict(db.conn.execute(f"SELECT count(*) as c FROM `{t}`").fetchone())['c']
|
|
# 가장 최근 데이터 시각
|
|
try:
|
|
cols = [dict(r)["Field"] for r in db.conn.execute(f"SHOW COLUMNS FROM `{t}`").fetchall()]
|
|
ts_col = next((c for c in ['snap_time','recv_ts','updated_at','created_at'] if c in cols), None)
|
|
if ts_col:
|
|
recent = db.conn.execute(f"SELECT MAX({ts_col}) as last FROM `{t}`").fetchone()
|
|
print(f" {t}: {cnt}건, 최근={dict(recent).get('last','?')}")
|
|
else:
|
|
print(f" {t}: {cnt}건")
|
|
except:
|
|
print(f" {t}: {cnt}건")
|
|
finally:
|
|
db.close()
|