예제 소스
한국투자증권 API와 LLM 주식 응용 프로그램 · 파일 19개
paper_trade.py
# -*- coding: utf-8 -*-
"""
모의(페이퍼) 자동매매 스케줄러 — 백그라운드 쓰레드 1개로 동작.
스케줄
· 매일 07:00 (로컬시각) — 한국 거래일 확인 후 '호재 종목' 선정 (메모리에 보관)
· 매일 16:30 (로컬시각) — 오늘 OHLC 데이터로 매수/매도 시뮬레이션 + CSV 저장
매매 룰 (실거래는 절대 발생하지 않습니다 / 모든 포지션은 당일 청산)
· 매수: 당일 시초가 (Open)
· 매도 (둘 중 먼저 도달):
- 익절: 고가 ≥ 매수가 × (1 + 5%) → 매도가 = 매수가 × 1.05
- 손절: 저가 ≤ 매수가 × (1 - 3%) → 매도가 = 매수가 × 0.97
- 동시 도달: 일봉 데이터로 순서를 알 수 없어 보수적으로 손절 적용
- 장 마감 매도: 익절/손절 모두 미도달이면 종가(Close) 에 무조건 매도
→ 다음 거래일로 포지션을 넘기지 않음 (당일 강제 청산)
결과 저장
· paper_trades_YYYY-MM-DD.csv (오늘 날짜로 매일 새 파일)
실행 동작
1) 실행 즉시 morning + close 작업을 1회 자동 수행 (호재 종목 선정 + 시뮬레이션)
2) 백그라운드 쓰레드가 다음 트리거(morning/close)까지 대기
3) 메인 쓰레드는 다음 작업까지 남은 시간을 1초마다 카운트다운 표시
4) 종료는 Ctrl-C
사용법
$ python paper_trade.py # 즉시 1회 실행 + 백그라운드 스케줄러 + 카운트다운
$ python paper_trade.py --runonce # 즉시 1회 실행 후 종료 (스케줄러 X)
$ python paper_trade.py --morning-at 07:00 --close-at 16:30
$ python paper_trade.py --force # 거래일이 아니어도 강제로 실행
$ python paper_trade.py --no-startup-run # 시작 시 1회 실행을 끔 (다음 트리거만 대기)
의존성: finance-datareader, pandas, requests, beautifulsoup4, lxml
"""
from __future__ import annotations
import argparse
import sys
import threading
import traceback
from datetime import date, datetime, time as dtime, timedelta
from pathlib import Path
import pandas as pd
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
# 한국거래소(XKRX) 거래일 캘린더. 없으면 fallback 사용.
try:
import exchange_calendars as _xcals # type: ignore
_XKRX = _xcals.get_calendar("XKRX")
except Exception:
_XKRX = None
try:
import holidays as _holidays # type: ignore
except Exception:
_holidays = None # type: ignore
# 같은 폴더의 호재 뉴스 수집 모듈
from fetch_positive_news import (
Article,
annotate,
collect_articles,
filter_positive,
load_stock_map,
)
from build_stock_list import EXCLUDE_DEPT_KEYWORDS
HERE = Path(__file__).resolve().parent
DEFAULTS = {
"sections": "401,402",
"pages": 5,
"min_score": 2,
"max_picks": 7,
"stop_loss_pct": 3.0,
"take_profit_pct": 5.0,
"capital": 5_000_000,
"delay": 0.5,
"morning_at": "07:00",
"close_at": "16:30",
}
# ──────────────────────────────────────────────────────────────────────────────
# 거래일 판정 (FDR 의 KOSPI 지수 데이터로 확인)
# ──────────────────────────────────────────────────────────────────────────────
def is_trading_day(d: date) -> bool:
"""한국거래소(KRX) 정규장 거래일 여부.
1순위: exchange_calendars 의 XKRX 캘린더 (가장 정확, 임시휴장/폐장일 포함)
2순위: 평일 + holidays.KR + 5/1(근로자의 날) + 12/31(폐장일) 보정
fdr 의 데이터 존재 여부로는 판단하지 않습니다 (장 시작 전에는 빈 DF 가 와서 오판).
"""
if _XKRX is not None:
try:
ts = pd.Timestamp(d)
return bool(_XKRX.is_session(ts))
except Exception:
pass
# ── fallback ───────────────────────────────────────────────────────────
if d.weekday() >= 5: # 토(5), 일(6)
return False
if _holidays is not None:
try:
kr = _holidays.country_holidays("KR", years=d.year)
if d in kr:
return False
except Exception:
pass
if d.month == 5 and d.day == 1: # 근로자의 날 (KRX 휴장)
return False
if d.month == 12 and d.day == 31: # 폐장일
return False
return True
# ──────────────────────────────────────────────────────────────────────────────
# 투자주의/관리/거래정지 등 위험 표시 종목코드 집합 (KRX 최신 정보)
# ──────────────────────────────────────────────────────────────────────────────
def fetch_flagged_codes() -> set[str]:
"""현재 KRX 기준 위험 표시(관리/투자주의/거래정지 등) 종목코드 집합."""
flagged: set[str] = set()
try:
df = fdr.StockListing("KRX")
except Exception as exc:
print(f" · KRX 종목정보 조회 실패(투자주의 검증 생략): {exc}", file=sys.stderr)
return flagged
if df is None or df.empty:
return flagged
if "Code" not in df.columns or "Dept" not in df.columns:
return flagged
for code, dept in zip(df["Code"], df["Dept"]):
if not isinstance(code, str) or not isinstance(dept, str):
continue
if any(kw in dept for kw in EXCLUDE_DEPT_KEYWORDS):
flagged.add(code)
return flagged
# ──────────────────────────────────────────────────────────────────────────────
# 호재 종목 후보 선정
# ──────────────────────────────────────────────────────────────────────────────
def aggregate_stock_scores(articles: list[Article]) -> list[dict]:
agg: dict[str, dict] = {}
for a in articles:
if a.positive_score <= 0:
continue
for name, code in zip(a.matched_names, a.matched_codes):
if code not in agg:
agg[code] = {
"종목코드": code,
"종목명": name,
"호재점수합": 0,
"기사수": 0,
"대표기사": "",
"대표키워드": "",
"대표URL": "",
}
agg[code]["호재점수합"] += a.positive_score
agg[code]["기사수"] += 1
if not agg[code]["대표기사"]:
agg[code]["대표기사"] = a.title
agg[code]["대표키워드"] = ", ".join(a.matched_pos[:5])
agg[code]["대표URL"] = a.url
picks = list(agg.values())
picks.sort(key=lambda x: (x["호재점수합"], x["기사수"]), reverse=True)
return picks
def pick_positive_stocks(args) -> list[dict]:
print("[호재선정] 종목 매핑 로드")
stock_pairs = load_stock_map(HERE / "stock_list.csv")
print(f" · 매칭 가능한 종목 수: {len(stock_pairs)}")
sections = [s.strip() for s in args.sections.split(",") if s.strip()]
print(f"[호재선정] 네이버 금융 뉴스 수집 (sections={','.join(sections)}, pages={args.pages})")
articles = collect_articles(sections, args.pages, args.delay)
print(f" · 수집된 기사 수: {len(articles)}")
articles = annotate(articles, stock_pairs)
positive = filter_positive(articles, min_score=args.min_score, require_stock=True)
print(f" · 호재 판정(>= {args.min_score}점): {len(positive)}")
all_picks = aggregate_stock_scores(positive)
# 투자주의 / 관리종목 / 거래정지 / 정리매매 / 단기과열 / 불성실공시 등 배제
print("[호재선정] 투자주의/관리/거래정지 등 위험 표시 종목 배제 (KRX 최신 정보)")
flagged = fetch_flagged_codes()
print(f" · 위험표시 종목 수: {len(flagged)}")
excluded: list[dict] = []
safe_picks: list[dict] = []
for p in all_picks:
if p["종목코드"] in flagged:
excluded.append(p)
else:
safe_picks.append(p)
if excluded:
print(f" · 배제된 후보 ({len(excluded)}건):")
for p in excluded[: args.max_picks * 2]:
print(f" - {p['종목코드']} {p['종목명']} (호재점수합={p['호재점수합']})")
picks = safe_picks[: args.max_picks]
print(f" · 상위 {len(picks)}개 최종 후보:")
for i, p in enumerate(picks, 1):
print(
f" {i:>2}. {p['종목코드']} {p['종목명']:<10} "
f"점수합={p['호재점수합']:>3} 기사={p['기사수']}건"
)
return picks
# ──────────────────────────────────────────────────────────────────────────────
# 시뮬레이션 (당일 OHLC 사용)
# ──────────────────────────────────────────────────────────────────────────────
def simulate_one(
code: str,
trade_day: date,
*,
stop_loss_pct: float,
take_profit_pct: float,
capital: int,
) -> dict | None:
try:
df = fdr.DataReader(code, trade_day, trade_day)
except Exception as exc:
print(f" · {code} 데이터 조회 실패: {exc}", file=sys.stderr)
return None
if df is None or df.empty:
return None
row = df.iloc[0]
try:
op = float(row["Open"])
hi = float(row["High"])
lo = float(row["Low"])
cl = float(row["Close"])
except Exception:
return None
if op <= 0:
return None
buy_price = op
take_price = buy_price * (1.0 + take_profit_pct / 100.0)
stop_price = buy_price * (1.0 - stop_loss_pct / 100.0)
hit_take = hi >= take_price
hit_stop = lo <= stop_price
if hit_take and hit_stop:
# 일봉만으로는 순서 판정 불가 → 보수적으로 손절 우선
sell_price = stop_price
exit_type = "손절(동시도달)"
elif hit_take:
sell_price = take_price
exit_type = "익절"
elif hit_stop:
sell_price = stop_price
exit_type = "손절"
else:
# 익절/손절 모두 미도달 → 장 마감 시 무조건 종가에 청산 (당일 청산 룰)
sell_price = cl
exit_type = "장마감매도"
qty = int(capital // buy_price)
pnl = (sell_price - buy_price) * qty
ret_pct = (sell_price - buy_price) / buy_price * 100.0
return {
"매수일": trade_day.isoformat(),
"종목코드": code,
"매수가": int(round(buy_price)),
"매도가": int(round(sell_price)),
"수량": qty,
"투자금": int(round(buy_price * qty)),
"평가금": int(round(sell_price * qty)),
"수익률(%)": round(ret_pct, 2),
"손익(원)": int(round(pnl)),
"체결유형": exit_type,
"당일시가": int(round(op)),
"당일고가": int(round(hi)),
"당일저가": int(round(lo)),
"당일종가": int(round(cl)),
}
def simulate_trades(
picks: list[dict],
trade_day: date,
*,
stop_loss_pct: float,
take_profit_pct: float,
capital_total: int,
) -> list[dict]:
if not picks:
return []
per_trade = capital_total // len(picks)
rows: list[dict] = []
for pick in picks:
sim = simulate_one(
pick["종목코드"], trade_day,
stop_loss_pct=stop_loss_pct,
take_profit_pct=take_profit_pct,
capital=per_trade,
)
if not sim:
print(f" · {pick['종목코드']} {pick['종목명']}: OHLC 데이터 없음 → 건너뜀")
continue
sim.update(
{
"종목명": pick["종목명"],
"호재점수합": pick["호재점수합"],
"기사수": pick["기사수"],
"대표키워드": pick["대표키워드"],
"대표기사": pick["대표기사"],
"대표URL": pick["대표URL"],
}
)
rows.append(sim)
return rows
# ──────────────────────────────────────────────────────────────────────────────
# CSV 저장 (날짜명)
# ──────────────────────────────────────────────────────────────────────────────
def save_csv(rows: list[dict], today: date) -> Path:
out_path = HERE / f"paper_trades_{today.isoformat()}.csv"
columns = [
"매수일", "종목코드", "종목명",
"매수가", "매도가", "수량", "투자금", "평가금",
"수익률(%)", "손익(원)", "체결유형",
"당일시가", "당일고가", "당일저가", "당일종가",
"호재점수합", "기사수", "대표키워드", "대표기사", "대표URL",
]
df = pd.DataFrame(rows, columns=columns)
if not df.empty:
total_invest = int(df["투자금"].sum())
total_eval = int(df["평가금"].sum())
total_pnl = int(df["손익(원)"].sum())
total_ret = (total_pnl / total_invest * 100.0) if total_invest > 0 else 0.0
summary = {c: "" for c in columns}
summary["매수일"] = "[합계]"
summary["수량"] = int(df["수량"].sum())
summary["투자금"] = total_invest
summary["평가금"] = total_eval
summary["손익(원)"] = total_pnl
summary["수익률(%)"] = round(total_ret, 2)
df = pd.concat([df, pd.DataFrame([summary])], ignore_index=True)
df.to_csv(out_path, encoding="utf-8-sig", index=False)
return out_path
# ──────────────────────────────────────────────────────────────────────────────
# 백그라운드 스케줄러 (단일 쓰레드)
# ──────────────────────────────────────────────────────────────────────────────
class TradeScheduler(threading.Thread):
"""매일 morning_at, close_at 두 시각에 작업을 트리거하는 단일 백그라운드 쓰레드."""
def __init__(self, args, morning_at: dtime, close_at: dtime) -> None:
super().__init__(name="paper-trade-scheduler", daemon=True)
self.args = args
self.morning_at = morning_at
self.close_at = close_at
self._stop_event = threading.Event()
self._busy_lock = threading.Lock() # 작업 중복 실행 방지
self._busy = threading.Event() # 작업 진행 중 여부 (카운트다운 일시정지용)
self._daily_picks: list[dict] = []
self._daily_picks_date: date | None = None
self._next_run: datetime | None = None
self._next_kind: str = ""
# ── 외부 제어 ────────────────────────────────────────────────────────
def stop(self) -> None:
self._stop_event.set()
def is_busy(self) -> bool:
return self._busy.is_set()
def next_run(self) -> tuple[datetime | None, str]:
return self._next_run, self._next_kind
def _run_with_busy(self, fn) -> None:
"""작업을 실행하면서 busy 플래그를 세우고, 카운트다운과 줄을 분리."""
self._busy.set()
try:
sys.stdout.write("\n")
sys.stdout.flush()
fn()
finally:
self._busy.clear()
# ── 작업 정의 ────────────────────────────────────────────────────────
def morning_task(self) -> None:
with self._busy_lock:
today = date.today()
print(f"\n=== [{datetime.now():%Y-%m-%d %H:%M:%S}] 아침 작업: 호재 종목 선정 ===")
if not is_trading_day(today) and not self.args.force:
print(f" · 오늘({today})은 한국 거래일이 아닙니다. 건너뜁니다.")
self._daily_picks = []
self._daily_picks_date = today
return
try:
self._daily_picks = pick_positive_stocks(self.args)
self._daily_picks_date = today
except Exception as exc:
print(f" ✗ morning_task 실패: {exc}", file=sys.stderr)
traceback.print_exc()
def close_task(self) -> None:
with self._busy_lock:
today = date.today()
print(f"\n=== [{datetime.now():%Y-%m-%d %H:%M:%S}] 마감 후 작업: 시뮬레이션 + CSV 저장 ===")
if not is_trading_day(today) and not self.args.force:
print(f" · 오늘({today})은 한국 거래일이 아닙니다. 건너뜁니다.")
return
picks = self._daily_picks if self._daily_picks_date == today else []
if not picks:
print(" · 메모리에 오늘 후보가 없어 새로 선정합니다.")
try:
picks = pick_positive_stocks(self.args)
except Exception as exc:
print(f" ✗ 호재 후보 재선정 실패: {exc}", file=sys.stderr)
return
if not picks:
print(" · 호재 후보 없음 → 빈 결과 CSV 저장")
out = save_csv([], today)
print(f" ✓ 저장: {out}")
return
print(
f"[시뮬레이션] 기준일=오늘({today}) "
f"익절=+{self.args.take_profit_pct:.1f}% 손절={-self.args.stop_loss_pct:.1f}% "
f"자본={self.args.capital:,}원"
)
try:
rows = simulate_trades(
picks, today,
stop_loss_pct=self.args.stop_loss_pct,
take_profit_pct=self.args.take_profit_pct,
capital_total=self.args.capital,
)
except Exception as exc:
print(f" ✗ 시뮬레이션 실패: {exc}", file=sys.stderr)
traceback.print_exc()
return
for r in rows:
print(
f" · {r['종목코드']} {r['종목명']:<10} "
f"매수 {r['매수가']:>7,} 매도 {r['매도가']:>7,} "
f"수익률 {r['수익률(%)']:>+6.2f}% ({r['체결유형']})"
)
out = save_csv(rows, today)
print(f" ✓ 저장: {out}")
# ── 다음 실행 시각 계산 ──────────────────────────────────────────────
def _compute_next(self) -> tuple[datetime, str]:
now = datetime.now()
today = now.date()
m_dt = datetime.combine(today, self.morning_at)
c_dt = datetime.combine(today, self.close_at)
if now < m_dt:
return m_dt, "morning"
if now < c_dt:
return c_dt, "close"
return datetime.combine(today + timedelta(days=1), self.morning_at), "morning"
def _interruptible_sleep(self, seconds: float) -> bool:
"""seconds 초 동안 sleep. stop 신호 받으면 True 반환."""
return self._stop_event.wait(timeout=max(0.0, seconds))
# ── 메인 루프 ────────────────────────────────────────────────────────
def run(self) -> None:
print(
f"[scheduler] 시작: morning={self.morning_at.strftime('%H:%M')} / "
f"close={self.close_at.strftime('%H:%M')}"
)
while not self._stop_event.is_set():
target, kind = self._compute_next()
self._next_run, self._next_kind = target, kind
wait_s = (target - datetime.now()).total_seconds()
print(
f"[scheduler] 다음 작업: {kind} @ {target:%Y-%m-%d %H:%M:%S} "
f"({int(wait_s)}초 대기)"
)
# 1분 단위로 끊어 자며, 시계 변경/슬립 복귀를 견고하게 처리
while not self._stop_event.is_set():
remaining = (target - datetime.now()).total_seconds()
if remaining <= 0:
break
if self._interruptible_sleep(min(remaining, 60.0)):
return
if self._stop_event.is_set():
return
try:
if kind == "morning":
self._run_with_busy(self.morning_task)
else:
self._run_with_busy(self.close_task)
except Exception as exc:
print(f"[scheduler] 작업 중 예외: {exc}", file=sys.stderr)
traceback.print_exc()
# 같은 시각 재진입 방지 (1초 이상 대기)
self._interruptible_sleep(1.0)
print("[scheduler] 종료")
# ──────────────────────────────────────────────────────────────────────────────
# CLI / 메인 루프
# ──────────────────────────────────────────────────────────────────────────────
def parse_hhmm(s: str) -> dtime:
try:
hh, mm = s.split(":")
return dtime(hour=int(hh), minute=int(mm))
except Exception as exc: # pragma: no cover
raise argparse.ArgumentTypeError(f"잘못된 시각 형식: {s!r} (HH:MM)") from exc
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="호재 종목 모의 자동매매 스케줄러 (실거래 아님)")
p.add_argument("--runonce", action="store_true",
help="스케줄러를 띄우지 않고 morning + close 를 즉시 1회 실행 후 종료")
p.add_argument("--force", action="store_true",
help="거래일이 아니어도 강제 실행")
p.add_argument("--no-startup-run", action="store_true",
help="시작 시 1회 자동 실행을 비활성화 (다음 트리거만 기다림)")
p.add_argument("--morning-at", type=str, default=DEFAULTS["morning_at"],
help="아침 작업 시각 HH:MM (기본 07:00, 호재 종목 선정)")
p.add_argument("--close-at", type=str, default=DEFAULTS["close_at"],
help="마감 후 작업 시각 HH:MM (기본 16:30, 시뮬레이션+CSV 저장)")
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("--stop-loss-pct", type=float, default=DEFAULTS["stop_loss_pct"],
help="손절 비율 %% (기본 3.0)")
p.add_argument("--take-profit-pct", type=float, default=DEFAULTS["take_profit_pct"],
help="익절 비율 %% (기본 5.0)")
p.add_argument("--capital", type=int, default=DEFAULTS["capital"])
p.add_argument("--delay", type=float, default=DEFAULTS["delay"])
return p.parse_args()
def _format_remaining(seconds: int) -> str:
if seconds < 0:
seconds = 0
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
return f"{h:02d}:{m:02d}:{s:02d}"
def countdown_loop(sched: TradeScheduler, stop_event: threading.Event) -> None:
"""메인 쓰레드에서 다음 트리거까지 남은 시간을 1초마다 한 줄로 갱신.
- 스케줄러가 작업 중이면(self._busy) 카운트다운을 잠시 멈춰
작업 로그가 카운트다운 줄과 섞이지 않게 함.
- 작업이 끝나면 다시 갱신을 재개.
"""
last_was_countdown_line = False
while not stop_event.is_set():
if sched.is_busy():
last_was_countdown_line = False
stop_event.wait(0.5)
continue
target, kind = sched.next_run()
if target is None:
stop_event.wait(0.3)
continue
secs = int((target - datetime.now()).total_seconds())
line = (
f"\r[다음 작업] {kind:<7} "
f"@ {target:%Y-%m-%d %H:%M:%S} "
f"남은시간 {_format_remaining(secs)} "
f"(Ctrl-C 로 종료)"
)
sys.stdout.write(line)
sys.stdout.flush()
last_was_countdown_line = True
stop_event.wait(1.0)
if last_was_countdown_line:
sys.stdout.write("\n")
sys.stdout.flush()
def main() -> int:
args = parse_args()
morning_at = parse_hhmm(args.morning_at)
close_at = parse_hhmm(args.close_at)
sched = TradeScheduler(args, morning_at, close_at)
if args.runonce:
# 스케줄러 없이 즉시 1회만 실행
sched.morning_task()
sched.close_task()
return 0
today = date.today()
cal_src = "XKRX" if _XKRX is not None else ("holidays.KR(+KRX 보정)" if _holidays is not None else "평일+KRX 보정")
print(f"=== 모의 자동매매 시작 @ {datetime.now():%Y-%m-%d %H:%M:%S} ===")
print(f" · morning={args.morning_at} / close={args.close_at} (실거래 아님)")
print(f" · 거래일 캘린더: {cal_src}")
print(f" · 오늘({today}) 거래일? {is_trading_day(today)}")
print(f" · 내일({today + timedelta(days=1)}) 거래일? {is_trading_day(today + timedelta(days=1))}")
if not args.no_startup_run:
print("[startup] 즉시 1회 실행 — morning + close")
sched.morning_task()
sched.close_task()
sched.start()
stop_event = threading.Event()
try:
countdown_loop(sched, stop_event)
except KeyboardInterrupt:
stop_event.set()
sys.stdout.write("\n")
sys.stdout.flush()
finally:
print("[main] 스케줄러 정지 중...")
sched.stop()
sched.join(timeout=5)
print("[main] 종료")
return 0
if __name__ == "__main__":
sys.exit(main())