From 789785fe3a4f79c379d5880956b453297448c4ca Mon Sep 17 00:00:00 2001 From: gahusb Date: Sun, 31 May 2026 22:11:45 +0900 Subject: [PATCH] feat(stock): compute_and_store + build_holdings_brief Co-Authored-By: Claude Opus 4.8 (1M context) --- stock/app/holdings_intel.py | 74 ++++++++++++++++++++++++++++++++ stock/app/test_holdings_intel.py | 22 ++++++++++ 2 files changed, 96 insertions(+) diff --git a/stock/app/holdings_intel.py b/stock/app/holdings_intel.py index 49a3b0c..838d3e8 100644 --- a/stock/app/holdings_intel.py +++ b/stock/app/holdings_intel.py @@ -238,6 +238,80 @@ def portfolio_health(holdings: list[dict], total_cash: int = 0) -> dict: } +DEFAULT_PARAMS = { + "stop_pct": 0.08, "take_pct": 0.25, "climax_vol_x": 3.0, + "move_pct": 7.0, "vol_z": 2.5, + "momentum_drop": 15.0, "momentum_low": 35.0, +} + + +def _load_ctx(asof: dt.date): + """ScreenContext.load를 감싸는 thin wrapper (테스트에서 monkeypatch 대상).""" + from .screener.engine import ScreenContext + with db._conn() as conn: + return ScreenContext.load(conn, asof) + + +def _today_kst() -> dt.date: + return (dt.datetime.utcnow() + dt.timedelta(hours=9)).date() + + +def compute_and_store(asof: Optional[dt.date] = None, use_llm: bool = True, + params: dict | None = None) -> dict: + """보유종목 시그널 계산 → holdings_signals upsert (멱등). + + Returns: + {"stored": N, "date": "YYYY-MM-DD"} or {"stored": 0, "reason": "..."} + """ + asof = asof or _today_kst() + p = {**DEFAULT_PARAMS, **(params or {})} + holdings = get_holdings() + if not holdings: + return {"stored": 0, "reason": "no_holdings"} + krx = [h for h in holdings if h.get("is_krx")] + ctx = _load_ctx(asof) + posture = technical_posture(ctx, [h["ticker"] for h in krx]) if krx else {} + date_iso = asof.isoformat() + issues_map = news_issues([h["ticker"] for h in holdings], date_iso, use_llm=use_llm) + stored = 0 + for h in holdings: + t = h["ticker"] + tp = ctx.prices[ctx.prices["ticker"] == t] if h.get("is_krx") else None + tf = ctx.flow[ctx.flow["ticker"] == t] if h.get("is_krx") else None + flags = exit_rules(h, tp, p) if h.get("is_krx") else {} + tech = posture.get(t) + # momentum_loss: 직전 저장 시그널 대비 하락 or 낮은 강도 + prev = db.get_holdings_signal_history(t, limit=2) + prev_score = next((r["tech_score"] for r in prev if r["date"] != date_iso), None) + if tech is not None and ( + (prev_score is not None and tech < prev_score - p["momentum_drop"]) + or tech < p["momentum_low"] + ): + flags["momentum_loss"] = True + evts = market_events(t, tp, tf, p) if h.get("is_krx") else [] + issues = list(issues_map.get(t, [])) + evts + action, reasons = decide_action(tech if tech is not None else 0.0, flags, h.get("pnl_rate")) + db.upsert_holdings_signal( + date=date_iso, ticker=t, name=h.get("name"), action=action, + tech_score=tech, exit_flags=flags, issues=issues, + close=h.get("current_price"), pnl_rate=h.get("pnl_rate"), reasons=reasons, + ) + stored += 1 + return {"stored": stored, "date": date_iso} + + +def build_holdings_brief(date: Optional[str] = None) -> dict: + """최신 시그널 + 포트 건강 조립 (브리핑/UI payload).""" + date = date or db.get_latest_holdings_date() + if not date: + return {"date": None, "holdings": [], "portfolio_health": {}} + signals = db.get_holdings_signals(date) + holdings = get_holdings() + total_cash = sum(c.get("cash", 0) for c in db.get_all_broker_cash()) + health = portfolio_health(holdings, total_cash=total_cash) + return {"date": date, "holdings": signals, "portfolio_health": health} + + def decide_action(tech_score: float, exit_flags: dict, pnl: float | None, add_score: float = ADD_SCORE) -> tuple[str, str]: """액션 결정 매트릭스: sell > trim > add > hold (우선순위 순). diff --git a/stock/app/test_holdings_intel.py b/stock/app/test_holdings_intel.py index 30dcee7..84825ef 100644 --- a/stock/app/test_holdings_intel.py +++ b/stock/app/test_holdings_intel.py @@ -273,3 +273,25 @@ def test_portfolio_health_empty_and_zero(): h1 = hi.portfolio_health([{"ticker": "X", "quantity": 1, "avg_price": 0, "current_price": 0, "is_krx": True}], total_cash=0) assert h1["total_pnl_rate"] == 0.0 + + +# ---- Phase 4 tests ---- + +def test_compute_and_store_and_brief(monkeypatch): + import os, tempfile + from app import db + monkeypatch.setattr(db, "DB_PATH", os.path.join(tempfile.mkdtemp(), "stock.db")) + db.init_db() + monkeypatch.setattr(hi, "get_holdings", lambda: [ + {"ticker": "005930", "name": "삼성전자", "quantity": 10, "avg_price": 1000, + "current_price": 1100, "pnl_rate": 10.0, "is_krx": True}]) + ctx = _toy_ctx(("005930",)) + monkeypatch.setattr(hi, "_load_ctx", lambda asof: ctx) + monkeypatch.setattr(hi, "_news_sentiment_map", lambda date: {}) + monkeypatch.setattr(hi.db, "get_all_broker_cash", lambda: [{"broker": "kis", "cash": 500000}]) + res = hi.compute_and_store(asof=ctx.asof, use_llm=False) + assert res["stored"] == 1 + brief = hi.build_holdings_brief() + assert brief["holdings"][0]["ticker"] == "005930" + assert "portfolio_health" in brief + assert brief["holdings"][0]["action"] in ("add", "hold", "trim", "sell")