hyowons
125 lines
5.1 KiB
Python
125 lines
5.1 KiB
Python
"""수급 백필 — ka10059 페이지네이션으로 종목별 투자자 순매수 다년치 → flow_history.sqlite.
|
||
|
||
백테스트가 수급(외국인·기관)을 반영하려면 과거 일자별 순매수가 필요한데 키움은 단일 콜 100일뿐
|
||
이라 연속조회로 깊이 받아 로컬 적재한다. 1회성(또는 가끔) 실행, idempotent(코드+일자 upsert).
|
||
|
||
스키마: flow(code, date, foreign_net, inst_net, indiv_net) 단위 천주, +순매수/−순매도.
|
||
backtest.load_flow 가 (foreign_net, inst_net) 를 읽는다.
|
||
|
||
rate limit(ka10059 stkinfo): 콜·종목 간 sleep + 429 백오프 재시도.
|
||
|
||
CLI:
|
||
python3 -m sim.backfill_flow [--pages N] [--force] [code ...]
|
||
--pages N : 종목당 페이지수(100거래일/페이지, 기본 6 ≈ 2.4년)
|
||
--force : 이미 충분히 쌓인 종목도 다시 받음
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
|
||
from . import config, universe
|
||
|
||
sys.path.insert(0, str(config.SCRIPTS))
|
||
import kiwoom_client as kc # noqa: E402
|
||
|
||
FLOW_DB = config.STATE_DIR / 'flow_history.sqlite'
|
||
DEFAULT_PAGES = 6 # 100거래일 × 6 ≈ 2.4년
|
||
PAGE_SLEEP = 0.35 # 페이지 간 간격
|
||
CODE_SLEEP = 0.5 # 종목 간 간격
|
||
RATE_BACKOFF = 3.0 # 429 시 추가 대기
|
||
|
||
|
||
def _conn() -> sqlite3.Connection:
|
||
FLOW_DB.parent.mkdir(parents=True, exist_ok=True)
|
||
c = sqlite3.connect(FLOW_DB, isolation_level=None)
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS flow (
|
||
code TEXT NOT NULL, date TEXT NOT NULL,
|
||
foreign_net INTEGER, inst_net INTEGER, indiv_net INTEGER,
|
||
PRIMARY KEY (code, date)
|
||
)""")
|
||
return c
|
||
|
||
|
||
def _is_rate_limit(resp: dict) -> bool:
|
||
return resp.get('return_code', 0) != 0 and (
|
||
'허용된 요청' in str(resp.get('return_msg') or '') or resp.get('return_code') == 5)
|
||
|
||
|
||
def fetch_code(code: str, pages: int = DEFAULT_PAGES) -> list[dict]:
|
||
"""ka10059 연속조회 — 최신부터 pages 페이지(페이지당 100거래일) 누적. 최신순 list."""
|
||
label = kc._default_account_label()
|
||
today = datetime.now(config.KST).strftime('%Y%m%d')
|
||
body = {'dt': today, 'stk_cd': kc._clean_code(code),
|
||
'amt_qty_tp': '2', 'trde_tp': '0', 'unit_tp': '1000'}
|
||
url = kc.base_url() + kc.ENDPOINT_STKINFO
|
||
cont, nk = 'N', ''
|
||
out: list[dict] = []
|
||
for _page in range(pages):
|
||
headers = kc.auth_headers(label, tr_id=kc.TR_INVESTOR_FLOW, cont_yn=cont, next_key=nk)
|
||
resp, hd = kc._http_post_full(url, body, headers)
|
||
if _is_rate_limit(resp):
|
||
time.sleep(RATE_BACKOFF)
|
||
resp, hd = kc._http_post_full(url, body, headers)
|
||
if resp.get('return_code', 0) != 0:
|
||
sys.stderr.write(f'[backfill] {code} page 실패: {resp.get("return_msg")}\n')
|
||
break
|
||
for r in resp.get('stk_invsr_orgn') or []:
|
||
d = (r.get('dt') or '').strip()
|
||
if d:
|
||
out.append({'date': d, 'foreign': kc._to_int(r.get('frgnr_invsr')),
|
||
'institution': kc._to_int(r.get('orgn')),
|
||
'individual': kc._to_int(r.get('ind_invsr'))})
|
||
cy = (hd.get('cont-yn') or '').strip().upper()
|
||
nkk = (hd.get('next-key') or '').strip()
|
||
if cy == 'Y' and nkk:
|
||
cont, nk = 'Y', nkk
|
||
time.sleep(PAGE_SLEEP)
|
||
else:
|
||
break
|
||
return out
|
||
|
||
|
||
def _existing_count(c: sqlite3.Connection, code: str) -> int:
|
||
return c.execute('SELECT COUNT(*) FROM flow WHERE code=?', (code,)).fetchone()[0]
|
||
|
||
|
||
def backfill(codes: list[str] | None = None, pages: int = DEFAULT_PAGES, force: bool = False) -> dict:
|
||
codes = codes or [e['code'] for e in universe.build_universe()]
|
||
c = _conn()
|
||
target = pages * 100
|
||
done = skipped = failed = 0
|
||
for code in codes:
|
||
if not force and _existing_count(c, code) >= target * 0.9:
|
||
skipped += 1
|
||
continue
|
||
rows = fetch_code(code, pages)
|
||
if not rows:
|
||
failed += 1
|
||
continue
|
||
c.executemany('INSERT OR REPLACE INTO flow(code,date,foreign_net,inst_net,indiv_net) VALUES (?,?,?,?,?)',
|
||
[(code, r['date'], r['foreign'], r['institution'], r['individual']) for r in rows])
|
||
done += 1
|
||
sys.stderr.write(f'[backfill] {code}: {len(rows)}행 ({rows[-1]["date"]}~{rows[0]["date"]})\n')
|
||
time.sleep(CODE_SLEEP)
|
||
total = c.execute('SELECT COUNT(*) FROM flow').fetchone()[0]
|
||
n_codes = c.execute('SELECT COUNT(DISTINCT code) FROM flow').fetchone()[0]
|
||
c.close()
|
||
return {'fetched': done, 'skipped': skipped, 'failed': failed,
|
||
'total_rows': total, 'codes_in_db': n_codes}
|
||
|
||
|
||
if __name__ == '__main__':
|
||
argv = sys.argv[1:]
|
||
force = '--force' in argv
|
||
pages = DEFAULT_PAGES
|
||
if '--pages' in argv:
|
||
pages = int(argv[argv.index('--pages') + 1])
|
||
explicit = [a for a in argv if a.isdigit() and len(a) >= 4]
|
||
res = backfill(codes=explicit or None, pages=pages, force=force)
|
||
print(f"백필 완료 — 신규 {res['fetched']} · 스킵 {res['skipped']} · 실패 {res['failed']} "
|
||
f"· DB {res['codes_in_db']}종목 {res['total_rows']}행")
|