feat(stock): holdings_signals 테이블 + CRUD

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

stock/app/db.py

@@ -1,6 +1,7 @@
 import sqlite3
 import os
 import hashlib
+import json
 from typing import List, Dict, Any, Optional
 
 from app.screener.schema import ensure_screener_schema
@@ -103,6 +104,27 @@ def init_db():
         if "commission" not in sh_cols:
             conn.execute("ALTER TABLE sell_history ADD COLUMN commission REAL NOT NULL DEFAULT 0")
 
+        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);")
+
         # Screener 스키마 부트스트랩 (7테이블 + 디폴트 설정 시드)
         ensure_screener_schema(conn)
 
@@ -297,3 +319,48 @@ def get_asset_snapshots(days: int = 30) -> List[Dict[str, Any]]:
             ).fetchall()
             rows = list(reversed(rows))
         return [dict(r) for r in rows]
+
+
+# --- Holdings Signals CRUD ---
+
+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]

stock/app/test_holdings_db.py

@@ -0,0 +1,24 @@
+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