feat(stock): edge diff(신규/해제/재무장) 순수 함수

stock/app/trade_alerts.py

@@ -76,3 +76,24 @@ def build_monitor_set(conn, session: str, exit_params: dict, buy_params: dict) -
         "buy_params": buy_params,
         "exit_params": exit_params,
     }
+
+
+def diff_firing(reported: list, prev: set) -> dict:
+    """워커 발화집합(reported) vs 직전 발화상태(prev) edge diff.
+
+    reported 각 항목: {ticker,kind,condition,price,detail,name?}.
+    key = (ticker,kind,condition).
+    반환 {"new":[신규 alert...], "cleared":[해제 key...], "seen":[현재 key...]}.
+    """
+    cur = {}
+    for a in reported:
+        key = (a["ticker"], a["kind"], a["condition"])
+        cur[key] = a
+    cur_keys = set(cur.keys())
+    new_keys = cur_keys - prev
+    cleared = sorted(prev - cur_keys)
+    return {
+        "new": [cur[k] for k in cur_keys if k in new_keys],
+        "cleared": cleared,
+        "seen": sorted(cur_keys),
+    }

stock/tests/test_trade_alerts_edge.py

@@ -0,0 +1,18 @@
+def test_diff_new_and_cleared_and_rearm():
+    from app.trade_alerts import diff_firing
+    reported = [{"ticker": "005930", "kind": "buy", "condition": "buy_breakout",
+                 "price": 71500, "detail": {}}]
+    # 최초: prev 비어있음 → 신규
+    d1 = diff_firing(reported, prev=set())
+    assert [a["condition"] for a in d1["new"]] == ["buy_breakout"]
+    assert d1["cleared"] == []
+    # 유지: prev에 이미 있음 → 신규 없음
+    prev = {("005930", "buy", "buy_breakout")}
+    d2 = diff_firing(reported, prev=prev)
+    assert d2["new"] == []
+    # 해제: reported 비었고 prev에 있음 → cleared
+    d3 = diff_firing([], prev=prev)
+    assert d3["cleared"] == [("005930", "buy", "buy_breakout")]
+    # 재무장 후 재발화: prev 다시 비면 신규
+    d4 = diff_firing(reported, prev=set())
+    assert len(d4["new"]) == 1