예제 소스

run_gemma4_chart_llm_only.py

"""
One-shot demo: build one chart from Market DB and get a trade opinion from Gemma4 (MLX-VLM) only.

No portfolio / no MA execution in code — only the model output is printed.

Requires: Apple Silicon, `pip install mlx-vlm`, and a Gemma4 MLX checkpoint under `models/`
(or pass an HF repo id that mlx-vlm can load).

Examples:
  python run_gemma4_chart_llm_only.py --model gemma-4-E4B-it
  python run_gemma4_chart_llm_only.py --model gemma-4-31b-it-8bit --bar 500
  python run_gemma4_chart_llm_only.py --model /abs/path/to/checkpoint --max-tokens 768

Torch path in this repo is Qwen-VL only; use this script (MLX) for Gemma4 weights.
"""

from __future__ import annotations

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

import numpy as np

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 ChartImageGenerator import ChartImageConfig, ChartImageGenerator  # noqa: E402
from Market import Market  # noqa: E402
from test_hf_img import _parse_opinion, _png_bytes_to_jpg_bytes  # noqa: E402

import main_llm_db_test_ollama as ollama_bt  # noqa: E402
from viewer import DEFAULT_MODELS_DIR, resolve_model_path  # noqa: E402

DEFAULT_GEMMA_DIR = "gemma-4-E4B-it"


def _build_gemma_delegated_prompt(ticker: str) -> str:
    """Prompt: model is sole decider; Korean labels on line 1/2 for _parse_opinion."""
    # Line 1 must match test_hf_img._parse_opinion (Korean opinion line).
    return f"""You are the ONLY decision maker for this single chart snapshot (ticker {ticker}).
The image shows candlesticks, the RED line is MA20, the BLUE line is MA60.

Use ONLY what you see. Do not assume live prices or order-book data.

Guidance (apply visually; your label on line 1 is final for this demo):
- Bullish: recent close crosses clearly ABOVE the red MA20 from below -> tend toward buy (\ub9e4\uc218) if structure supports it.
- Bearish: recent close crosses clearly BELOW the red MA20 from above -> tend toward sell (\ub9e4\ub3c4).
- If blue MA60 slopes clearly down (\ud558\ud5a5), be risk-off: do NOT recommend \ub9e4\uc218 on weak bounce / oversold-looking candles alone; prefer \ub9e4\ub3c4 or \uc720\uc9c0 unless reversal is strong.
- If unclear or noisy, choose \uc720\uc9c0.

Output format (MUST follow exactly for auto-parsing):
Line 1: \uc758\uacac: \ub9e4\uc218|\ub9e4\ub3c4|\uc720\uc9c0  (exactly one word after the colon)
Line 2+: \uadfc\uac70: 3-6 short sentences in Korean (MA20/MA60 vs price and risk).

Do not put any text before line 1."""


def main() -> None:
    ap = argparse.ArgumentParser(description="Gemma4 MLX: one chart -> LLM-only opinion (stdout)")
    ap.add_argument("--models-dir", default=DEFAULT_MODELS_DIR, help="Directory containing model folders")
    ap.add_argument(
        "--model",
        default=DEFAULT_GEMMA_DIR,
        help=f"Folder name under models-dir or absolute path (default: {DEFAULT_GEMMA_DIR})",
    )
    ap.add_argument(
        "--bar",
        type=int,
        default=None,
        help="Last row index (0-based) included in chart; default: last row of data",
    )
    ap.add_argument("--max-tokens", type=int, default=512)
    ap.add_argument("--temperature", type=float, default=0.0)
    ap.add_argument("--save-png", default=None, metavar="PATH", help="Save rendered chart PNG")
    ap.add_argument(
        "--prompt",
        default=None,
        help="Custom prompt text; if omitted, built-in Gemma-delegated prompt is used",
    )
    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(
            f"Model path not found: {model_path}\n"
            "  Hint: download an mlx-community Gemma4 VLM snapshot into models/ "
            "or pass a valid HF id/path."
        )

    logger.info("Market.get_data() …")
    market = Market()
    df = market.get_data()
    if df is None or len(df) < 10:
        raise SystemExit("No data or too few rows from Market.get_data().")

    df = ollama_bt._ensure_pb_mfi10(df)
    n = len(df)
    bar = int(args.bar) if args.bar is not None else n - 1
    if bar < 0 or bar >= n:
        raise SystemExit(f"--bar must be in 0..{n - 1}, got {bar}")

    df_hist = df.iloc[: bar + 1].copy()
    chart_idx = len(df_hist)
    price = float(df_hist.iloc[-1]["close"])

    chart_gen = ChartImageGenerator(
        ticker=ollama_bt.TICKER,
        config=ChartImageConfig(show=False),
    )
    base_png, _parts = 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,
    )
    if args.save_png:
        Path(args.save_png).write_bytes(base_png)
        logger.info("Wrote PNG: %s", args.save_png)

    img_bytes = _png_bytes_to_jpg_bytes(base_png)
    prompt = (args.prompt or "").strip() or _build_gemma_delegated_prompt(ollama_bt.TICKER)

    from local_vision_mlx import mlx_vision_chat  # noqa: WPS433

    logger.info("mlx_vision_chat … model=%s bar=%s close=%s", model_path.name, bar, price)
    reply = mlx_vision_chat(
        image_bytes=img_bytes,
        user_text=prompt,
        model_id=str(model_path),
        max_tokens=max(32, int(args.max_tokens)),
        temperature=float(args.temperature),
    )

    opinion, _raw = _parse_opinion(reply)
    print("--- meta ---", file=sys.stdout)
    print(f"ticker={ollama_bt.TICKER} bar_index={bar} close={price:,.0f}", file=sys.stdout)
    print(f"model={model_path}", file=sys.stdout)
    print("--- parsed opinion ---", file=sys.stdout)
    print(opinion or "(parse failed; see full reply)", file=sys.stdout)
    print("--- full reply ---", file=sys.stdout)
    print(reply.strip(), file=sys.stdout)


if __name__ == "__main__":
    main()