예제 소스
암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM · 파일 59개
main_llm_db_rl_viewer.py
"""
DB 캔들 → **캡쳐된 차트 PNG** 를 ``PngToStatusPolicy`` 가 보고 매수/매도/유지를
결정 → 모의 포트폴리오로 백테스트 + REINFORCE 온라인 학습.
이전 버전은 손수 만든 17차원 상태 벡터로 학습했지만 LLM 의 액션 정확도가
떨어져, **차트 이미지 자체** 를 입력으로 쓰는 정책으로 교체했습니다.
구버전은 ``main_llm_db_rl_viewer_legacy.py`` 에 백업되어 있습니다.
설정은 argparse 없이 다음 파일만 수정합니다.
- config/llm_rl_backend.py 백엔드·모델 경로
- config/llm_rl_model_download.py HF 스냅샷 다운로드 여부
- config/llm_rl_train.py 학습·뷰어·LABELS_CSV_PATH·PngToStatus 옵션
· 액션 CSV 모방학습 체크포인트 쓰려면 ``USE_PNG_STATUS_ACTION_SUPERVISED = True`` 및
``train_png_policy_from_action_csv.py`` 로 ``PNG_STATUS_ACTION_SUPERVISED_CKPT`` 생성
· 학습 맞춤 확인: ``PNG_STATUS_SHOW_VS_ACTION_CSV_EVAL`` + ``PNG_STATUS_EVAL_ACTION_CSV_PATH``
실행 (프로젝트 루트에서):
python train_png_policy_from_action_csv.py # 선택: ACTION CSV → PNG 정책 학습
python main_llm_db_rl_viewer.py
"""
from __future__ import annotations
import json
import logging.config
import warnings
from pathlib import Path
import numpy as np
import pandas as pd
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__)
print(
"[main_llm_db_rl_viewer] 무거운 의존성(torch, Tk, transformers 등) 로드 중… "
"최초 30~90초는 터미널에 아무것도 안 나올 수 있습니다.",
flush=True,
)
def _ensure_tkinter() -> None:
"""Homebrew python@3.12 등은 기본적으로 _tkinter 가 빠져 있을 수 있음."""
import sys
try:
import tkinter # noqa: F401
except ImportError as e:
mj, mn = sys.version_info.major, sys.version_info.minor
ver = f"{mj}.{mn}"
pybin = f"python{mj}.{mn}"
raise SystemExit(
"Tkinter(_tkinter)를 불러올 수 없습니다. 이 뷰어는 Tk GUI가 필요합니다.\n\n"
"[macOS + Homebrew 예 — Python 버전에 맞출 것]\n"
f" brew install python-tk@{ver}\n"
f" $(brew --prefix python-tk@{ver})/bin/{pybin} -m venv .venv\n"
" source .venv/bin/activate && pip install -r requirements.txt\n\n"
"또는 python.org 공식 macOS 설치본(Tk 포함)으로 새 venv 를 만드세요.\n"
f"(현재 인터프리터: {sys.executable})\n\n"
f"원인: {e}"
) from e
_ensure_tkinter()
from config import llm_rl_backend as be # noqa: E402
from config import llm_rl_model_download as dl # noqa: E402
from config import llm_rl_train as tr # noqa: E402
from lib.rl_data_labels import merge_extreme_labels_from_csv # noqa: E402
from lib.rl_policy_teacher import load_policy_markdown # noqa: E402
from lib.png_status_vision_viewer import PngStatusVisionBacktestViewer # noqa: E402
from Market import Market # noqa: E402
import main_llm_db_test_ollama as ollama_bt # noqa: E402
from PngToStatus import ( # noqa: E402
DEFAULT_INPUT_H,
DEFAULT_INPUT_W,
load_png_status_policy,
)
from viewer import ( # noqa: E402
DEFAULT_MLX_MODEL_DIR,
DEFAULT_TORCH_MODEL_DIR,
download_hf_snapshot,
make_local_vision_callable,
make_ollama_vision_callable,
resolve_model_path,
)
def _maybe_download_hf() -> None:
if not dl.ENABLE_SNAPSHOT:
return
models_dir = Path(be.MODELS_DIR).resolve()
models_dir.mkdir(parents=True, exist_ok=True)
logger.info("HF 스냅샷 다운로드: %s", dl.REPO_ID)
download_hf_snapshot(dl.REPO_ID.strip(), models_dir)
def _build_vision_callable():
if be.BACKEND == "ollama":
vision = make_ollama_vision_callable(
host=str(be.OLLAMA_HOST).rstrip("/"),
model=str(be.OLLAMA_MODEL).strip(),
)
v_retry = max(1, int(ollama_bt.OLLAMA_RETRY))
v_backoff = float(ollama_bt.OLLAMA_RETRY_BACKOFF_SEC)
status_lbl = f"ollama · {be.OLLAMA_MODEL}"
return vision, v_retry, v_backoff, status_lbl
models_dir = Path(be.MODELS_DIR).resolve()
folder = be.TORCH_MODEL_FOLDER if be.BACKEND == "torch" else be.MLX_MODEL_FOLDER
mp = resolve_model_path(models_dir, folder.strip())
if not mp.exists():
raise SystemExit(
f"모델 경로 없음: {mp}\n"
" config/llm_rl_model_download.py 에서 ENABLE_SNAPSHOT = True 로 받거나 폴더를 채우세요."
)
vision = make_local_vision_callable(
be.BACKEND,
str(mp),
max_new_tokens_torch=int(be.MAX_NEW_TOKENS_TORCH),
max_tokens_mlx=int(be.MAX_TOKENS_MLX),
)
import os
v_retry = int(os.environ.get("LOCAL_LLM_RETRY", "2"))
v_backoff = float(os.environ.get("LOCAL_LLM_RETRY_BACKOFF_SEC", "1.0"))
status_lbl = f"{be.BACKEND} · {mp.name}"
return vision, v_retry, v_backoff, status_lbl
def _load_action_csv_aligned_for_eval(path: Path, n: int) -> np.ndarray | None:
"""뷰어 df 행 수 n 에 맞게 tail 슬라이스한 정수 액션 배열."""
if not path.is_file():
logger.warning("액션 CSV 없음 — CSV 대비 학습 확인 패널 생략: %s", path)
return None
s = pd.read_csv(path, usecols=["action"]).squeeze("columns")
arr = s.astype(np.float64).round().astype(np.int64).to_numpy()
if len(arr) < n:
logger.warning(
"액션 CSV 행(%s)이 df(%s)보다 짧아 CSV 비교를 끕니다.",
len(arr),
n,
)
return None
if len(arr) > n:
logger.info("액션 CSV tail %s행만 사용 (df와 길이 맞춤)", n)
arr = arr[-n:]
return arr
def main() -> None:
print("[main_llm_db_rl_viewer] main() 시작", flush=True)
_maybe_download_hf()
vision, v_retry, v_backoff, status_lbl = _build_vision_callable()
logger.info("Market.get_data() …")
market = Market()
df = market.get_data()
if df is None or len(df) < tr.MIN_BARS + tr.BAR_STEP:
raise SystemExit("데이터가 부족합니다.")
n_full = len(df)
cap = getattr(tr, "RL_MAX_BARS", None)
bar_offset = 0
if cap is not None and n_full > int(cap):
bar_offset = n_full - int(cap)
df = df.iloc[-int(cap):].reset_index(drop=True)
logger.info(
"RL_MAX_BARS=%s 적용: tail %s봉 (전체 %s봉, bar_offset=%s)",
cap,
len(df),
n_full,
bar_offset,
)
df = merge_extreme_labels_from_csv(df, tr.LABELS_CSV_PATH, source_bar_offset=bar_offset)
eval_action_labels: np.ndarray | None = None
eval_action_csv_path: str | None = None
if bool(getattr(tr, "********", False)):
csv_path = Path(
getattr(
tr,
"PNG_STATUS_EVAL_ACTION_CSV_PATH",
getattr(tr, "ACTION_CSV_FOR_SUPERVISED_TRAIN", "cache/ACTION_repaired.CSV"),
)
)
eval_action_labels = _load_action_csv_aligned_for_eval(csv_path, len(df))
if eval_action_labels is not None:
eval_action_csv_path = str(csv_path)
logger.info("액션 CSV 비교 표시: %s", csv_path)
policy_doc_path = getattr(tr, "POLICY_DOC_PATH", None)
vision_policy_doc = (
load_policy_markdown(policy_doc_path)
if getattr(tr, "********", False)
else ""
)
# PngToStatus 정책 로드 (없으면 신규 생성)
if bool(getattr(tr, "********", False)):
pts_ckpt = getattr(
tr,
"********",
"cache/png_status_policy_action_repaired.pt",
)
logger.info("액션 CSV 모방학습 체크포인트 사용: %s", pts_ckpt)
if bool(getattr(tr, "ONLINE_TRAIN", False)):
logger.warning(
"ONLINE_TRAIN=True 이면 뷰어가 같은 경로에 정책을 덮어쓸 수 있습니다. "
"모방 가중치만 보려면 llm_rl_train.py 에서 ONLINE_TRAIN=False 권장."
)
else:
pts_ckpt = getattr(tr, "PNG_STATUS_CHECKPOINT_PATH", "cache/png_status_policy.pt")
pts_h = int(getattr(tr, "PNG_STATUS_INPUT_H", DEFAULT_INPUT_H))
pts_w = int(getattr(tr, "PNG_STATUS_INPUT_W", DEFAULT_INPUT_W))
pts_hidden = int(getattr(tr, "PNG_STATUS_HIDDEN", 256))
png_policy = load_png_status_policy(
pts_ckpt, h=pts_h, w=pts_w, hidden=pts_hidden
)
logger.info(
"PngToStatusPolicy 준비: ckpt=%s · 입력 %dx%d · hidden=%d",
pts_ckpt, pts_h, pts_w, pts_hidden,
)
port = ollama_bt.PaperLlmPortfolio(float(tr.INITIAL_KRW))
port.min_trade_bars = max(0, int(tr.MIN_TRADE_BARS))
app = PngStatusVisionBacktestViewer(
df=df,
portfolio=port,
vision_call=vision,
png_policy=png_policy,
online_train=bool(getattr(tr, "ONLINE_TRAIN", False)),
online_sample_action=bool(getattr(tr, "ONLINE_SAMPLE_ACTION", True)),
online_max_episodes=getattr(tr, "ONLINE_MAX_EPISODES", None),
online_save_every_ep=int(getattr(tr, "ONLINE_SAVE_EVERY_EP", 1)),
online_clear_text_each_ep=bool(getattr(tr, "ONLINE_CLEAR_TEXT_EACH_EP", False)),
checkpoint_path=str(pts_ckpt),
train_lr=float(getattr(tr, "PNG_STATUS_LR", tr.LEARNING_RATE)),
train_weight_decay=float(tr.WEIGHT_DECAY),
reward_win_weight=float(tr.REWARD_WIN_RATE_WEIGHT),
reward_profit_scale=float(tr.REWARD_PROFIT_SCALE),
initial_krw=float(tr.INITIAL_KRW),
min_trade_bars=int(tr.MIN_TRADE_BARS),
min_bars=max(50, tr.MIN_BARS),
bar_step=max(1, tr.BAR_STEP),
dry_run=tr.DRY_RUN,
auto_interval_sec=tr.AUTO_INTERVAL_SEC,
chart_max_width=tr.CHART_MAX_WIDTH,
vision_retries=max(1, int(be.VISION_RETRIES)),
vision_backoff_sec=float(be.VISION_BACKOFF_SEC),
window_title=(
getattr(tr, "PNG_STATUS_WINDOW_TITLE", "PNG→Status RL 차트 학습 뷰어 (캡쳐 이미지 기반)")
+ (
" · [액션모방 정책]"
if bool(getattr(tr, "********", False))
else ""
)
),
status_backend_label=status_lbl,
teacher_kl_coef=float(getattr(tr, "RL_TEACHER_KL_COEF", 0.0)),
teacher_temperature=float(getattr(tr, "RL_TEACHER_TEMPERATURE", 1.0)),
llm_align_weight=float(getattr(tr, "RL_LLM_ALIGN_WEIGHT", 0.0)),
vision_policy_doc=vision_policy_doc or None,
policy_doc_path_for_meta=str(policy_doc_path) if policy_doc_path else None,
save_train_meta_with_checkpoint=bool(
getattr(tr, "********", False)
),
eval_action_labels=eval_action_labels,
eval_action_csv_path=eval_action_csv_path,
)
if bool(getattr(tr, "ONLINE_TRAIN", False)) and bool(
getattr(tr, "ONLINE_AUTOSTART", False)
):
print(
"[main_llm_db_rl_viewer] 자동 스텝 시작 — torch 백엔드면 첫 LLM 호출에서 "
"VLM 로딩으로 수 분간 ‘처리 중…’만 보일 수 있습니다.",
flush=True,
)
app.auto_mode.set(True)
app.btn_stop.configure(state="normal")
app._start_step()
app.mainloop()
if __name__ == "__main__":
main()