feat(stock): technical_posture (스크리너 노드 보유종목 적용)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

stock/app/holdings_intel.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 import datetime as dt
 from typing import Any, Optional
 
+import pandas as pd
+
 from . import db
 from . import price_fetcher
 
@@ -30,3 +32,109 @@ def get_holdings() -> list[dict]:
             "is_krx": it["ticker"] in krx,
         })
     return out
+
+
+# ---- Task 2.1: technical_posture ----
+
+from .screener.engine import combine
+
+
+def _score_nodes_and_weights():
+    """NODE_REGISTRY에서 보유종목 매수강도 계산용 노드 인스턴스화."""
+    from .screener.registry import NODE_REGISTRY
+    weights = {"ma_alignment": 0.4, "momentum": 0.3, "rs_rating": 0.3}
+    nodes = [NODE_REGISTRY[k]() for k in weights]
+    return nodes, weights
+
+
+def technical_posture(ctx, tickers: list[str]) -> dict[str, float]:
+    """보유종목 restrict 후 score 노드 → 매수강도(0~100)."""
+    scoped = ctx.restrict(tickers)
+    if scoped.prices.empty:
+        return {}
+    nodes, weights = _score_nodes_and_weights()
+    scores = {}
+    for n in nodes:
+        try:
+            scores[n.name] = n.compute(scoped, {})
+        except Exception:
+            scores[n.name] = pd.Series(0.0, index=scoped.master.index)
+    total = combine(scores, weights)
+    return {t: float(total.get(t, 0.0)) for t in tickers if t in total.index}
+
+
+# ---- Task 2.2: exit_rules ----
+
+def _ma(closes: "pd.Series", window: int) -> Optional[float]:
+    if len(closes) < window:
+        return None
+    return float(closes.rolling(window).mean().iloc[-1])
+
+
+def exit_rules(holding: dict, ticker_prices: "pd.DataFrame", params: dict) -> dict:
+    """가격 기반 청산/리스크 flag. (momentum_loss는 compute 단계에서 합산.)"""
+    flags = {"stop_loss": False, "ma50_break": False, "ma200_break": False,
+             "take_profit": False, "climax": False}
+    avg = holding.get("avg_price")
+    cur = holding.get("current_price")
+    if ticker_prices is None or ticker_prices.empty:
+        closes = pd.Series(dtype=float)
+    else:
+        closes = ticker_prices.sort_values("date")["close"].astype(float).reset_index(drop=True)
+    last_close = float(closes.iloc[-1]) if len(closes) else cur
+    if cur is None:
+        cur = last_close
+    if cur and avg:
+        if cur < avg * (1 - params["stop_pct"]):
+            flags["stop_loss"] = True
+        if (cur - avg) / avg >= params["take_pct"]:
+            flags["take_profit"] = True
+    ma50 = _ma(closes, 50)
+    ma200 = _ma(closes, 200)
+    if ma50 is not None and last_close is not None and last_close < ma50:
+        flags["ma50_break"] = True
+    if ma200 is not None and last_close is not None and last_close < ma200:
+        flags["ma200_break"] = True
+    # climax: 최근 거래량이 20일 평균의 climax_vol_x배 이상 + 종가가 당일 고점 대비 하단(상단꼬리)
+    if ticker_prices is not None and not ticker_prices.empty and len(ticker_prices) >= 21:
+        tp = ticker_prices.sort_values("date")
+        vol = tp["volume"].astype(float).reset_index(drop=True)
+        avg_vol = vol.iloc[-21:-1].mean()
+        last_vol = vol.iloc[-1]
+        hi_ = float(tp["high"].astype(float).iloc[-1])
+        cl_ = float(tp["close"].astype(float).iloc[-1])
+        if avg_vol and last_vol >= avg_vol * params["climax_vol_x"] and hi_ > 0 and cl_ < hi_ * 0.97:
+            flags["climax"] = True
+    return flags
+
+
+# ---- Task 2.3: decide_action ----
+
+ADD_SCORE = 70.0   # 이 이상이면 추가매수 후보
+
+
+def decide_action(tech_score: float, exit_flags: dict, pnl: float | None) -> tuple[str, str]:
+    """우선순위: sell > trim > add > hold. 근거 텍스트 동봉."""
+    reasons = []
+    # 청산 (최우선)
+    if exit_flags.get("stop_loss"):
+        reasons.append("손절선 이탈")
+    if exit_flags.get("ma200_break"):
+        reasons.append("MA200 이탈")
+    if reasons:
+        return "sell", " · ".join(reasons)
+    # 축소
+    if exit_flags.get("ma50_break"):
+        reasons.append("MA50 이탈")
+    if exit_flags.get("momentum_loss"):
+        reasons.append("모멘텀 소멸")
+    if exit_flags.get("take_profit"):
+        reasons.append(f"목표 수익 도달(+{pnl:.0f}%)" if pnl is not None else "목표 수익 도달")
+    if exit_flags.get("climax"):
+        reasons.append("거래량 급증 분산 의심")
+    if reasons:
+        return "trim", " · ".join(reasons)
+    # 추가매수
+    if tech_score is not None and tech_score >= ADD_SCORE:
+        return "add", f"기술적 강도 양호({tech_score:.0f})"
+    return "hold", "특이 신호 없음"

