gahusb/web-page-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: rename stock-lab → stock (graduation)

Browse Source 
- git mv stock-lab/ → stock/
- docker-compose.yml: 서비스 키 + container_name + build.context +
  frontend.depends_on + agent-office STOCK_LAB_URL → STOCK_URL
- agent-office/app: config.py, service_proxy.py, agents/stock.py, tests/
  STOCK_LAB_URL → STOCK_URL
- nginx/default.conf: proxy_pass http://stock-lab → http://stock (3 lines)
- CLAUDE.md / README.md / STATUS.md / scripts/ 문구 갱신
- stock/ 내부 자기 참조 갱신

lab 네이밍 정책 (feedback_lab_naming.md) graduation.
API URL / Python import / DB 파일명 변경 없음.
This commit is contained in:
gahusb
2026-05-15 01:45:22 +09:00
parent 8812bd870a
commit ace0339d33
74 changed files with 67 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
stock/app/screener/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
"""Stock screener — KRX 강세주 분석 노드 기반 보드.
See docs/superpowers/specs/2026-05-12-stock-screener-board-design.md
"""
from .engine import Screener, ScreenContext, ScreenerResult
from .registry import NODE_REGISTRY, GATE_REGISTRY
__all__ = [
"Screener", "ScreenContext", "ScreenerResult",
"NODE_REGISTRY", "GATE_REGISTRY",
]

76
stock/app/screener/_test_fixtures.py Normal file
View File

@@ -0,0 +1,76 @@
"""Synthetic fixtures for screener tests — no DB / no FDR / no naver."""
import datetime as dt
import pandas as pd
def make_master(tickers: list[str], market_caps: dict | None = None,
preferred: set | None = None, managed: set | None = None) -> pd.DataFrame:
market_caps = market_caps or {t: 100_000_000_000 for t in tickers}
preferred = preferred or set()
managed = managed or set()
return pd.DataFrame([
{
"ticker": t,
"name": f"테스트{t}",
"market": "KOSPI",
"market_cap": market_caps.get(t),
"is_managed": int(t in managed),
"is_preferred": int(t in preferred),
"is_spac": 0,
"listed_date": None,
}
for t in tickers
]).set_index("ticker")
def make_prices(tickers: list[str], days: int = 260, start_close: int = 50000,
trend_pct: float = 0.0,
asof: dt.date = dt.date(2026, 5, 12)) -> pd.DataFrame:
"""trend_pct: 일별 종가 등락률(%). 양수면 상승 추세."""
rows = []
for t in tickers:
close = start_close
for i in range(days):
day_idx = days - 1 - i # asof가 마지막
date = asof - dt.timedelta(days=day_idx)
high = int(close * 1.012)
low = int(close * 0.988)
rows.append({
"ticker": t, "date": date.isoformat(),
"open": close, "high": high, "low": low, "close": close,
"volume": 1_000_000, "value": close * 1_000_000,
})
close = int(close * (1 + trend_pct / 100))
return pd.DataFrame(rows)
def make_flow(tickers: list[str], days: int = 260,
foreign_per_day: dict | None = None,
asof: dt.date = dt.date(2026, 5, 12)) -> pd.DataFrame:
foreign_per_day = foreign_per_day or {t: 0 for t in tickers}
rows = []
for t in tickers:
for i in range(days):
day_idx = days - 1 - i
date = asof - dt.timedelta(days=day_idx)
rows.append({
"ticker": t, "date": date.isoformat(),
"foreign_net": foreign_per_day.get(t, 0),
"institution_net": 0,
})
return pd.DataFrame(rows)
def make_kospi(days: int = 260, start: int = 2500, trend_pct: float = 0.0,
asof: dt.date = dt.date(2026, 5, 12)) -> pd.Series:
values = []
dates = []
v = start
for i in range(days):
day_idx = days - 1 - i
d = asof - dt.timedelta(days=day_idx)
dates.append(d.isoformat())
values.append(v)
v = v * (1 + trend_pct / 100)
return pd.Series(values, index=dates, name="kospi")

0
stock/app/screener/ai_news/__init__.py Normal file
View File

103
stock/app/screener/ai_news/analyzer.py Normal file
View File

@@ -0,0 +1,103 @@
"""Claude Haiku 기반 종목 뉴스 호재/악재 분석."""
from __future__ import annotations
import json
import logging
import os
from typing import Any, Dict, List
log = logging.getLogger(__name__)
DEFAULT_MODEL = os.getenv("AI_NEWS_MODEL", "claude-haiku-4-5-20251001")
PROMPT_TEMPLATE = """다음은 종목 {name}({ticker})에 대한 최근 뉴스 {n}개의 헤드라인입니다.
{news_block}
이 뉴스들이 종목에 호재인지 악재인지 평가하세요.
score: -10(매우 강한 악재) ~ +10(매우 강한 호재) 사이의 실수. 0은 중립.
reason: 30자 이내 한 줄 근거.
JSON으로만 응답하세요. 다른 텍스트 금지:
{{"score": <float>, "reason": "<string>"}}"""
def _clamp(x: float, lo: float = -10.0, hi: float = 10.0) -> float:
return max(lo, min(hi, x))
def _format_news_block(news: List[Dict[str, Any]]) -> str:
"""news dict 리스트 → prompt 에 들어가는 텍스트 블록.
summary 가 있으면 title 다음 줄에 indent 해서 포함 (최대 200자).
pub_date 가 있으면 title 앞에 표시.
"""
lines: List[str] = []
for n in news:
date = (n.get("pub_date") or "").strip()
title = (n.get("title") or "").strip()
summary = (n.get("summary") or "").strip()
prefix = f"[{date}] " if date else ""
if summary:
lines.append(f"- {prefix}{title}\n {summary[:200]}")
else:
lines.append(f"- {prefix}{title}")
return "\n".join(lines)
async def score_sentiment(
llm,
ticker: str,
news: List[Dict[str, Any]],
*,
name: str | None = None,
model: str = DEFAULT_MODEL,
) -> Dict[str, Any]:
"""Returns {ticker, score_raw, reason, news_count, tokens_input, tokens_output, model}."""
news_block = _format_news_block(news)
prompt = PROMPT_TEMPLATE.format(
name=name or ticker, ticker=ticker,
n=len(news), news_block=news_block,
)
resp = await llm.messages.create(
model=model,
max_tokens=200,
temperature=0,
system="너는 한국 주식 뉴스 감성 분석가다. JSON 객체 하나만 반환한다.",
messages=[
{"role": "user", "content": prompt},
# Assistant prefill — 첫 토큰을 강제로 '{' 로 시작해 JSON 응답을 보장
{"role": "assistant", "content": "{"},
],
)
raw = resp.content[0].text if resp.content else ""
# prefill '{' 이 응답에 포함되지 않으므로 다시 붙임
text = "{" + raw if not raw.lstrip().startswith("{") else raw
in_tokens = int(getattr(resp.usage, "input_tokens", 0) or 0)
out_tokens = int(getattr(resp.usage, "output_tokens", 0) or 0)
try:
data = json.loads(text)
score = _clamp(float(data["score"]))
reason = str(data["reason"])[:200]
return {
"ticker": ticker,
"score_raw": score,
"reason": reason,
"news_count": len(news),
"tokens_input": in_tokens,
"tokens_output": out_tokens,
"model": model,
}
except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e:
log.warning("ai_news parse fail for %s: %s (raw=%r)", ticker, e, text[:100])
return {
"ticker": ticker,
"score_raw": 0.0,
"reason": f"parse fail: {e!s}"[:200],
"news_count": len(news),
"tokens_input": in_tokens,
"tokens_output": out_tokens,
"model": model,
}

70
stock/app/screener/ai_news/articles_source.py Normal file
View File

@@ -0,0 +1,70 @@
"""기존 articles 테이블에서 종목별 뉴스 매핑."""
from __future__ import annotations
import datetime as dt
import logging
import sqlite3
from typing import Any, Dict, List, Tuple
log = logging.getLogger(__name__)
def gather_articles_for_tickers(
conn: sqlite3.Connection,
tickers: List[str],
asof: dt.date,
*,
window_days: int = 1,
max_per_ticker: int = 5,
) -> Tuple[Dict[str, List[Dict[str, Any]]], Dict[str, int]]:
"""articles 에서 ticker.name substring 매칭으로 종목별 뉴스 dict 반환.
Returns:
(
{ticker: [{"title": str, "summary": str, "press": str, "pub_date": str}, ...]},
{"total_articles": int, "matched_pairs": int, "hit_tickers": int},
)
"""
out: Dict[str, List[Dict[str, Any]]] = {t: [] for t in tickers}
stats = {"total_articles": 0, "matched_pairs": 0, "hit_tickers": 0}
if not tickers:
return out, stats
cutoff = (asof - dt.timedelta(days=window_days)).isoformat()
placeholders = ",".join("?" * len(tickers))
name_rows = conn.execute(
f"SELECT ticker, name FROM krx_master WHERE ticker IN ({placeholders})",
tickers,
).fetchall()
# 2글자 미만 회사명은 false positive 위험으로 제외
name_map = {r[0]: r[1] for r in name_rows if r[1] and len(r[1]) >= 2}
articles = conn.execute(
"SELECT title, summary, press, pub_date, crawled_at "
"FROM articles WHERE crawled_at >= ? ORDER BY crawled_at DESC",
(cutoff,),
).fetchall()
stats["total_articles"] = len(articles)
for a in articles:
title = (a[0] or "").strip()
summary = (a[1] or "").strip()
haystack = title + " " + summary
for ticker, name in name_map.items():
if name not in haystack:
continue
if len(out[ticker]) >= max_per_ticker:
continue
out[ticker].append({
"title": title,
"summary": summary,
"press": a[2] or "",
"pub_date": a[3] or "",
})
stats["matched_pairs"] += 1
stats["hit_tickers"] = sum(1 for arts in out.values() if arts)
return out, stats

