게시판차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기예제 소스

예제 소스

차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 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()