실전 프로젝트

# analysis/report.py — Claude 수급 리포트

import anthropic, json
from dataclasses import dataclass

@dataclass
class SupplyReport:
    code: str
    summary: str
    signal: str   # "매수" | "중립" | "매도"
    confidence: float

def generate_report(code, tick, investor, market, client) -> SupplyReport:
    prompt = f"""
종목코드: {code}
[틱 신호]
- 체결강도: {tick['buy_pressure']:.2f}
- VWAP 대비: {tick['vwap_dev']:+.2f}%
- 대량체결: {'있음' if tick['block_trade'] else '없음'}
[5일 외인·기관 추세]
- 외인 누적: {investor['foreign_net_cum']:,}주
- 기관 누적: {investor['inst_net_cum']:,}주
[시장 전체]
- 코스피 외인 순매수: {market['kospi_foreign_net']:,}억

JSON으로 답하세요: {{"signal":"매수|중립|매도","confidence":0.0~1.0,"summary":"3줄 이내"}}
"""
    resp = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=256,
        messages=[{"role": "user", "content": prompt}],
    )
    r = json.loads(resp.content[0].text)
    return SupplyReport(code=code, **r)

from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

def build_chatbot(docs_dir: str, db_dir: str):
    # 인덱싱
    chunks = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=80).split_documents(
        DirectoryLoader(docs_dir, glob="**/*.{md,txt}").load()
    )
    embedder = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")
    db = Chroma.from_documents(chunks, embedder, persist_directory=db_dir)

    # 체인
    prompt = ChatPromptTemplate.from_template(
        "문서:\n{context}\n\n질문: {question}\n\n문서에 없으면 '없습니다'라고 답하세요."
    )
    llm = ChatAnthropic(model="claude-sonnet-4-6", max_tokens=1024)
    return (
        {"context": db.as_retriever(search_kwargs={"k": 5})
                    | (lambda docs: "\n\n".join(d.page_content for d in docs)),
         "question": RunnablePassthrough()}
        | prompt | llm | StrOutputParser()
    )

if __name__ == "__main__":
    bot = build_chatbot("./docs", "./vector_db")
    while (q := input("질문: ").strip()) not in ("exit", "quit"):
        print(f"답변: {bot.invoke(q)}\n")

import schedule, time, httpx, sqlite3
from datetime import date
from pathlib import Path

DB = Path("data/supply.db")

def init_db():
    with sqlite3.connect(DB) as c:
        c.execute("""
            CREATE TABLE IF NOT EXISTS supply_daily (
                date TEXT, code TEXT, name TEXT,
                foreign_net INTEGER, buy_pressure REAL,
                PRIMARY KEY (date, code)
            )
        """)

def collect(session):
    today = date.today().isoformat()
    with sqlite3.connect(DB) as c:
        for s in session.get_top200():
            t = session.get_investor_trend(s.code, days=1)
            c.execute("INSERT OR REPLACE INTO supply_daily VALUES (?,?,?,?,?)",
                (today, s.code, s.name, t.foreign_net_total,
                 s.foreign_buy_vol / max(s.foreign_buy_vol + s.foreign_sell_vol, 1)))

def alert(webhook: str):
    with sqlite3.connect(DB) as c:
        rows = c.execute(
            "SELECT name, foreign_net FROM supply_daily WHERE date=? ORDER BY foreign_net DESC LIMIT 5",
            (date.today().isoformat(),)
        ).fetchall()
    msg = "*외인 순매수 상위*\n" + "\n".join(f"• {n}: {v:+,}주" for n, v in rows)
    httpx.post(webhook, json={"text": msg})

schedule.every().monday.at("16:30").do(lambda: (collect(session), alert(SLACK_URL)))

if __name__ == "__main__":
    init_db()
    while True:
        schedule.run_pending()
        time.sleep(60)