OnceAI오래된 문제를 다시 적는 곳

게시판암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM예제 소스

예제 소스

암호화폐 자동매매를 위한 CNN 시각 정책망과 VLM · 파일 59개

test_hf_img_unit.py

"""
test_hf_img.py 헬퍼 및 HF 호출 로직 단위 테스트 (네트워크 없음).

통합 실행 시 HTTP 400 model_not_supported 는 Hugging Face 계정에서 켠
Inference Provider 와 --model / HF_VISION_MODEL 조합 문제이며,
이 파일의 테스트는 그 경우를 “모킹한 400 응답”으로만 재현합니다.

실행: 프로젝트 루트에서  python -m unittest discover -s tests -v
실 Router + 이미지: tests/test_hf_img_integration.py (HF_INTEGRATION_TEST=1)
"""
from __future__ import annotations

import base64
import json
import re
import unittest
from io import BytesIO
from unittest.mock import MagicMock, patch

# 프로젝트 루트의 test_hf_img 모듈
import test_hf_img as hf


def _payload_from_mock(mock_urlopen: MagicMock) -> dict:
    req = mock_urlopen.call_args[0][0]
    return json.loads(req.data.decode("utf-8"))


def _image_url_from_payload(payload: dict) -> str:
    parts = payload["messages"][0]["content"]
    self0 = parts[0]
    assert self0["type"] == "image_url"
    return self0["image_url"]["url"]


def _fake_urlopen_cm(body: dict) -> MagicMock:
    """urlopen() 이 with 문에서 쓰는 컨텍스트 매니저를 흉내 냄."""
    raw = json.dumps(body, ensure_ascii=False).encode("utf-8")
    resp = MagicMock()
    resp.read.return_value = raw
    cm = MagicMock()
    cm.__enter__.return_value = resp
    cm.__exit__.return_value = None
    return cm


class TestPngToJpeg(unittest.TestCase):
    def test_jpg_magic_and_non_empty(self) -> None:
        try:
            from PIL import Image
        except ImportError:
            self.skipTest("Pillow 없음")

        buf = BytesIO()
        Image.new("RGB", (4, 4), color=(128, 64, 32)).save(buf, format="PNG")
        png = buf.getvalue()
        self.assertTrue(png.startswith(b"\x89PNG\r\n\x1a\n"))

        jpg = hf._png_bytes_to_jpg_bytes(png)
        self.assertTrue(jpg.startswith(b"\xff\xd8"), "JPEG SOI")
        self.assertGreater(len(jpg), 100)


class TestParseOpinion(unittest.TestCase):
    def test_colon_form(self) -> None:
        op, _ = hf._parse_opinion("의견: 매수\n근거: 테스트")
        self.assertEqual(op, "매수")

    def test_fullwidth_colon(self) -> None:
        op, _ = hf._parse_opinion("의견：유지\n근거")
        self.assertEqual(op, "유지")

    def test_fallback_keyword(self) -> None:
        op, _ = hf._parse_opinion("현재는 매도 쪽으로 보입니다.")
        self.assertEqual(op, "매도")

    def test_empty(self) -> None:
        op, full = hf._parse_opinion("")
        self.assertIsNone(op)
        self.assertEqual(full, "")

    def test_whitespace_only(self) -> None:
        op, full = hf._parse_opinion("   \n\t  ")
        self.assertIsNone(op)

    def test_no_match_returns_none(self) -> None:
        op, full = hf._parse_opinion("차트만 있고 의견 문구가 없습니다.")
        self.assertIsNone(op)
        self.assertEqual(full, "차트만 있고 의견 문구가 없습니다.")

    def test_opinion_precedence_over_fallback(self) -> None:
        """첫 줄 의견이 있으면 키워드 휴리스틱보다 우선."""
        text = "의견: 유지\n근거: 단기 매수 신호도 보이나 불확실"
        op, _ = hf._parse_opinion(text)
        self.assertEqual(op, "유지")

    def test_quoted_verdict(self) -> None:
        op, _ = hf._parse_opinion(
            '따라서 질문의 옵션 중 "매도"가 가장 논리적으로 보입니다.'
        )
        self.assertEqual(op, "매도")

    def test_long_text_conclusion_at_end(self) -> None:
        """긴 본문 뒤 결론(구버전은 앞 200자만 검사해 실패)."""
        long_prefix = "분석 내용. " * 80
        op, _ = hf._parse_opinion(long_prefix + "따라서 매도 쪽이 타당합니다.")
        self.assertEqual(op, "매도")

    def test_zonghe_heading(self) -> None:
        op, _ = hf._parse_opinion("**2. 종합 판단:**\n따라서 매도가 맞습니다.")
        self.assertEqual(op, "매도")