141
stock/app/screener/ai_news/pipeline.py Normal file
View File

@@ -0,0 +1,141 @@
"""ai_news refresh pipeline — 시총 상위 N종목 병렬 처리."""
from __future__ import annotations
import asyncio
import datetime as dt
import logging
import os
import sqlite3
import time
from typing import Any, Dict, List
from . import scraper as _scraper # legacy, kept for backward import
from . import analyzer as _analyzer
from . import articles_source # 신규
log = logging.getLogger(__name__)
DEFAULT_TOP_N = 100
DEFAULT_CONCURRENCY = 10
DEFAULT_NEWS_PER_TICKER = 5
def _top_market_cap_tickers(conn: sqlite3.Connection, n: int) -> List[str]:
rows = conn.execute(
"SELECT ticker FROM krx_master "
"WHERE market_cap IS NOT NULL AND is_preferred=0 AND is_spac=0 "
"ORDER BY market_cap DESC LIMIT ?",
(n,),
).fetchall()
return [r[0] for r in rows]
def _make_llm():
"""Anthropic AsyncClient — env에 ANTHROPIC_API_KEY 필수."""
from anthropic import AsyncAnthropic
return AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
async def _process_one(
ticker: str, name: str, articles: List[Dict[str, Any]],
sem: asyncio.Semaphore, llm, model: str,
) -> Dict[str, Any]:
async with sem:
return await _analyzer.score_sentiment(
llm, ticker, articles, name=name, model=model,
)
def _upsert_news_sentiment(
conn: sqlite3.Connection, asof: dt.date,
rows: List[Dict[str, Any]], *, source: str = "articles",
) -> None:
iso = asof.isoformat()
data = [
(
r["ticker"], iso, r["score_raw"], r["reason"], r["news_count"],
r["tokens_input"], r["tokens_output"], r["model"], source,
)
for r in rows
]
conn.executemany(
"""INSERT INTO news_sentiment
(ticker, date, score_raw, reason, news_count,
tokens_input, tokens_output, model, source)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(ticker, date) DO UPDATE SET
score_raw=excluded.score_raw,
reason=excluded.reason,
news_count=excluded.news_count,
tokens_input=excluded.tokens_input,
tokens_output=excluded.tokens_output,
model=excluded.model,
source=excluded.source
""",
data,
)
conn.commit()
async def refresh_daily(
conn: sqlite3.Connection,
asof: dt.date,
*,
top_n: int = DEFAULT_TOP_N,
concurrency: int = DEFAULT_CONCURRENCY,
max_news_per_ticker: int = DEFAULT_NEWS_PER_TICKER,
window_days: int = 1,
model: str = _analyzer.DEFAULT_MODEL,
) -> Dict[str, Any]:
started = time.time()
tickers = _top_market_cap_tickers(conn, n=top_n)
name_map = {
r[0]: r[1] for r in conn.execute(
f"SELECT ticker, name FROM krx_master WHERE ticker IN "
f"({','.join('?' * len(tickers))})", tickers,
).fetchall()
} if tickers else {}
articles_by_ticker, mapping_stats = articles_source.gather_articles_for_tickers(
conn, tickers, asof,
window_days=window_days,
max_per_ticker=max_news_per_ticker,
)
sem = asyncio.Semaphore(concurrency)
async with _make_llm() as llm:
tasks = []
for t in tickers:
arts = articles_by_ticker.get(t, [])
if not arts:
continue # 매핑 0 — score 미생성
tasks.append(_process_one(t, name_map.get(t, t), arts, sem, llm, model))
raw_results = await asyncio.gather(*tasks, return_exceptions=True)
successes: List[Dict[str, Any]] = []
failures: List[str] = []
for r in raw_results:
if isinstance(r, BaseException):
failures.append(repr(r))
elif isinstance(r, dict):
successes.append(r)
if successes:
_upsert_news_sentiment(conn, asof, successes, source="articles")
top_pos = sorted(successes, key=lambda r: -r["score_raw"])[:5]
top_neg = sorted(successes, key=lambda r: r["score_raw"])[:5]
return {
"asof": asof.isoformat(),
"updated": len(successes),
"failures": failures,
"duration_sec": round(time.time() - started, 2),
"tokens_input": sum(r["tokens_input"] for r in successes),
"tokens_output": sum(r["tokens_output"] for r in successes),
"top_pos": top_pos,
"top_neg": top_neg,
"model": model,
"mapping": mapping_stats,
}

46
stock/app/screener/ai_news/scraper.py Normal file
View File

