trading/kis-api-main.py

"""
주가 수집 + 매매 시그널 서비스 (네이버 금융 기반)
- API 키 불필요
- 시총 상위 200개 종목 현재가
- 매수/매도 목표가 + 손절가 자동 계산
"""
import asyncio,json,os,re
from datetime import datetime,timedelta
from typing import Optional
import asyncpg,httpx,redis.asyncio as aioredis,structlog
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI,Query
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware

structlog.configure(processors=[structlog.processors.TimeStamper(fmt="iso"),structlog.processors.add_log_level,structlog.processors.JSONRenderer()])
logger=structlog.get_logger()
REDIS_HOST=os.getenv("REDIS_HOST","redis")
REDIS_PASSWORD=os.getenv("REDIS_PASSWORD","")
PG_HOST=os.getenv("POSTGRES_HOST","postgres")
PG_PORT=int(os.getenv("POSTGRES_PORT","5432"))
PG_DB=os.getenv("POSTGRES_DB","trading_ai")
PG_USER=os.getenv("POSTGRES_USER","kyu")
PG_PASS=os.getenv("POSTGRES_PASSWORD","7895123")
HEADERS={"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}

pg_pool:Optional[asyncpg.Pool]=None
redis_cl:Optional[aioredis.Redis]=None
scheduler=AsyncIOScheduler(timezone="Asia/Seoul")

class Stats:
    collected=0;last_run="";errors=0
stats=Stats()

async def init_db():
    async with pg_pool.acquire() as c:
        await c.execute("""CREATE TABLE IF NOT EXISTS stock_prices(
            id SERIAL PRIMARY KEY,stock_code VARCHAR(10) NOT NULL,stock_name VARCHAR(100) DEFAULT '',
            price INTEGER DEFAULT 0,change_pct FLOAT DEFAULT 0,change_amount INTEGER DEFAULT 0,
            volume BIGINT DEFAULT 0,high INTEGER DEFAULT 0,low INTEGER DEFAULT 0,open_price INTEGER DEFAULT 0,
            market_cap BIGINT DEFAULT 0,per FLOAT DEFAULT 0,pbr FLOAT DEFAULT 0,
            high_52w INTEGER DEFAULT 0,low_52w INTEGER DEFAULT 0,collected_at TIMESTAMP DEFAULT NOW())""")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_sp_code ON stock_prices(stock_code)")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_sp_time ON stock_prices(collected_at DESC)")
        await c.execute("""CREATE TABLE IF NOT EXISTS trade_signals(
            id SERIAL PRIMARY KEY,stock_code VARCHAR(10) NOT NULL,stock_name VARCHAR(100) DEFAULT '',
            signal_type VARCHAR(10) NOT NULL,current_price INTEGER DEFAULT 0,target_price INTEGER DEFAULT 0,
            stop_loss INTEGER DEFAULT 0,expected_return_pct FLOAT DEFAULT 0,risk_reward_ratio FLOAT DEFAULT 0,
            confidence FLOAT DEFAULT 0,reason TEXT DEFAULT '',news_score FLOAT DEFAULT 0,
            dart_score FLOAT DEFAULT 0,price_momentum FLOAT DEFAULT 0,created_at TIMESTAMP DEFAULT NOW())""")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_ts_code ON trade_signals(stock_code)")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_ts_time ON trade_signals(created_at DESC)")
    logger.info("db.ok")

async def naver_price(client,code):
    try:
        r=await client.get(f"https://finance.naver.com/item/main.naver?code={code}",headers=HEADERS,timeout=15)
        r.encoding="euc-kr";h=r.text
        def ex(p,d="0"):
            m=re.search(p,h);return m.group(1).replace(",","").strip() if m else d
        name=ex(r'class="wrap_company".*?<h2><a[^>]*>([^<]+)</a>',"")
        price=int(ex(r'<p class="no_today">.*?<em.*?<span class="blind">([0-9,]+)</span>'))
        cpct=ex(r'class="blind">([0-9.\-+]+)%',"0")
        vol=ex(r'<td class="first">.*?<span class="blind">([0-9,]+)</span>')
        hi=ex(r'최고.*?<em.*?<span class="blind">([0-9,]+)</span>')
        lo=ex(r'최저.*?<em.*?<span class="blind">([0-9,]+)</span>')
        cap=ex(r'시가총액.*?<em>([0-9,]+)</em>',"0")
        per=ex(r'PER.*?<em>([0-9.]+)</em>',"0")
        pbr=ex(r'PBR.*?<em>([0-9.]+)</em>',"0")
        h52=ex(r'52주.*?최고.*?([0-9,]+)',"0")
        l52=ex(r'52주.*?최저.*?([0-9,]+)',"0")
        return {"code":code,"name":name,"price":price,"change_pct":float(cpct),
            "volume":int(vol),"high":int(hi),"low":int(lo),"market_cap":int(cap),
            "per":float(per),"pbr":float(pbr),"high_52w":int(h52),"low_52w":int(l52),
            "timestamp":datetime.now().isoformat()}
    except Exception as e:
        return None

