"""종목별 데이터 수집 — 키움 시세·일봉·수급 + FnGuide/WISEreport 애널. 비용 절약: - 시세: get_watchlist_quotes 로 universe 전체 1콜 batch. - 일봉: daily_candles_cache (어제까지 캐시 재활용). - 수급: ka10059, state/sim/flow_cache/{code}.json 에 60분 TTL 캐시 (5일 net 은 장중 거의 불변). - 애널: fnguide/wisereport 클라이언트 자체 디스크 캐시(12h) 사용. - 호가: 체결 직전 종목만 on-demand. """ from __future__ import annotations import json import sys import time from . import config sys.path.insert(0, str(config.SCRIPTS)) import kiwoom_client as kc # noqa: E402 import daily_candles_cache as dcc # noqa: E402 FLOW_CACHE_DIR = config.STATE_DIR / 'flow_cache' FLOW_TTL_SEC = 3600 # 애널 크롤링 페이싱 — 캐시 미스(실제 네트워크)일 때만 종목 간 간격을 둬 fnguide/wisereport 연속요청 차단 회피. FNGUIDE_CACHE_DIR = config.WORKSPACE / 'state' / 'fnguide_cache' WISEREPORT_CACHE_DIR = config.WORKSPACE / 'state' / 'wisereport_cache' ANALYST_CACHE_TTL = 12 * 3600 # fnguide/wisereport 클라이언트 TTL과 동일 ANALYST_PACING_SEC = 0.4 def _cache_fresh(path, ttl: int = ANALYST_CACHE_TTL) -> bool: """캐시 파일 mtime이 TTL 이내면 True(히트 예상). 없으면 False(곧 네트워크).""" try: return (time.time() - path.stat().st_mtime) < ttl except OSError: return False def batch_quotes(codes: list[str]) -> dict[str, dict]: """universe 전체 오늘 시세 1콜 (가격·시고저·거래량).""" if not codes: return {} try: return kc.get_watchlist_quotes(codes) except Exception as e: sys.stderr.write(f'[data] batch_quotes 실패: {e}\n') return {} MIN_BARS = config.SMA_LONG + 5 # 지표 계산 최소 봉수 def candles(code: str, count: int = 80) -> list[dict]: """어제까지 일봉 (오름차순). 캐시 우선 — 충분히 쌓여 있으면 sqlite 직접 읽기(네트워크 X)로 ka10081 rate limit 회피. 캐시가 부족한 종목만 1회 네트워크 fetch(캐시 warm). 약간 stale 해도 SMA/ATR엔 무방하며 오늘 봉은 engine 이 라이브 시세로 따로 결합한다. """ cached_desc = dcc._select_latest(code, count) if len(cached_desc) >= MIN_BARS: return list(reversed(cached_desc)) try: return dcc.get_candles(code, count=count) except Exception as e: sys.stderr.write(f'[data] candles {code} 실패: {e}\n') return list(reversed(cached_desc)) def build_series(hist: list[dict], today_quote: dict | None) -> list[dict]: """어제까지 일봉 + 오늘 라이브 봉 결합. 오늘 시세 없으면 hist 그대로.""" if not today_quote or not today_quote.get('price'): return hist today_bar = { 'date': 'TODAY', 'open': today_quote.get('open') or today_quote['price'], 'high': today_quote.get('high') or today_quote['price'], 'low': today_quote.get('low') or today_quote['price'], 'close': today_quote['price'], 'volume': today_quote.get('volume') or 0, } return hist + [today_bar] def _flow_cache_path(code: str): return FLOW_CACHE_DIR / f'{code}.json' def investor_flow(code: str, days: int = config.FLOW_DAYS) -> list[dict]: """최근 days 일 외국인·기관·개인 순매수 (최신순). 60분 TTL 캐시.""" path = _flow_cache_path(code) now = time.time() if path.exists(): try: cached = json.loads(path.read_text()) if now - cached.get('ts', 0) < FLOW_TTL_SEC: return cached.get('rows', [])[:days] except Exception: pass try: rows = kc.get_investor_flow(code, days=max(days, config.FLOW_DAYS)) except Exception as e: sys.stderr.write(f'[data] investor_flow {code} 실패: {e}\n') return [] FLOW_CACHE_DIR.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({'ts': now, 'rows': rows}, ensure_ascii=False)) return rows[:days] def flow_net(rows: list[dict]) -> dict: """수급 행 → 외국인·기관 누적 순매수(천주). 양수=순매수.""" foreign = sum(r.get('foreign', 0) for r in rows) institution = sum(r.get('institution', 0) for r in rows) return {'foreign': foreign, 'institution': institution, 'days': len(rows)} def analyst(code: str, price: int) -> dict | None: """애널 게이트/가점용. 캐시 미스(네트워크 발생) 시에만 페이싱 — 연속 크롤링 차단 회피.""" fresh = (_cache_fresh(FNGUIDE_CACHE_DIR / f'{code}.json') and _cache_fresh(WISEREPORT_CACHE_DIR / f'{code}.json')) try: return _analyst_query(code, price) finally: if not fresh: time.sleep(ANALYST_PACING_SEC) def _analyst_query(code: str, price: int) -> dict | None: """target_price·opinion·upside·revision·surprise. ETF·실패 시 None.""" target_price = opinion = None try: import fnguide_client as fg f = fg.get_fundamentals(code) cons = (f or {}).get('consensus') or {} target_price = cons.get('target_price') opinion = cons.get('opinion') except Exception: pass revision_up = surprise_pos = None try: import wisereport_client as wr c = wr.get_consensus(code) if c: rev = c.get('revision') or {} if rev.get('target_change_pct') is not None: revision_up = rev['target_change_pct'] > 0 if not target_price and rev.get('target_last'): target_price = rev['target_last'] surp = c.get('surprise') or {} items = surp.get('items') if isinstance(surp, dict) else None op_sp = (items or {}).get('영업이익', {}).get('fy0_surprise_pct') if items else None if op_sp is not None: surprise_pos = op_sp > 0 except Exception: pass if not target_price and opinion is None and revision_up is None: return None upside = ((target_price / price - 1) * 100) if (target_price and price) else None return { 'target_price': target_price, 'opinion': opinion, 'upside_pct': upside, 'revision_up': revision_up, 'surprise_pos': surprise_pos, } def quote_book(code: str) -> dict | None: """체결가용 호가 (매도1/매수1). 실패 시 None.""" try: return kc.get_quote_book(code) except Exception as e: sys.stderr.write(f'[data] quote_book {code} 실패: {e}\n') return None