hyowons
205 lines
7.5 KiB
Python
205 lines
7.5 KiB
Python
"""종목별 데이터 수집 — 키움 시세·일봉·수급 + 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 캐시 + 스캔 내 메모."""
|
|
mk = (code, days)
|
|
if mk in _FLOW_MEMO_SCAN:
|
|
return _FLOW_MEMO_SCAN[mk]
|
|
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:
|
|
rows = cached.get('rows', [])[:days]
|
|
_FLOW_MEMO_SCAN[mk] = rows
|
|
return rows
|
|
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))
|
|
_FLOW_MEMO_SCAN[mk] = rows[:days]
|
|
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:
|
|
"""애널 게이트/가점용. 캐시 미스(네트워크 발생) 시에만 페이싱 + 스캔 내 메모."""
|
|
if code in _ANL_MEMO_SCAN:
|
|
return _ANL_MEMO_SCAN[code]
|
|
fresh = (_cache_fresh(FNGUIDE_CACHE_DIR / f'{code}.json')
|
|
and _cache_fresh(WISEREPORT_CACHE_DIR / f'{code}.json'))
|
|
try:
|
|
out = _analyst_query(code, price)
|
|
_ANL_MEMO_SCAN[code] = out
|
|
return out
|
|
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,
|
|
}
|
|
|
|
|
|
# 스캔 1회 동안 호가 공유 — 같은 스캔에서 같은 종목은 같은 체결가 (선수 간 체결 공정성 + API 절감).
|
|
# engine._gather 가 스캔 시작마다 reset.
|
|
_BOOK_MEMO: dict = {}
|
|
_FLOW_MEMO_SCAN: dict = {} # 스캔 1회 동안 종목별 수급 공유 (파라미터 무관)
|
|
_ANL_MEMO_SCAN: dict = {} # 스캔 1회 동안 종목별 애널 공유 (파라미터 무관)
|
|
|
|
|
|
def reset_book_memo():
|
|
_BOOK_MEMO.clear()
|
|
_FLOW_MEMO_SCAN.clear()
|
|
_ANL_MEMO_SCAN.clear()
|
|
|
|
|
|
def quote_book(code: str) -> dict | None:
|
|
"""체결가용 호가 (매도1/매수1). 스캔 내 종목당 1회 조회 후 공유. 실패 시 None."""
|
|
if code in _BOOK_MEMO:
|
|
return _BOOK_MEMO[code]
|
|
try:
|
|
book = kc.get_quote_book(code)
|
|
except Exception as e:
|
|
sys.stderr.write(f'[data] quote_book {code} 실패: {e}\n')
|
|
book = None
|
|
_BOOK_MEMO[code] = book
|
|
return book
|