Stock-trading-programming/app/ai/morning.py

"""
app/ai/morning.py
장 전 데이터 수집 스크립트 (claude_morning 헬퍼)

Claude Code headless가 이 스크립트를 실행해 데이터를 수집한 뒤,
수집 결과를 직접 분석해 daily_context.json을 작성한다.

실행:
    python app/ai/morning.py          # 수집 + 파일 저장
    python app/ai/morning.py --print  # 수집 결과를 stdout으로 출력 (Claude 분석용)
"""

import asyncio
import json
import logging
import os
import re
import sys
from datetime import datetime
from pathlib import Path


def _load_env():
    env_path = Path(".env")
    if not env_path.exists():
        return
    with open(env_path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, _, v = line.partition("=")
            k = k.strip()
            v = v.strip()
            if " #" in v:
                v = v[: v.index(" #")]
            v = v.strip().strip('"').strip("'")
            if k and v and k not in os.environ:
                os.environ[k] = v


_load_env()

ROOT = Path(__file__).parent.parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

os.makedirs("logs", exist_ok=True)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.StreamHandler(sys.stderr),  # stdout은 Claude 분석용으로 비워둠
        logging.FileHandler("logs/stockbot.log", encoding="utf-8"),
    ],
)
logger = logging.getLogger(__name__)

import aiohttp
from bs4 import BeautifulSoup
from app.execution.kis_client import KISClient

TODAY = datetime.now().strftime("%Y-%m-%d")
NEWS_PATH = f"data/news/{TODAY}.json"
MARKET_PATH = f"data/market/{TODAY}.json"

_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept-Language": "ko-KR,ko;q=0.9",
    "Referer": "https://finance.naver.com/",
}


# ── 뉴스 크롤링 ─────────────────────────────────────────────────────────────

async def fetch_news() -> list[str]:
    url = "https://finance.naver.com/news/mainnews.naver"
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(
                url, headers=_HEADERS, timeout=aiohttp.ClientTimeout(total=15)
            ) as resp:
                raw = await resp.read()

        try:
            html = raw.decode("euc-kr")
        except Exception:
            html = raw.decode("utf-8", errors="replace")

        soup = BeautifulSoup(html, "html.parser")
        headlines: list[str] = []

        for tag in soup.select(
            "ul.simpleNewsList li a, "
            ".newsListBody .articleSubject a, "
            ".mainNewsList .articleSubject a"
        ):
            text = tag.get_text(strip=True)
            if len(text) >= 10:
                headlines.append(text)

        if not headlines:
            for tag in soup.find_all("a", href=re.compile(r"news")):
                text = tag.get_text(strip=True)
                if 10 <= len(text) <= 120:
                    headlines.append(text)

        seen: set[str] = set()
        result: list[str] = []
        for h in headlines:
            if h not in seen:
                seen.add(h)
                result.append(h)
            if len(result) >= 15:
                break

        logger.info(f"뉴스 {len(result)}건 수집")
        return result
    except Exception as e:
        logger.warning(f"뉴스 크롤링 실패: {e}")
        return []


# ── KIS 시장 데이터 ──────────────────────────────────────────────────────────

async def fetch_market_data(kis: KISClient) -> dict:
    data: dict = {
        "volume_rank": [],
        "foreign_buy": [],
        "institution_buy": [],
        "sectors": [],
    }

    try:
        data["volume_rank"] = await kis.get_volume_rank(top_n=20)
        logger.info(f"거래량 순위 {len(data['volume_rank'])}종목")
        await asyncio.sleep(1.1)
    except Exception as e:
        logger.warning(f"거래량 순위 실패: {e}")

    try:
        fi = await kis.get_foreign_institution_rank(top_n=20)
        data["foreign_buy"] = fi["foreign"][:10]
        data["institution_buy"] = fi["institution"][:10]
        logger.info("외국인/기관 수급 수집")
        await asyncio.sleep(1.1)
    except Exception as e:
        logger.warning(f"수급 데이터 실패: {e}")

    try:
        data["sectors"] = await kis.get_sector_trend()
        logger.info(f"업종 동향 {len(data['sectors'])}개")
        await asyncio.sleep(1.1)
    except Exception as e:
        logger.warning(f"업종 동향 실패: {e}")

    return data


# ── 메인 ─────────────────────────────────────────────────────────────────────

async def main(print_mode: bool = False):
    logger.info(f"데이터 수집 시작 [{TODAY}]")

    for d in ["data/news", "data/market"]:
        os.makedirs(d, exist_ok=True)

    # 뉴스 수집
    news = await fetch_news()
    Path(NEWS_PATH).write_text(
        json.dumps({"date": TODAY, "headlines": news}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    # KIS 시장 데이터
    kis = KISClient()
    market: dict = {"volume_rank": [], "foreign_buy": [], "institution_buy": [], "sectors": []}
    try:
        await kis.get_access_token()
        market = await fetch_market_data(kis)
    except Exception as e:
        logger.warning(f"KIS 수집 실패: {e}")

    Path(MARKET_PATH).write_text(
        json.dumps({"date": TODAY, **market}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    logger.info(f"수집 완료 → {NEWS_PATH}, {MARKET_PATH}")

    if print_mode:
        # Claude Code가 읽을 수 있도록 stdout으로 출력
        print(json.dumps(
            {
                "date": TODAY,
                "news_headlines": news,
                "volume_rank": market["volume_rank"][:10],
                "foreign_buy_top5": market["foreign_buy"][:5],
                "institution_buy_top5": market["institution_buy"][:5],
                "sectors": market["sectors"][:15],
            },
            ensure_ascii=False,
            indent=2,
        ))


if __name__ == "__main__":
    print_mode = "--print" in sys.argv
    asyncio.run(main(print_mode=print_mode))