예제 소스
한국투자증권 API와 LLM 주식 응용 프로그램 · 파일 19개
surge_monitor.py
# -*- coding: utf-8 -*-
"""
1분봉 기반 급등락 실시간 모니터링 데몬 (실거래 아님, 알림/로그 전용).
· KIS Open API 의 당일 1분봉(`inquire-time-itemchartprice`)을 일정 주기로 폴링
· 종목 watchlist 가 지정되지 않으면 fetch_positive_news 모듈로 호재 후보 자동 선정
(paper_trade 와 동일하게 KRX 의 투자주의/관리종목 등을 한 번 더 배제)
· 급등/급락 판정 (모두 동시 만족):
1) 최근 N분 누적 수익률 ≥ +X% (급등) 또는 ≤ -X% (급락)
2) 마지막 1분봉 거래량 ≥ 직전 20봉 평균의 K 배
3) 마지막 1분봉 캔들 폭 ((High-Low)/Close) ≥ A%
· 무한 루프 (Ctrl-C 종료) 로 동작하며, 매 초 같은 줄에 heartbeat 표시
· 트리거 발생 시 별도 줄로 알림만 화면에 출력 (CSV 저장 없음)
사용법
$ python surge_monitor.py # 호재 7종목 자동 선정 후 모니터링
$ python surge_monitor.py --codes 005930,000660,035420
$ python surge_monitor.py --interval-sec 60 --surge-pct 5 --plunge-pct -5
$ python surge_monitor.py --vol-spike 3 --candle-amp-pct 0.5
사전 준비
- .env / .env.local 에 APP_KEY, APP_SECRET 설정
- stock_list.csv (호재 자동 선정 시 필요)
"""
from __future__ import annotations
import argparse
import re
import socket
import sys
import time
from datetime import date, datetime, timedelta
from pathlib import Path
import pandas as pd
from requests.exceptions import ConnectionError as RequestsConnectionError
from requests.exceptions import Timeout as RequestsTimeout
try:
import FinanceDataReader as fdr
except ImportError as exc: # pragma: no cover
print("FinanceDataReader 가 필요합니다. `pip install finance-datareader`", file=sys.stderr)
raise SystemExit(1) from exc
# 같은 폴더의 KIS 클라이언트 / 호재 선정 모듈 재사용
from kis_client import KISClient
from paper_trade import is_trading_day, pick_positive_stocks
from build_stock_list import EXCLUDE_DEPT_KEYWORDS, is_preferred_stock
HERE = Path(__file__).resolve().parent
API_MINUTE_ITEMCHART = "********"
TR_MINUTE_ITEMCHART = "FHKST03010200"
DEFAULTS = {
"interval_sec": 30,
"surge_pct": 3.0,
"plunge_pct": -3.0,
"vol_spike": 5.0,
"candle_amp_pct": 1.0,
"ret_window": 5, # N분 누적 수익률
"vol_ma_window": 20, # 거래량 이동평균 봉 수
# watchlist 옵션
"top_n": 150, # >0 이면 거래대금 상위 N 모드. 0 이면 호재 자동 선정 사용
"top_by": "amount", # amount | marcap | volume
"per_call_sleep": 0.10, # 종목 1건 호출 후 sleep (rate limit 보호, 초당 ~10건)
"rate_limit_backoff": 1.0, # EGW00201(초당 건수 초과) 발생 시 추가 대기 초
# 네트워크 일시 장애(DNS 실패/연결 타임아웃 등) 자동 재시도
"net_retries": 3,
"net_retry_backoff": 2.0,
# paper_trade 측 옵션 (호재 자동 선정 시)
"sections": "401,402",
"pages": 5,
"min_score": 2,
"max_picks": 7,
"delay": 0.5,
}
# ──────────────────────────────────────────────────────────────────────────────
# KIS 1분봉 조회
# ──────────────────────────────────────────────────────────────────────────────
# ──────────────────────────────────────────────────────────────────────────────
# 네트워크 일시 장애 자동 재시도
# ──────────────────────────────────────────────────────────────────────────────
# 일시적 네트워크 오류로 간주할 예외들
_NET_TRANSIENT = (
RequestsConnectionError, # DNS 실패 / TCP RST / 호스트 unreachable 등
RequestsTimeout, # connect/read timeout
socket.gaierror, # getaddrinfo 실패 (DNS)
socket.timeout,
OSError, # 그 외 저수준 OS 에러 (Network is unreachable 등)
)
def call_with_net_retry(fn, *args, retries: int = 3, backoff: float = 2.0, label: str = ""):
"""일시적 네트워크 오류는 자동 재시도. 그 외 예외는 즉시 raise."""
last_exc: Exception | None = None
for i in range(max(1, retries)):
try:
return fn(*args)
except _NET_TRANSIENT as exc:
last_exc = exc
wait = backoff * (i + 1) # 2s, 4s, 6s ...
newline()
print(
f" · 네트워크 오류({label}) {i + 1}/{retries} → {wait:.1f}s 후 재시도: {exc}",
file=sys.stderr,
)
time.sleep(wait)
# 모든 재시도 실패
if last_exc is not None:
raise last_exc
def fetch_minute_candles(client: KISClient, code: str, hour: str | None = None) -> pd.DataFrame:
"""KIS 당일 1분봉 30개 조회. hour=None 이면 현재 시각 기준 가장 최근 30봉."""
params = {
"FID_ETC_CLS_CODE": "",
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": code,
"FID_INPUT_HOUR_1": hour or datetime.now().strftime("%H%M%S"),
"FID_PW_DATA_INCU_YN": "N",
}
body = client.get_json(API_MINUTE_ITEMCHART, TR_MINUTE_ITEMCHART, params)
rows = body.get("output2") or []
if not rows:
return pd.DataFrame(columns=["Open", "High", "Low", "Close", "Volume"])
df = pd.DataFrame(rows)
rename_map = {
"stck_bsop_date": "Date",
"stck_cntg_hour": "Time",
"stck_oprc": "Open",
"stck_hgpr": "High",
"stck_lwpr": "Low",
"stck_prpr": "Close", # 분봉은 stck_prpr 이 해당 봉의 종가
"cntg_vol": "Volume",
}
df = df.rename(columns=rename_map)
needed = ["Date", "Time", "Open", "High", "Low", "Close", "Volume"]
for c in needed:
if c not in df.columns:
df[c] = pd.NA
df["DateTime"] = pd.to_datetime(
df["Date"].astype(str) + df["Time"].astype(str),
format="%Y%m%d%H%M%S",
errors="coerce",
)
for col in ("Open", "High", "Low", "Close", "Volume"):
df[col] = pd.to_numeric(df[col], errors="coerce")
return (
df.dropna(subset=["DateTime"])
.set_index("DateTime")
.sort_index()[["Open", "High", "Low", "Close", "Volume"]]
)
# ──────────────────────────────────────────────────────────────────────────────
# 급등락 판정
# ──────────────────────────────────────────────────────────────────────────────
def detect_event(
df: pd.DataFrame,
*,
surge_pct: float,
plunge_pct: float,
vol_spike_th: float,
candle_amp_th_pct: float,
ret_window: int,
vol_ma_window: int,
) -> dict | None:
if len(df) < max(ret_window + 1, 5):
return None
last = df.iloc[-1]
base = df.iloc[-ret_window - 1]
if not base["Close"] or base["Close"] <= 0 or last["Close"] <= 0:
return None
ret_pct = (last["Close"] - base["Close"]) / base["Close"] * 100.0
candle_amp_pct = (last["High"] - last["Low"]) / last["Close"] * 100.0
# 직전 봉들의 거래량 이동평균 (마지막 봉 자체는 제외)
vol_ma = (
df["Volume"]
.iloc[:-1]
.rolling(vol_ma_window, min_periods=5)
.mean()
)
vma = float(vol_ma.iloc[-1]) if len(vol_ma) else 0.0
vol_spike = (last["Volume"] / vma) if vma and vma > 0 else 0.0
event = None
if (
ret_pct >= surge_pct
and vol_spike >= vol_spike_th
and candle_amp_pct >= candle_amp_th_pct
and last["Close"] >= last["Open"]
):
event = "SURGE"
elif (
ret_pct <= plunge_pct
and vol_spike >= vol_spike_th
and candle_amp_pct >= candle_amp_th_pct
and last["Close"] <= last["Open"]
):
event = "PLUNGE"
return {
"bar_time": df.index[-1].to_pydatetime(),
"close": float(last["Close"]),
"ret_pct": float(ret_pct),
"vol_spike": float(vol_spike),
"candle_amp_pct": float(candle_amp_pct),
"event": event,
}
# ──────────────────────────────────────────────────────────────────────────────
# Watchlist 결정
# ──────────────────────────────────────────────────────────────────────────────
def fetch_top_codes(top_n: int, by: str = "amount") -> list[tuple[str, str]]:
"""KRX 전종목 → 매매주의/관리/거래정지 등 위험표시 배제 → 상위 N 종목.
by: 'amount' (거래대금) | 'marcap' (시가총액) | 'volume' (거래량)
"""
print(f"[watchlist] KRX 전종목 조회 + 위험표시·우선주 배제 + 상위 {top_n}개 추출 (by={by})")
df = fdr.StockListing("KRX")
if df is None or df.empty:
return []
# 1) 6자리 숫자 종목코드만 (ETF/ETN/ELW 등 일부 비주식 제외)
if "Code" in df.columns:
df = df[df["Code"].astype(str).str.match(r"^\d{6}$", na=False)].copy()
before = len(df)
# 2) 우선주 배제 (종목명: …우, 2우B, 4우(전환) 등)
if "Name" in df.columns:
pref_mask = df.apply(
lambda r: is_preferred_stock(r.get("Name"), r.get("Code")),
axis=1,
)
df = df[~pref_mask].copy()
print(f" · 우선주 배제: {int(pref_mask.sum())}건")
# 3) Dept 컬럼 위험 키워드 배제
if "Dept" in df.columns:
excl_re = "|".join(map(re.escape, EXCLUDE_DEPT_KEYWORDS))
mask_excl = df["Dept"].fillna("").str.contains(excl_re, regex=True)
df = df[~mask_excl].copy()
after = len(df)
print(f" · 위험표시 배제: {before} → {after} (-{before - after})")
# 4) 정렬 컬럼 결정 (없으면 fallback)
sort_map = {"amount": "Amount", "marcap": "Marcap", "volume": "Volume"}
sort_col = sort_map.get(by)
if sort_col is None or sort_col not in df.columns:
for cand in ("Amount", "Marcap", "Volume"):
if cand in df.columns:
sort_col = cand
break
if sort_col is None:
print(" · 정렬 가능한 컬럼이 없어 그대로 상위 N 추출", file=sys.stderr)
else:
df[sort_col] = pd.to_numeric(df[sort_col], errors="coerce")
df = df.sort_values(sort_col, ascending=False, na_position="last")
print(f" · 정렬 기준: {sort_col} 내림차순")
df = df.head(top_n)
out: list[tuple[str, str]] = []
for _, row in df.iterrows():
code = str(row.get("Code", "")).strip()
name = str(row.get("Name", "")).strip()
if code and name:
out.append((code, name))
return out
def _filter_preferred_pairs(pairs: list[tuple[str, str]], *, label: str) -> list[tuple[str, str]]:
"""watchlist (코드, 종목명) 에서 우선주 제거."""
kept: list[tuple[str, str]] = []
dropped = 0
for code, name in pairs:
if is_preferred_stock(name, code):
dropped += 1
continue
kept.append((code, name))
if dropped:
print(f" · 우선주 배제({label}): {dropped}건 → {len(kept)}종목")
return kept
def resolve_watchlist(args) -> list[tuple[str, str]]:
"""(종목코드, 종목명) 리스트 반환. 우선순위: --codes > --top-n > 호재 자동 선정."""
if args.codes:
items = [c.strip() for c in args.codes.split(",") if c.strip()]
try:
df = pd.read_csv(HERE / "stock_list.csv", dtype={"종목코드": str})
name_map = dict(zip(df["종목코드"].astype(str), df["종목명"].astype(str)))
except Exception:
name_map = {}
print(f"[watchlist] --codes 지정: {len(items)} 종목")
return _filter_preferred_pairs(
[(c, name_map.get(c, c)) for c in items],
label="--codes",
)
if args.top_n and args.top_n > 0:
return fetch_top_codes(args.top_n, by=args.top_by)
print("[watchlist] --codes / --top-n 미지정 → 호재 자동 선정")
picks = pick_positive_stocks(args)
return _filter_preferred_pairs(
[(p["종목코드"], p["종목명"]) for p in picks],
label="호재",
)
# ──────────────────────────────────────────────────────────────────────────────
# Heartbeat 출력
# ──────────────────────────────────────────────────────────────────────────────
def heartbeat(line: str) -> None:
sys.stdout.write("\r\033[K" + line)
sys.stdout.flush()
def newline() -> None:
sys.stdout.write("\n")
sys.stdout.flush()
# ──────────────────────────────────────────────────────────────────────────────
# 모니터 메인 루프
# ──────────────────────────────────────────────────────────────────────────────
MARKET_OPEN = datetime.strptime("09:00", "%H:%M").time()
MARKET_CLOSE = datetime.strptime("15:30", "%H:%M").time()
def is_kr_market_hours(now: datetime) -> bool:
"""대략 한국 정규장 시간 (09:00 ~ 15:30) 안인지."""
t = now.time()
return MARKET_OPEN <= t <= MARKET_CLOSE
def next_market_open(now: datetime) -> datetime:
"""다음 정규장 시작 datetime 반환 (거래일까지 건너뜀)."""
today = now.date()
today_open = datetime.combine(today, MARKET_OPEN)
if now < today_open and is_trading_day(today):
return today_open
d = today + timedelta(days=1)
for _ in range(14):
if is_trading_day(d):
return datetime.combine(d, MARKET_OPEN)
d += timedelta(days=1)
return datetime.combine(today + timedelta(days=1), MARKET_OPEN) # fallback
def wait_until_market_open(watchlist_size: int) -> None:
"""장이 닫혀있는 동안 KIS 조회 없이 heartbeat 만 표시하며 대기."""
while True:
now = datetime.now()
today = now.date()
if is_trading_day(today) and is_kr_market_hours(now):
return
target = next_market_open(now)
left = (target - now).total_seconds()
if left <= 0:
return
td = is_trading_day(today)
status = "휴장일" if not td else ("장 시작 전" if now.time() < MARKET_OPEN else "장 마감 후")
h = int(left // 3600)
m = int((left % 3600) // 60)
s = int(left % 60)
heartbeat(
f"[{now:%Y-%m-%d %H:%M:%S}] {status} 대기 중 "
f"watch={watchlist_size} "
f"다음 장 개시 {target:%Y-%m-%d %H:%M} "
f"남은시간 {h:02d}:{m:02d}:{s:02d}"
)
time.sleep(1)
def run_monitor(args) -> int:
today = date.today()
watchlist = resolve_watchlist(args)
if not watchlist:
print("watchlist 가 비었습니다. 종료.", file=sys.stderr)
return 1
# watchlist 가 크면 폴링 1회 예상 시간 안내
est_cycle_sec = len(watchlist) * args.per_call_sleep + len(watchlist) * 0.05
if est_cycle_sec > args.interval_sec:
print(
f"[warn] watchlist 가 큽니다 (예상 폴링 {est_cycle_sec:.0f}초 > "
f"interval {args.interval_sec}초). --interval-sec 을 더 크게 설정하시는 걸 권장합니다.",
file=sys.stderr,
)
client = KISClient()
print(f"[init] KIS 액세스 토큰 발급 중...")
client.issue_access_token()
print()
print(f"=== 급등락 모니터 시작 @ {datetime.now():%Y-%m-%d %H:%M:%S} ===")
print(f" · 거래일 : {is_trading_day(today)}")
print(f" · watchlist : {len(watchlist)} 종목")
if len(watchlist) <= 30:
for c, n in watchlist:
print(f" - {c} {n}")
else:
head = watchlist[:10]
for c, n in head:
print(f" - {c} {n}")
print(f" ... (+{len(watchlist) - 10} 종목 생략)")
print(f" · 폴링 주기 : {args.interval_sec}초 "
f"(per-call sleep {args.per_call_sleep}s, 예상 1cycle ≈ {est_cycle_sec:.0f}s)")
print(
f" · 트리거 임계 : 수익률 ≥ +{args.surge_pct}% / ≤ {args.plunge_pct}% "
f"거래량 ≥ {args.vol_spike}배 캔들폭 ≥ {args.candle_amp_pct}% "
f"(누적 윈도우 {args.ret_window}분)"
)
print(f" · Ctrl-C 로 종료 (검출 결과는 화면에만 출력 / CSV 저장 없음)")
print()
last_event_bar: dict[str, datetime] = {} # 동일 봉 중복 알림 방지
cycle = 0
total_events = 0
try:
while True:
# 거래일/거래시간이 아니면 KIS 호출하지 않고 대기 (heartbeat 만 표시)
now0 = datetime.now()
if not (is_trading_day(now0.date()) and is_kr_market_hours(now0)):
wait_until_market_open(len(watchlist))
# 장 개시 직후 안내 한 줄 출력하고 다음 cycle 진입
newline()
print(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] 정규장 개시 — 모니터링 재개")
continue
cycle += 1
cycle_start = datetime.now()
mh_label = "정규장"
for code, name in watchlist:
try:
df = call_with_net_retry(
fetch_minute_candles, client, code,
retries=args.net_retries,
backoff=args.net_retry_backoff,
label=f"{code} {name}",
)
except Exception as exc:
msg = str(exc)
newline()
print(f" · {code} {name} 분봉 조회 실패: {msg}", file=sys.stderr)
# KIS '초당 거래건수 초과' (EGW00201) → 잠깐 backoff 후 다음 종목
if "EGW00201" in msg or "초당 거래건수" in msg:
if args.rate_limit_backoff > 0:
time.sleep(args.rate_limit_backoff)
elif args.per_call_sleep > 0:
time.sleep(args.per_call_sleep)
continue
# KIS rate limit 보호
if args.per_call_sleep > 0:
time.sleep(args.per_call_sleep)
if df.empty:
continue
det = detect_event(
df,
surge_pct=args.surge_pct,
plunge_pct=args.plunge_pct,
vol_spike_th=args.vol_spike,
candle_amp_th_pct=args.candle_amp_pct,
ret_window=args.ret_window,
vol_ma_window=args.vol_ma_window,
)
if not det or not det["event"]:
continue
# 같은 분봉 시각의 이벤트는 1회만
if last_event_bar.get(code) == det["bar_time"]:
continue
last_event_bar[code] = det["bar_time"]
total_events += 1
newline()
print(
f"[{datetime.now():%H:%M:%S}] !!! {det['event']:<6} !!! "
f"{code} {name} "
f"종가 {int(det['close']):>7,} "
f"수익률 {det['ret_pct']:+6.2f}% "
f"거래량 {det['vol_spike']:.1f}x "
f"캔들폭 {det['candle_amp_pct']:.2f}% "
f"@봉시각 {det['bar_time']:%H:%M}"
)
# ── 다음 폴링까지 heartbeat 카운트다운 ─────────────────────────────
cycle_dur = (datetime.now() - cycle_start).total_seconds()
remain = max(0.0, args.interval_sec - cycle_dur)
target = datetime.now() + timedelta(seconds=remain)
while True:
left = (target - datetime.now()).total_seconds()
if left <= 0:
break
heartbeat(
f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {mh_label} "
f"cycle=#{cycle:<4d} "
f"watch={len(watchlist)} "
f"events={total_events} "
f"next poll in {int(left):>3d}s"
)
time.sleep(min(1.0, left))
heartbeat(
f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {mh_label} "
f"cycle=#{cycle + 1:<4d} 폴링 중... "
)
except KeyboardInterrupt:
newline()
print(f"[stop] 사용자에 의해 종료 (Ctrl-C). 누적 이벤트 {total_events}건.")
return 0
except Exception as exc:
newline()
print(f"[error] 예기치 못한 오류로 종료: {exc}", file=sys.stderr)
import traceback
traceback.print_exc()
return 1
# ──────────────────────────────────────────────────────────────────────────────
# CLI
# ──────────────────────────────────────────────────────────────────────────────
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="KIS 1분봉 기반 급등락 모니터 (무한 루프)")
p.add_argument("--codes", type=str, default="",
help="감시 종목코드 콤마 구분 (예: 005930,000660). 우선순위 1 (지정 시 다른 모드 무시)")
p.add_argument("--top-n", type=int, default=DEFAULTS["top_n"],
help=(f"거래대금/시가총액 상위 N 종목으로 watchlist 구성 "
f"(기본 {DEFAULTS['top_n']}, 0 이면 호재 자동 선정 사용)"))
p.add_argument("--top-by", type=str, default=DEFAULTS["top_by"],
choices=["amount", "marcap", "volume"],
help="--top-n 사용 시 정렬 기준 (기본 amount=거래대금)")
p.add_argument("--per-call-sleep", type=float, default=DEFAULTS["per_call_sleep"],
help=f"종목 1건 KIS 호출 후 sleep 초 (rate limit 보호, 기본 {DEFAULTS['per_call_sleep']})")
p.add_argument("--rate-limit-backoff", type=float, default=DEFAULTS["rate_limit_backoff"],
help=(f"KIS 'EGW00201 초당 거래건수 초과' 발생 시 추가 대기 초 "
f"(기본 {DEFAULTS['rate_limit_backoff']})"))
p.add_argument("--net-retries", type=int, default=DEFAULTS["net_retries"],
help=(f"네트워크 일시 오류(DNS/연결타임아웃 등) 자동 재시도 횟수 "
f"(기본 {DEFAULTS['net_retries']})"))
p.add_argument("--net-retry-backoff", type=float, default=DEFAULTS["net_retry_backoff"],
help=(f"네트워크 재시도 간 대기 초 (선형 증가, 기본 {DEFAULTS['net_retry_backoff']})"))
p.add_argument("--interval-sec", type=int, default=DEFAULTS["interval_sec"],
help=f"폴링 주기 초 (기본 {DEFAULTS['interval_sec']}). watchlist 크면 더 크게")
# 트리거 임계치
p.add_argument("--surge-pct", type=float, default=DEFAULTS["surge_pct"],
help=f"급등 누적 수익률 %% (기본 +{DEFAULTS['surge_pct']})")
p.add_argument("--plunge-pct", type=float, default=DEFAULTS["plunge_pct"],
help=f"급락 누적 수익률 %% (기본 {DEFAULTS['plunge_pct']})")
p.add_argument("--vol-spike", type=float, default=DEFAULTS["vol_spike"],
help=f"거래량 평균 대비 배수 (기본 {DEFAULTS['vol_spike']}x)")
p.add_argument("--candle-amp-pct", type=float, default=DEFAULTS["candle_amp_pct"],
help=f"마지막 1분봉 (High-Low)/Close 비율 %% (기본 {DEFAULTS['candle_amp_pct']})")
p.add_argument("--ret-window", type=int, default=DEFAULTS["ret_window"],
help=f"누적 수익률 평가 봉 수 (기본 {DEFAULTS['ret_window']}분)")
p.add_argument("--vol-ma-window", type=int, default=DEFAULTS["vol_ma_window"],
help=f"거래량 이동평균 봉 수 (기본 {DEFAULTS['vol_ma_window']})")
# 호재 자동 선정 옵션 (paper_trade 와 동일)
p.add_argument("--sections", type=str, default=DEFAULTS["sections"])
p.add_argument("--pages", type=int, default=DEFAULTS["pages"])
p.add_argument("--min-score", type=int, default=DEFAULTS["min_score"])
p.add_argument("--max-picks", type=int, default=DEFAULTS["max_picks"])
p.add_argument("--delay", type=float, default=DEFAULTS["delay"])
p.add_argument("--force", action="store_true")
return p.parse_args()
def main() -> int:
args = parse_args()
return run_monitor(args)
if __name__ == "__main__":
sys.exit(main())