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

"""
app/ai/midday.py
장중 데이터 수집 스크립트 (claude_midday 헬퍼)

Claude Code headless가 이 스크립트를 실행해 장중 스냅샷을 수집한 뒤,
오전 daily_context와 비교 분석해 midday_context.json을 작성한다.

실행:
    python app/ai/midday.py --print
"""

import asyncio
import json
import logging
import os
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),
        logging.FileHandler("logs/stockbot.log", encoding="utf-8"),
    ],
)
logger = logging.getLogger(__name__)

from app.execution.kis_client import KISClient
from app.db.models import get_conn

TODAY = datetime.now().strftime("%Y-%m-%d")
DAILY_CONTEXT_PATH  = Path("data/daily_context.json")
MIDDAY_CONTEXT_PATH = Path("data/midday_context.json")


# ── DB 조회 ────────────────────────────────────────────────────────────────────

def get_today_trades() -> list:
    """오늘 체결된 거래 내역 (오전 결과)"""
    try:
        with get_conn() as conn:
            rows = conn.execute(
                """SELECT ticker, name, entry_time, exit_time,
                          entry_price, exit_price, quantity,
                          exit_reason, pnl
                   FROM trades
                   WHERE date=? AND exit_time IS NOT NULL
                   ORDER BY exit_time""",
                (TODAY,),
            ).fetchall()
        return [
            {
                "ticker"     : r[0],
                "name"       : r[1],
                "entry_time" : r[2],
                "exit_time"  : r[3],
                "entry_price": r[4],
                "exit_price" : r[5],
                "qty"        : r[6],
                "reason"     : r[7],
                "pnl"        : r[8],
            }
            for r in rows
        ]
    except Exception as e:
        logger.warning(f"거래 내역 조회 실패: {e}")
        return []


def get_current_positions() -> list:
    """현재 보유 포지션"""
    try:
        with get_conn() as conn:
            rows = conn.execute(
                """SELECT ticker, name, entry_time, entry_price, quantity,
                          target_price, stop_price
                   FROM positions""",
            ).fetchall()
        return [
            {
                "ticker"      : r[0],
                "name"        : r[1],
                "entry_time"  : r[2],
                "entry_price" : r[3],
                "qty"         : r[4],
                "target_price": r[5],
                "stop_price"  : r[6],
            }
            for r in rows
        ]
    except Exception as e:
        logger.warning(f"포지션 조회 실패: {e}")
        return []


def get_morning_context() -> dict:
    """아침 daily_context.json 로드"""
    if not DAILY_CONTEXT_PATH.exists():
        return {}
    try:
        return json.loads(DAILY_CONTEXT_PATH.read_text(encoding="utf-8"))
    except Exception:
        return {}


# ── KIS 시장 스냅샷 ────────────────────────────────────────────────────────────

async def fetch_market_snapshot(kis: KISClient) -> dict:
    """거래량 순위 + 업종 동향 현재 시점 스냅샷"""
    data: dict = {"volume_rank": [], "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:
        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


# ── 통계 계산 ──────────────────────────────────────────────────────────────────

def _calc_session_stats(trades: list) -> dict:
    closed = [t for t in trades if t["pnl"] is not None]
    wins   = [t for t in closed if t["pnl"] > 0]
    losses = [t for t in closed if t["pnl"] <= 0]
    net    = sum(t["pnl"] for t in closed)

    # 현재 연속 손절 수 계산 (역순으로 순회)
    consec = 0
    for t in reversed(closed):
        if t["pnl"] < 0:
            consec += 1
        else:
            break

    return {
        "total"       : len(closed),
        "wins"        : len(wins),
        "losses"      : len(losses),
        "net_pnl"     : round(net, 0),
        "win_rate_pct": round(len(wins) / len(closed) * 100, 1) if closed else 0,
        "consec_loss" : consec,
    }


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

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

    # 실거래 API로 KISClient 생성
    _orig = os.environ.get("KIS_MOCK", "true")
    os.environ["KIS_MOCK"] = "false"
    kis = KISClient()
    os.environ["KIS_MOCK"] = _orig

    market: dict = {"volume_rank": [], "sectors": []}
    try:
        await kis.get_access_token()
        market = await fetch_market_snapshot(kis)
    except Exception as e:
        logger.warning(f"KIS 장중 수집 실패: {e}")

    trades    = get_today_trades()
    positions = get_current_positions()
    morning   = get_morning_context()
    stats     = _calc_session_stats(trades)

    logger.info(f"오전 결과: {stats['total']}건 / 승{stats['wins']} 패{stats['losses']} "
                f"/ 연속손절 {stats['consec_loss']}회")

    if print_mode:
        print(json.dumps(
            {
                "date"            : TODAY,
                "generated_at"    : datetime.now().strftime("%H:%M:%S"),
                "morning_context" : morning,
                "session_stats"   : stats,
                "trades"          : trades,
                "current_positions": positions,
                "volume_rank"     : market["volume_rank"],
                "sectors"         : market["sectors"],
            },
            ensure_ascii=False,
            indent=2,
        ))


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