async def get_stock_codes(client,count=200):
    if pg_pool:
        try:
            rows=await pg_pool.fetch("SELECT stock_code FROM dart_corps LIMIT $1",count)
            codes=[r["stock_code"] for r in rows if r["stock_code"]];
            if len(codes)>50:return codes[:count]
        except:pass
    codes=[]
    for sosok in [0,1]:
        for page in range(1,20):
            try:
                r=await client.get(f"https://finance.naver.com/sise/sise_market_sum.naver?sosok={sosok}&page={page}",headers=HEADERS,timeout=15)
                r.encoding="euc-kr"
                found=re.findall(r'main\.naver\?code=(\d{6})',r.text)
                if not found:break
                codes.extend(found);await asyncio.sleep(0.2)
            except:break
            if len(codes)>=count:break
    return list(dict.fromkeys(codes))[:count]

def calc_signal(p,news_sc,dart_sc):
    price=p["price"]
    if price<=0:return None
    h52=p.get("high_52w",0);l52=p.get("low_52w",0);cpct=p.get("change_pct",0)
    total=news_sc*0.4+dart_sc*0.3+(cpct*10)*0.3
    total=max(-100,min(100,total))
    pos=(price-l52)/(h52-l52) if h52>l52>0 else 0.5

    if total>=30:
        sig="매수"
        tp=int(price+(h52-price)*(0.3+min(total,100)/200)) if h52>price else int(price*1.1)
        sl=int(price*(0.92+pos*0.03))
        conf=min(95,50+total*0.3+(1-pos)*20)
    elif total<=-30:
        sig="매도"
        tp=int(price-(price-l52)*(0.3+min(abs(total),100)/200)) if l52>0 and l52<price else int(price*0.9)
        sl=int(price*1.05)
        conf=min(95,50+abs(total)*0.3+pos*20)
    else:return None

    er=((tp-price)/price)*100 if sig=="매수" else ((price-tp)/price)*100
    risk=abs(sl-price);reward=abs(tp-price)
    rr=reward/risk if risk>0 else 0
    return {"signal_type":sig,"current_price":price,"target_price":tp,"stop_loss":sl,
        "expected_return_pct":round(er,2),"risk_reward_ratio":round(rr,2),
        "confidence":round(conf,1),"price_momentum":round(cpct,2),
        "news_score":round(news_sc,1),"dart_score":round(dart_sc,1)}

async def collect_prices():
    logger.info("prices.start")
    async with httpx.AsyncClient() as client:
        codes=await get_stock_codes(client,200)
        ok=0
        for code in codes:
            p=await naver_price(client,code)
            if not p or p["price"]<=0:continue
            if redis_cl:await redis_cl.set(f"price:{code}",json.dumps(p,ensure_ascii=False),ex=600)
            if pg_pool:
                try:
                    await pg_pool.execute("""INSERT INTO stock_prices(stock_code,stock_name,price,change_pct,
                        volume,high,low,open_price,market_cap,per,pbr,high_52w,low_52w,collected_at)
                        VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)""",
                        code,p["name"],p["price"],p["change_pct"],p["volume"],p["high"],p["low"],
                        0,p["market_cap"],p["per"],p["pbr"],p["high_52w"],p["low_52w"],datetime.now())
                except:pass
            ok+=1;await asyncio.sleep(0.3)
    await gen_signals()
    stats.collected+=ok;stats.last_run=datetime.now().isoformat()
    if redis_cl:await redis_cl.set("prices:last_update",datetime.now().isoformat())
    logger.info("prices.done",count=ok)

