From b4d6cf817b341aea48c3f950560444a904e02748 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 2 Jul 2026 21:13:28 +0900 Subject: [PATCH] =?UTF-8?q?feat(anchoring):=20dry-run=20=EB=AA=A8=EB=93=9C?= =?UTF-8?q?=C2=B7=EB=B0=95=EC=A0=9C=20=EC=A0=95=ED=95=A9=20=EA=B0=90?= =?UTF-8?q?=EC=8B=9C=C2=B7=EC=A0=84=ED=99=98=EA=B8=B0=20=EC=A0=90=ED=94=84?= =?UTF-8?q?=20=EC=A0=88=EC=B0=A8=20=E2=80=94=20=EC=8B=A4=EB=B0=B0=ED=8F=AC?= =?UTF-8?q?=20=EC=A4=80=EB=B9=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 종합 피드백에서 남긴 마지막 개선 3종: - --once --dry-run: 판정·예상 조정(조정예정 라인)·제외 예정 건수를 로그로만 보고 DB/Redis 를 일절 변경하지 않는 예행 연습 — 첫 운영 실행(레거시 전량 판정·마킹) 전에 규모를 눈으로 확인하는 안전장치. status=dry_run, 종료코드 0 - 박제 정합 감시: 스캔 시 정수식 tp×(1000−rate)//1000 과 박제 anchor 를 대조, 불일치 시 WARN + 요약에 snapshot_mismatch — negodata 이식 오류(float 잔재· 칸 해석 오류)를 적용 첫 주에 자동 감지. 전환기(rate 미박제)엔 자동 스킵 - 전환기 점프 절차화: 인수인계 적용 순서에 "negodata 적용 직전 current_rates 분포 확인 → 점프 감수/이력 리셋 정책 결정" 단계 삽입 + TODO 등재 테스트 2종 추가(dry-run 무변경·정합 WARN) — 모듈 20개·backend 57개 통과. 리허설 완료: dry-run 예상과 실제 실행 결과 일치 확인(제외 3·조정 1칸 10‰→30‰). Co-Authored-By: Claude Fable 5 --- schedules/anchoring/README.md | 1 + schedules/anchoring/TODO.md | 5 ++ schedules/anchoring/docs/운영및유지보수.md | 2 + schedules/anchoring/docs/인수인계.md | 8 ++- schedules/anchoring/src/anchoring/batch.py | 60 +++++++++++++++++----- schedules/anchoring/src/anchoring/main.py | 9 ++-- schedules/anchoring/tests/test_batch.py | 33 ++++++++++++ 7 files changed, 102 insertions(+), 16 deletions(-) diff --git a/schedules/anchoring/README.md b/schedules/anchoring/README.md index 920e089..680f8e7 100644 --- a/schedules/anchoring/README.md +++ b/schedules/anchoring/README.md @@ -42,6 +42,7 @@ psql -h 127.0.0.1 -U postgres -d negosium_db -f schema.sql # 로컬(가상환경) python3 -m venv .venv && .venv/bin/pip install -r requirements.txt cp config.toml.example config.toml # DB/Redis 채우기 (env 로 대체 가능) +PYTHONPATH=src .venv/bin/python -m anchoring.main --once --dry-run # 예행 연습(DB/Redis 무변경) PYTHONPATH=src .venv/bin/python -m anchoring.main --once # 수동 1회(격주 게이트 무시) PYTHONPATH=src .venv/bin/python -m anchoring.main # 스케줄러 상주 diff --git a/schedules/anchoring/TODO.md b/schedules/anchoring/TODO.md index 5e64295..6c8cf70 100644 --- a/schedules/anchoring/TODO.md +++ b/schedules/anchoring/TODO.md @@ -30,6 +30,11 @@ ## 백로그 (저우선 — 리뷰에서 식별, 착수 조건 명시) +- [ ] **전환기 점프 정책 결정 (negodata 적용 직전 필수)**: backend 배포~negodata 적용 사이에 + 학습된 rate 가 적용 순간 한 번에 반영된다("한 계단" 원칙의 1회 예외). + 적용 직전 `SELECT max(anchor_rate_after) FROM anchoring.current_rates` 로 폭 확인 후 + 점프 감수 vs 이력 아카이브·리셋을 결정할 것 — 절차는 `docs/인수인계.md` 적용 순서 ③. + - [ ] **percent 입력 모드 대비**: `chat_service.send` 는 `user_input_type == "price"` 만 가격으로 파싱한다. agent 에 percent 스크립트가 도입되면 percent 턴이 가격 흔적 없이 지나가 학습에서 조용히 빠진다(현재 agent 스크립트에 percent 없음 — 잠복). 도입 시 diff --git a/schedules/anchoring/docs/운영및유지보수.md b/schedules/anchoring/docs/운영및유지보수.md index b279eed..a0cc3a6 100644 --- a/schedules/anchoring/docs/운영및유지보수.md +++ b/schedules/anchoring/docs/운영및유지보수.md @@ -165,6 +165,7 @@ docker logs anchoring | tail -20 # 최근 상태 | 작업 | 명령 | |---|---| | 수동 배치 1회 (격주 게이트 무시) | `docker exec anchoring python -m anchoring.main --once` | +| **예행 연습** (DB/Redis 무변경, 예상 결과만 로그) | `docker exec anchoring python -m anchoring.main --once --dry-run` — 첫 운영 실행 전 필수 권장 | | 재기동 | `docker compose restart anchoring` | | 서비스 중지/시작 | `docker compose stop` / `docker compose up -d` | | 설정 변경 반영 | config.toml 수정 → `docker compose up -d --build` | @@ -181,6 +182,7 @@ docker logs anchoring | tail -20 # 최근 상태 | `redis 실패 누계 get=… set=…` WARN | 위와 동일(회차 요약) | 위와 동일 | | `가격 제시 흔적 0%` WARN | backend 의 가격 기록 배선이 끊김(배포 사고 등) — 학습이 조용히 멈추는 신호 | backend 팀에 `chat_service` 의 `last_offered_price` 갱신 경로 점검 요청 | | `칸 평가 실패 company=…` ERROR | 해당 칸 DB 오류/마킹 경합 | 스택 확인. 실패 칸은 마킹되지 않아 **다음 회차 자동 재시도** — 같은 칸이 연속 실패하면 개발 팀 문의 | +| `박제 정합 불일치 n건` WARN | negodata 의 앵커 산출 이식 오류 의심(정수식 ≠ 박제 anchor) | negodata 팀에 `docs/인수인계.md` §1.3 정수식 적용 여부 점검 요청 | | 종료 요약이 WARNING (`failed_cells > 0`) | 일부 칸 실패 | 바로 위 ERROR 라인들 확인 | | 토요일 00:00 에 서비스가 꺼져 있었음 | 배치 회차 누락 | 데이터 유실 없음(자동 이월). 재기동 후 `--once` 로 즉시 캐치업 | | 로그가 아무것도 안 나옴 | 컨테이너 죽음 | `docker ps -a` 로 상태 확인 → `docker logs anchoring` 마지막 로그 → 재기동 | diff --git a/schedules/anchoring/docs/인수인계.md b/schedules/anchoring/docs/인수인계.md index 987787e..3869878 100644 --- a/schedules/anchoring/docs/인수인계.md +++ b/schedules/anchoring/docs/인수인계.md @@ -85,7 +85,13 @@ sessions(..., target_anchoring_price=ap, anchor_rate_permille=rate, ...) ① DB 스키마 적용 (schedules/anchoring/schema.sql — rate_adjustments + sessions 컬럼 3개) ② anchoring 서비스 기동 (schedules/anchoring 컨테이너 — 격주 배치·Redis 캐시 시작) + backend 배포 (마지막 제시가 기록·박제값 소비 — 이 시점부터 표본·조정이 쌓이기 시작) -③ negodata 적용 (앵커 산출 교체 — 이 시점부터 조정된 rate 가 실제 기준가에 반영) +③ 전환기 점프 확인 (negodata 적용 직전): + SELECT max(anchor_rate_after) FROM anchoring.current_rates; + — ②~③ 사이에 학습이 진행되므로, 적용 순간 앵커가 학습된 rate 로 한 번에 이동한다 + ("조정일당 한 계단" 원칙이 이 순간만 예외). 값이 크게 벌어져 있으면 점프 감수 여부 + 또는 이력 리셋을 정책 결정 후 진행. +④ negodata 적용 (앵커 산출 교체 — 이 시점부터 조정된 rate 가 실제 기준가에 반영) + 적용 후 첫 배치 로그에서 "박제 정합 불일치" WARN 이 없는지 확인 — 이식 오류 자동 감지. ``` 각 단계는 독립적으로 안전하다(어느 단계까지만 적용돼도 기존 동작이 깨지지 않음). agent 는 변경 대상이 아니다. 문의는 backend 담당(민헌)에게. diff --git a/schedules/anchoring/src/anchoring/batch.py b/schedules/anchoring/src/anchoring/batch.py index 68209f8..c24b263 100644 --- a/schedules/anchoring/src/anchoring/batch.py +++ b/schedules/anchoring/src/anchoring/batch.py @@ -91,6 +91,7 @@ async def _scan_pending(db, company_ids: list | None = None) -> list: Session.bid_price, Session.target_price, Session.target_anchoring_price, + Session.anchor_rate_permille, Session.last_offered_price, Quotation.supplier_type, Item.company_id, @@ -132,10 +133,12 @@ async def _mark_sessions(db, session_ids: list, adjustment_id: int) -> int: return marked -async def _evaluate_cell(company_id, supplier_type: int, bracket: int, samples: list) -> dict | None: +async def _evaluate_cell(company_id, supplier_type: int, bracket: int, samples: list, + dry_run: bool = False) -> dict | None: """칸 1개 평가 — 조정 INSERT + 소비 마킹을 같은 세션 한 트랜잭션으로(§8 MUST). samples: [(session_id, sample_type_code)] — 유효 표본만, n ≥ 10 보장 후 호출. + dry_run: 계산만 하고 INSERT·마킹·캐시 SET 을 전부 생략(예상 결과 dict 반환). 반환: 요약용 dict / 마킹 경합 시 예외(트랜잭션 롤백). """ session_ids = [sid for sid, _ in samples] @@ -148,6 +151,15 @@ async def _evaluate_cell(company_id, supplier_type: int, bracket: int, samples: if rate_after is None: # 방어적 재확인(호출측에서 n>=10 보장) return None + if dry_run: + return { + "adjustment_id": None, + "before": rate_before, + "after": rate_after, + "success": sum(1 for st in sample_types if st == AnchoringSampleType.BID_SUCCESS.value), + "n": len(samples), + } + adjustment = RateAdjustment( company_id=company_id, supplier_type=supplier_type, @@ -183,11 +195,14 @@ def _new_company_agg() -> dict: return {"evaluated": 0, "up": 0, "hold": 0, "down": 0, "carryover": 0, "failed": 0, "excluded": 0} -async def run_evaluation_batch(force: bool = False, company_ids: list | None = None) -> dict: +async def run_evaluation_batch(force: bool = False, company_ids: list | None = None, + dry_run: bool = False) -> dict: """배치 1회. force=True 면 격주 게이트만 무시(정책 파라미터는 불변). company_ids: 대상 회사 한정 — 테스트가 공유 DB 의 실데이터를 소비하지 않게 하는 격리 장치이자, 특정 테넌트만 표적 수동 실행하는 운영 옵션. None = 전체(운영 기본). + dry_run: 판정·예상 조정을 로그로만 보고 DB/Redis 를 일절 변경하지 않는다 — + 첫 운영 실행 전 "이번 회차에 무슨 일이 일어날지" 확인용(§8 런북). 로그 규약: 모든 라인에 `[batch {run_id}]` 태그(회차 grep), 칸/회사 단위 라인은 `company=` `type=` `bracket=` key=value 형식(회사별 grep — `grep company=`). @@ -196,11 +211,14 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N run_id = now.strftime("%Y%m%d-%H%M%S") tag = f"[batch {run_id}]" scope = f", 대상 회사 {len(company_ids)}곳" if company_ids else "" - LOG.info(f"{tag} 시작 — ISO 주차 {now.isocalendar().week}, force={force}{scope}") + mode = ", DRY-RUN(변경 없음)" if dry_run else "" + LOG.info(f"{tag} 시작 — ISO 주차 {now.isocalendar().week}, force={force}{scope}{mode}") # 절차 0.5 — 캐시 정합(매주, 게이트 무관). Redis 다운이면 즉시 건너뜀 # (셀마다 timeout 을 태우며 수십 분 지연되는 것 방지 — TTL·다음 주 re-SET 이 회복) - if await ping(): + if dry_run: + reconciled = 0 + elif await ping(): reconciled = await _reconcile_cache() LOG.info(f"{tag} 캐시 re-SET {reconciled}칸") else: @@ -220,9 +238,17 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N cells: dict[tuple, list] = defaultdict(list) per_company: dict[str, dict] = defaultdict(_new_company_agg) priced = 0 + snapshot_mismatch = [] for r in rows: if r.last_offered_price is not None: priced += 1 + # 박제 정합 감시: negodata 가 rate 와 anchor 를 함께 박제하기 시작하면(인수인계 적용 후) + # 정수식 tp*(1000-rate)//1000 과 박제 anchor 가 일치해야 한다 — 불일치 = 이식 오류 신호. + # 전환기(rate 미박제 = NULL)에는 자동 스킵된다. + if (r.anchor_rate_permille is not None and r.target_anchoring_price is not None + and r.target_price is not None + and r.target_price * (1000 - r.anchor_rate_permille) // 1000 != r.target_anchoring_price): + snapshot_mismatch.append(r.session_id) if r.supplier_type not in SAMPLEABLE_SUPPLIER_TYPES or r.company_id is None: excluded_ids.append(r.session_id) # 칸 구성 불가 if r.company_id is not None: @@ -241,6 +267,11 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N bracket = calc_bracket_index(r.target_price) cells[(r.company_id, r.supplier_type, bracket)].append((r.session_id, sample_type)) + if snapshot_mismatch: + sample = ", ".join(str(sid) for sid in snapshot_mismatch[:5]) + LOG.warning(f"{tag} 박제 정합 불일치 {len(snapshot_mismatch)}건 — negodata 앵커 산출 이식 오류 의심 " + f"(정수식과 박제 anchor 불일치). 예: {sample}") + # 가격 제시율 — backend 의 last_offered_price 기록 배선 유실(무증상 학습 동결) 감지(§8 절차 5) if rows and priced == 0: LOG.warning(f"{tag} 가격 제시 흔적 0% (종료 재협상 {len(rows)}건 중 last_offered_price 전무) " @@ -250,11 +281,14 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N # 원자성이 불필요하고(중단 시 다음 회차가 이어서 마킹), 첫 실행의 레거시 대량 # 마킹이 장시간 단일 트랜잭션(WAL·락)을 만드는 것을 방지한다. if excluded_ids: - marked_total = 0 - for i in range(0, len(excluded_ids), _MARK_CHUNK): - async with session_scope() as db: - marked_total += await _mark_sessions(db, excluded_ids[i:i + _MARK_CHUNK], MARK_EXCLUDED) - LOG.info(f"{tag} 제외 확정 마킹 {marked_total}건") + if dry_run: + LOG.info(f"{tag} 제외 확정 마킹(예정) {len(excluded_ids)}건 — DRY-RUN, 미실행") + else: + marked_total = 0 + for i in range(0, len(excluded_ids), _MARK_CHUNK): + async with session_scope() as db: + marked_total += await _mark_sessions(db, excluded_ids[i:i + _MARK_CHUNK], MARK_EXCLUDED) + LOG.info(f"{tag} 제외 확정 마킹 {marked_total}건") # 절차 3~4 — 칸별 평가(칸 단위 독립 트랜잭션 — 한 칸 실패가 전파되지 않음) evaluated = up = hold = down = clamped = failed = 0 @@ -266,7 +300,7 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N agg["carryover"] += 1 continue try: - result = await _evaluate_cell(company_id, stype, bracket, samples) + result = await _evaluate_cell(company_id, stype, bracket, samples, dry_run=dry_run) except Exception as ex: failed += 1 agg["failed"] += 1 @@ -279,7 +313,8 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N evaluated += 1 agg["evaluated"] += 1 # 칸별 조정 상세 — 로그만으로 "어느 칸이 왜 바뀌었나" 추적 + DB(adj_id) 교차 확인 - LOG.info(f"{tag} 조정 company={company_id} type={stype} bracket={bracket} " + label = "조정예정" if dry_run else "조정" + LOG.info(f"{tag} {label} company={company_id} type={stype} bracket={bracket} " f"n={result['n']} 성공={result['success']} {result['before']}‰→{result['after']}‰ " f"adj_id={result['adjustment_id']}") if result["after"] > result["before"]: @@ -304,7 +339,7 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N summary = { "run_id": run_id, - "status": "done" if failed == 0 else "partial", + "status": ("dry_run" if dry_run else "done") if failed == 0 else "partial", "scanned": len(rows), "priced_rate": (priced / len(rows)) if rows else None, "excluded_marked": len(excluded_ids), @@ -313,6 +348,7 @@ async def run_evaluation_batch(force: bool = False, company_ids: list | None = N "carryover_cells": carryover, "failed_cells": failed, "companies": len(per_company), + "snapshot_mismatch": len(snapshot_mismatch), "cache_reconciled": reconciled, } # 칸 실패가 있으면 요약을 WARNING 으로 승격 — "WARN 이상 알람" 정책에 걸리도록 diff --git a/schedules/anchoring/src/anchoring/main.py b/schedules/anchoring/src/anchoring/main.py index 11084fe..275de9c 100644 --- a/schedules/anchoring/src/anchoring/main.py +++ b/schedules/anchoring/src/anchoring/main.py @@ -4,6 +4,8 @@ PYTHONPATH=src python -m anchoring.main 수동 1회(격주 게이트 무시 — 미스파이어 캐치업/운영 점검 런북): PYTHONPATH=src python -m anchoring.main --once +예행 연습(판정·예상 조정을 로그로만 — DB/Redis 무변경, 첫 운영 실행 전 확인용): + PYTHONPATH=src python -m anchoring.main --once --dry-run 기동 시 정적 테이블 검증 실패 → 예외로 즉시 중단(§13-7 MUST). """ @@ -32,8 +34,9 @@ async def _run(once: bool) -> None: try: if once: - LOG.info("[main] 수동 1회 실행(--once, 격주 게이트 무시)") - result = await run_evaluation_batch(force=True) + dry = "--dry-run" in sys.argv + LOG.info(f"[main] 수동 1회 실행(--once, 격주 게이트 무시{', dry-run' if dry else ''})") + result = await run_evaluation_batch(force=True, dry_run=dry) LOG.info(f"[main] 결과: {result}") return result @@ -59,7 +62,7 @@ async def _run(once: bool) -> None: def main() -> None: result = asyncio.run(_run(once="--once" in sys.argv)) # --once 가 부분 실패(partial)로 끝나면 비정상 종료코드 — 런북/cron 에서 감지 가능해야 한다 - if result is not None and result.get("status") not in ("done", "skipped"): + if result is not None and result.get("status") not in ("done", "skipped", "dry_run"): sys.exit(1) diff --git a/schedules/anchoring/tests/test_batch.py b/schedules/anchoring/tests/test_batch.py index 95c194b..74d17d0 100644 --- a/schedules/anchoring/tests/test_batch.py +++ b/schedules/anchoring/tests/test_batch.py @@ -189,3 +189,36 @@ async def test_priced_rate_zero_warns(seeder, caplog): with caplog.at_level(logging.WARNING, logger="anchoring"): await run_evaluation_batch(force=True, company_ids=[seeder.company_id]) assert any("가격 제시 흔적 0%" in m for m in caplog.messages) + + +# ── dry-run: 판정·예상 조정만 로그, DB 무변경 ───────────── +async def test_dry_run_changes_nothing(seeder, caplog): + async with adb.session_scope() as db: + ids = await _seed_mixed(db, seeder, success=10, fail=0) + excluded = [await seeder.seed_session(db, status=5, last_offered_price=None)] + + with caplog.at_level(logging.INFO, logger="anchoring"): + result = await run_evaluation_batch(force=True, company_ids=[seeder.company_id], dry_run=True) + + assert result["status"] == "dry_run" and result["evaluated_cells"] == 1 + assert any("조정예정" in m and f"company={seeder.company_id}" in m for m in caplog.messages) + async with adb.session_scope() as db: + assert await _adjustments(db, seeder) == [] # INSERT 없음 + marks = await _marks(db, ids + excluded) + assert all(v is None for v in marks.values()) # 마킹 없음(제외 포함) + + # 이어서 실제 실행하면 그대로 반영된다 (dry-run 이 상태를 소비하지 않았음을 증명) + await _run_batch(seeder) + async with adb.session_scope() as db: + assert len(await _adjustments(db, seeder)) == 1 + + +# ── 박제 정합 감시: 정수식과 박제 anchor 불일치 → WARN ──── +async def test_snapshot_mismatch_warns(seeder, caplog): + async with adb.session_scope() as db: + # rate 10‰ 기준 정수식 anchor 는 29,700 — 29,000 으로 박제된 세션은 이식 오류 신호 + await seeder.seed_session(db, rate=10, anchor_price=29_000, bid_price=28_000) + + with caplog.at_level(logging.WARNING, logger="anchoring"): + await run_evaluation_batch(force=True, company_ids=[seeder.company_id], dry_run=True) + assert any("박제 정합 불일치 1건" in m for m in caplog.messages)