반응형
방법 구조
- 데이터 소스
- KRX 장전(Pre-market) 수급 데이터
(외국인·기관 매수/매도 TOP 종목, 업종별 흐름) - HTS/MTS API (예: 키움 Open API, 대신증권 API, NH투자증권 API 등)
→ 실시간 체결·호가·수급 데이터 가능 - 뉴스/속보 API (네이버 금융, 연합뉴스, 인포스탁, 이데일리)
- 테마주·대장주 랭킹 사이트 (FnGuide, 매경, 팍스넷, 씽크풀)
- KRX 장전(Pre-market) 수급 데이터
- 수집 시간대
- 08:00~08:50 : 장전 시간외 거래·예상체결가, 수급 매수세 체크
- 08:50~09:00 : 시초가 직전 변동 종목·거래량 급증 종목 감지
- 출력 형태
- CSV / 엑셀 파일
- 텔레그램·카카오톡 알림
- 블로그 자동 포스팅 (티스토리 API)
- 자동화 구현 예시
- Python + 증권사 API + News RSS
→ cron 스케줄러로 매일 08:00 실행 - 예:
- Python + 증권사 API + News RSS
# 1. 외국인 매수 상위 종목 API 호출
# 2. 거래량·등락률·뉴스 제목 매칭
# 3. 테마 분석 후 상위 종목 필터링
# 4. 파일 저장 및 텔레그램 발송포함 가능 정보
- 외국인 매수 TOP 10
- 장전 예상체결가 급등 종목
- 오늘 뉴스 기반 테마주
- 거래량 급증 상위주
- 전일 대비 호재/악재 뉴스 정리
08 Preopen Foreign Monitor
"""
08_preopen_foreign_monitor.py
Korea market pre-open monitor (08:00-09:00) — 외국인 매수/특징주/예상체결/테마 요약을
아침 8~9시 사이(특히 08:50~08:59)에 실행해 Telegram으로 요약을 전송합니다.
설명
- Kiwoom OpenAPI+ (WINDOWS, KOA Studio) 를 이용해 TR(opt10034, opt10029 등)을 호출합니다.
- pykiwoom의 KiwoomManager를 사용해 TR을 실행하고 결과를 DataFrame으로 정리합니다.
- Telegram Bot으로 요약 메시지(+CSV)를 전송합니다.
중요: 이 스크립트는 Windows 환경에서 Kiwoom OpenAPI+ (및 KOA Studio)가 설치되어
있고, 사용자가 키움 로그인을 완료한 상태에서 실행되어야 정상 동작합니다.
참고 자료:
- pykiwoom (GitHub) : https://github.com/sharebook-kr/pykiwoom
- Kiwoom OpenAPI TR 목록 및 설명 (예: opt10034, opt10029 등) : 여러 가이드/블로그 정리 자료
(예시: https://jackerlab.com/kiwoom-api-tr-list/ , https://trustyou.tistory.com/228)
- KRX Open Data (참고): https://data.krx.co.kr/
사용법 요약
1) Python 환경 준비 (권장 Python 3.9+)
pip install pykiwoom pandas pyyaml requests
2) config.yaml 생성 (아래 예시를 참고)
- telegram_token, telegram_chat_id, run_time 등 설정
3) 윈도우 작업 스케줄러에 등록하여 매일 08:50에 실행하도록 설정
4) Kiwoom KOA Studio 로그인 필요 (자동 로그인 설정 가능)
구성
- fetch 외국인 상위: opt10034
- fetch 예상체결 상위: opt10029
- fetch 거래량/급등: opt10023/opt10027 (원하면 추가)
- 메시지 전송: Telegram
주의: TR 입력 파라미터 이름(예: '시장구분','매매구분','기간')이나 출력 필드
은 키움 OpenAPI 버전/KOA Studio 버전에 따라 달라질 수 있습니다. 필요 시 KOA의
TR 문서를 확인해 입력 키/출력 필드를 조정하세요.
"""
import os
import sys
import time
import yaml
import logging
import datetime
import pandas as pd
import requests
# pykiwoom 패키지(sharebook-kr/pykiwoom)를 사용합니다.
# 설치: pip install pykiwoom
try:
from pykiwoom.kiwoom import KiwoomManager
except Exception as e:
KiwoomManager = None
# ---------------------------
# 설정파일 예시 (config.yaml)
# ---------------------------
CONFIG_EXAMPLE = '''
telegram_token: "<YOUR_TELEGRAM_BOT_TOKEN>"
telegram_chat_id: "<YOUR_CHAT_ID>"
# 사용할 TR/시장 관련 기본값
market: "0" # 0: 전체(통상), 1: KOSPI, 2: KOSDAQ (환경에 따라 다름)
foreign_period: "1" # 기간: 1(당일) 등 (KOA TR에 맞춰 조정)
run_time: "08:50" # 스케줄러에서 실행하는 시간(참고용)
output_csv_dir: "./outputs"
# 디버그 모드(참고용). True 면 Kiwoom 대신 로컬 모의 데이터 사용 가능
debug_mode: False
'''
# ---------------------------
# 로깅 설정
# ---------------------------
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger('preopen')
# ---------------------------
# 헬퍼 함수들
# ---------------------------
def ensure_dir(path):
os.makedirs(path, exist_ok=True)
def load_config(path='config.yaml'):
if not os.path.exists(path):
with open('config.example.yaml', 'w', encoding='utf-8') as f:
f.write(CONFIG_EXAMPLE)
raise FileNotFoundError(f"{path} not found. A sample 'config.example.yaml' was created. Please edit and retry.")
with open(path, 'r', encoding='utf-8') as f:
cfg = yaml.safe_load(f)
return cfg
# Kiwoom TR 호출 래퍼 (KiwoomManager 기반)
def run_tr(km, rqname, trcode, input_params, output_fields, screen='0101', next_flag='0'):
"""
km: KiwoomManager 인스턴스
rqname: 임의 이름
trcode: ex) 'opt10034'
input_params: dict, ex) {"시장구분": "0", "매매구분": "1", "기간": "1"}
output_fields: list of field names expected (키움 TR의 출력 필드명)
반환: pandas.DataFrame
주의: TR의 입력 파라미터 이름(키)과 출력 필드명은 KOA Studio/TR문서를
참고해 정확히 기입해야 합니다. 예시는 보편적으로 사용되는 필드를 제안합니다.
"""
if km is None:
raise RuntimeError('KiwoomManager is not available. Make sure pykiwoom installed and Kiwoom OpenAPI set up.')
tr_cmd = {
'rqname': rqname,
'trcode': trcode,
'next': next_flag,
'screen': screen,
'input': input_params,
'output': output_fields
}
log.info(f"TR request {trcode} (rqname={rqname}) inputs={input_params}")
km.put_tr(tr_cmd)
data = km.get_tr()
# pykiwoom의 get_tr은 list of dict 또는 DataFrame을 반환할 수 있으니 안전하게 처리
if isinstance(data, pd.DataFrame):
df = data
else:
try:
df = pd.DataFrame(data)
except Exception:
# 빈 DataFrame 반환
df = pd.DataFrame()
return df
# 외국인 기간별 매매 상위(opt10034) 가져오기
def fetch_foreign_top(km, market='0', trade_type='1', period='1', topn=20):
"""market/trade_type/period 파라미터는 KOA/OpenAPI 문서를 참고해 조정하세요."""
input_params = {
'시장구분': market,
'매매구분': trade_type,
'기간': period
}
# 출력필드는 TR마다 다를 수 있으니 필요시 KOA 문서에 맞춰 수정하세요.
output_fields = [
'순위','종목코드','종목명','현재가','전일대비기호','전일대비','매도호가','매수호가','거래량','순매수량','취득가능주식수'
]
df = run_tr(km, 'opt10034_foreign_top', 'opt10034', input_params, output_fields)
if df.empty:
log.warning('opt10034 returned empty result')
return df
# 정리: 종목코드가 없을 때 인덱스 정리 등
df = df.head(topn)
return df
# 예상체결 등락률 상위(opt10029) 가져오기
def fetch_predicted_open(km, market='0', topn=50):
# input 파라미터 명은 KOA 문서와 버전에 따라 다름. 아래는 예시.
input_params = {
'조회시장': market,
# '조회조건': '1' # 필요하면 추가
}
output_fields = [
'순위','종목코드','종목명','예상체결가','기준가','대비','등락률','예상체결량','매도호가','매수호가'
]
df = run_tr(km, 'opt10029_predicted', 'opt10029', input_params, output_fields)
if df.empty:
log.warning('opt10029 returned empty result')
return df
return df.head(topn)
# 메시지 생성 (간단한 텍스트 요약)
def build_message(df_foreign, df_pred):
lines = []
lines.append('\U00002705 <b>장전(예상) 외국인 매수 상위</b>')
if df_foreign is None or df_foreign.empty:
lines.append('데이터 없음 (opt10034 결과 없음)')
else:
for _, r in df_foreign.head(10).iterrows():
code = r.get('종목코드','-')
name = r.get('종목명','-')
net = r.get('순매수량', r.get('순매수','-'))
price = r.get('현재가','-')
lines.append(f"{name} ({code}) — 순매수: {net} / 현재가: {price}")
lines.append('\n\U0001F50D <b>동시호가 예상체결 상위</b>')
if df_pred is None or df_pred.empty:
lines.append('데이터 없음 (opt10029 결과 없음)')
else:
for _, r in df_pred.head(10).iterrows():
lines.append(f"{r.get('종목명','-')} — 예상등락률: {r.get('등락률','-')}, 예상체결가: {r.get('예상체결가','-')}")
footer = f"\n(생성시간: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')})"
return '\n'.join(lines) + footer
# Telegram 전송
def send_telegram_message(token, chat_id, text, csv_path=None):
base = f'https://api.telegram.org/bot{token}'
send_url = base + '/sendMessage'
payload = {
'chat_id': chat_id,
'text': text,
'parse_mode': 'HTML',
'disable_web_page_preview': True
}
r = requests.post(send_url, data=payload, timeout=15)
log.info(f'Telegram message send status: {r.status_code}')
if csv_path and os.path.exists(csv_path):
files = {'document': open(csv_path, 'rb')}
doc_url = base + '/sendDocument'
data = {'chat_id': chat_id, 'caption': '장전(예상) 요약 CSV'}
r2 = requests.post(doc_url, data=data, files=files, timeout=30)
log.info(f'Telegram document send status: {r2.status_code}')
# 메인 플로우
def main():
try:
cfg = load_config('config.yaml')
except FileNotFoundError as e:
log.error(str(e))
sys.exit(1)
ensure_dir(cfg.get('output_csv_dir','./outputs'))
# Kiwoom 초기화
if cfg.get('debug_mode'):
km = None
log.warning('Debug mode ON — Kiwoom 호출을 스킵합니다. (테스트용)')
else:
if KiwoomManager is None:
log.error('pykiwoom 모듈을 찾을 수 없습니다. pip install pykiwoom 후 재시도하세요.')
sys.exit(1)
km = KiwoomManager()
# KiwoomManager는 내부서브프로세스로 키움 OCX를 띄워서 통신합니다.
# KOA Studio에서 로그인이 되어 있거나, 프로그램이 로그인 창을 띄우게 됩니다.
log.info('KiwoomManager 초기화 완료 — 로그인 창에서 로그인(또는 자동로그인)하세요.')
# 잠시 대기(로그인 안정화)
time.sleep(1)
try:
df_fore = pd.DataFrame()
df_pred = pd.DataFrame()
if not cfg.get('debug_mode'):
# 외국인 상위
try:
df_fore = fetch_foreign_top(km, market=str(cfg.get('market','0')), trade_type='1', period=str(cfg.get('foreign_period','1')), topn=30)
log.info(f'외국인 상위 rows: {len(df_fore)}')
except Exception as e:
log.exception('외국인 상위 조회 중 오류')
# 예상체결
try:
df_pred = fetch_predicted_open(km, market=str(cfg.get('market','0')))
log.info(f'예상체결 rows: {len(df_pred)}')
except Exception as e:
log.exception('예상체결 조회 중 오류')
else:
# debug 샘플 DataFrame
df_fore = pd.DataFrame([{'순위':1, '종목코드':'005930', '종목명':'삼성전자', '현재가':60000, '순매수량':12345}])
df_pred = pd.DataFrame([{'순위':1, '종목코드':'005930', '종목명':'삼성전자', '예상체결가':60500, '등락률': '0.83%'}])
# 병합하여 CSV로 저장
today = datetime.date.today().isoformat()
out_csv = os.path.join(cfg.get('output_csv_dir','./outputs'), f'preopen_summary_{today}.csv')
if not df_fore.empty and not df_pred.empty:
# 양쪽에 종목코드가 있다면 병합
try:
left = df_fore.copy()
right = df_pred[['종목코드','예상체결가','등락률']].copy()
merged = pd.merge(left, right, on='종목코드', how='left')
merged.to_csv(out_csv, index=False, encoding='utf-8-sig')
except Exception:
# fallback
df_fore.to_csv(out_csv, index=False, encoding='utf-8-sig')
elif not df_fore.empty:
df_fore.to_csv(out_csv, index=False, encoding='utf-8-sig')
elif not df_pred.empty:
df_pred.to_csv(out_csv, index=False, encoding='utf-8-sig')
else:
# 빈 파일 생성
pd.DataFrame().to_csv(out_csv, index=False)
msg = build_message(df_fore, df_pred)
send_telegram_message(cfg['telegram_token'], cfg['telegram_chat_id'], msg, csv_path=out_csv)
log.info('모니터링 실행 완료 — Telegram 발송 완료')
except Exception as e:
log.exception('메인 실행 중 오류 발생')
if __name__ == '__main__':
main()반응형