예제 소스
암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM · 파일 59개
main_llm_db_test_ollama.py
"""
Market.get_data()로 DB 과거 캔들을 불러와 main_llm_real.py와 동일한
비전 차트 → 매수/매도/유지 파이프라인으로 모의 매매 백테스트.
- 초기 자금 기본 5,000,000원 (Paper, Upbit 미연동)
- 실거래 계정/주문 없음
- LLM 백엔드: Ollama(로컬) 비전 모델
필수:
- Ollama가 실행 중이어야 함 (기본: http://127.0.0.1:11434)
- 비전 모델을 pull 해두어야 함 (예: llava 계열)
"""
from __future__ import annotations
import argparse
import base64
import json
import logging
import logging.config
import math
import os
import time
import warnings
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
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 Market import Market # noqa: E402
from test_hf_img import ( # noqa: E402
_build_analysis_prompt,
_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
DEFAULT_INITIAL_KRW = 5_000_000
MIN_ORDER_KRW = 5000
DEFAULT_OLLAMA_HOST = os.environ.get("OLLAMA_HOST", "http://127.0.0.1:11434").rstrip("/")
DEFAULT_OLLAMA_MODEL = os.environ.get("OLLAMA_VISION_MODEL", "llava:latest").strip() or "llava:latest"
OLLAMA_TIMEOUT_SEC = float(os.environ.get("OLLAMA_TIMEOUT_SEC", "180"))
OLLAMA_RETRY = int(os.environ.get("OLLAMA_RETRY", "1"))
OLLAMA_RETRY_BACKOFF_SEC = float(os.environ.get("OLLAMA_RETRY_BACKOFF_SEC", "1.0"))
OLLAMA_LOG_REPLY_HEAD = int(os.environ.get("OLLAMA_LOG_REPLY_HEAD", "240"))
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
def _ollama_vision_chat(
*,
host: str,
model: str,
image_bytes: bytes,
user_text: str,
timeout_sec: float,
) -> str:
"""
Ollama Chat API 사용 (images: base64).
문서 기준: POST /api/chat
"""
url = f"{host}/api/chat"
img_b64 = base64.b64encode(image_bytes).decode("ascii")
payload = {
"model": model,
"stream": False,
"messages": [
{
"role": "user",
"content": user_text,
"images": [img_b64],
}
],
}
req = Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urlopen(req, timeout=timeout_sec) as resp:
raw = resp.read().decode("utf-8", errors="replace")
except HTTPError as e:
body = ""
try:
body = e.read().decode("utf-8", errors="replace")
except Exception:
body = ""
raise RuntimeError(f"Ollama HTTPError {e.code}: {body or str(e)}") from e
except URLError as e:
raise RuntimeError(f"Ollama URLError: {e}") from e
data = json.loads(raw)
msg = (data.get("message") or {}).get("content")
if not isinstance(msg, str):
raise RuntimeError(f"Ollama 응답 파싱 실패: keys={list(data.keys())}")
return msg
def _ollama_post_json(*, url: str, payload: dict, timeout_sec: float) -> dict:
req = Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
with urlopen(req, timeout=timeout_sec) as resp:
raw = resp.read().decode("utf-8", errors="replace")
return json.loads(raw)
def _ollama_get_json(*, url: str, timeout_sec: float) -> dict:
req = Request(url, headers={"Accept": "application/json"}, method="GET")
with urlopen(req, timeout=timeout_sec) as resp:
raw = resp.read().decode("utf-8", errors="replace")
return json.loads(raw)
def _log_ollama_server_info(*, host: str, model: str) -> None:
"""
Ollama가 MLX를 쓰는지의 여부는 클라이언트 API만으로 확정하기 어렵습니다.
대신, 서버/모델 메타(arch/quant/context 등)를 뽑아 로그로 남겨 진단에 도움을 줍니다.
"""
# version
try:
v = _ollama_get_json(url=f"{host}/api/version", timeout_sec=min(5.0, OLLAMA_TIMEOUT_SEC))
logger.info("Ollama version: %s", v.get("version", v))
except Exception as e:
logger.warning("Ollama version 조회 실패: %s", e)
# show model meta
try:
s = _ollama_post_json(
url=f"{host}/api/show",
payload={"name": model},
timeout_sec=min(15.0, OLLAMA_TIMEOUT_SEC),
)
details = s.get("details") or {}
model_info = s.get("model_info") or {}
# 너무 큰 딕셔너리는 핵심만 찍기
slim = {
"details": {
k: details.get(k)
for k in ("family", "families", "format", "parameter_size", "quantization_level")
if k in details
},
"model_info": {
k: model_info.get(k)
for k in (
"general.architecture",
"general.name",
"llama.context_length",
"llama.embedding_length",
)
if k in model_info
},
}
logger.info("Ollama model meta: %s", slim)
logger.info(
"MLX 체크: Ollama는 보통 llama.cpp/Metal 경로를 사용합니다. "
"API만으로 'MLX 사용'을 확정하기는 어렵고, 서버 로그/빌드/모델 형태로 확인이 필요합니다."
)
except Exception as e:
logger.warning("Ollama model meta 조회 실패: %s", e)
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_buy_price = 0.0
self.last_buy_unit = 0.0
self.last_buy_amount_krw = 0.0
self.last_sell_price = 0.0
self.last_sell_profit_rate = 0.0
self.last_sell_profit_krw = 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.last_buy_price = price
self.last_buy_unit = float(coins)
self.last_buy_amount_krw = float(buy_amount)
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.last_sell_price = price
self.last_sell_profit_rate = float(self.trading_profit_rate)
self.last_sell_profit_krw = float(proceeds - cost)
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,
"last_buy_price": self.last_buy_price,
"last_buy_unit": self.last_buy_unit,
"last_buy_amount_krw": self.last_buy_amount_krw,
"last_sell_price": self.last_sell_price,
"last_sell_profit_rate": self.last_sell_profit_rate,
"last_sell_profit_krw": self.last_sell_profit_krw,
}
def run_backtest(
*,
initial_krw: float,
bar_step: int,
min_bars: int,
min_trade_bars: int,
max_llm_calls: int | None,
dry_run: bool,
ollama_host: str,
ollama_model: str,
) -> None:
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
logger.info(
f"백테스트 시작: 행 수={n}, 초기자금={initial_krw:,.0f}원, "
f"bar_step={bar_step}, min_bars={min_bars}, 백엔드=Ollama, 모델={ollama_model}, host={ollama_host}"
)
_log_ollama_server_info(host=ollama_host, model=ollama_model)
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"])
t0 = time.perf_counter()
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,
)
t_render = time.perf_counter() - t0
t1 = time.perf_counter()
img_bytes = _png_bytes_to_jpg_bytes(base_png)
t_imgconv = time.perf_counter() - t1
prompt = _build_analysis_prompt(TICKER)
prompt_len = len(prompt or "")
if dry_run:
opinion = "유지"
reply = ""
t_llm = 0.0
t_parse = 0.0
else:
reply = ""
last_err: Exception | None = None
t_llm = 0.0
for attempt in range(max(1, OLLAMA_RETRY)):
try:
t2 = time.perf_counter()
reply = _ollama_vision_chat(
host=ollama_host,
model=ollama_model,
image_bytes=img_bytes,
user_text=prompt,
timeout_sec=OLLAMA_TIMEOUT_SEC,
)
t_llm = time.perf_counter() - t2
last_err = None
break
except Exception as e:
last_err = e
logger.warning(
"Ollama 호출 실패 (attempt %d/%d): %s",
attempt + 1,
max(1, OLLAMA_RETRY),
e,
)
if attempt + 1 < max(1, OLLAMA_RETRY):
time.sleep(max(0.0, OLLAMA_RETRY_BACKOFF_SEC) * (attempt + 1))
if last_err is not None:
logger.error("Ollama 최종 실패로 bar 스킵: %s", last_err)
i += bar_step
continue
llm_calls += 1
t3 = time.perf_counter()
opinion, _raw = _parse_opinion(reply)
t_parse = time.perf_counter() - t3
if opinion is None:
logger.warning(
"응답에서 매수·매도·유지를 찾지 못해 유지로 처리합니다. reply_head=%r",
(reply[:OLLAMA_LOG_REPLY_HEAD] 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)
reply_len = len(reply or "")
img_kb = len(img_bytes) / 1024.0
t_total = time.perf_counter() - t0
logger.info(
"bar=%d/%d 가격=%.0f LLM=%s 신호=%s 포트=%.0f원(%+.2f%%) "
"timing[s]=render=%.3f img=%.3f llm=%.3f parse=%.3f total=%.3f "
"payload=img=%.1fKB prompt_len=%d reply_len=%d",
i,
n - 1,
current_price,
(opinion or "N/A"),
signal_str,
st["portfolio_value"],
st["profit_rate"] * 100,
t_render,
t_imgconv,
t_llm,
t_parse,
t_total,
img_kb,
prompt_len,
reply_len,
)
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 데이터, Ollama)")
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="LLM 호출 최대 횟수(시간 제한용, 미지정이면 데이터 끝까지)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="LLM 호출 없이 차트만 생성·유지 신호로 파이프라인 점검",
)
parser.add_argument(
"--ollama-host",
default=DEFAULT_OLLAMA_HOST,
help="Ollama 호스트 (기본: OLLAMA_HOST 또는 http://127.0.0.1:11434)",
)
parser.add_argument(
"--ollama-model",
default=DEFAULT_OLLAMA_MODEL,
help="Ollama 비전 모델 (기본: OLLAMA_VISION_MODEL 또는 llava:latest)",
)
args = parser.parse_args()
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,
ollama_host=str(args.ollama_host).rstrip("/"),
ollama_model=str(args.ollama_model).strip(),
)
if __name__ == "__main__":
main()