"""기존 Q-Table(UCB) vs action-as-feature DQN 공정 비교 — 고객사 성향 조건화 환경 (최종). 같은 환경(FeatureBuyer 2축 + 협력사·고객사성향 랜덤)에서 동일 에피소드로 학습·평가. - Q-Table: 이산 state 162칸 + 카드=슬롯. 성향(고객사) 입력 자체가 불가능 → 평균 성향에 수렴 - DQN : 연속 상태 + 성향 벡터 + 카드 특징(임베딩+전략/톤 one-hot) 평가 4종: ① 학습 카드 9장 — 평균보상(진짜 목적함수) + top3 적중(MC 정답 기준) ② zero-shot 11장 — 안 본 카드 2장 포함 ③ 새 카드 첫 턴 사용률 — 구조적 차이 ④ 성향 극단 테스트 — 같은 협력사, 성향만 바꿨을 때 카드를 바꾸는가 실행: APP_ENV=local python -m tools.compare_qtable_vs_dqn """ import random import numpy as np import torch from eval_harness.buyer import Scenario from eval_harness.feature_buyer import FeatureBuyer, SupplierProfile, sample_supplier from negotiation.policies.feature_dqn_policy import FeatureDQNPolicy from negotiation.policies.qtable_policy import UCBQTablePolicy from negotiation.policies.base import EpisodeState, PolicyContext, Transition from negotiation.qtable.domain.model.q_table import QTable from negotiation.qtable.domain.model.snapshot import NegotiationOutcome from negotiation.qtable.domain.service.feature_builder import ( STATE_FEATURE_DIM, TENANT_FEATURE_DIM, build_state_features) from negotiation.qtable.domain.service.reward_calculator import RewardCalculator from negotiation.qtable.domain.service.state_calculator import state_index from tenancy.config_loader import TenantConfigLoader from tools.train_feature_dqn import ( ANCHOR, HOLDOUT, MAX_TURNS, TARGET, load_cards, make_snapshot, pref_config, sample_tenant_pref) # ---- 정책 어댑터 ------------------------------------------------------------------ class DQNAdapter: name = "feature_dqn" def __init__(self, policy, feat): self.p, self.feat = policy, feat def _sf(self, snap, tf): return np.concatenate([build_state_features(snap), tf]) def choose(self, snap, tf, avail, greedy): self.p.greedy = greedy i, _, _ = self.p.select(self._sf(snap, tf), np.stack([self.feat[c] for c in avail])) return avail[i] def learn(self, snap, tf, card, reward, next_snap, next_avail, done): sf = self._sf(snap, tf) if done or next_snap is None: self.p.remember(sf, self.feat[card], reward, None, None, True) else: self.p.remember(sf, self.feat[card], reward, self._sf(next_snap, tf), np.stack([self.feat[c] for c in next_avail]), False) self.p.train_step() class QTableAdapter: """기존 UCBQTablePolicy. 성향(tf)은 구조상 받을 수 없다 — 이산 state 162칸에 그 축이 없음.""" name = "qtable_ucb" def __init__(self, all_numbers, state_cfg, lr=0.1, gamma=0.95): self.numbers = list(all_numbers) self.a_of = {n: i for i, n in enumerate(self.numbers)} self.state_cfg = state_cfg self.qt = QTable(162, len(self.numbers), learning_rate=lr, discount_factor=gamma) self.pol = UCBQTablePolicy(self.qt) def choose(self, snap, tf, avail, greedy): idx = state_index(snap, self.state_cfg) if greedy: q = self.qt.row(idx) return max(avail, key=lambda c: q[self.a_of[c]]) mask = np.zeros(len(self.numbers), dtype=bool) for c in avail: mask[self.a_of[c]] = True ctx = PolicyContext(state_index=idx, snapshot=snap, action_space_size=len(self.numbers), episode=EpisodeState(), available_mask=mask) return self.numbers[self.pol.select(ctx).action_id] def learn(self, snap, tf, card, reward, next_snap, next_avail, done): idx = state_index(snap, self.state_cfg) nidx = state_index(next_snap, self.state_cfg) if (next_snap is not None and not done) else None self.pol.update(Transition(state_index=idx, action_id=self.a_of[card], reward=reward, next_state_index=nidx, done=done)) class RandomAdapter: name = "random" def __init__(self, seed=0): self.rng = np.random.default_rng(seed) def choose(self, snap, tf, avail, greedy): return avail[self.rng.integers(len(avail))] def learn(self, *a, **k): pass # ---- 공용 에피소드 ----------------------------------------------------------------- def run_episode(adapter, sup, tf, pool, strat, rc, seed, learn=True, greedy=False, forced_first=None): buyer = FeatureBuyer(sup, strat, seed=seed, max_turns=MAX_TURNS) scenario = Scenario(anchor_price=ANCHOR, target_price=TARGET, revenue_amount=sup.revenue_amount, distribution_code=sup.distribution_code, partner_count=sup.partner_count) price0 = TARGET * 1.15 price, used, total_r, first_card = price0, set(), 0.0, None for turn in range(1, MAX_TURNS + 1): acceptance = max(0.0, (price0 - price) / price0) snap = make_snapshot(sup, price, turn, acceptance) avail = [c for c in pool if c not in used] or list(pool) if turn == 1 and forced_first is not None: card = forced_first else: card = adapter.choose(snap, tf, avail, greedy) used.add(card) if first_card is None: first_card = card resp = buyer.respond(card, scenario, turn, price) price = resp.new_price done = resp.accept or price <= ANCHOR or turn >= MAX_TURNS success = resp.accept or price <= ANCHOR outcome = (NegotiationOutcome.SUCCESS if success else NegotiationOutcome.FAILURE if done else NegotiationOutcome.ONGOING) # 채점은 최종 결과 시점만 (중간 턴 0 → γ 부트스트랩으로 전파). # 진행 중 보상을 누적하면 '질질 끄는 전략'이 부당하게 유리해지는 인공물이 생긴다. r = rc.calculate(make_snapshot(sup, price, turn, acceptance, outcome)).total if done else 0.0 total_r += r if learn: if done: adapter.learn(snap, tf, card, r, None, None, True) else: acc2 = max(0.0, (price0 - price) / price0) nsnap = make_snapshot(sup, price, turn + 1, acc2) navail = [c for c in pool if c not in used] or list(pool) adapter.learn(snap, tf, card, r, nsnap, navail, False) if done: return total_r, success, price, first_card return total_r, False, price, first_card # ---- MC 정답 랭킹: 이 (협력사, 성향)에서 진짜 좋은 첫 카드 top-k --------------------- _rand = RandomAdapter(seed=1) def rank_cards_mc(sup, tf, pool, strat, rc, seed, sims=6, k=3): means = {} for c in pool: rs = [run_episode(_rand, sup, tf, pool, strat, rc, seed=seed + 17 * s, learn=False, greedy=False, forced_first=c)[0] for s in range(sims)] means[c] = np.mean(rs) return sorted(means, key=lambda c: -means[c])[:k] # ---- 학습/평가 --------------------------------------------------------------------- def train(adapter, pool, strat, base_reward, state_cfg, episodes, seed): rng = np.random.default_rng(seed) for ep in range(1, episodes + 1): sup = sample_supplier(rng) rcfg, tf = sample_tenant_pref(rng, base_reward) rc = RewardCalculator(rcfg, state_cfg) run_episode(adapter, sup, tf, pool, strat, rc, seed=seed * 100 + ep, learn=True) def evaluate(adapter, pool, strat, base_reward, state_cfg, n=300, seed0=777, label=""): from negotiation.qtable.domain.service.feature_builder import build_tenant_features rng = np.random.default_rng(seed0) rewards, succ, ratios, hits, holdout_first = [], 0, [], 0, 0 for i in range(n): sup = sample_supplier(rng) rcfg, tf = sample_tenant_pref(rng, base_reward) rc = RewardCalculator(rcfg, state_cfg) good = rank_cards_mc(sup, tf, pool, strat, rc, seed=seed0 * 7 + i) r, ok, price, first = run_episode(adapter, sup, tf, pool, strat, rc, seed=seed0 * 1000 + i, learn=False, greedy=True) rewards.append(r); succ += ok; ratios.append(price / TARGET) hits += (first in good); holdout_first += (first in HOLDOUT) m, ci = float(np.mean(rewards)), float(1.96 * np.std(rewards) / np.sqrt(n)) print(f"{label:<14} mean_rwd={m:.4f} ±{ci:.4f} success={succ/n:.3f} " f"settled/tgt={np.mean(ratios):.3f} top3_hit={hits/n:.3f} 새카드첫턴={holdout_first/n:.3f}") def pref_behavior_test(adapters, pool, strat, base_reward, state_cfg): """④ 같은 협력사, 성향만 바꿨을 때 카드를 바꾸는가 (greedy). 첫 턴은 '일단 깎기'가 공통 정답이라 성향 차이가 잘 안 드러난다. → 협상 중반(가격이 이미 target 근처, 3턴째) 상태를 함께 프로브: 여기서 성사중시는 '마무리(수락 잘 되는) 카드', 가격중시는 '더 깎는 카드'가 갈려야 한다. """ from negotiation.qtable.domain.service.feature_builder import build_tenant_features sups = [SupplierProfile(5_000_000, 3, "A"), # 소형·경쟁多 SupplierProfile(200_000_000, 1, "A")] # 대형·단독 probes = [("첫턴", TARGET * 1.15, 1, 0.0), ("중반(3턴,가격↓)", TARGET * 1.02, 3, 0.11)] for pr_name, price, turn, acc in probes: print(f"\n ── 프로브: {pr_name} (price={price:.0f}) ──") print(f" {'협력사':<13} {'성향':<9} " + " ".join(f"{a.name:<15}" for a in adapters)) for sup in sups: row = {} for p, pname in [(0.05, "성사중시"), (0.95, "가격중시")]: rcfg = pref_config(base_reward, p) tf = build_tenant_features(rcfg) picks = [] for a in adapters: snap = make_snapshot(sup, price, turn, acc) picks.append(a.choose(snap, tf, pool, True)) seg = f"{sup.segment[0]}·{sup.segment[1]}" print(f" {seg:<13} {pname:<9} " + " ".join(f"{c}(전략{strat[c]})".ljust(15) for c in picks)) def main(episodes=10000, seed=42): random.seed(seed); np.random.seed(seed); torch.manual_seed(seed) numbers, feat, strat = load_cards() train_pool = [c for c in numbers if c not in HOLDOUT] tcfg = TenantConfigLoader().load("ktcommerce") card_dim = feat[numbers[0]].shape[0] print(f"환경: 2축 FeatureBuyer + 성향 랜덤 · 학습 {episodes}ep · 카드특징 {card_dim}차원 " f"(임베딩384+전략4+톤4) · 학습 {len(train_pool)}장 / 홀드아웃 {HOLDOUT}") qt = QTableAdapter(numbers, tcfg.state) dqn = DQNAdapter(FeatureDQNPolicy(state_dim=STATE_FEATURE_DIM + TENANT_FEATURE_DIM, card_dim=card_dim, eps_decay=4000), feat) print("\n[학습] qtable_ucb ...") train(qt, train_pool, strat, tcfg.reward, tcfg.state, episodes, seed) print("[학습] feature_dqn ...") train(dqn, train_pool, strat, tcfg.reward, tcfg.state, episodes, seed) print("\n=== ① 학습 카드 9장 풀 ===") evaluate(RandomAdapter(seed), train_pool, strat, tcfg.reward, tcfg.state, label="random") evaluate(qt, train_pool, strat, tcfg.reward, tcfg.state, label="qtable_ucb") evaluate(dqn, train_pool, strat, tcfg.reward, tcfg.state, label="feature_dqn") print("\n=== ② zero-shot 11장 풀 (안 본 카드 2장 포함) ===") evaluate(RandomAdapter(seed), numbers, strat, tcfg.reward, tcfg.state, label="random") evaluate(qt, numbers, strat, tcfg.reward, tcfg.state, label="qtable_ucb") evaluate(dqn, numbers, strat, tcfg.reward, tcfg.state, label="feature_dqn") print("\n=== ④ 성향 극단 테스트 — 같은 협력사, 성향만 바꾸면 카드를 바꾸는가 (11장 풀) ===") pref_behavior_test([qt, dqn], numbers, strat, tcfg.reward, tcfg.state) if __name__ == "__main__": main()