# 주식 보유종목 인텔리전스 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 시장용 스크리너 엔진을 내 보유종목에 restrict 적용하고, 신규 매도/리스크 룰·이슈 감지·포트 건강을 얹어 매일 advisory 브리핑(텔레그램+UI)한다. **Architecture:** stock에 순수연산 `holdings_intel.py` + 집계테이블 `holdings_signals` 추가. 기존 `screener/engine.py`의 `ScreenContext.restrict()`로 보유종목 기술분석, 신규 exit_rules/decide_action으로 매도자세 결정, market_events+news_issues로 이슈, portfolio_health로 포트요약. agent-office가 EOD 계산(16:40)·아침 브리핑(08:30)·장중 가드(30분)를 orchestrate. KIS 실주문 미사용(advisory). **Tech Stack:** Python 3.12, FastAPI, SQLite, pandas, APScheduler, Claude Haiku(ai_summarizer), pytest / React+Vite(web-ui 별도 repo). **Spec:** `docs/superpowers/specs/2026-05-31-stock-holdings-intelligence-design.md` --- ## 기존 자산 (재사용 — 시그니처 확인됨) - `stock/app/db.py`: `_conn()`, `init_db()`(CREATE TABLE IF NOT EXISTS + `_ensure`류 마이그레이션), `get_all_portfolio() -> [{id,broker,ticker,name,quantity,avg_price,purchase_price}]`, `get_all_broker_cash() -> [{broker,cash}]`, `get_latest_articles(limit,category)`. 테스트는 monkeypatch로 DB_PATH류 격리(기존 test 파일 참조). - `stock/app/price_fetcher.py`: `get_current_prices(tickers) -> {ticker:int}`, `get_current_prices_detail(tickers) -> {ticker:{...}}`. - `stock/app/screener/engine.py`: `ScreenContext.load(conn, asof, lookback_days=504) -> ctx`(master/prices/flow/news_sentiment), `ctx.restrict(tickers)`, `ctx.latest_close()`. `combine(scores, weights)`. - `stock/app/screener/nodes/base.py`: `ScoreNode.compute(ctx, params) -> pd.Series(0..100, index=ticker)`. nodes: ma_alignment/momentum/rs_rating/vcp_lite/volume_surge/foreign_buy/high52w. `stock/app/screener/registry.py`에 GATE/SCORE 레지스트리. - `stock/app/ai_summarizer.py`: `async summarize_news(articles) -> {summary, tokens, model, duration_ms}` (Claude/Ollama). - `news_sentiment` 테이블(date,ticker,score_raw,news_count) — 종목별 감성. `krx_daily_prices`(ticker,date,o/h/l/c,volume,value), `krx_flow`(ticker,date,foreign_net,institution_net). - agent-office: `service_proxy`(STOCK_URL httpx), `StockAgent`(on_schedule/on_screener_schedule/on_ai_news_schedule/on_command), `scheduler.py`(`_run_stock_*` wrappers + init_scheduler), `telegram.messaging.send_raw`. ## 알려진 제약 (plan 전반 반영) - `articles`는 종목 태깅 없음 → 종목별 이슈는 `news_sentiment` 기반 + 회사명 substring 매칭으로 article best-effort. - MA200/momentum 노드는 ~252일 일봉 필요 → 누적 부족 종목은 NaN→0(노드가 이미 처리). 신규 보유·운영 초기엔 tech_score 낮을 수 있음(graceful). - KRX 외 종목(미국주): krx_daily_prices 밖 → `is_krx=False`로 기술분석 skip, 뉴스·손익만. --- # Phase 1 — holdings_signals 테이블 + get_holdings ## Task 1.1: holdings_signals 테이블 + CRUD **Files:** - Modify: `stock/app/db.py` (init_db + CRUD) - Test: `stock/app/test_holdings_db.py` - [ ] **Step 1: 실패 테스트** `stock/app/test_holdings_db.py`: ```python import os, tempfile, importlib def _fresh_db(monkeypatch): tmp = tempfile.mkdtemp() from app import db monkeypatch.setattr(db, "DB_PATH", os.path.join(tmp, "stock.db")) db.init_db() return db def test_holdings_signals_table_and_upsert(monkeypatch): db = _fresh_db(monkeypatch) db.upsert_holdings_signal(date="2026-05-29", ticker="005930", name="삼성전자", action="hold", tech_score=72.0, exit_flags={"stop_loss": False}, issues=[{"type": "news", "severity": "low", "summary": "x"}], close=80000, pnl_rate=5.2, reasons="강건") db.upsert_holdings_signal(date="2026-05-29", ticker="005930", name="삼성전자", action="trim", tech_score=60.0, exit_flags={"ma50_break": True}, issues=[], close=79000, pnl_rate=3.0, reasons="MA50 이탈") rows = db.get_holdings_signals(date="2026-05-29") assert len(rows) == 1 # upsert 멱등 assert rows[0]["action"] == "trim" assert rows[0]["exit_flags"]["ma50_break"] is True # JSON 역직렬화 hist = db.get_holdings_signal_history("005930", days=30) assert len(hist) == 1 ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_db.py -v` Expected: FAIL (`upsert_holdings_signal` 없음) - [ ] **Step 3: 테이블 DDL** — `stock/app/db.py` `init_db()` 안 sell_history 테이블 블록 뒤에: ```python conn.execute( """ CREATE TABLE IF NOT EXISTS holdings_signals ( date TEXT NOT NULL, ticker TEXT NOT NULL, name TEXT, action TEXT NOT NULL, tech_score REAL, exit_flags TEXT NOT NULL DEFAULT '{}', issues TEXT NOT NULL DEFAULT '[]', close INTEGER, pnl_rate REAL, reasons TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), PRIMARY KEY (date, ticker) ); """ ) conn.execute("CREATE INDEX IF NOT EXISTS idx_holdings_sig_ticker " "ON holdings_signals(ticker, date DESC);") ``` - [ ] **Step 4: CRUD 함수** — `stock/app/db.py` 끝에 (`import json`은 파일 상단에 이미 있으면 재사용, 없으면 추가): ```python def upsert_holdings_signal(date, ticker, name, action, tech_score, exit_flags, issues, close, pnl_rate, reasons) -> None: with _conn() as conn: conn.execute( """ INSERT INTO holdings_signals (date, ticker, name, action, tech_score, exit_flags, issues, close, pnl_rate, reasons) VALUES (?,?,?,?,?,?,?,?,?,?) ON CONFLICT(date, ticker) DO UPDATE SET name=excluded.name, action=excluded.action, tech_score=excluded.tech_score, exit_flags=excluded.exit_flags, issues=excluded.issues, close=excluded.close, pnl_rate=excluded.pnl_rate, reasons=excluded.reasons """, (date, ticker, name, action, tech_score, json.dumps(exit_flags, ensure_ascii=False), json.dumps(issues, ensure_ascii=False), close, pnl_rate, reasons), ) def _row_to_signal(r) -> dict: d = dict(r) d["exit_flags"] = json.loads(d.get("exit_flags") or "{}") d["issues"] = json.loads(d.get("issues") or "[]") return d def get_holdings_signals(date: str) -> list: with _conn() as conn: rows = conn.execute( "SELECT * FROM holdings_signals WHERE date=? ORDER BY ticker", (date,)).fetchall() return [_row_to_signal(r) for r in rows] def get_latest_holdings_date() -> str | None: with _conn() as conn: r = conn.execute("SELECT MAX(date) AS d FROM holdings_signals").fetchone() return r["d"] if r and r["d"] else None def get_holdings_signal_history(ticker: str, days: int = 30) -> list: with _conn() as conn: rows = conn.execute( "SELECT * FROM holdings_signals WHERE ticker=? ORDER BY date DESC LIMIT ?", (ticker, days)).fetchall() return [_row_to_signal(r) for r in rows] ``` > `_conn()` row_factory가 `sqlite3.Row`인지 확인(기존 db.py 패턴). 아니면 dict 변환 보장. - [ ] **Step 5: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_db.py -v` Expected: PASS - [ ] **Step 6: Commit** ```bash git add stock/app/db.py stock/app/test_holdings_db.py git commit -m "feat(stock): holdings_signals 테이블 + CRUD" ``` ## Task 1.2: get_holdings — 보유종목 + 현재가 + 손익 + KRX 판별 **Files:** - Create: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: 실패 테스트** `stock/app/test_holdings_intel.py`: ```python 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": "삼성전자", "quantity": 10, "avg_price": 70000, "purchase_price": 70000}, {"id": 2, "broker": "kis", "ticker": "AAPL", "name": "Apple", "quantity": 5, "avg_price": 200, "purchase_price": 200}, ]) monkeypatch.setattr(hi.price_fetcher, "get_current_prices", lambda tickers: {"005930": 77000}) # AAPL 미조회(비KRX) monkeypatch.setattr(hi, "_krx_tickers", lambda: {"005930"}) hs = hi.get_holdings() s = {h["ticker"]: h for h in hs} assert s["005930"]["is_krx"] is True assert round(s["005930"]["pnl_rate"], 1) == 10.0 # (77000-70000)/70000 assert s["AAPL"]["is_krx"] is False # KRX 외 ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_get_holdings_merges_price_and_pnl -v` Expected: FAIL - [ ] **Step 3: 구현** — `stock/app/holdings_intel.py`: ```python """보유종목 인텔리전스 — 순수연산 중심 (advisory). KIS 실주문 미사용.""" from __future__ import annotations import datetime as dt from typing import Any, Optional from . import db from . import price_fetcher def _krx_tickers() -> set: """krx_master에 존재하는 ticker 집합 (KRX 판별용).""" with db._conn() as conn: try: rows = conn.execute("SELECT ticker FROM krx_master").fetchall() except Exception: return set() return {r["ticker"] for r in rows} def get_holdings() -> list[dict]: """portfolio + 현재가 + pnl_rate + is_krx.""" items = db.get_all_portfolio() tickers = [it["ticker"] for it in items] prices = price_fetcher.get_current_prices(tickers) if tickers else {} krx = _krx_tickers() out = [] for it in items: cur = prices.get(it["ticker"]) avg = it["avg_price"] pnl = ((cur - avg) / avg * 100.0) if (cur and avg) else None out.append({ **it, "current_price": cur, "pnl_rate": pnl, "is_krx": it["ticker"] in krx, }) return out ``` - [ ] **Step 4: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py -v` Expected: PASS - [ ] **Step 5: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): get_holdings (현재가·손익·KRX판별)" ``` --- # Phase 2 — 기술분석 + 매도룰 + 액션 결정 (핵심 신규 로직) ## Task 2.1: technical_posture — 스크리너 노드를 보유종목에 적용 **Files:** - Modify: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: registry 확인** — Run: `cd stock && python -c "from app.screener import registry; print(dir(registry))"` 로 SCORE 노드 레지스트리/기본 weights 접근법 확인. (예: `registry.SCORE_REGISTRY` dict[name->NodeClass], `registry.DEFAULT_WEIGHTS`. 실제 이름은 registry.py를 읽어 확인.) - [ ] **Step 2: 실패 테스트** — `test_holdings_intel.py`에 추가: ```python import datetime as dt import pandas as pd 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 # 상승추세 → 양수 점수 ``` - [ ] **Step 3: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_technical_posture_returns_scores -v` Expected: FAIL - [ ] **Step 4: 구현** — `holdings_intel.py`에 추가 (registry의 실제 SCORE 노드/weights 이름은 Step 1에서 확인한 것을 사용): ```python from .screener.engine import combine # 보유종목 매수강도에 쓸 score 노드 (registry에서 인스턴스화). # registry.py 실제 구조에 맞춰 import — 아래는 직접 인스턴스화 예시. def _score_nodes_and_weights(): # NODE_REGISTRY(검증됨): {"momentum": Momentum20, "rs_rating": RsRating, "ma_alignment": MaAlignment, ...} 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} ``` > Step 1에서 확인한 노드 클래스명/모듈경로/`Momentum`·`RsRating` 실제 이름에 맞춰 import 수정. `compute`가 빈 params 허용하는지 확인(MaAlignment는 default_params 사용 → `{}` OK). - [ ] **Step 5: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py -v` Expected: PASS - [ ] **Step 6: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): technical_posture (스크리너 노드 보유종목 적용)" ``` ## Task 2.2: exit_rules — 손절·MA이탈·익절·클라이맥스 (가격 기반 flag) **Files:** - Modify: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: 실패 테스트** — `test_holdings_intel.py`에 추가: ```python 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% ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py -k exit_rules -v` Expected: FAIL - [ ] **Step 3: 구현** — `holdings_intel.py`에 추가: ```python 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 ``` - [ ] **Step 4: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py -k exit_rules -v` Expected: PASS - [ ] **Step 5: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): exit_rules (손절·MA이탈·익절·클라이맥스)" ``` ## Task 2.3: decide_action — 매수강도+flag → 액션 매트릭스 **Files:** - Modify: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: 실패 테스트** — 추가: ```python 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" ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_decide_action_matrix -v` Expected: FAIL - [ ] **Step 3: 구현** — `holdings_intel.py`에 추가: ```python 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", "특이 신호 없음" ``` - [ ] **Step 4: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_decide_action_matrix -v` Expected: PASS - [ ] **Step 5: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): decide_action 매트릭스 (sell>trim>add>hold)" ``` --- # Phase 3 — 이슈 감지 + 포트 건강 ## Task 3.1: market_events — 급변·거래량·외인 (기존 데이터) **Files:** - Modify: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: 실패 테스트** — 추가: ```python DEFAULT_EVENT = {"move_pct": 7.0, "vol_z": 2.5} def test_market_events_detects_move_and_volume(): closes = [1000]*30 + [1100] # 마지막날 +10% vols = [1000]*30 + [10000] # 거래량 급증 df = _ticker_prices(closes, vols) evts = hi.market_events("005930", df, None, DEFAULT_EVENT) types = {e["type"] for e in evts} assert "price_move" in types assert "volume_surge" in types ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_market_events_detects_move_and_volume -v` Expected: FAIL - [ ] **Step 3: 구현** — `holdings_intel.py`에 추가: ```python def market_events(ticker: str, ticker_prices: "pd.DataFrame", ticker_flow: "pd.DataFrame | None", params: dict) -> list[dict]: """일봉/flow 기반 시장 이벤트 (급변·거래량 Z·외인 순매도).""" events = [] if ticker_prices is None or ticker_prices.empty or len(ticker_prices) < 2: return events tp = ticker_prices.sort_values("date").reset_index(drop=True) close = tp["close"].astype(float) pct = (close.iloc[-1] - close.iloc[-2]) / close.iloc[-2] * 100.0 if close.iloc[-2] else 0.0 if abs(pct) >= params["move_pct"]: events.append({"type": "price_move", "severity": "high" if abs(pct) >= params["move_pct"]*1.5 else "med", "summary": f"전일 대비 {pct:+.1f}%"}) vol = tp["volume"].astype(float) if len(vol) >= 21: base = vol.iloc[-21:-1] mu, sd = base.mean(), base.std(ddof=0) if sd and (vol.iloc[-1] - mu) / sd >= params["vol_z"]: events.append({"type": "volume_surge", "severity": "med", "summary": f"거래량 평소 대비 급증(Z={ (vol.iloc[-1]-mu)/sd:.1f })"}) if ticker_flow is not None and not ticker_flow.empty: tf = ticker_flow.sort_values("date") recent = tf["foreign_net"].astype(float).iloc[-3:] if len(recent) >= 3 and (recent < 0).all(): events.append({"type": "foreign_selling", "severity": "med", "summary": "외국인 3일 연속 순매도"}) return events ``` - [ ] **Step 4: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_market_events_detects_move_and_volume -v` Expected: PASS - [ ] **Step 5: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): market_events (급변·거래량Z·외인순매도)" ``` ## Task 3.2: news_issues — news_sentiment 기반 악재 flag (+LLM best-effort) **Files:** - Modify: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: 실패 테스트** — 추가: ```python def test_news_issues_flags_negative_sentiment(monkeypatch): # news_sentiment: 005930 음수 점수 → 악재 flag monkeypatch.setattr(hi, "_news_sentiment_map", lambda date: { "005930": {"score_raw": -0.6, "news_count": 8}}) issues = hi.news_issues(["005930"], date="2026-05-29", use_llm=False) assert "005930" in issues assert issues["005930"][0]["type"] == "news" assert issues["005930"][0]["severity"] in ("med", "high") ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_news_issues_flags_negative_sentiment -v` Expected: FAIL - [ ] **Step 3: 구현** — `holdings_intel.py`에 추가: ```python NEG_SENTIMENT = -0.3 # 이하면 악재 후보 def _news_sentiment_map(date: str) -> dict: with db._conn() as conn: try: rows = conn.execute( "SELECT ticker, score_raw, news_count FROM news_sentiment WHERE date=?", (date,)).fetchall() except Exception: return {} return {r["ticker"]: {"score_raw": r["score_raw"], "news_count": r["news_count"]} for r in rows} def news_issues(tickers: list[str], date: str, use_llm: bool = True) -> dict[str, list]: """news_sentiment 음수 → 악재 flag. (LLM 요약은 best-effort, 기본 비활성 테스트.)""" senti = _news_sentiment_map(date) out: dict[str, list] = {} for t in tickers: s = senti.get(t) if not s or s["score_raw"] is None: continue if s["score_raw"] <= NEG_SENTIMENT: sev = "high" if s["score_raw"] <= NEG_SENTIMENT * 2 else "med" out.setdefault(t, []).append({ "type": "news", "severity": sev, "summary": f"부정 뉴스 감성({s['score_raw']:+.2f}, {s.get('news_count',0)}건)", }) return out ``` > LLM 요약(`use_llm=True`)은 후속 — articles가 종목 태깅이 없어 회사명 substring 매칭이 필요. v1은 sentiment 기반 flag로 충분(spec §3). LLM 통합은 Phase 4 compute에서 옵션으로 호출하되 실패 graceful. - [ ] **Step 4: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_news_issues_flags_negative_sentiment -v` Expected: PASS - [ ] **Step 5: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): news_issues (감성 기반 악재 flag)" ``` ## Task 3.3: portfolio_health — 집중도·시장mix·현금·손익 **Files:** - Modify: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: 실패 테스트** — 추가: ```python def test_portfolio_health(): holdings = [ {"ticker": "005930", "quantity": 10, "avg_price": 70000, "current_price": 77000, "is_krx": True}, {"ticker": "000660", "quantity": 5, "avg_price": 100000, "current_price": 90000, "is_krx": True}, ] h = hi.portfolio_health(holdings, total_cash=1000000) assert h["positions"] == 2 assert 0 <= h["max_weight"] <= 1.0 assert "total_eval" in h and "total_pnl" in h and "cash_ratio" in h ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_portfolio_health -v` Expected: FAIL - [ ] **Step 3: 구현** — `holdings_intel.py`에 추가: ```python def portfolio_health(holdings: list[dict], total_cash: int = 0) -> dict: """비중 집중도(최대비중·HHI) + 시장 mix + 현금비중 + 총손익.""" evals, buys = [], [] for h in holdings: cur = h.get("current_price") or h.get("avg_price") or 0 ev = cur * h.get("quantity", 0) bu = (h.get("avg_price") or 0) * h.get("quantity", 0) evals.append(ev); buys.append(bu) total_eval = sum(evals) total_buy = sum(buys) weights = [e / total_eval for e in evals] if total_eval else [] hhi = sum(w*w for w in weights) total_assets = total_eval + (total_cash or 0) return { "positions": len(holdings), "total_eval": total_eval, "total_buy": total_buy, "total_pnl": total_eval - total_buy, "total_pnl_rate": ((total_eval - total_buy) / total_buy * 100.0) if total_buy else 0.0, "max_weight": max(weights) if weights else 0.0, "hhi": round(hhi, 4), "cash_ratio": ((total_cash or 0) / total_assets) if total_assets else 0.0, } ``` - [ ] **Step 4: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_portfolio_health -v` Expected: PASS - [ ] **Step 5: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): portfolio_health (집중도·현금·손익)" ``` --- # Phase 4 — compute_and_store + 브리핑 조립 + API ## Task 4.1: compute_and_store + build_holdings_brief **Files:** - Modify: `stock/app/holdings_intel.py` - Test: `stock/app/test_holdings_intel.py` - [ ] **Step 1: 실패 테스트** — 추가 (DB + ctx 통합; monkeypatch로 ScreenContext.load·get_holdings·news를 결정적으로): ```python 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") ``` - [ ] **Step 2: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py::test_compute_and_store_and_brief -v` Expected: FAIL - [ ] **Step 3: 구현** — `holdings_intel.py`에 추가: ```python 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): 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 (멱등).""" 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 {} issues_map = news_issues([h["ticker"] for h in holdings], asof.isoformat(), use_llm=use_llm) date_iso = asof.isoformat() 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, days=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() hmap = {h["ticker"]: h for h in 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} ``` - [ ] **Step 4: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_intel.py -v` Expected: PASS (전체) - [ ] **Step 5: Commit** ```bash git add stock/app/holdings_intel.py stock/app/test_holdings_intel.py git commit -m "feat(stock): compute_and_store + build_holdings_brief" ``` ## Task 4.2: API 라우터 + main 등록 **Files:** - Modify: `stock/app/main.py` - Test: `stock/app/test_holdings_api.py` - [ ] **Step 1: main.py 라우팅 패턴 확인** — Run: `cd stock && grep -nE "@app.(get|post)|include_router|FastAPI\(" app/main.py | head` 로 stock이 `@app.get` 직접 정의인지 라우터인지 확인. (stock/main.py는 직접 `@app.get` 패턴으로 보임 — 그에 맞춰 엔드포인트 추가.) - [ ] **Step 2: 실패 테스트** `stock/app/test_holdings_api.py`: ```python import os, tempfile, sys from fastapi.testclient import TestClient def _client(monkeypatch): sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from app import db monkeypatch.setattr(db, "DB_PATH", os.path.join(tempfile.mkdtemp(), "stock.db")) db.init_db() from app.main import app return TestClient(app) def test_holdings_intel_endpoint(monkeypatch): client = _client(monkeypatch) r = client.get("/api/stock/holdings/intel") assert r.status_code == 200 body = r.json() assert "holdings" in body and "portfolio_health" in body ``` - [ ] **Step 3: 실패 확인** — Run: `cd stock && python -m pytest app/test_holdings_api.py -v` Expected: FAIL (404) - [ ] **Step 4: 엔드포인트 추가** — `stock/app/main.py`에 (`from . import holdings_intel` import 추가, 기존 `@app.get` 패턴 사용): ```python from . import holdings_intel @app.get("/api/stock/holdings/intel") def holdings_intel_brief(): return holdings_intel.build_holdings_brief() @app.get("/api/stock/holdings/intel/history") def holdings_intel_history(ticker: str, days: int = 30): from . import db return {"ticker": ticker, "history": db.get_holdings_signal_history(ticker, days)} @app.post("/api/stock/holdings/intel/run") def holdings_intel_run(background_tasks: BackgroundTasks, use_llm: bool = True): background_tasks.add_task(holdings_intel.compute_and_store, None, use_llm) return {"ok": True, "queued": True} ``` > **확인됨**: main.py의 기존 import는 `from fastapi import FastAPI, Query, Header, Depends, HTTPException`로 **`BackgroundTasks`가 없음** → 그 줄에 `, BackgroundTasks`를 추가할 것. `holdings_intel.compute_and_store(None, use_llm)`은 첫 인자 asof=None(오늘) 의미. - [ ] **Step 5: 통과 확인** — Run: `cd stock && python -m pytest app/test_holdings_api.py -v` Expected: PASS - [ ] **Step 6: Commit** ```bash git add stock/app/main.py stock/app/test_holdings_api.py git commit -m "feat(stock): holdings intel API (intel/history/run)" ``` --- # Phase 5 — agent-office (EOD 계산·아침 브리핑·장중 가드) ## Task 5.1: service_proxy 호출 **Files:** - Modify: `agent-office/app/service_proxy.py` - [ ] **Step 1: 함수 추가** — stock 섹션에: ```python async def stock_holdings_run() -> Dict[str, Any]: async with httpx.AsyncClient(timeout=120) as client: resp = await client.post(f"{STOCK_URL}/api/stock/holdings/intel/run", params={"use_llm": True}) resp.raise_for_status() return resp.json() async def stock_holdings_brief() -> Dict[str, Any]: resp = await _client.get(f"{STOCK_URL}/api/stock/holdings/intel") resp.raise_for_status() return resp.json() ``` > 파일 상단 httpx import / `_client` 패턴 확인 후 일치시킬 것. - [ ] **Step 2: import 확인** — Run: `cd agent-office && python -c "from app import service_proxy"` Expected: 에러 없음 - [ ] **Step 3: Commit** ```bash git add agent-office/app/service_proxy.py git commit -m "feat(agent-office): stock holdings run/brief 프록시" ``` ## Task 5.2: 브리핑 텔레그램 포매터 **Files:** - Create: `agent-office/app/notifiers/telegram_stock.py` (없으면 생성; 있으면 함수 추가) - Test: `agent-office/tests/test_holdings_brief_format.py` - [ ] **Step 1: 실패 테스트** `agent-office/tests/test_holdings_brief_format.py`: ```python from app.notifiers import telegram_stock as ts def test_format_holdings_brief(): payload = { "date": "2026-05-29", "holdings": [ {"ticker": "005930", "name": "삼성전자", "action": "trim", "tech_score": 60.0, "exit_flags": {"ma50_break": True}, "issues": [{"type":"news","severity":"high","summary":"악재"}], "pnl_rate": 5.2, "reasons": "MA50 이탈"}, {"ticker": "000660", "name": "SK하이닉스", "action": "hold", "tech_score": 75.0, "exit_flags": {}, "issues": [], "pnl_rate": -2.0, "reasons": "특이 신호 없음"}, ], "portfolio_health": {"positions": 2, "total_pnl_rate": 3.1, "max_weight": 0.6, "cash_ratio": 0.2}, } txt = ts.format_holdings_brief(payload) assert "삼성전자" in txt assert "축소" in txt or "trim" in txt assert "%" in txt ``` - [ ] **Step 2: 실패 확인** — Run: `cd agent-office && python -m pytest tests/test_holdings_brief_format.py -v` Expected: FAIL - [ ] **Step 3: 구현** — `agent-office/app/notifiers/telegram_stock.py`: ```python """보유종목 인텔리전스 텔레그램 포매터 (advisory).""" import logging from typing import Any, Dict logger = logging.getLogger("agent-office") _ACTION_KR = {"add": "🟢 추가매수", "hold": "⚪ 보유", "trim": "🟡 축소", "sell": "🔴 매도"} _SEV = {"high": "🔴", "med": "🟠", "low": "🟡"} def format_holdings_brief(payload: Dict[str, Any]) -> str: date = payload.get("date") or "?" lines = [f"📊 보유종목 인텔리전스 ({date})", ""] ph = payload.get("portfolio_health") or {} if ph: lines.append(f"포트 손익 {ph.get('total_pnl_rate',0):+.1f}% · " f"종목 {ph.get('positions',0)} · 최대비중 {ph.get('max_weight',0)*100:.0f}% · " f"현금 {ph.get('cash_ratio',0)*100:.0f}%") lines.append("") for h in payload.get("holdings", []): act = _ACTION_KR.get(h.get("action"), h.get("action", "?")) pnl = h.get("pnl_rate") pnl_txt = f"{pnl:+.1f}%" if pnl is not None else "—" line = f"{act} {h.get('name') or h.get('ticker')} ({pnl_txt})" if h.get("reasons"): line += f" — {h['reasons']}" lines.append(line) for iss in (h.get("issues") or [])[:3]: lines.append(f" {_SEV.get(iss.get('severity'),'•')} {iss.get('summary','')}") lines.append("") lines.append("ℹ️ 투자 판단 보조용 제안입니다(자동매매 아님).") return "\n".join(lines) async def send_holdings_brief(payload: Dict[str, Any]) -> None: from ..telegram.messaging import send_raw text = format_holdings_brief(payload) try: await send_raw(text) except Exception as e: logger.warning(f"[telegram_stock] holdings brief send failed: {e}") ``` - [ ] **Step 4: 통과 확인** — Run: `cd agent-office && python -m pytest tests/test_holdings_brief_format.py -v` Expected: PASS - [ ] **Step 5: Commit** ```bash git add agent-office/app/notifiers/telegram_stock.py agent-office/tests/test_holdings_brief_format.py git commit -m "feat(agent-office): 보유종목 브리핑 텔레그램 포매터" ``` ## Task 5.3: StockAgent 메서드 + 장중 가드 + scheduler cron **Files:** - Modify: `agent-office/app/agents/stock.py` - Modify: `agent-office/app/scheduler.py` - [ ] **Step 1: StockAgent 메서드 추가** — `agents/stock.py` `StockAgent`에: ```python async def run_holdings_eod(self) -> dict: """평일 16:40 — 보유종목 시그널 계산·저장.""" from ..service_proxy import stock_holdings_run from ..db import create_task, update_task_status, add_log task_id = create_task(self.agent_id, "holdings_eod", {}) try: res = await stock_holdings_run() update_task_status(task_id, "succeeded", res) add_log(self.agent_id, f"holdings_eod: {res}", "info", task_id) return {"ok": True, **res} except Exception as e: update_task_status(task_id, "failed", {"error": str(e)}) add_log(self.agent_id, f"holdings_eod 실패: {e}", "error", task_id) return {"ok": False, "message": str(e)} async def run_holdings_brief(self) -> dict: """평일 08:30 — 저장된 시그널 브리핑 텔레그램.""" from ..service_proxy import stock_holdings_brief from ..notifiers.telegram_stock import send_holdings_brief from ..db import create_task, update_task_status, add_log task_id = create_task(self.agent_id, "holdings_brief", {}) try: payload = await stock_holdings_brief() await send_holdings_brief(payload) update_task_status(task_id, "succeeded", {"date": payload.get("date"), "count": len(payload.get("holdings", []))}) add_log(self.agent_id, f"holdings_brief 발송: {payload.get('date')}", "info", task_id) return {"ok": True} except Exception as e: update_task_status(task_id, "failed", {"error": str(e)}) add_log(self.agent_id, f"holdings_brief 실패: {e}", "error", task_id) return {"ok": False, "message": str(e)} ``` 그리고 `on_command`에 분기 추가: ```python if command == "holdings_eod": return await self.run_holdings_eod() if command == "holdings_brief": return await self.run_holdings_brief() ``` - [ ] **Step 2: scheduler cron** — `agent-office/app/scheduler.py`에 wrapper + 등록: ```python async def _run_stock_holdings_eod(): agent = AGENT_REGISTRY.get("stock") if agent: await agent.run_holdings_eod() async def _run_stock_holdings_brief(): agent = AGENT_REGISTRY.get("stock") if agent: await agent.run_holdings_brief() ``` `init_scheduler()` stock cron 그룹에: ```python scheduler.add_job(_run_stock_holdings_eod, "cron", day_of_week="mon-fri", hour=16, minute=40, id="stock_holdings_eod") scheduler.add_job(_run_stock_holdings_brief, "cron", day_of_week="mon-fri", hour=8, minute=30, id="stock_holdings_brief") ``` > 장중 경량 가드(30분 간격 손절·급변 alert)는 **후속 슬라이스**로 분리 — 본 plan은 EOD 계산 + 아침 브리핑까지. (가드는 holdings_signals의 exit_flags + 현재가 비교가 필요해 별도 설계가 깔끔; spec §4.3은 다음 사이클로 명시.) - [ ] **Step 3: import 확인** — Run: `cd agent-office && python -c "from app import scheduler; from app.agents.stock import StockAgent; from app.notifiers import telegram_stock"` Expected: 에러 없음 - [ ] **Step 4: Commit** ```bash git add agent-office/app/agents/stock.py agent-office/app/scheduler.py git commit -m "feat(agent-office): StockAgent holdings EOD(16:40)+브리핑(08:30) cron" ``` > **Note (장중 가드 분리):** spec §4.3의 장중 경량 가드는 별도 후속 슬라이스로 미룬다(throttle/cap 설계가 로또 시그널 수준의 별도 작업). 본 plan은 EOD+아침브리핑으로 완결된 advisory 루프를 제공. --- # Phase 6 — web-ui 보유종목 인텔리전스 탭 (별도 repo: web-ui) > **주의:** web-ui는 별도 Git 저장소. 커밋은 `web-ui/`에서([[feedback-commit-repo]]). 배포 `npm run release:nas` 수동. 먼저 feature 브랜치 생성. ## Task 6.1: api.js 헬퍼 **Files:** Modify `web-ui/src/api.js` - [ ] **Step 1: 헬퍼 추가** — 기존 `apiGet` 패턴 사용(확인됨): ```javascript export const stockHoldingsIntel = () => apiGet('/api/stock/holdings/intel'); export const stockHoldingsHistory = (ticker, days = 30) => apiGet(`/api/stock/holdings/intel/history?ticker=${ticker}&days=${days}`); ``` - [ ] **Step 2: Commit** (web-ui repo, feature 브랜치) ```bash cd ../web-ui && git checkout -b feat/stock-holdings-ui && git add src/api.js && git commit -m "feat: 보유종목 인텔리전스 API 헬퍼" ``` ## Task 6.2: HoldingsIntel 컴포넌트 + 포트폴리오 페이지 통합 **Files:** - Create: `web-ui/src/pages/stock/HoldingsIntel.jsx` (경로는 Step 1 확인 결과에 맞춤) - Modify: stock/포트폴리오 페이지 컨테이너 - [ ] **Step 1: 페이지 구조 확인** — Run: `cd ../web-ui && ls src/pages/stock/ 2>/dev/null; grep -rln "portfolio\|Portfolio\|포트폴리오" src/pages/ | head` 로 포트폴리오 페이지 + 탭 패턴 + 기존 카드 컴포넌트 스타일 확인. - [ ] **Step 2: HoldingsIntel 컴포넌트** — 기존 카드/탭 컨벤션에 맞춰 작성 (액션별 색상, 이슈 severity 뱃지, 포트 건강 요약). 예시 골격: ```jsx import { useEffect, useState } from 'react'; import { stockHoldingsIntel } from '../../api'; const ACTION = { add: ['추가매수', '#22c55e'], hold: ['보유', '#94a3b8'], trim: ['축소', '#f59e0b'], sell: ['매도', '#ef4444'] }; export default function HoldingsIntel() { const [data, setData] = useState(null); useEffect(() => { stockHoldingsIntel().then(setData).catch(() => {}); }, []); if (!data) return null; const ph = data.portfolio_health || {}; return (
손익 {(ph.total_pnl_rate ?? 0).toFixed(1)}% · 종목 {ph.positions ?? 0} · 최대비중 {((ph.max_weight ?? 0) * 100).toFixed(0)}% · 현금 {((ph.cash_ratio ?? 0) * 100).toFixed(0)}%
{(data.holdings || []).map((h) => { const [label, color] = ACTION[h.action] || [h.action, '#94a3b8']; return (
{label} {h.name || h.ticker} {h.pnl_rate != null ? `${h.pnl_rate.toFixed(1)}%` : '—'}
{h.reasons}
{(h.issues || []).slice(0, 3).map((iss, i) => (
{iss.summary}
))}
); })}
); } ``` > 실제 디자인 토큰·CSS 클래스·탭 통합 지점은 Step 1에서 확인한 기존 포트폴리오 페이지 컨벤션에 맞춰 조정. 신규 라우트보다 **기존 페이지 탭 통합** 선호([[feedback-new-page-or-tab]]). - [ ] **Step 3: 페이지 통합** — Step 1에서 찾은 포트폴리오 페이지에 탭/섹션으로 `` 추가. - [ ] **Step 4: 빌드 확인** — Run: `cd ../web-ui && npm run build` Expected: exit 0 - [ ] **Step 5: Commit** (web-ui) ```bash git add src/ && git commit -m "feat: 보유종목 인텔리전스 탭 (액션·이슈·포트건강)" ``` --- # Phase 7 — 통합 검증 ## Task 7.1: 전체 회귀 - [ ] **Step 1: stock 테스트** — Run: `cd stock && python -m pytest app/ -q` Expected: 신규 통과 + 기존 회귀 없음(사전 실패가 있으면 별도 식별) - [ ] **Step 2: agent-office 테스트** — Run: `cd agent-office && python -m pytest -q` Expected: 신규 통과 + 기존 회귀 없음 - [ ] **Step 3: 배포 후 수동 트리거 안내** — NAS 배포 후 `POST /api/stock/holdings/intel/run`으로 첫 시그널 생성, `GET /api/stock/holdings/intel`로 확인. 평일 EOD/아침 cron이 이후 자동 운영. --- ## Self-Review 체크리스트 결과 - **Spec 커버리지**: 데이터모델(1.1) / get_holdings(1.2) / technical_posture(2.1) / 매도룰(2.2) / decide_action(2.3) / market_events(3.1) / news_issues(3.2) / portfolio_health(3.3) / compute+brief(4.1) / API(4.2) / agent-office(5.x) / UI(6.x). 장중 가드(spec §4.3)는 Phase 5 Note에서 후속 슬라이스로 명시 분리(스코프 관리). - **Placeholder**: 모든 코드 step에 실제 코드. registry 노드명·main.py 라우팅 패턴·web-ui 컨벤션은 "Step에서 확인 후 맞춤" 명시(코드베이스 의존, 합리적). - **타입 일관성**: exit_flags dict 키(stop_loss/ma50_break/ma200_break/momentum_loss/take_profit/climax)가 exit_rules·decide_action·compute·포매터에서 일치. holdings_signals 컬럼 ↔ upsert ↔ build_brief ↔ telegram_stock 키 일치. action 값(add/hold/trim/sell) 전 구간 일치.