예제 소스
암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM · 파일 59개
viewer.py
"""
비전 LLM 백테스트 Tk 뷰어 (차트 + 오른쪽 판단/근거/이력·수익히스토리 + 하단 포트폴리오 요약).
다른 스크립트에서는 `VisionBacktestViewer` 와 `make_*_vision_callable` 만 가져다 쓰면 됩니다.
from viewer import VisionBacktestViewer, make_local_vision_callable, resolve_model_path, download_hf_snapshot
로깅: 호출 측에서 `config/logging.json` dictConfig 후 import 하는 것을 권장합니다.
"""
from __future__ import annotations
import io
import logging
import os
import re
import threading
import time
from collections.abc import Callable
from pathlib import Path
from tkinter import (
BooleanVar,
Button,
Checkbutton,
Frame,
Label,
PanedWindow,
PhotoImage,
Scrollbar,
Text,
Tk,
messagebox,
)
from typing import Literal
import numpy as np
from PIL import Image, ImageTk
from ChartImageGenerator import ChartImageConfig, ChartImageGenerator
from test_hf_img import (
_build_analysis_prompt,
_parse_opinion,
_png_bytes_to_jpg_bytes,
)
import main_llm_db_test_ollama as ollama_bt
logger = logging.getLogger(__name__)
BG = "#ffffff"
FG = "#111111"
ACCENT = "#1a5fb4"
FOOTER_BG = "#f4f4f4"
# Right panel min width: judgment label uses wraplength; text columns less clipped.
RIGHT_PANEL_WIDTH = 600
DEFAULT_MODELS_DIR = "models"
DEFAULT_TORCH_MODEL_DIR = "Qwen2.5-VL-3B-Instruct"
DEFAULT_MLX_MODEL_DIR = "Qwen2-VL-2B-Instruct-4bit"
_ENV_LOCAL_RETRY = int(os.environ.get("LOCAL_LLM_RETRY", "2"))
_ENV_LOCAL_BACKOFF = float(os.environ.get("LOCAL_LLM_RETRY_BACKOFF_SEC", "1.0"))
def extract_llm_reason(full_text: str) -> str:
if not full_text:
return ""
m = re.search(r"근거\s*[:：]\s*(.*)", full_text, re.DOTALL)
if m:
return m.group(1).strip()
lines = full_text.strip().splitlines()
if len(lines) <= 1:
return full_text.strip()
if re.match(r"^\s*의견\s*[:：]", lines[0]):
return "\n".join(lines[1:]).strip()
return full_text.strip()
def opinion_to_signal_label(opinion: str | None) -> str:
if not opinion:
return "파악 불가 (유지로 처리)"
sig, _ = ollama_bt._opinion_to_signal(opinion)
return ["매도", "매수", "유지"][sig]
def resolve_model_path(models_dir: Path, model: str) -> Path:
p = Path(model).expanduser()
if p.is_absolute():
return p.resolve()
return (models_dir / model).resolve()
def download_hf_snapshot(repo_id: str, models_dir: Path) -> Path:
try:
from huggingface_hub import snapshot_download
except ImportError as e:
raise SystemExit("huggingface_hub 가 필요합니다: pip install huggingface_hub") from e
models_dir.mkdir(parents=True, exist_ok=True)
folder_name = repo_id.split("/")[-1]
target = (models_dir / folder_name).resolve()
target.mkdir(parents=True, exist_ok=True)
logger.info("snapshot_download: %s → %s", repo_id, target)
snapshot_download(repo_id, local_dir=str(target))
print(f"[OK] 저장됨: {target}", flush=True)
return target
VisionImageBytes = bytes
VisionPrompt = str
VisionReply = str
VisionCallable = Callable[[VisionImageBytes, VisionPrompt], VisionReply]
def make_ollama_vision_callable(
*,
host: str,
model: str,
timeout_sec: float | None = None,
) -> VisionCallable:
"""단일 호출만 수행. 재시도는 VisionBacktestViewer(vision_retries)에서."""
to = float(timeout_sec if timeout_sec is not None else ollama_bt.OLLAMA_TIMEOUT_SEC)
h = host.rstrip("/")
m = model.strip()
def _call(img_bytes: bytes, prompt: str) -> str:
return ollama_bt._ollama_vision_chat(
host=h,
model=m,
image_bytes=img_bytes,
user_text=prompt,
timeout_sec=to,
)
return _call
def make_local_vision_callable(
backend: Literal["torch", "mlx"],
model_path: str,
*,
max_new_tokens_torch: int = 512,
max_tokens_mlx: int = 512,
) -> VisionCallable:
"""단일 호출만 수행. 재시도는 VisionBacktestViewer(vision_retries)에서."""
mid = model_path
def _read_model_type(path: str) -> str | None:
"""로컬 HF 체크포인트(config.json)에서 model_type 힌트 추출."""
try:
p = Path(path)
cfg = p / "config.json"
if not cfg.exists():
return None
import json
with cfg.open("r", encoding="utf-8") as f:
data = json.load(f)
mt = data.get("model_type")
return str(mt) if mt else None
except Exception:
return None
def _call(img_bytes: bytes, prompt: str) -> str:
if backend == "torch":
# local_vision_llm 은 Qwen2.5-VL 계열 전용 구현이라,
# Gemma 등 다른 아키텍처를 여기로 넣으면 weight mismatch로 실패한다.
mt = _read_model_type(mid)
if mt and mt not in ("qwen2_5_vl", "qwen2_vl"):
raise RuntimeError(
"이 프로젝트의 --backend torch 경로는 Qwen2.5-VL 계열(local_vision_llm)만 지원합니다.\n"
f"- 감지된 model_type: {mt}\n"
"- Gemma4 같은 모델은 --backend mlx (mlx-vlm) 또는 Ollama 백엔드를 사용하세요.\n"
"예: python main_llm_db_test_local_viewer.py --backend mlx --model <models/아래 폴더명>\n"
)
from local_vision_llm import local_vision_chat
return local_vision_chat(
image_bytes=img_bytes,
user_text=prompt,
model_id=mid,
max_new_tokens=max_new_tokens_torch,
)
if backend == "mlx":
from local_vision_mlx import mlx_vision_chat
return mlx_vision_chat(
image_bytes=img_bytes,
user_text=prompt,
model_id=mid,
max_tokens=max_tokens_mlx,
temperature=0.0,
)
raise RuntimeError(f"unknown backend: {backend}")
return _call
class VisionBacktestViewer(Tk):
"""
차트 + 오른쪽 LLM 패널(이력·수익 히스토리 분할) + 하단 포트폴리오 요약.
비전 추론은 `vision_call`(JPEG bytes, 프롬프트) -> 응답 문자열 로 주입합니다.
"""
def __init__(
self,
*,
df,
portfolio: ollama_bt.PaperLlmPortfolio,
vision_call: VisionCallable,
min_bars: int = 100,
bar_step: int = 3,
dry_run: bool = False,
auto_interval_sec: float = 3.0,
chart_max_width: int = 900,
vision_retries: int = 2,
vision_backoff_sec: float = 1.0,
window_title: str = "비전 백테스트 뷰어",
status_backend_label: str = "",
) -> None:
super().__init__()
self.title(window_title)
self.configure(bg=BG)
self.vision_call = vision_call
self.vision_retries = max(1, int(vision_retries))
self.vision_backoff_sec = max(0.0, float(vision_backoff_sec))
self.status_backend_label = (status_backend_label or "").strip()
self.portfolio = portfolio
self.df = ollama_bt._ensure_pb_mfi10(df)
self.n = len(self.df)
self.idx = max(50, min_bars)
self.bar_step = max(1, bar_step)
self.dry_run = dry_run
self.auto_interval_sec = max(0.2, float(auto_interval_sec))
self.chart_max_width = max(400, int(chart_max_width))
self._busy = False
self._after_id: str | None = None
self._photo: PhotoImage | None = None
self.auto_mode = BooleanVar(value=False)
# MA 돌파 규칙 상태(직전 종가/MA20)
self._prev_close: float | None = None
self._prev_ma20: float | None = None
self._build_ui()
self._render_chart_sync()
p0 = float(self.df.iloc[self.idx]["close"])
self._refresh_portfolio_summary(p0)
tag = f" · {self.status_backend_label}" if self.status_backend_label else ""
self._set_status(
f"데이터 {self.n}행{tag} · bar={self.idx} · 초기 {portfolio.orgin_money:,.0f}원"
)
def _build_ui(self) -> None:
top = Frame(self, bg=BG, padx=8, pady=8)
top.pack(fill="x")
self.btn_next = Button(
top,
text="다음 (수동)",
command=self._on_next_manual,
bg=BG,
fg=FG,
highlightbackground=BG,
)
self.btn_next.pack(side="left", padx=(0, 8))
self.chk_auto = Checkbutton(
top,
text="자동 진행",
variable=self.auto_mode,
command=self._on_auto_toggle,
bg=BG,
fg=FG,
highlightbackground=BG,
selectcolor=BG,
activebackground=BG,
activeforeground=FG,
)
self.chk_auto.pack(side="left", padx=(0, 8))
self.btn_stop = Button(
top,
text="정지",
command=self._on_stop,
bg=BG,
fg=FG,
highlightbackground=BG,
state="disabled",
)
self.btn_stop.pack(side="left", padx=(0, 16))
self.lbl_status = Label(top, text="", bg=BG, fg=FG, anchor="w")
self.lbl_status.pack(side="left", fill="x", expand=True)
body = Frame(self, bg=BG)
body.pack(fill="both", expand=True, padx=8, pady=(0, 4))
left = Frame(body, bg=BG, relief="groove", bd=1)
left.pack(side="left", fill="both", expand=True, padx=(0, 8))
Label(left, text="차트 (현재 bar까지)", bg=BG, fg=FG, font=("", 11, "bold")).pack(
anchor="w", padx=4, pady=4
)
self.lbl_chart = Label(left, bg=BG)
self.lbl_chart.pack(fill="both", expand=True, padx=4, pady=4)
right = Frame(body, bg=BG, width=RIGHT_PANEL_WIDTH)
right.pack(side="right", fill="y")
right.pack_propagate(False)
Label(right, text="LLM 판단", bg=BG, fg=FG, font=("", 11, "bold")).pack(anchor="w")
self.lbl_judgment = Label(
right,
text="—",
bg="#f0f4ff",
fg=ACCENT,
font=("", 18, "bold"),
pady=12,
anchor="center",
justify="center",
wraplength=RIGHT_PANEL_WIDTH - 24,
relief="solid",
bd=1,
)
self.lbl_judgment.pack(fill="x", pady=(4, 8))
Label(right, text="판단 이유 (근거)", bg=BG, fg=FG, font=("", 11, "bold")).pack(anchor="w")
reason_frame = Frame(right, bg=BG)
reason_frame.pack(fill="x", pady=(4, 8))
self.txt_reason = Text(
reason_frame,
height=7,
wrap="word",
bg="#fafafa",
fg=FG,
insertbackground=FG,
relief="solid",
bd=1,
font=("", 11),
)
self.txt_reason.pack(side="left", fill="both", expand=True)
sb_r = Scrollbar(reason_frame, command=self.txt_reason.yview)
sb_r.pack(side="right", fill="y")
self.txt_reason.configure(yscrollcommand=sb_r.set)
right_pane = PanedWindow(right, orient="vertical", sashrelief="groove", bg=BG, sashwidth=4)
right_pane.pack(fill="both", expand=True, pady=(4, 0))
hist_block = Frame(right_pane, bg=BG)
Label(hist_block, text="이력 (자동 시 누적·스크롤)", bg=BG, fg=FG, font=("", 11, "bold")).pack(
anchor="w"
)
hist_frame = Frame(hist_block, bg=BG)
hist_frame.pack(fill="both", expand=True, pady=(4, 0))
mono = ("Menlo", 10) if self.tk.call("tk", "windowingsystem") == "aqua" else ("Consolas", 10)
self.txt_history = Text(
hist_frame,
wrap="word",
bg="#ffffff",
fg=FG,
insertbackground=FG,
relief="solid",
bd=1,
font=mono,
)
self.txt_history.pack(side="left", fill="both", expand=True)
sb_h = Scrollbar(hist_frame, command=self.txt_history.yview)
sb_h.pack(side="right", fill="y")
self.txt_history.configure(yscrollcommand=sb_h.set)
right_pane.add(hist_block, minsize=140)
self.chart_gen = ChartImageGenerator(
ticker=ollama_bt.TICKER,
config=ChartImageConfig(show=False),
)
foot = Frame(self, bg=FOOTER_BG, relief="groove", bd=1, padx=10, pady=8)
foot.pack(fill="x", padx=8, pady=(0, 8))
Label(
foot,
text="모의 포트폴리오 · 수익",
bg=FOOTER_BG,
fg=FG,
font=("", 12, "bold"),
anchor="w",
).pack(fill="x")
foot_body = Frame(foot, bg=FOOTER_BG)
foot_body.pack(fill="both", expand=True, pady=(6, 4))
summary = Frame(foot_body, bg=FOOTER_BG)
summary.pack(side="left", fill="x", expand=True)
profit_panel = Frame(foot_body, bg=FOOTER_BG)
profit_panel.pack(side="right", fill="both", expand=False, padx=(12, 0))
Label(
profit_panel,
text="수익 히스토리",
bg=FOOTER_BG,
fg=FG,
font=("", 11, "bold"),
anchor="w",
).pack(anchor="w")
phist = Frame(profit_panel, bg=FOOTER_BG)
phist.pack(fill="both", expand=True, pady=(4, 0))
mono_f = ("Menlo", 9) if self.tk.call("tk", "windowingsystem") == "aqua" else ("Consolas", 9)
self.txt_profit_hist = Text(
phist,
wrap="none",
height=8,
bg="#ffffff",
fg=FG,
insertbackground=FG,
relief="solid",
bd=1,
font=mono_f,
)
self.txt_profit_hist.pack(side="left", fill="both", expand=True)
sb_p = Scrollbar(phist, command=self.txt_profit_hist.yview)
sb_p.pack(side="right", fill="y")
self.txt_profit_hist.configure(yscrollcommand=sb_p.set)
def _stat(parent: Frame, title: str, row: int, col: int) -> Label:
f = Frame(parent, bg=FOOTER_BG)
f.grid(row=row, column=col, sticky="w", padx=(0, 18), pady=3)
Label(f, text=title, bg=FOOTER_BG, fg="#666", font=("", 9)).pack(anchor="w")
lb = Label(f, text="—", bg=FOOTER_BG, fg=FG, font=("", 11, "bold"))
lb.pack(anchor="w")
return lb
self.lbl_pf_initial = _stat(summary, "초기 자금", 0, 0)
self.lbl_pf_cash = _stat(summary, "현금 잔고", 0, 1)
self.lbl_pf_units = _stat(summary, f"보유 ({ollama_bt.TICKER})", 0, 2)
self.lbl_pf_pv = _stat(summary, "평가액 (원)", 0, 3)
self.lbl_pf_pnl = _stat(summary, "누적 수익률", 1, 0)
self.lbl_pf_trades = _stat(summary, "매도 체결", 1, 1)
self.lbl_pf_win = _stat(summary, "매도 승률", 1, 2)
self.lbl_pf_price = _stat(summary, "현재가 (종가)", 1, 3)
self.lbl_pf_buy_price = _stat(summary, "매수 가격", 2, 0)
self.lbl_pf_buy_units = _stat(summary, "매수 수량", 2, 1)
self.lbl_pf_sell_price = _stat(summary, "매도 가격", 2, 2)
self.lbl_pf_sell_pnl = _stat(summary, "매도 수익", 2, 3)
self.lbl_pf_reward = _stat(summary, "학습 리워드(최근)", 3, 0)
self._last_training_reward: float | None = None
def _refresh_portfolio_summary(self, current_price: float) -> None:
st = self.portfolio.get_status(current_price)
self.lbl_pf_initial.configure(text=f"{self.portfolio.orgin_money:,.0f} 원")
self.lbl_pf_cash.configure(text=f"{st['balance']:,.0f} 원")
self.lbl_pf_units.configure(text=f"{st['unit']:.6f}")
self.lbl_pf_pv.configure(text=f"{st['portfolio_value']:,.0f} 원")
pr = st["profit_rate"] * 100.0
self.lbl_pf_pnl.configure(
text=f"{pr:+.2f} %",
fg="#0d6e3f" if pr >= 0 else "#a30f0f",
)
self.lbl_pf_trades.configure(text=f"{st['total_trades']} 회")
self.lbl_pf_win.configure(text=f"{st['win_rate'] * 100:.1f} %")
self.lbl_pf_price.configure(text=f"{current_price:,.0f} 원")
self.lbl_pf_buy_price.configure(
text=f"{float(st.get('last_buy_price', 0.0)):,.0f} 원"
if float(st.get('last_buy_price', 0.0)) > 0
else "—"
)
bu = float(st.get("last_buy_unit", 0.0))
self.lbl_pf_buy_units.configure(text=f"{bu:.6f}" if bu > 0 else "—")
sp = float(st.get("last_sell_price", 0.0))
self.lbl_pf_sell_price.configure(text=f"{sp:,.0f} 원" if sp > 0 else "—")
spr = float(st.get("last_sell_profit_rate", 0.0))
spk = float(st.get("last_sell_profit_krw", 0.0))
if sp > 0:
self.lbl_pf_sell_pnl.configure(
text=f"{spr:+.2f}% ({spk:+,.0f}원)",
fg="#0d6e3f" if spr >= 0 else "#a30f0f",
)
else:
self.lbl_pf_sell_pnl.configure(text="—", fg=FG)
if self._last_training_reward is None:
self.lbl_pf_reward.configure(text="—", fg=FG)
else:
r = float(self._last_training_reward)
self.lbl_pf_reward.configure(
text=f"{r:+.4f}",
fg="#0d6e3f" if r >= 0 else "#a30f0f",
)
def set_training_reward(self, reward: float | None) -> None:
self._last_training_reward = None if reward is None else float(reward)
def _append_profit_history(
self,
*,
bar: int,
price: float,
opinion_label: str,
trade_executed: bool,
) -> None:
st = self.portfolio.get_status(price)
line = (
f"bar={bar:6d} | 종가 {price:>12,.0f} | 평가 {st['portfolio_value']:>14,.0f} | "
f"누적 {st['profit_rate'] * 100:+.3f}% | 신호 {opinion_label}"
)
if trade_executed:
line += " | ★체결"
line += "\n"
self.txt_profit_hist.insert("end", line)
self.txt_profit_hist.see("end")
def _set_status(self, s: str) -> None:
self.lbl_status.configure(text=s)
def _set_busy(self, busy: bool) -> None:
self._busy = busy
state = "disabled" if busy else "normal"
self.btn_next.configure(state=state)
self.chk_auto.configure(state="disabled" if busy else "normal")
def _png_to_tk(self, png_bytes: bytes) -> PhotoImage:
im = Image.open(io.BytesIO(png_bytes)).convert("RGB")
w, h = im.size
mw = self.chart_max_width
if w > mw:
ratio = mw / w
im = im.resize((int(w * ratio), int(h * ratio)), Image.Resampling.LANCZOS)
return ImageTk.PhotoImage(im)
def _render_chart_sync(self) -> None:
df_hist = self.df.iloc[: self.idx + 1].copy()
chart_idx = len(df_hist)
base_png, _parts = self.chart_gen.render_and_slice_by_x(
df_hist,
idx=chart_idx,
max_data=min(ollama_bt.CHART_MAX_DATA, len(df_hist)),
ticker=ollama_bt.TICKER,
n_slices=ollama_bt.CHART_N_SLICES,
overlap_px=ollama_bt.CHART_OVERLAP_PX,
scale=ollama_bt.CHART_SCALE,
)
self._photo = self._png_to_tk(base_png)
self.lbl_chart.configure(image=self._photo)
def _on_next_manual(self) -> None:
if self._busy:
return
if self.auto_mode.get():
messagebox.showinfo("안내", "자동 진행 중입니다. 정지한 뒤 수동으로 진행하세요.")
return
self._start_step()
def _on_auto_toggle(self) -> None:
if self.auto_mode.get():
self.btn_stop.configure(state="normal")
if not self._busy:
self._start_step()
else:
self._cancel_auto_timer()
self.btn_stop.configure(state="disabled")
def _on_stop(self) -> None:
self.auto_mode.set(False)
self._cancel_auto_timer()
self.btn_stop.configure(state="disabled")
def _cancel_auto_timer(self) -> None:
if self._after_id is not None:
try:
self.after_cancel(self._after_id)
except Exception:
pass
self._after_id = None
def _schedule_auto_next(self) -> None:
self._cancel_auto_timer()
if not self.auto_mode.get():
return
ms = int(self.auto_interval_sec * 1000)
self._after_id = self.after(ms, self._auto_tick)
def _auto_tick(self) -> None:
self._after_id = None
if not self.auto_mode.get() or self._busy:
return
if self.idx >= self.n:
self.auto_mode.set(False)
self.btn_stop.configure(state="disabled")
self._set_status("데이터 끝 — 자동 정지")
self.btn_next.configure(state="disabled")
return
self._start_step()
def _start_step(self) -> None:
if self._busy:
return
if self.idx >= self.n:
self._set_status("데이터 끝입니다.")
self.auto_mode.set(False)
self._cancel_auto_timer()
self.btn_next.configure(state="disabled")
return
self._set_busy(True)
self._set_status(f"처리 중… bar={self.idx}")
t = threading.Thread(target=self._worker_step, daemon=True)
t.start()
def _call_vision_with_retries(self, img_bytes: bytes, prompt: str) -> str:
last: Exception | None = None
for attempt in range(self.vision_retries):
try:
return self.vision_call(img_bytes, prompt)
except Exception as e:
last = e
logger.warning("vision_call 실패 (%s/%s): %s", attempt + 1, self.vision_retries, e)
if attempt + 1 < self.vision_retries:
time.sleep(self.vision_backoff_sec * (attempt + 1))
raise last or RuntimeError("vision_call 실패")
def _ma20_cross_signal(self, *, close: float, ma20: float) -> int:
"""
규칙(빨간 MA=MA20 기준, 종가로 판정):
- 직전 종가가 MA20 위(또는 접촉)이고, 이번 종가가 MA20 아래로 마감 => 매도(하향 돌파)
- 직전 종가가 MA20 아래(또는 접촉)이고, 이번 종가가 MA20 위로 마감 => 매수(상향 돌파)
- 그 외 => 유지
"""
prev_close = self._prev_close
prev_ma20 = self._prev_ma20
signal = 2 # HOLD
if prev_close is not None and prev_ma20 is not None:
if prev_close >= prev_ma20 and close < ma20:
signal = 0
elif prev_close <= prev_ma20 and close > ma20:
signal = 1
self._prev_close = close
self._prev_ma20 = ma20
return signal
def _ma60_is_downward_curve(self, df_hist) -> bool:
"""최근 3봉에서 MA60(파란 라인)이 국소 하향이면 True. 이 경우 MA20 상향 돌파(매수)는 체결하지 않음."""
if "ma60" not in getattr(df_hist, "columns", ()) or len(df_hist) < 3:
return False
try:
s = df_hist["ma60"]
a, b, c = float(s.iloc[-1]), float(s.iloc[-2]), float(s.iloc[-3])
except (TypeError, ValueError, KeyError, IndexError):
return False
if a != a or b != b or c != c:
return False
return a < b <= c
def _worker_step(self) -> None:
err: str | None = None
png_bytes: bytes | None = None
reply = ""
llm_opinion: str | None = None
opinion: str | None = None
ma_opinion: str | None = None
ma_cross_opinion: str | None = None
ma60_down = False
reason = ""
price = 0.0
t_llm = 0.0
bar = self.idx
trade_executed = False
try:
df_hist = self.df.iloc[: bar + 1].copy()
chart_idx = len(df_hist)
price = float(df_hist.iloc[-1]["close"])
t0 = time.perf_counter()
base_png, _parts = self.chart_gen.render_and_slice_by_x(
df_hist,
idx=chart_idx,
max_data=min(ollama_bt.CHART_MAX_DATA, len(df_hist)),
ticker=ollama_bt.TICKER,
n_slices=ollama_bt.CHART_N_SLICES,
overlap_px=ollama_bt.CHART_OVERLAP_PX,
scale=ollama_bt.CHART_SCALE,
)
png_bytes = base_png
img_bytes = _png_bytes_to_jpg_bytes(base_png)
prompt = _build_analysis_prompt(ollama_bt.TICKER)
if self.dry_run:
reply = "참고의견: 유지\n근거: dry-run 모드입니다."
llm_opinion = "유지"
reason = extract_llm_reason(reply)
else:
t2 = time.perf_counter()
reply = self._call_vision_with_retries(img_bytes, prompt)
t_llm = time.perf_counter() - t2
llm_opinion, raw = _parse_opinion(reply)
reason = extract_llm_reason(raw or reply)
if llm_opinion is None:
llm_opinion = "유지"
# MA20(빨간 MA) 종가 돌파 → 매수/매도/유지 (체결의 유일한 근거)
if "ma20" in df_hist.columns and len(df_hist) >= 2:
ma20 = float(df_hist.iloc[-1]["ma20"])
ma_signal = self._ma20_cross_signal(close=price, ma20=ma20)
if ma_signal == 0:
ma_cross_opinion = "매도"
elif ma_signal == 1:
ma_cross_opinion = "매수"
else:
ma_cross_opinion = "유지"
else:
ma_cross_opinion = None
ma60_down = self._ma60_is_downward_curve(df_hist)
ma_opinion = ma_cross_opinion
if ma_opinion == "매수" and ma60_down:
ma_opinion = "유지"
# LLM 출력은 참고(표시/근거)만. 실제 체결은 MA20 + MA60 필터만 사용.
opinion = ma_opinion if ma_opinion else "유지"
signal, probabilities = ollama_bt._opinion_to_signal(opinion)
trade_executed = self.portfolio.execute_llm_trade(
signal,
price,
current_bar=bar,
probabilities=probabilities,
)
elapsed = time.perf_counter() - t0
except Exception as e:
err = str(e)
logger.exception("step 실패 bar=%s", bar)
elapsed = 0.0
self.after(
0,
lambda: self._apply_step_ui(
bar=bar,
png_bytes=png_bytes,
llm_opinion=llm_opinion,
opinion=opinion,
ma_opinion=ma_opinion,
ma_cross_opinion=ma_cross_opinion,
ma60_down=ma60_down,
reason=reason,
reply=reply,
price=price,
err=err,
t_llm=t_llm,
elapsed=elapsed,
trade_executed=trade_executed,
),
)
def _apply_step_ui(
self,
*,
bar: int,
png_bytes: bytes | None,
llm_opinion: str | None,
opinion: str | None,
ma_opinion: str | None,
ma_cross_opinion: str | None = None,
ma60_down: bool = False,
reason: str,
reply: str,
price: float,
err: str | None,
t_llm: float,
elapsed: float,
trade_executed: bool = False,
) -> None:
try:
if png_bytes:
self._photo = self._png_to_tk(png_bytes)
self.lbl_chart.configure(image=self._photo)
if err:
self.lbl_judgment.configure(text="오류", fg="#b2182b")
self.txt_reason.delete("1.0", "end")
self.txt_reason.insert("1.0", err)
hist_line = f"--- bar={bar} 가격={price:,.0f} ERROR ---\n{err}\n\n"
else:
label = opinion_to_signal_label(opinion)
llm_label = opinion_to_signal_label(llm_opinion) if llm_opinion else "N/A"
ma_label = opinion_to_signal_label(ma_opinion) if ma_opinion else "N/A"
cross_label = (
opinion_to_signal_label(ma_cross_opinion) if ma_cross_opinion else "N/A"
)
ma60_txt = "예" if ma60_down else "아니오"
self.lbl_judgment.configure(
text=f"{label} (체결=MA, MA20돌파:{cross_label}, MA60↓:{ma60_txt}, LLM참고:{llm_label})",
fg=ACCENT,
)
self.txt_reason.delete("1.0", "end")
self.txt_reason.insert("1.0", reason or "(근거 없음)")
hist_line = (
f"--- bar={bar}/{self.n - 1} 가격={price:,.0f} "
f"판단={label} (체결=MA 최종:{ma_label}, MA20돌파:{cross_label}, MA60↓:{ma60_txt}, "
f"LLM참고:{llm_label}) llm={t_llm:.2f}s total={elapsed:.2f}s ---\n"
f"{reply.strip()}\n\n"
)
self._refresh_portfolio_summary(price)
self._append_profit_history(
bar=bar,
price=price,
opinion_label=label,
trade_executed=trade_executed,
)
self.txt_history.insert("end", hist_line)
self.txt_history.see("end")
if err is None:
nxt = bar + self.bar_step
if nxt >= self.n:
self.idx = self.n
self.btn_next.configure(state="disabled")
self.auto_mode.set(False)
self._cancel_auto_timer()
self.btn_stop.configure(state="disabled")
else:
self.idx = nxt
tag = f" · {self.status_backend_label}" if self.status_backend_label else ""
if err:
self._set_status(f"오류 bar={bar} — {err[:100]}")
else:
st = self.portfolio.get_status(price)
self._set_status(
f"bar={self.idx}/{self.n - 1} 가격={price:,.0f} "
f"판단={opinion_to_signal_label(opinion)} · 평가 {st['portfolio_value']:,.0f}원 "
f"({st['profit_rate'] * 100:+.2f}%) · llm={t_llm:.2f}s{tag}"
)
finally:
self._set_busy(False)
if self.auto_mode.get() and err is None:
self._schedule_auto_next()
__all__ = [
"VisionBacktestViewer",
"VisionCallable",
"make_local_vision_callable",
"make_ollama_vision_callable",
"download_hf_snapshot",
"resolve_model_path",
"extract_llm_reason",
"opinion_to_signal_label",
"DEFAULT_MODELS_DIR",
"DEFAULT_TORCH_MODEL_DIR",
"DEFAULT_MLX_MODEL_DIR",
"BG",
"FG",
"ACCENT",
"FOOTER_BG",
]