stock/app/test_holdings_intel.py

@@ -1,5 +1,9 @@
+import datetime as dt
+import pandas as pd
+
 from app import holdings_intel as hi
 
+
 def test_get_holdings_merges_price_and_pnl(monkeypatch):
     monkeypatch.setattr(hi.db, "get_all_portfolio", lambda: [
         {"id": 1, "broker": "kis", "ticker": "005930", "name": "삼성전자",
@@ -54,3 +58,86 @@ def test_get_holdings_price_missing(monkeypatch):
     hs = hi.get_holdings()
     assert hs[0]["current_price"] is None
     assert hs[0]["pnl_rate"] is None
+
+
+# ---- Phase 2 tests ----
+
+def _toy_ctx(tickers=("005930",), n=300):
+    """결정적 일봉으로 ScreenContext 유사 객체 구성."""
+    from app.screener.engine import ScreenContext
+    rows = []
+    base = dt.date(2025, 1, 1)
+    for t in tickers:
+        price = 1000
+        for i in range(n):
+            price = int(price * 1.002)  # 완만한 상승 → 정배열
+            d = (base + dt.timedelta(days=i)).isoformat()
+            rows.append({"ticker": t, "date": d, "open": price, "high": price,
+                         "low": price, "close": price, "volume": 1000, "value": price*1000})
+    prices = pd.DataFrame(rows)
+    master = pd.DataFrame({"name": [f"n{t}" for t in tickers],
+                           "market": ["KOSPI"]*len(tickers),
+                           "market_cap": [1e12]*len(tickers)},
+                          index=pd.Index(tickers, name="ticker"))
+    flow = pd.DataFrame(columns=["ticker","date","foreign_net","institution_net"])
+    return ScreenContext(master=master, prices=prices, flow=flow,
+                         kospi=pd.Series(dtype=float), asof=base+dt.timedelta(days=n-1))
+
+
+def test_technical_posture_returns_scores():
+    ctx = _toy_ctx(("005930",))
+    scores = hi.technical_posture(ctx, ["005930"])
+    assert "005930" in scores
+    assert 0.0 <= scores["005930"] <= 100.0     # 상승추세 → 양수 점수
+
+
+# ---- Task 2.2 tests ----
+
+def _ticker_prices(closes, vols=None):
+    n = len(closes)
+    base = dt.date(2025, 1, 1)
+    vols = vols or [1000]*n
+    return pd.DataFrame({
+        "ticker": ["005930"]*n,
+        "date": [(base+dt.timedelta(days=i)).isoformat() for i in range(n)],
+        "open": closes, "high": closes, "low": closes, "close": closes, "volume": vols,
+    })
+
+
+DEFAULT_EXIT = {"stop_pct": 0.08, "take_pct": 0.25, "climax_vol_x": 3.0}
+
+
+def test_exit_rules_stop_and_ma():
+    closes = [1000]*60 + [1100]*200  # 충분한 길이, 최근 평탄
+    df = _ticker_prices(closes)
+    # 현재가가 평단(2000) 대비 -45% → stop_loss
+    flags = hi.exit_rules({"avg_price": 2000, "current_price": 1100}, df, DEFAULT_EXIT)
+    assert flags["stop_loss"] is True
+    # 종가 1100 > MA50≈1100, MA200은 더 낮음 → ma 이탈 아님
+    assert flags["ma200_break"] is False
+
+
+def test_exit_rules_take_profit():
+    df = _ticker_prices([1000]*260)
+    flags = hi.exit_rules({"avg_price": 1000, "current_price": 1300}, df, DEFAULT_EXIT)
+    assert flags["take_profit"] is True   # +30% ≥ 25%
+
+
+# ---- Task 2.3 tests ----
+
+def test_decide_action_matrix():
+    # 강건 + 이탈 없음 + 높은 강도 → add
+    a, r = hi.decide_action(tech_score=80, exit_flags={}, pnl=5)
+    assert a == "add"
+    # ma200 이탈 → sell
+    a, r = hi.decide_action(70, {"ma200_break": True}, 2)
+    assert a == "sell"
+    # stop_loss → sell
+    a, _ = hi.decide_action(70, {"stop_loss": True}, -10)
+    assert a == "sell"
+    # ma50 이탈만 → trim
+    a, _ = hi.decide_action(60, {"ma50_break": True}, 3)
+    assert a == "trim"
+    # 이탈 없음 보통 강도 → hold
+    a, _ = hi.decide_action(50, {}, 1)
+    assert a == "hold"