예제 소스
한국투자증권 API와 LLM 주식 응용 프로그램 · 파일 19개
surge_paper_trade.py
# -*- coding: utf-8 -*-
"""
급등(1분봉) 감지 시 모의투자 도메인(KIS paper) 시세 기준 가상 매매.
· 시세: 기본은 KIS **모의투자** 분봉 API. ``--redis-url`` 이면 Redis 적재분만 읽음 (writer 별도 실행).
실제 모의 계좌 주문은 넣지 않으며, 체결·잔고는 메모리·CSV 로만 시뮬레이션합니다.
· 매수(기본 ``--entry-style dip_peak``): 급등(SURGE) 감지 후 **즉시 종가 매수하지 않음**.
신호 후 ``--post-signal-minutes`` 동안 분봉 저/고 갱신 + 호가 스프레드/잔량 우위를 관찰한 뒤,
(스프레드가 한 번은 충분히 벌어졌다가 다시 좁혀지고, 매수잔량 우위) + (저점권 가격 도달) 조건을 만족하면 진입합니다.
보유 후 목표가(최소 +5%, ``--profit-target-pct``) 도달 시 매도, 손절은 매수가 기준 ``--stop-loss-pct`` 유지.
Redis 분봉만 쓸 때도 호가 조회를 켜면(기본) 같은 모의 KIS 토큰으로 호가만 추가 호출합니다.
· ``--entry-style surge-close``: 예전 방식 — 신호 봉 종가로 즉시 가상 매수.
· 한 종목당 투자 한도 1,000,000원(주식 수 = floor(한도 / 체결가)), 동시 (보유+관찰대기) 합산 최대 3종목(기본) → 총 300만원 상한.
· 동일 종목 중복 진입 없음(이미 보유 중이거나 **관찰 대기** 중이면 신규 신호 무시).
· 매도: ``surge-close`` 는 분봉 고저로 익절·손절 % 판정. ``dip_peak`` 은 **목표 매도가**(신호 고가·매도1호가 등) 도달 시 매도, 손절만 % 유지.
· 장 종료: 미청산은 마지막 분봉 종가로 청산, 미체결 **관찰 대기**는 취소.
· 장중·장외 대기 중 모두 터미널 **맨 아래 한 줄**을 \\r 로 갱신(tqdm 과 유사)하며
총자산·실현손익·평가손익·순손익·수익률·현금·주식평가·청산/승률·보유 수를 표시합니다.
사용법
$ export KIS_PAPER=1
$ python surge_paper_trade.py
$ python surge_paper_trade.py --codes 005930,000660 --interval-sec 45
$ python surge_paper_trade.py --take-profit-pct 5 --stop-loss-pct 3
$ python surge_paper_trade.py --exit-after-session # 이번 거래일 장마감 후 종료 (스케줄 러너용)
$ python surge_paper_trade.py --no-train-after-close # 장마감 후 LoRA 학습 생략
Redis 분리 (권장: 터미널 2개)
$ export REDIS_URL=redis://127.0.0.1:6379/0
$ python surge_minute_redis_writer.py --interval-sec 30 --top-n 150
$ python surge_paper_trade.py --redis-url $REDIS_URL --interval-sec 30 --top-n 150
· 동일 종목 연속 진입 완화: ``--cooldown-sec-after-exit``, ``--max-entries-per-code-per-day``
· 기본: 거래일 장 마감 후 당일 CSV → JSONL → LoRA 학습(서브프로세스, ``--no-train-after-close`` 로 끔)
사전 준비: `.env` / `.env.local` 에 APP_KEY, APP_SECRET, 모의투자 시 KIS_PAPER=1 권장 (Redis writer 는 KIS 필요)
LoRA 학습: `pip install -r training/requirements_lora.txt`
"""
from __future__ import annotations
import argparse
import csv
import os
import subprocess
import sys
import time
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from pathlib import Path
import pandas as pd
from kis_client import KISClient, parse_orderbook_output1
from paper_trade import is_trading_day
from surge_redis_common import DEFAULT_MINUTE_KEY_PREFIX, df_from_redis_value, minute_key
from surge_monitor import (
DEFAULTS as SM_DEFAULTS,
MARKET_OPEN,
call_with_net_retry,
detect_event,
fetch_minute_candles,
heartbeat,
is_kr_market_hours,
newline,
next_market_open,
resolve_watchlist,
)
HERE = Path(__file__).resolve().parent
DEFAULTS = {
"interval_sec": SM_DEFAULTS["interval_sec"],
"per_call_sleep": SM_DEFAULTS["per_call_sleep"],
"rate_limit_backoff": SM_DEFAULTS["rate_limit_backoff"],
"net_retries": SM_DEFAULTS["net_retries"],
"net_retry_backoff": SM_DEFAULTS["net_retry_backoff"],
"surge_pct": SM_DEFAULTS["surge_pct"],
"plunge_pct": SM_DEFAULTS["plunge_pct"],
"vol_spike": SM_DEFAULTS["vol_spike"],
"candle_amp_pct": SM_DEFAULTS["candle_amp_pct"],
"ret_window": SM_DEFAULTS["ret_window"],
"vol_ma_window": SM_DEFAULTS["vol_ma_window"],
"position_krw": 1_000_000,
"max_positions": 3,
"initial_cash": 5_000_000,
"take_profit_pct": 5.0,
"stop_loss_pct": 3.0,
# dip_peak(관찰형) 고점 목표 최소 수익률
"profit_target_pct": 5.0,
# 급등 신호 후 관찰(저점 형성 확인) 시간
"post_signal_minutes": 5,
# 스프레드/잔량 게이트 (dip_peak 진입 필터)
"min_spread_pct": 0.18,
"enter_spread_pct": 0.10,
"min_bid_ask_ratio": 1.20,
# (선택) 풀백 비율. 0이면 끔. 예: 1.0 => 최근 고점 대비 1% 이상 되돌림에서만 진입
"pullback_pct": 0.0,
# (체결 가정) low 대비 완충. 예: 0.10 => low*1.001 수준으로 지정가 가정
"entry_buffer_pct": 0.10,
# 신호가 대비 추가 상승 시 추세 돌파 진입 (되돌림 없이 계속 오를 때)
"breakout_entry_pct": 0.8,
# 관찰 시간 종료 시에도 신호가 위면 추세 추종 진입
"timeout_trend_entry": True,
"top_n": SM_DEFAULTS["top_n"],
"top_by": SM_DEFAULTS["top_by"],
"sections": SM_DEFAULTS["sections"],
"pages": SM_DEFAULTS["pages"],
"min_score": SM_DEFAULTS["min_score"],
"max_picks": SM_DEFAULTS["max_picks"],
"delay": SM_DEFAULTS["delay"],
}
@dataclass
class OpenPosition:
code: str
name: str
shares: int
entry_price: float
entry_time: datetime
cost: float
peak: float = 0.0
trough: float = 0.0
# >0 이면 익절 % 대신 이 가격(고점 목표) 도달 시 매도
sell_target: float = 0.0
def __post_init__(self) -> None:
if self.peak <= 0:
self.peak = self.entry_price
if self.trough <= 0:
self.trough = self.entry_price
@dataclass
class PendingEntry:
"""(deprecated) 구버전 저점 지정가 대기."""
code: str
name: str
buy_limit: float
sell_limit: float
signal_bar_time: datetime
signal_close: float
created_at: datetime
@dataclass
class PendingSignal:
"""급등 신호 후 관찰(저점 형성 + 호가 우위) 후 진입 대기."""
code: str
name: str
signal_bar_time: datetime
signal_close: float
created_at: datetime
expires_at: datetime
min_low: float
max_high: float
spread_max_seen_pct: float = 0.0
last_bid1: float = 0.0
last_ask1: float = 0.0
last_spread_pct: float = 0.0
last_bid_ask_ratio: float = 0.0
@dataclass
class SessionStats:
initial_cash: float
cash: float
realized_pnl: float = 0.0
trades: int = 0
wins: int = 0
last_equity: float = 0.0
def _shares_for_budget(price: float, budget: float) -> int:
if price <= 0:
return 0
return int(budget // price)
def _mark_to_market(positions: list[OpenPosition], last_prices: dict[str, float]) -> float:
v = 0.0
for p in positions:
px = last_prices.get(p.code, p.entry_price)
v += p.shares * px
return v
def _unrealized_pnl(positions: list[OpenPosition], last_prices: dict[str, float]) -> float:
u = 0.0
for p in positions:
px = last_prices.get(p.code, p.entry_price)
u += p.shares * px - p.cost
return u
def _fmt_paper_status_bar(
stats: SessionStats,
positions: list[OpenPosition],
last_prices: dict[str, float],
*,
tail: str = "",
) -> tuple[float, float, str]:
"""(주식평가액, 총자산, 한 줄 상태 문자열). tail: 예) '다음폴링 42s'."""
mtm = _mark_to_market(positions, last_prices)
equity = stats.cash + mtm
total_pnl = equity - stats.initial_cash
ret_pct = (total_pnl / stats.initial_cash * 100.0) if stats.initial_cash else 0.0
parts = [f"총 {equity:,.0f}원", f"손익 {total_pnl:+,.0f}({ret_pct:+.2f}%)", f"현금 {stats.cash:,.0f}", f"보유 {len(positions)}"]
if tail:
parts.append(tail)
return mtm, equity, " | ".join(parts)
def _heartbeat_paper_status(
stats: SessionStats,
positions: list[OpenPosition],
last_prices: dict[str, float],
*,
tail: str = "",
) -> None:
"""tqdm 과 같이 같은 줄(\\r)에 요약 상태 유지."""
ts = datetime.now().strftime("%H:%M:%S")
_, _, line = _fmt_paper_status_bar(stats, positions, last_prices, tail=tail)
heartbeat(f"[{ts}] {line}")
def wait_until_market_open_paper(
watchlist_size: int,
stats: SessionStats,
positions: list[OpenPosition],
last_prices: dict[str, float],
) -> None:
"""장 외 시간: 맨 아래 줄에 계좌 요약 + 다음 장 개시 카운트다운."""
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_sec = (target - now).total_seconds()
if left_sec <= 0:
return
td = is_trading_day(today)
mh = "휴장일" if not td else ("장 시작 전" if now.time() < MARKET_OPEN else "장 마감 후")
h = int(left_sec // 3600)
m = int((left_sec % 3600) // 60)
s = int(left_sec % 60)
_, _, summary = _fmt_paper_status_bar(stats, positions, last_prices, tail="")
heartbeat(
f"[{now:%Y-%m-%d %H:%M:%S}] {mh} | {summary} | "
f"다음장 {target:%Y-%m-%d %H:%M} | 남은 {h:02d}:{m:02d}:{s:02d}"
)
time.sleep(1)
def _update_extremes_from_df(p: OpenPosition, df) -> None:
"""분봉 창에서 entry 이후 구간의 고·저로 누적 peak/trough 갱신."""
if df is None or df.empty:
return
sub = df[df.index >= p.entry_time]
if sub.empty:
sub = df
hi = float(sub["High"].max())
lo = float(sub["Low"].min())
p.peak = max(p.peak, hi)
p.trough = min(p.trough, lo)
def _exit_decision(
p: OpenPosition,
*,
take_profit_pct: float,
stop_loss_pct: float,
) -> tuple[float, str] | None:
"""매도가·사유 또는 None."""
sl = 1.0 - stop_loss_pct / 100.0
sl_price = p.entry_price * sl
if p.trough <= sl_price:
return sl_price, "손절"
if p.sell_target and p.sell_target > 0:
if p.peak >= p.sell_target:
return p.sell_target, "고점목표"
return None
tp = 1.0 + take_profit_pct / 100.0
tp_price = p.entry_price * tp
if p.peak >= tp_price:
return tp_price, "익절"
return None
def _row_at_bar_time(df, bar_time: datetime):
"""급등 판정 시각에 해당하는 분봉 한 줄."""
if df is None or df.empty:
return None
ts = pd.Timestamp(bar_time)
if ts in df.index:
r = df.loc[ts]
return r.iloc[-1] if isinstance(r, pd.DataFrame) else r
try:
pos = df.index.get_indexer([ts], method="nearest")[0]
if pos >= 0:
return df.iloc[pos]
except Exception:
pass
return None
def _dip_peak_limits(
df,
det: dict,
quote_lv: dict[str, float | None],
) -> tuple[float, float] | None:
"""(매수 한도가, 목표 매도가). 호가 없으면 분봉 저·고만 사용."""
row = _row_at_bar_time(df, det["bar_time"])
if row is None:
if df is None or df.empty:
return None
row = df.iloc[-1]
slo = float(row["Low"])
shi = float(row["High"])
bid1 = quote_lv.get("bidp1")
ask1 = quote_lv.get("askp1")
buy_limit = min(slo, bid1) if bid1 else slo
sell_candidates = [shi]
if ask1:
sell_candidates.append(ask1)
dh = quote_lv.get("stck_hgpr")
if dh:
sell_candidates.append(dh)
sell_limit = max(sell_candidates)
if buy_limit <= 0 or sell_limit <= 0:
return None
if sell_limit <= buy_limit:
ref = max(float(det.get("close") or buy_limit), buy_limit)
sell_limit = ref * 1.005
return buy_limit, sell_limit
def _quote_levels_from_kis(client: KISClient | None, code: str) -> dict[str, float | None]:
empty = {k: None for k in ("bidp1", "askp1", "stck_prpr", "stck_hgpr", "stck_lwpr")}
if client is None:
return empty
try:
body = client.inquire_asking_price_exp_ccn(code)
if str(body.get("rt_cd", "0")) != "0":
return empty
o1 = body.get("output1")
if isinstance(o1, list) and o1:
o1 = o1[0]
return parse_orderbook_output1(o1 if isinstance(o1, dict) else None)
except Exception:
return empty
def _pending_fill_hit(df, quote_lv: dict[str, float | None], buy_limit: float) -> bool:
if buy_limit <= 0:
return False
if df is not None and not df.empty:
low = float(df.iloc[-1]["Low"])
cl = float(df.iloc[-1]["Close"])
if low <= buy_limit + 1e-9 or cl <= buy_limit + 1e-9:
return True
pr = quote_lv.get("stck_prpr")
if pr is not None and pr <= buy_limit + 1e-9:
return True
bid1 = quote_lv.get("bidp1")
if bid1 is not None and bid1 <= buy_limit + 1e-9:
return True
return False
def _spread_pct(bid1: float | None, ask1: float | None) -> float:
if not bid1 or not ask1 or bid1 <= 0 or ask1 <= 0:
return 0.0
mid = (bid1 + ask1) / 2.0
if mid <= 0:
return 0.0
return (ask1 - bid1) / mid * 100.0
def _bid_ask_ratio_from_quote(quote_lv: dict[str, float | None]) -> float:
"""호가 잔량 기반 우위. total_* 있으면 total 기준, 아니면 1호가 잔량 기준."""
tb = quote_lv.get("total_bidp_rsqn")
ta = quote_lv.get("total_askp_rsqn")
if tb and ta and ta > 0:
return float(tb) / float(ta)
b1 = quote_lv.get("bidp_rsqn1")
a1 = quote_lv.get("askp_rsqn1")
if b1 and a1 and a1 > 0:
return float(b1) / float(a1)
return 0.0
def _quote_has_book(quote_lv: dict[str, float | None]) -> bool:
return bool(quote_lv.get("bidp1") and quote_lv.get("askp1"))
def _current_px(df, quote_lv: dict[str, float | None]) -> float:
pr = quote_lv.get("stck_prpr")
if pr and pr > 0:
return float(pr)
if df is not None and not df.empty:
return float(df["Close"].iloc[-1])
return 0.0
def _ratio_ok(ratio: float, quote_lv: dict[str, float | None], min_ratio: float) -> bool:
if ratio >= min_ratio:
return True
# 호가 없으면 분봉만으로 추세 판단 허용
return not _quote_has_book(quote_lv)
def _spread_gate_strict(sp: float, spread_max: float, args) -> bool:
return spread_max >= float(args.min_spread_pct) and sp <= float(args.enter_spread_pct) + 1e-12
def _spread_gate_relaxed(sp: float, args) -> bool:
"""돌파/타임아웃: 현재 스프레드만 완화 기준."""
if sp <= 0:
return True
return sp <= float(args.enter_spread_pct) * 2.0 + 1e-12
def _sell_target_from_obs(entry_px: float, ps: "PendingSignal", args) -> float:
return max(
entry_px * (1.0 + float(args.profit_target_pct) / 100.0),
ps.max_high * (1.0 + float(args.entry_buffer_pct) / 100.0),
)
def _pending_entry_decision(
ps: PendingSignal,
df,
quote_lv: dict[str, float | None],
args,
*,
on_timeout: bool = False,
) -> tuple[float, str] | None:
"""관찰 중 진입 가격·사유. None 이면 아직/불가."""
cur_px = _current_px(df, quote_lv)
if cur_px <= 0:
return None
bid1 = quote_lv.get("bidp1")
ask1 = quote_lv.get("askp1")
sp = _spread_pct(bid1, ask1)
ratio = _bid_ask_ratio_from_quote(quote_lv)
sig = float(ps.signal_close)
if sig <= 0:
return None
buy_cap = ps.min_low * (1.0 + float(args.entry_buffer_pct) / 100.0)
price_hit = cur_px <= buy_cap + 1e-9
if df is not None and not df.empty:
last = df.iloc[-1]
price_hit = price_hit or float(last["Low"]) <= buy_cap + 1e-9 or float(last["Close"]) <= buy_cap + 1e-9
pullback_ok = True
if args.pullback_pct and args.pullback_pct > 0 and ps.max_high > 0:
pullback_ok = cur_px <= ps.max_high * (1.0 - float(args.pullback_pct) / 100.0) + 1e-9
dip_gate = (
_spread_gate_strict(sp, ps.spread_max_seen_pct, args)
and _ratio_ok(ratio, quote_lv, float(args.min_bid_ask_ratio))
and pullback_ok
and price_hit
)
if dip_gate:
entry_px = float(bid1 or buy_cap or cur_px)
if buy_cap > 0:
entry_px = min(entry_px, buy_cap)
return entry_px, "SURGE_DIP"
breakout_thr = sig * (1.0 + float(args.breakout_entry_pct) / 100.0)
breakout_gate = (
not on_timeout
and cur_px >= breakout_thr - 1e-9
and cur_px >= sig - 1e-9
and _spread_gate_relaxed(sp, args)
and _ratio_ok(ratio, quote_lv, float(args.min_bid_ask_ratio))
)
if breakout_gate:
entry_px = float(ask1 or cur_px)
return entry_px, "SURGE_BREAKOUT"
if on_timeout and args.timeout_trend_entry:
if cur_px >= sig - 1e-9 and _ratio_ok(ratio, quote_lv, float(args.min_bid_ask_ratio)):
entry_px = float(ask1 or cur_px)
return entry_px, "SURGE_TREND_TIMEOUT"
return None
def _fmt_pending_skip(ps: PendingSignal, cur_px: float, sp: float, ratio: float) -> str:
sig = ps.signal_close
ret = (cur_px / sig - 1.0) * 100.0 if sig > 0 and cur_px > 0 else 0.0
return (
f"signal={sig:,.0f} cur={cur_px:,.0f}({ret:+.1f}%) "
f"low={ps.min_low:,.0f} high={ps.max_high:,.0f} "
f"sp={sp:.3f}% sp_max={ps.spread_max_seen_pct:.3f}% ratio={ratio:.2f}"
)
def _append_csv(path: Path, row: dict[str, object]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
new_file = not path.is_file()
with path.open("a", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=list(row.keys()))
if new_file:
w.writeheader()
w.writerow(row)
def _run_lora_after_close(args: argparse.Namespace, trade_day: date) -> None:
"""당일 surge_paper_trades CSV → build_surge_dataset → train_surge_lora (자식 프로세스)."""
newline()
csv_path = HERE / f"surge_paper_trades_{trade_day:%Y-%m-%d}.csv"
py = sys.executable
train_data = Path(args.lora_train_data)
out_dir = Path(args.lora_out)
print(f"[LoRA] {trade_day} 장마감 후 학습 파이프라인", flush=True)
if not csv_path.is_file():
print(f" · CSV 없음 → 건너뜀 ({csv_path.name})", flush=True)
return
ds_cmd = [
py,
str(HERE / "training" / "build_surge_dataset.py"),
"--csv",
str(csv_path),
"--out",
str(train_data),
]
print(f" · {' '.join(ds_cmd)}", flush=True)
r1 = subprocess.run(ds_cmd, cwd=str(HERE))
if r1.returncode != 0:
print(f" · [warn] 데이터셋 생성 실패 (코드 {r1.returncode})", flush=True)
return
if not train_data.is_file():
print(" · [warn] JSONL 이 생성되지 않음", flush=True)
return
tr_cmd = [
py,
"-m",
"training",
"train_surge_lora.py",
"--data",
str(train_data),
"--out",
str(out_dir),
"--base-model",
str(args.lora_base_model),
"--epochs",
str(args.lora_epochs),
]
print(f" · {' '.join(tr_cmd)}", flush=True)
r2 = subprocess.run(tr_cmd, cwd=str(HERE))
if r2.returncode != 0:
print(f" · [warn] LoRA 학습 실패 (코드 {r2.returncode})", flush=True)
else:
print(f" · [ok] LoRA 저장: {out_dir}", flush=True)
def _minute_df_from_redis(
r, code: str, *, prefix: str, stale_sec: int, now: datetime
):
"""Redis GET → 분봉 DataFrame. 오래되면 빈 프레임."""
import pandas as pd
raw = r.get(minute_key(code, prefix=prefix))
df, updated = df_from_redis_value(raw)
if df.empty:
return df
if stale_sec > 0 and updated is not None and (now - updated).total_seconds() > stale_sec:
return pd.DataFrame(columns=["Open", "High", "Low", "Close", "Volume"])
return df
def run(args: argparse.Namespace) -> int:
# 호환: 예전 옵션 --pending-timeout-sec 을 쓰면 post-signal 관찰 시간으로 환산
try:
pt = int(getattr(args, "pending_timeout_sec", 0) or 0)
except Exception:
pt = 0
if pt > 0:
default_psm = int(DEFAULTS.get("post_signal_minutes", 5))
if int(getattr(args, "post_signal_minutes", default_psm) or default_psm) == default_psm:
args.post_signal_minutes = max(1, int((pt + 59) // 60))
args.timeout_trend_entry = not bool(getattr(args, "no_timeout_trend_entry", False))
watchlist = resolve_watchlist(args)
if not watchlist:
print("watchlist 가 비었습니다.", file=sys.stderr)
return 1
redis_client = None
client: KISClient | None = None
quote_client: KISClient | None = None
redis_url = (getattr(args, "redis_url", None) or "").strip()
if redis_url:
try:
import redis as redis_mod
except ImportError:
print("Redis 모드에는 `pip install redis` 가 필요합니다.", file=sys.stderr)
return 1
try:
redis_client = redis_mod.Redis.from_url(redis_url, decode_responses=True)
redis_client.ping()
except Exception as exc:
print(f"Redis 연결 실패: {exc}", file=sys.stderr)
return 1
print("[init] 분봉 소스: Redis")
else:
client = KISClient(paper=True)
print("[init] KIS 모의투자 도메인 토큰 발급(분봉)...")
client.issue_access_token()
use_dip = args.entry_style == "dip_peak"
if use_dip and (not args.no_quote_on_pending):
if client is not None:
quote_client = client
else:
try:
quote_client = KISClient(paper=True)
print("[init] KIS 모의투자 토큰 발급(호가 전용)...")
quote_client.issue_access_token()
except Exception as exc:
print(f"[warn] 호가 조회용 KIS 연결 실패 → 분봉만으로 저·고 추정: {exc}", file=sys.stderr)
quote_client = None
stats = SessionStats(initial_cash=float(args.initial_cash), cash=float(args.initial_cash))
positions: list[OpenPosition] = []
codes_open = {p.code for p in positions}
last_event_bar: dict[str, datetime] = {}
last_prices: dict[str, float] = {}
pending: dict[str, PendingSignal] = {}
print()
print(f"=== 급등 가상매매(모의 시세) @ {datetime.now():%Y-%m-%d %H:%M:%S} ===")
print(f" · 초기 현금 : {int(stats.initial_cash):,} 원")
print(f" · 종목당 매수 : {int(args.position_krw):,} 원 / 최대 {args.max_positions} 종목")
print(
f" · 익절/손절 : +{args.take_profit_pct}% / -{args.stop_loss_pct}% "
f"(dip_peak 목표수익≥{args.profit_target_pct}%)"
)
print(f" · 진입 방식 : {args.entry_style}")
if args.entry_style == "dip_peak":
print(
f" · 관찰진입 : {args.post_signal_minutes}분 "
f"저점(sp+ratio) / 돌파≥{args.breakout_entry_pct}% / "
f"종료추세={'ON' if args.timeout_trend_entry else 'OFF'} "
f"호가 {'ON' if not args.no_quote_on_pending else 'OFF'}"
)
print(f" · 폴링 : {args.interval_sec}초")
print(f" · 거래 기록 CSV : surge_paper_trades_YYYY-MM-DD.csv (당일 자동)")
print(
f" · 장마감 LoRA : {'끔' if args.no_train_after_close else '켬'} "
f"(데이터 {args.lora_train_data})"
)
if redis_url:
print(f" · Redis : {redis_url} prefix={args.redis_prefix!r} stale>{args.redis_stale_sec}s 스킵")
print(
f" · 재진입 제한 : 쿨다운 {args.cooldown_sec_after_exit}s "
f"종목당일진입≤{args.max_entries_per_code_per_day or '무제한'}"
)
print()
cycle = 0
session_trading_date: date | None = None
last_lora_train_date: date | None = None
entries_per_code: dict[str, int] = {}
last_exit_at: dict[str, datetime] = {}
try:
while True:
now0 = datetime.now()
in_session = is_trading_day(now0.date()) and is_kr_market_hours(now0)
if not in_session:
if positions and not is_kr_market_hours(now0):
# 장 마감 직후 일괄 청산
newline()
for p in list(positions):
px = last_prices.get(p.code, p.entry_price)
proceeds = p.shares * px
pnl = proceeds - p.cost
stats.cash += proceeds
stats.realized_pnl += pnl
stats.trades += 1
if pnl > 0:
stats.wins += 1
print(
f"[청산] {p.code} {p.name} 장종료 @ {px:,.0f} "
f"손익 {pnl:+,.0f}원 ({pnl / p.cost * 100:+.2f}%)"
)
last_exit_at[p.code] = datetime.now()
_append_csv(
HERE / f"surge_paper_trades_{now0.date():%Y-%m-%d}.csv",
{
"시각": datetime.now().isoformat(timespec="seconds"),
"종목코드": p.code,
"종목명": p.name,
"구분": "매도",
"사유": "장종료",
"가격": round(px, 2),
"수량": p.shares,
"금액": round(proceeds, 2),
"실현손익": round(pnl, 2),
"누적실현": round(stats.realized_pnl, 2),
"잔고현금": round(stats.cash, 2),
},
)
positions.clear()
codes_open.clear()
_heartbeat_paper_status(stats, positions, last_prices)
if pending and not is_kr_market_hours(now0):
newline()
for c, ps in list(pending.items()):
print(f"[취소] {c} {ps.name} 장마감 — 관찰 대기 취소")
pending.clear()
_heartbeat_paper_status(stats, positions, last_prices)
closed_this_trading_day = (
is_trading_day(now0.date())
and not is_kr_market_hours(now0)
and session_trading_date is not None
and session_trading_date == now0.date()
)
if closed_this_trading_day and not args.no_train_after_close:
if last_lora_train_date != session_trading_date:
_run_lora_after_close(args, session_trading_date)
last_lora_train_date = session_trading_date
if args.exit_after_session and closed_this_trading_day:
newline()
print(
f"[{now0:%Y-%m-%d %H:%M:%S}] 장 마감 — 이번 거래일 세션 종료 (--exit-after-session)"
)
return 0
wait_until_market_open_paper(len(watchlist), stats, positions, last_prices)
session_trading_date = None
newline()
print(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] 정규장 개시 — 가상매매 재개")
_heartbeat_paper_status(stats, positions, last_prices)
continue
if session_trading_date is None:
session_trading_date = now0.date()
entries_per_code.clear()
pending.clear()
day_csv = HERE / f"surge_paper_trades_{now0.date():%Y-%m-%d}.csv"
cycle += 1
cycle_start = datetime.now()
for code, name in watchlist:
try:
if redis_client is not None:
df = _minute_df_from_redis(
redis_client,
code,
prefix=args.redis_prefix,
stale_sec=args.redis_stale_sec,
now=now0,
)
else:
assert client is not None
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} 분봉 실패: {msg}", file=sys.stderr)
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
if redis_client is None and args.per_call_sleep > 0:
time.sleep(args.per_call_sleep)
if not df.empty:
last_prices[code] = float(df["Close"].iloc[-1])
ps = pending.get(code)
if ps is not None:
on_timeout = now0 >= ps.expires_at
if df is not None and not df.empty:
last_bar = df.iloc[-1]
ps.min_low = min(ps.min_low, float(last_bar["Low"]))
ps.max_high = max(ps.max_high, float(last_bar["High"]))
ql = _quote_levels_from_kis(quote_client, code)
bid1 = ql.get("bidp1")
ask1 = ql.get("askp1")
sp = _spread_pct(bid1, ask1)
ratio = _bid_ask_ratio_from_quote(ql)
ps.last_bid1 = float(bid1 or 0.0)
ps.last_ask1 = float(ask1 or 0.0)
ps.last_spread_pct = float(sp)
ps.last_bid_ask_ratio = float(ratio)
ps.spread_max_seen_pct = max(ps.spread_max_seen_pct, float(sp))
decision = _pending_entry_decision(ps, df, ql, args, on_timeout=on_timeout)
if decision is None and on_timeout:
pending.pop(code, None)
cur_px = _current_px(df, ql)
newline()
print(
f"[{datetime.now():%H:%M:%S}] 관찰취소(시간초과) {code} {name} "
f"({args.post_signal_minutes}분) "
f"{_fmt_pending_skip(ps, cur_px, sp, ratio)}"
)
_heartbeat_paper_status(stats, positions, last_prices)
elif decision is not None:
entry_px, entry_reason = decision
shares = _shares_for_budget(entry_px, float(args.position_krw))
cost = shares * entry_px
if shares > 0 and cost <= stats.cash + 1e-6 and stats.cash - cost >= 0:
pending.pop(code, None)
sell_target = _sell_target_from_obs(entry_px, ps, args)
op = OpenPosition(
code=code,
name=name,
shares=shares,
entry_price=entry_px,
entry_time=now0,
cost=cost,
peak=ps.max_high if ps.max_high > 0 else entry_px,
trough=ps.min_low if ps.min_low > 0 else entry_px,
sell_target=sell_target,
)
stats.cash -= cost
positions.append(op)
codes_open.add(code)
entries_per_code[code] = entries_per_code.get(code, 0) + 1
reason_ko = {
"SURGE_DIP": "저점",
"SURGE_BREAKOUT": "돌파",
"SURGE_TREND_TIMEOUT": "관찰종료추세",
}.get(entry_reason, entry_reason)
newline()
print(
f"[{datetime.now():%H:%M:%S}] 매수(가상) {reason_ko} {code} {name} "
f"{entry_px:,.0f}원 ×{shares} = {cost:,.0f}원 "
f"목표 {sell_target:,.0f}원(+{args.profit_target_pct}%) "
f"signal={ps.signal_close:,.0f} sp={sp:.3f}% ratio={ratio:.2f}"
)
_append_csv(
day_csv,
{
"시각": datetime.now().isoformat(timespec="seconds"),
"종목코드": code,
"종목명": name,
"구분": "매수",
"사유": entry_reason,
"가격": round(entry_px, 2),
"수량": shares,
"금액": round(cost, 2),
"실현손익": "",
"누적실현": round(stats.realized_pnl, 2),
"잔고현금": round(stats.cash, 2),
},
)
_heartbeat_paper_status(stats, positions, last_prices)
# ── 보유 포지션: 고저 갱신 후 익절/손절
for p in list(positions):
if p.code != code:
continue
_update_extremes_from_df(p, df)
ex = _exit_decision(
p,
take_profit_pct=args.take_profit_pct,
stop_loss_pct=args.stop_loss_pct,
)
if ex is None:
continue
sell_px, reason = ex
proceeds = p.shares * sell_px
pnl = proceeds - p.cost
stats.cash += proceeds
stats.realized_pnl += pnl
stats.trades += 1
if pnl > 0:
stats.wins += 1
newline()
print(
f"[{datetime.now():%H:%M:%S}] 매도 {reason} {code} {name} "
f"{sell_px:,.0f}원 ×{p.shares} 손익 {pnl:+,.0f}원 ({pnl / p.cost * 100:+.2f}%)"
)
_append_csv(
day_csv,
{
"시각": datetime.now().isoformat(timespec="seconds"),
"종목코드": p.code,
"종목명": p.name,
"구분": "매도",
"사유": reason,
"가격": round(sell_px, 2),
"수량": p.shares,
"금액": round(proceeds, 2),
"실현손익": round(pnl, 2),
"누적실현": round(stats.realized_pnl, 2),
"잔고현금": round(stats.cash, 2),
},
)
positions.remove(p)
codes_open.discard(p.code)
last_exit_at[p.code] = datetime.now()
_heartbeat_paper_status(stats, positions, last_prices)
if df.empty:
continue
blocked = codes_open | pending.keys()
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 det.get("event") != "SURGE":
continue
if last_event_bar.get(code) == det["bar_time"]:
continue
# 신규 매수(또는 대기) 조건
if code in blocked:
last_event_bar[code] = det["bar_time"]
continue
if len(positions) + len(pending) >= args.max_positions:
last_event_bar[code] = det["bar_time"]
continue
if args.cooldown_sec_after_exit > 0 and code in last_exit_at:
if (now0 - last_exit_at[code]).total_seconds() < args.cooldown_sec_after_exit:
last_event_bar[code] = det["bar_time"]
continue
if args.max_entries_per_code_per_day > 0:
if entries_per_code.get(code, 0) >= args.max_entries_per_code_per_day:
last_event_bar[code] = det["bar_time"]
continue
if args.entry_style == "surge_close":
entry_px = float(det["close"])
shares = _shares_for_budget(entry_px, float(args.position_krw))
if shares <= 0:
last_event_bar[code] = det["bar_time"]
continue
cost = shares * entry_px
if cost > stats.cash + 1e-6:
last_event_bar[code] = det["bar_time"]
continue
if stats.cash - cost < 0:
last_event_bar[code] = det["bar_time"]
continue
last_bar = df.iloc[-1]
op = OpenPosition(
code=code,
name=name,
shares=shares,
entry_price=entry_px,
entry_time=det["bar_time"],
cost=cost,
peak=float(last_bar["High"]),
trough=float(last_bar["Low"]),
sell_target=0.0,
)
stats.cash -= cost
positions.append(op)
codes_open.add(code)
last_event_bar[code] = det["bar_time"]
entries_per_code[code] = entries_per_code.get(code, 0) + 1
newline()
print(
f"[{datetime.now():%H:%M:%S}] 매수(가상) 급등 {code} {name} "
f"{entry_px:,.0f}원 ×{shares} = {cost:,.0f}원 "
f"(보유 {len(positions)}/{args.max_positions})"
)
_append_csv(
day_csv,
{
"시각": datetime.now().isoformat(timespec="seconds"),
"종목코드": code,
"종목명": name,
"구분": "매수",
"사유": "SURGE",
"가격": round(entry_px, 2),
"수량": shares,
"금액": round(cost, 2),
"실현손익": "",
"누적실현": round(stats.realized_pnl, 2),
"잔고현금": round(stats.cash, 2),
},
)
_heartbeat_paper_status(stats, positions, last_prices)
continue
# dip_peak: 급등 신호 후 관찰창(기본 5분) 동안 저점/고점 + 호가 우위 확인 후 진입
last_bar = df.iloc[-1]
min_low = float(last_bar["Low"])
max_high = float(last_bar["High"])
pending[code] = PendingSignal(
code=code,
name=name,
signal_bar_time=det["bar_time"],
signal_close=float(det["close"]),
created_at=now0,
expires_at=now0 + timedelta(minutes=int(args.post_signal_minutes)),
min_low=min_low,
max_high=max_high,
)
last_event_bar[code] = det["bar_time"]
newline()
print(
f"[{datetime.now():%H:%M:%S}] 급등·관찰대기 {code} {name} "
f"신호종가 {det['close']:,.0f} "
f"관찰 {args.post_signal_minutes}분 "
f"(대기 {len(pending)} / 보유 {len(positions)})"
)
_heartbeat_paper_status(stats, positions, last_prices)
mtm = _mark_to_market(positions, last_prices)
stats.last_equity = stats.cash + mtm
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_paper_status(
stats,
positions,
last_prices,
tail=f"cycle#{cycle} 다음폴링 {int(left)}s",
)
time.sleep(min(1.0, left))
except KeyboardInterrupt:
newline()
mtm = _mark_to_market(positions, last_prices)
equity = stats.cash + mtm
unreal = _unrealized_pnl(positions, last_prices)
total_pnl = equity - stats.initial_cash
ret_pct = (total_pnl / stats.initial_cash * 100.0) if stats.initial_cash else 0.0
print(
f"[종료] Ctrl-C 총자산 {equity:,.0f}원 실현 {stats.realized_pnl:+,.0f} "
f"평가손익 {unreal:+,.0f} 순손익 {total_pnl:+,.0f} ({ret_pct:+.2f}%) "
f"미청산 {len(positions)}건"
)
return 0
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="급등 감지 + 모의 시세 기준 가상 매매(주문 없음)")
p.add_argument("--codes", type=str, default="")
p.add_argument("--top-n", type=int, default=DEFAULTS["top_n"])
p.add_argument("--top-by", type=str, default=DEFAULTS["top_by"], choices=["amount", "marcap", "volume"])
p.add_argument("--interval-sec", type=int, default=DEFAULTS["interval_sec"])
p.add_argument("--per-call-sleep", type=float, default=DEFAULTS["per_call_sleep"])
p.add_argument("--rate-limit-backoff", type=float, default=DEFAULTS["rate_limit_backoff"])
p.add_argument("--net-retries", type=int, default=DEFAULTS["net_retries"])
p.add_argument("--net-retry-backoff", type=float, default=DEFAULTS["net_retry_backoff"])
p.add_argument("--surge-pct", type=float, default=DEFAULTS["surge_pct"])
p.add_argument("--plunge-pct", type=float, default=DEFAULTS["plunge_pct"])
p.add_argument("--vol-spike", type=float, default=DEFAULTS["vol_spike"])
p.add_argument("--candle-amp-pct", type=float, default=DEFAULTS["candle_amp_pct"])
p.add_argument("--ret-window", type=int, default=DEFAULTS["ret_window"])
p.add_argument("--vol-ma-window", type=int, default=DEFAULTS["vol_ma_window"])
p.add_argument(
"--redis-url",
type=str,
default=os.environ.get("REDIS_URL", ""),
help="설정 시 분봉은 KIS 대신 Redis에서 읽음 (surge_minute_redis_writer 별도 실행)",
)
p.add_argument("--redis-prefix", type=str, default=DEFAULT_MINUTE_KEY_PREFIX)
p.add_argument(
"--redis-stale-sec",
type=int,
default=120,
help="Redis payload 의 updated_at 이 이보다 오래되면 해당 종목 분봉 스킵",
)
p.add_argument(
"--cooldown-sec-after-exit",
type=int,
default=180,
help="매도(손절/익절/장종료) 후 같은 종목 재매수 금지 시간(초). 0 이면 끔",
)
p.add_argument(
"--max-entries-per-code-per-day",
type=int,
default=3,
help="종목당 당일 신규 매수 횟수 상한. 0 이면 제한 없음",
)
p.add_argument("--position-krw", type=int, default=DEFAULTS["position_krw"], help="종목당 매수 예산(원)")
p.add_argument("--max-positions", type=int, default=DEFAULTS["max_positions"], help="동시 보유 종목 수 상한")
p.add_argument("--initial-cash", type=int, default=DEFAULTS["initial_cash"], help="세션 초기 현금(원)")
p.add_argument("--take-profit-pct", type=float, default=DEFAULTS["take_profit_pct"])
p.add_argument("--stop-loss-pct", type=float, default=DEFAULTS["stop_loss_pct"])
p.add_argument(
"--entry-style",
type=str,
default="dip_peak",
choices=["dip_peak", "surge_close"],
help="dip_peak: 급등 후 저점(호가·분봉)에서만 가상 매수, 고가 목표가에서 매도. surge_close: 신호 봉 종가 즉시 매수",
)
p.add_argument(
"--no-quote-on-pending",
action="store_true",
help="dip_peak 에서 KIS 호가 API 를 쓰지 않음(분봉만으로 관찰/진입)",
)
p.add_argument(
"--pending-timeout-sec",
type=int,
default=300,
help="[deprecated] dip_peak 관찰 대기 최대 시간(초). 0 이면 제한 없음",
)
p.add_argument(
"--post-signal-minutes",
type=int,
default=DEFAULTS["post_signal_minutes"],
help="dip_peak: 급등 신호 후 관찰 시간(분). 기본 5",
)
p.add_argument(
"--profit-target-pct",
type=float,
default=DEFAULTS["profit_target_pct"],
help="dip_peak: 목표 매도 최소 수익률(%) (기본 5.0)",
)
p.add_argument(
"--min-spread-pct",
type=float,
default=DEFAULTS["min_spread_pct"],
help="dip_peak: 관찰 중 한 번이라도 스프레드가 이 % 이상은 나왔어야 함",
)
p.add_argument(
"--enter-spread-pct",
type=float,
default=DEFAULTS["enter_spread_pct"],
help="dip_peak: 진입 시점 현재 스프레드가 이 % 이하로 좁혀져야 함",
)
p.add_argument(
"--min-bid-ask-ratio",
type=float,
default=DEFAULTS["min_bid_ask_ratio"],
help="dip_peak: (총)매수잔량/(총)매도잔량 최소 비율",
)
p.add_argument(
"--pullback-pct",
type=float,
default=DEFAULTS["pullback_pct"],
help="dip_peak: 관찰 중 단기고점 대비 되돌림(%)이 이 값 이상일 때만 진입. 0이면 끔",
)
p.add_argument(
"--entry-buffer-pct",
type=float,
default=DEFAULTS["entry_buffer_pct"],
help="dip_peak: min_low 대비 매수 한도 완충(%) (예: 0.10 => low*1.001)",
)
p.add_argument(
"--breakout-entry-pct",
type=float,
default=DEFAULTS["breakout_entry_pct"],
help="dip_peak: 신호가 대비 이 % 이상 추가 상승 시 되돌림 없이 돌파 진입",
)
p.add_argument(
"--no-timeout-trend-entry",
action="store_true",
help="dip_peak: 관찰 시간 종료 시 신호가 위 추세 추종 진입을 끔",
)
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")
p.add_argument(
"--exit-after-session",
action="store_true",
help="이번 거래일 정규장이 끝나면(청산 처리 후) 프로세스를 종료합니다. 스케줄 러너용.",
)
p.add_argument(
"--no-train-after-close",
action="store_true",
help="거래일 장마감 후 LoRA 학습(build_surge_dataset + train_surge_lora)을 하지 않습니다.",
)
p.add_argument(
"--lora-train-data",
type=str,
default=str(HERE / "training" / "data" / "surge_analysis.jsonl"),
help="build_surge_dataset 출력 JSONL 경로",
)
p.add_argument(
"--lora-out",
type=str,
default=str(HERE / "training" / "outputs" / "surge_lora"),
help="LoRA 저장 디렉터리",
)
p.add_argument(
"--lora-base-model",
type=str,
default=os.environ.get("BASE_MODEL", "Qwen/Qwen2.5-0.5B-Instruct"),
help="train_surge_lora --base-model",
)
p.add_argument("--lora-epochs", type=float, default=4.0, help="train_surge_lora --epochs")
return p.parse_args()
def main() -> int:
return run(parse_args())
if __name__ == "__main__":
sys.exit(main())