예제 소스

main_llm_db_portfolio_viewer.py

"""
비전 LLM 백테스트 뷰어 (ollama / 로컬 torch / mlx).

UI는 viewer.VisionBacktestViewer 단일 클래스를 사용합니다.

  python main_llm_db_portfolio_viewer.py --backend ollama
  python main_llm_db_portfolio_viewer.py --backend torch --model 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,
    make_ollama_vision_callable,
    resolve_model_path,
)


def main() -> None:
    parser = argparse.ArgumentParser(description="비전 백테스트 뷰어 (viewer.VisionBacktestViewer)")
    parser.add_argument("--backend", choices=("ollama", "torch", "mlx"), default="ollama")
    parser.add_argument("--models-dir", default=DEFAULT_MODELS_DIR)
    parser.add_argument("--model", default=None)
    parser.add_argument("--ollama-host", default=ollama_bt.DEFAULT_OLLAMA_HOST)
    parser.add_argument("--ollama-model", default=ollama_bt.DEFAULT_OLLAMA_MODEL)
    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("--initial", type=float, default=ollama_bt.DEFAULT_INITIAL_KRW)
    parser.add_argument("--min-trade-bars", type=int, default=5)
    parser.add_argument("--dry-run", action="store_true")
    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

    if args.backend == "ollama":
        vision = make_ollama_vision_callable(
            host=str(args.ollama_host),
            model=str(args.ollama_model),
        )
        v_retry = max(1, int(ollama_bt.OLLAMA_RETRY))
        v_backoff = float(ollama_bt.OLLAMA_RETRY_BACKOFF_SEC)
        status_lbl = f"ollama · {args.ollama_model}"
        title = "비전 백테스트 뷰어 (Ollama)"
    else:
        default_name = DEFAULT_TORCH_MODEL_DIR if args.backend == "torch" else DEFAULT_MLX_MODEL_DIR
        mp = resolve_model_path(models_dir, (args.model or default_name).strip())
        if not mp.exists():
            raise SystemExit(f"모델 경로 없음: {mp}\n  --download-hf 로 받거나 경로 확인")
        vision = make_local_vision_callable(
            args.backend,
            str(mp),
            max_new_tokens_torch=args.max_new_tokens,
            max_tokens_mlx=args.max_tokens_mlx,
        )
        v_retry = int(__import__("os").environ.get("LOCAL_LLM_RETRY", "2"))
        v_backoff = float(__import__("os").environ.get("LOCAL_LLM_RETRY_BACKOFF_SEC", "1.0"))
        status_lbl = f"{args.backend} · {mp.name}"
        title = "비전 백테스트 뷰어 (로컬)"

    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=v_retry,
        vision_backoff_sec=v_backoff,
        window_title=title,
        status_backend_label=status_lbl,
    )
    app.mainloop()


if __name__ == "__main__":
    main()