From 116b2540c2bbb782817a8f2be694a94c76825fa1 Mon Sep 17 00:00:00 2001 From: gahusb Date: Sun, 31 May 2026 21:48:01 +0900 Subject: [PATCH] =?UTF-8?q?feat(stock):=20technical=5Fposture=20(=EC=8A=A4?= =?UTF-8?q?=ED=81=AC=EB=A6=AC=EB=84=88=20=EB=85=B8=EB=93=9C=20=EB=B3=B4?= =?UTF-8?q?=EC=9C=A0=EC=A2=85=EB=AA=A9=20=EC=A0=81=EC=9A=A9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) --- stock/app/holdings_intel.py | 108 +++++++++++++++++++++++++++++++ stock/app/test_holdings_intel.py | 87 +++++++++++++++++++++++++ 2 files changed, 195 insertions(+) diff --git a/stock/app/holdings_intel.py b/stock/app/holdings_intel.py index 0e83634..5cc43ee 100644 --- a/stock/app/holdings_intel.py +++ b/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", "특이 신호 없음" diff --git a/stock/app/test_holdings_intel.py b/stock/app/test_holdings_intel.py index 01d4dc7..dcd4545 100644 --- a/stock/app/test_holdings_intel.py +++ b/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"