325 lines
11 KiB
Python
325 lines
11 KiB
Python
import time
|
|
import json
|
|
import datetime
|
|
import pandas as pd
|
|
import numpy as np
|
|
import os
|
|
|
|
# =========================================================
|
|
# [Part 1] 데이터 표준화 (Data Class)
|
|
# 증권사마다 주는 데이터 이름이 다르므로, 우리 봇이 쓸 공통 이름으로 정의
|
|
# =========================================================
|
|
class StockData:
|
|
def __init__(self, code, name, price, open_p, high, low, close, volume):
|
|
self.code = code
|
|
self.name = name
|
|
self.current_price = price # 현재가
|
|
self.open = open_p # 시가
|
|
self.high = high # 고가
|
|
self.low = low # 저가
|
|
self.close = close # 종가
|
|
self.volume = volume # 거래량
|
|
|
|
|
|
class OrderbookData:
|
|
def __init__(self, total_bid, total_ask):
|
|
self.total_bid = total_bid # 총 매수 잔량 (살 사람)
|
|
self.total_ask = total_ask # 총 매도 잔량 (팔 사람)
|
|
|
|
|
|
# =========================================================
|
|
# [Part 2] 증권사 연결 인터페이스 (이 부분만 채우세요!)
|
|
# =========================================================
|
|
class BrokerAPI:
|
|
"""
|
|
[안내] 여기에 한국투자증권(KIS)이나 다른 API 코드를 연결합니다.
|
|
봇 로직은 이 함수들을 호출해서 데이터를 받아옵니다.
|
|
"""
|
|
|
|
def get_rank_list(self):
|
|
# [To-Do] 거래대금 상위 종목 리스트 반환 (API 호출)
|
|
# return [{"code": "005930", "price": 60000}, ...]
|
|
return [] # 테스트용 빈 리스트
|
|
|
|
def get_ohlcv_limit(self, code, timeframe='3m', limit=100):
|
|
# [To-Do] 특정 종목의 캔들(OHLCV) 데이터 반환 (RSI, MA 계산용)
|
|
# DataFrame 형태로 반환: columns=['open', 'high', 'low', 'close', 'volume']
|
|
return pd.DataFrame()
|
|
|
|
def get_daily_ohlc_yesterday(self, code):
|
|
# [To-Do] 어제 일봉 데이터 반환 (Dynamic K 계산용)
|
|
# return StockData(...)
|
|
pass
|
|
|
|
def get_current_data(self, code):
|
|
# [To-Do] 현재가 데이터 반환
|
|
pass
|
|
|
|
def get_orderbook(self, code):
|
|
# [To-Do] 호가창 데이터 반환
|
|
# return OrderbookData(bid, ask)
|
|
pass
|
|
|
|
def buy_market_order(self, code, qty):
|
|
# [To-Do] 시장가 매수 주문
|
|
print(f"✅ [API] {code} {qty}주 매수 주문 전송 완료")
|
|
|
|
def sell_market_order(self, code, qty):
|
|
# [To-Do] 시장가 매도 주문
|
|
print(f"✅ [API] {code} {qty}주 매도 주문 전송 완료")
|
|
|
|
|
|
# =========================================================
|
|
# [Part 3] 정훈님의 정글 서바이버 봇 (핵심 로직)
|
|
# =========================================================
|
|
class JungleSurvivorBot:
|
|
def __init__(self, broker_api, budget=100000):
|
|
self.api = broker_api # 증권사 API 객체 연결
|
|
self.budget = budget # 운용 자금 (10만원)
|
|
self.target_list = [] # 감시 종목 리스트
|
|
self.portfolio = {} # 보유 종목: {'code': {'buy_price': 1000, 'max_price': 1100, 'qty': 10}}
|
|
|
|
# --- [전략 파라미터] ---
|
|
self.rsi_period = 11 # RSI 기간 (남들 14보다 빠르게)
|
|
self.rsi_sell_std = 68 # RSI 매도 기준 (70보다 조금 낮게 선취매도)
|
|
self.trailing_gap = 0.01 # 트레일링 스탑 (고점 대비 1% 하락 시 매도)
|
|
self.target_profit = 0.035 # 목표 수익률 (3.5%면 무조건 익절)
|
|
self.stop_loss = -0.02 # 손절매 (-2%)
|
|
|
|
# -----------------------------------------------------
|
|
# [3-1] 계산기 (Indicators)
|
|
# -----------------------------------------------------
|
|
def calc_rsi(self, df):
|
|
"""RSI 지표 계산"""
|
|
delta = df['close'].diff()
|
|
gain = (delta.where(delta > 0, 0)).rolling(window=self.rsi_period).mean()
|
|
loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean()
|
|
rs = gain / loss
|
|
return 100 - (100 / (1 + rs))
|
|
|
|
def calc_ma(self, df, period=20):
|
|
"""이동평균선 계산"""
|
|
return df['close'].rolling(window=period).mean()
|
|
|
|
def calc_dynamic_k(self, yesterday_data):
|
|
"""
|
|
변동성 돌파 K값 실시간 변경 (노이즈 비율 활용)
|
|
노이즈가 심하면 K값을 높여서 진입 장벽을 높임
|
|
"""
|
|
if yesterday_data is None: return 0.5 # 데이터 없으면 기본값
|
|
|
|
total_range = yesterday_data.high - yesterday_data.low
|
|
body_range = abs(yesterday_data.open - yesterday_data.close)
|
|
|
|
if total_range == 0: return 0.5
|
|
|
|
noise_ratio = 1 - (body_range / total_range)
|
|
# K값은 최소 0.3 ~ 최대 0.9 사이로 제한
|
|
return max(0.3, min(noise_ratio, 0.9))
|
|
|
|
# -----------------------------------------------------
|
|
# [3-2] 타겟 선정 (Dynamic Universe)
|
|
# -----------------------------------------------------
|
|
def update_universe(self):
|
|
print("\n🔄 [리스트 갱신] 실시간 주도주 스캔 중...")
|
|
raw_list = self.api.get_rank_list()
|
|
|
|
new_targets = []
|
|
for item in raw_list:
|
|
code = item['code']
|
|
price = item['price']
|
|
|
|
# [필터] 1. 가격 (2천원 ~ 5만원) / 10만원으로 살 수 있어야 함
|
|
if not (2000 <= price <= 50000) or (price > self.budget):
|
|
continue
|
|
|
|
# [필터] 2. 프로그램 매매 추이 (API 제공 시 추가)
|
|
# if item['prog_buy'] < 0: continue
|
|
|
|
new_targets.append(code)
|
|
|
|
# JSON 파일 저장 (로그용)
|
|
with open('target_universe.json', 'w') as f:
|
|
json.dump(new_targets, f)
|
|
|
|
self.target_list = new_targets
|
|
print(f"✅ 타겟 갱신 완료: {len(self.target_list)}개 종목 감시 시작")
|
|
|
|
# -----------------------------------------------------
|
|
# [3-3] 매수 판단 로직 (살까 말까?)
|
|
# -----------------------------------------------------
|
|
def check_buy(self, code):
|
|
# 1. 데이터 수집
|
|
df_candle = self.api.get_ohlcv_limit(code) # 3분봉
|
|
curr_data = self.api.get_current_data(code)
|
|
yesterday = self.api.get_daily_ohlc_yesterday(code)
|
|
orderbook = self.api.get_orderbook(code)
|
|
prog_net_buy = self.api.get_program_net_buy(code) # 프로그램 수급 확인
|
|
if prog_net_buy < 0:
|
|
print(f"🚫 {code} 패스: 프로그램 매도세 ({prog_net_buy})")
|
|
return False
|
|
if len(df_candle) < 20: return False # 데이터 부족
|
|
|
|
current_price = curr_data.current_price
|
|
|
|
# 2. 지표 계산
|
|
ma20 = self.calc_ma(df_candle, 20).iloc[-1]
|
|
k_val = self.calc_dynamic_k(yesterday)
|
|
|
|
# 3. 변동성 돌파 목표가 계산
|
|
# 목표가 = 오늘 시가 + (어제 변동폭 * K)
|
|
prev_range = yesterday.high - yesterday.low
|
|
breakout_price = curr_data.open + (prev_range * k_val)
|
|
|
|
print(f"🧐 {code} 분석: 현재가 {current_price} | 20선 {ma20:.0f} | 목표가 {breakout_price:.0f} (K:{k_val:.2f})")
|
|
|
|
# --- [매수 조건 (AND)] ---
|
|
# A. 추세: 현재가가 20일선 위에 있는가?
|
|
cond_trend = current_price >= ma20
|
|
|
|
# B. 변동성 돌파: 의미 있는 가격(목표가)을 넘었는가?
|
|
cond_breakout = current_price >= breakout_price
|
|
|
|
# C. 수급 안전판: 매수 잔량이 매도 잔량보다 2배 많은가?
|
|
cond_safe = orderbook.total_bid >= (orderbook.total_ask * 2)
|
|
|
|
if cond_trend and cond_breakout and cond_safe:
|
|
print(f"🚀 [매수 신호] {code} 발견! (Trend+Breakout+Safe)")
|
|
return True
|
|
|
|
return False
|
|
|
|
# -----------------------------------------------------
|
|
# [3-4] 매도 판단 로직 (팔까 말까?)
|
|
# -----------------------------------------------------
|
|
def check_sell(self, code):
|
|
if code not in self.portfolio: return False
|
|
|
|
info = self.portfolio[code]
|
|
buy_price = info['buy_price']
|
|
max_price = info['max_price'] # 트레일링 스탑용 고점
|
|
qty = info['qty']
|
|
|
|
curr_data = self.api.get_current_data(code)
|
|
curr_price = curr_data.current_price
|
|
|
|
# 고점 갱신 (트레일링 스탑용)
|
|
if curr_price > max_price:
|
|
self.portfolio[code]['max_price'] = curr_price
|
|
max_price = curr_price
|
|
|
|
# 지표 계산
|
|
df_candle = self.api.get_ohlcv_limit(code)
|
|
rsi = self.calc_rsi(df_candle).iloc[-1]
|
|
|
|
# 수익률 계산
|
|
profit_rate = (curr_price - buy_price) / buy_price
|
|
drop_from_high = (max_price - curr_price) / max_price
|
|
|
|
print(f"👀 {code} 보유중: 수익률 {profit_rate * 100:.2f}% | 고점대비하락 {drop_from_high * 100:.2f}% | RSI {rsi:.1f}")
|
|
|
|
# --- [매도 조건 (OR)] ---
|
|
|
|
# 1. [목표 달성] 3.5% 먹으면 묻지도 따지지도 말고 익절
|
|
if profit_rate >= self.target_profit:
|
|
print("💰 목표 수익 달성! 매도!")
|
|
return True
|
|
|
|
# 2. [트레일링 스탑] 고점 대비 1% 빠지면 익절/청산
|
|
if drop_from_high >= self.trailing_gap and profit_rate > 0:
|
|
print("📉 트레일링 스탑 발동! 매도!")
|
|
return True
|
|
|
|
# 3. [조기 퇴근] RSI가 68 넘으면 과열이므로 선취 매도
|
|
if rsi >= self.rsi_sell_std:
|
|
print("🔥 RSI 과열! 조기 매도!")
|
|
return True
|
|
|
|
# 4. [손절] -2% 살려주세요
|
|
if profit_rate <= self.stop_loss:
|
|
print("😭 손절매 실행...")
|
|
return True
|
|
|
|
return False
|
|
|
|
# -----------------------------------------------------
|
|
# [3-5] 메인 루프 (실행기)
|
|
# -----------------------------------------------------
|
|
def run(self):
|
|
print("🤖 정글 서바이버 봇 가동 시작!")
|
|
|
|
# [수정 1] 봇 켜자마자 일단 유니버스(감시 종목)부터 만들고 시작!
|
|
# 파일이 없거나 비어있을 수 있으니 강제 갱신 1회 실행
|
|
if not os.path.exists('target_universe.json'):
|
|
print("📂 타겟 파일이 없어서 새로 만듭니다...")
|
|
self.update_universe()
|
|
else:
|
|
# 파일이 있어도 비어있는지 확인
|
|
try:
|
|
with open('target_universe.json', 'r') as f:
|
|
json.load(f)
|
|
except json.JSONDecodeError:
|
|
print("📂 타겟 파일이 비어있어서 새로 채웁니다...")
|
|
self.update_universe()
|
|
|
|
while True:
|
|
# 1. 현재 시간 체크
|
|
now = datetime.datetime.now()
|
|
|
|
# 2. 유니버스 갱신 (30분마다)
|
|
if now.minute % 30 == 0 and now.second < 5:
|
|
self.update_universe()
|
|
time.sleep(5)
|
|
|
|
# 3. 매도 감시 (보유 종목 먼저 체크)
|
|
sell_list = []
|
|
for code in list(self.portfolio.keys()):
|
|
if self.check_sell(code):
|
|
# 매도 실행
|
|
qty = self.portfolio[code]['qty']
|
|
self.api.sell_market_order(code, qty)
|
|
sell_list.append(code)
|
|
|
|
# 포트폴리오에서 삭제
|
|
for code in sell_list:
|
|
del self.portfolio[code]
|
|
|
|
# 4. 매수 감시 (보유 종목 없을 때 or 자금 남을 때)
|
|
if len(self.portfolio) < 3:
|
|
# [수정 2] 읽을 때도 안전장치 추가
|
|
try:
|
|
with open('target_universe.json', 'r') as f:
|
|
targets = json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
targets = [] # 에러 나면 그냥 빈 리스트로 처리
|
|
|
|
for code in targets:
|
|
if code in self.portfolio: continue # 이미 가진 건 패스
|
|
|
|
if self.check_buy(code):
|
|
# 매수 실행 (예: 10만원어치 계산)
|
|
curr_price = self.api.get_current_data(code).current_price
|
|
qty = int(self.budget / 3 / curr_price) # 자금의 1/3 투입
|
|
if qty > 0:
|
|
self.api.buy_market_order(code, qty)
|
|
# 포트폴리오 등록
|
|
self.portfolio[code] = {
|
|
'buy_price': curr_price,
|
|
'max_price': curr_price,
|
|
'qty': qty
|
|
}
|
|
break # 한 턴에 하나만 산다 (서버 부하 방지)
|
|
|
|
time.sleep(1) # 1초 대기
|
|
|
|
|
|
# =========================================================
|
|
# 실행부
|
|
# =========================================================
|
|
if __name__ == "__main__":
|
|
# 1. API 객체 생성 (나중에 키움/한투 코드로 교체될 부분)
|
|
my_broker = BrokerAPI()
|
|
|
|
# 2. 봇 생성 및 실행
|
|
bot = JungleSurvivorBot(my_broker, budget=100000)
|
|
bot.run() |