docs(CLAUDE.md): realestate 매물 알림+안전마진 엔드포인트 + agent-office notify-listing 등재

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9
fix(realestate): 시세 표본 광역폴백(M1)+area-NULL upsert skip(M3)+matches JSON파싱(M4)
2026-07-09 14:08:14 +09:00 · 2026-07-09 14:03:54 +09:00 · 2026-07-09 13:50:04 +09:00 · 2026-07-09 13:43:34 +09:00 · 2026-07-09 13:39:19 +09:00 · 2026-07-09 13:36:20 +09:00
31 changed files with 1511 additions and 25 deletions
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -332,10 +332,12 @@ docker compose up -d
 | POST | `/api/internal/insta/update` | Windows 워커 결과 webhook |
 ### realestate-lab (realestate-lab/)
-공공데이터포털 청약 분양정보 수집 + 자치구 5티어 매칭 + agent-office push 알림.
+공공데이터포털 청약 분양정보 수집 + 자치구 5티어 매칭 + agent-office push 알림. **+ 매물 알림(전월세·매매) + 안전마진/매수 적정성 판정 + 재무·규제 예산 산정**(2026-07-09 추가).
- 핵심 파일: `main.py`, `db.py`, `collector.py`, `matcher.py`, `notifier.py`, `models.py`
+- 핵심 파일(청약): `main.py`, `db.py`, `collector.py`, `matcher.py`, `notifier.py`, `models.py`
 - 핵심 파일(매물): `listing_collector.py`(국토부 실거래[합법]+네이버 호가[폴백]), `listing_matcher.py`(안전마진·적정성), `finance_rules.py`+`finance_rules_config.py`(전세대출·LTV·DSR·토허), `lawd_codes.py`(법정동코드)
 - 매칭 100점: 지역35 / 주택유형10 / 면적15 / 가격15 / 자격25
- 📌 상세(DB 스키마·스케줄러 4단계·매칭 모델·notifier 멱등 흐름·env): **`service_realestate.md`**
+- 매물 판정: 임차 전세가율 ≤0.70🟢/≤0.80🟡/>0.80🔴, 매매 호가율 ≤0.97🟢/≤1.05🟡/>1.05🔴, 실거래 표본<3 ⚪보류. 네이버 차단돼도 실거래·안전마진·청약 무중단(`naver_ok`)
 - 📌 상세(DB 스키마·스케줄러·매칭 모델·notifier 멱등·매물 4테이블·판정 경계·env): **`service_realestate.md`**
 | 메서드 | 경로 | 설명 |
 |--------|------|------|
@@ -349,6 +351,12 @@ docker compose up -d
 | POST | `/api/realestate/matches/refresh` | 매칭 재계산 |
 | PATCH | `/api/realestate/matches/{id}/read` | 신규 알림 읽음 |
 | GET | `/api/realestate/dashboard` | 요약 (진행중·신규매칭·일정) |
 | GET | `/api/realestate/listings` | 매물 목록 (dong/deal_type/tier/matched_only/page/size) |
 | POST/GET | `/api/realestate/listings/collect` (+ `/collect/status`) | 매물 수동 수집(collect→match→notify)/상태 |
 | GET/PUT | `/api/realestate/listings/criteria` | 매물 조건(동·거래유형·보증금상한·자기자금·연소득 등) 조회/수정 |
 | GET | `/api/realestate/listings/matches` | 매물 매칭+판정 결과 |
 | POST | `/api/realestate/safety-check` | 단건 안전마진/적정성 판정 (실거래 median 대비) |
 | POST | `/api/realestate/budget` | 전세/매수 예산·규제 산정 (전세대출·LTV·DSR·토허) |
 ### agent-office (agent-office/)
 AI 에이전트 가상 오피스 — 기존 서비스 API를 프록시로 호출, 실시간 WebSocket + 텔레그램 봇.
@@ -365,7 +373,8 @@ AI 에이전트 가상 오피스 — 기존 서비스 API를 프록시로 호출
 | POST | `/api/agent-office/command` | 에이전트 명령 전송 |
 | POST | `/api/agent-office/approve` | 작업 승인/거부 |
 | POST | `/api/agent-office/telegram/webhook` | 텔레그램 Webhook (realestate_bookmark_* 콜백 포함) |
-| POST | `/api/agent-office/realestate/notify` | realestate-lab 전용 push 수신 → 텔레그램 |
+| POST | `/api/agent-office/realestate/notify` | realestate-lab 청약 매칭 push 수신 → 텔레그램 |
 | POST | `/api/agent-office/realestate/notify-listing` | realestate-lab 매물 알림 push 수신 → 텔레그램(너+아내, 안전마진/적정성 렌더) |
 | GET | `/api/agent-office/states` | 전체 에이전트 상태 |
 | GET | `/api/agent-office/nodes` | 분산 워커(NAS↔Windows) 관측 — heartbeat 생사+큐깊이+dead-letter 집계 (web-ui `/infra` Three.js 시각화 소비). 상세 → `infra_distributed_workers.md` |
 | GET | `/api/agent-office/activity` | 전 에이전트 통합 활동 피드 (tasks+logs UNION). 필터 `agent_id`/`type`(task\|log)/`status`/`days` + `limit`/`offset` |
