145 lines
5.7 KiB
Python
145 lines
5.7 KiB
Python
"""P2 검증 (계획서 P2 _검증_, 클린룸 반영).
|
|
|
|
기준 변경: "신규 결과 == Chat_server 결과"(독점 복제)가 아니라
|
|
"동일 config + 동일 입력 → 결정론적 동일 출력 + config 주입이 실제로 반영"으로 검증한다.
|
|
|
|
1. build_state/state_index 결정론 + 범위 [0, state_space_size).
|
|
2. mixed-radix 인코딩이 전 차원조합에 대해 [0,162) 전단사(bijection).
|
|
3. config 주입 효과: 테넌트별 임계값이 다르면 같은 입력이 다른 상태로 분류된다.
|
|
4. reward 결정론 + config(failure_penalty 등) 반영.
|
|
5. ActionCardMapper 라운드트립 + 중복방지 마스킹.
|
|
"""
|
|
|
|
import itertools
|
|
import os
|
|
|
|
import numpy as np
|
|
|
|
from negotiation.cards.action_card_mapper import ActionCardMapper
|
|
from negotiation.qtable.domain.model.snapshot import NegotiationOutcome, NegotiationSnapshot
|
|
from negotiation.qtable.domain.model.state import encode_index
|
|
from negotiation.qtable.domain.service.reward_calculator import RewardCalculator
|
|
from negotiation.qtable.domain.service.state_calculator import build_state, state_dims, state_index
|
|
from tenancy.config_loader import TenantConfigLoader
|
|
|
|
_TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants")
|
|
|
|
|
|
def _cfg(tid: str):
|
|
return TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0).load(tid)
|
|
|
|
|
|
def _snapshot(**over) -> NegotiationSnapshot:
|
|
base = dict(
|
|
revenue_amount=5_000_000,
|
|
distribution_code="A",
|
|
partner_count=1,
|
|
acceptance_ratio=0.05,
|
|
input_price=9800,
|
|
anchor_price=9900, # KT 앵커링가 (anchor < target)
|
|
target_price=10000, # KT 목표 매입가
|
|
round_number=1,
|
|
outcome=NegotiationOutcome.ONGOING,
|
|
)
|
|
base.update(over)
|
|
return NegotiationSnapshot(**base)
|
|
|
|
|
|
def test_build_state_deterministic_and_in_range():
|
|
cfg = _cfg("ktcommerce")
|
|
snap = _snapshot()
|
|
s1 = build_state(snap, cfg.state)
|
|
s2 = build_state(snap, cfg.state)
|
|
assert s1 == s2 # 결정론
|
|
idx = state_index(snap, cfg.state)
|
|
assert idx == state_index(snap, cfg.state)
|
|
assert 0 <= idx < cfg.state.state_space_size
|
|
|
|
|
|
def test_encode_index_known_example():
|
|
# dims=[3,3,3,3,2], 인덱스 (1,0,0,0,1) → ((((1)*3+0)*3+0)*3+0)*2+1 = 54*1 +1 = 55
|
|
assert encode_index([1, 0, 0, 0, 1], [3, 3, 3, 3, 2]) == 55
|
|
# 최소/최대
|
|
assert encode_index([0, 0, 0, 0, 0], [3, 3, 3, 3, 2]) == 0
|
|
assert encode_index([2, 2, 2, 2, 1], [3, 3, 3, 3, 2]) == 161
|
|
|
|
|
|
def test_mixed_radix_bijection_over_full_space():
|
|
cfg = _cfg("ktcommerce")
|
|
dims = state_dims(cfg.state)
|
|
assert dims == [3, 3, 3, 3, 2]
|
|
seen = set()
|
|
for combo in itertools.product(*[range(d) for d in dims]):
|
|
idx = encode_index(list(combo), dims)
|
|
seen.add(idx)
|
|
# 162개 조합이 0..161 에 1:1
|
|
assert seen == set(range(cfg.state.state_space_size))
|
|
|
|
|
|
def test_config_injection_changes_classification():
|
|
# revenue=20,000,000 원: ktcommerce(th=[10M,50M]) → mid(1), imarketkorea(th=[30M,100M]) → low(0)
|
|
snap = _snapshot(revenue_amount=20_000_000)
|
|
kt = build_state(snap, _cfg("ktcommerce").state)
|
|
imk = build_state(snap, _cfg("imarketkorea").state)
|
|
assert kt.revenue_idx == 1
|
|
assert imk.revenue_idx == 0
|
|
assert kt != imk # 같은 입력이 테넌트 config 에 따라 다른 상태
|
|
|
|
|
|
def test_distribution_unknown_code_raises():
|
|
cfg = _cfg("ktcommerce")
|
|
snap = _snapshot(distribution_code="Z") # code_map 에 없음
|
|
try:
|
|
build_state(snap, cfg.state)
|
|
assert False, "unknown distribution code should raise"
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
def test_price_zone_and_partner_buckets():
|
|
cfg = _cfg("ktcommerce").state
|
|
# 제시가 ≤ 앵커가(9900) → 우선협상 구간(0)
|
|
assert build_state(_snapshot(input_price=9800), cfg).price_zone_idx == 0
|
|
# 제시가 > 앵커가 → 협상 지속 구간(1)
|
|
assert build_state(_snapshot(input_price=10500), cfg).price_zone_idx == 1
|
|
# partner: 0->none(2), 1->single(0), 3->multiple(1)
|
|
assert build_state(_snapshot(partner_count=0), cfg).partner_idx == 2
|
|
assert build_state(_snapshot(partner_count=1), cfg).partner_idx == 0
|
|
assert build_state(_snapshot(partner_count=3), cfg).partner_idx == 1
|
|
|
|
|
|
def test_reward_deterministic_and_config_driven():
|
|
snap = _snapshot(outcome=NegotiationOutcome.FAILURE, round_number=2)
|
|
kt_rc = RewardCalculator(_cfg("ktcommerce").reward) # failure_penalty -0.5
|
|
imk_rc = RewardCalculator(_cfg("imarketkorea").reward) # failure_penalty -0.7
|
|
|
|
r1 = kt_rc.calculate(snap)
|
|
r2 = kt_rc.calculate(snap)
|
|
assert r1 == r2 # 결정론
|
|
assert r1.end_reward == -0.5 # config 반영
|
|
assert imk_rc.calculate(snap).end_reward == -0.7
|
|
|
|
# 성공 라운드 보상이 실패보다 크다 (방향성)
|
|
success = _snapshot(outcome=NegotiationOutcome.SUCCESS, round_number=0)
|
|
assert kt_rc.calculate(success).total > kt_rc.calculate(_snapshot(outcome=NegotiationOutcome.FAILURE, round_number=0)).total
|
|
|
|
|
|
def test_action_card_mapper_roundtrip_and_mask():
|
|
cfg = _cfg("ktcommerce")
|
|
mapper = ActionCardMapper(cfg.action_mapping)
|
|
assert mapper.action_space_size == 9
|
|
assert mapper.get_card_id(0) == "NGC-A001"
|
|
assert mapper.get_action_id("NGC-A001") == 0
|
|
assert mapper.get_card_id(99) is None
|
|
|
|
# 중복방지 마스킹: 사용한 action 제외
|
|
mask = mapper.available_mask(used_action_ids={0, 3})
|
|
assert mask.dtype == np.bool_
|
|
assert mask[0] == False and mask[3] == False
|
|
assert mask[1] == True
|
|
assert mask.sum() == 7
|
|
|
|
# reload 로 다른 테넌트 카드셋 교체
|
|
mapper.reload(_cfg("imarketkorea").action_mapping)
|
|
assert mapper.get_card_id(0) == "NGC-B001"
|