class TestBuildPrompt(unittest.TestCase):
    def test_contains_ticker(self) -> None:
        p = hf._build_analysis_prompt("KRW-BTC")
        self.assertIn("KRW-BTC", p)
        self.assertIn("매수", p)


class TestHfVisionChatMocked(unittest.TestCase):
    @patch("urllib.request.urlopen")
    def test_returns_string_content(self, mock_urlopen: MagicMock) -> None:
        mock_urlopen.return_value = _fake_urlopen_cm(
            {"choices": [{"message": {"content": "의견: 유지\n근거: 안전"}}]}
        )
        out = hf._hf_vision_chat(
            token="t",
            image_bytes=b"\xff\xd8\xff",  # JPEG-ish for mime
            user_text="hi",
            model="m/M",
            timeout_sec=5,
        )
        self.assertIn("유지", out)
        mock_urlopen.assert_called_once()
        req = mock_urlopen.call_args[0][0]
        self.assertEqual(req.get_header("Authorization"), "Bearer t")
        payload = json.loads(req.data.decode("utf-8"))
        self.assertEqual(payload["model"], "m/M")
        self.assertEqual(payload["messages"][0]["role"], "user")

    @patch("urllib.request.urlopen")
    def test_png_mime_in_payload(self, mock_urlopen: MagicMock) -> None:
        png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 20
        mock_urlopen.return_value = _fake_urlopen_cm(
            {"choices": [{"message": {"content": "ok"}}]}
        )
        hf._hf_vision_chat(
            token="x",
            image_bytes=png,
            user_text="q",
            model="vision/x",
            timeout_sec=5,
        )
        payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8"))
        parts = payload["messages"][0]["content"]
        url = parts[0]["image_url"]["url"]
        self.assertIn("data:image/png;base64,", url)

    @patch("urllib.request.urlopen")
    def test_list_content_parts(self, mock_urlopen: MagicMock) -> None:
        mock_urlopen.return_value = _fake_urlopen_cm(
            {
                "choices": [
                    {
                        "message": {
                            "content": [
                                {"type": "text", "text": "의견: 매도"},
                                {"type": "text", "text": "\n둘째"},
                            ]
                        }
                    }
                ]
            }
        )
        out = hf._hf_vision_chat(
            token="t",
            image_bytes=b"\xff\xd8\xff",
            user_text="q",
            model="m",
            timeout_sec=5,
        )
        self.assertIn("매도", out)

    @patch("urllib.request.urlopen")
    def test_http_error_message(self, mock_urlopen: MagicMock) -> None:
        import urllib.error

        fp = BytesIO(
            b'{"error":{"message":"model_not_supported","type":"invalid_request_error"}}'
        )
        err = urllib.error.HTTPError(
            hf.HF_CHAT_URL, 400, "Bad Request", hdrs={}, fp=fp
        )
        mock_urlopen.side_effect = err

        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="t",
                image_bytes=b"\xff\xd8\xff",
                user_text="q",
                model="Bad/Model",
                timeout_sec=5,
            )
        self.assertIn("400", str(ctx.exception))
        self.assertIn("Bad/Model", str(ctx.exception))

    @patch("urllib.request.urlopen")
    def test_http_400_router_model_not_supported_body(self, mock_urlopen: MagicMock) -> None:
        """터미널에 나온 Router 400 JSON 형태를 그대로 모킹."""
        import urllib.error

        body = {
            "error": {
                "message": (
                    "The requested model 'Qwen/Qwen2.5-VL-7B-Instruct' "
                    "is not supported by any provider you have enabled."
                ),
                "type": "invalid_request_error",
                "param": "model",
                "code": "model_not_supported",
            }
        }
        fp = BytesIO(json.dumps(body).encode("utf-8"))
        err = urllib.error.HTTPError(
            hf.HF_CHAT_URL, 400, "Bad Request", hdrs={}, fp=fp
        )
        mock_urlopen.side_effect = err

        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="********",
                image_bytes=b"\xff\xd8\xff\xe0\x00\x10JFIF",
                user_text="분석",
                model="Qwen/Qwen2.5-VL-7B-Instruct",
                timeout_sec=30,
            )
        msg = str(ctx.exception)
        self.assertIn("HTTP 400", msg)
        self.assertIn("model_not_supported", msg)
        self.assertIn("provider you have enabled", msg)
        self.assertIn("Inference Providers", msg)

    @patch("urllib.request.urlopen")
    def test_http_400_not_a_chat_model_hint(self, mock_urlopen: MagicMock) -> None:
        import urllib.error

        body = {
            "error": {
                "message": "The requested model 'google/gemma-4-E4B-it' is not a chat model.",
                "type": "invalid_request_error",
                "param": "model",
                "code": "model_not_supported",
            }
        }
        fp = BytesIO(json.dumps(body).encode("utf-8"))
        err = urllib.error.HTTPError(
            hf.HF_CHAT_URL, 400, "Bad Request", hdrs={}, fp=fp
        )
        mock_urlopen.side_effect = err

        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="********",
                image_bytes=b"\xff\xd8\xff",
                user_text="hi",
                model="google/gemma-4-E4B-it",
                timeout_sec=5,
            )
        msg = str(ctx.exception)
        self.assertIn("not a chat model", msg.lower())
        self.assertIn("Conversational", msg)

    @patch("urllib.request.urlopen")
    def test_urlopen_receives_timeout_kwarg(self, mock_urlopen: MagicMock) -> None:
        mock_urlopen.return_value = _fake_urlopen_cm(
            {"choices": [{"message": {"content": "ok"}}]}
        )
        hf._hf_vision_chat(
            token="t",
            image_bytes=b"\xff\xd8\xff",
            user_text="x",
            model="m",
            timeout_sec=77,
        )
        _, kwargs = mock_urlopen.call_args
        self.assertEqual(kwargs.get("timeout"), 77)

    @patch("urllib.request.urlopen")
    def test_urllib_error_wrapped(self, mock_urlopen: MagicMock) -> None:
        import urllib.error

        mock_urlopen.side_effect = urllib.error.URLError("no route")
        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="t",
                image_bytes=b"\xff\xd8\xff",
                user_text="x",
                model="m",
                timeout_sec=5,
            )
        self.assertIn("연결 실패", str(ctx.exception))


