diff --git a/stock/app/holdings_intel.py b/stock/app/holdings_intel.py index faa24e1..759794b 100644 --- a/stock/app/holdings_intel.py +++ b/stock/app/holdings_intel.py @@ -7,6 +7,7 @@ import pandas as pd from . import db from . import price_fetcher +from .screener.engine import combine def _krx_tickers() -> set: @@ -36,8 +37,6 @@ def get_holdings() -> list[dict]: # ---- Task 2.1: technical_posture ---- -from .screener.engine import combine - def _score_nodes_and_weights(): """NODE_REGISTRY에서 보유종목 매수강도 계산용 노드 인스턴스화.""" @@ -59,16 +58,24 @@ def technical_posture(ctx, tickers: list[str]) -> dict[str, float]: scores[n.name] = n.compute(scoped, {}) except Exception: scores[n.name] = pd.Series(0.0, index=scoped.master.index) - total = combine(scores, weights) + scores_ne = {k: s for k, s in scores.items() if not s.empty} + weights_ne = {k: w for k, w in weights.items() if k in scores_ne} + if not weights_ne: + return {} + total = combine(scores_ne, weights_ne) return {t: float(total.get(t, 0.0)) for t in tickers if t in total.index} # ---- Task 2.2: exit_rules ---- +_DEFAULT_EXIT_PARAMS = {"stop_pct": 0.08, "take_pct": 0.25, "climax_vol_x": 3.0} + + def _ma(closes: "pd.Series", window: int) -> Optional[float]: if len(closes) < window: return None - return float(closes.rolling(window).mean().iloc[-1]) + val = closes.rolling(window).mean().iloc[-1] + return float(val) if pd.notna(val) else None def exit_rules(holding: dict, ticker_prices: "pd.DataFrame", params: dict) -> dict: @@ -76,6 +83,7 @@ def exit_rules(holding: dict, ticker_prices: "pd.DataFrame", params: dict) -> di Note: momentum_loss는 compute_and_store 단계에서 집계하므로 여기서 설정하지 않는다. """ + p = {**_DEFAULT_EXIT_PARAMS, **(params or {})} flags = {"stop_loss": False, "ma50_break": False, "ma200_break": False, "take_profit": False, "climax": False} avg = holding.get("avg_price") @@ -87,10 +95,10 @@ def exit_rules(holding: dict, ticker_prices: "pd.DataFrame", params: dict) -> di 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"]): + if cur is not None and avg: + if cur < avg * (1 - p["stop_pct"]): flags["stop_loss"] = True - if (cur - avg) / avg >= params["take_pct"]: + if avg > 0 and (cur - avg) / avg >= p["take_pct"]: flags["take_profit"] = True ma50 = _ma(closes, 50) ma200 = _ma(closes, 200) @@ -106,7 +114,7 @@ def exit_rules(holding: dict, ticker_prices: "pd.DataFrame", params: dict) -> di 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: + if avg_vol and last_vol >= avg_vol * p["climax_vol_x"] and hi_ > 0 and cl_ < hi_ * 0.97: flags["climax"] = True return flags @@ -116,7 +124,8 @@ def exit_rules(holding: dict, ticker_prices: "pd.DataFrame", params: dict) -> di ADD_SCORE = 70.0 # 이 이상이면 추가매수 후보 -def decide_action(tech_score: float, exit_flags: dict, pnl: float | None) -> tuple[str, str]: +def decide_action(tech_score: float, exit_flags: dict, pnl: float | None, + add_score: float = ADD_SCORE) -> tuple[str, str]: """액션 결정 매트릭스: sell > trim > add > hold (우선순위 순). Returns: @@ -142,6 +151,6 @@ def decide_action(tech_score: float, exit_flags: dict, pnl: float | None) -> tup if reasons: return "trim", " · ".join(reasons) # 추가매수 - if tech_score is not None and tech_score >= ADD_SCORE: + 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 dcd4545..4c18ec4 100644 --- a/stock/app/test_holdings_intel.py +++ b/stock/app/test_holdings_intel.py @@ -141,3 +141,50 @@ def test_decide_action_matrix(): # 이탈 없음 보통 강도 → hold a, _ = hi.decide_action(50, {}, 1) assert a == "hold" + + +# ---- Phase 2 hardening tests (m3) ---- + +def _ticker_prices_hl(closes, highs, vols): + n = len(closes) + base = dt.date(2025, 1, 1) + return pd.DataFrame({ + "ticker": ["005930"] * n, + "date": [(base + dt.timedelta(days=i)).isoformat() for i in range(n)], + "open": closes, + "high": highs, + "low": closes, + "close": closes, + "volume": vols, + }) + + +def test_exit_rules_climax(): + closes = [1000] * 30 + highs = [1000] * 29 + [1100] # 마지막날 상단꼬리(종가1000 < 고가1100*0.97) + vols = [1000] * 29 + [5000] # 거래량 5x + flags = hi.exit_rules({"avg_price": 900, "current_price": 1000}, + _ticker_prices_hl(closes, highs, vols), {}) + assert flags["climax"] is True + + +def test_exit_rules_ma200_break(): + closes = list(range(1000, 1000 + 260))[::-1] # 하락 추세 → 종가 < MA200 + df = _ticker_prices(closes) + flags = hi.exit_rules({"avg_price": 2000, "current_price": closes[-1]}, df, {}) + assert flags["ma200_break"] is True + + +def test_technical_posture_short_history_returns_low_not_crash(): + ctx = _toy_ctx(("005930",), n=100) # <252 → MA 노드 NaN→0, but no crash + scores = hi.technical_posture(ctx, ["005930"]) + assert "005930" in scores + assert 0.0 <= scores["005930"] <= 100.0 + + +def test_technical_posture_empty_kospi_not_penalized(): + # rs_rating는 빈 kospi에서 빈 Series → combine에서 제외되어야 (C1) + ctx = _toy_ctx(("005930",), n=300) # kospi 빈 fixture + scores = hi.technical_posture(ctx, ["005930"]) + # ma_alignment+momentum만으로 정규화 → 상승추세면 충분히 높은 점수 + assert scores["005930"] > 50.0