예제 소스
암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM · 파일 59개
png_status_vision_viewer.py
"""
viewer.VisionBacktestViewer 확장: 캡쳐된 차트 PNG → ``PngToStatusPolicy`` →
매수/매도/유지 액션 결정. LLM/MA 는 참고만 표시.
기존 ``RlVisionBacktestViewer`` 와 같은 UI/포트폴리오 흐름을 유지하되,
정책 입력을 손수 만든 17차원 상태 벡터가 아니라 **차트 이미지 자체**로 바꿔
시각 정보를 직접 학습합니다. (REINFORCE + 선택적 teacher KL/LLM 정렬)
"""
from __future__ import annotations
import logging
import math
import time
from pathlib import Path
from typing import Optional
import numpy as np
import torch
import torch.nn.functional as F
import torch.optim as optim
from torch.distributions import Categorical
from test_hf_img import _build_analysis_prompt, _parse_opinion, _png_bytes_to_jpg_bytes
import main_llm_db_test_ollama as ollama_bt
from PngToStatus import PngToStatusPolicy, save_png_status_policy
from viewer import ACCENT, BG, FG, VisionBacktestViewer, extract_llm_reason, opinion_to_signal_label
from lib.rl_chart_trainer import action_to_opinion
from lib.rl_policy_teacher import (
kl_student_from_teacher_logits,
opinion_to_action_index,
teacher_logits_from_chart,
write_rl_training_meta,
)
logger = logging.getLogger(__name__)
def _act_idx_to_kor(i: int) -> str:
if int(i) == 0:
return "매도"
if int(i) == 1:
return "매수"
return "유지"
class PngStatusVisionBacktestViewer(VisionBacktestViewer):
"""차트 PNG → ``PngToStatusPolicy`` 로 액션을 선택하는 RL 뷰어."""
def __init__(
self,
*,
png_policy: PngToStatusPolicy,
online_train: bool = False,
online_sample_action: bool = True,
online_max_episodes: int | None = None,
online_save_every_ep: int = 1,
online_clear_text_each_ep: bool = False,
checkpoint_path: str | None = None,
train_lr: float = 1e-3,
train_weight_decay: float = 1e-5,
reward_win_weight: float = 2.0,
reward_profit_scale: float = 5.0,
initial_krw: float | None = None,
min_trade_bars: int | None = None,
teacher_kl_coef: float = 0.0,
teacher_temperature: float = 1.0,
llm_align_weight: float = 0.0,
vision_policy_doc: str | None = None,
policy_doc_path_for_meta: str | None = None,
save_train_meta_with_checkpoint: bool = False,
eval_action_labels: np.ndarray | None = None,
eval_action_csv_path: str | None = None,
**kwargs,
) -> None:
self._png_policy = png_policy
self._online_train = bool(online_train)
self._online_sample_action = bool(online_sample_action)
self._online_max_episodes = (
online_max_episodes if online_max_episodes is None else int(online_max_episodes)
)
self._online_save_every_ep = max(1, int(online_save_every_ep))
self._online_clear_text_each_ep = bool(online_clear_text_each_ep)
self._checkpoint_path = checkpoint_path
self._reward_win_weight = float(reward_win_weight)
self._reward_profit_scale = float(reward_profit_scale)
self._teacher_kl_coef = float(teacher_kl_coef)
self._teacher_temperature = float(teacher_temperature)
self._llm_align_weight = float(llm_align_weight)
self._vision_policy_doc = (vision_policy_doc or "").strip()
self._policy_doc_path_for_meta = policy_doc_path_for_meta
self._save_train_meta_with_checkpoint = bool(save_train_meta_with_checkpoint)
self._eval_action_labels: np.ndarray | None = None
self._eval_action_csv_path = (eval_action_csv_path or "").strip() or None
if eval_action_labels is not None:
self._eval_action_labels = np.asarray(eval_action_labels, dtype=np.int64).reshape(-1)
self._eval_hits_argmax = 0
self._eval_hits_exec = 0
self._eval_steps = 0
self._lbl_action_eval: Optional[object] = None
self._train_opt: Optional[optim.Optimizer] = None
if self._online_train:
self._png_policy.train()
self._train_opt = optim.Adam(
self._png_policy.parameters(),
lr=float(train_lr),
weight_decay=float(train_weight_decay),
)
else:
self._png_policy.eval()
self._episode = 1
self._traj_log_probs: list[torch.Tensor] = []
self._traj_teacher_kl: list[torch.Tensor] = []
self._traj_llm_ce: list[torch.Tensor] = []
self._initial_krw = float(initial_krw) if initial_krw is not None else None
self._min_trade_bars = int(min_trade_bars) if min_trade_bars is not None else None
self._reset_to_idx: int | None = None
super().__init__(**kwargs)
self._reset_to_idx = int(self.idx)
if self._eval_action_labels is not None and len(self._eval_action_labels) != self.n:
raise ValueError(
f"eval_action_labels 길이({len(self._eval_action_labels)}) != df 행({self.n})"
)
if self._eval_action_labels is not None:
from tkinter import Frame, Label
ch = self.winfo_children()
if len(ch) >= 2:
# Frame 생성자에는 padx/pady 튜플을 넣지 않는다(macOS Tk 에서 TclError).
row = Frame(self, bg=BG)
row.pack(fill="x", before=ch[1], padx=8, pady=(0, 2))
src = self._eval_action_csv_path or "액션CSV"
Label(row, text=f"{src} vs 정책", bg=BG, fg=FG, font=("", 10, "bold")).pack(
side="left", padx=(0, 8)
)
self._lbl_action_eval = Label(
row,
text="누적: argmax — | 체결 — (스텝 0)",
bg=BG,
fg="#0a5f38",
font=("", 10, "bold"),
anchor="w",
)
self._lbl_action_eval.pack(side="left", fill="x", expand=True)
def _reset_action_eval_stats(self) -> None:
self._eval_hits_argmax = 0
self._eval_hits_exec = 0
self._eval_steps = 0
if self._lbl_action_eval is not None:
self._lbl_action_eval.configure(text="누적: argmax — | 체결 — (스텝 0)")
# ------------------------------------------------------------------
# 정책 입력 헬퍼
# ------------------------------------------------------------------
def _build_side_features(self, price: float) -> torch.Tensor:
port = self.portfolio
in_pos = port.unit > 1e-12
bp = float(getattr(port, "buy_price", 0.0))
unreal = (price - bp) / max(bp, 1e-9) if in_pos and bp > 0 else 0.0
pv = port.balance + port.unit * float(price)
cash_ratio = port.balance / max(pv, 1e-9)
return self._png_policy.make_side_tensor(
in_position=in_pos,
unrealized_pnl=float(unreal),
cash_ratio=float(cash_ratio),
)
def _policy_logits_from_png(self, png_bytes: bytes, price: float) -> torch.Tensor:
x = self._png_policy.png_bytes_to_tensor(png_bytes)
side = self._build_side_features(price)
return self._png_policy(x, side).squeeze(0)
# ------------------------------------------------------------------
# 메인 step (worker thread)
# ------------------------------------------------------------------
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
rl_opinion: str | None = None
reason = ""
price = 0.0
t_llm = 0.0
bar = self.idx
trade_executed = False
teacher_y: int | None = None
pred_argmax = 2
act = 2
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
# LLM 은 참고용 (체결은 PngToStatus 가 결정)
if self.dry_run:
reply = "참고의견: 유지\n근거: dry-run 모드입니다."
llm_opinion = "유지"
reason = extract_llm_reason(reply)
else:
img_bytes = _png_bytes_to_jpg_bytes(base_png)
prompt = _build_analysis_prompt(
ollama_bt.TICKER,
policy_reference_doc=self._vision_policy_doc or None,
)
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 = "유지"
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 = "유지"
# ── 핵심: 캡쳐된 PNG 를 PngToStatusPolicy 에 그대로 주입 ──
logits = self._policy_logits_from_png(png_bytes, price)
pred_argmax = int(torch.argmax(logits, dim=-1).item())
if self._online_train and self._online_sample_action:
dist = Categorical(logits=logits)
a = dist.sample()
self._traj_log_probs.append(dist.log_prob(a))
if self._teacher_kl_coef > 0:
t_np = teacher_logits_from_chart(
self.df, bar, self.portfolio, price
)
self._traj_teacher_kl.append(
kl_student_from_teacher_logits(
logits, t_np, temperature=self._teacher_temperature
)
)
if self._llm_align_weight > 0 and llm_opinion:
log_p = F.log_softmax(logits, dim=-1)
ai = opinion_to_action_index(llm_opinion)
self._traj_llm_ce.append(-log_p[ai])
act = int(a.item())
else:
act = pred_argmax
rl_opinion = action_to_opinion(act)
opinion = rl_opinion
if self._eval_action_labels is not None and bar < len(self._eval_action_labels):
teacher_y = int(self._eval_action_labels[bar])
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("png_status step 실패 bar=%s", bar)
elapsed = 0.0
self.after(
0,
lambda b=bar,
pb=png_bytes,
lo=llm_opinion,
op=opinion,
rlo=rl_opinion,
mao=ma_opinion,
mco=ma_cross_opinion,
m60=ma60_down,
rs=reason,
rp=reply,
pr=price,
er=err,
tllm=t_llm,
el=elapsed,
te=trade_executed,
tea=teacher_y,
pa=pred_argmax,
exa=(act if err is None else None): self._apply_step_ui_pts(
bar=b,
png_bytes=pb,
llm_opinion=lo,
opinion=op,
rl_opinion=rlo,
ma_opinion=mao,
ma_cross_opinion=mco,
ma60_down=m60,
reason=rs,
reply=rp,
price=pr,
err=er,
t_llm=tllm,
elapsed=el,
trade_executed=te,
eval_teacher=tea,
eval_argmax=pa,
eval_exec_act=exa,
),
)
# ------------------------------------------------------------------
# UI 적용 (UI thread)
# ------------------------------------------------------------------
def _apply_step_ui_pts(
self,
*,
bar: int,
png_bytes: bytes | None,
llm_opinion: str | None,
opinion: str | None,
rl_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,
eval_teacher: int | None = None,
eval_argmax: int | None = None,
eval_exec_act: int | None = None,
) -> 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 "아니오"
rl_label = opinion_to_signal_label(rl_opinion) if rl_opinion else label
self.lbl_judgment.configure(
text=(
f"{label} (체결=PNG:{rl_label}, MA20돌파:{cross_label}, "
f"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} (체결=PNG, MA참고:{ma_label}, MA20돌파:{cross_label}, "
f"MA60↓:{ma60_txt}, LLM참고:{llm_label}) "
f"llm={t_llm:.2f}s total={elapsed:.2f}s ---\n"
f"{reply.strip()}\n"
)
if (
eval_teacher is not None
and eval_argmax is not None
and eval_exec_act is not None
):
t_lab = _act_idx_to_kor(eval_teacher)
a_lab = _act_idx_to_kor(eval_argmax)
e_lab = _act_idx_to_kor(eval_exec_act)
ma_a = "✓" if int(eval_teacher) == int(eval_argmax) else "✗"
ma_e = "✓" if int(eval_teacher) == int(eval_exec_act) else "✗"
hist_line += (
f" [액션CSV] teacher={t_lab} | 정책argmax={a_lab}{ma_a} | "
f"체결(샘플)={e_lab}{ma_e}\n"
)
self._eval_steps += 1
if int(eval_teacher) == int(eval_argmax):
self._eval_hits_argmax += 1
if int(eval_teacher) == int(eval_exec_act):
self._eval_hits_exec += 1
if self._lbl_action_eval is not None and self._eval_steps > 0:
pa = 100.0 * self._eval_hits_argmax / self._eval_steps
pe = 100.0 * self._eval_hits_exec / self._eval_steps
self._lbl_action_eval.configure(
text=(
f"누적 일치: argmax {self._eval_hits_argmax}/{self._eval_steps} "
f"({pa:.1f}%) | 체결(샘플) {self._eval_hits_exec}/{self._eval_steps} ({pe:.1f}%)"
)
)
hist_line += "\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:
if self._online_train:
self._finish_episode_and_reset(last_price=price)
else:
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}%) · 매도승률 {st['win_rate']*100:.1f}% · "
f"llm={t_llm:.2f}s{tag}"
)
finally:
self._set_busy(False)
if self.auto_mode.get() and err is None:
self._schedule_auto_next()
# ------------------------------------------------------------------
# 에피소드 종료 → REINFORCE 업데이트
# ------------------------------------------------------------------
def _finish_episode_and_reset(self, *, last_price: float) -> None:
st = self.portfolio.get_status(float(last_price))
wr = float(st["win_rate"])
pr = float(st["profit_rate"])
reward = self._reward_win_weight * wr + math.tanh(pr * self._reward_profit_scale)
try:
self.set_training_reward(float(reward))
self._refresh_portfolio_summary(float(last_price))
except Exception:
pass
if self._train_opt is not None and self._traj_log_probs:
self._train_opt.zero_grad()
lp = torch.stack(self._traj_log_probs).sum()
loss = -lp * reward
if self._traj_teacher_kl:
loss = loss + self._teacher_kl_coef * torch.stack(self._traj_teacher_kl).mean()
if self._traj_llm_ce:
loss = loss + self._llm_align_weight * torch.stack(self._traj_llm_ce).mean()
loss.backward()
torch.nn.utils.clip_grad_norm_(self._png_policy.parameters(), max_norm=5.0)
self._train_opt.step()
self.txt_history.insert(
"end",
(
f"=== EPISODE {self._episode} END | reward={reward:.4f} "
f"win_rate={wr*100:.1f}% pnl={pr*100:+.2f}% trades={st['total_trades']} ===\n\n"
),
)
self.txt_history.see("end")
if self._checkpoint_path and (self._episode % self._online_save_every_ep == 0):
try:
save_png_status_policy(self._png_policy, self._checkpoint_path)
if self._save_train_meta_with_checkpoint and self._policy_doc_path_for_meta:
write_rl_training_meta(
self._checkpoint_path,
policy_doc_path=str(self._policy_doc_path_for_meta),
extra={
"online_train": True,
"policy": "PngToStatusPolicy",
"episode": int(self._episode),
"teacher_kl_coef": self._teacher_kl_coef,
"teacher_temperature": self._teacher_temperature,
"llm_align_weight": self._llm_align_weight,
},
)
except Exception as e:
logger.warning("PngToStatus 체크포인트 저장 실패: %s", e)
if self._online_max_episodes is not None and self._episode >= self._online_max_episodes:
self.auto_mode.set(False)
self._cancel_auto_timer()
self.btn_stop.configure(state="disabled")
self.btn_next.configure(state="disabled")
self._set_status(f"온라인 학습 종료: episodes={self._episode}")
self.idx = self.n
self._traj_log_probs.clear()
self._traj_teacher_kl.clear()
self._traj_llm_ce.clear()
return
self._episode += 1
self._traj_log_probs.clear()
self._traj_teacher_kl.clear()
self._traj_llm_ce.clear()
self._prev_close = None
self._prev_ma20 = None
if self._online_clear_text_each_ep:
self.txt_profit_hist.delete("1.0", "end")
self.txt_history.delete("1.0", "end")
init_krw = self._initial_krw if self._initial_krw is not None else float(
getattr(self.portfolio, "orgin_money", 5_000_000.0)
)
new_port = ollama_bt.PaperLlmPortfolio(init_krw)
if self._min_trade_bars is not None:
new_port.min_trade_bars = max(0, int(self._min_trade_bars))
self.portfolio = new_port
self.idx = max(50, int(self._reset_to_idx or 100))
self._set_status(f"온라인 학습: EP {self._episode} 시작 (자동 진행 유지)")