class TestHfRequestPayloadContract(unittest.TestCase):
    """Router로 나가는 POST 본문이 OpenAI 호환 멀티모달 형태인지 검증."""

    def setUp(self) -> None:
        self._patcher = patch("urllib.request.urlopen")
        self.mock_urlopen = self._patcher.start()
        self.mock_urlopen.return_value = _fake_urlopen_cm(
            {"choices": [{"message": {"content": "의견: 유지"}}]}
        )

    def tearDown(self) -> None:
        self._patcher.stop()

    def test_request_url_method_headers(self) -> None:
        hf._hf_vision_chat(
            token="********",
            image_bytes=b"\xff\xd8\xff\xe0",
            user_text="hello",
            model="org/model-id",
            timeout_sec=5,
        )
        req = self.mock_urlopen.call_args[0][0]
        self.assertEqual(req.full_url, hf.HF_CHAT_URL)
        self.assertEqual(req.get_method(), "POST")
        self.assertEqual(req.get_header("Content-type"), "application/json")
        self.assertEqual(req.get_header("Accept"), "application/json")
        self.assertEqual(req.get_header("Authorization"), "Bearer secret")

    def test_payload_openai_multimodal_shape(self) -> None:
        hf._hf_vision_chat(
            token="t",
            image_bytes=b"\xff\xd8\xff",
            user_text="프롬프트",
            model="my/vision",
            timeout_sec=5,
        )
        p = _payload_from_mock(self.mock_urlopen)
        self.assertEqual(p["model"], "my/vision")
        self.assertEqual(p["max_tokens"], hf.DEFAULT_HF_MAX_TOKENS)
        self.assertEqual(p["temperature"], 0.2)
        self.assertEqual(len(p["messages"]), 1)
        self.assertEqual(p["messages"][0]["role"], "user")
        parts = p["messages"][0]["content"]
        self.assertEqual(len(parts), 2)
        self.assertEqual(parts[0]["type"], "image_url")
        self.assertIn("url", parts[0]["image_url"])
        self.assertEqual(parts[1]["type"], "text")
        self.assertEqual(parts[1]["text"], "프롬프트")

    def test_jpeg_data_url_roundtrips_bytes(self) -> None:
        raw = b"\xff\xd8\xff\xe0\x00\x10JFIF" + b"\xab" * 32
        hf._hf_vision_chat(
            token="t",
            image_bytes=raw,
            user_text="q",
            model="m",
            timeout_sec=5,
        )
        url = _image_url_from_payload(_payload_from_mock(self.mock_urlopen))
        m = re.match(r"^data:(image/jpeg);base64,(.+)$", url)
        self.assertIsNotNone(m)
        decoded = base64.standard_b64decode(m.group(2))
        self.assertEqual(decoded, raw)

    def test_unicode_user_text_json_safe(self) -> None:
        hf._hf_vision_chat(
            token="t",
            image_bytes=b"\xff\xd8\xff",
            user_text="티커: KRW-비트코인 🔍",
            model="m",
            timeout_sec=5,
        )
        p = _payload_from_mock(self.mock_urlopen)
        self.assertIn("비트코인", p["messages"][0]["content"][1]["text"])


