o2o-negosium-original/negodata/backend/services/learning_service.py

import uuid

from fastapi import Depends

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import quotations
from common.anchoring.constants import UPPER_BOUNDS
from common.enums import DBWRType, ErrorType, SupplierType
from crud.learning_crud import ILearningCRUD, LearningCRUD
from router.v1.learning.protocol import (
    AnchoringCell,
    AnchoringHistoryRow,
    CardUsageRow,
    LearningKpi,
    Res_AnchoringStatus,
    Res_LearningStatus,
)

HISTORY_LIMIT = 50  # 앵커링 조정 이력 표시 개수 — 한 화면에서 훑는 용도

def _price_range_label(index: int) -> str:
    """가격대 인덱스 → 실제 금액 범위. 사다리는 common.anchoring 정본(UPPER_BOUNDS)만 참조한다."""
    if index < 0 or index >= len(UPPER_BOUNDS):
        return ""
    low = UPPER_BOUNDS[index - 1] if index else 0
    return f"{low:,}~{UPPER_BOUNDS[index]:,}원"


_SUPPLIER_TYPE_LABEL = {
    SupplierType.NONE.value: "미지정",
    SupplierType.DISTRIBUTION.value: "유통",
    SupplierType.MANUFACTURE.value: "제조",
    SupplierType.SOLE_AGENCY.value: "총판",
}


class LearningService:
    """협상 학습 현황 — 협상카드 학습(agent Q-learning)과 앵커링 조정 이력을 읽어 보여준다.

    두 값 모두 negodata 가 만드는 값이 아니라 agent·anchoring 서비스가 쌓은 결과다(읽기 전용).
    """

    def __init__(self, crud: ILearningCRUD = Depends(LearningCRUD)):
        self.crud = crud

    async def get_learning_status(self, company_id: str) -> Res_LearningStatus:
        """카드 사용 현황 — 담아둔 카드가 실제로 나가는지, 어느 국면에 나가는지, 쏠리지는 않는지.

        협상 성과(타결·가격)는 카드별로 나누지 않는다 — 한 협상에 여러 장이 나가 어느 장의 몫인지
        가릴 수 없고, 카드 배정도 무작위가 아니라 국면에 따라 정해지기 때문이다.
        """
        res = Res_LearningStatus()
        # learning 의 company_id 는 agent 가 테넌트 키를 그대로 넣는 문자열 컬럼이다(UUID 타입 아님).
        cid = str(company_id)

        sessions, records, _settled, last_at = await self._read(
            lambda s: self.crud.learning_summary(s, cid), default=(0, 0, 0, None))
        rows = await self._read(lambda s: self.crud.card_usage(s, cid), default=[])
        names = await self._read(lambda s: self.crud.card_names(s), default=[])
        name_map = {str(number): (name, is_wild, card_pk) for number, name, is_wild, card_pk in names if number}

        total_uses = sum(int(r[2] or 0) for r in rows) or 1
        for card_id, used_sessions, uses, avg_turn, last_used, avg_q, avg_ucb, visits in rows:
            number = str(card_id)
            name, is_wild, card_pk = name_map.get(number, (None, 0, None))
            res.cards.append(CardUsageRow(
                number=number,
                name=name,
                card_id=str(card_pk) if card_pk else None,
                type="wild" if is_wild else "nego",
                used_sessions=int(used_sessions or 0),
                uses=int(uses or 0),
                share=round(int(uses or 0) / total_uses, 3),
                avg_turn=round(float(avg_turn), 1) if avg_turn is not None else 0.0,
                last_used_at=last_used,
                avg_q=round(float(avg_q), 4) if avg_q is not None else None,
                avg_ucb=round(float(avg_ucb), 4) if avg_ucb is not None else None,
                # 탐색보너스는 두 값이 다 있을 때만 — 한쪽만 있으면 차이가 의미를 잃는다.
                explore_bonus=(round(float(avg_ucb) - float(avg_q), 4)
                               if avg_q is not None and avg_ucb is not None else None),
                visits=int(visits or 0),
            ))

        res.kpi = LearningKpi(
            learned_sessions=int(sessions or 0),
            records=int(records or 0),
            used_cards=len(res.cards),
            unused_cards=max(0, len(name_map) - len(res.cards)),
            top3_share=round(sum(c.uses for c in res.cards[:3]) / total_uses, 3),
            last_learned_at=last_at,
        )
        return res

    async def get_anchoring_status(self, company_id: str) -> Res_AnchoringStatus:
        res = Res_AnchoringStatus()
        # anchoring 스키마의 company_id 는 UUID 타입 — learning(문자열 테넌트 키)과 다르다.
        cid = uuid.UUID(str(company_id))

        cells = await self._read(lambda s: self.crud.anchoring_current(s, cid), default=[])
        for supplier_type, price_range_index, value, adjusted_at in cells:
            res.cells.append(AnchoringCell(
                supplier_type=int(supplier_type or 0),
                supplier_type_label=_SUPPLIER_TYPE_LABEL.get(int(supplier_type or 0), "미지정"),
                price_range_index=int(price_range_index or 0),
                price_range_label=_price_range_label(int(price_range_index or 0)),
                anchoring_value=float(value) if value is not None else 0.0,
                last_adjusted_at=adjusted_at,
            ))

        history = await self._read(lambda s: self.crud.anchoring_history(s, cid, HISTORY_LIMIT), default=[])
        for st, pri, before, after, sample, success, rate, created_at in history:
            res.history.append(AnchoringHistoryRow(
                supplier_type_label=_SUPPLIER_TYPE_LABEL.get(int(st or 0), "미지정"),
                price_range_index=int(pri or 0),
                price_range_label=_price_range_label(int(pri or 0)),
                value_before=float(before) if before is not None else 0.0,
                value_after=float(after) if after is not None else 0.0,
                sample_count=int(sample or 0),
                success_count=int(success or 0),
                success_rate=round(float(rate), 3) if rate is not None else 0.0,
                created_at=created_at,
            ))
        res.adjusted_count = len(res.history)
        return res

    async def _read(self, fn, default):
        """crud 한 건 실행 — 실패해도 화면은 떠야 하므로 기본값으로 떨어진다."""
        err, rows = await DB_SESSION_MNG.execute_lambda(quotations.DBType(), DBWRType.DB_READ.value, fn)
        return rows if err == ErrorType.SUCCESS else default