@@ -0,0 +1,46 @@
"""[DEPRECATED] 네이버 finance 종목 뉴스 스크래핑.
본 모듈은 ai_news Phase 1 (2026-05-14) 에서 더 이상 파이프라인에서 사용되지 않음.
데이터 소스는 stock 의 articles 테이블 (ai_news/articles_source.py) 로 전환됨.
삭제 시점: Phase 2 (DART 도입) 결정 후. IC 검증 4주 누적 후 노드 활성화
여부에 따라 본 모듈을 (a) 완전 삭제 또는 (b) ensemble fallback 으로 재활용.
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List
from bs4 import BeautifulSoup
log = logging.getLogger(__name__)
NAVER_NEWS_URL = "https://finance.naver.com/item/news_news.naver"
NAVER_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "https://finance.naver.com/",
}
async def fetch_news(client, ticker: str, n: int = 5) -> List[Dict[str, Any]]:
"""Scrape top N news headlines for a ticker. Returns [] on any failure."""
try:
r = await client.get(NAVER_NEWS_URL, params={"code": ticker, "page": 1})
except Exception as e:
log.warning("ai_news scrape http error for %s: %s", ticker, e)
return []
if r.status_code != 200:
return []
soup = BeautifulSoup(r.text, "lxml")
out: List[Dict[str, Any]] = []
for row in soup.select("table.type5 tbody tr")[:n]:
title_el = row.select_one("td.title a")
date_el = row.select_one("td.date")
if not title_el or not date_el:
continue
out.append({
"title": title_el.get_text(strip=True),
"date": date_el.get_text(strip=True),
})
return out

73
stock/app/screener/ai_news/telegram.py Normal file
View File

@@ -0,0 +1,73 @@
"""ai_news Top 5/5 텔레그램 메시지 빌더 (MarkdownV2)."""
from __future__ import annotations
from typing import Any, Dict, List
_MD_SPECIAL = r"_*[]()~`>#+-=|{}.!\\"
def _escape(text: str) -> str:
return "".join("\\" + c if c in _MD_SPECIAL else c for c in str(text))
def _cost_won(tokens_input: int, tokens_output: int) -> int:
"""Claude Haiku 가격 환산 (대략): in $1/M × ₩1300, out $5/M × ₩1300."""
return int(tokens_input * 0.0013 + tokens_output * 0.0065)
def _row_line(idx: int, r: Dict[str, Any]) -> str:
score = r["score_raw"]
# score 문자열 자체를 _escape 통과 — '+', '-', '.' 모두 MarkdownV2 reserved
score_str = _escape(f"{score:+.1f}")
name = r.get("name") or ""
ticker = r["ticker"]
label = (
f"{_escape(name)} \\({_escape(ticker)}\\)"
if name else _escape(ticker)
)
return f"{idx}\\. {label} \\({score_str}\\) — {_escape(r['reason'])}"
def build_message(
*,
asof: str,
top_pos: List[Dict[str, Any]],
top_neg: List[Dict[str, Any]],
tokens_input: int,
tokens_output: int,
mapping: Dict[str, int] | None = None,
) -> str:
lines: List[str] = [
f"🌅 *AI 뉴스 분석* \\({_escape(asof)} 08:00\\)",
"",
"📈 *호재 Top 5*",
]
if top_pos:
for i, r in enumerate(top_pos, 1):
lines.append(_row_line(i, r))
else:
lines.append(_escape("- (없음)"))
lines += ["", "📉 *악재 Top 5*"]
if top_neg:
for i, r in enumerate(top_neg, 1):
lines.append(_row_line(i, r))
else:
lines.append(_escape("- (없음)"))
cost = _cost_won(tokens_input, tokens_output)
mapping_part = ""
if mapping:
mapping_part = (
f"매핑 {mapping['hit_tickers']}/100 ticker "
f"\\({mapping['matched_pairs']}쌍 / articles {mapping['total_articles']}건\\) · "
)
lines += [
"",
f"_분석: 시총 상위 100종목 · {mapping_part}"
f"토큰 {tokens_input:,} in / {tokens_output:,} out · "
f"약 ₩{cost:,}_",
]
return "\n".join(lines)

125
stock/app/screener/ai_news/validation.py Normal file
View File

@@ -0,0 +1,125 @@
"""AI news sentiment validation — Spearman IC vs forward returns.
핵심 metric: 일자별 score_raw 와 다음 N일 forward return 의 Spearman 상관.
4주+ 누적 후 IC mean > 0.05 면 weight 활성화 가치 있음.
"""
from __future__ import annotations
import datetime as dt
import sqlite3
from typing import Any, Dict, List, Optional
import pandas as pd
def _spearman(a: pd.Series, b: pd.Series) -> Optional[float]:
"""Spearman rank correlation. None if insufficient/degenerate data."""
if len(a) < 5 or len(b) < 5:
return None
if a.std(ddof=0) == 0 or b.std(ddof=0) == 0:
return None
return float(a.rank().corr(b.rank()))
def compute_ic(
conn: sqlite3.Connection,
*,
days: int = 30,
horizon: int = 1,
min_news_count: int = 1,
asof_today: Optional[dt.date] = None,
) -> Dict[str, Any]:
"""Compute daily Spearman IC of ai_news.score_raw vs forward return.
Returns:
{
"horizon_days": int,
"min_news_count": int,
"window_days": int,
"ic_count": int, # 유효 일수
"ic_mean": float | None,
"ic_std": float | None,
"ic_per_day": [{"date": "YYYY-MM-DD", "ic": float, "n": int}, ...],
"verdict": "skip" | "weak" | "strong",
}
verdict:
- skip: ic_count < 10
- weak: ic_mean in [-0.05, 0.05]
- strong: |ic_mean| > 0.05
"""
asof_today = asof_today or dt.date.today()
cutoff = (asof_today - dt.timedelta(days=days)).isoformat()
sentiment = pd.read_sql_query(
"SELECT ticker, date, score_raw, news_count "
"FROM news_sentiment WHERE date >= ? AND news_count >= ? ORDER BY date",
conn, params=(cutoff, min_news_count),
)
if sentiment.empty:
return _empty_result(days, horizon, min_news_count)
# forward return 조회: 각 (ticker, date) 에 대해 close[date+horizon] / close[date] - 1
prices = pd.read_sql_query(
"SELECT ticker, date, close FROM krx_daily_prices "
"WHERE date >= ? ORDER BY ticker, date",
conn, params=(cutoff,),
)
if prices.empty:
return _empty_result(days, horizon, min_news_count)
prices = prices.sort_values(["ticker", "date"])
prices["fwd_close"] = prices.groupby("ticker", group_keys=False)["close"].shift(-horizon)
prices["fwd_ret"] = prices["fwd_close"] / prices["close"] - 1.0
merged = sentiment.merge(
prices[["ticker", "date", "fwd_ret"]], on=["ticker", "date"], how="inner"
)
merged = merged.dropna(subset=["fwd_ret"])
if merged.empty:
return _empty_result(days, horizon, min_news_count)
ic_rows: List[Dict[str, Any]] = []
for date, grp in merged.groupby("date"):
ic = _spearman(grp["score_raw"], grp["fwd_ret"])
if ic is not None:
ic_rows.append({"date": date, "ic": ic, "n": int(len(grp))})
if not ic_rows:
return _empty_result(days, horizon, min_news_count)
ic_series = pd.Series([r["ic"] for r in ic_rows], dtype=float)
ic_mean = float(ic_series.mean())
ic_std = float(ic_series.std(ddof=0)) if len(ic_series) > 1 else 0.0
if len(ic_rows) < 10:
verdict = "skip"
elif abs(ic_mean) > 0.05:
verdict = "strong"
else:
verdict = "weak"
return {
"horizon_days": horizon,
"min_news_count": min_news_count,
"window_days": days,
"ic_count": len(ic_rows),
"ic_mean": round(ic_mean, 4),
"ic_std": round(ic_std, 4),
"ic_per_day": ic_rows,
"verdict": verdict,
}
def _empty_result(days: int, horizon: int, min_news_count: int) -> Dict[str, Any]:
return {
"horizon_days": horizon,
"min_news_count": min_news_count,
"window_days": days,
"ic_count": 0,
"ic_mean": None,
"ic_std": None,
"ic_per_day": [],
"verdict": "skip",
}

167
stock/app/screener/engine.py Normal file
View File

@@ -0,0 +1,167 @@
"""Screener engine — ScreenContext (Phase 0) + Screener/combine (Phase 2)."""
from __future__ import annotations
import datetime as dt
import sqlite3
from dataclasses import dataclass, replace
import pandas as pd
@dataclass(frozen=True)
class ScreenContext:
"""1회 실행 동안 공유되는 읽기 전용 데이터 컨테이너."""
master: pd.DataFrame # index=ticker
prices: pd.DataFrame # cols: ticker,date,open,high,low,close,volume,value
flow: pd.DataFrame # cols: ticker,date,foreign_net,institution_net
kospi: pd.Series # index=date(str), name="kospi"
asof: dt.date
news_sentiment: "pd.DataFrame | None" = None
@classmethod
def load(cls, conn: sqlite3.Connection, asof: dt.date,
lookback_days: int = 252 * 2) -> "ScreenContext":
cutoff = (asof - dt.timedelta(days=int(lookback_days * 1.5))).isoformat()
asof_iso = asof.isoformat()
master = pd.read_sql_query(
"SELECT * FROM krx_master",
conn, index_col="ticker",
)
prices = pd.read_sql_query(
"SELECT ticker,date,open,high,low,close,volume,value "
"FROM krx_daily_prices WHERE date BETWEEN ? AND ? ORDER BY date",
conn, params=(cutoff, asof_iso),
)
flow = pd.read_sql_query(
"SELECT ticker,date,foreign_net,institution_net "
"FROM krx_flow WHERE date BETWEEN ? AND ? ORDER BY date",
conn, params=(cutoff, asof_iso),
)
news_sentiment = pd.read_sql_query(
"SELECT ticker, score_raw, news_count FROM news_sentiment WHERE date = ?",
conn, params=(asof_iso,),
)
# KOSPI 지수: MVP에서는 005930(삼성전자) 종가를 시장 대용으로 사용.
# 후속 슬라이스에서 ^KS11 별도 캐시.
kospi = pd.Series(dtype=float, name="kospi")
if "005930" in master.index and not prices.empty:
sub = prices[prices["ticker"] == "005930"].set_index("date")["close"]
kospi = sub.copy()
kospi.name = "kospi"
return cls(master=master, prices=prices, flow=flow, kospi=kospi, asof=asof,
news_sentiment=news_sentiment)
def restrict(self, tickers) -> "ScreenContext":
tickers = pd.Index(tickers)
return replace(
self,
master=self.master.loc[self.master.index.intersection(tickers)],
prices=self.prices[self.prices["ticker"].isin(tickers)],
flow=self.flow[self.flow["ticker"].isin(tickers)],
)
def latest_close(self) -> pd.Series:
if self.prices.empty:
return pd.Series(dtype=float)
return self.prices.sort_values("date").groupby("ticker")["close"].last()
def latest_high(self) -> pd.Series:
if self.prices.empty:
return pd.Series(dtype=float)
return self.prices.sort_values("date").groupby("ticker")["high"].last()
# ---- combine + Screener (Phase 2) ----
from . import position_sizer as _ps
def combine(scores: dict, weights: dict) -> pd.Series:
"""Weighted average across score nodes. ValueError if all weights = 0."""
active = {k: w for k, w in weights.items() if w > 0 and k in scores}
if not active:
raise ValueError("no active score nodes (all weights = 0)")
df = pd.DataFrame({k: scores[k] for k in active})
w = pd.Series(active)
weighted = (df.fillna(0).multiply(w, axis=1)).sum(axis=1) / w.sum()
return weighted
@dataclass
class ScreenerResult:
asof: dt.date
survivors_count: int
scores: dict # node name → pd.Series
weights: dict
ranked: pd.Series # ticker → total_score (sorted desc, head=top_n)
rows: list # list of dicts (for serialization)
warnings: list
class Screener:
def __init__(self, gate, score_nodes, weights: dict, node_params: dict,
gate_params: dict, top_n: int = 20, sizer_params: dict = None):
self.gate = gate
self.score_nodes = score_nodes
self.weights = weights
self.node_params = node_params
self.gate_params = gate_params
self.top_n = top_n
self.sizer_params = sizer_params or {"atr_window": 14, "atr_stop_mult": 2.0, "rr_ratio": 2.0}
def run(self, ctx: ScreenContext) -> ScreenerResult:
warnings: list = []
survivors = self.gate.filter(ctx, self.gate_params)
if len(survivors) == 0:
raise ValueError("no survivors after hygiene gate")
if len(survivors) < 100:
warnings.append(f"survivors_count={len(survivors)} < 100 — 백분위 정규화 신뢰도 낮음")
scoped = ctx.restrict(survivors)
scores: dict = {}
for n in self.score_nodes:
w = self.weights.get(n.name, 0)
if w <= 0:
continue
try:
scores[n.name] = n.compute(scoped, self.node_params.get(n.name, {}))
except Exception as e:
warnings.append(f"node '{n.name}' failed: {e}")
scores[n.name] = pd.Series(0.0, index=scoped.master.index)
total = combine(scores, self.weights)
ranked = total.sort_values(ascending=False).head(self.top_n)
sizing = _ps.plan_positions(scoped, list(ranked.index), self.sizer_params)
latest_close = scoped.latest_close()
rows = []
for rank_idx, ticker in enumerate(ranked.index, start=1):
s = sizing.get(ticker, {})
row = {
"rank": rank_idx,
"ticker": ticker,
"name": str(scoped.master.loc[ticker, "name"]),
"total_score": float(ranked.loc[ticker]),
"scores": {k: float(v.get(ticker, 0.0)) for k, v in scores.items()},
"close": int(latest_close.get(ticker, 0)),
"market_cap": int(scoped.master.loc[ticker, "market_cap"] or 0),
"entry_price": s.get("entry_price"),
"stop_price": s.get("stop_price"),
"target_price": s.get("target_price"),
"atr14": s.get("atr14"),
"r_pct": s.get("r_pct"),
}
rows.append(row)
return ScreenerResult(
asof=ctx.asof, survivors_count=len(survivors),
scores=scores, weights=self.weights,
ranked=ranked, rows=rows, warnings=warnings,
)

0
stock/app/screener/nodes/__init__.py Normal file
View File

36
stock/app/screener/nodes/ai_news.py Normal file
View File

@@ -0,0 +1,36 @@
"""AI 뉴스 호재/악재 점수 노드.
ScreenContext.news_sentiment (DataFrame: ticker, score_raw, news_count) 를
min_news_count 로 필터한 뒤 percentile_rank 로 0~100 변환.
"""
from __future__ import annotations
import pandas as pd
from .base import ScoreNode, percentile_rank
class AiNewsSentiment(ScoreNode):
name = "ai_news"
label = "AI 뉴스 호재/악재"
default_params = {"min_news_count": 1}
param_schema = {
"type": "object",
"properties": {
"min_news_count": {
"type": "integer", "minimum": 0, "default": 1,
"description": "최소 분석 뉴스 수. 미만이면 점수 미산출.",
},
},
}
def compute(self, ctx, params: dict) -> pd.Series:
df = getattr(ctx, "news_sentiment", None)
if df is None or df.empty:
return pd.Series(dtype=float)
min_news = int(params.get("min_news_count", 1))
df = df[df["news_count"] >= min_news]
if df.empty:
return pd.Series(dtype=float)
return percentile_rank(df.set_index("ticker")["score_raw"])

40
stock/app/screener/nodes/base.py Normal file
View File

@@ -0,0 +1,40 @@
"""Node base classes + helpers."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any, ClassVar
import pandas as pd
class ScoreNode(ABC):
name: ClassVar[str]
label: ClassVar[str]
default_params: ClassVar[dict]
param_schema: ClassVar[dict]
@abstractmethod
def compute(self, ctx: "Any", params: dict) -> pd.Series:
"""returns Series indexed by ticker, 0..100 float."""
class GateNode(ABC):
name: ClassVar[str]
label: ClassVar[str]
default_params: ClassVar[dict]
param_schema: ClassVar[dict]
@abstractmethod
def filter(self, ctx: "Any", params: dict) -> pd.Index:
"""returns surviving tickers."""
def percentile_rank(series: pd.Series) -> pd.Series:
"""Percentile rank in [0, 100]. All-equal → 50. NaN preserved."""
if series.empty:
return series.astype(float)
if series.dropna().nunique() == 1:
return pd.Series(50.0, index=series.index)
ranked = series.rank(pct=True, na_option="keep") * 100.0
return ranked

