Convert reasons column to JSONB and update related logic

- Migrate `reasons` column in `stock_analysis_results` to JSONB to enhance data handling and performance.
- Refactor related services and routers to handle JSONB format properly, replacing ad-hoc JSON parsing with a new `_normalize_reasons` utility.
- Update model definitions and data access to reflect column type changes, ensuring backward compatibility where necessary.
- Add comprehensive test coverage for reasons parsing and handling across different workflows.

Author: robin-watcha

alembic/versions/1b7e2a9a0a9d_change_reasons_to_jsonb.py

@@ -0,0 +1,45 @@
+"""Change reasons column to JSONB
+
+Revision ID: 1b7e2a9a0a9d
+Revises: 7cff05b5aa4d
+Create Date: 2025-12-04 12:55:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "1b7e2a9a0a9d"
+down_revision: Union[str, Sequence[str], None] = "7cff05b5aa4d"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Alter reasons to JSONB while preserving existing JSON text data."""
+    op.alter_column(
+        "stock_analysis_results",
+        "reasons",
+        existing_type=sa.Text(),
+        type_=postgresql.JSONB(astext_type=sa.Text()),
+        existing_nullable=True,
+        postgresql_using=(
+            "CASE WHEN reasons IS NULL OR trim(reasons) = '' "
+            "THEN NULL ELSE reasons::jsonb END"
+        ),
+    )
+
+
+def downgrade() -> None:
+    """Revert reasons to TEXT."""
+    op.alter_column(
+        "stock_analysis_results",
+        "reasons",
+        existing_type=postgresql.JSONB(astext_type=sa.Text()),
+        type_=sa.Text(),
+        existing_nullable=True,
+        postgresql_using="reasons::text",
+    )

app/analysis/analyzer.py

@@ -336,10 +336,7 @@ async def _save_json_analysis_to_db(
                 instrument_type=instrument_type
             )
 
-            # 2. 근거를 JSON 문자열로 변환
-            reasons_json = json.dumps(result.reasons, ensure_ascii=False)
-
-            # 3. 분석 결과 저장
+            # 2. 분석 결과 저장
             record = StockAnalysisResult(
                 stock_info_id=stock_info.id,  # 주식 정보와 연결
                 prompt=prompt,
@@ -354,7 +351,7 @@ async def _save_json_analysis_to_db(
                 buy_hope_max=result.price_analysis.buy_hope_range.max,
                 sell_target_min=result.price_analysis.sell_target_range.min,
                 sell_target_max=result.price_analysis.sell_target_range.max,
-                reasons=reasons_json,
+                reasons=result.reasons,
                 detailed_text=result.detailed_text,
             )
             db.add(record)

app/models/analysis.py

@@ -2,6 +2,7 @@
 from sqlalchemy import Column, Integer, String, Text, DateTime, Float, ForeignKey, Boolean
 from sqlalchemy.sql import func
 from sqlalchemy.orm import relationship
+from sqlalchemy.dialects.postgresql import JSONB
 from app.models.base import Base
 
 
@@ -53,7 +54,7 @@ class StockAnalysisResult(Base):
     sell_target_max = Column(Float, nullable=True, comment="매도 목표 범위 최대값")
 
     # 근거 및 상세 분석
-    reasons = Column(Text, nullable=True, comment="분석 근거 (JSON 형태로 저장)")
+    reasons = Column(JSONB, nullable=True, comment="분석 근거 (JSON 형태로 저장)")
     detailed_text = Column(Text, nullable=True, comment="상세 분석 텍스트")
 
     # 원본 프롬프트

app/routers/analysis_json.py

@@ -1,3 +1,4 @@
+import json
 import time
 from typing import List, Optional
 
@@ -38,6 +39,23 @@
 )
 
 
+def _normalize_reasons(raw_reasons) -> List[str]:
+    """Convert stored reasons to a string list, handling JSON/text/null."""
+    if raw_reasons is None:
+        return []
+    if isinstance(raw_reasons, list):
+        return [str(r) for r in raw_reasons]
+    if isinstance(raw_reasons, str):
+        try:
+            parsed = json.loads(raw_reasons)
+            if isinstance(parsed, list):
+                return [str(r) for r in parsed]
+            return [str(parsed)]
+        except Exception:
+            return [raw_reasons]
+    return []
+
+
 @router.get("/", response_class=HTMLResponse)
 async def analysis_json_dashboard(request: Request):
     """JSON 분석 결과 대시보드 페이지"""
@@ -145,7 +163,7 @@ async def get_analysis_results(
                     "buy_hope_max": analysis_result.buy_hope_max,
                     "sell_target_min": analysis_result.sell_target_min,
                     "sell_target_max": analysis_result.sell_target_max,
-                    "reasons": analysis_result.reasons,
+                    "reasons": _normalize_reasons(analysis_result.reasons),
                     "detailed_text": analysis_result.detailed_text,
                     "created_at": analysis_result.created_at.isoformat() if analysis_result.created_at else None,
                 })
