예제 소스
한국투자증권 API와 LLM 주식 응용 프로그램 · 파일 19개
build_surge_dataset.py
# -*- coding: utf-8 -*-
"""surge_paper_trades CSV → 급등 신호 후 손익 원인 분석용 JSONL 학습 데이터.
- 매수(SURGE) 직후 짝이 되는 매도(익절/손절/장종료)까지 한 건으로 묶습니다.
- 삼표시멘트 등 손실 건은 '급등 신호 ≠ 이후 가격 방향 보장' 관점에서 설명 문장을 붙입니다.
- 샘플 수가 적으므로 동일 구조의 **패러프레이즈 증강**을 몇 배 생성합니다.
사용
python training/build_surge_dataset.py --csv surge_paper_trades_2026-05-13.csv
python training/build_surge_dataset.py # 기본: chart 루트에서 오늘 날짜 CSV 자동 탐색
"""
from __future__ import annotations
import argparse
import json
import random
from collections import defaultdict
from pathlib import Path
import pandas as pd
HERE = Path(__file__).resolve().parent
CHART_ROOT = HERE.parent
def _default_csv() -> Path | None:
today = pd.Timestamp.now(tz=None).strftime("%Y-%m-%d")
p = CHART_ROOT / f"surge_paper_trades_{today}.csv"
return p if p.is_file() else None
def _pair_trades(df: pd.DataFrame) -> list[dict]:
"""시간순으로 매수 뒤에 오는 첫 매도를 한 쌍으로."""
df = df.sort_values("시각").reset_index(drop=True)
out: list[dict] = []
nth_today: dict[str, int] = defaultdict(int)
i = 0
while i < len(df):
row = df.iloc[i]
if row["구분"] != "매수":
i += 1
continue
buy = row
j = i + 1
while j < len(df):
if df.iloc[j]["종목코드"] == buy["종목코드"] and df.iloc[j]["구분"] == "매도":
sell = df.iloc[j]
pnl = float(sell["실현손익"]) if pd.notna(sell["실현손익"]) and sell["실현손익"] != "" else 0.0
cost = float(buy["금액"])
pnl_pct = (pnl / cost * 100.0) if cost else 0.0
code = str(buy["종목코드"])
nth_today[code] += 1
out.append(
{
"종목코드": code,
"종목명": str(buy["종목명"]),
"매수가": float(buy["가격"]),
"매도가": float(sell["가격"]),
"수량": int(buy["수량"]),
"매도사유": str(sell["사유"]),
"실현손익": pnl,
"수익률_pct": pnl_pct,
"당일진입차수": nth_today[code],
}
)
i = j + 1
break
j += 1
else:
i += 1
return out
def _build_explanation(rec: dict) -> str:
"""규칙 기반 한국어 설명(LoRA가 학습할 '교사' 텍스트)."""
name = rec["종목명"]
code = rec["종목코드"]
buy = rec["매수가"]
sell = rec["매도가"]
reason = rec["매도사유"]
pnl = rec["실현손익"]
pctp = rec["수익률_pct"]
common = (
f"이 거래는 1분봉에서 '급등(SURGE)' 조건(단기 누적 상승·거래량 스파이크·봉 폭 확대)이 "
f"동시에 만족될 때 신호가 뜬 뒤, 종가 {buy:,.0f}원 부근에 가상 매수한 사례입니다. "
f"이 신호는 당시 모멘텀과 유동성을 반영할 뿐, 이후 가격이 같은 방향으로 갈 것을 보장하지 않습니다."
)
if reason == "손절" and pnl < 0:
tail = (
f" 이후 분봉에서 저가가 매수가 대비 약 -3% 아래로 내려가며 손절 규칙이 발동해 "
f"{sell:,.0f}원 부근에서 청산되었고, 결과는 약 {pctp:+.2f}%({pnl:,.0f}원)입니다. "
f"급등 직후에는 단기 과열로 차익 실현 매물이 나오거나 고점에서 되돌림(mean reversion)이 흔해, "
f"진입 직후 변동성만 크고 방향이 바로 꺾이는 경우가 있습니다."
)
if int(rec.get("당일진입차수", 1)) >= 2:
tail += (
f" 같은 종목({name})에 당일 이전 진입과는 별개의 분봉 신호이므로, "
f"앞선 손익과 무관하게 다시 손절이 날 수 있습니다."
)
if "삼표" in name or code == "038500":
tail += (
" 삼표시멘트처럼 장중 테마·수급으로 급등이 반복돼도, 매 신호마다 "
"눌림 폭이 손절선보다 깊게 올 수 있어 규칙형 전략에서는 연속 손실이 발생할 수 있습니다."
)
return common + tail
if reason == "익절" and pnl > 0:
return (
common
+ f" 이후 고가가 매수가 대비 +5% 이상 올라 익절 규칙이 발동해 {sell:,.0f}원 부근에서 청산되었고, "
f"약 {pctp:+.2f}%({pnl:,.0f}원) 수익입니다. 급등 신호 이후에도 추세가 잠시 이어지면 익절가에 도달할 수 있으나, "
f"이는 확률적 결과이며 모든 신호에서 재현되지는 않습니다."
)
return (
common
+ f" 청산 사유는 '{reason}'이며 실현손익은 약 {pnl:,.0f}원({pctp:+.2f}%)입니다."
)
def _augment_instruction(rec: dict, rng: random.Random) -> list[tuple[str, str, str]]:
"""(instruction, input, output) 여러 변형."""
user_input = (
f"종목: {rec['종목명']}({rec['종목코드']}). "
f"SURGE 신호 후 매수가 {rec['매수가']:,.0f}원, "
f"청산 {rec['매도사유']}, 매도가 {rec['매도가']:,.0f}원, "
f"실현 {rec['실현손익']:+,.0f}원 ({rec['수익률_pct']:+.2f}%)."
)
expl = _build_explanation(rec)
templates = [
"급등(SURGE) 신호로 들어간 모의 매매인데 결과가 이렇게 나왔습니다. 원인을 분석해 주세요.",
"1분봉 급등 알고리즘으로 샀는데 손익이 위와 같습니다. 왜 그럴 수 있는지 설명해 주세요.",
"모멘텀 돌파 신호 진입 후 청산 결과입니다. 신호의 의미와 손익에 대한 해석을 해 주세요.",
]
rows: list[tuple[str, str, str]] = []
for t in templates:
rows.append((t, user_input, expl))
if rng.random() < 0.5:
rows.append(
(
"위 진입·청산 결과를 바탕으로 리스크 관점에서 짧게 정리해 주세요.",
user_input,
expl[:800] + ("…" if len(expl) > 800 else ""),
)
)
return rows
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--csv", type=str, default="", help="surge_paper_trades_*.csv 경로")
ap.add_argument(
"--out",
type=str,
default=str(HERE / "data" / "surge_analysis.jsonl"),
help="출력 JSONL",
)
ap.add_argument("--seed", type=int, default=42)
args = ap.parse_args()
csv_path = Path(args.csv) if args.csv else _default_csv()
if not csv_path or not csv_path.is_file():
print("CSV 를 찾을 수 없습니다. --csv 로 지정하거나 오늘 날짜 surge_paper_trades_YYYY-MM-DD.csv 를 chart 루트에 두세요.", file=__import__("sys").stderr)
return 1
df = pd.read_csv(csv_path)
pairs = _pair_trades(df)
if not pairs:
print("매수-매도 쌍을 찾지 못했습니다.", file=__import__("sys").stderr)
return 1
out_path = Path(args.out)
out_path.parent.mkdir(parents=True, exist_ok=True)
rng = random.Random(args.seed)
n = 0
with out_path.open("w", encoding="utf-8") as f:
for rec in pairs:
for instr, inp, out in _augment_instruction(rec, rng):
json.dump(
{"instruction": instr, "input": inp, "output": out},
f,
ensure_ascii=False,
)
f.write("\n")
n += 1
# 일반 원칙 샘플 (신호 한계)
general = [
(
"1분봉 급등(SURGE) 전략으로 손실이 났는데, 신호 자체가 잘못된 건가요?",
"손절 -3%, 익절 +5% 규칙을 쓰고 있습니다.",
"신호가 '틀렸다'기보다, 정의상 '최근 몇 분간 가격·거래량·봉 폭이 임계치를 넘었다'는 사실만 알려 줍니다. "
"이후 시장이 같은 방향으로 이어질지는 별개라, 짧은 되돌림에 손절이 먼저 걸리면 손실로 끝납니다. "
"급등 직후 변동성이 크면 손절선이 익절선보다 가까운 경우가 많아 기대값이 나빠질 수 있습니다.",
),
(
"삼표시멘트처럼 같은 종목에서 급등 신호가 여러 번 났을 때 주의할 점은?",
"당일 두 번 매수 후 두 번 모두 손절된 기록이 있습니다.",
"각 신호는 서로 다른 시각의 분봉 패턴입니다. 첫 번째 손실이 '학습'되어 두 번째가 안전해지지는 않습니다. "
"오히려 같은 테마로 변동성이 반복되면 손절이 연속으로 나올 수 있으니, 동일 종목 재진입 필터(쿨다운·손실 횟수)를 "
"두는 편이 리스크 관리에 유리합니다.",
),
]
for instr, inp, out in general:
json.dump({"instruction": instr, "input": inp, "output": out}, f, ensure_ascii=False)
f.write("\n")
n += 1
print(f"[ok] {csv_path.name} → {out_path} (총 {n}줄)")
return 0
if __name__ == "__main__":
raise SystemExit(main())