b362edf0d7
- 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>
56 lines
2.6 KiB
Python
56 lines
2.6 KiB
Python
"""현재 앵커링 값 조회(읽기 경로) — negodata 이식판.
|
|
|
|
원본 reader(schedules/anchoring)는 Redis 캐시를 먼저 보지만, 이식판은 DB 직조회 한 문장만 쓴다
|
|
(2026-07-03 단순화 결정 — 조회가 견적 생성 시에만 일어나 캐시가 불필요, Redis 의존 제거).
|
|
칸별 최신 조정 rate 는 모듈 소유 뷰 `anchoring.current_rates` 가 제공하고,
|
|
조정 이력이 없는 칸은 결과에 없으므로 호출측이 정적 테이블 시작값으로 폴백한다.
|
|
|
|
견적 생성이 앵커 조회 때문에 실패해서는 안 된다(인수인계.md §1 규칙 6) — anchoring 스키마
|
|
미적용 환경을 포함해 어떤 실패도 밖으로 던지지 않고 빈 결과(전량 정적 폴백) + WARN 으로 처리한다.
|
|
"""
|
|
from sqlalchemy import column, select, table
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from common.anchoring.constants import ANCHOR_RATE_MAX, ANCHOR_RATE_MIN, SAMPLEABLE_SUPPLIER_TYPES
|
|
from common.logger import LOG
|
|
|
|
# 모듈 소유 DDL(schedules/anchoring/schema.sql)의 조회용 뷰 — negodata 는 ORM 모델 없이 읽기만 한다.
|
|
_current_rates = table(
|
|
"current_rates",
|
|
column("company_id"),
|
|
column("price_bracket_index"),
|
|
column("anchor_rate_permille"),
|
|
column("supplier_type"),
|
|
schema="anchoring",
|
|
)
|
|
|
|
|
|
async def fetch_current_rates(db: AsyncSession, company_ids: list, supplier_type: int) -> dict:
|
|
"""칸별 현재 앵커링 값 일괄 조회. {(company_id, bracket_index): rate‰} 반환.
|
|
|
|
supplier_type 은 견적 단위로 하나뿐이라 키에 넣지 않는다.
|
|
조정 이력이 없는 칸은 결과에 없다(호출측 정적 폴백). 조회 실패 시 빈 dict."""
|
|
if not company_ids or supplier_type not in SAMPLEABLE_SUPPLIER_TYPES:
|
|
return {}
|
|
try:
|
|
stmt = select(
|
|
_current_rates.c.company_id,
|
|
_current_rates.c.price_bracket_index,
|
|
_current_rates.c.anchor_rate_permille,
|
|
).where(
|
|
_current_rates.c.company_id.in_(company_ids),
|
|
_current_rates.c.supplier_type == supplier_type,
|
|
)
|
|
rows = (await db.execute(stmt)).all()
|
|
except Exception as ex:
|
|
LOG.w(f"[앵커링] current_rates 조회 실패 — 전량 정적 테이블 폴백: {ex}")
|
|
return {}
|
|
|
|
out = {}
|
|
for company_id, bracket_index, rate in rows:
|
|
if not ANCHOR_RATE_MIN <= rate <= ANCHOR_RATE_MAX: # 범위 밖 값은 오염 방어 — 버리고 정적 폴백
|
|
LOG.w(f"[앵커링] rate 범위 밖 — 무시(정적 폴백): company={company_id} bracket={bracket_index} rate={rate}")
|
|
continue
|
|
out[(company_id, bracket_index)] = rate
|
|
return out
|