async def gen_signals():
    if not pg_pool:return
    async with pg_pool.acquire() as c:
        scores=await c.fetch("""SELECT stock_code,stock_name,news_score,dart_score,total_score
            FROM stock_scores WHERE score_date=(SELECT MAX(score_date) FROM stock_scores)
            AND (total_score>=30 OR total_score<=-30)""")
        n=0
        for row in scores:
            code=row["stock_code"]
            if not code or len(code)!=6:continue
            if redis_cl:
                cached=await redis_cl.get(f"price:{code}")
                if not cached:continue
                pd=json.loads(cached)
            else:continue
            sig=calc_signal(pd,row["news_score"],row["dart_score"])
            if not sig:continue
            reasons=await c.fetch("SELECT reason FROM news_analysis WHERE primary_stock=$1 AND intensity>=2 ORDER BY analyzed_at DESC LIMIT 2",code)
            reason=" | ".join([r["reason"][:100] for r in reasons])
            await c.execute("""INSERT INTO trade_signals(stock_code,stock_name,signal_type,current_price,
                target_price,stop_loss,expected_return_pct,risk_reward_ratio,confidence,reason,
                news_score,dart_score,price_momentum) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)""",
                code,row["stock_name"],sig["signal_type"],sig["current_price"],sig["target_price"],
                sig["stop_loss"],sig["expected_return_pct"],sig["risk_reward_ratio"],sig["confidence"],
                reason,sig["news_score"],sig["dart_score"],sig["price_momentum"])
            n+=1
    logger.info("signals.done",count=n)

app=FastAPI(title="주가+시그널")
app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_methods=["*"],allow_headers=["*"])

@app.on_event("startup")
async def startup():
    global pg_pool,redis_cl
    pg_pool=await asyncpg.create_pool(host=PG_HOST,port=PG_PORT,database=PG_DB,user=PG_USER,password=PG_PASS,min_size=2,max_size=5)
    redis_cl=aioredis.Redis(host=REDIS_HOST,port=6379,password=REDIS_PASSWORD,db=3,decode_responses=True)
    await init_db()
    scheduler.add_job(collect_prices,"cron",day_of_week="mon-fri",hour="9-16",minute="*/5",id="prices",replace_existing=True)
    scheduler.start();logger.info("started")

@app.on_event("shutdown")
async def shutdown():
    scheduler.shutdown()
    if pg_pool:await pg_pool.close()
    if redis_cl:await redis_cl.aclose()

@app.get("/health")
async def health():
    lu=await redis_cl.get("prices:last_update") if redis_cl else ""
    return JSONResponse(content={"status":"ok","collected":stats.collected,"last_run":stats.last_run,"last_update":lu or ""})

@app.get("/price/{code}")
async def price(code:str):
    if redis_cl:
        c=await redis_cl.get(f"price:{code}")
        if c:return JSONResponse(content=json.loads(c))
    async with httpx.AsyncClient() as cl:
        p=await naver_price(cl,code)
        if p:return JSONResponse(content=p)
    return JSONResponse(content={"error":"not found"},status_code=404)

@app.get("/prices")
async def prices(limit:int=Query(default=50)):
    if not redis_cl:return JSONResponse(content={"data":{}})
    keys=await redis_cl.keys("price:*")
    r={};
    for k in keys[:limit]:
        c=await redis_cl.get(k)
        if c:d=json.loads(c);r[d["code"]]=d
    return JSONResponse(content={"count":len(r),"data":r})

@app.get("/signals")
async def signals(days:int=Query(default=7)):
    async with pg_pool.acquire() as c:
        rows=await c.fetch("SELECT * FROM trade_signals WHERE created_at>=NOW()-INTERVAL '%s days' ORDER BY confidence DESC LIMIT 30"%days)
    return [dict(r) for r in rows]

@app.get("/signals/{code}")
async def stock_signals(code:str):
    async with pg_pool.acquire() as c:
        rows=await c.fetch("SELECT * FROM trade_signals WHERE stock_code=$1 ORDER BY created_at DESC LIMIT 10",code)
    return [dict(r) for r in rows]

@app.post("/collect")
async def manual():
    asyncio.create_task(collect_prices());return {"status":"started"}