class TestHfResponseEdgeCases(unittest.TestCase):
    def setUp(self) -> None:
        self.p = patch("urllib.request.urlopen")
        self.mock = self.p.start()

    def tearDown(self) -> None:
        self.p.stop()

    def test_top_level_error_in_json_raises(self) -> None:
        self.mock.return_value = _fake_urlopen_cm({"error": {"message": "rate limited"}})
        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="t",
                image_bytes=b"\xff\xd8\xff",
                user_text="q",
                model="m",
                timeout_sec=5,
            )
        self.assertIn("HF API error", str(ctx.exception))

    def test_empty_choices_raises(self) -> None:
        self.mock.return_value = _fake_urlopen_cm({"choices": []})
        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="t",
                image_bytes=b"\xff\xd8\xff",
                user_text="q",
                model="m",
                timeout_sec=5,
            )
        self.assertIn("choices 없음", str(ctx.exception))

    def test_invalid_json_body_raises(self) -> None:
        resp = MagicMock()
        resp.read.return_value = b"not json {"
        cm = MagicMock()
        cm.__enter__.return_value = resp
        self.mock.return_value = cm
        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="t",
                image_bytes=b"\xff\xd8\xff",
                user_text="q",
                model="m",
                timeout_sec=5,
            )
        self.assertIn("JSON 파싱 실패", str(ctx.exception))

    def test_list_content_skips_non_text(self) -> None:
        self.mock.return_value = _fake_urlopen_cm(
            {
                "choices": [
                    {
                        "message": {
                            "content": [
                                {"type": "image_url", "image_url": {"url": "x"}},
                                {"type": "text", "text": "의견: 매수"},
                            ]
                        }
                    }
                ]
            }
        )
        out = hf._hf_vision_chat(
            token="t",
            image_bytes=b"\xff\xd8\xff",
            user_text="q",
            model="m",
            timeout_sec=5,
        )
        self.assertEqual(out, "의견: 매수")

    def test_content_none_raises_with_debug(self) -> None:
        self.mock.return_value = _fake_urlopen_cm(
            {"choices": [{"message": {"content": None}}]}
        )
        with self.assertRaises(RuntimeError) as ctx:
            hf._hf_vision_chat(
                token="t",
                image_bytes=b"\xff\xd8\xff",
                user_text="q",
                model="m",
                timeout_sec=5,
            )
        self.assertIn("텍스트를 찾을 수 없습니다", str(ctx.exception))

    def test_reasoning_content_when_content_null_qwen_style(self) -> None:
        """Qwen3 등: content null, reasoning_content 에 본문."""
        self.mock.return_value = _fake_urlopen_cm(
            {
                "choices": [
                    {
                        "message": {
                            "content": None,
                            "reasoning_content": "의견: 유지\n근거: 단색 차트.",
                        }
                    }
                ]
            }
        )
        out = hf._hf_vision_chat(
            token="t",
            image_bytes=b"\xff\xd8\xff",
            user_text="q",
            model="Qwen/Qwen3.5-397B-A17B",
            timeout_sec=5,
        )
        self.assertIn("유지", out)


class TestDefaultVisionModelEnv(unittest.TestCase):
    def test_default_is_non_empty_string(self) -> None:
        self.assertIsInstance(hf.DEFAULT_VISION_MODEL, str)
        self.assertTrue(len(hf.DEFAULT_VISION_MODEL.strip()) > 0)


if __name__ == "__main__":
    unittest.main()