o2o-negosium-original/negodata/backend/common/anchoring/base_table.py
민헌 b362edf0d7 feat(negodata): 앵커링 v1.2 적용 — 칸 rate 조회·정수 박제·재생성 앵커 상속 폐지
- common/anchoring 이식 패키지 신설: 원본(schedules/anchoring)에서 읽기 경로 발췌
  (constants·base_table·service) + reader 이식판(Redis 미사용 — current_rates 뷰
  단일 쿼리, 실패·무이력 시 정적 테이블 폴백으로 견적 생성 무중단)
- _build_quotation: 목표가 산정과 앵커 산출 분리 — 칸(items.company_id ×
  quotations.supplier_type × 가격구간) rate 로 tp*(1000-rate)//1000 정수 박제,
  anchor_rate_permille 동시 기록. quotation_settings.anchoring_value 계산 사용 중단
- 재생성(regenerate_next_round): target_price 만 상속, 앵커는 생성 시점 rate 재계산
  (인수인계 규칙 1 — 상속 폐지)
- sessions 모델 anchor_rate_permille 매핑, crud get_item_companies 신설
- 테스트 6종 신설(스키마 부재 폴백·칸별 조정 반영·유형 미지정·재생성 재계산·
  bracket 경계 골든 벡터) — 전체 스위트 50 통과
- negodata 담당자 승인 하 직접 적용. 6개월 압축 시뮬레이션(격주 배치 13회,
  세션 522건)으로 박제→소비→재박제 루프·클램프·이월·이중소비 방지 검증 완료

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-04 16:37:27 +09:00

60 lines
2.5 KiB
Python

"""정적 기본 테이블 — 칸 시작값의 유일한 소스. 원본: schedules/anchoring/src/anchoring/base_table.py
resources/anchoring_base.json(46행 사다리, 불변)을 최초 사용 시 메모리에 로드한다.
DB 에 저장하지 않으며 런타임에 절대 수정하지 않는다. 검증 실패 시 예외(견적 생성 차단이 아니라
잘못된 리소스 배포를 조기에 드러내기 위함 — 파일은 코드와 함께 배포되므로 정상 배포에선 실패하지 않는다).
"""
import json
from pathlib import Path
from common.anchoring.constants import BRACKET_COUNT, UPPER_BOUNDS
_RESOURCE = Path(__file__).parent / "resources" / "anchoring_base.json"
_rates: list[int] | None = None # bracket_index → 시작값(‰)
class BaseTableError(RuntimeError):
"""정적 테이블 로드/검증 실패."""
def _validate(rows: list) -> list[int]:
"""행 검증 후 천분율 정수 리스트로 변환. 실패 시 BaseTableError.
규약: 46행 · idx 1..46 연속 · upper_bound == 사다리(UPPER_BOUNDS) · 값 0.01~0.20.
"""
if not isinstance(rows, list) or len(rows) != BRACKET_COUNT:
raise BaseTableError(f"정적 테이블 행 수 불일치: {len(rows) if isinstance(rows, list) else type(rows)} != {BRACKET_COUNT}")
rates: list[int] = []
for i, row in enumerate(rows):
idx = row.get("idx")
ub = row.get("upper_bound")
av = row.get("anchoring_value")
if idx != i + 1:
raise BaseTableError(f"idx 불연속: 위치 {i} 의 idx={idx} (기대 {i + 1})")
if ub != UPPER_BOUNDS[i]:
raise BaseTableError(f"upper_bound 사다리 불일치: idx={idx} upper_bound={ub} (기대 {UPPER_BOUNDS[i]})")
if not isinstance(av, (int, float)) or av != av or not (0.01 <= av <= 0.20):
raise BaseTableError(f"anchoring_value 범위 밖: idx={idx} value={av}")
rates.append(int(round(av * 1000)))
return rates
def load_base_table() -> None:
"""리소스 파일 로드 + 검증. 최초 1회 호출(멱등)."""
global _rates
if _rates is not None:
return
try:
rows = json.loads(_RESOURCE.read_text())
except Exception as ex:
raise BaseTableError(f"정적 테이블 파일 로드 실패: {_RESOURCE}: {ex}") from ex
_rates = _validate(rows)
def get_base_rate_permille(bracket_index: int) -> int:
"""구간 인덱스 → 시작 앵커링 값(‰)."""
if _rates is None:
load_base_table()
return _rates[bracket_index]