o2o-negosium-original/agent/negotiation/policy/autonomy_store.py

"""AutonomyStore — 완전 자율 협상 정책 서빙 (룰 대체, numpy 전용).

AUTONOMY_MODE=1 이면 가격협상 판정 룰(앵커 이하 타결 / 와일드카드 존 / 라운드 상한)과
카드 선택을 전부 이 정책의 행동 결정으로 대체한다:
  accept  → 협상완료 (제시가 타결)          walk → 협상실패
  counter → "C원이면 수락" 역제안 스텝       press → 전략별 압박 멘트 스텝

행동의 유일한 유인은 보상 함수다. 남는 제한은 두 가지뿐이며 비즈니스 룰이 아니다:
  - 역제안 후보 격자가 [anchor−5%span, target] 안 (행동 공간 정의)
  - 세션 턴 상한(엔지니어링 타임아웃, ChatEngine._AUTONOMY_TURN_CAP)

번들: artifacts/autonomy_serving.npz (tools/export_autonomy_serving.py).
불가(플래그 꺼짐/번들 없음)면 None → 기존 룰 엔진 그대로 (즉시 롤백 경로).
"""

import os
from typing import Optional

import numpy as np

from common.logger import LOG
from negotiation.policies.autonomy_actions import (
    ACTIONS, Action, extra_state, internet_gap_feat, settle_norm)
from negotiation.qtable.domain.model.snapshot import NegotiationSnapshot
from negotiation.qtable.domain.service.feature_builder import (
    build_state_features, build_tenant_features)

_HERE = os.path.dirname(os.path.abspath(__file__))
BUNDLE_PATH = os.path.join(_HERE, "..", "..", "artifacts", "autonomy_serving.npz")