@@ -218,14 +236,7 @@ async def get_analysis_detail(
 
     analysis_result, stock_info = row
 
-    # 근거를 JSON에서 파싱
-    import json
-    reasons = []
-    try:
-        if analysis_result.reasons:
-            reasons = json.loads(analysis_result.reasons)
-    except:
-        reasons = []
+    reasons = _normalize_reasons(analysis_result.reasons)
 
     return {
         "id": analysis_result.id,

app/routers/stock_latest.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 from asyncio import AbstractEventLoop
 from typing import List, Optional
 from fastapi import APIRouter, Depends, Request, Query, HTTPException, BackgroundTasks
@@ -17,6 +18,23 @@
 router = APIRouter(prefix="/stock-latest", tags=["Stock Latest Analysis"])
 
 
+def _normalize_reasons(raw_reasons) -> List[str]:
+    """Convert stored reasons to a string list, handling JSON/text/null."""
+    if raw_reasons is None:
+        return []
+    if isinstance(raw_reasons, list):
+        return [str(r) for r in raw_reasons]
+    if isinstance(raw_reasons, str):
+        try:
+            parsed = json.loads(raw_reasons)
+            if isinstance(parsed, list):
+                return [str(r) for r in parsed]
+            return [str(parsed)]
+        except Exception:
+            return [raw_reasons]
+    return []
+
+
 @router.get("/", response_class=HTMLResponse)
 async def stock_latest_dashboard(request: Request):
     """종목별 최신 분석 결과 대시보드 페이지"""
@@ -173,14 +191,7 @@ async def get_stock_analysis_history(
     # 응답 데이터 구성
     history_results = []
     for analysis in analysis_history:
-        # 근거를 JSON에서 파싱
-        import json
-        reasons = []
-        try:
-            if analysis.reasons:
-                reasons = json.loads(analysis.reasons)
-        except:
-            reasons = []
+        reasons = _normalize_reasons(analysis.reasons)
 
         history_results.append({
             "id": analysis.id,

app/tasks/kis.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import logging
 from typing import Any, Callable, Dict, List, Optional
 
@@ -110,7 +111,23 @@ async def _send_toss_recommendation_async(
 
         decision = analysis.decision.lower() if analysis.decision else "hold"
         confidence = analysis.confidence if analysis.confidence else 0
-        reasons = analysis.reasons if analysis.reasons else []
+        raw_reasons = analysis.reasons
+        if isinstance(raw_reasons, list):
+            reasons = [str(r) for r in raw_reasons]
+        elif isinstance(raw_reasons, str):
+            try:
+                parsed = json.loads(raw_reasons)
+                reasons = [str(r) for r in parsed] if isinstance(parsed, list) else [str(parsed)]
+            except Exception as parse_error:
+                logger.debug(
+                    "Failed to parse analysis reasons for %s(%s): %s",
+                    name,
+                    code,
+                    parse_error,
+                )
+                reasons = [raw_reasons]
+        else:
+            reasons = []
 
         # AI 결정과 무관하게 항상 가격 제안 알림 발송
         await notifier.notify_toss_price_recommendation(

tests/test_kis_manual_holdings_integration.py

@@ -3,6 +3,7 @@
 
 수동 잔고(토스 등)를 자동화 태스크에 통합하는 기능 테스트
 """
+import json
 import pytest
 from decimal import Decimal
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -726,6 +727,62 @@ async def get_latest_analysis_by_symbol(self, symbol):
         assert notification_sent[0]["appropriate_buy_min"] == 23000.0
         assert notification_sent[0]["appropriate_sell_max"] == 28000.0
 
+    @pytest.mark.asyncio
+    async def test_send_toss_price_recommendation_parses_json_reasons(self, monkeypatch):
+        """DB에 JSON 문자열로 저장된 근거를 리스트로 파싱해 전달"""
+        from app.tasks.kis import _send_toss_recommendation_async
+        from app.models.analysis import StockAnalysisResult
+
+        notification_sent = []
+
+        class MockNotifier:
+            _enabled = True
+
+            async def notify_toss_price_recommendation(self, **kwargs):
+                notification_sent.append(kwargs)
+                return True
+
+        mock_analysis = MagicMock(spec=StockAnalysisResult)
+        mock_analysis.decision = "buy"
+        mock_analysis.confidence = 70
+        mock_analysis.reasons = json.dumps(
+            ["첫째 근거", "둘째 근거", "셋째 근거"], ensure_ascii=False
+        )
+        mock_analysis.appropriate_buy_min = 23000.0
+        mock_analysis.appropriate_buy_max = 24000.0
+        mock_analysis.appropriate_sell_min = 26000.0
+        mock_analysis.appropriate_sell_max = 28000.0
+        mock_analysis.buy_hope_min = 22000.0
+        mock_analysis.buy_hope_max = 23000.0
+        mock_analysis.sell_target_min = 28000.0
+        mock_analysis.sell_target_max = 30000.0
+
+        class MockAnalysisService:
+            def __init__(self, db):
+                pass
+
+            async def get_latest_analysis_by_symbol(self, symbol):
+                return mock_analysis
+
+        mock_db_session = MagicMock()
+        mock_db_session.__aenter__ = AsyncMock(return_value=MagicMock())
+        mock_db_session.__aexit__ = AsyncMock(return_value=None)
+
+        with patch('app.tasks.kis.get_trade_notifier', return_value=MockNotifier()), \
+             patch('app.core.db.AsyncSessionLocal', return_value=mock_db_session), \
+             patch('app.services.stock_info_service.StockAnalysisService', MockAnalysisService):
+
+            await _send_toss_recommendation_async(
+                code="015760",
+                name="한국전력",
+                current_price=25000.0,
+                toss_quantity=10,
+                toss_avg_price=23000.0,
+            )
+
+        assert len(notification_sent) == 1
+        assert notification_sent[0]["reasons"] == ["첫째 근거", "둘째 근거", "셋째 근거"]
+
     def test_format_toss_price_recommendation_html_escapes_special_chars(self):
         """HTML 포맷 메시지가 특수문자를 올바르게 이스케이프하는지 확인"""
         from app.monitoring.trade_notifier import TradeNotifier