upbit-trader/archive/tests/test_llm_advisor.py

"""core/llm_advisor 단위 테스트.

실행:
    .venv/bin/python3 -m pytest tests/test_llm_advisor.py -v
"""
import json
import sys
import os
from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

os.environ.setdefault('ORACLE_USER', 'x')
os.environ.setdefault('ORACLE_PASSWORD', 'x')
os.environ.setdefault('ORACLE_DSN', 'x')
os.environ.setdefault('ANTHROPIC_API_KEY', 'test-key')

from core.llm_advisor import (
    get_exit_price,
    _describe_bars,
    _build_prompt,
    _execute_tool,
)

TICKER = 'KRW-XRP'


def _make_bars(n: int = 20, base: float = 2100.0) -> list[dict]:
    now = datetime.now()
    bars = []
    for i in range(n):
        p = base + i * 0.5
        bars.append({
            'open':   p,
            'high':   p + 2,
            'low':    p - 2,
            'close':  p + 1,
            'volume': 100.0,
            'ts':     now - timedelta(seconds=(n - i) * 20),
        })
    return bars


def _make_pos(entry: float = 2084.0, seconds_ago: int = 120,
              sell_price: float = 2126.0) -> dict:
    return {
        'entry_price': entry,
        'entry_ts':    datetime.now() - timedelta(seconds=seconds_ago),
        'sell_price':  sell_price,
        'sell_uuid':   'uuid-1',
        'qty':         2399.0,
        'stage':       0,
        'llm_last_ts': None,
    }


def _mock_response(action: str, price: float = 0):
    """Anthropic API 응답 Mock."""
    if action == 'hold':
        text = '{"action": "hold"}'
    else:
        text = json.dumps({'action': 'sell', 'price': price})

    content_block = MagicMock()
    content_block.type = 'text'
    content_block.text = text

    response = MagicMock()
    response.content      = [content_block]
    response.stop_reason  = 'end_turn'
    return response


# ── _describe_bars ────────────────────────────────────────────────────────────

class TestDescribeBars:

    def test_returns_string(self):
        bars = _make_bars(20, base=2100.0)
        desc = _describe_bars(bars, current_price=2110.0)
        assert isinstance(desc, str)
        assert '패턴 요약' in desc

    def test_empty_bars_returns_fallback(self):
        desc = _describe_bars([], current_price=2100.0)
        assert desc == '봉 데이터 없음'

    def test_trend_direction_detected(self):
        bars = _make_bars(20, base=2100.0)    # 상승 bars (base + i*0.5)
        desc = _describe_bars(bars, current_price=2110.0)
        assert '상승▲' in desc


# ── get_exit_price: hold ──────────────────────────────────────────────────────

class TestGetExitPriceHold:

    def test_returns_none_on_hold(self):
        pos      = _make_pos()
        bar_list = _make_bars()
        with patch('anthropic.Anthropic') as MockClient:
            MockClient.return_value.messages.create.return_value = _mock_response('hold')
            result = get_exit_price(TICKER, pos, bar_list, current_price=2109.0)
        assert result is None

    def test_returns_none_when_no_api_key(self):
        pos      = _make_pos()
        bar_list = _make_bars()
        with patch.dict(os.environ, {'ANTHROPIC_API_KEY': ''}):
            result = get_exit_price(TICKER, pos, bar_list, current_price=2109.0)
        assert result is None


# ── get_exit_price: sell ─────────────────────────────────────────────────────

class TestGetExitPriceSell:

    def test_returns_llm_suggested_price(self):
        pos      = _make_pos(entry=2084.0, sell_price=2126.0)
        bar_list = _make_bars()
        with patch('anthropic.Anthropic') as MockClient:
            MockClient.return_value.messages.create.return_value = _mock_response('sell', 2112.0)
            result = get_exit_price(TICKER, pos, bar_list, current_price=2109.0)
        assert result == 2112.0

    def test_llm_can_suggest_below_current_price(self):
        """LLM의 판단을 신뢰 — 현재가 이하 제안도 그대로 반환."""
        pos      = _make_pos(entry=2084.0, sell_price=2126.0)
        bar_list = _make_bars()
        with patch('anthropic.Anthropic') as MockClient:
            MockClient.return_value.messages.create.return_value = _mock_response('sell', 2080.0)
            result = get_exit_price(TICKER, pos, bar_list, current_price=2109.0)
        assert result == 2080.0  # 가드 없음 — LLM 신뢰

    def test_llm_can_suggest_high_price(self):
        """LLM의 판단을 신뢰 — 진입가 대비 10% 높은 제안도 그대로 반환."""
        pos      = _make_pos(entry=2084.0, sell_price=2126.0)
        bar_list = _make_bars()
        with patch('anthropic.Anthropic') as MockClient:
            MockClient.return_value.messages.create.return_value = _mock_response('sell', 2300.0)
            result = get_exit_price(TICKER, pos, bar_list, current_price=2109.0)
        assert result == 2300.0  # 상한 가드 없음 — LLM 신뢰


# ── get_exit_price: 오류 처리 ─────────────────────────────────────────────────

class TestGetExitPriceErrors:

    def test_returns_none_on_json_error(self):
        """JSON 파싱 실패 → None (cascade fallback)."""
        pos      = _make_pos()
        bar_list = _make_bars()
        bad_resp = MagicMock()
        bad_resp.content     = [MagicMock(type='text', text='not json')]
        bad_resp.stop_reason = 'end_turn'
        with patch('anthropic.Anthropic') as MockClient:
            MockClient.return_value.messages.create.return_value = bad_resp
            result = get_exit_price(TICKER, pos, bar_list, current_price=2109.0)
        assert result is None

    def test_returns_none_on_api_exception(self):
        """API 오류 → None (cascade fallback)."""
        pos      = _make_pos()
        bar_list = _make_bars()
        with patch('anthropic.Anthropic') as MockClient:
            MockClient.return_value.messages.create.side_effect = Exception('API Error')
            result = get_exit_price(TICKER, pos, bar_list, current_price=2109.0)
        assert result is None


# ── tool 실행 ─────────────────────────────────────────────────────────────────

class TestExecuteTool:

    def test_unknown_tool_returns_error_string(self):
        result = _execute_tool('unknown_tool', {'ticker': TICKER})
        assert '알 수 없는 tool' in result

    def test_get_price_ticks_db_error_returns_string(self):
        """DB 연결 실패 시 오류 문자열 반환 (예외 아님)."""
        result = _execute_tool('get_price_ticks', {'ticker': TICKER, 'minutes': 5})
        assert isinstance(result, str)

    def test_get_ohlcv_db_error_returns_string(self):
        result = _execute_tool('get_ohlcv', {'ticker': TICKER, 'limit': 10})
        assert isinstance(result, str)