예제 소스
차트 REINFORCE: 규칙 교사와 KL로 정책망 준비하기 · 파일 59개
test_hf_img.py
"""
Redis(또는 CSV/더미) 데이터로 차트를 렌더한 뒤, 이미지는 **메모리(JPEG bytes)** 에만 두고
Hugging Face 비전 LLM에 넘겨 의견을 **매수 / 매도 / 유지** 중 하나로 받습니다.
**기본 출력:** 표준 출력에는 위 세 가지 중 **한 단어만** 출력합니다. 전체 응답·진행 로그는 `--verbose`.
디스크에 origin.jpg 등을 남기려면 `--save-images` 를 지정하세요.
`--skip-generate` 는 기존 파일 경로(`--origin`)에서 읽습니다.
필요: plotly, kaleido, pillow, Redis(옵션), HF 토큰(private/hf_key.py 또는 환경변수 HF_TOKEN)
로컬 비전 모델: 환경변수 HF_VISION_LOCAL=1 이면 Router API 대신 로컬 추론(local_vision_llm.py).
이 경우 HF 토큰 불필요. HF_LOCAL_VISION_MODEL 로 모델 ID/경로 지정(기본 Qwen2.5-VL-3B).
원격 기본(397B 등)은 로컬에서 자동으로 3B instruct 로 대체됩니다.
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import re
import sys
import urllib.error
import urllib.request
from datetime import datetime, timedelta
from io import BytesIO, StringIO
import numpy as np
import pandas as pd
from ChartImageGenerator import ChartImageConfig, ChartImageGenerator
# Hugging Face 토큰: 환경변수 우선, 없으면 private/hf_key.py의 HG_TOKEN
try:
from private.hf_key import HG_TOKEN as _HG_TOKEN_FROM_FILE
except ImportError:
_HG_TOKEN_FROM_FILE = ""
HF_TOKEN = os.environ.get("HF_TOKEN", "").strip() or (
_HG_TOKEN_FROM_FILE.strip() if isinstance(_HG_TOKEN_FROM_FILE, str) else str(_HG_TOKEN_FROM_FILE)
)
# OpenAI 호환 채팅 (HF Inference Providers Router).
# 베이스: https://huggingface.co/changelog/inference-providers-openai-compatible
# 전체 URL은 쿼리스트링(?...) 없이 POST JSON 본문만 사용합니다 (OpenAI /v1/chat/completions 규약).
HF_CHAT_URL = os.environ.get(
"HF_CHAT_URL",
"https://router.huggingface.co/v1/chat/completions",
)
# Router /v1/chat/completions 은 **대화형(chat) VLM**만 허용합니다.
# google/gemma-4-E4B-it 등 any-to-any·일부 image-text-to-text 모델은
# 허브에 있어도 "not a chat model" 400 이 날 수 있습니다.
# 후보: [Models](https://huggingface.co/models) 에서 Image-Text-to-Text + Inference,
# 그리고 [Chat Completion — Conversational VLMs](https://huggingface.co/docs/api-inference/en/tasks/chat-completion) 예: Qwen2.5-VL-Instruct, GLM-4.5V 등.
# 계정별 Provider 제한이 있으면 HF_VISION_MODEL / --model 로 조정.
# (실사용에서 Router 성공 사례: Qwen/Qwen3.5-397B-A17B, max_tokens 8192)
DEFAULT_VISION_MODEL = os.environ.get(
"HF_VISION_MODEL",
"Qwen/Qwen3.5-397B-A17B",
)
DEFAULT_HF_MAX_TOKENS = int(os.environ.get("HF_MAX_TOKENS", "8192"))
def _dummy_df(rows: int = 600) -> pd.DataFrame:
rng = np.random.default_rng(42)
base = 100_000 + np.cumsum(rng.normal(0, 200, size=rows))
high = base + rng.uniform(50, 400, size=rows)
low = base - rng.uniform(50, 400, size=rows)
open_ = base + rng.normal(0, 50, size=rows)
close = base + rng.normal(0, 50, size=rows)
volume = rng.uniform(10, 1000, size=rows)
df = pd.DataFrame(
{
"open": open_,
"high": high,
"low": low,
"close": close,
"volume": volume,
}
)
df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()
df["ma120"] = df["close"].rolling(120).mean()
mid = df["close"].rolling(40).mean()
std = df["close"].rolling(40).std()
df["middle"] = mid
df["upper"] = mid + (std * 2.3)
df["lower"] = mid - (std * 2.3)
df["ma20_deg"] = (df["ma20"] - df["ma20"].shift()).fillna(0)
df["PB"] = ((df["close"] - df["lower"]) / (df["upper"] - df["lower"])).replace(
[np.inf, -np.inf], np.nan
)
df["PB"] = df["PB"].fillna(0.5).clip(0, 1)
df["MFI10"] = (50 + rng.normal(0, 10, size=rows)).clip(0, 100)
start = datetime.now() - timedelta(minutes=rows * 3)
df["time"] = [start + timedelta(minutes=3 * i) for i in range(rows)]
return df.ffill().fillna(0)
def _load_df(path: str | None) -> pd.DataFrame:
if not path:
return _dummy_df()
if not os.path.exists(path):
raise FileNotFoundError(path)
if path.lower().endswith(".csv"):
return pd.read_csv(path)
if path.lower().endswith(".parquet"):
return pd.read_parquet(path)
raise ValueError("지원 포맷: .csv 또는 .parquet")
def _load_df_from_redis(
*,
host: str = "localhost",
port: int = 6379,
db: int = 0,
key: str = "cul_df",
) -> pd.DataFrame:
try:
import redis # type: ignore
except ImportError as e:
raise ImportError("Redis에서 읽으려면 redis 패키지가 필요합니다. `pip install redis`") from e
r = redis.StrictRedis(host=host, port=port, db=db)
raw = r.get(key)
if not raw:
raise KeyError(f"Redis key '{key}'에 데이터가 없습니다.")
df_str = raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else str(raw)
df = pd.read_json(StringIO(df_str), orient="records")
if df is None or df.empty:
raise ValueError(f"Redis key '{key}'의 DataFrame이 비어있습니다.")
return df
def _png_bytes_to_jpg_bytes(png_bytes: bytes, quality: int = 92) -> bytes:
"""PNG(kaleido 출력) → JPEG 바이트(메모리만, 파일 없음)."""
try:
from PIL import Image
except ImportError as e:
raise ImportError("JPEG 변환에는 Pillow가 필요합니다. `pip install pillow`") from e
im = Image.open(BytesIO(png_bytes)).convert("RGB")
buf = BytesIO()
im.save(buf, format="JPEG", quality=quality, optimize=True)
return buf.getvalue()
def _png_bytes_to_jpg_file(png_bytes: bytes, out_path: str, quality: int = 92) -> None:
data = _png_bytes_to_jpg_bytes(png_bytes, quality=quality)
parent = os.path.dirname(os.path.abspath(out_path))
if parent:
os.makedirs(parent, exist_ok=True)
with open(out_path, "wb") as f:
f.write(data)
def _read_image_bytes(path: str) -> bytes:
with open(path, "rb") as f:
return f.read()
def _build_analysis_prompt(ticker: str, policy_reference_doc: str | None = None) -> str:
doc_block = ""
if policy_reference_doc and policy_reference_doc.strip():
doc_block = (
"\n\n---\n**Desk policy (Korean notes; align your commentary with these principles, "
"execution remains rule-based in the app):**\n"
+ policy_reference_doc.strip()
+ "\n---\n"
)
return f"""You are an expert at reading candlestick charts for crypto/stocks.
The attached image shows the candlestick chart and moving averages for ticker {ticker}.
{doc_block}
IMPORTANT — roles (do not confuse them):
- **Automated trading in this app does NOT use your 매수/매도/유지.** Buy/sell execution is decided only by numeric rules on the data (red MA = MA20 close cross; blue MA = MA60 trend filter). Your job is **commentary / reference opinion only**.
What you MUST look at (for your written analysis):
1) **Blue line = MA60.** If it is clearly curving / sloping **downward** over the last part of the chart, describe that as a **bearish / risk-off** backdrop (in Korean: e.g. 하향 곡선, 둔화, 역풍). In that regime, a discretionary trader would often **avoid new longs** and favor **defensive / sell-side** bias — the program will **block buys** when MA60 is downward; it does **not** take orders from your text.
**Oversold trap (CRITICAL for Korean commentary):** When MA60 is **downward**, short-term **과매도** looks (long lower wicks, sharp dip, stretched candles) often mean only a **bounce inside a downtrend**, not a proven bottom. **Do not** justify optimism or waiting out losses using 과매도 alone. Delaying defense (no exit) in this structure can leave a holder exposed to a **sharp crash** and **large drawdowns**. In `근거`, you MUST add one explicit sentence on whether this **과매도-in-downtrend trap** applies, in Korean (적용됨 또는 해당 없음 + 짧은 이유).
2) **Red line = MA20.** Describe whether the latest candle **closes above** (상향 돌파 후보) or **below** (하향 돌파 후보) the red MA, vs the immediately previous candle relationship. Bullish cross ≈ price crosses **up** through red MA; bearish cross ≈ price crosses **down** through red MA. Again: this is your **visual explanation**; execution uses the same rule in code, not your label.
**Line 1 (참고의견) guideline when MA60 bends down:** Do **not** output **매수** unless reversal is **strong** (e.g. clear higher lows, sustained reclaim above MA20/MA60 with follow-through). Default bias: **매도** (risk-off, stress protecting capital / respecting downtrend) or **유지** only if structure is genuinely not bearish **and** `근거` explains why short-term 과매도 is **not** a bull trap. Never sound complacent from oversold alone while MA60 still slopes down.
Output format (MUST follow, for auto-parsing of your **reference** opinion only):
Line 1: 참고의견: 매수|매도|유지 (one word; follow the MA60-downward bias rules above)
Line 2+: 근거: 3-6 short sentences in Korean (MA60 shape, MA20 vs close, and the **과매도·하락 추세 함정** sentence whenever MA60 is downward)
You may also use the legacy prefix `의견:` instead of `참고의견:` on line 1 if needed, but `참고의견:` is preferred.
Do not omit the first line with either `참고의견:` or `의견:`."""
def _extract_assistant_text(msg: dict) -> str:
"""OpenAI 호환 assistant message 에서 텍스트 추출.
Qwen3 계열 등은 `content` 가 null 이고 `reasoning_content`(또는 `reasoning`)에만
본문이 오는 경우가 있어, 해당 필드를 함께 본다.
"""
if not isinstance(msg, dict):
return ""
def _from_content_list(parts: list) -> str:
chunks: list[str] = []
for p in parts:
if not isinstance(p, dict):
continue
t = p.get("type")
if t == "text":
chunks.append(p.get("text") or "")
elif t in ("reasoning", "output_text"):
chunks.append(p.get("text") or "")
elif p.get("text") is not None:
chunks.append(str(p.get("text")))
return "\n".join(chunks).strip()
content = msg.get("content")
if isinstance(content, list):
s = _from_content_list(content)
if s:
return s
if isinstance(content, str) and content.strip():
return content.strip()
for key in ("reasoning_content", "reasoning"):
alt = msg.get(key)
if isinstance(alt, str) and alt.strip():
return alt.strip()
refusal = msg.get("refusal")
if isinstance(refusal, str) and refusal.strip():
return refusal.strip()
if isinstance(content, list):
return _from_content_list(content)
return ""
def _hf_vision_chat(
*,
token: str,
image_bytes: bytes,
user_text: str,
model: str,
timeout_sec: int = 120,
max_tokens: int = DEFAULT_HF_MAX_TOKENS,
) -> str:
"""HF Router OpenAI 호환 chat/completions (image_url + text).
HF_VISION_LOCAL=1 이면 Hugging Face Router 대신 로컬 VLM(transformers) 사용.
이 경우 token 은 무시됩니다.
"""
if os.environ.get("HF_VISION_LOCAL", "").strip().lower() in (
"1",
"true",
"yes",
"on",
):
from local_vision_llm import local_vision_chat
return local_vision_chat(
image_bytes=image_bytes,
user_text=user_text,
model_id=model,
max_new_tokens=max_tokens,
)
mime = "image/jpeg"
if image_bytes[:8] == b"\x89PNG\r\n\x1a\n":
mime = "image/png"
b64 = base64.standard_b64encode(image_bytes).decode("ascii")
data_url = f"data:{mime};base64,{b64}"
payload = {
"model": model,
"messages": [
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": data_url}},
{"type": "text", "text": user_text},
],
}
],
"max_tokens": max_tokens,
"temperature": 0.2,
}
req = urllib.request.Request(
HF_CHAT_URL,
data=json.dumps(payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Accept": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout_sec) as resp:
raw = resp.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as e:
err_body = e.read().decode("utf-8", errors="replace") if e.fp else ""
extra = ""
low = err_body.lower()
if "not a chat model" in low:
extra = (
"\n→ 이 모델 ID는 Router chat/completions 용 대화형 모델이 아닙니다. "
"HF_VISION_MODEL 또는 --model 을 Qwen2.5-VL-*-Instruct 등 "
"Chat Completion 문서의 Conversational VLM 으로 바꾸세요."
)
elif "not supported by any provider" in low or "model_not_supported" in low:
extra = (
"\n→ 계정 Inference Providers 에서 해당 모델을 제공하는지 확인하거나 "
"다른 모델 ID(또는 model:provider 형식)를 쓰세요."
)
raise RuntimeError(
f"HF API HTTP {e.code}: {err_body[:2000] or e.reason}\n"
f"URL={HF_CHAT_URL} model={model}{extra}"
) from e
except urllib.error.URLError as e:
raise RuntimeError(f"HF API 연결 실패: {e}") from e
try:
body = json.loads(raw)
except json.JSONDecodeError as e:
raise RuntimeError(f"HF 응답 JSON 파싱 실패: {raw[:500]}") from e
if "error" in body:
err = body["error"]
raise RuntimeError(f"HF API error: {err}")
choices = body.get("choices") or []
if not choices:
raise RuntimeError(f"HF 응답에 choices 없음: {raw[:800]}")
msg = choices[0].get("message") or {}
text = _extract_assistant_text(msg)
if not text:
raise RuntimeError(
"HF 응답에서 assistant 텍스트를 찾을 수 없습니다. "
"(Qwen3 등: content 가 비고 reasoning_content 만 오는 경우가 있음 — 파서 보강됨). "
f"message={json.dumps(msg, ensure_ascii=False)[:2500]}\nraw[:1500]={raw[:1500]}"
)
return text
def _parse_opinion(text: str) -> tuple[str | None, str]:
"""응답에서 매수/매도/유지 추출.
긴 분석 뒤에 결론이 오거나, 따옴표·종합 판단 문장에만 키워드가 있는 경우를 처리한다.
"""
if not text:
return None, ""
m = re.search(r"참고의견\s*[:：]\s*(매수|매도|유지)", text)
if m:
return m.group(1), text
m = re.search(r"의견\s*[:：]\s*(매수|매도|유지)", text)
if m:
return m.group(1), text
# "매도", '유지', 「매도」 등 따옴표·각괄호로 감싼 최종 선택
m = re.search(
r'(?:["\u201c\u201d\u2018\u2019]|[\u300c「])(매수|매도|유지)(?:["\u201c\u201d\u2018\u2019]|[\u300d」])',
text,
)
if m:
return m.group(1), text
# 종합/결론/따라서/옵션 등 결론부 문장
m = re.search(
r"(?:종합\s*판단|결론|최종|따라서|따라\s*서|질문의\s*옵션|옵션\s*중)[^。\n]{0,1200}?(매수|매도|유지)",
text,
re.DOTALL,
)
if m:
return m.group(1), text
# 전체 본문에서 한글 단어 경계 매칭 후 **마지막** (구버전은 앞 200자만 보아 긴 응답에서 실패)
boundary = re.compile(r"(?<![가-힣])(매수|매도|유지)(?![가-힣])")
matches = list(boundary.finditer(text))
if matches:
return matches[-1].group(1), text
# 끝부분: 마지막 4000자에서 구두점·공백 뒤에 오는 세 단어 중 **가장 마지막** 등장
tail = text[-4000:] if len(text) > 4000 else text
loose = re.compile(
r"(?:^|[\s,.;:!?(\[\{⟨«])(매수|매도|유지)(?=[\s,.;:!?)\]\}⟩»]|이다|입니다|습니다|다\b|$)"
)
lm = list(loose.finditer(tail))
if lm:
return lm[-1].group(1), text
return None, text
def main() -> None:
ap = argparse.ArgumentParser(
description="Redis 등에서 데이터 로드 → 차트는 메모리 JPEG로 HF 비전 LLM에 전달 → 매수/매도/유지 의견"
)
ap.add_argument("--csv", default=None, help="입력 데이터(.csv/.parquet). 없으면 더미")
ap.add_argument("--redis", action="store_true", help="Redis cul_df 사용")
ap.add_argument("--redis_host", default="localhost")
ap.add_argument("--redis_port", type=int, default=6379)
ap.add_argument("--redis_db", type=int, default=0)
ap.add_argument("--redis_key", default="cul_df")
ap.add_argument("--outdir", default="out_imgs", help="--save-images 시 저장 폴더")
ap.add_argument(
"--save-images",
action="store_true",
help="생성한 origin/slice JPG를 디스크에 저장 (기본: 메모리만, 파일 안 씀)",
)
ap.add_argument("--ticker", default="KRW-ETH")
ap.add_argument("--idx", type=int, default=500)
ap.add_argument("--max_data", type=int, default=300)
ap.add_argument("--n_slices", type=int, default=3)
ap.add_argument("--overlap_px", type=int, default=0)
ap.add_argument("--scale", type=int, default=2)
ap.add_argument("--debug_env", action="store_true")
ap.add_argument(
"--skip-generate",
action="store_true",
help="이미지 생성 생략, 기존 origin.jpg만 분석",
)
ap.add_argument("--origin", default=None, help="분석할 이미지 경로 (기본: outdir/origin.jpg)")
ap.add_argument(
"--model",
default=DEFAULT_VISION_MODEL,
help="HF Router chat용 비전 모델 id (기본: HF_VISION_MODEL 또는 Qwen3.5-397B-A17B)",
)
ap.add_argument("--timeout", type=int, default=120, help="HF API 타임아웃(초)")
ap.add_argument(
"--max-tokens",
type=int,
default=DEFAULT_HF_MAX_TOKENS,
help="생성 토큰 상한 (기본 8192). 환경변수 HF_MAX_TOKENS 로도 설정 가능",
)
ap.add_argument(
"-v",
"--verbose",
action="store_true",
help="진행 로그·모델 전체 응답 출력 (기본: stdout에는 매수/매도/유지 한 줄만)",
)
args = ap.parse_args()
if args.debug_env:
print("python:", sys.executable)
print("HF_CHAT_URL:", HF_CHAT_URL)
print("model:", args.model)
local_on = os.environ.get("HF_VISION_LOCAL", "").strip().lower() in (
"1",
"true",
"yes",
"on",
)
if not HF_TOKEN and not local_on:
print(
"HF 토큰이 없습니다. private/hf_key.py에 HG_TOKEN을 두거나 환경변수 HF_TOKEN을 설정하세요. "
"로컬 모델만 쓰려면 HF_VISION_LOCAL=1 을 설정하세요.",
file=sys.stderr,
)
sys.exit(2)
origin_path = args.origin or os.path.join(args.outdir, "origin.jpg")
if not args.skip_generate:
if args.redis:
df = _load_df_from_redis(
host=args.redis_host,
port=args.redis_port,
db=args.redis_db,
key=args.redis_key,
)
else:
df = _load_df(args.csv)
gen = ChartImageGenerator(
ticker=args.ticker,
config=ChartImageConfig(show=False),
)
base_png, parts_png = gen.render_and_slice_by_x(
df,
idx=args.idx,
max_data=args.max_data,
ticker=args.ticker,
n_slices=args.n_slices,
overlap_px=args.overlap_px,
scale=args.scale,
)
# LLM에는 항상 메모리상 JPEG 바이트만 사용 (디스크 불필요)
img_bytes = _png_bytes_to_jpg_bytes(base_png)
if args.save_images:
os.makedirs(args.outdir, exist_ok=True)
_png_bytes_to_jpg_file(base_png, origin_path)
for i, p in enumerate(parts_png):
_png_bytes_to_jpg_file(p, os.path.join(args.outdir, f"slice_{i:02d}.jpg"))
if args.verbose:
print(f"saved: {origin_path}", file=sys.stderr)
print(f"saved slices: {len(parts_png)} (dir: {args.outdir})", file=sys.stderr)
else:
if args.verbose:
print(
f"chart JPEG in memory: {len(img_bytes)} bytes (slices={len(parts_png)}, disk skip)",
file=sys.stderr,
)
else:
if not os.path.isfile(origin_path):
print(f"파일 없음: {origin_path}", file=sys.stderr)
sys.exit(1)
img_bytes = _read_image_bytes(origin_path)
prompt = _build_analysis_prompt(args.ticker)
if args.verbose:
print("\n--- HF 비전 분석 요청 ---", file=sys.stderr)
reply = _hf_vision_chat(
token=HF_TOKEN,
image_bytes=img_bytes,
user_text=prompt,
model=args.model,
timeout_sec=args.timeout,
max_tokens=args.max_tokens,
)
opinion, _ = _parse_opinion(reply)
if args.verbose:
print("\n=== 모델 전체 응답 ===\n", file=sys.stderr)
print(reply, file=sys.stderr)
print("\n=== 최종 의견 ===", file=sys.stderr)
if opinion:
print(opinion)
else:
print(
"파싱 실패: 응답에서 매수·매도·유지를 찾지 못했습니다. "
"모델 전문을 보려면 --verbose 로 다시 실행하세요.",
file=sys.stderr,
)
sys.exit(1)
if __name__ == "__main__":
main()