예제 소스
Qwen2.5-VL 차트 LoRA: PNG로 매수·매도·유지 학습하기 · 파일 59개
main_llm_db_test.py
"""
Market.get_data()로 DB 과거 캔들을 불러와 main_llm_real.py와 동일한
비전 차트 → 매수/매도/유지 파이프라인으로 모의 매매 백테스트.
- 초기 자금 기본 5,000,000원 (Paper, Upbit 미연동)
- 실거래 계정/주문 없음
- `--local`: local_vision_llm.py 로컬 VLM (HF 토큰 불필요). 또는 환경변수 HF_VISION_LOCAL=1
- 기본(로컬 아님): Hugging Face Router API (test_hf_img._hf_vision_chat)
"""
from __future__ import annotations
import argparse
import json
import logging
import logging.config
import math
import os
import warnings
import numpy as np
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 local_vision_llm import local_vision_chat # noqa: E402
from Market import Market # 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,
)
# main_llm_real.py 와 동일한 기본값
TICKER = "KRW-ETH"
UPDATE_INTERVAL_SEC = 180
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)
HF_LOCAL_MAX_CAP = int(os.environ.get("HF_LOCAL_MAX_NEW_TOKENS_CAP", "4096"))
DEFAULT_INITIAL_KRW = 5_000_000
MIN_ORDER_KRW = 5000
def _opinion_to_signal(opinion: str | None) -> tuple[int, np.ndarray]:
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):
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
class PaperLlmPortfolio:
"""
RealTimeXGBoostTrader.execute_llm_trade 와 동일한 신호 규칙을 모의 적용.
- 매수: 보유 0일 때만, 잔액의 99.5%까지 (Account.unit_buy 와 유사)
- 매도: 보유 있을 때만
- 최소 주문 5,000원
"""
BASIC_FEES = 0.0005
def __init__(self, initial_krw: float):
self.orgin_money = float(initial_krw)
self.balance = float(initial_krw)
self.unit = 0.0
self.rate = 0.0
self.trading_profit_rate = 0.0
self.buy_price = 0.0
self.last_trade_bar = -10**9
self.total_trades = 0
self.profitable_trades = 0
self.min_trade_bars = 5
def _mark_rate(self, price: float) -> None:
total = self.balance + self.unit * price
self.rate = (total - self.orgin_money) * 100.0 / self.orgin_money
def execute_llm_trade(
self,
signal: int,
price: float,
*,
current_bar: int,
probabilities: np.ndarray | None = None,
) -> bool:
if current_bar - self.last_trade_bar < self.min_trade_bars:
return False
price = float(price)
executed = False
if signal == 1 and self.unit == 0:
buy_amount = math.floor(self.balance * 0.995)
if buy_amount < MIN_ORDER_KRW:
return False
coins = (buy_amount / price) * (1.0 - self.BASIC_FEES)
self.balance -= buy_amount
self.unit = coins
self.buy_price = price
self._mark_rate(price)
self.last_trade_bar = current_bar
executed = True
logger.info(
f"[모의 매수] bar={current_bar} 가격={price:,.0f} 금액={buy_amount:,} 수량={self.unit:.6f}"
)
elif signal == 0 and self.unit > 0:
gross = self.unit * price
proceeds = gross * (1.0 - self.BASIC_FEES)
cost = self.unit * self.buy_price
self.trading_profit_rate = (proceeds - cost) / cost * 100.0 if cost > 0 else 0.0
self.balance += proceeds
self.unit = 0.0
self.total_trades += 1
if self.trading_profit_rate > 0:
self.profitable_trades += 1
self._mark_rate(price)
self.last_trade_bar = current_bar
executed = True
logger.info(
f"[모의 매도] bar={current_bar} 가격={price:,.0f} "
f"거래수익률={self.trading_profit_rate:+.2f}% 잔액={self.balance:,.0f}"
)
else:
self._mark_rate(price)
return executed
def get_status(self, current_price: float) -> dict:
current_price = float(current_price)
pv = self.balance + self.unit * current_price
profit_rate = (pv - self.orgin_money) / self.orgin_money
return {
"balance": self.balance,
"unit": self.unit,
"portfolio_value": pv,
"profit_rate": profit_rate,
"rate_pct": self.rate,
"total_trades": self.total_trades,
"win_rate": self.profitable_trades / max(self.total_trades, 1),
"current_price": current_price,
}
def _vision_uses_local(*, use_local_flag: bool) -> bool:
"""CLI --local 또는 환경변수 HF_VISION_LOCAL."""
if use_local_flag:
return True
return os.environ.get("HF_VISION_LOCAL", "").strip().lower() in (
"1",
"true",
"yes",
"on",
)
def run_backtest(
*,
initial_krw: float,
bar_step: int,
min_bars: int,
min_trade_bars: int,
max_llm_calls: int | None,
dry_run: bool,
use_local: bool = False,
local_model: str | None = None,
) -> None:
vision_local = _vision_uses_local(use_local_flag=use_local)
if (
not dry_run
and not vision_local
and (not HF_TOKEN or not str(HF_TOKEN).strip())
):
logger.error(
"HF 토큰이 없습니다. private/hf_key.py 또는 HF_TOKEN 환경변수를 설정하거나 "
"--local / HF_VISION_LOCAL=1(로컬 VLM) 또는 --dry-run 으로 실행하세요."
)
return
logger.info("Market.get_data() 로 과거 데이터 로드 중...")
market = Market()
df = market.get_data()
if df is None or len(df) < min_bars + bar_step:
logger.error("데이터가 부족합니다.")
return
df = _ensure_pb_mfi10(df)
n = len(df)
portfolio = PaperLlmPortfolio(initial_krw)
portfolio.min_trade_bars = min_trade_bars
chart_gen = ChartImageGenerator(
ticker=TICKER,
config=ChartImageConfig(show=False),
)
llm_calls = 0
trades_done = 0
_lm = (local_model or os.environ.get("HF_LOCAL_VISION_MODEL") or "").strip() or HF_VISION_MODEL
_backend = "local_vision_llm" if vision_local else "HF Router"
logger.info(
f"백테스트 시작: 행 수={n}, 초기자금={initial_krw:,.0f}원, "
f"bar_step={bar_step}, min_bars={min_bars}, 백엔드={_backend}, 모델(참고)={_lm}"
)
i = min_bars
while i < n:
if max_llm_calls is not None and llm_calls >= max_llm_calls:
logger.info(f"LLM 호출 상한({max_llm_calls}) 도달 — 중단합니다.")
break
df_hist = df.iloc[: i + 1].copy()
chart_idx = len(df_hist)
current_price = float(df_hist.iloc[-1]["close"])
base_png, _parts = chart_gen.render_and_slice_by_x(
df_hist,
idx=chart_idx,
max_data=min(CHART_MAX_DATA, len(df_hist)),
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)
if dry_run:
opinion = "유지"
reply = ""
else:
try:
if vision_local:
mid = (local_model or "").strip() or HF_VISION_MODEL
reply = local_vision_chat(
image_bytes=img_bytes,
user_text=prompt,
model_id=mid,
max_new_tokens=min(HF_MAX_TOKENS, HF_LOCAL_MAX_CAP),
)
else:
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"비전 LLM 오류: {e}")
i += bar_step
continue
llm_calls += 1
opinion, _raw = _parse_opinion(reply)
if opinion is None:
logger.warning(
"응답에서 매수·매도·유지를 찾지 못해 유지로 처리합니다. "
"응답 앞 300자: %s",
reply[:300] if reply else "",
)
signal, probabilities = _opinion_to_signal(opinion)
signal_str = ["SELL", "BUY", "HOLD"][signal]
trade_executed = portfolio.execute_llm_trade(
signal,
current_price,
current_bar=i,
probabilities=probabilities,
)
if trade_executed:
trades_done += 1
st = portfolio.get_status(current_price)
logger.info(
f"bar={i}/{n-1} 가격={current_price:,.0f} LLM={opinion or 'N/A'} 신호={signal_str} "
f"포트폴리오={st['portfolio_value']:,.0f}원 ({st['profit_rate'] * 100:+.2f}%)"
)
i += bar_step
final_price = float(df.iloc[-1]["close"])
final = portfolio.get_status(final_price)
logger.info("")
logger.info("=== 백테스트 결과 ===")
logger.info(f"초기 자금: {initial_krw:,.0f}원")
logger.info(f"최종 평가액: {final['portfolio_value']:,.0f}원")
logger.info(f"순손익: {final['portfolio_value'] - initial_krw:,.0f}원")
logger.info(f"누적 수익률: {final['profit_rate'] * 100:+.2f}%")
logger.info(f"LLM 호출 수: {llm_calls}")
logger.info(f"체결된 모의 거래 횟수(매수+매도 쌍은 매도 기준): {trades_done}, 매도 건수: {final['total_trades']}")
logger.info(f"최종 현금: {final['balance']:,.0f}원, 최종 보유: {final['unit']:.6f}")
def main() -> None:
parser = argparse.ArgumentParser(description="LLM 비전 모의 백테스트 (Market DB 데이터)")
parser.add_argument("--initial", type=float, default=DEFAULT_INITIAL_KRW, help="초기 자금(원)")
parser.add_argument(
"--bar-step",
type=int,
default=3,
help="몇 개 봉마다 한 번 LLM 호출할지 (실시간 180초 간격에 가깝게 하려면 1분봉 기준 3 등)",
)
parser.add_argument(
"--min-bars",
type=int,
default=100,
help="최소 봉 수(이 인덱스부터 백테스트 시작)",
)
parser.add_argument(
"--min-trade-bars",
type=int,
default=5,
help="모의 매매 최소 간격(봉 수). RealTimeXGBoostTrader 300초에 가깝게(1분봉=5)",
)
parser.add_argument(
"--max-llm-calls",
type=int,
default=None,
help="HF 호출 최대 횟수(비용/시간 제한용, 미지정이면 데이터 끝까지)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="LLM 호출 없이 차트만 생성·유지 신호로 파이프라인 점검",
)
parser.add_argument(
"--local",
action="store_true",
help="local_vision_llm 로컬 VLM 사용 (HF 토큰 불필요). HF_VISION_LOCAL=1 과 동일 효과",
)
parser.add_argument(
"--local-model",
default=None,
help="로컬 모델 허브 ID 또는 디렉터리 (미지정 시 HF_LOCAL_VISION_MODEL 또는 HF_VISION_MODEL)",
)
args = parser.parse_args()
if args.local:
os.environ["HF_VISION_LOCAL"] = "1"
run_backtest(
initial_krw=args.initial,
bar_step=max(1, args.bar_step),
min_bars=max(50, args.min_bars),
min_trade_bars=max(0, args.min_trade_bars),
max_llm_calls=args.max_llm_calls,
dry_run=args.dry_run,
use_local=args.local,
local_model=args.local_model,
)
if __name__ == "__main__":
main()