f29e9723b0
schedules/anchoring — backend 를 import 하지 않는 독립 컨테이너 배치 서비스.
회사 × 협력사유형(1유통/2제조/3총판) × 가격구간(3,000원, 33,334칸)별 앵커링
값(정수 천분율)을 격주 토 00:00 KST 에 협상 성공률로 자동 조정한다.
- 판정 = "가격 흔적" 기준: last_offered_price 가 있는 종료 재협상만 표본,
DONE & bid ≤ 박제 앵커만 성공, 나머지(초과 합의·결렬·가격 쓰고 이탈) 실패.
앵커는 비노출(엔진 내부 체결 임계) — agent 무변경
- 저장 = anchoring.rate_adjustments 1개(append-only, consumed_session_ids 박제),
소비 경계 = sessions.anchoring_adjustment_id 마킹(멱등·이월). DDL 은 모듈
소유(schema.sql, sessions 3컬럼 ALTER 포함)
- 안정성: 조정 INSERT+마킹 한 트랜잭션 + rowcount 불일치 전체 롤백,
Redis TTL 7일 + 매주 조정 칸 re-SET, socket timeout 0.3s, DB 폴백,
가격 제시율 0% WARN, --once 수동 캐치업
- 정적 기본 테이블(전 구간 10‰, 상한 정확히 1억·초과분 마지막 인덱스 클램프)
기동 검증 실패 시 기동 중단
- 전체 async(SQLAlchemy+asyncpg, redis.asyncio) — negodata 가 reader 를 그대로
이식 가능(docs/인수인계.md). 최종 문서 docs/{개발용,기획용,워크플로우}.md
- 테스트 15종: 골든 벡터(§11) + DB 통합(멱등·이월·격리·rowcount 롤백)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
155 lines
7.1 KiB
Python
155 lines
7.1 KiB
Python
"""배치 통합 테스트 (스펙 §11.5 — 실제 Postgres, Redis 없음(무Redis 폴백 경로)).
|
|
|
|
실행: cd schedules/anchoring && PYTHONPATH=src .venv/bin/python -m pytest tests/test_batch.py -q
|
|
"""
|
|
import uuid
|
|
|
|
from sqlalchemy import select, text
|
|
|
|
from anchoring import db as adb
|
|
from anchoring.batch import MarkingConflictError, _evaluate_cell, run_evaluation_batch
|
|
from anchoring.models import RateAdjustment, Session
|
|
from anchoring.reader import get_anchor_rate
|
|
from conftest import requires_db
|
|
|
|
pytestmark = requires_db
|
|
|
|
# 시드 기본값: target 30,000 / rate 10‰ / anchor 29,700 → bracket = 30000//3000 = 10
|
|
BRACKET = 10
|
|
SUCCESS_BID = 29_000 # ≤ anchor → BID_SUCCESS
|
|
FAIL_BID = 29_999 # > anchor → BID_FAIL
|
|
|
|
|
|
async def _adjustments(db, seeder):
|
|
stmt = (
|
|
select(RateAdjustment)
|
|
.where(RateAdjustment.company_id == seeder.company_id)
|
|
.order_by(RateAdjustment.id)
|
|
)
|
|
return (await db.execute(stmt)).scalars().all()
|
|
|
|
|
|
async def _marks(db, session_ids):
|
|
stmt = select(Session.session_id, Session.anchoring_adjustment_id).where(Session.session_id.in_(session_ids))
|
|
return dict((await db.execute(stmt)).all())
|
|
|
|
|
|
async def _seed_mixed(db, seeder, success: int, fail: int, **kw):
|
|
ids = []
|
|
for _ in range(success):
|
|
ids.append(await seeder.seed_session(db, bid_price=SUCCESS_BID, **kw))
|
|
for _ in range(fail):
|
|
ids.append(await seeder.seed_session(db, bid_price=FAIL_BID, **kw))
|
|
return ids
|
|
|
|
|
|
# ── §11.5: 13건 전량 평가 + 멱등 (실패 3종 혼합) ──────────
|
|
async def test_full_cycle_and_idempotency(seeder):
|
|
async with adb.session_scope() as db:
|
|
ids = await _seed_mixed(db, seeder, success=8, fail=2) # DONE 인데 앵커 초과(와일드카드 상단 등)
|
|
for _ in range(2): # 가격 쓰고 결렬(REJECTED) = 실패
|
|
ids.append(await seeder.seed_session(db, status=5, bid_price=None, last_offered_price=FAIL_BID))
|
|
# 가격 쓰고 이탈 → 견적 마감 시 일괄 NOT_PARTICIPATED = 실패 (중간 이탈 시나리오)
|
|
ids.append(await seeder.seed_session(db, status=4, bid_price=None, last_offered_price=FAIL_BID))
|
|
# 합계 13건, 성공 8 → r≈0.615 → +20
|
|
|
|
await run_evaluation_batch(force=True)
|
|
|
|
async with adb.session_scope() as db:
|
|
adjustments = await _adjustments(db, seeder)
|
|
assert len(adjustments) == 1
|
|
adj = adjustments[0]
|
|
assert (adj.nego_count, adj.success_count) == (13, 8)
|
|
assert (adj.anchor_rate_before, adj.anchor_rate_after) == (10, 30)
|
|
assert sorted(adj.consumed_session_ids) == sorted(str(i) for i in ids)
|
|
marks = await _marks(db, ids)
|
|
assert all(v == adj.id for v in marks.values()) # 13건 모두 소비 마킹
|
|
|
|
# 재실행 — 마킹 멱등: 우리 칸 조정은 그대로 1건
|
|
await run_evaluation_batch(force=True)
|
|
async with adb.session_scope() as db:
|
|
assert len(await _adjustments(db, seeder)) == 1
|
|
|
|
|
|
# ── §11.5: 이월(7건 스킵 → 누적 13건 단일 평가) ──────────
|
|
async def test_carryover(seeder):
|
|
async with adb.session_scope() as db:
|
|
first = await _seed_mixed(db, seeder, success=5, fail=2) # 7건 < 10
|
|
|
|
await run_evaluation_batch(force=True)
|
|
async with adb.session_scope() as db:
|
|
assert await _adjustments(db, seeder) == []
|
|
marks = await _marks(db, first)
|
|
assert all(v is None for v in marks.values()) # 마킹 없음 = 이월
|
|
|
|
second = await _seed_mixed(db, seeder, success=3, fail=3) # 누적 13건 (8S/5F)
|
|
|
|
await run_evaluation_batch(force=True)
|
|
async with adb.session_scope() as db:
|
|
adjustments = await _adjustments(db, seeder)
|
|
assert len(adjustments) == 1
|
|
assert adjustments[0].nego_count == 13 # 4주치 전량 1회 평가
|
|
assert adjustments[0].anchor_rate_after == 30
|
|
marks = await _marks(db, first + second)
|
|
assert all(v == adjustments[0].id for v in marks.values())
|
|
|
|
|
|
# ── §11.5: 회사 격리 + 현재값 조회(무Redis DB 폴백) + δ 유형 차원 ──
|
|
async def test_company_isolation_and_reader(seeder):
|
|
async with adb.session_scope() as db:
|
|
await _seed_mixed(db, seeder, success=10, fail=0, supplier_type=1) # 유통 → +20
|
|
await _seed_mixed(db, seeder, success=10, fail=0, supplier_type=2) # 제조 → +10 (δ 스왑 가드)
|
|
|
|
await run_evaluation_batch(force=True)
|
|
|
|
other_company = uuid.uuid4()
|
|
async with adb.session_scope() as db:
|
|
adjustments = await _adjustments(db, seeder)
|
|
by_type = {a.supplier_type: a.anchor_rate_after for a in adjustments}
|
|
assert by_type == {1: 30, 2: 20}
|
|
# 조정된 칸은 새 rate, 타사 같은 (유형,구간) 칸은 정적 테이블 시작값
|
|
assert await get_anchor_rate(db, seeder.company_id, 1, BRACKET) == 30
|
|
assert await get_anchor_rate(db, other_company, 1, BRACKET) == 10
|
|
|
|
|
|
# ── §11.5: EXCLUDED 마킹 0 + 유효 n<10 이월 + supplier_type 미지정 ──
|
|
async def test_excluded_and_unsampleable(seeder):
|
|
async with adb.session_scope() as db:
|
|
# 가격 흔적 없는 종료(무가격 결렬·미참여) → EXCLUDED
|
|
excluded = [await seeder.seed_session(db, status=5, last_offered_price=None) for _ in range(8)]
|
|
excluded += [await seeder.seed_session(db, status=4, last_offered_price=None) for _ in range(7)]
|
|
valid = await _seed_mixed(db, seeder, success=5, fail=0) # 유효 5 < 10
|
|
untyped = [await seeder.seed_session(db, supplier_type=0, bid_price=SUCCESS_BID)] # 칸 구성 불가
|
|
|
|
await run_evaluation_batch(force=True)
|
|
|
|
async with adb.session_scope() as db:
|
|
assert await _adjustments(db, seeder) == [] # 유효 5 < 10 → 평가 없음
|
|
marks = await _marks(db, excluded + untyped)
|
|
assert all(v == 0 for v in marks.values()) # 제외 확정 마킹(재스캔 방지)
|
|
marks = await _marks(db, valid)
|
|
assert all(v is None for v in marks.values()) # 유효 표본은 이월
|
|
|
|
|
|
# ── 개정 1: 마킹 rowcount ≠ n → 조정 INSERT 포함 전체 롤백 ──
|
|
async def test_marking_conflict_rolls_back(seeder):
|
|
async with adb.session_scope() as db:
|
|
ids = await _seed_mixed(db, seeder, success=10, fail=0)
|
|
# 경합 시뮬레이션: 1건을 다른 실행이 먼저 소비한 상태로 만든다
|
|
await db.execute(text(
|
|
"UPDATE negotiation.sessions SET anchoring_adjustment_id = 999999 WHERE session_id = :sid"
|
|
), {"sid": ids[0]})
|
|
|
|
samples = [(sid, 1) for sid in ids] # 10건 전부 BID_SUCCESS 로 평가 시도
|
|
try:
|
|
await _evaluate_cell(seeder.company_id, 1, BRACKET, samples)
|
|
raised = False
|
|
except MarkingConflictError:
|
|
raised = True
|
|
assert raised
|
|
|
|
async with adb.session_scope() as db:
|
|
assert await _adjustments(db, seeder) == [] # 롤백 — 이중 조정 없음
|
|
marks = await _marks(db, ids[1:])
|
|
assert all(v is None for v in marks.values()) # 나머지 9건 마킹도 롤백
|