kis_trader/remove/legacy_root/kiwoom_rest_api/trader.py

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()