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>
92 lines
3.9 KiB
Python
92 lines
3.9 KiB
Python
"""build_state — NegotiationSnapshot + StateConfig → State/state_index (우리 자체 구현).
|
|
|
|
각 차원 인덱스 산출 규칙(기능적 방법)은 우리 설계이며, 임계값/코드맵은 config 주입이다.
|
|
도메인은 tenant-agnostic: 같은 config + 같은 snapshot 이면 항상 같은 출력(결정론).
|
|
"""
|
|
|
|
from typing import List
|
|
|
|
from negotiation.qtable.domain.model.snapshot import NegotiationSnapshot
|
|
from negotiation.qtable.domain.model.state import State, encode_index
|
|
from tenancy.config import StateConfig
|
|
|
|
|
|
def _threshold_bucket(value: float, thresholds: List[float]) -> int:
|
|
"""thresholds 경계로 구간 인덱스 산출. value <= thresholds[i] 이면 i, 모두 초과면 마지막 구간.
|
|
|
|
경계 포함 규칙: value <= threshold → 해당 구간(하한). (예: th=[10,30] → ≤10:0, ≤30:1, else:2)
|
|
"""
|
|
for i, th in enumerate(thresholds):
|
|
if value <= th:
|
|
return i
|
|
return len(thresholds) # 마지막 구간 (= dim-1, dim = len(thresholds)+1)
|
|
|
|
|
|
def _acceptance_bucket(ratio: float, thresholds: List[float]) -> int:
|
|
"""수용률 구간: < thresholds[0] → 0, <= thresholds[1] → 1, ... else 마지막.
|
|
|
|
하한은 strict-less, 이후 경계는 inclusive (low 는 미만, mid 이상은 이하).
|
|
"""
|
|
if ratio < thresholds[0]:
|
|
return 0
|
|
for i in range(1, len(thresholds)):
|
|
if ratio <= thresholds[i]:
|
|
return i
|
|
return len(thresholds)
|
|
|
|
|
|
def _partner_bucket(count: int) -> int:
|
|
"""파트너 수 → 인덱스. 규약: single=0, multiple=1, none=2."""
|
|
if count <= 0:
|
|
return 2 # none
|
|
if count == 1:
|
|
return 0 # single
|
|
return 1 # multiple
|
|
|
|
|
|
def _price_zone_bucket(input_price: float, anchor_price: float, target_price: float) -> int:
|
|
"""입력가격 구간 (KT 구매자 관점, anchor 앵커링가 < target 목표 매입가).
|
|
|
|
협력사 제시가가 앵커링가 이하면 우선협상(즉시 타결) 구간(0), 초과면 협상 지속 구간(1).
|
|
설계 명세 v4 는 target 을 경계로 두지만, 확정 경제모델(제시가≤anchor→우선협상 타결)의
|
|
실제 의사결정 경계는 anchor 이므로 의도적으로 anchor 를 경계로 사용한다.
|
|
"""
|
|
if anchor_price <= 0 or target_price <= 0:
|
|
raise ValueError("anchor_price/target_price must be positive")
|
|
if input_price <= anchor_price:
|
|
return 0 # at_or_below_anchor (우선협상 가능)
|
|
return 1 # above_anchor (협상 지속)
|
|
|
|
|
|
def state_dims(cfg: StateConfig) -> List[int]:
|
|
"""각 차원의 크기. config 의 weights/code_map 길이로 결정."""
|
|
return [
|
|
len(cfg.revenue.weights),
|
|
len(cfg.distribution.weights),
|
|
len(cfg.partner.weights),
|
|
len(cfg.acceptance.weights),
|
|
len(cfg.price_zone.weights),
|
|
]
|
|
|
|
|
|
def build_state(snapshot: NegotiationSnapshot, cfg: StateConfig) -> State:
|
|
"""snapshot 을 config 기준으로 이산 State 로 변환."""
|
|
revenue_idx = _threshold_bucket(snapshot.revenue_amount, cfg.revenue.thresholds)
|
|
|
|
code = (snapshot.distribution_code or "").strip()
|
|
if code not in cfg.distribution.code_map:
|
|
raise ValueError(f"unknown distribution code: {snapshot.distribution_code!r} (code_map keys={list(cfg.distribution.code_map)})")
|
|
distribution_idx = cfg.distribution.code_map[code]
|
|
|
|
partner_idx = _partner_bucket(snapshot.partner_count)
|
|
acceptance_idx = _acceptance_bucket(snapshot.acceptance_ratio, cfg.acceptance.thresholds)
|
|
price_zone_idx = _price_zone_bucket(snapshot.input_price, snapshot.anchor_price, snapshot.target_price)
|
|
|
|
return State(revenue_idx, distribution_idx, partner_idx, acceptance_idx, price_zone_idx)
|
|
|
|
|
|
def state_index(snapshot: NegotiationSnapshot, cfg: StateConfig) -> int:
|
|
"""snapshot → 단일 정수 state_index (mixed-radix)."""
|
|
st = build_state(snapshot, cfg)
|
|
return encode_index(list(st.to_tuple()), state_dims(cfg))
|