예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
Account.py
"""실제 Upbit 거래를 위한 Account 클래스."""
import importlib.util
import math
import sys
import time
import logging
from collections import deque
from pathlib import Path
import pymysql
import pyupbit
from private.db import *
def _load_upbit_keys():
"""private/upbit.key.py 의 access_key, secret_key 를 읽는다."""
path = Path(__file__).resolve().parent / "private" / "upbit.key.py"
spec = importlib.util.spec_from_file_location("upbit_key", path)
if spec is None or spec.loader is None:
raise FileNotFoundError(path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
access = getattr(mod, "access_key", "") or ""
secret = getattr(mod, "secret_key", "") or ""
if not access or not secret:
raise RuntimeError(f"{path.name} 에 access_key, secret_key 가 없습니다.")
return access, secret
logger = logging.getLogger(__name__)
def _record_trade_to_db(*args, **kwargs):
"""메인 스크립트에 정의된 record_trade_to_db 호출 (직접 실행 / 모듈 import 모두)."""
for modname in ("__main__", "main_ri_findsell_dqn_v8_real", "main_llm_real"):
mod = sys.modules.get(modname)
if mod is not None and hasattr(mod, "record_trade_to_db"):
return mod.record_trade_to_db(*args, **kwargs)
raise RuntimeError("record_trade_to_db를 찾을 수 없습니다. 메인 모듈이 먼저 로드되어야 합니다.")
class Account:
"""실제 Upbit 거래를 위한 Account 클래스"""
def connect_upbit(self):
sqlcon = None
cursor = None
try:
# 데이터베이스 연결
sqlcon = pymysql.connect(host=oracle_cloud, user=db_user, password=db_password, db=db_name, connect_timeout=10)
cursor = sqlcon.cursor(pymysql.cursors.DictCursor)
# SQL 쿼리 실행
self.app_key, self.secu_key = _load_upbit_keys()
query = """
SELECT
BASE_BALANCE,
TRADE_YN
FROM main.TN_PAYMENT_ACNT
WHERE USER_ID = '%s'
""" % self.user_id
cursor.execute(query)
result = cursor.fetchone()
# 결과 처리
if result:
self.orgin_money = result["BASE_BALANCE"]
self.trade_yn = result["TRADE_YN"]
# 실제 투자 원금 확인 및 조정
logger.info(f"DB BASE_BALANCE: {self.orgin_money:,}원")
return pyupbit.Upbit(access=self.app_key, secret=self.secu_key)
except Exception as e:
logger.error(f"Upbit 연결 오류: {e}")
return None
finally:
if cursor is not None:
cursor.close()
if sqlcon is not None:
sqlcon.close()
def _auto_estimate_investment(self):
"""현재 포트폴리오 가치를 기반으로 투자 원금 자동 추정"""
try:
upbit_temp = pyupbit.Upbit(access=self.app_key, secret=self.secu_key)
temp_balance = upbit_temp.get_balance() or 0
temp_unit = upbit_temp.get_balance_t(self.ticker) or 0
current_price = pyupbit.get_current_price(self.ticker)
except Exception as e:
logger.error(f"자동 추정 실패: {str(e)}, 기본값 유지")
def get_trade_yn(self):
sqlcon = None
cursor = None
try:
# 데이터베이스 연결
sqlcon = pymysql.connect(host=oracle_cloud, user=db_user, password=db_password, db=db_name, connect_timeout=10)
cursor = sqlcon.cursor(pymysql.cursors.DictCursor)
# SQL 쿼리 실행
query = """
SELECT
TRADE_YN
FROM main.TN_PAYMENT_ACNT
WHERE USER_ID = '%s'
""" % self.user_id
cursor.execute(query)
result = cursor.fetchone()
# 결과 처리
if result:
self.trade_yn = result["TRADE_YN"]
return
except Exception as e:
logger.error(f"거래 여부 확인 오류: {e}")
return None
finally:
if cursor is not None:
cursor.close()
if sqlcon is not None:
sqlcon.close()
def __init__(self, ticker, user_id) -> None:
self.ticker = ticker
self.trade_yn = None
self.user_id = user_id
self.upbit = self.connect_upbit()
if self.upbit is None:
raise Exception("Upbit connection failed")
self.balance = self.upbit.get_balance()
self.balance = self.balance if self.balance is not None else 0
self.money = self.balance
self.unit = self.upbit.get_balance_t(self.ticker)
self.unit = self.unit if self.unit is not None else 0
self.BASIC_FEES = 0.0005
self.hold_score = 0
self.byu_score = 0
self.sell_score = 0
self.buy_index = 0
self.sell_index = 0
self.max_rate = 0
self.rate = 0
self.old_rate = 0
self.bak_unit = 0
self.bak_balance = 0
self.position = 0
self.sell_price = 0
self.buy_price = 0
self.buy_backup = None
self.sell_backup = None
self.trading_profit = 0
self.trading_profit_rate = 0
self.dq = deque()
self.trade_profit = 0
def unit_buy(self, price, probabilities=None):
"""실제 시장가 매수 실행"""
self.get_trade_yn()
if self.trade_yn != "Y":
logger.warning(f"거래 비활성화: {self.trade_yn}")
return False
self.unit = self.upbit.get_balance_t(self.ticker)
self.unit = self.unit if self.unit is not None else 0
if self.unit != 0:
logger.info("이미 보유 중 - 매수 생략")
return False
self.back_up_balance()
all_balance = self.upbit.get_balance()
logger.info(f"전체 잔액: {all_balance}")
self.old_rate = self.rate
self.position = 1
# 시장가 매수: 금액 기반으로 주문 (전체 잔액의 99.5% 사용)
buy_amount = math.floor(all_balance * 0.995) # 0.5% 수수료 마진 고려
logger.info(f"시장가 매수 금액: {buy_amount:,}원")
if buy_amount < 5000: # 최소 주문 금액 5000원
logger.warning(f"매수 금액이 너무 작습니다: {buy_amount:,}원")
return False
# 시장가 매수 주문 (금액 지정)
result = self.upbit.buy_market_order(ticker=self.ticker, price=buy_amount)
logger.info(f"시장가 매수 결과: {result}")
# 주문 오류 확인
if result and "error" in result:
logger.error(f"매수 주문 오류: {result['error']}")
return False
elif result and "uuid" in result:
self.uuid = result["uuid"]
logger.info(f"매수 주문 성공: {self.uuid}")
# 짧은 대기 후 체결 확인 (시장가는 즉시 체결)
time.sleep(1)
else:
logger.error(f"매수 주문 실패: {result}")
return False
# 매수 후 상태 업데이트
self.unit = self.upbit.get_balance_t(self.ticker)
self.unit = self.unit if self.unit is not None else 0
self.balance = self.upbit.get_balance()
self.balance = self.balance if self.balance is not None else 0
# 시장가 매수 시 실제 체결 가격 확인
try:
avg_buy_price = self.upbit.get_avg_buy_price(self.ticker)
if avg_buy_price and avg_buy_price > 0:
actual_price = avg_buy_price
logger.info(f"실제 체결 가격: {actual_price:,.0f}원")
else:
actual_price = price # 폴백으로 예상 가격 사용
logger.warning("체결 가격 확인 실패, 예상 가격 사용")
except Exception as e:
actual_price = price
logger.warning(f"체결 가격 조회 오류: {str(e)}, 예상 가격 사용")
total = self.unit * actual_price + self.balance
self.rate = (total - self.orgin_money) * 100 / self.orgin_money
logger.info(f"[시장가 매수 완료] 체결가: {actual_price:,.0f}, 수량: {self.unit:.6f}, 잔액: {self.balance:,.0f}, 수익률: {self.rate:.2f}%")
self.buy_price = actual_price
# 거래 내역을 DB에 기록
_record_trade_to_db(
user_id=self.user_id,
ticker=self.ticker,
action="BUY",
price=price,
units=self.unit,
balance=self.balance,
profit_rate=self.rate,
trading_profit=0,
trading_profit_rate=0,
probabilities=probabilities,
)
return True
def unit_sell(self, price, probabilities=None):
"""실제 시장가 매도 실행"""
self.get_trade_yn()
if self.trade_yn != "Y":
logger.warning(f"거래 비활성화: {self.trade_yn}")
return False
self.unit = self.upbit.get_balance_t(self.ticker)
self.unit = self.unit if self.unit is not None else 0
if self.unit == 0:
logger.info("보유 수량이 0 - 매도 생략")
return False
self.back_up_unit()
self.old_rate = self.rate
# 시장가 매도: 보유 수량 전체 매도
result = self.upbit.sell_market_order(ticker=self.ticker, volume=self.unit)
logger.info(f"시장가 매도 결과: {result}")
# 주문 오류 확인
if result and "error" in result:
logger.error(f"매도 주문 오류: {result['error']}")
return False
elif result and "uuid" in result:
logger.info(f"매도 주문 성공: {result['uuid']}")
# 짧은 대기 후 체결 확인 (시장가는 즉시 체결)
time.sleep(1)
else:
logger.error(f"매도 주문 실패: {result}")
return False
# 매도 후 상태 업데이트
self.balance = self.upbit.get_balance()
self.balance = self.balance if self.balance is not None else 0
self.unit = self.upbit.get_balance_t(self.ticker)
self.unit = self.unit if self.unit is not None else 0
# 시장가 매도 시 실제 체결 가격 확인 (거래 내역에서 확인)
try:
# 매도는 시장가로 체결되므로 예상 가격 사용
actual_sell_price = price
logger.info(f"시장가 매도 체결 (예상가: {price:,.0f}원)")
except Exception as e:
actual_sell_price = price
logger.warning(f"매도 가격 설정 오류: {str(e)}")
total = self.unit * actual_sell_price + self.balance
self.rate = (total - self.orgin_money) * 100 / self.orgin_money
self.money = self.balance
self.max_rate = 0
self.position = 0
self.sell_price = actual_sell_price
# 거래 내역을 DB에 기록
_record_trade_to_db(
user_id=self.user_id,
ticker=self.ticker,
action="SELL",
price=price,
units=0, # 매도 후 보유량
balance=self.balance,
profit_rate=self.rate,
trading_profit=self.trading_profit,
trading_profit_rate=self.trading_profit_rate,
probabilities=probabilities,
)
return True
def unit_hold(self, price):
"""보유 상태 확인"""
self.unit = self.upbit.get_balance_t(self.ticker)
self.unit = self.unit if self.unit is not None else 0
self.balance = self.upbit.get_balance()
self.balance = self.balance if self.balance is not None else 0
self.old_rate = self.rate
total = self.unit * price + self.balance
self.rate = (total - self.orgin_money) * 100 / self.orgin_money
logger.info(f"[보유] 가격: {price:,.0f}, 수량: {self.unit:.6f}, 잔액: {self.balance:,.0f}, 수익률: {self.rate:.2f}%")
return True
def get_current_balance(self, price):
"""현재 포트폴리오 가치 반환"""
return self.balance + self.unit * price
def do_action(self, action, price, probabilities=None):
"""액션에 따른 거래 실행"""
if action == 0: # Sell
return self.unit_sell(price, probabilities)
elif action == 1: # Buy
return self.unit_buy(price, probabilities)
else: # Hold
return self.unit_hold(price)
def back_up_balance(self):
self.bak_balance = self.balance
def back_up_unit(self):
self.bak_unit = self.unit
def back_up_position(self):
self.bak_position = self.position
def cancel_all_order(self):
"""모든 주문 취소"""
try:
orders = self.upbit.get_order(self.ticker)
if not orders:
logger.info("취소할 주문이 없습니다")
return
for order in orders:
uuid = order["uuid"]
result = self.upbit.cancel_order(uuid)
if result is None or "error" in result:
error_message = result.get("error", {}).get("message", "Unknown error") if result else "None result"
logger.error(f"주문 취소 실패 {uuid}: {error_message}")
else:
logger.info(f"주문 취소 성공 {uuid}")
except Exception as e:
logger.error(f"주문 취소 중 오류: {str(e)}")