328 lines
13 KiB
Python
328 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
fetch_stock_meta.py — 키움 테마 API로 stock_meta 자동 채우기
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
[역할]
|
|
- 키움 REST API (ka90001 / ka90002) 로 전체 테마 목록 + 구성 종목을 조회
|
|
- stock_meta 테이블에 종목별 테마명·테마순위 저장
|
|
- 이후 스캘핑/꼬리잡기 매수 시 "이 종목의 테마 60분봉 RSI" 확인 가능
|
|
|
|
[사용방법]
|
|
python3 fetch_stock_meta.py # 전체 테마 스캔 (기본)
|
|
python3 fetch_stock_meta.py --top 30 # 상위 30개 테마만 (빠름)
|
|
python3 fetch_stock_meta.py --theme AI # 특정 테마명 검색
|
|
|
|
[systemd timer 등록 예시 — 장 마감 후 1일 1회]
|
|
systemctl --user enable kis_fetch_meta.timer
|
|
|
|
[키움 API 엔드포인트]
|
|
ka90001: 테마그룹별 조회 (테마 목록)
|
|
ka90002: 테마구성종목 조회 (각 테마 내 종목 리스트)
|
|
"""
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
import time
|
|
import random
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import requests
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
|
|
from database import TradeDB
|
|
# _get_kiwoom_creds: DB에서 키움 키 꺼내기
|
|
# _get_kiwoom_token_cached: 23시간 캐시 토큰 (au10001 rate limit 방지)
|
|
from kis_ws import _get_kiwoom_creds, _get_kiwoom_token_cached
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="[%(asctime)s] %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
)
|
|
logger = logging.getLogger("FetchStockMeta")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# 키움 토큰: kis_ws._get_kiwoom_token_cached() 재사용
|
|
# (23시간 모듈 레벨 캐시 → au10001 rate limit 방지)
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# ka90001: 테마 목록 조회
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
def fetch_theme_list(
|
|
token: str,
|
|
app_key: str,
|
|
app_secret: str,
|
|
is_mock: bool,
|
|
theme_name_filter: str = "",
|
|
) -> list:
|
|
"""
|
|
키움 ka90001 API 로 전체 테마 그룹 목록 조회.
|
|
|
|
Returns:
|
|
[{'thema_grp_cd': '100', 'thema_nm': 'AI반도체', 'stk_num': '15', ...}, ...]
|
|
"""
|
|
domain = "mockapi.kiwoom.com" if is_mock else "api.kiwoom.com"
|
|
url = f"https://{domain}/api/dostk/thme"
|
|
headers = {
|
|
"content-type": "application/json;charset=UTF-8",
|
|
"appkey": app_key,
|
|
"appsecret": app_secret,
|
|
"authorization": f"Bearer {token}",
|
|
"api-id": "ka90001",
|
|
"cont-yn": "N",
|
|
"next-key": "",
|
|
}
|
|
# qry_tp=0: 전체검색 / date_tp=1: 1일 / flu_pl_amt_tp=1: 상위기간수익률 / stex_tp=1: KRX
|
|
body = {
|
|
"qry_tp": "1" if theme_name_filter else "0",
|
|
"thema_nm": theme_name_filter,
|
|
"stk_cd": "",
|
|
"date_tp": "1",
|
|
"flu_pl_amt_tp":"1",
|
|
"stex_tp": "1",
|
|
}
|
|
|
|
themes = []
|
|
while True:
|
|
try:
|
|
resp = requests.post(url, json=body, headers=headers, timeout=15)
|
|
data = resp.json()
|
|
except Exception as e:
|
|
logger.warning("ka90001 조회 실패: %s", e)
|
|
break
|
|
|
|
groups = data.get("thema_grp") or []
|
|
themes.extend(groups)
|
|
|
|
cont_yn = str(resp.headers.get("cont-yn", "N")).upper()
|
|
next_key = str(resp.headers.get("next-key", "")).strip()
|
|
if cont_yn != "Y" or not next_key:
|
|
break
|
|
headers["cont-yn"] = cont_yn
|
|
headers["next-key"] = next_key
|
|
time.sleep(0.4)
|
|
|
|
logger.info("✅ 테마 목록 조회 완료: %d개", len(themes))
|
|
return themes
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# ka90002: 테마 구성 종목 조회
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
def fetch_theme_stocks(
|
|
token: str,
|
|
app_key: str,
|
|
app_secret: str,
|
|
is_mock: bool,
|
|
thema_grp_cd: str,
|
|
) -> list:
|
|
"""
|
|
키움 ka90002 API 로 특정 테마의 구성 종목 조회.
|
|
|
|
Returns:
|
|
[{'stk_cd': '005930', 'stk_nm': '삼성전자', ...}, ...]
|
|
"""
|
|
domain = "mockapi.kiwoom.com" if is_mock else "api.kiwoom.com"
|
|
url = f"https://{domain}/api/dostk/thme"
|
|
headers = {
|
|
"content-type": "application/json;charset=UTF-8",
|
|
"appkey": app_key,
|
|
"appsecret": app_secret,
|
|
"authorization": f"Bearer {token}",
|
|
"api-id": "ka90002",
|
|
"cont-yn": "N",
|
|
"next-key": "",
|
|
}
|
|
body = {
|
|
"thema_grp_cd": thema_grp_cd,
|
|
"stex_tp": "1",
|
|
"date_tp": "1",
|
|
}
|
|
|
|
stocks = []
|
|
while True:
|
|
try:
|
|
resp = requests.post(url, json=body, headers=headers, timeout=15)
|
|
data = resp.json()
|
|
except Exception as e:
|
|
logger.warning("ka90002 조회 실패 (테마%s): %s", thema_grp_cd, e)
|
|
break
|
|
|
|
items = data.get("thema_comp_stk") or []
|
|
stocks.extend(items)
|
|
|
|
cont_yn = str(resp.headers.get("cont-yn", "N")).upper()
|
|
next_key = str(resp.headers.get("next-key", "")).strip()
|
|
if cont_yn != "Y" or not next_key:
|
|
break
|
|
headers["cont-yn"] = cont_yn
|
|
headers["next-key"] = next_key
|
|
time.sleep(0.35)
|
|
|
|
return stocks
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# 종목코드로 시장구분 추정 (KOSPI / KOSDAQ / ETF)
|
|
# 정확한 구분은 별도 API 필요, 여기선 휴리스틱
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
def _guess_market(code: str) -> str:
|
|
"""
|
|
KOSPI ETF 코드 범위 (069XXX, 102XXX 등) / 기타 판별은 어려워
|
|
가장 보수적으로: 6자리 숫자 기준 추정.
|
|
- 0XXXXX, 1XXXXX: 주로 KOSPI (삼성 005930, 현대 005380)
|
|
- 2XXXXX, 3XXXXX: 주로 KOSDAQ (카카오 035720)
|
|
- ETF는 069XXX, 102XXX, 114XXX, 122XXX, 133XXX, 229XXX 등
|
|
완벽하지 않음 — 향후 KIS FHKST01010100 API로 정확히 수정 가능
|
|
"""
|
|
if not code or len(code) != 6 or not code.isdigit():
|
|
return "Q"
|
|
n = int(code)
|
|
# 대표적 ETF 코드 대역
|
|
etf_ranges = [
|
|
(69000, 70000), (102000, 103000), (114000, 115000),
|
|
(122000, 123000), (133000, 134000), (229000, 230000),
|
|
(252000, 253000), (261000, 262000), (278000, 279000),
|
|
(364000, 365000),
|
|
]
|
|
for lo, hi in etf_ranges:
|
|
if lo <= n < hi:
|
|
return "E"
|
|
# KOSPI 범위 (0~200000 사이, 대략)
|
|
if n < 200000:
|
|
return "K"
|
|
return "Q"
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
# 메인: 테마 → 종목 → stock_meta 저장
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
def run(top_n: int = 0, theme_filter: str = "", dry_run: bool = False):
|
|
"""
|
|
Args:
|
|
top_n : 상위 N개 테마만 처리 (0=전체)
|
|
theme_filter : 특정 테마명 검색 (예: "AI", "반도체")
|
|
dry_run : DB 저장 없이 출력만
|
|
"""
|
|
db = TradeDB()
|
|
|
|
# 1. DB에서 키움 키 꺼내기
|
|
kw_key, kw_secret, kw_mock = _get_kiwoom_creds(db)
|
|
if not kw_key or not kw_secret:
|
|
logger.error("❌ 키움 앱키/시크릿 없음 → update_env_simple.py 로 KIWOOM_APP_KEY_REAL 설정 필요")
|
|
sys.exit(1)
|
|
|
|
mode_str = "모의" if kw_mock else "실전"
|
|
logger.info("🔑 키움 키 확인 완료 [%s] (앞8자: %s…)", mode_str, kw_key[:8])
|
|
|
|
# 2. 키움 OAuth 토큰 (캐시 사용)
|
|
token = _get_kiwoom_token_cached(kw_key, kw_secret, kw_mock)
|
|
if not token:
|
|
sys.exit(1)
|
|
logger.info("✅ 키움 토큰 확인 완료")
|
|
|
|
# 3. 테마 목록 조회
|
|
themes = fetch_theme_list(token, kw_key, kw_secret, kw_mock, theme_filter)
|
|
if not themes:
|
|
logger.error("❌ 테마 목록 없음 (API 응답 확인 필요)")
|
|
sys.exit(1)
|
|
|
|
if top_n > 0:
|
|
themes = themes[:top_n]
|
|
logger.info("📋 상위 %d개 테마만 처리", top_n)
|
|
|
|
# 4. 테마별 구성 종목 조회 + stock_meta 저장
|
|
total_saved = 0
|
|
total_skipped = 0
|
|
|
|
for idx, tg in enumerate(themes, 1):
|
|
grp_cd = str(tg.get("thema_grp_cd") or "").strip()
|
|
theme_nm = str(tg.get("thema_nm") or "").strip()
|
|
stk_num = str(tg.get("stk_num") or "0").strip()
|
|
|
|
if not grp_cd or not theme_nm:
|
|
continue
|
|
|
|
logger.info("[%d/%d] 테마: %s (%s종목)", idx, len(themes), theme_nm, stk_num)
|
|
|
|
stocks = fetch_theme_stocks(token, kw_key, kw_secret, kw_mock, grp_cd)
|
|
if not stocks:
|
|
logger.debug(" → 구성 종목 없음 (스킵)")
|
|
total_skipped += 1
|
|
continue
|
|
|
|
for rank, stk in enumerate(stocks, 1):
|
|
code = str(stk.get("stk_cd") or "").strip()
|
|
name = str(stk.get("stk_nm") or "").strip()
|
|
if not code or len(code) != 6:
|
|
continue
|
|
|
|
# 테마 내 순위: 상위 3위까지 핵심주(1), 4~10위 연관주(2), 나머지 주변주(3)
|
|
theme_rank = 1 if rank <= 3 else (2 if rank <= 10 else 3)
|
|
market = _guess_market(code)
|
|
|
|
if dry_run:
|
|
print(f" {code} {name:12s} market={market} theme={theme_nm} rank={theme_rank}")
|
|
else:
|
|
ok = db.upsert_stock_meta(
|
|
code = code,
|
|
name = name,
|
|
market = market,
|
|
sector = "", # 섹터는 별도 API 필요 (KIS FHKST01010100)
|
|
theme = theme_nm,
|
|
theme_rank = theme_rank,
|
|
)
|
|
if ok:
|
|
total_saved += 1
|
|
if dry_run:
|
|
total_saved += 1
|
|
|
|
time.sleep(random.uniform(0.3, 0.6)) # API 레이트리밋
|
|
|
|
# 5. 결과 요약
|
|
if not dry_run:
|
|
total_stocks = db.conn.execute("SELECT COUNT(*) AS n FROM stock_meta").fetchone()['n']
|
|
logger.info("=" * 50)
|
|
logger.info("✅ 저장 완료: %d건 | stock_meta 전체: %d종목", total_saved, total_stocks)
|
|
|
|
# 테마별 종목 수 TOP 10
|
|
rows = db.conn.execute("""
|
|
SELECT theme, COUNT(*) AS cnt
|
|
FROM stock_meta
|
|
WHERE theme IS NOT NULL AND theme != ''
|
|
GROUP BY theme
|
|
ORDER BY cnt DESC
|
|
LIMIT 10
|
|
""").fetchall()
|
|
logger.info("📊 테마별 종목 수 TOP 10:")
|
|
for r in rows:
|
|
logger.info(" %-25s: %d종목", r['theme'], r['cnt'])
|
|
else:
|
|
logger.info("(dry-run 모드: DB 저장 안 함, 조회만 %d건)", total_saved)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
ap = argparse.ArgumentParser(description="키움 테마 API → stock_meta DB 자동 채우기")
|
|
ap.add_argument("--top", type=int, default=0, help="상위 N개 테마만 처리 (0=전체)")
|
|
ap.add_argument("--theme", type=str, default="", help="특정 테마명 검색 (예: AI, 반도체)")
|
|
ap.add_argument("--dry-run",action="store_true", help="DB 저장 없이 출력만")
|
|
args = ap.parse_args()
|
|
|
|
run(
|
|
top_n = args.top,
|
|
theme_filter= args.theme,
|
|
dry_run = args.dry_run,
|
|
)
|