feat(stock): 매매알람 쿨다운 중복억제 + 종목명 해석

- 쿨다운(TRADE_ALERT_COOLDOWN_HOURS 기본 6h): 같은 종목·조건 해제→재발화 오실레이션 시
  재알림 억제(set_alert_firing mark_fired=False로 firing 유지·발동시각 미갱신, suppressed 카운트).
- 종목명: 워커 firing에 name 없어도 NAS가 watchlist→portfolio→krx_master로 해석해 알림·이력에 포함.

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:

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

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:

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]):
-            set_alert_firing(a["ticker"], a["kind"], a["condition"], firing=True, at_iso=req.as_of)
+        key = (a["ticker"], a["kind"], a["condition"])
+        # 쿨다운: 같은 종목·조건이 최근 발동됐으면(해제→재발화 오실레이션) 재알림 억제
+        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)

stock/tests/test_trade_alerts_report_api.py

@@ -46,11 +46,40 @@ 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, [])                        # 해제
-        r = _report(client, firing)                # 재발화
+        _report(client, [])
+        r = _report(client, firing)
     assert r.json()["new_alerts"] == 1
+
+
+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하이닉스"