From e3088f7cc69e95d57acf1e5201bfcaf7940b9ad4 Mon Sep 17 00:00:00 2001 From: gahusb Date: Sun, 31 May 2026 21:33:55 +0900 Subject: [PATCH] =?UTF-8?q?docs(plan):=20=EC=A3=BC=EC=8B=9D=20=EB=B3=B4?= =?UTF-8?q?=EC=9C=A0=EC=A2=85=EB=AA=A9=20=EC=9D=B8=ED=85=94=EB=A6=AC?= =?UTF-8?q?=EC=A0=84=EC=8A=A4=20=EA=B5=AC=ED=98=84=20plan=20(7=20Phase,=20?= =?UTF-8?q?TDD)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 데이터모델+get_holdings → 2 기술분석·매도룰·decide_action → 3 이슈(market_events·news·portfolio_health) → 4 compute+brief+API → 5 agent-office EOD·아침브리핑 → 6 web-ui 탭 → 7 검증. 장중 가드는 후속. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../2026-05-31-stock-holdings-intelligence.md | 1102 +++++++++++++++++ 1 file changed, 1102 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-31-stock-holdings-intelligence.md diff --git a/docs/superpowers/plans/2026-05-31-stock-holdings-intelligence.md b/docs/superpowers/plans/2026-05-31-stock-holdings-intelligence.md new file mode 100644 index 0000000..edbc836 --- /dev/null +++ b/docs/superpowers/plans/2026-05-31-stock-holdings-intelligence.md @@ -0,0 +1,1102 @@ +# 주식 보유종목 인텔리전스 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) 전 구간 일치.