class AutonomyPolicy:
    """세션 컨텍스트 → 상태특징 → 행동(greedy). ChatEngine 에 decider 로 주입된다."""

    name = "full_autonomy"

    def __init__(self, z, reward_cfg):
        self._W = (z["W0"], z["b0"], z["W1"], z["b1"], z["W2"], z["b2"])
        self._state_dim = int(z["state_dim"])
        self._tenant_feat = build_tenant_features(reward_cfg)

    @staticmethod
    def _acceptance(ctx: dict) -> float:
        base = ctx.get("item_price") or ctx.get("first_offer_price") or 0
        cur = ctx.get("input_price") or 0
        if base <= 0 or cur <= 0:
            return 0.0
        return max(0.0, (base - cur) / base)

    def decide(self, ctx: dict) -> Action:
        """ChatSession.context → Action. 상태 구성은 ChatService._snapshot 과 동일 규칙."""
        snap = NegotiationSnapshot(
            revenue_amount=ctx["revenue_amount"], distribution_code=ctx["distribution_code"],
            partner_count=ctx["partner_count"], acceptance_ratio=self._acceptance(ctx),
            input_price=ctx.get("input_price", ctx["anchor_price"]), anchor_price=ctx["anchor_price"],
            target_price=ctx["target_price"], round_number=ctx.get("round", 0),
        )
        # v3 추가 특징: 직전 역제안 기억 + 마감 잔여율 + 협력사 이력 + 인터넷최저가 갭.
        # 소스가 없으면 전부 중립값(0.5/0) — 학습 시뮬의 '미상' 표현과 동일해야 한다.
        # 역제안 기억은 autonomy_last_counter(역제안만 갱신) — autonomy_last(마지막 행동)를 쓰면
        # 사이에 낀 설득이 기억을 지워 단조 봉투가 뚫린다(counter→press→counter 철회 실버그).
        # 시뮬의 last_kind/last_q 도 역제안만 추적하므로 이쪽이 학습 분포와도 일치한다.
        last = ctx.get("autonomy_last_counter") or {}
        deadline = 0.5
        end_ts, total_s = ctx.get("deadline_end_ts"), ctx.get("deadline_total_s")
        if end_ts and total_s:
            import time
            deadline = float(np.clip((end_ts - time.time()) / total_s, 0.0, 1.0))
        hist_n = int(ctx.get("hist_n") or 0)
        hist_success = float(ctx["hist_success"]) if ctx.get("hist_success") is not None else 0.5
        hist_settle = (settle_norm(float(ctx["hist_settle_ratio"]))
                       if ctx.get("hist_settle_ratio") is not None else 0.5)
        sf = np.concatenate([build_state_features(snap), self._tenant_feat, extra_state(
            last.get("kind", ""), float(last.get("q", 0.0)),
            deadline=deadline, hist_n=min(hist_n, 5) / 5.0,
            hist_success=hist_success if hist_n else 0.5,
            hist_settle=hist_settle,
            internet_gap=internet_gap_feat(float(ctx.get("internet_lowest_price") or 0),
                                           float(snap.anchor_price)),
        )])
        span = max(snap.target_price - snap.anchor_price, 1.0)
        pos = (snap.input_price - snap.anchor_price) / span
        # 행동 봉투 (학습 available_actions 와 동일해야 한다):
        #  ① 목표가 초과 제시가는 '수락' 제외 — 매입 승인 범위(v3.1 착취 방지)
        #  ② 직전 역제안보다 낮은 금액의 역제안 제외 — 단조 양보 원칙(제안 철회는 협상 예절 위반;
        #     올리는 '속도'는 정책 학습, 후퇴 '금지'만 구조로 보장)
        #  ③ 역제시 해금 조건 — 옛 제품 의미론 복원(제품 결정 2026-07-10): 일반 카드는 설득만,
        #     역제시(숫자 제안)는 와일드카드처럼 마무리 수단. 최소 AUTONOMY_MIN_PRESS(기본 2)회
        #     설득 이후에만 역제시 후보가 열린다. 해금 후의 타이밍·금액은 정책 학습.
        #  ④ 마무리 국면 — 제시가가 목표가 0.5% 이내로 붙으면 압박 제외(+역제시 잠금 해제):
        #     푼돈 차이에서 '재검토 부탁' 반복은 상대만 지치게 한다. 클로징(역제안/최종제안)하거나 끝내거나.
        min_press = int(os.getenv("AUTONOMY_MIN_PRESS", "2"))
        near_target = snap.input_price <= snap.target_price * 1.005
        counter_locked = (int(ctx.get("autonomy_press_n") or 0) < min_press) and not near_target
        last_counter_q = float(last["q"]) if last.get("kind") == "counter" else None
        if last_counter_q is not None:
            counter_locked = False   # 이미 역제시를 시작했으면 잠그지 않는다(단조 봉투가 관리)
        #  ⑤ 첫 역제안은 앵커가 이하(q ≤ 0)만 — 낮게 개시해 목표가까지 천천히 올라간다
        #     (제품 결정: 사다리를 다 쓰는 앵커링 개시. 이후 단조 봉투가 상향을 관리).
        #  ⑥ 결렬(walk)도 해금 전 금지 — 설득 0회에 walk 를 고르면 최종제안 보장(엔진)과 결합해
        #     '첫 턴 목표가 통보'가 된다(v3.4 라이브 결함). 해금 전에는 설득만 가능.
        cands = [a for a in ACTIONS
                 if not (a.kind == "accept" and snap.input_price > snap.target_price)
                 and not (a.kind == "counter" and counter_locked)
                 and not (a.kind == "walk" and counter_locked)
                 and not (a.kind == "press" and near_target)
                 and not (a.kind == "counter" and last_counter_q is None and a.counter_q > 1e-9)
                 and not (a.kind == "counter" and last_counter_q is not None
                          and a.counter_q < last_counter_q - 1e-9)]
        feats = []
        for a in cands:
            cut = 0.0
            if a.kind == "counter":
                c = snap.anchor_price + a.counter_q * span
                cut = max(0.0, (snap.input_price - c) / max(snap.input_price, 1.0))
            feats.append(a.feat(pos, cut))
        feats = np.stack(feats)
        W0, b0, W1, b1, W2, b2 = self._W
        x = np.concatenate([np.repeat(sf[None, :], feats.shape[0], axis=0), feats], axis=1)
        h = np.maximum(x @ W0.T + b0, 0.0)
        h = np.maximum(h @ W1.T + b1, 0.0)
        scores = (h @ W2.T + b2).squeeze(-1)
        return cands[int(np.argmax(scores))]

    @staticmethod
    def counter_price(ctx: dict, act: Action) -> int:
        span = max(ctx["target_price"] - ctx["anchor_price"], 1.0)
        return int(round(ctx["anchor_price"] + act.counter_q * span))


class AutonomyStore:
    _z = None
    _load_failed = False

    @classmethod
    def enabled(cls) -> bool:
        return os.getenv("AUTONOMY_MODE", "0").lower() in ("1", "true", "yes")

    @classmethod
    def policy_for(cls, engine) -> Optional[AutonomyPolicy]:
        """engine: tenancy.registry.TenantEngine. 비활성/번들 없음 → None (룰 엔진 유지)."""
        if not cls.enabled() or cls._load_failed:
            return None
        if cls._z is None:
            try:
                cls._z = np.load(BUNDLE_PATH, allow_pickle=False)
                LOG.i("[Autonomy] 완전 자율 정책 번들 로드 완료 — 협상 판정 룰 대체 모드")
            except Exception as ex:
                cls._load_failed = True
                LOG.e_no_callstack(f"[Autonomy] 번들 로드 실패 → 룰 엔진 유지: {ex}")
                return None
        return AutonomyPolicy(cls._z, engine.config.reward)