33
stock/app/screener/nodes/foreign_buy.py Normal file
View File

@@ -0,0 +1,33 @@
"""외국인 N일 누적 순매수 강도 (시총 대비)."""
import pandas as pd
from .base import ScoreNode, percentile_rank
class ForeignBuy(ScoreNode):
name = "foreign_buy"
label = "외국인 누적 순매수"
default_params = {"window_days": 5}
param_schema = {
"type": "object",
"properties": {
"window_days": {"type": "integer", "minimum": 1, "maximum": 60, "default": 5}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
window = int(params.get("window_days", 5))
flow = ctx.flow
if flow.empty:
return pd.Series(dtype=float)
last_dates = (
flow.sort_values("date").groupby("ticker").tail(window)
)
net_sum = last_dates.groupby("ticker")["foreign_net"].sum()
market_cap = ctx.master["market_cap"].fillna(0).reindex(net_sum.index)
raw = (net_sum / market_cap.replace(0, pd.NA)).astype(float)
return percentile_rank(raw).fillna(50.0)

30
stock/app/screener/nodes/high52w.py Normal file
View File

@@ -0,0 +1,30 @@
"""52주 신고가 근접도 (룰 기반: 70% 미만 0점, 100% 도달 100점, 선형)."""
import pandas as pd
from .base import ScoreNode
class High52WProximity(ScoreNode):
name = "high52w"
label = "52주 신고가 근접도"
default_params = {"window_days": 252}
param_schema = {
"type": "object",
"properties": {
"window_days": {"type": "integer", "minimum": 60, "maximum": 504, "default": 252}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
window = int(params.get("window_days", 252))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
last = ordered.groupby("ticker").tail(window)
agg = last.groupby("ticker").agg(close=("close", "last"), high=("high", "max"))
proximity = (agg["close"] / agg["high"]).clip(upper=1.0)
score = ((proximity - 0.7) / 0.3).clip(lower=0.0, upper=1.0) * 100.0
return score.fillna(0.0)

81
stock/app/screener/nodes/hygiene.py Normal file
View File

@@ -0,0 +1,81 @@
"""HygieneGate — pre-filter for screener."""
from __future__ import annotations
import pandas as pd
from .base import GateNode
class HygieneGate(GateNode):
name = "hygiene"
label = "위생 게이트"
default_params = {
"min_market_cap_won": 50_000_000_000,
"min_avg_value_won": 500_000_000,
"min_listed_days": 60,
"skip_managed": True,
"skip_preferred": True,
"skip_spac": True,
"skip_halted_days": 3,
}
param_schema = {
"type": "object",
"properties": {
"min_market_cap_won": {"type": "integer", "minimum": 0},
"min_avg_value_won": {"type": "integer", "minimum": 0},
"min_listed_days": {"type": "integer", "minimum": 0},
"skip_managed": {"type": "boolean"},
"skip_preferred": {"type": "boolean"},
"skip_spac": {"type": "boolean"},
"skip_halted_days": {"type": "integer", "minimum": 0},
},
}
def filter(self, ctx, params: dict) -> pd.Index:
master = ctx.master.copy()
prices = ctx.prices
# 시총
master = master[master["market_cap"].fillna(0) >= params["min_market_cap_won"]]
# 우선주·관리·스팩
if params.get("skip_preferred", True):
master = master[master["is_preferred"] == 0]
if params.get("skip_managed", True):
master = master[master["is_managed"] == 0]
if params.get("skip_spac", True):
master = master[master["is_spac"] == 0]
candidates = master.index
# 20일 평균 거래대금
if not prices.empty:
recent20 = (
prices[prices["ticker"].isin(candidates)]
.sort_values("date")
.groupby("ticker")
.tail(20)
)
avg_value = recent20.groupby("ticker")["value"].mean()
ok = avg_value[avg_value >= params["min_avg_value_won"]].index
candidates = candidates.intersection(ok)
# 최근 N일 거래정지 (volume==0 N일 이상)
halted_days = params.get("skip_halted_days", 3)
if halted_days > 0 and not prices.empty:
recent = (
prices[prices["ticker"].isin(candidates)]
.sort_values("date")
.groupby("ticker")
.tail(halted_days)
)
zero_count = (
recent.assign(z=lambda d: (d["volume"] == 0).astype(int))
.groupby("ticker")["z"].sum()
)
healthy = zero_count[zero_count < halted_days].index
candidates = candidates.intersection(healthy)
# 상장 N일 — MVP에선 listed_date null 허용, null이면 통과
return pd.Index(candidates)

51
stock/app/screener/nodes/ma_alignment.py Normal file
View File

@@ -0,0 +1,51 @@
"""이평선 정배열 점수 — 5개 조건 충족 개수 / 5 × 100."""
import pandas as pd
from .base import ScoreNode
class MaAlignment(ScoreNode):
name = "ma_alignment"
label = "이평선 정배열"
default_params = {"ma_periods": [50, 150, 200]}
param_schema = {
"type": "object",
"properties": {
"ma_periods": {"type": "array", "items": {"type": "integer"}}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
ma_periods = params.get("ma_periods", self.default_params["ma_periods"])
if len(ma_periods) != 3:
raise ValueError("ma_periods must have 3 entries (short, medium, long)")
ma_s, ma_m, ma_l = (int(x) for x in ma_periods)
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
min_history = max(252, ma_l)
def _score(s: pd.Series) -> float:
closes = s.astype(float).reset_index(drop=True)
if len(closes) < min_history:
return float("nan")
close = closes.iloc[-1]
ma_short = closes.rolling(ma_s).mean().iloc[-1]
ma_medium = closes.rolling(ma_m).mean().iloc[-1]
ma_long = closes.rolling(ma_l).mean().iloc[-1]
low52 = closes.iloc[-252:].min()
conds = [
close > ma_short,
ma_short > ma_medium,
ma_medium > ma_long,
close > ma_long,
close >= low52 * 1.25,
]
return sum(conds) / 5 * 100.0
raw = ordered.groupby("ticker", group_keys=False)["close"].apply(_score)
return raw.fillna(0.0)

34
stock/app/screener/nodes/momentum.py Normal file
View File

@@ -0,0 +1,34 @@
"""20일 모멘텀."""
import pandas as pd
from .base import ScoreNode, percentile_rank
class Momentum20(ScoreNode):
name = "momentum"
label = "20일 모멘텀"
default_params = {"window_days": 20}
param_schema = {
"type": "object",
"properties": {
"window_days": {"type": "integer", "minimum": 5, "maximum": 120, "default": 20}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
window = int(params.get("window_days", 20))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
last = ordered.groupby("ticker").tail(window + 1)
def _ret(s):
if len(s) < window + 1:
return float("nan")
return s.iloc[-1] / s.iloc[0] - 1
raw = last.groupby("ticker")["close"].apply(_ret)
return percentile_rank(raw).fillna(50.0)

48
stock/app/screener/nodes/rs_rating.py Normal file
View File

@@ -0,0 +1,48 @@
"""RS Rating — IBD 가중 (3m=2,6m=1,9m=1,12m=1)."""
import pandas as pd
from .base import ScoreNode, percentile_rank
_PERIOD_TO_DAYS = {"3m": 63, "6m": 126, "9m": 189, "12m": 252}
class RsRating(ScoreNode):
name = "rs_rating"
label = "RS Rating (시장 대비 상대강도)"
default_params = {"weights": {"3m": 2, "6m": 1, "9m": 1, "12m": 1}}
param_schema = {
"type": "object",
"properties": {
"weights": {"type": "object"}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
weights: dict = params.get("weights", self.default_params["weights"])
prices = ctx.prices
kospi = ctx.kospi
if prices.empty or kospi.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
def _excess_for_ticker(g: pd.DataFrame) -> float:
closes = g.set_index("date")["close"]
total = 0.0
wsum = 0.0
for period, w in weights.items():
k = _PERIOD_TO_DAYS.get(period, 0)
if len(closes) <= k or len(kospi) <= k:
continue
r_stock = closes.iloc[-1] / closes.iloc[-(k + 1)] - 1
r_market = kospi.iloc[-1] / kospi.iloc[-(k + 1)] - 1
total += w * (r_stock - r_market)
wsum += w
return total / wsum if wsum else float("nan")
raw = ordered.groupby("ticker", group_keys=False).apply(
_excess_for_ticker, include_groups=False
)
return percentile_rank(raw).fillna(50.0)

40
stock/app/screener/nodes/vcp_lite.py Normal file
View File

@@ -0,0 +1,40 @@
"""VCP-lite — 단기/장기 일중 변동성 비율 기반 수축률."""
import pandas as pd
from .base import ScoreNode, percentile_rank
class VcpLite(ScoreNode):
name = "vcp_lite"
label = "VCP-lite (변동성 수축)"
default_params = {"short_window": 40, "long_window": 252}
param_schema = {
"type": "object",
"properties": {
"short_window": {"type": "integer", "minimum": 10, "maximum": 120, "default": 40},
"long_window": {"type": "integer", "minimum": 60, "maximum": 504, "default": 252},
},
}
def compute(self, ctx, params: dict) -> pd.Series:
short_w = int(params.get("short_window", 40))
long_w = int(params.get("long_window", 252))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date").copy()
ordered["range_pct"] = (ordered["high"] - ordered["low"]) / ordered["close"]
def _ratio(s: pd.Series) -> float:
if len(s) < long_w:
return float("nan")
short_vol = s.tail(short_w).mean()
long_vol = s.tail(long_w).mean()
if long_vol == 0 or pd.isna(long_vol):
return float("nan")
return 1 - (short_vol / long_vol)
raw = ordered.groupby("ticker", group_keys=False)["range_pct"].apply(_ratio)
return percentile_rank(raw).fillna(50.0)

40
stock/app/screener/nodes/volume_surge.py Normal file
View File

@@ -0,0 +1,40 @@
"""거래량 급증 — log1p(recent/baseline)."""
import numpy as np
import pandas as pd
from .base import ScoreNode, percentile_rank
class VolumeSurge(ScoreNode):
name = "volume_surge"
label = "거래량 급증"
default_params = {"baseline_days": 20, "eval_days": 3}
param_schema = {
"type": "object",
"properties": {
"baseline_days": {"type": "integer", "minimum": 5, "maximum": 60, "default": 20},
"eval_days": {"type": "integer", "minimum": 1, "maximum": 10, "default": 3},
},
}
def compute(self, ctx, params: dict) -> pd.Series:
baseline = int(params.get("baseline_days", 20))
eval_d = int(params.get("eval_days", 3))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
last_recent = ordered.groupby("ticker").tail(eval_d).groupby("ticker")["volume"].mean()
last_baseline = (
ordered.groupby("ticker")
.tail(baseline + eval_d)
.groupby("ticker")
.head(baseline)
.groupby("ticker")["volume"]
.mean()
)
ratio = last_recent / last_baseline.replace(0, pd.NA)
raw = np.log1p(ratio.astype(float))
return percentile_rank(raw).fillna(50.0)

51
stock/app/screener/position_sizer.py Normal file
View File

@@ -0,0 +1,51 @@
"""ATR Wilder smoothing + entry/stop/target 계산."""
import pandas as pd
def compute_atr_wilder(df_one_ticker: pd.DataFrame, window: int = 14) -> float:
"""단일 종목 DataFrame(date·open·high·low·close)에 대해 Wilder ATR 마지막 값."""
g = df_one_ticker.sort_values("date").copy()
high = g["high"].astype(float)
low = g["low"].astype(float)
close = g["close"].astype(float)
prev_close = close.shift(1)
tr = pd.concat([
(high - low),
(high - prev_close).abs(),
(low - prev_close).abs(),
], axis=1).max(axis=1)
atr = tr.ewm(alpha=1 / window, adjust=False).mean()
return float(atr.iloc[-1])
def round_won(x: float) -> int:
return int(round(x))
def plan_positions(ctx, tickers: list, params: dict) -> dict:
"""각 ticker 에 대해 entry/stop/target/atr14 반환."""
atr_window = int(params.get("atr_window", 14))
stop_mult = float(params.get("atr_stop_mult", 2.0))
rr = float(params.get("rr_ratio", 2.0))
prices = ctx.prices.sort_values("date")
out: dict = {}
for t in tickers:
sub = prices[prices["ticker"] == t]
if sub.empty:
continue
close = float(sub["close"].iloc[-1])
atr14 = compute_atr_wilder(sub, window=atr_window)
entry = round_won(close * 1.005)
stop = round_won(close - stop_mult * atr14)
target = round_won(entry + rr * (entry - stop))
r_pct = (entry - stop) / entry * 100 if entry else 0.0
out[t] = {
"entry_price": entry,
"stop_price": stop,
"target_price": target,
"atr14": atr14,
"r_pct": r_pct,
}
return out

26
stock/app/screener/registry.py Normal file
View File

@@ -0,0 +1,26 @@
"""Registry of node classes (single source of truth for /nodes endpoint)."""
from .nodes.hygiene import HygieneGate
from .nodes.foreign_buy import ForeignBuy
from .nodes.volume_surge import VolumeSurge
from .nodes.momentum import Momentum20
from .nodes.high52w import High52WProximity
from .nodes.rs_rating import RsRating
from .nodes.ma_alignment import MaAlignment
from .nodes.vcp_lite import VcpLite
from .nodes.ai_news import AiNewsSentiment
NODE_REGISTRY: dict = {
"foreign_buy": ForeignBuy,
"volume_surge": VolumeSurge,
"momentum": Momentum20,
"high52w": High52WProximity,
"rs_rating": RsRating,
"ma_alignment": MaAlignment,
"vcp_lite": VcpLite,
"ai_news": AiNewsSentiment,
}
GATE_REGISTRY: dict = {
"hygiene": HygieneGate,
}

371
stock/app/screener/router.py Normal file
View File

@@ -0,0 +1,371 @@
"""FastAPI router for /api/stock/screener/*"""
from __future__ import annotations
import datetime as dt
import json
import os
import sqlite3
from typing import Optional
from fastapi import APIRouter, HTTPException
from . import schemas
from .registry import NODE_REGISTRY, GATE_REGISTRY
router = APIRouter(prefix="/api/stock/screener")
import json as _json
import pathlib as _pathlib
_HOLIDAYS_CACHE = None
def _holidays():
global _HOLIDAYS_CACHE
if _HOLIDAYS_CACHE is None:
path = _pathlib.Path(__file__).resolve().parent.parent / "holidays.json"
try:
with path.open(encoding="utf-8") as f:
data = _json.load(f)
_HOLIDAYS_CACHE = set(data) if isinstance(data, list) else set(data.keys())
except FileNotFoundError:
_HOLIDAYS_CACHE = set()
return _HOLIDAYS_CACHE
def _is_holiday(d: dt.date) -> bool:
return d.weekday() >= 5 or d.isoformat() in _holidays()
def _db_path() -> str:
return os.environ.get("STOCK_DB_PATH", "/app/data/stock.db")
def _conn() -> sqlite3.Connection:
# WAL 모드 + busy_timeout으로 동시 read/write lock 회피
# WAL은 reader vs writer 동시성만 해결 — writer 두 명은 직렬이므로 busy_timeout이
# snapshot/refresh의 write 시간보다 길어야 함 (네이버 스크래핑 ~20초 + DB upsert).
conn = sqlite3.connect(_db_path(), timeout=120.0)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=120000")
return conn
# ---------- /nodes ----------
@router.get("/nodes", response_model=schemas.NodesResponse)
def get_nodes():
score_nodes = [
schemas.NodeMeta(
name=cls.name, label=cls.label,
default_params=cls.default_params, param_schema=cls.param_schema,
)
for cls in NODE_REGISTRY.values()
]
gate_nodes = [
schemas.NodeMeta(
name=cls.name, label=cls.label,
default_params=cls.default_params, param_schema=cls.param_schema,
)
for cls in GATE_REGISTRY.values()
]
return schemas.NodesResponse(score_nodes=score_nodes, gate_nodes=gate_nodes)
# ---------- /settings ----------
@router.get("/settings", response_model=schemas.SettingsResponse)
def get_settings():
with _conn() as c:
row = c.execute(
"SELECT weights_json, node_params_json, gate_params_json, "
"top_n, rr_ratio, atr_window, atr_stop_mult, updated_at "
"FROM screener_settings WHERE id=1"
).fetchone()
if row is None:
raise HTTPException(503, "settings not initialized")
return schemas.SettingsResponse(
weights=json.loads(row[0]),
node_params=json.loads(row[1]),
gate_params=json.loads(row[2]),
top_n=row[3], rr_ratio=row[4], atr_window=row[5], atr_stop_mult=row[6],
updated_at=row[7],
)
@router.put("/settings", response_model=schemas.SettingsResponse)
def put_settings(body: schemas.SettingsBody):
now = dt.datetime.utcnow().isoformat()
with _conn() as c:
c.execute(
"""UPDATE screener_settings SET
weights_json=?, node_params_json=?, gate_params_json=?,
top_n=?, rr_ratio=?, atr_window=?, atr_stop_mult=?, updated_at=?
WHERE id=1""",
(
json.dumps(body.weights), json.dumps(body.node_params),
json.dumps(body.gate_params),
body.top_n, body.rr_ratio, body.atr_window, body.atr_stop_mult, now,
),
)
c.commit()
return schemas.SettingsResponse(**body.model_dump(), updated_at=now)
# ---------- /run ----------
from . import telegram as _tg
from .engine import Screener, ScreenContext
def _resolve_asof(asof_str, conn: sqlite3.Connection) -> dt.date:
if asof_str:
return dt.date.fromisoformat(asof_str)
row = conn.execute("SELECT max(date) FROM krx_daily_prices").fetchone()
if not row or row[0] is None:
raise HTTPException(503, "no snapshot available — run /snapshot/refresh first")
return dt.date.fromisoformat(row[0])
def _load_settings(conn) -> dict:
row = conn.execute(
"SELECT weights_json,node_params_json,gate_params_json,top_n,"
"rr_ratio,atr_window,atr_stop_mult FROM screener_settings WHERE id=1"
).fetchone()
return {
"weights": json.loads(row[0]),
"node_params": json.loads(row[1]),
"gate_params": json.loads(row[2]),
"top_n": row[3],
"rr_ratio": row[4],
"atr_window": row[5],
"atr_stop_mult": row[6],
}
def _persist_run(conn, asof, mode, weights, node_params, gate_params, top_n,
result, started_at, finished_at) -> int:
cur = conn.execute(
"""INSERT INTO screener_runs (asof,mode,status,started_at,finished_at,
weights_json,node_params_json,gate_params_json,top_n,survivors_count,telegram_sent)
VALUES (?,?,?,?,?,?,?,?,?,?,0)""",
(asof.isoformat(), mode, "success", started_at, finished_at,
json.dumps(weights), json.dumps(node_params), json.dumps(gate_params),
top_n, result.survivors_count),
)
run_id = cur.lastrowid
for row in result.rows:
conn.execute(
"""INSERT INTO screener_results (run_id,rank,ticker,name,total_score,
scores_json,close,market_cap,entry_price,stop_price,target_price,atr14)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
(run_id, row["rank"], row["ticker"], row["name"], row["total_score"],
json.dumps(row["scores"]), row["close"], row["market_cap"],
row["entry_price"], row["stop_price"], row["target_price"], row["atr14"]),
)
conn.commit()
return run_id
@router.post("/run", response_model=schemas.RunResponse)
def post_run(body: schemas.RunRequest):
from .registry import NODE_REGISTRY as _NR, GATE_REGISTRY as _GR
started_at = dt.datetime.utcnow().isoformat()
with _conn() as c:
asof = _resolve_asof(body.asof, c)
# Skipped holiday handling for mode='auto'
if body.mode == "auto" and _is_holiday(asof):
return schemas.RunResponse(
asof=asof.isoformat(), mode="auto", status="skipped_holiday",
run_id=None, survivors_count=None,
weights={}, top_n=0,
results=[], telegram_payload=None,
warnings=[f"{asof.isoformat()} is a holiday — skipped"],
)
defaults = _load_settings(c)
if body.mode == "auto":
weights = defaults["weights"]
node_params = defaults["node_params"]
gate_params = defaults["gate_params"]
top_n = defaults["top_n"]
else:
weights = body.weights if body.weights is not None else defaults["weights"]
node_params = body.node_params if body.node_params is not None else defaults["node_params"]
gate_params = body.gate_params if body.gate_params is not None else defaults["gate_params"]
top_n = body.top_n if body.top_n is not None else defaults["top_n"]
sizer_params = {
"atr_window": defaults["atr_window"],
"atr_stop_mult": defaults["atr_stop_mult"],
"rr_ratio": defaults["rr_ratio"],
}
ctx = ScreenContext.load(c, asof)
score_nodes = [cls() for name, cls in _NR.items() if weights.get(name, 0) > 0]
gate = _GR["hygiene"]()
try:
screener = Screener(
gate=gate, score_nodes=score_nodes, weights=weights,
node_params=node_params, gate_params=gate_params,
top_n=top_n, sizer_params=sizer_params,
)
result = screener.run(ctx)
except ValueError as e:
raise HTTPException(422, str(e))
finished_at = dt.datetime.utcnow().isoformat()
run_id = None
if body.mode in ("manual_save", "auto"):
run_id = _persist_run(c, asof, body.mode, weights, node_params, gate_params,
top_n, result, started_at, finished_at)
payload = _tg.build_telegram_payload(
asof=asof, mode=body.mode, survivors_count=result.survivors_count,
top_n=top_n, rows=result.rows, run_id=run_id,
)
return schemas.RunResponse(
asof=asof.isoformat(), mode=body.mode, status="success",
run_id=run_id, survivors_count=result.survivors_count,
weights=weights, top_n=top_n,
results=result.rows,
telegram_payload=schemas.TelegramPayload(**payload),
warnings=result.warnings,
)
# ---------- /snapshot/refresh ----------
from . import snapshot as _snap
@router.post("/snapshot/refresh")
def post_snapshot_refresh(asof: Optional[str] = None):
asof_date = dt.date.fromisoformat(asof) if asof else dt.date.today()
if asof_date.weekday() >= 5:
return {"asof": asof_date.isoformat(), "status": "skipped_weekend"}
with _conn() as c:
summary = _snap.refresh_daily(c, asof_date)
return summary
# ---------- /runs ----------
@router.get("/runs", response_model=list[schemas.RunSummary])
def list_runs(limit: int = 30):
with _conn() as c:
rows = c.execute(
"SELECT id,asof,mode,status,started_at,finished_at,top_n,"
"survivors_count,telegram_sent FROM screener_runs "
"ORDER BY asof DESC, id DESC LIMIT ?", (limit,),
).fetchall()
return [
schemas.RunSummary(
id=r[0], asof=r[1], mode=r[2], status=r[3],
started_at=r[4], finished_at=r[5], top_n=r[6],
survivors_count=r[7], telegram_sent=bool(r[8]),
)
for r in rows
]
# ---------- /snapshot/refresh-news-sentiment ----------
from .ai_news import pipeline as _ai_pipeline
from .ai_news import telegram as _ai_telegram
from .ai_news import validation as _ai_validation
@router.post("/snapshot/refresh-news-sentiment")
async def post_refresh_news_sentiment(asof: Optional[str] = None):
asof_date = dt.date.fromisoformat(asof) if asof else dt.date.today()
if asof_date.weekday() >= 5:
return {"asof": asof_date.isoformat(), "status": "skipped_weekend"}
if _is_holiday(asof_date):
return {"asof": asof_date.isoformat(), "status": "skipped_holiday"}
with _conn() as c:
summary = await _ai_pipeline.refresh_daily(c, asof_date)
# top_pos/top_neg 항목에 종목명 주입 (텔레그램 가독성)
tickers = {r["ticker"] for r in summary["top_pos"] + summary["top_neg"]}
if tickers:
placeholders = ",".join("?" * len(tickers))
name_map = {
row[0]: row[1] for row in c.execute(
f"SELECT ticker, name FROM krx_master WHERE ticker IN ({placeholders})",
list(tickers),
).fetchall()
}
for r in summary["top_pos"] + summary["top_neg"]:
r["name"] = name_map.get(r["ticker"], "")
summary["telegram_text"] = _ai_telegram.build_message(
asof=summary["asof"],
top_pos=summary["top_pos"], top_neg=summary["top_neg"],
tokens_input=summary["tokens_input"],
tokens_output=summary["tokens_output"],
mapping=summary.get("mapping"),
)
return summary
# ---------- /ai-news/ic ----------
@router.get("/ai-news/ic")
def get_ai_news_ic(days: int = 30, horizon: int = 1, min_news_count: int = 1):
"""ai_news.score_raw 의 forward return IC (Spearman) 계산.
verdict:
- skip: ic_count < 10 (데이터 부족)
- weak: |ic_mean| <= 0.05
- strong: |ic_mean| > 0.05 (gradient 활성화 가치 있음)
"""
with _conn() as c:
return _ai_validation.compute_ic(
c, days=days, horizon=horizon, min_news_count=min_news_count,
)
@router.get("/runs/{run_id}")
def get_run(run_id: int):
with _conn() as c:
meta = c.execute(
"SELECT id,asof,mode,status,started_at,finished_at,top_n,"
"survivors_count,telegram_sent,weights_json,node_params_json,gate_params_json "
"FROM screener_runs WHERE id=?",
(run_id,),
).fetchone()
if not meta:
raise HTTPException(404, "run not found")
rows = c.execute(
"SELECT rank,ticker,name,total_score,scores_json,close,market_cap,"
"entry_price,stop_price,target_price,atr14 "
"FROM screener_results WHERE run_id=? ORDER BY rank",
(run_id,),
).fetchall()
return {
"meta": {
"id": meta[0], "asof": meta[1], "mode": meta[2], "status": meta[3],
"started_at": meta[4], "finished_at": meta[5], "top_n": meta[6],
"survivors_count": meta[7], "telegram_sent": bool(meta[8]),
"weights": json.loads(meta[9]),
"node_params": json.loads(meta[10]),
"gate_params": json.loads(meta[11]),
},
"results": [
{
"rank": r[0], "ticker": r[1], "name": r[2],
"total_score": r[3], "scores": json.loads(r[4]),
"close": r[5], "market_cap": r[6],
"entry_price": r[7], "stop_price": r[8], "target_price": r[9],
"atr14": r[10],
}
for r in rows
],
}

204
stock/app/screener/schema.py Normal file
View File

@@ -0,0 +1,204 @@
"""Screener schema bootstrap. Called once at module import via db.py."""
import json
import sqlite3
from datetime import datetime, timezone
DEFAULT_WEIGHTS = {
"foreign_buy": 1.0,
"volume_surge": 1.0,
"momentum": 1.0,
"high52w": 1.2,
"rs_rating": 1.2,
"ma_alignment": 1.0,
"vcp_lite": 0.8,
# ai_news: 검증 전 gradient 차단 (4주 IC > 0.05 확인 후 활성화).
# 데이터 수집은 계속, 가중합 영향만 0.
"ai_news": 0.0,
}
DEFAULT_NODE_PARAMS = {
"foreign_buy": {"window_days": 5},
"volume_surge": {"baseline_days": 20, "eval_days": 3},
"momentum": {"window_days": 20},
"high52w": {"window_days": 252},
"rs_rating": {"weights": {"3m": 2, "6m": 1, "9m": 1, "12m": 1}},
"ma_alignment": {"ma_periods": [50, 150, 200]},
"vcp_lite": {"short_window": 40, "long_window": 252},
"ai_news": {"min_news_count": 1},
}
DEFAULT_GATE_PARAMS = {
"min_market_cap_won": 50_000_000_000,
"min_avg_value_won": 500_000_000,
"min_listed_days": 60,
"skip_managed": True,
"skip_preferred": True,
"skip_spac": True,
"skip_halted_days": 3,
}
DDL = """
CREATE TABLE IF NOT EXISTS krx_master (
ticker TEXT PRIMARY KEY,
name TEXT NOT NULL,
market TEXT NOT NULL,
market_cap INTEGER,
is_managed INTEGER NOT NULL DEFAULT 0,
is_preferred INTEGER NOT NULL DEFAULT 0,
is_spac INTEGER NOT NULL DEFAULT 0,
listed_date TEXT,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS krx_daily_prices (
ticker TEXT NOT NULL,
date TEXT NOT NULL,
open INTEGER, high INTEGER, low INTEGER, close INTEGER,
volume INTEGER,
value INTEGER,
PRIMARY KEY (ticker, date)
);
CREATE INDEX IF NOT EXISTS idx_prices_date ON krx_daily_prices(date);
CREATE TABLE IF NOT EXISTS krx_flow (
ticker TEXT NOT NULL,
date TEXT NOT NULL,
foreign_net INTEGER,
institution_net INTEGER,
PRIMARY KEY (ticker, date)
);
CREATE INDEX IF NOT EXISTS idx_flow_date ON krx_flow(date);
CREATE TABLE IF NOT EXISTS screener_settings (
id INTEGER PRIMARY KEY CHECK (id = 1),
weights_json TEXT NOT NULL,
node_params_json TEXT NOT NULL,
gate_params_json TEXT NOT NULL,
top_n INTEGER NOT NULL DEFAULT 20,
rr_ratio REAL NOT NULL DEFAULT 2.0,
atr_window INTEGER NOT NULL DEFAULT 14,
atr_stop_mult REAL NOT NULL DEFAULT 2.0,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS screener_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
asof TEXT NOT NULL,
mode TEXT NOT NULL,
status TEXT NOT NULL,
error TEXT,
started_at TEXT NOT NULL,
finished_at TEXT,
weights_json TEXT NOT NULL,
node_params_json TEXT NOT NULL,
gate_params_json TEXT NOT NULL,
top_n INTEGER NOT NULL,
survivors_count INTEGER,
telegram_sent INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_runs_asof ON screener_runs(asof DESC);
CREATE TABLE IF NOT EXISTS screener_results (
run_id INTEGER NOT NULL,
rank INTEGER NOT NULL,
ticker TEXT NOT NULL,
name TEXT NOT NULL,
total_score REAL NOT NULL,
scores_json TEXT NOT NULL,
close INTEGER,
market_cap INTEGER,
entry_price INTEGER,
stop_price INTEGER,
target_price INTEGER,
atr14 REAL,
PRIMARY KEY (run_id, ticker),
FOREIGN KEY (run_id) REFERENCES screener_runs(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_results_run_rank ON screener_results(run_id, rank);
-- articles 테이블 (도메스틱/해외 뉴스 원본).
-- 메인 app.db.init_db() 에서도 생성하지만, 테스트 환경 및 단독 screener 컨텍스트
-- (ai_news.articles_source 등)에서도 참조 가능하도록 idempotent 하게 보장한다.
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
hash TEXT UNIQUE NOT NULL,
category TEXT DEFAULT 'domestic',
title TEXT NOT NULL,
link TEXT,
summary TEXT,
press TEXT,
pub_date TEXT,
crawled_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_articles_crawled ON articles(crawled_at DESC);
CREATE TABLE IF NOT EXISTS news_sentiment (
ticker TEXT NOT NULL,
date TEXT NOT NULL,
score_raw REAL NOT NULL,
reason TEXT NOT NULL DEFAULT '',
news_count INTEGER NOT NULL DEFAULT 0,
tokens_input INTEGER NOT NULL DEFAULT 0,
tokens_output INTEGER NOT NULL DEFAULT 0,
model TEXT NOT NULL DEFAULT 'claude-haiku-4-5-20251001',
source TEXT NOT NULL DEFAULT 'articles',
created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
PRIMARY KEY (ticker, date)
);
CREATE INDEX IF NOT EXISTS idx_news_sentiment_date ON news_sentiment(date DESC);
"""
def ensure_screener_schema(conn: sqlite3.Connection) -> None:
"""Create tables and seed default settings (idempotent)."""
conn.executescript(DDL)
# ai_news 키 누락 시 1회 보충 (이미 운영 중인 환경에 대해)
row = conn.execute(
"SELECT weights_json, node_params_json FROM screener_settings WHERE id=1"
).fetchone()
if row is not None:
w = json.loads(row[0])
p = json.loads(row[1])
changed = False
if "ai_news" not in w:
w["ai_news"] = DEFAULT_WEIGHTS["ai_news"]
changed = True
# One-time reset: ai_news default 0.8 → 0.0 (검증 전 gradient 차단).
# 사용자가 명시적으로 0.8 외 값을 설정했다면 영향 없음.
elif w.get("ai_news") == 0.8:
w["ai_news"] = 0.0
changed = True
if "ai_news" not in p:
p["ai_news"] = DEFAULT_NODE_PARAMS["ai_news"]
changed = True
if changed:
conn.execute(
"UPDATE screener_settings SET weights_json=?, node_params_json=? WHERE id=1",
(json.dumps(w), json.dumps(p)),
)
# news_sentiment.source 컬럼 1회 추가 (기존 운영 환경)
cols = {r[1] for r in conn.execute(
"PRAGMA table_info(news_sentiment)"
).fetchall()}
if "source" not in cols:
conn.execute(
"ALTER TABLE news_sentiment "
"ADD COLUMN source TEXT NOT NULL DEFAULT 'articles'"
)
existing = conn.execute("SELECT id FROM screener_settings WHERE id=1").fetchone()
if existing is None:
now = datetime.now(timezone.utc).isoformat()
conn.execute(
"""
INSERT INTO screener_settings (
id, weights_json, node_params_json, gate_params_json,
top_n, rr_ratio, atr_window, atr_stop_mult, updated_at
) VALUES (1, ?, ?, ?, 20, 2.0, 14, 2.0, ?)
""",
(
json.dumps(DEFAULT_WEIGHTS),
json.dumps(DEFAULT_NODE_PARAMS),
json.dumps(DEFAULT_GATE_PARAMS),
now,
),
)
conn.commit()

85
stock/app/screener/schemas.py Normal file
View File

@@ -0,0 +1,85 @@
from __future__ import annotations
from typing import Literal, Optional
from pydantic import BaseModel, Field
class NodeMeta(BaseModel):
name: str
label: str
default_params: dict
param_schema: dict
class NodesResponse(BaseModel):
score_nodes: list[NodeMeta]
gate_nodes: list[NodeMeta]
class SettingsBody(BaseModel):
weights: dict[str, float]
node_params: dict[str, dict] = Field(default_factory=dict)
gate_params: dict
top_n: int = 20
rr_ratio: float = 2.0
atr_window: int = 14
atr_stop_mult: float = 2.0
class SettingsResponse(SettingsBody):
updated_at: str
class RunRequest(BaseModel):
mode: Literal["preview", "manual_save", "auto"] = "preview"
asof: Optional[str] = None
weights: Optional[dict[str, float]] = None
node_params: Optional[dict[str, dict]] = None
gate_params: Optional[dict] = None
top_n: Optional[int] = None
class ResultRow(BaseModel):
rank: int
ticker: str
name: str
total_score: float
scores: dict[str, float]
close: int
market_cap: int
entry_price: Optional[int] = None
stop_price: Optional[int] = None
target_price: Optional[int] = None
atr14: Optional[float] = None
r_pct: Optional[float] = None
class TelegramPayload(BaseModel):
chat_target: str
parse_mode: str
text: str
class RunResponse(BaseModel):
asof: str
mode: str
status: Literal["success", "failed", "skipped_holiday"]
run_id: Optional[int] = None
survivors_count: Optional[int] = None
weights: dict[str, float]
top_n: int
results: list[ResultRow] = Field(default_factory=list)
telegram_payload: Optional[TelegramPayload] = None
warnings: list[str] = Field(default_factory=list)
error: Optional[str] = None
class RunSummary(BaseModel):
id: int
asof: str
mode: str
status: str
started_at: str
finished_at: Optional[str] = None
top_n: int
survivors_count: Optional[int] = None
telegram_sent: bool

250
stock/app/screener/snapshot.py Normal file
View File

@@ -0,0 +1,250 @@
"""KRX daily snapshot loader (FDR + naver finance scraping)."""
from __future__ import annotations
import datetime as dt
import logging
import re
import sqlite3
import time
from dataclasses import dataclass
import FinanceDataReader as fdr
import httpx
import pandas as pd
from bs4 import BeautifulSoup
log = logging.getLogger(__name__)
NAVER_FRGN_URL = "https://finance.naver.com/item/frgn.naver"
NAVER_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "https://finance.naver.com/",
}
DEFAULT_FLOW_TOP_N = 100
DEFAULT_RATE_LIMIT_SEC = 0.2
# 시총 상위 100종목 × 0.2초 = ~20초 — agent-office httpx timeout(180s) 안에 여유롭게 완료
# 외국인 매수 시그널은 대형주에서 의미가 크므로 상위 100종목으로 충분.
# 더 많은 종목이 필요하면 별도 cron으로 분리 권장.
@dataclass
class RefreshSummary:
asof: dt.date
master_count: int
prices_count: int
flow_count: int
failures: list[str]
def asdict(self) -> dict:
return {
"asof": self.asof.isoformat(),
"master_count": self.master_count,
"prices_count": self.prices_count,
"flow_count": self.flow_count,
"failures": self.failures,
}
def _iso(d: dt.date) -> str:
return d.isoformat()
def _is_preferred(name: str) -> int:
"""우선주 휴리스틱: 종목명이 '우'로 끝나거나 '우[A-Z]?'/'우\\d?' 패턴."""
n = name or ""
return 1 if re.search(r"우[A-Z]?$|우\d?$", n) else 0
def _is_spac(name: str) -> int:
return 1 if "스팩" in (name or "") else 0
def fetch_master_listing() -> pd.DataFrame:
"""fdr.StockListing('KRX'). Wrapped for stub-ability in tests."""
return fdr.StockListing("KRX")
def fetch_ohlcv_for_ticker(ticker: str, start: str, end: str) -> pd.DataFrame:
"""fdr.DataReader for backfill."""
return fdr.DataReader(ticker, start, end)
def fetch_flow_naver(ticker: str, *, client) -> dict | None:
"""Scrape naver frgn page; return latest-day flow dict, or None."""
r = client.get(NAVER_FRGN_URL, params={"code": ticker, "page": 1})
if r.status_code != 200:
return None
soup = BeautifulSoup(r.text, "lxml")
for row in soup.select("table.type2 tr"):
cells = [c.get_text(strip=True).replace(",", "") for c in row.select("td")]
if not cells or not cells[0]:
continue
if not re.match(r"\d{4}\.\d{2}\.\d{2}", cells[0]):
continue
try:
inst = int(cells[5]) if cells[5] not in ("", "-") else 0
foreign = int(cells[6]) if cells[6] not in ("", "-") else 0
return {
"date": cells[0].replace(".", "-"),
"foreign_net": foreign,
"institution_net": inst,
}
except (IndexError, ValueError):
return None
return None
def _master_and_prices_rows(asof: dt.date,
df: pd.DataFrame) -> tuple[list[tuple], list[tuple]]:
iso = _iso(asof)
now_iso = dt.datetime.utcnow().isoformat()
master_rows: list[tuple] = []
price_rows: list[tuple] = []
for _, row in df.iterrows():
ticker = str(row.get("Code") or "").strip()
name = str(row.get("Name") or "").strip()
if not ticker or not name:
continue
market_raw = str(row.get("Market") or "").upper()
market = "KOSDAQ" if "KOSDAQ" in market_raw else "KOSPI"
try:
market_cap = int(row["Marcap"]) if pd.notna(row.get("Marcap")) else None
except (TypeError, ValueError):
market_cap = None
master_rows.append((
ticker, name, market, market_cap,
0, _is_preferred(name), _is_spac(name),
None, now_iso,
))
try:
o = int(row["Open"]) if pd.notna(row.get("Open")) else None
h = int(row["High"]) if pd.notna(row.get("High")) else None
l = int(row["Low"]) if pd.notna(row.get("Low")) else None
c = int(row["Close"]) if pd.notna(row.get("Close")) else None
v = int(row["Volume"]) if pd.notna(row.get("Volume")) else None
amt = row.get("Amount")
a = int(amt) if pd.notna(amt) else None
if c is not None and v is not None:
price_rows.append((ticker, iso, o, h, l, c, v, a))
except (TypeError, KeyError):
pass
return master_rows, price_rows
def _gather_flow_naver(asof: dt.date, tickers: list[str],
*, rate_limit_sec: float = DEFAULT_RATE_LIMIT_SEC) -> list[tuple]:
iso = _iso(asof)
rows: list[tuple] = []
if not tickers:
return rows
with httpx.Client(timeout=10, headers=NAVER_HEADERS) as client:
for t in tickers:
try:
data = fetch_flow_naver(t, client=client)
if data and data["date"] == iso:
rows.append((t, iso, data["foreign_net"], data["institution_net"]))
except Exception as e:
log.warning("flow scrape failed for %s: %s", t, e)
if rate_limit_sec > 0:
time.sleep(rate_limit_sec)
return rows
def refresh_daily(conn: sqlite3.Connection, asof: dt.date,
flow_top_n: int = DEFAULT_FLOW_TOP_N,
rate_limit_sec: float = DEFAULT_RATE_LIMIT_SEC) -> dict:
"""Pull master + prices (FDR) + flow (naver scraping for top N by market cap)."""
df = fetch_master_listing()
master_rows, price_rows = _master_and_prices_rows(asof, df)
conn.executemany("""
INSERT INTO krx_master (
ticker, name, market, market_cap,
is_managed, is_preferred, is_spac,
listed_date, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(ticker) DO UPDATE SET
name=excluded.name, market=excluded.market,
market_cap=excluded.market_cap,
is_managed=excluded.is_managed,
is_preferred=excluded.is_preferred,
is_spac=excluded.is_spac,
updated_at=excluded.updated_at
""", master_rows)
conn.executemany("""
INSERT OR REPLACE INTO krx_daily_prices
(ticker, date, open, high, low, close, volume, value)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", price_rows)
# 외국인/기관: 시총 상위 N종목만 (rate limit 보호)
if flow_top_n > 0:
top = sorted(master_rows, key=lambda r: r[3] or 0, reverse=True)[:flow_top_n]
flow_tickers = [r[0] for r in top]
else:
flow_tickers = []
flow_rows = _gather_flow_naver(asof, flow_tickers, rate_limit_sec=rate_limit_sec)
conn.executemany("""
INSERT OR REPLACE INTO krx_flow
(ticker, date, foreign_net, institution_net)
VALUES (?, ?, ?, ?)
""", flow_rows)
conn.commit()
return RefreshSummary(
asof=asof, master_count=len(master_rows),
prices_count=len(price_rows), flow_count=len(flow_rows),
failures=[],
).asdict()
def backfill(conn: sqlite3.Connection, start: dt.date, end: dt.date) -> list[dict]:
"""5년치 일봉 백필 — 종목별 fdr.DataReader 호출. Master는 end 기준 (FDR은 historical master 미지원)."""
df = fetch_master_listing()
master_rows, _ = _master_and_prices_rows(end, df)
conn.executemany("""
INSERT INTO krx_master (
ticker, name, market, market_cap,
is_managed, is_preferred, is_spac,
listed_date, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(ticker) DO UPDATE SET name=excluded.name
""", master_rows)
iso_start = start.isoformat()
iso_end = end.isoformat()
results = []
for r in master_rows:
t = r[0]
try:
ddf = fetch_ohlcv_for_ticker(t, iso_start, iso_end)
if ddf is None or ddf.empty:
continue
ddf = ddf.reset_index()
ddf["Date"] = pd.to_datetime(ddf["Date"]).dt.strftime("%Y-%m-%d")
rows = []
for _, rr in ddf.iterrows():
if pd.isna(rr["Close"]) or pd.isna(rr["Volume"]):
continue
rows.append((
t, rr["Date"],
int(rr["Open"]) if pd.notna(rr["Open"]) else None,
int(rr["High"]) if pd.notna(rr["High"]) else None,
int(rr["Low"]) if pd.notna(rr["Low"]) else None,
int(rr["Close"]),
int(rr["Volume"]),
int(rr["Close"] * rr["Volume"]),
))
conn.executemany("""
INSERT OR REPLACE INTO krx_daily_prices
(ticker, date, open, high, low, close, volume, value)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", rows)
results.append({"ticker": t, "count": len(rows)})
except Exception as e:
log.error("backfill failed for %s: %s", t, e)
results.append({"ticker": t, "error": str(e)})
conn.commit()
return results

82
stock/app/screener/telegram.py Normal file
View File

@@ -0,0 +1,82 @@
"""Telegram payload builder. Caller (agent-office) handles actual delivery."""
from __future__ import annotations
import datetime as dt
# 노드별 풀 라벨 (아이콘 대신 사용 — 사용자가 명확한 이름 선호)
NODE_LABELS = {
"foreign_buy": "외국인",
"volume_surge": "거래량급증",
"momentum": "20일모멘텀",
"high52w": "52주신고가",
"rs_rating": "RS레이팅",
"ma_alignment": "이평선정배열",
"vcp_lite": "VCP수축",
}
PAGE_BASE = "https://gahusb.synology.me/stock/screener"
def _escape_md(s: str) -> str:
"""Minimal MarkdownV2 escape — extend if formatting breaks."""
for ch in r"\_*[]()~`>#+-=|{}.!":
s = s.replace(ch, "\\" + ch)
return s
def _format_won(n) -> str:
"""1,234,567원 형태 (None 시 '-')."""
if n is None:
return "\\-"
return f"{int(n):,}원"
def _format_active_nodes(scores: dict, threshold: int = 70) -> str:
"""70점 이상 노드를 '라벨 점수' 형태로 나열, 콤마 구분."""
active = []
for name, sc in scores.items():
label = NODE_LABELS.get(name)
if label is None or sc < threshold:
continue
active.append(f"{_escape_md(label)} {int(sc)}")
return " · ".join(active) if active else "\\(70점 이상 노드 없음\\)"
def build_telegram_payload(asof: dt.date, mode: str, survivors_count: int,
top_n: int, rows: list, run_id) -> dict:
title = "*KRX 강세주 스크리너*"
header = (
f"🎯 {title} — {_escape_md(asof.isoformat())} \\({_escape_md(mode)}\\)\n"
f"통과 {survivors_count}종 / Top {top_n} / 본문 1\\-10"
)
lines = []
for r in rows[:10]:
nodes_str = _format_active_nodes(r.get("scores", {}))
score_str = f"{r['total_score']:.1f}"
r_pct = r.get("r_pct")
r_pct_str = f"{r_pct:.1f}" if r_pct is not None else "-"
lines.append(
f"{r['rank']}\\. *{_escape_md(r['name'])}* `{r['ticker']}` "
f"⭐ {_escape_md(score_str)}\n"
f" {nodes_str}\n"
f" 진입 {_format_won(r.get('entry_price'))} "
f"손절 {_format_won(r.get('stop_price'))} "
f"익절 {_format_won(r.get('target_price'))} "
f"\\(R {_escape_md(r_pct_str)}%\\)"
)
# URL은 inline link로 감싸 URL 내부 . - ? = 이스케이프 회피
link = (
f"🔗 [전체 결과·11\\~20위]({PAGE_BASE}?run_id={run_id})"
if run_id else ""
)
text = header + "\n\n" + "\n\n".join(lines) + ("\n\n" + link if link else "")
return {
"chat_target": "default",
"parse_mode": "MarkdownV2",
"text": text,
}