@@ -492,6 +501,7 @@ Gitea Webhook 수신 → 자동 배포. HMAC SHA256 검증(`X-Gitea-Signature`
 - **공휴일 목록**: `stock/app/holidays.json` 매년 수동 갱신 (KRX 기준)
 - **Windows AI 서버 IP**: `192.168.45.59` (DHCP 고정 예약). Tailscale은 Synology userspace 모드라 TCP 불가 → 로컬 IP 사용
 - **렌더/생성 워커 분리**: music/video/image/insta 무거운 작업은 Windows `web-ai` 워커. NAS 코드의 `*_provider.py`/`card_renderer.py`가 DEPRECATED stub면 실 로직은 web-ai 쪽이 authoritative
 - **[팀 규칙] 모든 WSL(docker) 워커는 `/infra`에서 관측 가능해야 한다**: 새 워커 추가 시 필수 3단계 — ① 워커가 `worker:<name>:heartbeat`(EX45, ~15초) 발신 ② BE가 `agent-office/app/node_monitor.py`의 `WORKER_REGISTRY`에 `{name,kind,queue}` 등재 ③ → `/api/agent-office/nodes`·web-ui `/infra` 노출 + 다운/복구/dead-letter 텔레그램 경보. 미준수 = "사일런트 사망"(insta-render 2주 무관측 사고) 재발 위험. 워커 신규/변경 PR 머지 게이트. web-ai/web-ui repo CLAUDE.md에도 동일 규칙 명시 필요. 상세는 `infra_distributed_workers.md` 메모리(관측 계약 2)
 - **Playwright Dockerfile**: bookworm 고정 + 수동 chromium deps, `--with-deps` 금지 (`feedback_playwright_dockerfile.md`)
 - **lab 네이밍**: `-lab`은 개발/연구 단계에만, 정식 서비스엔 미사용 (`feedback_lab_naming.md`)
--- a/README.md
+++ b/README.md
@@ -115,6 +115,7 @@ curl http://localhost:18500/health
 - **실계좌**: Windows AI 서버(192.168.45.59:8000) 프록시 → KIS Open API (잔고/주문)
 - **포트폴리오**: 종목·예수금·매도 히스토리 관리, 현재가 자동 조회
 - **자산 스냅샷**: 평일 15:40 자동 저장 (KRX 공휴일 판별, `holidays.json` 매년 갱신)
 - **실시간 매매 알람** (2026-07-02): 장중(+시간외) 1분 폴링으로 매수(watchlist ∪ 스크리너 후보, TA 시그널)·매도(보유종목, exit 룰 + 트레일링 스톱) 조건 충족 시 텔레그램(본인+아내) 알람. **TA 계산은 Windows `trade-monitor` WSL2 docker 워커**, NAS는 감시대상 조립 + edge 중복판정(영속) + 발송 담당. 관심종목은 `/api/stock/watchlist` CRUD 또는 텔레그램 `/watch` 봇 명령. webai 계약: `GET /api/webai/trade-alert/monitor-set` · `POST /report`. 워커/프론트 탭은 web-ai/web-ui repo (설계: `docs/superpowers/specs/2026-07-02-realtime-trade-alerts-design.md`)
 **LLM provider 전환** — `LLM_PROVIDER` 환경변수
 - `claude` (기본): Anthropic Messages API (`claude-haiku-4-5`)
@@ -169,6 +170,8 @@ AI 에이전트 가상 오피스 — 2D 픽셀아트 사무실에서 4명의 에
 - **텔레그램 연동**: 양방향 알림 + 인라인 키보드 승인
  - 봇이 작업 결과를 텔레그램으로 푸시, 명령은 텔레그램에서 바로 에이전트에 전달
  - Webhook 검증 후 `chat.id` 기준 라우팅
  - **실시간 매매 알람 수신**: `POST /api/agent-office/stock/trade-alert` (stock이 edge 판정한 알람 push) → 텔레그램 본인+아내 발송. 봇 명령 `/watch`·`/unwatch`·`/watchlist`로 관심종목 관리
 - **분산 워커 관측**: `GET /api/agent-office/nodes`가 `worker:<name>:heartbeat`를 집계 → web-ui `/infra` 시각화 + 다운/복구/dead-letter 텔레그램 경보. WSL docker 워커는 `node_monitor.WORKER_REGISTRY` 등재 필수(위 주의사항 팀 규칙)
 #### 에이전트 구성
@@ -283,11 +286,11 @@ git push → Gitea → X-Gitea-Signature (HMAC SHA256)
 | DB | 소유 서비스 | 주요 테이블 |
 |----|------------|-----------|
 | `lotto.db` | lotto | draws, recommendations, simulation_runs/candidates, best_picks, purchase_history, strategy_performance/weights, weekly_reports, lotto_briefings |
-| `stock.db` | stock | articles, portfolio, broker_cash, asset_snapshots, sell_history |
+| `stock.db` | stock | articles, portfolio, broker_cash, asset_snapshots, sell_history, holdings_signals, news_sentiment, **watchlist, trade_alert_state, trade_alert_history** (실시간 매매 알람) |
 | `music.db` | music-lab | music_tasks, music_library (provider, lyrics, image_url, suno_id, file_hash, cover_images, wav_url, video_url, stem_urls), video_projects, revenue_records, market_trends, trend_reports |
 | `insta.db` | insta-lab | news_articles, trending_keywords (source 컬럼), card_slates, card_assets, generation_tasks, prompt_templates, account_preferences |
 | `realestate.db` | realestate-lab | announcements, announcement_models, user_profile, match_results, collect_log |
-| `agent_office.db` | agent-office | agent_config, agent_tasks, agent_logs, telegram_state, conversation_messages |
+| `agent_office.db` | agent-office | agent_config, agent_tasks, agent_logs, telegram_state, conversation_messages, youtube_research_jobs, lotto_signals/baselines, notified_failed_pipelines (파이프라인 실패 알림 dedup) |
 | `personal.db` | personal | profile, careers, projects, skills, introductions, todos, blog_posts |
 | `travel.db` | travel-proxy | photos (album, filename, mtime, has_thumb), album_covers |
 | `pack_files` (외부 Supabase) | packs-lab | filename, host_path, mime, byte_size, sha256, deleted_at |
@@ -384,6 +387,52 @@ PORTFOLIO_EDIT_PASSWORD=
 - **Suno CDN** — `cdn1.suno.ai` URL은 임시 만료 → 생성 즉시 로컬 다운로드 필수
 - **LLM provider 롤백** — Claude API 장애 시 `.env`의 `LLM_PROVIDER=ollama`로 전환 후 `docker compose up -d`
 - **시뮬레이션 교체 방식** — `best_picks`는 교체형 (`is_active=0` 비활성화 후 신규 입력)
 - **[팀 규칙] 모든 WSL docker 워커는 `/infra` 관측 필수** — 새 워커는 ① `worker:<name>:heartbeat`(EX45) 발신 ② BE가 `agent-office/app/node_monitor.py`의 `WORKER_REGISTRY`에 등재 ③ → `/api/agent-office/nodes`·web-ui `/infra` 노출 + 다운/복구/dead-letter 경보. 미준수 = 사일런트 사망 재발(insta-render 2주 사고). 워커 PR 머지 게이트
 - **Alpine + tzdata 함정** — stock 컨테이너는 `python:3.12-alpine` + tzdata 미설치라 `TZ=Asia/Seoul`이 무효 → `date.today()`가 UTC. KST 날짜는 `_today_kst()`(=`utcnow()+9h`) 명시 변환 필수 (아침 스케줄 리포트 하루 밀림 방지)
 ---
 ## 하네스 엔지니어링 (Claude Code 제어)
 이 레포는 Claude Code 세션의 동작을 `.claude/` 설정으로 **제어(harness engineering)** 한다. 모든 산출물은 git 추적되어 이 체크아웃의 모든 세션(co-gahusb 팀버스의 BE 역할 포함)에 공유된다.
 ### 제어 표면 (무엇을 통제하는가)
 | 레이어 | 메커니즘 | 위치 | 역할 |
 |--------|---------|------|------|
 | 컨텍스트 주입 | CLAUDE.md 계층 + 서비스 메모리 | `CLAUDE.md`, `memory/service_*.md` | 항상 로딩되는 카탈로그(불변) ↔ 관련 시 recall(가변) 2계층 |
 | 권한 가드 | permissions allow/deny/ask | `.claude/settings.json` | 읽기전용 명령 무프롬프트 / 시크릿·DB 차단 / push·reset 확인 |
 | 행동 강제 | PreToolUse·PostToolUse·SessionStart hook | `.claude/hooks/` | CLAUDE.md 주석 규칙을 하네스가 실제 차단·환기 |
 | 반복 워크플로우 | slash commands | `.claude/commands/` | `/co-inbox`, `/svc`, `/harness-audit` |
 | 전문 역할 | subagents | `.claude/agents/` | `be-developer`, `evaluator` |
 | 협업 버스 | MCP 서버 | `.mcp.json` | co-gahusb 팀버스(세션 간 메시지·작업·락) |
 ### 적용된 가드 (hook)
 | hook | 이벤트 / matcher | 동작 | 근거 |
 |------|-----------------|------|------|
 | `pretooluse-guard.sh` | PreToolUse · `Bash\|PowerShell` | **차단** 로컬 docker 변경(`up/down/build/restart/exec…`; ps·logs·config·images는 허용) | `feedback_docker_nas` |
 | 〃 | 〃 | **차단** `git commit --amend` · `git push --force`(`--force-with-lease`는 허용) | `feedback_concurrent_session_git_collision` |
 | 〃 | 〃 | **차단** PowerShell `>`/`>>` 파일 리다이렉트(UTF-16 BOM; `2>$null`·`> $null`은 허용) | `feedback_powershell_redirect_encoding` |
 | `posttooluse-memory.sh` | PostToolUse · `Edit\|Write` | 서비스 `db.py`/`models.py`/스케줄러/`.sql` 편집 시 `service_<name>.md` 갱신 환기(비차단) | 메모리 디스플린 |
 | `session-start.sh` | SessionStart · `startup\|resume` | BE 역할 + 수신함/락 넛지 주입 | 협업 버스 프로토콜 |
 차단 판단 로직은 `.claude/hooks/_guard.py`(Python). 래퍼는 파서 부재 시 **fail-open**(통과)하고, 출력은 UTF-8로 고정한다.
 ### slash commands
 | 커맨드 | 용도 |
 |--------|------|
 | `/co-inbox` | co-gahusb 팀버스 BE 수신함(inbox + tasks + locks) 일괄 확인 |
 | `/svc <name>` | 해당 `service_<name>.md` 메모리 + 핵심 파일 위치를 즉시 로드 |
 | `/harness-audit` | 서브에이전트 fan-out으로 CLAUDE.md 카탈로그 ↔ 실제 코드 드리프트 감사 |
 ### 확장 / 유지보수
 - **hook 경로는 이 머신 기준 절대경로**(`/c/Users/jaeoh/Desktop/workspace/web-backend/.claude/hooks/…`)다. 레포를 다른 경로로 클론하면 `settings.json`의 3개 hook command 경로를 갱신해야 한다.
 - 가드 패턴 추가/수정은 `_guard.py`만 고치면 된다(설정 변경 불필요).
 - hook은 새 세션에서 자동 로드된다. 진행 중 세션에 즉시 반영하려면 `/hooks` 메뉴를 열거나 재시작한다.
 - 메모리 디스플린: 코드 구조가 바뀌면 **CLAUDE.md(불변 카탈로그)** 가 아니라 **`service_*.md`(가변 상세)** 를 갱신한다.
 ---
@@ -391,3 +440,4 @@ PORTFOLIO_EDIT_PASSWORD=
 - `CLAUDE.md` — Claude Code 작업용 상세 컨텍스트 (API 전체 목록, 테이블 스키마 등)
 - `docs/` — 서비스별 기획·설계 문서
 - `.claude/` — 하네스 설정(settings·hooks·commands·agents). 위 "하네스 엔지니어링" 섹션 참조
--- a/agent-office/app/main.py
+++ b/agent-office/app/main.py
@@ -227,6 +227,24 @@ async def realestate_notify(body: RealestateNotifyBody):
    return await agent.on_new_matches(body.matches)
 # --- Realestate Listing Notify Endpoint (매물 알림, 청약 notify와 별도) ---
 class ListingNotifyBody(BaseModel):
    listings: List[Dict[str, Any]] = []
@app.post("/api/agent-office/realestate/notify-listing")
 async def realestate_notify_listing(body: ListingNotifyBody):
    from .notifiers.telegram_realestate_listing import send_listing_alerts
    from .db import add_log
    res = await send_listing_alerts(body.listings)
    for a in body.listings:
        add_log("realestate",
                 f"매물알림 {a.get('deal_type')} {a.get('complex_name')} "
                 f"{a.get('safety_tier') or a.get('valuation_tier')}", "info")
    return res
 # --- YouTube Research Agent Endpoints ---
 class YouTubeResearchBody(BaseModel):
--- a/agent-office/app/node_monitor.py
+++ b/agent-office/app/node_monitor.py
@@ -16,6 +16,7 @@ WORKER_REGISTRY = [
    {"name": "insta-render", "kind": "render", "queue": "queue:insta-render"},
    {"name": "task-watcher",  "kind": "watcher", "queue": None},
    {"name": "ai_trade",      "kind": "trader",  "queue": None},
    {"name": "trade-monitor", "kind": "trader",  "queue": None},
 ]
 _redis = None
@@ -92,7 +93,7 @@ async def collect_status(redis=None) -> dict:
    for w in out["workers"]:
        if w["kind"] == "trader":
-            out["links"].append({"from": "ai_trade", "to": "nas-stock", "type": "http-pull",
+            out["links"].append({"from": w["name"], "to": "nas-stock", "type": "http-pull",
                                 "status": "healthy" if w["alive"] else "down"})
        elif w["kind"] == "render":
            out["links"].append({"from": "nas", "to": w["name"], "type": "redis-queue",
--- a/agent-office/app/notifiers/telegram_realestate_listing.py
+++ b/agent-office/app/notifiers/telegram_realestate_listing.py
@@ -0,0 +1,83 @@
 """매물 알림 텔레그램 포맷+전송 (본인+아내 각각). realestate-lab notify_new_listings 수신.
 telegram_trade.py(매매알람)와 대칭 구조: send_raw 저수준 전송 + chat_id 리스트 순회.
 """
 import logging
 from typing import Any, Dict, List
 from ..telegram.messaging import send_raw
 from ..config import TELEGRAM_CHAT_ID, TELEGRAM_WIFE_CHAT_ID
 logger = logging.getLogger("agent-office")
 _TIER_EMOJI = {
    "안전": "🟢 안전", "주의": "🟡 주의", "위험": "🔴 위험", "보류": "⚪ 보류(표본부족)",
    "저평가": "🟢 저평가", "시세": "🟡 시세 수준", "고가": "🔴 고가",
 }
 def _manwon(v) -> str:
    if not v:
        return "-"
    return f"{v / 10000:.1f}억" if v >= 10000 else f"{v:,}만"
 def format_listing_alert(a: Dict[str, Any]) -> str:
    cat = a.get("category") or ("매매" if a.get("deal_type") == "매매" else "임차")
    if cat == "임차":
        price = _manwon(a.get("deposit"))
        tier = _TIER_EMOJI.get(a.get("safety_tier"), a.get("safety_tier") or "")
        ratio = a.get("jeonse_ratio")
        if ratio is not None:
            judge = (f"{tier} (전세가율 {int(ratio * 100)}% · 시세 {_manwon(a.get('market_median'))}, "
                     f"실거래 {a.get('sample_size', 0)}건)")
        else:
            judge = f"{tier} (실거래 표본 부족 — 수동 확인)"
        warn = "⚠️ 등기부 선순위 근저당 수동 확인 필수(인터넷등기소)"
    else:
        price = _manwon(a.get("sale_price"))
        tier = _TIER_EMOJI.get(a.get("valuation_tier"), a.get("valuation_tier") or "")
        ratio = a.get("price_ratio")
        budget = " · 예산 내 ✅" if a.get("budget_ok") else ""
        if ratio is not None:
            judge = (f"{tier} (호가율 {int(ratio * 100)}% · 실거래 median {_manwon(a.get('market_median'))}, "
                     f"{a.get('sample_size', 0)}건){budget}")
        else:
            judge = f"{tier} (실거래 표본 부족 — 수동 확인)"
        flags = a.get("regulation_flags") or []
        warn = "⚠️ " + (" · ".join(flags) if flags else "비토허") + " · 등기부 선순위 수동 확인 필수"
    area = f"전용 {a.get('area_exclusive')}㎡" if a.get("area_exclusive") else ""
    return (
        f"🏠 [{a.get('deal_type') or ''}] {a.get('complex_name') or ''} · {price} · {area} · {a.get('floor') or ''}\n"
        f"{judge}\n"
        f"📍 {a.get('dong') or ''} · {warn}\n"
        f"🔗 {a.get('url') or ''}"
    )
 async def send_listing_alerts(listings: List[Dict[str, Any]]) -> dict:
    """매물마다 본인+아내 chat_id 각각으로 send_raw. 1건 이상 성공하면 delivered→sent_ids에 id 수집.
    실패해도 나머지 계속 진행(per-send try/except)."""
    sent = 0
    sent_ids: List[Any] = []
    all_ok = True
    chat_ids = [c for c in (TELEGRAM_CHAT_ID, TELEGRAM_WIFE_CHAT_ID) if c]
    for a in listings:
        text = format_listing_alert(a)
        delivered = False
        for cid in chat_ids:
            try:
                r = await send_raw(text, chat_id=cid)
            except Exception as e:
                logger.warning(f"[telegram_realestate_listing] send failed (chat_id={cid}): {e}")
                all_ok = False
                continue
            if r.get("ok"):
                sent += 1
                delivered = True
            else:
                all_ok = False
        if delivered and a.get("id") is not None:
            sent_ids.append(a["id"])
    return {"sent": sent, "sent_ids": sent_ids, "ok": all_ok}
--- a/agent-office/app/notifiers/telegram_trade.py
+++ b/agent-office/app/notifiers/telegram_trade.py
@@ -13,15 +13,31 @@ _COND_LABEL = {
    "sell_stop_loss": "손절", "sell_ma_break": "이평 이탈", "sell_take_profit": "익절",
    "sell_climax": "급등 소진", "sell_trailing_stop": "트레일링 스톱",
 }
 # 조건별 "왜 이 시점에 매수/매도인가" 한 줄 근거
 _COND_REASON = {
    "buy_ma20_pullback": "상승추세 중 MA20 지지선 눌림목 반등 — 저가 진입 기회",
    "buy_breakout":      "전고점·저항 돌파 + 거래량 증가 — 추세 상승 진입 신호",
    "buy_rsi_bounce":    "RSI 과매도(30↓)에서 반등 — 단기 낙폭과대 되돌림",
    "sell_stop_loss":    "평단 대비 손절선 도달 — 추가 하락 리스크 차단",
    "sell_ma_break":     "주요 이평선(MA50/200) 이탈 — 추세 훼손, 보유 재검토",
    "sell_take_profit":  "목표 수익 도달 — 이익 실현 구간",
    "sell_climax":       "거래량 급증 + 윗꼬리(고점 대비 하락 마감) — 분산·소진 의심",
    "sell_trailing_stop":"보유기간 고점 대비 하락 — 수익 반납 방어(트레일링 스톱)",
 }
 def format_trade_alert(a: Dict[str, Any]) -> str:
    kind = _KIND_LABEL.get(a["kind"], a["kind"])
    cond = _COND_LABEL.get(a["condition"], a["condition"])
    reason = _COND_REASON.get(a["condition"], "")
    name = a.get("name") or a["ticker"]
    price = a.get("price")
    price_s = f"{int(price):,}원" if price else "-"
-    return f"{kind} 알람\n<b>{name}</b> ({a['ticker']})\n조건: {cond}\n현재가: {price_s}"
+    lines = [f"{kind} 알람", f"<b>{name}</b> ({a['ticker']})", f"조건: {cond}"]
    if reason:
        lines.append(f"💡 {reason}")
    lines.append(f"현재가: {price_s}")
    return "\n".join(lines)
 async def send_trade_alerts(alerts: List[Dict[str, Any]]) -> dict:
--- a/agent-office/tests/test_node_monitor.py
+++ b/agent-office/tests/test_node_monitor.py
@@ -68,6 +68,18 @@ async def test_trader_http_pull_link():
    link = next(l for l in st["links"] if l["from"] == "ai_trade")
    assert link["type"] == "http-pull" and link["status"] == "healthy"
@pytest.mark.asyncio
 async def test_trade_monitor_registered_and_own_link():
    """WSL 워커 trade-monitor가 registry에 있어 /nodes에 노출되고, 링크 from은
    ai_trade 하드코딩이 아니라 자기 이름(trade-monitor)이어야 한다 (다중 trader 구분)."""
    r = FakeRedis(kv={"worker:trade-monitor:heartbeat": _hb("trade-monitor", "trader", "market_open")})
    st = await node_monitor.collect_status(redis=r)
    tm = next(w for w in st["workers"] if w["name"] == "trade-monitor")
    assert tm["alive"] is True and tm["kind"] == "trader"
    link = next(l for l in st["links"] if l["from"] == "trade-monitor")
    assert link["type"] == "http-pull" and link["to"] == "nas-stock" and link["status"] == "healthy"
@pytest.mark.asyncio
 async def test_paused_no_watcher_heartbeat_fallback_reason():
    """paused=True인데 watcher heartbeat 없으면 paused_reason == 'trading' 폴백."""
--- a/agent-office/tests/test_realestate_listing_notify.py
+++ b/agent-office/tests/test_realestate_listing_notify.py
@@ -0,0 +1,78 @@
 """매물 알림 텔레그램 렌더+전송 테스트. 기존 test_trade_alert_notify.py 패턴(async mock + chat_id patch) 준수."""
 import pytest
 from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
 async def test_format_listing_rent_and_sale():
    from app.notifiers.telegram_realestate_listing import format_listing_alert
    rent = format_listing_alert({"id": 1, "category": "임차", "deal_type": "전세",
        "complex_name": "OO", "deposit": 29000, "area_exclusive": 42.0, "floor": "5/15",
        "safety_tier": "안전", "jeonse_ratio": 0.68, "market_median": 43000, "sample_size": 7,
        "dong": "신대방동", "url": "http://x", "regulation_flags": []})
    assert "전세" in rent and "안전" in rent and "OO" in rent and "68" in rent
    sale = format_listing_alert({"id": 2, "category": "매매", "deal_type": "매매",
        "complex_name": "PP", "sale_price": 89000, "area_exclusive": 59.0, "floor": "12/20",
        "valuation_tier": "시세", "price_ratio": 1.01, "market_median": 88000, "sample_size": 7,
        "dong": "상도동", "url": "http://y", "budget_ok": 1, "regulation_flags": []})
    assert "매매" in sale and "시세" in sale
@pytest.mark.asyncio
 async def test_format_listing_pending_sample_and_regulation_flags():
    """표본<3(보류)·토허 등 regulation_flags 경고 라인이 포함되는지."""
    from app.notifiers.telegram_realestate_listing import format_listing_alert
    pending = format_listing_alert({"id": 3, "category": "임차", "deal_type": "전세",
        "complex_name": "QQ", "deposit": 28000, "area_exclusive": 40.0, "floor": "3/10",
        "safety_tier": "보류", "jeonse_ratio": None, "market_median": None, "sample_size": 1,
        "dong": "봉천동", "url": "http://z", "regulation_flags": []})
    assert "보류" in pending
    toheo = format_listing_alert({"id": 4, "category": "매매", "deal_type": "매매",
        "complex_name": "RR", "sale_price": 150000, "area_exclusive": 84.0, "floor": "7/15",
        "valuation_tier": "고가", "price_ratio": 1.1, "market_median": 136000, "sample_size": 5,
        "dong": "대치동", "url": "http://w", "budget_ok": 0, "regulation_flags": ["토허"]})
    assert "토허" in toheo
@pytest.mark.asyncio
 async def test_send_listing_alerts_returns_sent_ids():
    from app.notifiers import telegram_realestate_listing as t
    items = [{"id": 5, "category": "임차", "deal_type": "전세", "complex_name": "OO",
              "deposit": 29000, "area_exclusive": 42.0, "safety_tier": "안전",
              "jeonse_ratio": 0.68, "market_median": 43000, "sample_size": 7,
              "dong": "신대방동", "url": "http://x", "regulation_flags": []}]
    with patch("app.notifiers.telegram_realestate_listing.send_raw",
               new=AsyncMock(return_value={"ok": True})) as m, \
         patch("app.notifiers.telegram_realestate_listing.TELEGRAM_CHAT_ID", "U"), \
         patch("app.notifiers.telegram_realestate_listing.TELEGRAM_WIFE_CHAT_ID", "W"):
        res = await t.send_listing_alerts(items)
    assert res["sent_ids"] == [5] and res["ok"] is True
    assert res["sent"] == 2   # 너+아내 둘 다
    chat_ids = {c.kwargs.get("chat_id") for c in m.await_args_list}
    assert chat_ids == {"U", "W"}
@pytest.mark.asyncio
 async def test_send_listing_alerts_partial_failure_not_marked_sent():
    """한쪽 chat_id 전송이 예외를 던져도 나머지는 계속 발송되고, 최소 1건 성공이면 sent_ids에 포함."""
    from app.notifiers import telegram_realestate_listing as t
    items = [{"id": 6, "category": "매매", "deal_type": "매매", "complex_name": "PP",
              "sale_price": 89000, "area_exclusive": 59.0, "valuation_tier": "시세",
              "price_ratio": 1.01, "market_median": 88000, "sample_size": 7,
              "dong": "상도동", "url": "http://y", "budget_ok": 1, "regulation_flags": []}]
    async def _flaky(text, chat_id=None, **kw):
        if chat_id == "U":
            raise RuntimeError("network error")
        return {"ok": True}
    with patch("app.notifiers.telegram_realestate_listing.send_raw", side_effect=_flaky), \
         patch("app.notifiers.telegram_realestate_listing.TELEGRAM_CHAT_ID", "U"), \
         patch("app.notifiers.telegram_realestate_listing.TELEGRAM_WIFE_CHAT_ID", "W"):
        res = await t.send_listing_alerts(items)
    assert res["sent_ids"] == [6]
    assert res["ok"] is False   # 일부 실패했으므로 all_ok=False
--- a/agent-office/tests/test_trade_alert_notify.py
+++ b/agent-office/tests/test_trade_alert_notify.py
@@ -53,3 +53,15 @@ async def test_format_trade_alert_has_direction():
    txt = format_trade_alert({"ticker": "005930", "name": "삼성전자", "kind": "sell",
                              "condition": "sell_stop_loss", "price": 60000, "detail": {}})
    assert "매도" in txt and "삼성전자" in txt
 def test_format_trade_alert_includes_reason_line():
    """조건별 '왜 매수/매도해야 하는지' 한 줄 이유(💡)가 메시지에 포함된다."""
    from app.notifiers.telegram_trade import format_trade_alert
    for cond in ("buy_breakout", "sell_stop_loss", "sell_trailing_stop"):
        txt = format_trade_alert({"ticker": "005930", "name": "삼성전자", "kind": cond.split("_")[0],
                                  "condition": cond, "price": 60000, "detail": {}})
        assert "💡" in txt, f"{cond}: 이유 한 줄 누락"
        # 이유 라인이 조건 라벨을 그대로 반복하지 않고 실제 설명을 담아야 함
        reason_line = next(l for l in txt.split("\n") if l.startswith("💡"))
        assert len(reason_line) > 6
--- a/realestate-lab/app/db.py
+++ b/realestate-lab/app/db.py
@@ -10,6 +10,9 @@ logger = logging.getLogger("realestate-lab")
 DB_PATH = os.getenv("REALESTATE_DB_PATH", "/app/data/realestate.db")
 # listing_matcher.MIN_SAMPLE와 동일 값 유지 — 시세 표본이 이 미만이면 광역 폴백.
 _MARKET_SAMPLE_MIN = 3
 def _conn():
    c = sqlite3.connect(DB_PATH, timeout=120.0)
@@ -177,6 +180,72 @@ def init_db():
            );
        """)
        # ── listings (매물 파이프라인) ──────────────────────────────────────
        conn.execute("""
            CREATE TABLE IF NOT EXISTS listings (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                article_no TEXT UNIQUE, source TEXT NOT NULL, deal_type TEXT NOT NULL,
                deposit INTEGER, monthly_rent INTEGER, sale_price INTEGER,
                area_exclusive REAL, floor TEXT, complex_name TEXT, address TEXT,
                dong TEXT, dong_code TEXT, url TEXT, posted_at TEXT, raw_json TEXT,
                created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
                notified_at TEXT
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS market_deals (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                house_type TEXT, dong_code TEXT, dong TEXT, complex_name TEXT, area REAL,
                deal_type TEXT, deposit INTEGER, monthly_rent INTEGER, sale_amount INTEGER,
                deal_ym TEXT, floor TEXT, source TEXT DEFAULT 'molit',
                created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
            )
        """)
        # 플랜 원안은 table-level UNIQUE(...)였으나 SQLite는 UNIQUE 제약에서 NULL을 서로 다른 값으로
        # 취급해 deposit/sale_amount(둘 중 하나는 항상 NULL)가 있는 행에서 dedup이 깨진다.
        # (표현식은 table-level UNIQUE에 사용 불가 → CREATE UNIQUE INDEX + COALESCE로 대체.
        #  실제 컬럼값은 그대로 NULL 유지 — downstream get_market_deals_for/listing_matcher 영향 없음.)
        conn.execute("""
            CREATE UNIQUE INDEX IF NOT EXISTS ux_market_deals_dedup ON market_deals(
                dong_code, COALESCE(complex_name,''), area, deal_ym, deal_type,
                COALESCE(deposit,-1), COALESCE(sale_amount,-1), COALESCE(floor,'')
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS listing_matches (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                listing_id INTEGER UNIQUE REFERENCES listings(id) ON DELETE CASCADE,
                category TEXT, passed INTEGER, match_score INTEGER,
                jeonse_ratio REAL, safety_tier TEXT, price_ratio REAL, valuation_tier TEXT,
                market_median INTEGER, sample_size INTEGER, budget_ok INTEGER,
                regulation_flags TEXT, reasons TEXT, is_new INTEGER,
                notified_at TEXT,
                created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS listing_criteria (
                id INTEGER PRIMARY KEY DEFAULT 1,
                dongs TEXT NOT NULL DEFAULT '[]', deal_types TEXT NOT NULL DEFAULT '[]',
                max_deposit INTEGER, max_sale_price INTEGER, min_area REAL,
                house_types TEXT NOT NULL DEFAULT '[]', min_safety_tier TEXT,
                notify_enabled INTEGER NOT NULL DEFAULT 1,
                equity INTEGER, annual_income INTEGER,
                is_homeless INTEGER NOT NULL DEFAULT 1, is_householder INTEGER NOT NULL DEFAULT 0,
                is_first_home INTEGER NOT NULL DEFAULT 0,
                updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS listing_collect_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                new_count INTEGER, total_count INTEGER, naver_ok INTEGER,
                error TEXT, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_md_lookup ON market_deals(dong_code, deal_type, area, deal_ym);")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_listing_dong ON listings(dong, deal_type);")
 # ── 상태 자동 계산 ───────────────────────────────────────────────────────────
@@ -841,3 +910,205 @@ def get_dashboard() -> Dict[str, Any]:
        "upcoming_schedules": schedules,
        "bookmarked": bookmarked_items,
    }
 # ── 매물 파이프라인 (listings/market_deals/listing_matches/listing_criteria) ──
 _LISTING_CRITERIA_SEED = {
    "dongs": ["신대방동", "대방동", "상도동", "봉천동", "대림동", "신길동"],
    "deal_types": ["전세", "반전세", "매매"],
    "max_deposit": 32000, "max_sale_price": None, "min_area": 40.0,
    "house_types": ["아파트", "오피스텔", "빌라"], "min_safety_tier": None,
    "notify_enabled": 1, "equity": None, "annual_income": None,
    "is_homeless": 1, "is_householder": 0, "is_first_home": 0,
 }
 _LC_JSON_COLS = ("dongs", "deal_types", "house_types")
 def get_listing_criteria() -> Dict[str, Any]:
    with _conn() as conn:
        row = conn.execute("SELECT * FROM listing_criteria WHERE id=1").fetchone()
        if row is None:
            conn.execute(
                """INSERT INTO listing_criteria
                   (id,dongs,deal_types,max_deposit,max_sale_price,min_area,house_types,
                    min_safety_tier,notify_enabled,equity,annual_income,is_homeless,is_householder,is_first_home)
                   VALUES (1,:dongs,:deal_types,:max_deposit,:max_sale_price,:min_area,:house_types,
                    :min_safety_tier,:notify_enabled,:equity,:annual_income,:is_homeless,:is_householder,:is_first_home)""",
                {**_LISTING_CRITERIA_SEED,
                 "dongs": json.dumps(_LISTING_CRITERIA_SEED["dongs"], ensure_ascii=False),
                 "deal_types": json.dumps(_LISTING_CRITERIA_SEED["deal_types"], ensure_ascii=False),
                 "house_types": json.dumps(_LISTING_CRITERIA_SEED["house_types"], ensure_ascii=False)},
            )
            row = conn.execute("SELECT * FROM listing_criteria WHERE id=1").fetchone()
    d = dict(row)
    for c in _LC_JSON_COLS:
        d[c] = json.loads(d[c] or "[]")
    return d
 def update_listing_criteria(fields: Dict[str, Any]) -> None:
    if not fields:
        return
    sets, vals = [], []
    for k, v in fields.items():
        if v is None and k not in ("max_deposit", "max_sale_price", "equity", "annual_income", "min_safety_tier"):
            continue
        sets.append(f"{k}=?")
        vals.append(json.dumps(v, ensure_ascii=False) if k in _LC_JSON_COLS else v)
    if not sets:
        return
    sets.append("updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')")
    with _conn() as conn:
        conn.execute("INSERT OR IGNORE INTO listing_criteria(id) VALUES(1)")
        conn.execute(f"UPDATE listing_criteria SET {','.join(sets)} WHERE id=1", vals)
 def upsert_listing(data: Dict[str, Any]) -> tuple:
    cols = ("article_no", "source", "deal_type", "deposit", "monthly_rent", "sale_price",
            "area_exclusive", "floor", "complex_name", "address", "dong", "dong_code",
            "url", "posted_at", "raw_json")
    d = {c: data.get(c) for c in cols}
    with _conn() as conn:
        exists = conn.execute("SELECT id FROM listings WHERE article_no=?", (d["article_no"],)).fetchone()
        is_new = exists is None
        conn.execute(f"""
            INSERT INTO listings ({','.join(cols)}) VALUES ({','.join(':'+c for c in cols)})
            ON CONFLICT(article_no) DO UPDATE SET
              deposit=excluded.deposit, monthly_rent=excluded.monthly_rent, sale_price=excluded.sale_price,
              floor=excluded.floor, complex_name=excluded.complex_name, address=excluded.address,
              url=excluded.url, posted_at=excluded.posted_at, raw_json=excluded.raw_json
        """, d)
        row = conn.execute("SELECT * FROM listings WHERE article_no=?", (d["article_no"],)).fetchone()
    return dict(row), is_new
 def upsert_market_deal(data: Dict[str, Any]) -> None:
    if data.get("area") is None:
        # area 없는 실거래는 area BETWEEN 조회에서 절대 매칭 안 돼 median 계산에 무용하고,
        # dedup 인덱스도 못 걸려(area가 NOT NULL 컬럼이 아니라 조건 매칭 불가) 무한 중복을 유발한다.
        return
    cols = ("house_type", "dong_code", "dong", "complex_name", "area", "deal_type",
            "deposit", "monthly_rent", "sale_amount", "deal_ym", "floor", "source")
    d = {c: data.get(c) for c in cols}
    d.setdefault("source", "molit")
    with _conn() as conn:
        conn.execute(f"""INSERT OR IGNORE INTO market_deals ({','.join(cols)})
                         VALUES ({','.join(':'+c for c in cols)})""", d)
 def get_market_deals_for(dong_code, complex_name, area, deal_type, months=6) -> List[Dict[str, Any]]:
    with _conn() as conn:
        def _query(cn):
            rows = conn.execute(
                """SELECT * FROM market_deals
                   WHERE dong_code=? AND deal_type=?
                     AND (complex_name=? OR ? IS NULL OR complex_name IS NULL)
                     AND area BETWEEN ? AND ?
                     AND deal_ym >= strftime('%Y%m', 'now', ?)""",
                (dong_code, deal_type, cn, cn,
                 (area or 0) - 5, (area or 0) + 5, f"-{int(months)} months"),
            ).fetchall()
            return [dict(r) for r in rows]
        deals = _query(complex_name)
        if len(deals) < _MARKET_SAMPLE_MIN and complex_name:
            # 단지 표본이 부족하면 complex 제약을 풀고 같은 동/거래유형/면적±5/최근성으로
            # 광역 재조회한다. 표본을 늘리기만 하므로 안전(잘못된 판정 유발 X).
            deals = _query(None)
    return deals
 def upsert_listing_match(data: Dict[str, Any]) -> None:
    cols = ("listing_id", "category", "passed", "match_score", "jeonse_ratio", "safety_tier",
            "price_ratio", "valuation_tier", "market_median", "sample_size", "budget_ok",
            "regulation_flags", "reasons", "is_new")
    d = {c: data.get(c) for c in cols}
    d["regulation_flags"] = json.dumps(data.get("regulation_flags") or [], ensure_ascii=False)
    d["reasons"] = json.dumps(data.get("reasons") or [], ensure_ascii=False)
    with _conn() as conn:
        conn.execute(f"""
            INSERT INTO listing_matches ({','.join(cols)}) VALUES ({','.join(':'+c for c in cols)})
            ON CONFLICT(listing_id) DO UPDATE SET
              passed=excluded.passed, match_score=excluded.match_score,
              jeonse_ratio=excluded.jeonse_ratio, safety_tier=excluded.safety_tier,
              price_ratio=excluded.price_ratio, valuation_tier=excluded.valuation_tier,
              market_median=excluded.market_median, sample_size=excluded.sample_size,
              budget_ok=excluded.budget_ok, regulation_flags=excluded.regulation_flags,
              reasons=excluded.reasons, category=excluded.category
        """, d)
 def get_listings(dong=None, deal_type=None, tier=None, matched_only=False, limit=50, offset=0) -> List[Dict[str, Any]]:
    sql = ("SELECT l.*, m.safety_tier, m.valuation_tier, m.category, m.match_score, m.passed "
           "FROM listings l LEFT JOIN listing_matches m ON m.listing_id=l.id WHERE 1=1")
    p = []
    if dong: sql += " AND l.dong=?"; p.append(dong)
    if deal_type: sql += " AND l.deal_type=?"; p.append(deal_type)
    if tier: sql += " AND (m.safety_tier=? OR m.valuation_tier=?)"; p += [tier, tier]
    if matched_only: sql += " AND m.passed=1"
    sql += " ORDER BY l.created_at DESC LIMIT ? OFFSET ?"; p += [limit, offset]
    with _conn() as conn:
        return [dict(r) for r in conn.execute(sql, p).fetchall()]
 def get_listing_matches() -> List[Dict[str, Any]]:
    with _conn() as conn:
        rows = conn.execute(
            "SELECT m.*, l.complex_name, l.dong, l.deal_type, l.deposit, l.sale_price, l.area_exclusive, l.url "
            "FROM listing_matches m JOIN listings l ON l.id=m.listing_id ORDER BY m.created_at DESC"
        ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        d["regulation_flags"] = json.loads(d.get("regulation_flags") or "[]")
        d["reasons"] = json.loads(d.get("reasons") or "[]")
        out.append(d)
    return out
 _TIER_ORDER = {"위험": 0, "고가": 0, "주의": 1, "시세": 1, "안전": 2, "저평가": 2, "보류": -1}
 def get_unnotified_listing_matches(min_tier=None) -> List[Dict[str, Any]]:
    with _conn() as conn:
        rows = conn.execute(
            "SELECT m.*, l.complex_name, l.dong, l.deal_type, l.deposit, l.monthly_rent, "
            "l.sale_price, l.area_exclusive, l.floor, l.url "
            "FROM listing_matches m JOIN listings l ON l.id=m.listing_id "
            "WHERE m.passed=1 AND m.notified_at IS NULL"
        ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        d["regulation_flags"] = json.loads(d.get("regulation_flags") or "[]")
        d["reasons"] = json.loads(d.get("reasons") or "[]")
        out.append(d)
    if min_tier:
        floor = _TIER_ORDER.get(min_tier, -1)
        out = [d for d in out if _TIER_ORDER.get(d.get("safety_tier") or d.get("valuation_tier"), -1) >= floor]
    return out
 def mark_listings_notified(match_ids: List[int]) -> None:
    if not match_ids:
        return
    with _conn() as conn:
        conn.executemany(
            "UPDATE listing_matches SET notified_at=strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id=?",
            [(i,) for i in match_ids],
        )
 def save_listing_collect_log(new_count, total_count, naver_ok, error=None) -> None:
    with _conn() as conn:
        conn.execute(
            "INSERT INTO listing_collect_log (new_count,total_count,naver_ok,error) VALUES (?,?,?,?)",
            (new_count, total_count, 1 if naver_ok else 0, error),
        )
 def get_last_listing_collect_log() -> Optional[Dict[str, Any]]:
    with _conn() as conn:
        row = conn.execute("SELECT * FROM listing_collect_log ORDER BY id DESC LIMIT 1").fetchone()
    return dict(row) if row else None
--- a/realestate-lab/app/finance_rules.py
+++ b/realestate-lab/app/finance_rules.py
@@ -0,0 +1,67 @@
 """전세/매수 예산·규제 판정 (순수 함수, 외부 호출 없음)."""
 from . import finance_rules_config as cfg
 def region_regulation(dong: str) -> dict:
    d = (dong or "").strip()
    is_toheo = d in cfg.TOHEO_DONGS
    is_regulated = d in cfg.REGULATED_DONGS or is_toheo
    notes = []
    if is_toheo:
        notes.append("토지거래허가구역 — 매수 시 실거주 의무(2년)+허가 필요, 갭투자 금지")
    if is_regulated:
        notes.append("규제지역 — LTV 강화")
    notes.append("토허/규제 여부는 서울시 최신 고시로 확인 필요")
    return {"is_toheo": is_toheo, "is_regulated": is_regulated, "notes": " · ".join(notes)}
 def _dsr_max_loan(annual_income: int) -> int:
    """DSR 한도 내 대출 원금 근사(만원). 원리금균등, 가정 금리·만기."""
    if not annual_income or annual_income <= 0:
        return 0
    annual_pay_cap = annual_income * cfg.DSR_LIMIT_PCT / 100.0     # 연 상환 가능액
    r = cfg.ASSUMED_RATE_PCT / 100.0 / 12
    n = cfg.ASSUMED_TERM_YEARS * 12
    monthly_cap = annual_pay_cap / 12.0
    if r == 0:
        return int(monthly_cap * n)
    # PV = PMT * (1-(1+r)^-n)/r
    pv = monthly_cap * (1 - (1 + r) ** (-n)) / r
    return int(pv)
 def estimate_jeonse_budget(equity: int, annual_income: int, jeonse_loan_ratio: float = None) -> dict:
    equity = equity or 0
    ratio = jeonse_loan_ratio if jeonse_loan_ratio is not None else cfg.JEONSE_LOAN_RATIO
    # 전세대출 한도 = min(보증기관 절대상한, DSR 한도). 보증금 비례는 매물별로 matcher에서 재확인.
    loan_limit = min(cfg.JEONSE_LOAN_CAP, _dsr_max_loan(annual_income) or cfg.JEONSE_LOAN_CAP)
    max_deposit = equity + loan_limit
    return {
        "loan_limit": loan_limit,
        "max_deposit": max_deposit,
        "notes": f"전세대출 한도≈{loan_limit}만(보증기관·DSR 근사, 보증금 대비 {int(ratio*100)}% 이내). {cfg.DISCLAIMER}",
    }
 def estimate_purchase_budget(equity: int, annual_income: int, region_flags: dict,
                             is_first_home: bool, is_homeless: bool) -> dict:
    equity = equity or 0
    reg = "regulated" if (region_flags or {}).get("is_regulated") else "normal"
    first = "first" if is_first_home else "normal"
    ltv_pct = cfg.LTV_TABLE[(reg, first)]
    # 대출가능액 = min(수도권 한도, DSR 한도). LTV는 매매가에 따라 matcher에서 상한 재확인.
    dsr_cap = _dsr_max_loan(annual_income)
    loan_cap = min(cfg.SUDOGWON_MORTGAGE_CAP, dsr_cap) if dsr_cap else cfg.SUDOGWON_MORTGAGE_CAP
    max_price = equity + loan_cap
    flags = []
    if (region_flags or {}).get("is_toheo"):
        flags.append("토허")
    if (region_flags or {}).get("is_regulated"):
        flags.append("규제지역")
    return {
        "ltv_pct": ltv_pct,
        "loan_cap": loan_cap,
        "max_price": max_price,
        "dsr_note": f"DSR {cfg.DSR_LIMIT_PCT}% 한도 근사(금리{cfg.ASSUMED_RATE_PCT}%·{cfg.ASSUMED_TERM_YEARS}년)",
        "regulation_flags": flags,
    }
--- a/realestate-lab/app/finance_rules_config.py
+++ b/realestate-lab/app/finance_rules_config.py
@@ -0,0 +1,35 @@
 """정책 파라미터 (외부 호출 없음). ⚠️ 정부 정책 수시 변동 — 계약·대출 전 은행/서울시 고시로 최종 확인.
 기준: 2025 6·27 대책(수도권 주담대 6억 한도) 등. 실제 집행 전 최신값으로 갱신."""
 # 수도권 주택담보대출 총액 한도 (만원). 6·27 대책: 6억.
 SUDOGWON_MORTGAGE_CAP = 60000
 # LTV(%) — (규제지역 여부, 생애최초 여부) → 한도. 무주택 기준.
 LTV_TABLE = {
    ("regulated", "first"):  70,
    ("regulated", "normal"): 50,
    ("normal",    "first"):  80,
    ("normal",    "normal"): 70,
 }
 # 스트레스 DSR 한도(%) — 연소득 대비 연원리금.
 DSR_LIMIT_PCT = 40
 # 대출 산정용 가정 금리(연%)·만기(년) — 원리금균등 근사.
 ASSUMED_RATE_PCT = 4.5
 ASSUMED_TERM_YEARS = 30
 # 전세대출: 보증기관 보증비율(보증금 대비) 상한 근사.
 JEONSE_LOAN_RATIO = 0.80
 # 전세대출 절대 상한(만원) — 보증기관 한도 근사.
 JEONSE_LOAN_CAP = 44000
 # 토지거래허가구역(구·동 단위, 서울시 고시 기준, 수동 갱신). 소수만.
 TOHEO_DONGS = {
    "대치동", "삼성동", "청담동", "잠실동",      # 강남·송파 국제교류복합지구권
    "압구정동", "여의도동", "목동", "성수동",     # 재건축 토허
    "이촌동", "한남동",                          # 용산
 }
 # 규제지역(투기과열/조정) — 매수 LTV 강화 대상.
 REGULATED_DONGS = set(TOHEO_DONGS) | {"반포동", "서초동"}
 DISCLAIMER = "정책(LTV·DSR·수도권 한도·토허)은 수시 변동 — 계약·대출 전 은행/중개사/서울시 고시로 최종 확인."
--- a/realestate-lab/app/lawd_codes.py
+++ b/realestate-lab/app/lawd_codes.py
@@ -0,0 +1,24 @@
 """대상 동명 → 법정동코드(시군구 5자리, LAWD_CD). 국토부 실거래가 API 파라미터용.
 동작구/관악구/영등포구 중심(스펙 초기 대상 동). 신규 동 추가 시 여기에.
 출처: 행정표준코드관리시스템 시군구 코드."""
 # 시군구(구) 단위 5자리 — 실거래가 API는 LAWD_CD=시군구코드를 받는다.
 _DISTRICT_CODE = {
    "동작구": "11590",
    "관악구": "11620",
    "영등포구": "11560",
    "서초구": "11650",
    "강남구": "11680",
 }
 # 동 → 소속 구
 _DONG_TO_DISTRICT = {
    "신대방동": "동작구", "대방동": "동작구", "상도동": "동작구", "노량진동": "동작구", "사당동": "동작구",
    "봉천동": "관악구", "신림동": "관악구",
    "신길동": "영등포구", "대림동": "영등포구", "영등포동": "영등포구",
 }
 LAWD_CODES = {dong: _DISTRICT_CODE[d] for dong, d in _DONG_TO_DISTRICT.items() if d in _DISTRICT_CODE}
 def lawd_code(dong: str) -> str | None:
    return LAWD_CODES.get((dong or "").strip())
--- a/realestate-lab/app/listing_collector.py
+++ b/realestate-lab/app/listing_collector.py
@@ -0,0 +1,166 @@
 """매물 수집: 국토부 실거래가(합법, market_deals) + 네이버부동산 호가(회색·폴백, listings)."""
 import os
 import re
 import time
 import logging
 import requests
 from datetime import date
 from .lawd_codes import LAWD_CODES
 from .db import (get_listing_criteria, upsert_listing, upsert_market_deal,
                 save_listing_collect_log)
 logger = logging.getLogger("realestate-lab")
 MOLIT_KEY = os.getenv("DATA_GO_KR_API_KEY", "")
 NAVER_BASE = "https://new.land.naver.com/api"
 # MOLIT operation ID (스펙 §3, 라이브 확인 필요). house_type → (rent_op, trade_op)
 MOLIT_OPS = {
    "아파트": ("getRTMSDataSvcAptRent", "getRTMSDataSvcAptTradeDev"),
    "오피스텔": ("getRTMSDataSvcOffiRent", "getRTMSDataSvcOffiTrade"),
    "연립다세대": ("getRTMSDataSvcRHRent", "getRTMSDataSvcRHTrade"),
 }
 _HOUSE_TYPE_ALIAS = {"빌라": "연립다세대"}
 def _won_to_manwon(s) -> int | None:
    """'43,000' 또는 '2억 9,000' → 만원 정수."""
    if s is None:
        return None
    t = str(s).replace(",", "").strip()
    if not t:
        return None
    m = re.match(r"(?:(\d+)억)?\s*(\d+)?", t)
    if m and (m.group(1) or m.group(2)):
        eok = int(m.group(1)) if m.group(1) else 0
        man = int(m.group(2)) if m.group(2) else 0
        return eok * 10000 + man
    try:
        return int(float(t))
    except ValueError:
        return None
 def _parse_molit_rent(raw: dict, house_type: str) -> dict:
    deposit = _won_to_manwon(raw.get("deposit"))
    rent = _won_to_manwon(raw.get("monthlyRent"))
    return {"house_type": house_type, "complex_name": (raw.get("aptNm") or raw.get("offiNm") or "").strip(),
            "area": float(raw.get("excluUseAr") or 0) or None, "deal_type": "전세" if not rent else "월세",
            "deposit": deposit, "monthly_rent": rent, "sale_amount": None,
            "deal_ym": f"{raw.get('dealYear','')}{str(raw.get('dealMonth','')).zfill(2)}",
            "floor": str(raw.get("floor") or ""), "source": "molit"}
 def _parse_molit_trade(raw: dict, house_type: str) -> dict:
    return {"house_type": house_type, "complex_name": (raw.get("aptNm") or raw.get("offiNm") or "").strip(),
            "area": float(raw.get("excluUseAr") or 0) or None, "deal_type": "매매",
            "deposit": None, "monthly_rent": None,
            "sale_amount": _won_to_manwon(raw.get("dealAmount")),
            "deal_ym": f"{raw.get('dealYear','')}{str(raw.get('dealMonth','')).zfill(2)}",
            "floor": str(raw.get("floor") or ""), "source": "molit"}
 def _parse_naver_article(raw: dict) -> dict:
    dt = raw.get("tradeTypeName") or ""
    prc = _won_to_manwon(raw.get("dealOrWarrantPrc"))
    rent = _won_to_manwon(raw.get("rentPrc"))
    is_sale = dt == "매매"
    return {"article_no": str(raw.get("articleNo") or ""), "source": "naver", "deal_type": dt,
            "deposit": None if is_sale else prc, "monthly_rent": rent,
            "sale_price": prc if is_sale else None,
            "area_exclusive": float(raw.get("area2") or raw.get("areaName") or 0) or None,
            "floor": raw.get("floorInfo"), "complex_name": raw.get("articleName") or raw.get("buildingName"),
            "dong_code": (raw.get("cortarNo") or "")[:5] or None,
            "url": f"https://new.land.naver.com/houses?articleNo={raw.get('articleNo')}",
            "raw_json": None}
 def _molit_call(op: str, lawd: str, ym: str) -> list:
    """국토부 실거래 단건 호출 (기존 collector 재시도 관용). 실패 시 []."""
    if not MOLIT_KEY:
        return []
    url = f"https://apis.data.go.kr/1613000/RTMSDataSvc{op[len('getRTMSDataSvc'):]}/{op}"
    for attempt in range(3):
        try:
            resp = requests.get(url, params={"serviceKey": MOLIT_KEY, "LAWD_CD": lawd,
                                             "DEAL_YMD": ym, "numOfRows": 200, "_type": "json"}, timeout=30)
            resp.raise_for_status()
            body = resp.json()
            items = (((body.get("response") or {}).get("body") or {}).get("items") or {}).get("item") or []
            return items if isinstance(items, list) else [items]
        except requests.RequestException as e:
            if attempt < 2:
                time.sleep(2 ** attempt)
            else:
                logger.error("MOLIT %s 실패: %s", op, e)
    return []
 def _fetch_molit_all() -> int:
    """대상 동 시군구코드 × 유형 × 최근 2개월 전월세·매매 실거래 수집→market_deals. 저장건수 반환."""
    saved = 0
    lawds = set(LAWD_CODES.values())
    today = date.today()
    yms = [today.strftime("%Y%m"),
           (today.replace(day=1) - __import__("datetime").timedelta(days=1)).strftime("%Y%m")]
    for lawd in lawds:
        for htype, (rent_op, trade_op) in MOLIT_OPS.items():
            for ym in yms:
                for op, parser in ((rent_op, _parse_molit_rent), (trade_op, _parse_molit_trade)):
                    for raw in _molit_call(op, lawd, ym):
                        try:
                            d = parser(raw, htype)
                            d["dong_code"] = lawd
                            d["dong"] = (raw.get("umdNm") or "").strip() or None
                            upsert_market_deal(d)
                            saved += 1
                        except Exception as e:
                            logger.warning("MOLIT 파싱 실패: %s", e)
                    time.sleep(0.3)
    return saved
 def _fetch_naver_all() -> tuple:
    """대상 동 네이버 호가 매물 수집→listings. (new_count, total_count) 반환. 예외는 상위로 전파."""
    new_count = total = 0
    crit = get_listing_criteria()
    for dong in crit.get("dongs", []):
        code = LAWD_CODES.get(dong)
        if not code:
            continue
        # 저빈도: 동당 1회, UA/Referer 현실값. 응답 articleList 순회.
        resp = requests.get(f"{NAVER_BASE}/articles",
                            params={"cortarNo": code, "order": "dateDesc", "page": 1},
                            headers={"User-Agent": "Mozilla/5.0", "Referer": "https://new.land.naver.com/"},
                            timeout=20)
        resp.raise_for_status()
        for raw in (resp.json().get("articleList") or []):
            try:
                d = _parse_naver_article(raw)
                if not d["article_no"]:
                    continue
                d["dong"] = dong
                _, is_new = upsert_listing(d)
                total += 1
                if is_new:
                    new_count += 1
            except Exception as e:
                logger.warning("네이버 파싱 실패: %s", e)
        time.sleep(1.5)
    return new_count, total
 def collect_listings() -> dict:
    naver_ok = True
    new_count = total = 0
    try:
        _fetch_molit_all()   # 실거래(합법)는 항상 우선 — 안전마진 baseline
    except Exception as e:
        logger.error("MOLIT 수집 오류(안전마진 baseline): %s", e)
    try:
        new_count, total = _fetch_naver_all()
    except requests.RequestException as e:
        naver_ok = False
        logger.warning("네이버 매물수집 skip(차단/오류) — 안전마진·실거래는 유지: %s", e)
    save_listing_collect_log(new_count, total, naver_ok)
    return {"new_count": new_count, "total_count": total, "naver_ok": naver_ok}
--- a/realestate-lab/app/listing_matcher.py
+++ b/realestate-lab/app/listing_matcher.py
@@ -0,0 +1,114 @@
 """매물 조건 매칭 + 임차 안전마진(전세가율)·매매 적정성(호가율) 판정."""
 import logging
 import statistics
 from .db import (get_listing_criteria, get_market_deals_for, upsert_listing_match,
                 get_listings)
 from . import finance_rules
 logger = logging.getLogger("realestate-lab")
 # 경계 상수(조정 가능)
 JEONSE_SAFE, JEONSE_WARN = 0.70, 0.80
 VALUATION_LOW, VALUATION_HIGH = 0.97, 1.05
 MIN_SAMPLE = 3
 DISCLAIMER = "등기부 선순위 근저당은 자동 확인 불가 — 계약 전 인터넷등기소 열람 필수."
 def compute_safety(deposit, deal_type, deals) -> dict:
    vals = [d["deposit"] for d in deals if d.get("deposit")]
    if len(vals) < MIN_SAMPLE:
        return {"jeonse_ratio": None, "tier": "보류", "median": None, "sample": len(vals)}
    median = statistics.median(vals)
    ratio = (deposit or 0) / median if median else None
    if ratio is None:
        tier = "보류"
    elif ratio <= JEONSE_SAFE:
        tier = "안전"
    elif ratio <= JEONSE_WARN:
        tier = "주의"
    else:
        tier = "위험"
    return {"jeonse_ratio": round(ratio, 3) if ratio else None,
            "tier": tier, "median": int(median), "sample": len(vals)}
 def compute_valuation(sale_price, deals) -> dict:
    vals = [d["sale_amount"] for d in deals if d.get("sale_amount")]
    if len(vals) < MIN_SAMPLE:
        return {"price_ratio": None, "tier": "보류", "median": None, "sample": len(vals)}
    median = statistics.median(vals)
    ratio = (sale_price or 0) / median if median else None
    if ratio is None:
        tier = "보류"
    elif ratio <= VALUATION_LOW:
        tier = "저평가"
    elif ratio <= VALUATION_HIGH:
        tier = "시세"
    else:
        tier = "고가"
    return {"price_ratio": round(ratio, 3) if ratio else None,
            "tier": tier, "median": int(median), "sample": len(vals)}
 def match_listing(criteria, listing, budget) -> dict:
    reasons, passed = [], True
    if criteria.get("dongs") and listing.get("dong") not in criteria["dongs"]:
        passed = False; reasons.append("동 불일치")
    if criteria.get("deal_types") and listing.get("deal_type") not in criteria["deal_types"]:
        passed = False; reasons.append("거래유형 불일치")
    if criteria.get("min_area") and (listing.get("area_exclusive") or 0) < criteria["min_area"]:
        passed = False; reasons.append("면적 미달")
    if listing.get("deal_type") == "매매":
        cap = criteria.get("max_sale_price") or (budget or {}).get("max_price")
        if cap and (listing.get("sale_price") or 0) > cap:
            passed = False; reasons.append("매수가능 상한 초과")
    else:  # 임차
        cap = criteria.get("max_deposit") or (budget or {}).get("max_deposit")
        if cap and (listing.get("deposit") or 0) > cap:
            passed = False; reasons.append("보증금 상한 초과")
    score = 100 if passed else 0
    if passed:
        reasons.append("조건 통과")
    return {"passed": passed, "score": score, "reasons": reasons}
 def run_listing_matching() -> None:
    """미평가 포함 전체 매물 재평가 → listing_matches upsert (청약 run_matching 대칭)."""
    criteria = get_listing_criteria()
    equity = criteria.get("equity") or 0
    income = criteria.get("annual_income") or 0
    jeonse_budget = finance_rules.estimate_jeonse_budget(equity, income)
    for l in get_listings(limit=1000):
        is_rent = l["deal_type"] != "매매"
        reg = finance_rules.region_regulation(l.get("dong"))
        if is_rent:
            budget = jeonse_budget
        else:
            budget = finance_rules.estimate_purchase_budget(
                equity, income, reg, criteria.get("is_first_home"), criteria.get("is_homeless"))
        m = match_listing(criteria, l, budget)
        deals = get_market_deals_for(l.get("dong_code"), l.get("complex_name"),
                                     l.get("area_exclusive"), "전세" if is_rent else "매매")
        rec = {"listing_id": l["id"], "category": "임차" if is_rent else "매매",
               "passed": 1 if m["passed"] else 0, "match_score": m["score"],
               "sample_size": 0, "reasons": m["reasons"], "is_new": 1,
               "regulation_flags": reg_flags(reg)}
        if is_rent:
            s = compute_safety(l.get("deposit"), l["deal_type"], deals)
            rec.update({"jeonse_ratio": s["jeonse_ratio"], "safety_tier": s["tier"],
                        "market_median": s["median"], "sample_size": s["sample"]})
        else:
            v = compute_valuation(l.get("sale_price"), deals)
            rec.update({"price_ratio": v["price_ratio"], "valuation_tier": v["tier"],
                        "market_median": v["median"], "sample_size": v["sample"],
                        "budget_ok": 1 if (l.get("sale_price") or 0) <= (budget.get("max_price") or 0) else 0})
        upsert_listing_match(rec)
    logger.info("매물 매칭 완료")
 def reg_flags(reg: dict) -> list:
    f = []
    if reg.get("is_toheo"): f.append("토허")
    if reg.get("is_regulated"): f.append("규제지역")
    return f
--- a/realestate-lab/app/main.py
+++ b/realestate-lab/app/main.py
@@ -15,11 +15,20 @@ from .db import (
    get_profile, upsert_profile, get_matches, mark_match_read,
    get_last_collect_log, get_dashboard,
    delete_old_completed_announcements,
    get_listing_criteria, update_listing_criteria, get_listings, get_listing_matches,
    get_last_listing_collect_log, get_market_deals_for,
 )
 from .collector import collect_all
 from .matcher import run_matching
-from .notifier import notify_new_matches
+from .notifier import notify_new_matches, notify_new_listings
-from .models import AnnouncementCreate, AnnouncementUpdate, ProfileUpdate
+from .listing_collector import collect_listings
 from .listing_matcher import run_listing_matching, compute_safety, compute_valuation
 from . import finance_rules
 from .finance_rules_config import DISCLAIMER
 from .models import (
    AnnouncementCreate, AnnouncementUpdate, ProfileUpdate,
    ListingCriteriaUpdate, SafetyCheckRequest, BudgetRequest,
 )
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s %(message)s")
 logger = logging.getLogger("realestate-lab")
@@ -56,12 +65,29 @@ def scheduled_status_update():
    logger.info("상태 갱신 + 재매칭 완료")
 _listing_collect_lock = threading.Lock()
 def _run_listing_pipeline():
    if not _listing_collect_lock.acquire(blocking=False):
        logger.info("매물 수집 이미 진행 중 — 건너뜀"); return
    try:
        collect_listings(); run_listing_matching(); notify_new_listings()
    finally:
        _listing_collect_lock.release()
 def scheduled_listing_collect():
    logger.info("매물 스케줄 수집 시작"); _run_listing_pipeline()
@asynccontextmanager
 async def lifespan(app: FastAPI):
    init_db()
    # 09:00 cron 스태거링 — agent-office 09:00/05/10 이후 (CHECK_POINT FU-A)
    scheduler.add_job(scheduled_collect, "cron", hour=9, minute=15, id="collect")
    scheduler.add_job(scheduled_status_update, "cron", hour=0, minute=0, id="status_update")
    scheduler.add_job(scheduled_listing_collect, "cron", hour="8,11,14,17,20", minute=40, id="listing_collect")
    scheduler.start()
    logger.info("realestate-lab 시작")
    yield
@@ -217,3 +243,66 @@ def api_match_read(match_id: int):
@app.get("/api/realestate/dashboard")
 def api_dashboard():
    return get_dashboard()
 # ── 매물 API ─────────────────────────────────────────────────────────────────
@app.get("/api/realestate/listings")
 def api_listings(dong: str = None, deal_type: str = None, tier: str = None,
                 matched_only: bool = False, page: int = 1, size: int = 50):
    return {"listings": get_listings(dong, deal_type, tier, matched_only, size, (page - 1) * size)}
@app.post("/api/realestate/listings/collect")
 def api_listings_collect(background_tasks: BackgroundTasks):
    background_tasks.add_task(_run_listing_pipeline)
    return {"ok": True, "message": "매물 수집 시작됨"}
@app.get("/api/realestate/listings/collect/status")
 def api_listings_collect_status():
    return get_last_listing_collect_log() or {"status": "never_run"}
@app.get("/api/realestate/listings/criteria")
 def api_get_listing_criteria():
    return get_listing_criteria()
@app.put("/api/realestate/listings/criteria")
 def api_put_listing_criteria(req: ListingCriteriaUpdate):
    update_listing_criteria(req.model_dump(exclude_none=True))
    return get_listing_criteria()
@app.get("/api/realestate/listings/matches")
 def api_listing_matches():
    return {"matches": get_listing_matches()}
@app.post("/api/realestate/safety-check")
 def api_safety_check(req: SafetyCheckRequest):
    deals = get_market_deals_for(req.dong_code, req.complex_name, req.area,
                                 "전세" if req.deal_type != "매매" else "매매")
    reg = finance_rules.region_regulation(req.dong)   # dong명(없으면 토허 판정 생략)
    if req.deal_type == "매매":
        res = compute_valuation(req.amount, deals)
        ratio_key = "price_ratio"
    else:
        res = compute_safety(req.amount, req.deal_type, deals)
        ratio_key = "jeonse_ratio"
    return {"median": res["median"], "ratio": res.get(ratio_key), "tier": res["tier"],
            "sample": res["sample"], "is_toheo": reg["is_toheo"],
            "disclaimer": "등기부 선순위·토허 허가·실거주 의무는 계약 전 수동 확인 필수. " + DISCLAIMER}
@app.post("/api/realestate/budget")
 def api_budget(req: BudgetRequest):
    reg = finance_rules.region_regulation(req.target_dong)
    return {
        "jeonse": finance_rules.estimate_jeonse_budget(req.equity, req.annual_income),
        "purchase": finance_rules.estimate_purchase_budget(req.equity, req.annual_income, reg,
                                                           req.is_first_home, req.is_homeless),
        "region": reg,
        "disclaimer": DISCLAIMER,
    }
--- a/realestate-lab/app/models.py
+++ b/realestate-lab/app/models.py
@@ -84,3 +84,38 @@ class ProfileUpdate(BaseModel):
    preferred_districts: Optional[Dict[str, List[str]]] = None
    min_match_score: Optional[int] = Field(default=None, ge=0, le=100)
    notify_enabled: Optional[bool] = None
 class ListingCriteriaUpdate(BaseModel):
    dongs: Optional[List[str]] = None
    deal_types: Optional[List[str]] = None
    max_deposit: Optional[int] = None
    max_sale_price: Optional[int] = None
    min_area: Optional[float] = None
    house_types: Optional[List[str]] = None
    min_safety_tier: Optional[str] = None
    notify_enabled: Optional[bool] = None
    equity: Optional[int] = None
    annual_income: Optional[int] = None
    is_homeless: Optional[bool] = None
    is_householder: Optional[bool] = None
    is_first_home: Optional[bool] = None
 class SafetyCheckRequest(BaseModel):
    complex_name: Optional[str] = None
    address: Optional[str] = None
    dong: Optional[str] = None
    dong_code: Optional[str] = None
    area: float
    deal_type: str
    amount: int
 class BudgetRequest(BaseModel):
    equity: int
    annual_income: int
    is_homeless: bool = True
    is_householder: bool = False
    is_first_home: bool = False
    target_dong: Optional[str] = None
--- a/realestate-lab/app/notifier.py
+++ b/realestate-lab/app/notifier.py
@@ -3,7 +3,8 @@ import os
 import logging
 import requests
-from .db import get_profile, get_unnotified_matches, mark_matches_notified
+from .db import (get_profile, get_unnotified_matches, mark_matches_notified,
                 get_listing_criteria, get_unnotified_listing_matches, mark_listings_notified)
 logger = logging.getLogger("realestate-lab")
@@ -44,3 +45,27 @@ def notify_new_matches() -> dict:
        mark_matches_notified(sent_ids)
        logger.info("알림 송신: %d건", len(sent_ids))
    return body
 def notify_new_listings() -> dict:
    """임계 통과 미알림 매물을 agent-office로 push (notify_new_matches 대칭).
    passed 매물은 전 안전등급 알림(위험도 회피 목적 유용). criteria.min_safety_tier 설정 시 그 이상만."""
    crit = get_listing_criteria()
    if not crit.get("notify_enabled"):
        return {"sent": 0, "skipped": "notify_disabled"}
    listings = get_unnotified_listing_matches(min_tier=crit.get("min_safety_tier"))
    if not listings:
        return {"sent": 0}
    url = f"{AGENT_OFFICE_URL}/api/agent-office/realestate/notify-listing"
    try:
        resp = requests.post(url, json={"listings": listings}, timeout=NOTIFY_TIMEOUT_SECONDS)
        resp.raise_for_status()
        body = resp.json()
    except requests.RequestException as e:
        logger.error("agent-office 매물 push 실패: %s", e)
        return {"sent": 0, "error": str(e)}
    sent_ids = body.get("sent_ids") or []
    if sent_ids:
        mark_listings_notified(sent_ids)
        logger.info("매물 알림 송신: %d건", len(sent_ids))
    return body
--- a/realestate-lab/tests/conftest.py
+++ b/realestate-lab/tests/conftest.py
@@ -16,6 +16,11 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 # 테이블 목록 — init_db가 생성하는 모든 테이블
 _USER_TABLES = (
    "listing_matches",      # FK CASCADE 대비 자식 테이블 먼저
    "listings",
    "market_deals",
    "listing_criteria",
    "listing_collect_log",
    "match_results",       # FK CASCADE 대비 자식 테이블 먼저
    "announcement_models",
    "announcements",
--- a/realestate-lab/tests/test_finance_rules.py
+++ b/realestate-lab/tests/test_finance_rules.py
@@ -0,0 +1,21 @@
 def test_region_regulation_toheo_and_normal():
    from app.finance_rules import region_regulation
    assert region_regulation("대치동")["is_toheo"] is True       # 강남 토허
    r = region_regulation("신대방동")
    assert r["is_toheo"] is False and "확인" in r["notes"]
 def test_jeonse_budget_equity_plus_loan():
    from app.finance_rules import estimate_jeonse_budget
    b = estimate_jeonse_budget(equity=15000, annual_income=6000)   # 자기자금 1.5억
    assert b["max_deposit"] == b["loan_limit"] + 15000
    assert b["loan_limit"] > 0
 def test_purchase_budget_seoul_cap_applies():
    from app.finance_rules import region_regulation, estimate_purchase_budget
    flags = region_regulation("상도동")
    b = estimate_purchase_budget(equity=30000, annual_income=8000,
                                 region_flags=flags, is_first_home=False, is_homeless=True)
    # 수도권 주담대 한도(6·27 대책 6억=60000만원) 캡이 대출가능액에 반영
    assert b["loan_cap"] <= 60000
    assert b["max_price"] == b["loan_cap"] + 30000
    assert 0 < b["ltv_pct"] <= 100
--- a/realestate-lab/tests/test_lawd_codes.py
+++ b/realestate-lab/tests/test_lawd_codes.py
@@ -0,0 +1,6 @@
 def test_lawd_code_maps_target_dongs():
    from app.lawd_codes import lawd_code
    assert lawd_code("신대방동") == "11590"   # 동작구
    assert lawd_code("봉천동") == "11620"      # 관악구
    assert lawd_code("신길동") == "11560"      # 영등포구
    assert lawd_code("없는동") is None
--- a/realestate-lab/tests/test_listing_api.py
+++ b/realestate-lab/tests/test_listing_api.py
@@ -0,0 +1,33 @@
 from fastapi.testclient import TestClient
 def _client():
    from app.main import app
    return TestClient(app)
 def test_criteria_get_default_and_put():
    c = _client()
    assert c.get("/api/realestate/listings/criteria").json()["max_deposit"] == 32000
    c.put("/api/realestate/listings/criteria", json={"max_deposit": 30000})
    assert c.get("/api/realestate/listings/criteria").json()["max_deposit"] == 30000
 def test_budget_endpoint():
    r = _client().post("/api/realestate/budget", json={"equity": 30000, "annual_income": 8000,
                       "is_homeless": True, "is_householder": True, "is_first_home": False, "target_dong": "상도동"})
    b = r.json()
    assert "jeonse" in b and "purchase" in b and "region" in b
    assert b["purchase"]["max_price"] == b["purchase"]["loan_cap"] + 30000
 def test_safety_check_rent():
    from app import db
    for v in (40000, 42000, 44000):
        db.upsert_market_deal({"house_type": "아파트", "dong_code": "11590", "dong": "신대방동",
                               "complex_name": "OO", "area": 42.0, "deal_type": "전세",
                               "deposit": v, "sale_amount": None, "deal_ym": "202606", "floor": "5"})
    r = _client().post("/api/realestate/safety-check",
                       json={"complex_name": "OO", "area": 42.0, "deal_type": "전세",
                             "amount": 28000, "dong_code": "11590"})
    j = r.json()
    assert j["tier"] == "안전" and "disclaimer" in j
 def test_listings_collect_status():
    assert _client().get("/api/realestate/listings/collect/status").status_code == 200
--- a/realestate-lab/tests/test_listing_collector.py
+++ b/realestate-lab/tests/test_listing_collector.py
@@ -0,0 +1,29 @@
 from unittest.mock import patch
 def test_parse_molit_rent_fields():
    from app.listing_collector import _parse_molit_rent
    raw = {"aptNm": "OO아파트", "excluUseAr": "42.5", "deposit": "43,000",
           "monthlyRent": "0", "dealYear": "2026", "dealMonth": "5", "floor": "5"}
    d = _parse_molit_rent(raw, "아파트")
    assert d["complex_name"] == "OO아파트" and d["area"] == 42.5
    assert d["deposit"] == 43000 and d["deal_type"] == "전세" and d["deal_ym"] == "202605"
 def test_parse_naver_article_fields():
    from app.listing_collector import _parse_naver_article
    raw = {"articleNo": "A123", "tradeTypeName": "전세", "dealOrWarrantPrc": "2억 9,000",
           "areaName": "42", "area2": "42.5", "floorInfo": "5/15", "articleName": "OO아파트",
           "buildingName": "OO", "cortarNo": "1159010100"}
    d = _parse_naver_article(raw)
    assert d["article_no"] == "A123" and d["deal_type"] == "전세" and d["deposit"] == 29000
    assert d["source"] == "naver"
 def test_collect_naver_block_sets_naver_ok_false(monkeypatch):
    """네이버 403 → 매물수집 skip + naver_ok=False, 실거래는 계속."""
    from app import listing_collector as lc
    monkeypatch.setattr(lc, "_fetch_molit_all", lambda: 2)      # 실거래 2건 저장됨 가정
    def boom(*a, **k):
        import requests
        raise requests.RequestException("403")
    monkeypatch.setattr(lc, "_fetch_naver_all", boom)
    r = lc.collect_listings()
    assert r["naver_ok"] is False   # 안전마진(실거래)은 유지
--- a/realestate-lab/tests/test_listing_db.py
+++ b/realestate-lab/tests/test_listing_db.py
@@ -0,0 +1,90 @@
 def test_listing_criteria_seed_and_update():
    from app import db
    c = db.get_listing_criteria()
    assert c["id"] == 1 and "신대방동" in c["dongs"] and c["max_deposit"] == 32000
    db.update_listing_criteria({"max_deposit": 30000, "equity": 15000})
    assert db.get_listing_criteria()["max_deposit"] == 30000
    assert db.get_listing_criteria()["equity"] == 15000
 def test_upsert_listing_idempotent_and_is_new():
    from app import db
    row = {"article_no": "A1", "source": "naver", "deal_type": "전세", "deposit": 29000,
           "area_exclusive": 42.0, "complex_name": "OO", "dong": "신대방동", "dong_code": "11590"}
    _, is_new = db.upsert_listing(row)
    assert is_new is True
    _, is_new2 = db.upsert_listing({**row, "deposit": 28000})
    assert is_new2 is False
    ls = db.get_listings()
    assert len(ls) == 1 and ls[0]["deposit"] == 28000
 def test_market_deals_dedup_and_query():
    from app import db
    base = {"house_type": "아파트", "dong_code": "11590", "dong": "신대방동",
            "complex_name": "OO", "area": 42.0, "deal_type": "전세",
            "deposit": 43000, "sale_amount": None, "deal_ym": "202605", "floor": "5", "source": "molit"}
    db.upsert_market_deal(base)
    db.upsert_market_deal(base)  # 동일 → dedup
    deals = db.get_market_deals_for("11590", "OO", 42.0, "전세", months=120)
    assert len(deals) == 1
 def test_unnotified_listing_match_and_mark():
    from app import db
    lid, _ = db.upsert_listing({"article_no": "A2", "source": "naver", "deal_type": "전세",
                                "deposit": 29000, "area_exclusive": 42.0, "dong": "신대방동"})
    db.upsert_listing_match({"listing_id": lid[0] if isinstance(lid, tuple) else lid["id"],
                             "category": "임차", "passed": 1, "match_score": 90,
                             "safety_tier": "안전", "sample_size": 7, "reasons": ["ok"], "is_new": 1})
    un = db.get_unnotified_listing_matches()
    assert len(un) == 1
    db.mark_listings_notified([un[0]["id"]])
    assert db.get_unnotified_listing_matches() == []
 # ── 최종 리뷰 fix (M1/M3/M4) ─────────────────────────────────────────────────
 def test_market_deals_broad_fallback_when_complex_sample_thin():
    """complex_name 지정 시 표본이 MIN_SAMPLE(3) 미만이면 complex 제약을 풀고
    같은 dong_code/deal_type/area±5/recency 조건으로 광역 재조회해야 한다."""
    from app import db
    base = {"house_type": "아파트", "dong_code": "11590", "dong": "신대방동",
            "area": 42.0, "deal_type": "전세", "sale_amount": None, "floor": "5", "source": "molit"}
    # complex "A" 단독 1건 (< MIN_SAMPLE=3)
    db.upsert_market_deal({**base, "complex_name": "A", "deposit": 30000, "deal_ym": "202601"})
    # complex "B" 3건 (dedup 인덱스 회피 위해 deal_ym 다르게)
    db.upsert_market_deal({**base, "complex_name": "B", "deposit": 31000, "deal_ym": "202602"})
    db.upsert_market_deal({**base, "complex_name": "B", "deposit": 32000, "deal_ym": "202603"})
    db.upsert_market_deal({**base, "complex_name": "B", "deposit": 33000, "deal_ym": "202604"})
    deals = db.get_market_deals_for("11590", "A", 42.0, "전세", months=120)
    assert len(deals) == 4  # A 1건 + B 3건 광역 폴백
 def test_upsert_market_deal_area_null_skipped():
    """area=None인 실거래는 median 계산에 무용 + dedup 인덱스도 못 걸려 무한중복
    유발하므로 저장 자체를 skip해야 한다."""
    from app import db
    d = {"house_type": "아파트", "dong_code": "11590", "dong": "신대방동",
         "complex_name": "OO", "area": None, "deal_type": "전세",
         "deposit": 43000, "sale_amount": None, "deal_ym": "202605", "floor": "5", "source": "molit"}
    db.upsert_market_deal(d)
    db.upsert_market_deal(d)  # 재호출해도 여전히 저장 안 됨(무한중복 방지 확인)
    with db._conn() as conn:
        cnt = conn.execute("SELECT COUNT(*) FROM market_deals").fetchone()[0]
    assert cnt == 0
 def test_get_listing_matches_json_parsed():
    """get_listing_matches()도 get_unnotified_listing_matches()와 동일하게
    regulation_flags/reasons를 list로 파싱해 반환해야 한다."""
    from app import db
    lid, _ = db.upsert_listing({"article_no": "A3", "source": "naver", "deal_type": "전세",
                                "deposit": 29000, "area_exclusive": 42.0, "dong": "신대방동"})
    db.upsert_listing_match({"listing_id": lid["id"], "category": "임차", "passed": 1,
                             "match_score": 90, "safety_tier": "안전", "sample_size": 7,
                             "regulation_flags": ["토지거래허가구역"], "reasons": ["ok"], "is_new": 1})
    matches = db.get_listing_matches()
    assert len(matches) == 1
    assert isinstance(matches[0]["regulation_flags"], list)
    assert matches[0]["regulation_flags"] == ["토지거래허가구역"]
    assert isinstance(matches[0]["reasons"], list)
    assert matches[0]["reasons"] == ["ok"]
--- a/realestate-lab/tests/test_listing_matcher.py
+++ b/realestate-lab/tests/test_listing_matcher.py
@@ -0,0 +1,29 @@
 import statistics
 def _deals(vals, key="deposit", dt="전세"):
    return [{key: v, "deal_type": dt} for v in vals]
 def test_compute_safety_tiers():
    from app.listing_matcher import compute_safety
    deals = _deals([40000, 42000, 44000, 43000])  # median ~42500
    assert compute_safety(28000, "전세", deals)["tier"] == "안전"    # 0.66
    assert compute_safety(32000, "전세", deals)["tier"] == "주의"    # 0.75
    assert compute_safety(38000, "전세", deals)["tier"] == "위험"    # 0.89
    assert compute_safety(28000, "전세", _deals([40000, 41000]))["tier"] == "보류"  # 표본<3
 def test_compute_valuation_tiers():
    from app.listing_matcher import compute_valuation
    deals = _deals([80000, 88000, 90000, 87000], key="sale_amount", dt="매매")  # median ~87500
    assert compute_valuation(83000, deals)["tier"] == "저평가"    # 0.95
    assert compute_valuation(89000, deals)["tier"] == "시세"      # 1.02
    assert compute_valuation(95000, deals)["tier"] == "고가"      # 1.09
 def test_match_listing_rent_deposit_cap():
    from app.listing_matcher import match_listing
    crit = {"dongs": ["신대방동"], "deal_types": ["전세"], "max_deposit": 32000,
            "min_area": 40.0, "house_types": ["아파트"]}
    ok = match_listing(crit, {"deal_type": "전세", "deposit": 29000, "area_exclusive": 42.0,
                              "dong": "신대방동"}, budget={"max_deposit": 45000})
    assert ok["passed"] is True
    over = match_listing(crit, {"deal_type": "전세", "deposit": 40000, "area_exclusive": 42.0,
                                "dong": "신대방동"}, budget={"max_deposit": 45000})
    assert over["passed"] is False
--- a/realestate-lab/tests/test_listing_notifier.py
+++ b/realestate-lab/tests/test_listing_notifier.py
@@ -0,0 +1,28 @@
 from unittest.mock import patch, MagicMock
 def _seed_passed_match():
    from app import db
    row, _ = db.upsert_listing({"article_no": "N1", "source": "naver", "deal_type": "전세",
                                "deposit": 29000, "area_exclusive": 42.0, "dong": "신대방동"})
    db.upsert_listing_match({"listing_id": row["id"], "category": "임차", "passed": 1,
                             "match_score": 100, "safety_tier": "안전", "sample_size": 7,
                             "reasons": ["ok"], "is_new": 1})
    return db.get_unnotified_listing_matches()[0]["id"]
 def test_notify_listings_pushes_and_marks():
    from app import notifier, db
    mid = _seed_passed_match()
    fake = MagicMock(); fake.json.return_value = {"sent": 1, "sent_ids": [mid]}; fake.raise_for_status.return_value = None
    with patch.object(notifier.requests, "post", return_value=fake) as post:
        res = notifier.notify_new_listings()
    assert post.call_count == 1
    assert db.get_unnotified_listing_matches() == []   # 마킹됨
 def test_notify_listings_no_mark_on_failure():
    from app import notifier, db
    import requests as rq
    mid = _seed_passed_match()
    with patch.object(notifier.requests, "post", side_effect=rq.RequestException("down")):
        notifier.notify_new_listings()
    assert len(db.get_unnotified_listing_matches()) == 1   # 미마킹→재시도
--- a/stock/app/db.py
+++ b/stock/app/db.py
@@ -465,10 +465,17 @@ def get_alert_state_firing() -> set:
    return {(r["ticker"], r["kind"], r["condition"]) for r in rows}
-def set_alert_firing(ticker: str, kind: str, condition: str, firing: bool, at_iso: str = None) -> None:
+def set_alert_firing(ticker: str, kind: str, condition: str, firing: bool,
                     at_iso: str = None, mark_fired: bool = True) -> None:
    """currently_firing 상태 갱신.
    mark_fired=True(기본): 실제 알림 발송 → first/last_fired_at 갱신.
    mark_fired=False: 쿨다운으로 발송 억제하되 firing 상태만 유지 → 발동시각 미갱신
      (쿨다운이 계속 연장되지 않도록).
    """
    now = at_iso or _now_iso()
    with _conn() as conn:
-        if firing:
+        if firing and mark_fired:
            conn.execute(
                """INSERT INTO trade_alert_state(ticker,kind,condition,currently_firing,first_fired_at,last_fired_at,last_seen_at)
                   VALUES(?,?,?,1,?,?,?)
@@ -479,6 +486,14 @@ def set_alert_firing(ticker: str, kind: str, condition: str, firing: bool, at_is
                     last_seen_at=excluded.last_seen_at""",
                (ticker, kind, condition, now, now, now),
            )
        elif firing and not mark_fired:
            conn.execute(
                """INSERT INTO trade_alert_state(ticker,kind,condition,currently_firing,last_seen_at)
                   VALUES(?,?,?,1,?)
                   ON CONFLICT(ticker,kind,condition) DO UPDATE SET
                     currently_firing=1, last_seen_at=excluded.last_seen_at""",
                (ticker, kind, condition, now),
            )
        else:
            conn.execute(
                "UPDATE trade_alert_state SET currently_firing=0, last_seen_at=? WHERE ticker=? AND kind=? AND condition=?",
@@ -486,6 +501,32 @@ def set_alert_firing(ticker: str, kind: str, condition: str, firing: bool, at_is
            )
 def get_alert_last_fired_map() -> dict:
    """{(ticker,kind,condition): last_fired_at ISO} — 쿨다운 판정용."""
    with _conn() as conn:
        rows = conn.execute(
            "SELECT ticker,kind,condition,last_fired_at FROM trade_alert_state"
        ).fetchall()
    return {(r["ticker"], r["kind"], r["condition"]): r["last_fired_at"] for r in rows}
 def get_ticker_name(ticker: str) -> Optional[str]:
    """종목명 해석 — watchlist → portfolio → krx_master 순. 없으면 None."""
    with _conn() as conn:
        for sql in (
            "SELECT name FROM watchlist WHERE ticker=?",
            "SELECT name FROM portfolio WHERE ticker=? LIMIT 1",
            "SELECT name FROM krx_master WHERE ticker=?",
        ):
            try:
                row = conn.execute(sql, (ticker,)).fetchone()
            except sqlite3.OperationalError:
                continue  # 일부 테스트 DB엔 해당 테이블 부재
            if row and row["name"]:
                return row["name"]
    return None
 def touch_alert_seen(keys: list, at_iso: str) -> None:
    with _conn() as conn:
        for (ticker, kind, condition) in keys:
--- a/stock/app/main.py
+++ b/stock/app/main.py
@@ -23,6 +23,7 @@ from .db import (
    add_sell_history, get_sell_history, update_sell_history, delete_sell_history,
    add_watchlist, remove_watchlist, get_watchlist, get_alert_history,
    get_alert_state_firing, set_alert_firing, touch_alert_seen, add_alert_history,
    get_alert_last_fired_map, get_ticker_name,
 )
 from .scraper import fetch_market_news, fetch_major_indices
 from .price_fetcher import get_current_prices, get_current_prices_detail
@@ -548,15 +549,30 @@ def post_trade_alert_report(req: TradeAlertReport):
    전송 실패 시 상태를 채택하지 않아 다음 사이클에 동일 alert가 다시
    "신규"로 잡혀 재시도된다(멱등). 해제(cleared)는 전송과 무관하게 firing=False.
    """
    from datetime import datetime, timedelta
    cooldown_h = float(os.getenv("TRADE_ALERT_COOLDOWN_HOURS", "6"))
    now = datetime.utcnow()
    prev = get_alert_state_firing()
    last_fired = get_alert_last_fired_map()
    d = diff_firing(req.firing, prev)
    new_count = 0
    suppressed = 0
    for a in d["new"]:
-        if trade_alerts.notify_agent_office([a]):
+        key = (a["ticker"], a["kind"], a["condition"])
-            set_alert_firing(a["ticker"], a["kind"], a["condition"], firing=True, at_iso=req.as_of)
+        # 쿨다운: 같은 종목·조건이 최근 발동됐으면(해제→재발화 오실레이션) 재알림 억제
        lf = last_fired.get(key)
        if cooldown_h > 0 and _within_cooldown(now, lf, timedelta(hours=cooldown_h)):
            set_alert_firing(*key, firing=True, mark_fired=False)   # firing 유지, 발동시각 미갱신
            suppressed += 1
            continue
        name = a.get("name") or get_ticker_name(a["ticker"])
        alert = {**a, "name": name}
        if trade_alerts.notify_agent_office([alert]):
            set_alert_firing(*key, firing=True)                    # 발동시각 갱신(UTC)
            add_alert_history(
-                a["ticker"], a.get("name"), a["kind"], a["condition"],
+                a["ticker"], name, a["kind"], a["condition"],
                a.get("price"), a.get("detail") or {},
            )
            new_count += 1
@@ -566,7 +582,19 @@ def post_trade_alert_report(req: TradeAlertReport):
    touch_alert_seen(d["seen"], req.as_of or "")
-    return {"new_alerts": new_count, "cleared": len(d["cleared"])}
+    return {"new_alerts": new_count, "cleared": len(d["cleared"]), "suppressed": suppressed}
 def _within_cooldown(now, last_iso, cooldown) -> bool:
    """last_iso(UTC ISO `%Y-%m-%dT%H:%M:%fZ`)가 now 기준 cooldown 이내면 True."""
    if not last_iso:
        return False
    from datetime import datetime
    try:
        lf = datetime.strptime(last_iso, "%Y-%m-%dT%H:%M:%fZ")
    except (ValueError, TypeError):
        return False
    return (now - lf) < cooldown
@app.post("/api/portfolio", status_code=201)
@@ -747,8 +775,11 @@ def delete_watchlist_item(ticker: str):
@app.get("/api/stock/trade-alerts")
 def list_trade_alerts(days: int = 7):
-    """매매 알람 이력 조회 (최근 N일)"""
+    """매매 알람 이력 조회 (최근 N일). 조건별 근거(reason) 문자열 포함 — FE가 detail 객체 대신 렌더."""
-    return {"alerts": get_alert_history(days)}
+    alerts = get_alert_history(days)
    for a in alerts:
        a["reason"] = trade_alerts.condition_reason(a.get("condition", ""))
    return {"alerts": alerts}
 # --- Holdings Intelligence API ---
--- a/stock/app/trade_alerts.py
+++ b/stock/app/trade_alerts.py
@@ -32,9 +32,28 @@ def current_session(now_kst) -> str:
    return "closed"
-DEFAULT_EXIT_PARAMS = {"stop_pct": 0.08, "take_pct": 0.25, "trailing_pct": 0.10}
+DEFAULT_EXIT_PARAMS = {"stop_pct": 0.08, "take_pct": 0.25, "trailing_pct": 0.10,
                       "climax_vol_x": 3.0, "climax_close_pct": 0.97}
 DEFAULT_BUY_PARAMS = {"rsi_oversold": 30, "breakout_vol_mult": 1.5, "pullback_pct": 0.02}
 # 조건별 "왜 매수/매도인가" 한 줄 근거 — agent-office `telegram_trade._COND_REASON`과 동일 유지(sync).
 # API(/api/stock/trade-alerts)가 이 문자열을 반환해 FE가 detail(객체) 대신 안전하게 렌더한다.
 _COND_REASON = {
    "buy_ma20_pullback": "상승추세 중 MA20 지지선 눌림목 반등 — 저가 진입 기회",
    "buy_breakout":      "전고점·저항 돌파 + 거래량 증가 — 추세 상승 진입 신호",
    "buy_rsi_bounce":    "RSI 과매도(30↓)에서 반등 — 단기 낙폭과대 되돌림",
    "sell_stop_loss":    "평단 대비 손절선 도달 — 추가 하락 리스크 차단",
    "sell_ma_break":     "주요 이평선(MA50/200) 이탈 — 추세 훼손, 보유 재검토",
    "sell_take_profit":  "목표 수익 도달 — 이익 실현 구간",
    "sell_climax":       "거래량 급증 + 윗꼬리(고점 대비 하락 마감) — 분산·소진 의심",
    "sell_trailing_stop":"보유기간 고점 대비 하락 — 수익 반납 방어(트레일링 스톱)",
 }
 def condition_reason(condition: str) -> str:
    """조건 키 → 사람이 읽는 근거 한 줄. 미지의 조건이면 조건 키 그대로."""
    return _COND_REASON.get(condition, condition)
 def latest_screener_candidates(conn) -> list:
    """최신 성공(status='success') screener run의 후보 {ticker,name} 목록."""
--- a/stock/tests/test_trade_alerts_monitorset_api.py
+++ b/stock/tests/test_trade_alerts_monitorset_api.py
@@ -33,3 +33,11 @@ def test_monitor_set_ok(client):
    assert body["session"] in ("pre", "regular", "after", "closed")
    assert "buy_targets" in body and "sell_targets" in body
    assert body["exit_params"]["trailing_pct"] == 0.10
 def test_monitor_set_exit_params_include_climax(client):
    """climax 파라미터 중앙화 — 워커가 하드코딩 대신 NAS exit_params에서 받아 튜닝."""
    ep = client.get("/api/webai/trade-alert/monitor-set",
                    headers={"X-WebAI-Key": "k"}).json()["exit_params"]
    assert ep["climax_vol_x"] == 3.0
    assert ep["climax_close_pct"] == 0.97
--- a/stock/tests/test_trade_alerts_report_api.py
+++ b/stock/tests/test_trade_alerts_report_api.py
@@ -46,11 +46,51 @@ def test_report_send_failure_does_not_persist(client):
    assert r2.json()["new_alerts"] == 1
-def test_report_cleared_rearm(client):
+def test_report_cooldown_suppresses_immediate_refire(client):
    """같은 종목·조건이 해제됐다 곧바로 재발화해도 쿨다운(기본 6h) 내면 재알림 억제."""
    firing = [{"ticker": "005930", "name": "삼성", "kind": "buy",
               "condition": "buy_breakout", "price": 71500, "detail": {}}]
    with patch("app.trade_alerts.notify_agent_office", return_value=True):
        assert _report(client, firing).json()["new_alerts"] == 1   # 최초 알림
        _report(client, [])                                        # 해제
        r = _report(client, firing)                                # 즉시 재발화 → 쿨다운 억제
    assert r.json()["new_alerts"] == 0
    assert r.json()["suppressed"] == 1
 def test_report_refire_after_cooldown_alerts(client, monkeypatch):
    """쿨다운=0이면 해제 후 재발화 시 재알림."""
    monkeypatch.setenv("TRADE_ALERT_COOLDOWN_HOURS", "0")
    firing = [{"ticker": "005930", "name": "삼성", "kind": "buy",
               "condition": "buy_breakout", "price": 71500, "detail": {}}]
    with patch("app.trade_alerts.notify_agent_office", return_value=True):
        _report(client, firing)
-        _report(client, [])                        # 해제
+        _report(client, [])
-        r = _report(client, firing)                # 재발화
+        r = _report(client, firing)
    assert r.json()["new_alerts"] == 1
 def test_trade_alerts_history_includes_reason_string(client):
    """이력 API가 조건별 사람이 읽는 근거(reason) 문자열을 함께 반환한다(FE가 detail 객체 대신 렌더)."""
    firing = [{"ticker": "005930", "name": "삼성", "kind": "sell",
               "condition": "sell_ma_break", "price": 60000,
               "detail": {"ma50": 1, "ma200": 2, "severity": "high"}}]
    with patch("app.trade_alerts.notify_agent_office", return_value=True):
        _report(client, firing)
    alerts = client.get("/api/stock/trade-alerts?days=1").json()["alerts"]
    assert isinstance(alerts[0]["reason"], str) and len(alerts[0]["reason"]) > 3
 def test_report_resolves_stock_name_from_watchlist(client):
    """워커 firing에 name이 없어도 NAS가 종목명을 해석해 알림에 포함한다."""
    from app import db
    db.add_watchlist("000660", "SK하이닉스")
    firing = [{"ticker": "000660", "kind": "buy", "condition": "buy_breakout",
               "price": 180000, "detail": {}}]   # name 없음
    with patch("app.trade_alerts.notify_agent_office", return_value=True) as m:
        _report(client, firing)
    sent_alert = m.call_args[0][0][0]
    assert sent_alert["name"] == "SK하이닉스"
    # 이력에도 종목명 기록
    alerts = client.get("/api/stock/trade-alerts?days=1").json()["alerts"]
    assert alerts[0]["name"] == "SK하이닉스"
Author	SHA1	Message	Date
gahusb	a7be8f76bf	docs(CLAUDE.md): realestate 매물 알림+안전마진 엔드포인트 + agent-office notify-listing 등재 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9	2026-07-09 14:08:14 +09:00
gahusb	95fadaa8ef	fix(realestate): 시세 표본 광역폴백(M1)+area-NULL upsert skip(M3)+matches JSON파싱(M4) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9	2026-07-09 14:03:54 +09:00
gahusb	c6b969443f	feat(agent-office): 매물 알림 텔레그램 notify-listing(너+아내) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9	2026-07-09 13:50:04 +09:00
gahusb	b4fb3998fe	feat(realestate): 매물 라우트 6종+safety-check/budget+3h cron Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9	2026-07-09 13:43:34 +09:00
gahusb	4847626424	feat(realestate): notify_new_listings — 매물 알림 agent-office push	2026-07-09 13:39:19 +09:00
gahusb	08ac800910	feat(realestate): 매물 수집(국토부 실거래[합법]+네이버 호가[폴백]) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9	2026-07-09 13:36:20 +09:00
gahusb	d752675e9d	feat(realestate): 매물 매칭+안전마진(전세가율)/매매 적정성(호가율) 판정 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9	2026-07-09 13:33:11 +09:00
gahusb	c0a50f4ee6	feat(realestate): 매물 DB 4테이블(listings/market_deals/matches/criteria)+헬퍼 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9	2026-07-09 13:29:36 +09:00
gahusb	cd15504f86	feat(realestate): 재무·규제 규칙(전세/매수 예산·토허·LTV·DSR config)	2026-07-09 13:24:00 +09:00
gahusb	e8998a4098	feat(realestate): 매물 파이프라인 법정동코드(LAWD_CD) 매핑	2026-07-09 13:20:59 +09:00
gahusb	b8229c0ffa	feat(stock): /api/stock/trade-alerts 응답에 조건별 근거(reason) 문자열 추가 FE가 detail(객체)을 그대로 렌더해 React #31 크래시 → 안전한 텍스트 필드 제공. condition_reason(조건→한줄근거), agent-office telegram_trade._COND_REASON과 동일 유지.	2026-07-09 12:47:34 +09:00
gahusb	9baea3a0e2	feat(stock): 매매알람 쿨다운 중복억제 + 종목명 해석 - 쿨다운(TRADE_ALERT_COOLDOWN_HOURS 기본 6h): 같은 종목·조건 해제→재발화 오실레이션 시 재알림 억제(set_alert_firing mark_fired=False로 firing 유지·발동시각 미갱신, suppressed 카운트). - 종목명: 워커 firing에 name 없어도 NAS가 watchlist→portfolio→krx_master로 해석해 알림·이력에 포함.	2026-07-03 16:14:51 +09:00
gahusb	80daa53558	feat(agent-office): 매매알람에 조건별 '왜 매수/매도' 한 줄 근거(💡) 추가	2026-07-03 16:14:51 +09:00
gahusb	35795abb0f	docs(README): 실시간 매매 알람 + WSL워커 /infra 관측 팀규칙 + Alpine tzdata 함정 반영 stock 실시간 매매알람(watchlist/trade_alert_state/history·webai 계약·1분 Windows 워커), agent-office 매매알람 notify+/watch 봇·분산워커 관측, 주의사항에 팀규칙·tzdata, DB 테이블 목록 최신화. (기존 하네스 엔지니어링 섹션도 함께 커밋)	2026-07-03 11:01:24 +09:00
gahusb	4e47f5dd43	docs(CLAUDE.md): [팀 규칙] 모든 WSL docker 워커는 /infra 관측 필수 (node_monitor WORKER_REGISTRY 등재+heartbeat 3단계)	2026-07-03 10:48:17 +09:00
gahusb	246c8d5328	feat(agent-office): node_monitor에 trade-monitor 워커 등재 + trader 링크 from을 워커명으로 수정 WSL 워커 관측 규칙 — 모든 WSL docker 워커는 /infra에서 모니터링 가능해야 함. trade-monitor(kind=trader) 등재 → /nodes·/infra 노출. 링크 from 하드코딩('ai_trade')을 w[name]으로 고쳐 다중 trader가 각자 링크를 갖도록 함. 미배포 워커는 prev=None이라 다운 경보 없음.	2026-07-03 10:45:45 +09:00
gahusb	ed17193945	feat(stock): 매매알람 exit_params에 climax 파라미터 중앙화 (climax_vol_x 3.0, climax_close_pct 0.97)	2026-07-03 10:37:57 +09:00