hbyang
be1b4b968f
- 동적 가중치 W: 라운드 감쇠 → 상태 5차원 가중합 clip(Σwᵢ·Sᵢ, 0.2, 0.8) (식 12~13, 기존 w1~w5 연결) - 종료보상에 (1−W) 적용: R = W×R_price + (1−W)×R_end − λ×round (식 8) - R_price 3단계: P<anchor 시 1+β·(anchor−P)/anchor 초과달성 보너스 추가 (식 9~11, beta 의미 재정의) - price zone 경계는 명세(T)와 달리 anchor 유지(우선협상 규칙이 실제 의사결정 경계) — 사유 docstring 명시 - state_calculator/config 의 낡은 반대 컨벤션(anchor≥target) 주석 정정 - RewardCalculator(RewardConfig, StateConfig) 시그니처 변경 + 호출부 5곳 갱신, 테스트 기대값 정정 (76/76 PASS) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
84 lines
3.9 KiB
Python
84 lines
3.9 KiB
Python
"""에피소드 시뮬레이터 — 정책 vs 구매자의 한 협상 시퀀스 (H5).
|
|
|
|
한 에피소드: 앵커에서 시작, 매 턴 정책이 카드 선택 → 구매자 반응(수락/계속/이탈) → 보상.
|
|
수락 시 타결가로 성공 종료, 마지막 턴까지 미수락이면 결렬. 보상은 RewardCalculator(테넌트 config)로 채점.
|
|
정책의 update(Transition)로 온라인 학습(시퀀스 보상). 동일 seed 시 재현 가능(페어드 비교).
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from eval_harness.buyer import HeuristicBuyer, Scenario
|
|
from negotiation.policies.base import EpisodeState, PolicyContext, Transition
|
|
from negotiation.qtable.domain.model.snapshot import NegotiationOutcome, NegotiationSnapshot
|
|
from negotiation.qtable.domain.service.reward_calculator import RewardCalculator
|
|
from negotiation.qtable.domain.service.state_calculator import state_index
|
|
from tenancy.config import RewardConfig, StateConfig
|
|
|
|
|
|
@dataclass
|
|
class EpisodeResult:
|
|
success: bool
|
|
settled_price: float
|
|
turns: int
|
|
total_reward: float
|
|
target_price: float
|
|
first_action: int = -1 # 첫 턴 선택 카드(학습 수렴 측정용)
|
|
|
|
@property
|
|
def settled_ratio(self) -> float:
|
|
return self.settled_price / self.target_price if self.target_price else 0.0
|
|
|
|
|
|
def run_episode(policy, buyer: HeuristicBuyer, scenario: Scenario,
|
|
state_cfg: StateConfig, reward_cfg: RewardConfig, action_space_size: int,
|
|
max_turns: int = 5, learn: bool = True) -> EpisodeResult:
|
|
rc = RewardCalculator(reward_cfg, state_cfg)
|
|
episode = EpisodeState()
|
|
total_r = 0.0
|
|
# 협력사는 목표가보다 높게 시작(협상 여지). KT 가 카드로 앵커가 이하까지 끌어내린다.
|
|
price = scenario.target_price * 1.15
|
|
last_idx = last_action = None
|
|
first_action = -1
|
|
|
|
def snap(p, turn, outcome):
|
|
return NegotiationSnapshot(
|
|
revenue_amount=scenario.revenue_amount, distribution_code=scenario.distribution_code,
|
|
partner_count=scenario.partner_count, acceptance_ratio=scenario.acceptance_ratio,
|
|
input_price=p, anchor_price=scenario.anchor_price, target_price=scenario.target_price,
|
|
round_number=turn, outcome=outcome,
|
|
)
|
|
|
|
for turn in range(1, max_turns + 1):
|
|
s = snap(price, turn, NegotiationOutcome.ONGOING)
|
|
idx = state_index(s, state_cfg)
|
|
ctx = PolicyContext(state_index=idx, snapshot=s, action_space_size=action_space_size, episode=episode)
|
|
decision = policy.select(ctx)
|
|
last_idx, last_action = idx, decision.action_id
|
|
if turn == 1:
|
|
first_action = decision.action_id
|
|
|
|
resp = buyer.respond(decision.action_id, scenario, turn, price)
|
|
price = resp.new_price # 협력사가 양보한 새 가격
|
|
# 우선협상(제시가 ≤ anchor) 또는 협력사 수락 → 타결
|
|
if resp.accept or price <= scenario.anchor_price:
|
|
fs = snap(price, turn, NegotiationOutcome.SUCCESS)
|
|
r = rc.calculate(fs).total # 낮은 타결가일수록 보상↑
|
|
if learn:
|
|
policy.update(Transition(state_index=idx, action_id=decision.action_id, reward=r, done=True))
|
|
total_r += r
|
|
return EpisodeResult(True, price, turn, total_r, scenario.target_price, first_action)
|
|
|
|
# 미타결: 진행 보상 후 다음 턴(가격은 계속 내려간 상태)
|
|
r = rc.calculate(s).total
|
|
if learn:
|
|
policy.update(Transition(state_index=idx, action_id=decision.action_id, reward=r, done=False))
|
|
total_r += r
|
|
|
|
# 카드 소진/라운드 종료까지 우선협상 미달 → 결렬
|
|
fs = snap(price, max_turns, NegotiationOutcome.FAILURE)
|
|
r = rc.calculate(fs).total
|
|
if learn and last_idx is not None:
|
|
policy.update(Transition(state_index=last_idx, action_id=last_action, reward=r, done=True))
|
|
total_r += r
|
|
return EpisodeResult(False, price, max_turns, total_r, scenario.target_price, first_action)
|