예제 소스
Qwen2.5-VL 차트 LoRA: PNG로 매수·매도·유지 학습하기 · 파일 59개
main_llm_real.py
"""
Redis 실시간 데이터 → 차트 이미지 → Hugging Face 비전 LLM(매수/매도/유지) → Upbit 실제 매매.
`test_hf_img.py`와 동일한 HF 호출·파싱 경로를 사용합니다.
XGBoost 학습/예측은 사용하지 않습니다.
"""
import json
import logging
import logging.config
import os
import time
import warnings
import numpy as np
import pymysql
warnings.filterwarnings("ignore")
np.seterr(divide="ignore", invalid="ignore")
with open("config/logging.json", "r") as f:
logging.config.dictConfig(json.load(f))
logger = logging.getLogger(__name__)
from ChartImageGenerator import ChartImageConfig, ChartImageGenerator # noqa: E402
from RealTimeMarket import RealTimeMarket # noqa: E402
from RealTimeXGBoostTrader import RealTimeXGBoostTrader # noqa: E402
from test_hf_img import ( # noqa: E402
DEFAULT_HF_MAX_TOKENS,
DEFAULT_VISION_MODEL,
HF_TOKEN,
_build_analysis_prompt,
_hf_vision_chat,
_parse_opinion,
_png_bytes_to_jpg_bytes,
)
from private.localdb import *
logger.info("=== LLM 비전 + Upbit 실제 거래 ===")
TICKER = "KRW-ETH"
UPDATE_INTERVAL = 180
# 차트·HF (test_hf_img 기본과 맞춤)
CHART_MAX_DATA = 300
CHART_N_SLICES = 3
CHART_OVERLAP_PX = 0
CHART_SCALE = 2
HF_TIMEOUT_SEC = int(os.environ.get("HF_TIMEOUT_SEC", "120"))
HF_MAX_TOKENS = DEFAULT_HF_MAX_TOKENS
HF_VISION_MODEL = os.environ.get("HF_VISION_MODEL", DEFAULT_VISION_MODEL)
def record_trade_to_db(
user_id,
ticker,
action,
price,
units=0,
balance=0,
profit_rate=0,
trading_profit=0,
trading_profit_rate=0,
probabilities=None,
):
"""거래 내역을 로컬 MariaDB에 기록"""
sqlcon = None
cursor = None
try:
sqlcon = pymysql.connect(
host=LOCAL_DB_HOST,
user=LOCAL_DB_USER,
password=LOCAL_DB_PASSWORD,
db=LOCAL_DB_NAME,
connect_timeout=10,
)
cursor = sqlcon.cursor()
prob_json = None
if probabilities is not None:
try:
if hasattr(probabilities, "tolist"):
prob_data = probabilities.tolist()
elif isinstance(probabilities, (list, tuple)):
prob_data = list(probabilities)
else:
prob_data = probabilities
prob_json = json.dumps(prob_data)
except Exception as json_error:
logger.warning(f"probabilities JSON 변환 실패: {str(json_error)}")
prob_json = None
query = """
INSERT INTO trade_history
(user_id, ticker, action, price, units, balance, profit_rate,
trading_profit, trading_profit_rate, probabilities)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
cursor.execute(
query,
(
user_id,
ticker,
action,
price,
units,
balance,
profit_rate,
trading_profit,
trading_profit_rate,
prob_json,
),
)
sqlcon.commit()
logger.info(f"거래 기록 저장 완료: {user_id} - {action} - {price:,.0f}원")
except Exception as e:
logger.error(f"거래 기록 저장 실패: {str(e)}")
if sqlcon:
sqlcon.rollback()
finally:
if cursor:
cursor.close()
if sqlcon:
sqlcon.close()
def _opinion_to_signal(opinion: str | None) -> tuple[int, np.ndarray]:
"""매수/매도/유지 → RealTimeXGBoostTrader 시그널(0 SELL, 1 BUY, 2 HOLD) 및 의사확률."""
if opinion == "매수":
return 1, np.array([0.0, 1.0, 0.0], dtype=np.float64)
if opinion == "매도":
return 0, np.array([1.0, 0.0, 0.0], dtype=np.float64)
if opinion == "유지":
return 2, np.array([0.0, 0.0, 1.0], dtype=np.float64)
return 2, np.array([0.0, 0.0, 1.0], dtype=np.float64)
def _ensure_pb_mfi10(df):
"""Redis DF에 PB·MFI10이 없으면 차트 보조 마커용으로 대략 계산."""
out = df
if "PB" not in out.columns and "upper" in out.columns and "lower" in out.columns:
denom = out["upper"] - out["lower"]
out = out.copy()
pb = (out["close"] - out["lower"]) / denom.replace(0, np.nan)
out["PB"] = pb.replace([np.inf, -np.inf], np.nan).fillna(0.5).clip(0, 1)
if "MFI10" not in out.columns:
out = out.copy() if out is df else out
out["MFI10"] = 50.0
return out
def train_llm_real_trading():
"""HF 비전 신호 기반 실제 거래 루프."""
try:
from private.user import user_id
except ImportError:
user_id = "********"
_local = os.environ.get("HF_VISION_LOCAL", "").strip().lower() in (
"1",
"true",
"yes",
"on",
)
if (not HF_TOKEN or not str(HF_TOKEN).strip()) and not _local:
logger.error(
"HF 토큰이 없습니다. private/hf_key.py(HG_TOKEN) 또는 환경변수 HF_TOKEN을 설정하세요. "
"로컬 비전 모델만 쓰려면 HF_VISION_LOCAL=1 을 설정하세요."
)
return
market = RealTimeMarket(TICKER)
trader = RealTimeXGBoostTrader(TICKER, user_id)
if not trader.real_trading:
logger.error("Upbit 연결이 실패했습니다. 프로그램을 종료합니다.")
return
trader.account.cancel_all_order()
chart_gen = ChartImageGenerator(
ticker=TICKER,
config=ChartImageConfig(show=False),
)
last_update_time = time.time()
df = None
logger.info("HF 비전 + Upbit 실제 거래 시작... (Ctrl+C로 종료)")
logger.info(f"모델: {HF_VISION_MODEL}, 차트 max_data={CHART_MAX_DATA}")
logger.info(f"초기 잔액: {trader.account.balance:,}원")
logger.info(f"초기 보유: {trader.account.unit:.6f} {TICKER}")
try:
while True:
current_time = time.time()
if current_time - last_update_time < UPDATE_INTERVAL:
time.sleep(10)
continue
df = market.get_latest_data()
if df is None or len(df) < 100:
logger.error("데이터 부족 또는 업데이트 실패")
time.sleep(30)
continue
df = _ensure_pb_mfi10(df)
current_idx = len(df) - 1
current_price = float(df.iloc[current_idx]["close"])
base_png, _parts = chart_gen.render_and_slice_by_x(
df,
idx=current_idx,
max_data=min(CHART_MAX_DATA, len(df)),
ticker=TICKER,
n_slices=CHART_N_SLICES,
overlap_px=CHART_OVERLAP_PX,
scale=CHART_SCALE,
)
img_bytes = _png_bytes_to_jpg_bytes(base_png)
prompt = _build_analysis_prompt(TICKER)
try:
reply = _hf_vision_chat(
token=HF_TOKEN,
image_bytes=img_bytes,
user_text=prompt,
model=HF_VISION_MODEL,
timeout_sec=HF_TIMEOUT_SEC,
max_tokens=HF_MAX_TOKENS,
)
except Exception as e:
logger.error(f"HF 비전 API 오류: {e}")
last_update_time = current_time
time.sleep(30)
continue
opinion, _raw = _parse_opinion(reply)
if opinion is None:
logger.warning(
"응답에서 매수·매도·유지를 찾지 못해 유지로 처리합니다. "
"응답 앞 500자: %s",
reply[:500] if reply else "",
)
signal, probabilities = _opinion_to_signal(opinion)
signal_str = ["SELL", "BUY", "HOLD"][signal]
trade_executed = trader.execute_llm_trade(
signal, current_price, probabilities
)
status = trader.get_status(current_price)
logger.info(f"가격: {current_price:,.0f}원, LLM 의견: {opinion or 'N/A'}, 신호: {signal_str}")
logger.info(
f"포트폴리오: {status.get('portfolio_value', 0):,.0f}원 "
f"({status.get('profit_rate', 0) * 100:+.2f}%)"
)
logger.info(
f"잔액: {status.get('balance', 0):,.0f}원, 보유: {status.get('unit', 0):.6f}"
)
if trade_executed:
logger.info("*** LLM 기준 거래가 실행되었습니다! ***")
last_update_time = current_time
except KeyboardInterrupt:
logger.info("\n\n거래 종료...")
trader.account.cancel_all_order()
if df is not None and len(df) > 0:
final_price = float(df.iloc[-1]["close"])
final_status = trader.get_status(final_price)
logger.info("\n=== 최종 결과 ===")
logger.info(f"최종 잔액: {final_status.get('balance', 0):,}원")
logger.info(f"최종 보유: {final_status.get('unit', 0):.6f} {TICKER}")
logger.info(
f"포트폴리오 가치: {final_status.get('portfolio_value', 0):,.0f}원"
)
logger.info(f"수익률: {final_status.get('profit_rate', 0) * 100:+.2f}%")
if trader.trade_history:
trader._save_trade_history_to_file()
logger.info("최종 거래 내역 파일 저장 완료")
if __name__ == "__main__":
train_llm_real_trading()