예제 소스

main_llm_db_test_local_viewer.py

"""
로컬 디렉터리(models/ 아래) 비전 모델 백테스트 뷰어.

UI·로직은 viewer.VisionBacktestViewer 를 사용합니다.

  python main_llm_db_test_local_viewer.py --backend torch --model Qwen2.5-VL-3B-Instruct
  python main_llm_db_test_local_viewer.py --download-hf Qwen/Qwen2.5-VL-3B-Instruct
"""
from __future__ import annotations

import argparse
import json
import logging.config
import warnings
from pathlib import Path

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))

import logging

logger = logging.getLogger(__name__)

from Market import Market  # noqa: E402

import main_llm_db_test_ollama as ollama_bt  # noqa: E402
from viewer import (  # noqa: E402
    DEFAULT_MODELS_DIR,
    DEFAULT_MLX_MODEL_DIR,
    DEFAULT_TORCH_MODEL_DIR,
    VisionBacktestViewer,
    download_hf_snapshot,
    make_local_vision_callable,
    resolve_model_path,
)

_ENV_LOCAL_RETRY = int(__import__("os").environ.get("LOCAL_LLM_RETRY", "2"))
_ENV_LOCAL_BACKOFF = float(__import__("os").environ.get("LOCAL_LLM_RETRY_BACKOFF_SEC", "1.0"))


def main() -> None:
    parser = argparse.ArgumentParser(description="로컬 비전 백테스트 GUI (viewer.VisionBacktestViewer)")
    parser.add_argument("--models-dir", default=DEFAULT_MODELS_DIR)
    parser.add_argument(
        "--model",
        default=None,
        help="models-dir 아래 폴더 또는 절대 경로. torch/mlx 기본 폴더명은 viewer 모듈 참고",
    )
    parser.add_argument("--backend", choices=("torch", "mlx"), default="torch")
    parser.add_argument("--download-hf", metavar="REPO_ID", default=None)
    parser.add_argument("--min-bars", type=int, default=100)
    parser.add_argument("--bar-step", type=int, default=3)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--initial", type=float, default=ollama_bt.DEFAULT_INITIAL_KRW)
    parser.add_argument("--min-trade-bars", type=int, default=5)
    parser.add_argument("--auto-interval-sec", type=float, default=3.0)
    parser.add_argument("--chart-max-width", type=int, default=900)
    parser.add_argument("--max-new-tokens", type=int, default=512)
    parser.add_argument("--max-tokens-mlx", type=int, default=512)
    args = parser.parse_args()

    models_dir = Path(args.models_dir).resolve()
    models_dir.mkdir(parents=True, exist_ok=True)

    if args.download_hf:
        download_hf_snapshot(args.download_hf.strip(), models_dir)
        return

    default_name = DEFAULT_TORCH_MODEL_DIR if args.backend == "torch" else DEFAULT_MLX_MODEL_DIR
    model_path = resolve_model_path(models_dir, (args.model or default_name).strip())
    if not model_path.exists():
        raise SystemExit(
            f"모델 경로가 없습니다: {model_path}\n"
            f"  python main_llm_db_test_local_viewer.py --download-hf <HF_REPO>"
        )

    vision = make_local_vision_callable(
        args.backend,
        str(model_path),
        max_new_tokens_torch=args.max_new_tokens,
        max_tokens_mlx=args.max_tokens_mlx,
    )

    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("데이터가 부족합니다.")

    port = ollama_bt.PaperLlmPortfolio(float(args.initial))
    port.min_trade_bars = max(0, int(args.min_trade_bars))

    app = VisionBacktestViewer(
        df=df,
        portfolio=port,
        vision_call=vision,
        min_bars=max(50, args.min_bars),
        bar_step=max(1, args.bar_step),
        dry_run=args.dry_run,
        auto_interval_sec=args.auto_interval_sec,
        chart_max_width=args.chart_max_width,
        vision_retries=_ENV_LOCAL_RETRY,
        vision_backoff_sec=_ENV_LOCAL_BACKOFF,
        window_title="로컬 비전 백테스트 뷰어",
        status_backend_label=f"{args.backend} · {model_path.name}",
    )
    app.mainloop()


if __name__ == "__main__":
    main()