예제 소스
암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM · 파일 59개
main_llm_db_gemma31_viewer.py
"""
Gemma4-31B (MLX) chart viewer: same chart as VisionBacktestViewer; opinion and paper trades from Gemma only (no MA rules).
python main_llm_db_gemma31_viewer.py
python main_llm_db_gemma31_viewer.py --model gemma-4-31b-it-8bit --min-bars 120
Footer matches viewer.py (mock portfolio, PnL, sell win rate). Requires Apple Silicon, mlx-vlm.
"""
from __future__ import annotations
import argparse
import io
import json
import logging.config
import threading
import time
import warnings
from pathlib import Path
import numpy as np
from PIL import Image, ImageTk
warnings.filterwarnings("ignore")
np.seterr(divide="ignore", invalid="ignore")
_cfg = Path(__file__).resolve().parent / "config" / "logging.json"
if _cfg.exists():
with open(_cfg, "r", encoding="utf-8") as f:
logging.config.dictConfig(json.load(f))
import logging
logger = logging.getLogger(__name__)
from tkinter import (
BooleanVar,
Button,
Checkbutton,
Frame,
Label,
PhotoImage,
Scrollbar,
Text,
Tk,
messagebox,
)
from ChartImageGenerator import ChartImageConfig, ChartImageGenerator
from Market import Market
from run_gemma4_chart_llm_only import _build_gemma_delegated_prompt
from test_hf_img import _parse_opinion, _png_bytes_to_jpg_bytes
import main_llm_db_test_ollama as ollama_bt
from viewer import (
ACCENT,
BG,
DEFAULT_MODELS_DIR,
FG,
FOOTER_BG,
RIGHT_PANEL_WIDTH,
extract_llm_reason,
opinion_to_signal_label,
resolve_model_path,
)
DEFAULT_GEMMA31 = "gemma-4-31b-it-8bit"
_ENV_RETRY = int(__import__("os").environ.get("LOCAL_LLM_RETRY", "2"))
_ENV_BACKOFF = float(__import__("os").environ.get("LOCAL_LLM_RETRY_BACKOFF_SEC", "1.0"))
class GemmaOnlyChartViewer(Tk):
"""Same chart as the main viewer; Gemma4-only signal + PaperLlmPortfolio (like viewer footer)."""
def __init__(
self,
*,
df,
portfolio: ollama_bt.PaperLlmPortfolio,
model_path: Path,
min_bars: int = 100,
bar_step: int = 3,
dry_run: bool = False,
auto_interval_sec: float = 3.0,
chart_max_width: int = 900,
max_tokens_mlx: int = 512,
vision_retries: int = 2,
vision_backoff_sec: float = 1.0,
) -> None:
super().__init__()
self.title("Gemma4 " + "\ucc28\ud2b8" + " (\uc758\uacac + \ubaa8\uc758\ud3ec\ud2b8)")
self.configure(bg=BG)
self.portfolio = portfolio
self.model_path_str = str(model_path.resolve())
self.max_tokens_mlx = max(32, int(max_tokens_mlx))
self.vision_retries = max(1, int(vision_retries))
self.vision_backoff_sec = max(0.0, float(vision_backoff_sec))
self.df = ollama_bt._ensure_pb_mfi10(df)
self.n = len(self.df)
self.idx = max(50, int(min_bars))
self.bar_step = max(1, int(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)
self.chart_gen = ChartImageGenerator(
ticker=ollama_bt.TICKER,
config=ChartImageConfig(show=False),
)
self._build_ui()
self._render_chart_sync()
p0 = float(self.df.iloc[self.idx]["close"])
self._refresh_portfolio_summary(p0)
self._set_status(
"\ub370\uc774\ud130 %d\ud589 \xb7 bar=%d \xb7 %s \xb7 \ucd08\uae30 %s\uc6d0"
% (
self.n,
self.idx,
model_path.name,
f"{self.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="\ub2e4\uc74c (\uc218\ub3d9)",
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="\uc790\ub3d9 \uc9c4\ud589",
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="\uc815\uc9c0",
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="\ucc28\ud2b8 (\ud604\uc7ac bar\uae4c\uc9c0 \xb7 \uae30\uc874\uacfc \ub3d9\uc77c \ub80c\ub354)",
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="Gemma4 \ud310\ub2e8 (\uc804\uc6a9)", bg=BG, fg=FG, font=("", 11, "bold")).pack(
anchor="w"
)
self.lbl_judgment = Label(
right,
text="\u2014\n(Gemma \uc2e0\ud638 \xb7 \ubaa8\uc758 \uccb4\uacb0)",
bg="#f0f4ff",
fg=ACCENT,
font=("", 16, "bold"),
pady=10,
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="\uadfc\uac70 (\ubaa8\ub378 \uc6d0\ubb38)", 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=8,
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)
Label(right, text="\uc774\ub825", bg=BG, fg=FG, font=("", 11, "bold")).pack(anchor="w")
hist_frame = Frame(right, 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)
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="\ubaa8\uc758 \ud3ec\ud2b8\ud3f4\ub9ac\uc624 \xb7 \uc218\uc775",
bg=FOOTER_BG,
fg=FG,
font=("", 12, "bold"),
anchor="w",
).pack(fill="x")
summary = Frame(foot, bg=FOOTER_BG)
summary.pack(fill="x", pady=(6, 4))
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="\u2014", bg=FOOTER_BG, fg=FG, font=("", 11, "bold"))
lb.pack(anchor="w")
return lb
self.lbl_pf_initial = _stat(summary, "\ucd08\uae30 \uc790\uae08", 0, 0)
self.lbl_pf_cash = _stat(summary, "\ud604\uae08 \uc7ac\uace0", 0, 1)
self.lbl_pf_units = _stat(summary, f"\ubcf4\uc720 ({ollama_bt.TICKER})", 0, 2)
self.lbl_pf_pv = _stat(summary, "\ud3c9\uac00\uc561 (\uc6d0)", 0, 3)
self.lbl_pf_pnl = _stat(summary, "\ub204\uc801 \uc218\uc775\ub960", 1, 0)
self.lbl_pf_trades = _stat(summary, "\ub9e4\ub3c4 \uccb4\uacb0", 1, 1)
self.lbl_pf_win = _stat(summary, "\ub9e4\ub3c4 \uc2b9\ub960", 1, 2)
self.lbl_pf_price = _stat(summary, "\ud604\uc7ac\uac00 (\uc885\uac00)", 1, 3)
Label(
foot,
text="\uc218\uc775 \ud788\uc2a4\ud1a0\ub9ac (\uc2a4\ud15d\ub9c8\ub2e4 \ud3c9\uac00\xb7\ub204\uc801\uc218\uc775\ub960)",
bg=FOOTER_BG,
fg=FG,
font=("", 10, "bold"),
).pack(anchor="w", pady=(4, 2))
phist = Frame(foot, bg=FOOTER_BG)
phist.pack(fill="x", expand=False)
mono_f = ("Menlo", 9) if self.tk.call("tk", "windowingsystem") == "aqua" else ("Consolas", 9)
self.txt_profit_hist = Text(
phist,
height=5,
wrap="none",
bg="#ffffff",
fg=FG,
relief="solid",
bd=1,
font=mono_f,
)
self.txt_profit_hist.pack(side="left", fill="x", 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 _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} \uc6d0")
self.lbl_pf_cash.configure(text=f"{st['balance']:,.0f} \uc6d0")
self.lbl_pf_units.configure(text=f"{st['unit']:.6f}")
self.lbl_pf_pv.configure(text=f"{st['portfolio_value']:,.0f} \uc6d0")
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']} \ud68c")
self.lbl_pf_win.configure(text=f"{st['win_rate'] * 100:.1f} %")
self.lbl_pf_price.configure(text=f"{current_price:,.0f} \uc6d0")
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} | \uc885\uac00 {price:>12,.0f} | \ud3c9\uac00 {st['portfolio_value']:>14,.0f} | "
f"\ub204\uc801 {st['profit_rate'] * 100:+.3f}% | \uc2e0\ud638 {opinion_label}"
)
if trade_executed:
line += " | \u2605\uccb4\uacb0"
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 _call_mlx_with_retries(self, img_bytes: bytes, prompt: str) -> str:
from local_vision_mlx import mlx_vision_chat
last: Exception | None = None
for attempt in range(self.vision_retries):
try:
return mlx_vision_chat(
image_bytes=img_bytes,
user_text=prompt,
model_id=self.model_path_str,
max_tokens=self.max_tokens_mlx,
temperature=0.0,
)
except Exception as e:
last = e
logger.warning("mlx_vision_chat failed (%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("mlx_vision_chat failed")
def _on_next_manual(self) -> None:
if self._busy:
return
if self.auto_mode.get():
messagebox.showinfo(
"\uc548\ub0b4",
"\uc790\ub3d9 \uc9c4\ud589 \uc911\uc785\ub2c8\ub2e4. \uc815\uc9c0 \ud6c4 \uc218\ub3d9\uc73c\ub85c \uc9c4\ud589\ud558\uc138\uc694.",
)
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("\ub370\uc774\ud130 \ub05d \u2014 \uc790\ub3d9 \uc815\uc9c0")
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("\ub370\uc774\ud130 \ub05d\uc785\ub2c8\ub2e4.")
self.auto_mode.set(False)
self._cancel_auto_timer()
self.btn_next.configure(state="disabled")
return
self._set_busy(True)
self._set_status("\ucc98\ub9ac \uc911\u2026 bar=%d" % self.idx)
t = threading.Thread(target=self._worker_step, daemon=True)
t.start()
def _worker_step(self) -> None:
err: str | None = None
png_bytes: bytes | None = None
reply = ""
opinion: str | None = None
reason = ""
price = 0.0
t_llm = 0.0
bar = self.idx
elapsed = 0.0
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_gemma_delegated_prompt(ollama_bt.TICKER)
if self.dry_run:
reply = "\uc758\uacac: \uc720\uc9c0\n\uadfc\uac70: dry-run \ubaa8\ub4dc\uc785\ub2c8\ub2e4."
opinion = "\uc720\uc9c0"
reason = extract_llm_reason(reply)
else:
t2 = time.perf_counter()
reply = self._call_mlx_with_retries(img_bytes, prompt)
t_llm = time.perf_counter() - t2
opinion, raw = _parse_opinion(reply)
reason = extract_llm_reason(raw or reply)
if opinion is None:
opinion = "\uc720\uc9c0"
if not self.dry_run:
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 failed bar=%s", bar)
self.after(
0,
lambda b=bar,
pb=png_bytes,
o=opinion,
rs=reason,
rep=reply,
pr=price,
e=err,
tl=t_llm,
el=elapsed,
te=trade_executed: self._apply_step_ui(
bar=b,
png_bytes=pb,
opinion=o,
reason=rs,
reply=rep,
price=pr,
err=e,
t_llm=tl,
elapsed=el,
trade_executed=te,
),
)
def _apply_step_ui(
self,
*,
bar: int,
png_bytes: bytes | None,
opinion: str | None,
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="\uc624\ub958\n" + err[:200], fg="#b2182b")
self.txt_reason.delete("1.0", "end")
self.txt_reason.insert("1.0", err)
hist_line = "--- bar=%d \uac00\uaca9=%.0f ERROR ---\n%s\n\n" % (bar, price, err)
else:
label = opinion_to_signal_label(opinion)
self.lbl_judgment.configure(
text=label + "\n(Gemma4 \xb7 \ubaa8\uc758 \uccb4\uacb0)",
fg=ACCENT,
)
self.txt_reason.delete("1.0", "end")
self.txt_reason.insert("1.0", reason or "(\uadfc\uac70 \uc5c6\uc74c)")
star = " \u2605\uccb4\uacb0" if trade_executed else ""
hist_line = (
"--- bar=%d/%d \uac00\uaca9=%.0f Gemma=%s%s llm=%.2fs total=%.2fs ---\n%s\n\n"
% (bar, self.n - 1, price, label, star, t_llm, elapsed, reply.strip())
)
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
if err:
self._set_status("\uc624\ub958 bar=%d \u2014 %s" % (bar, err[:120]))
else:
st = self.portfolio.get_status(price)
self._set_status(
"bar=%d/%d \uac00\uaca9=%.0f \ud310\ub2e8=%s \xb7 \ud3c9\uac00 %s\uc6d0 (%+.2f%%) \xb7 llm=%.2fs"
% (
self.idx,
self.n - 1,
price,
opinion_to_signal_label(opinion),
f"{st['portfolio_value']:,.0f}",
st["profit_rate"] * 100.0,
t_llm,
)
)
finally:
self._set_busy(False)
if self.auto_mode.get() and err is None:
self._schedule_auto_next()
def main() -> None:
ap = argparse.ArgumentParser(description="Gemma4 chart viewer (Gemma signal + mock portfolio footer)")
ap.add_argument("--models-dir", default=DEFAULT_MODELS_DIR)
ap.add_argument("--model", default=DEFAULT_GEMMA31, help="Folder under models/ or absolute path")
ap.add_argument("--min-bars", type=int, default=100)
ap.add_argument("--bar-step", type=int, default=3)
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--auto-interval-sec", type=float, default=3.0)
ap.add_argument("--chart-max-width", type=int, default=900)
ap.add_argument("--max-tokens-mlx", type=int, default=512)
ap.add_argument("--initial", type=float, default=ollama_bt.DEFAULT_INITIAL_KRW)
ap.add_argument("--min-trade-bars", type=int, default=5)
args = ap.parse_args()
models_dir = Path(args.models_dir).resolve()
model_path = resolve_model_path(models_dir, args.model.strip())
if not model_path.exists():
raise SystemExit("Model path not found: %s" % model_path)
logger.info("Market.get_data() …")
market = Market()
df = market.get_data()
if df is None or len(df) < args.min_bars + args.bar_step:
raise SystemExit("Not enough data from Market.get_data()")
port = ollama_bt.PaperLlmPortfolio(float(args.initial))
port.min_trade_bars = max(0, int(args.min_trade_bars))
app = GemmaOnlyChartViewer(
df=df,
portfolio=port,
model_path=model_path,
min_bars=args.min_bars,
bar_step=args.bar_step,
dry_run=args.dry_run,
auto_interval_sec=args.auto_interval_sec,
chart_max_width=args.chart_max_width,
max_tokens_mlx=args.max_tokens_mlx,
vision_retries=_ENV_RETRY,
vision_backoff_sec=_ENV_BACKOFF,
)
app.mainloop()
if __name__ == "__main__":
main()