kyu
a56227ace9
장 시작 09:00 토큰 만료 시 여러 잡이 동시 재발급 → 키움 앱키당 토큰 1개 정책으로 무효 토큰을 붙들어 ka10001 전건 실패(ok:0) → 시세 stale(텔레그램 현재가 불일치). KiwoomToken.get()에 락+재확인 추가, _refresh 분리. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1247 lines
54 KiB
Python
1247 lines
54 KiB
Python
"""
|
|
키움증권 REST API 기반 주가·수급·공매도 수집 서비스
|
|
- ka10001: 현재가 + 재무지표 (PER/PBR/ROE/EPS/BPS/외국인비중/시가총액)
|
|
- ka10005: 일봉 OHLCV + 외국인·기관 순매수
|
|
- ka10008: 외국인 종목별 매매동향 (일자별 보유비중 변화)
|
|
- ka10014: 공매도 추이 (잔고수량·거래비중)
|
|
- Redis db=3 price:{code} 형식 유지 (score-engine·ta-engine 호환)
|
|
"""
|
|
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", "")
|
|
KIWOOM_APP_KEY = os.getenv("KIWOOM_APP_KEY", "")
|
|
KIWOOM_SECRET_KEY = os.getenv("KIWOOM_SECRET_KEY", "")
|
|
KIWOOM_BASE_URL = os.getenv("KIWOOM_BASE_URL", "https://api.kiwoom.com")
|
|
TG_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "")
|
|
TG_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "")
|
|
|
|
pg_pool: Optional[asyncpg.Pool] = None
|
|
redis_cl: Optional[aioredis.Redis] = None
|
|
scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
|
|
|
|
# ── 키움 토큰 관리 ────────────────────────────────────────
|
|
|
|
class KiwoomToken:
|
|
token: str = ""
|
|
expires_at: datetime = datetime.min
|
|
_lock = asyncio.Lock()
|
|
|
|
async def get(self, client: httpx.AsyncClient) -> str:
|
|
if self.token and datetime.now() < self.expires_at:
|
|
return self.token
|
|
# 동시 갱신 방지: 락 안에서 재확인(double-check) — 키움은 앱키당 토큰 1개만
|
|
# 유효(재발급 시 이전 토큰 무효화)하므로 thundering-herd 시 무효 토큰을 붙들게 됨
|
|
async with self._lock:
|
|
if self.token and datetime.now() < self.expires_at:
|
|
return self.token
|
|
return await self._refresh(client)
|
|
|
|
async def _refresh(self, client: httpx.AsyncClient) -> str:
|
|
resp = await client.post(
|
|
f"{KIWOOM_BASE_URL}/oauth2/token",
|
|
json={"grant_type": "client_credentials",
|
|
"appkey": KIWOOM_APP_KEY,
|
|
"secretkey": KIWOOM_SECRET_KEY},
|
|
headers={"Content-Type": "application/json;charset=UTF-8"},
|
|
timeout=10)
|
|
data = resp.json()
|
|
if data.get("return_code", -1) != 0:
|
|
raise RuntimeError(f"토큰 발급 실패: {data.get('return_msg')}")
|
|
self.token = data["token"]
|
|
# expires_dt: "20260508151325" 형식
|
|
exp_str = data.get("expires_dt", "")
|
|
try:
|
|
self.expires_at = datetime.strptime(exp_str, "%Y%m%d%H%M%S") - timedelta(minutes=5)
|
|
except Exception:
|
|
self.expires_at = datetime.now() + timedelta(hours=23)
|
|
logger.info("kiwoom.token.refreshed", expires=exp_str)
|
|
return self.token
|
|
|
|
kiwoom_token = KiwoomToken()
|
|
|
|
|
|
class _RateLimiter:
|
|
"""Kiwoom 글로벌 레이트리밋 — 초당 max_rate건으로 직렬 페이싱 (429 방지)."""
|
|
def __init__(self, max_rate: float):
|
|
self.min_gap = 1.0 / max_rate
|
|
self._last = 0.0
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def acquire(self):
|
|
async with self._lock:
|
|
loop = asyncio.get_running_loop()
|
|
wait = self.min_gap - (loop.time() - self._last)
|
|
if wait > 0:
|
|
await asyncio.sleep(wait)
|
|
self._last = loop.time()
|
|
|
|
# 실측: 페이싱 시 10/s도 정상. 보수적으로 6/s (≈2,300종목 6분 주기).
|
|
kiwoom_rate = _RateLimiter(6.0)
|
|
|
|
# ── 키움 API 호출 헬퍼 ────────────────────────────────────
|
|
|
|
async def kiwoom_post(client: httpx.AsyncClient, endpoint: str, api_id: str,
|
|
body: dict, cont_yn: str = "N", next_key: str = "",
|
|
return_headers: bool = False):
|
|
token = await kiwoom_token.get(client)
|
|
headers = {
|
|
"Content-Type": "application/json;charset=UTF-8",
|
|
"Authorization": f"Bearer {token}",
|
|
"api-id": api_id,
|
|
"cont-yn": cont_yn,
|
|
}
|
|
if next_key:
|
|
headers["next-key"] = next_key
|
|
# 글로벌 페이싱 + 429(요청한도 초과) 재시도
|
|
for attempt in range(3):
|
|
await kiwoom_rate.acquire()
|
|
r = await client.post(f"{KIWOOM_BASE_URL}{endpoint}",
|
|
headers=headers, json=body, timeout=15)
|
|
if r.status_code == 429 and attempt < 2:
|
|
await asyncio.sleep(0.6 * (attempt + 1))
|
|
continue
|
|
break
|
|
if return_headers: # 연속조회용 cont-yn/next-key 응답헤더 필요
|
|
return r.json(), r.headers
|
|
return r.json()
|
|
|
|
def to_int(v: str) -> int:
|
|
try:
|
|
return int(str(v).replace(",", "").lstrip("+").replace(" ", "") or 0)
|
|
except:
|
|
return 0
|
|
|
|
def to_float(v: str) -> float:
|
|
try:
|
|
return float(str(v).replace(",", "").lstrip("+").replace(" ", "") or 0)
|
|
except:
|
|
return 0.0
|
|
|
|
# ── DB 초기화 ─────────────────────────────────────────────
|
|
|
|
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,
|
|
eps FLOAT DEFAULT 0,
|
|
bps FLOAT DEFAULT 0,
|
|
roe FLOAT DEFAULT 0,
|
|
ev FLOAT DEFAULT 0,
|
|
high_52w INTEGER DEFAULT 0,
|
|
low_52w INTEGER DEFAULT 0,
|
|
foreign_ratio FLOAT DEFAULT 0,
|
|
credit_ratio FLOAT DEFAULT 0,
|
|
float_shares BIGINT 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 stock_ohlcv (
|
|
id SERIAL PRIMARY KEY,
|
|
stock_code VARCHAR(10) NOT NULL,
|
|
dt DATE NOT NULL,
|
|
open_price INTEGER DEFAULT 0,
|
|
high_price INTEGER DEFAULT 0,
|
|
low_price INTEGER DEFAULT 0,
|
|
close_price INTEGER DEFAULT 0,
|
|
volume BIGINT DEFAULT 0,
|
|
trade_amount BIGINT DEFAULT 0,
|
|
foreign_ratio FLOAT DEFAULT 0,
|
|
foreign_net BIGINT DEFAULT 0,
|
|
institution_net BIGINT DEFAULT 0,
|
|
individual_net BIGINT DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT NOW(),
|
|
UNIQUE(stock_code, dt)
|
|
)""")
|
|
await c.execute("CREATE INDEX IF NOT EXISTS idx_ohlcv_code_dt ON stock_ohlcv(stock_code, dt DESC)")
|
|
|
|
await c.execute("""
|
|
CREATE TABLE IF NOT EXISTS stock_foreign_flow (
|
|
id SERIAL PRIMARY KEY,
|
|
stock_code VARCHAR(10) NOT NULL,
|
|
dt DATE NOT NULL,
|
|
close_price INTEGER DEFAULT 0,
|
|
change_qty BIGINT DEFAULT 0,
|
|
hold_qty BIGINT DEFAULT 0,
|
|
hold_ratio FLOAT DEFAULT 0,
|
|
limit_ratio FLOAT DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT NOW(),
|
|
UNIQUE(stock_code, dt)
|
|
)""")
|
|
await c.execute("CREATE INDEX IF NOT EXISTS idx_ff_code_dt ON stock_foreign_flow(stock_code, dt DESC)")
|
|
|
|
await c.execute("""
|
|
CREATE TABLE IF NOT EXISTS stock_short_sale (
|
|
id SERIAL PRIMARY KEY,
|
|
stock_code VARCHAR(10) NOT NULL,
|
|
dt DATE NOT NULL,
|
|
close_price INTEGER DEFAULT 0,
|
|
short_qty BIGINT DEFAULT 0,
|
|
short_balance_qty BIGINT DEFAULT 0,
|
|
trade_weight FLOAT DEFAULT 0,
|
|
short_avg_price INTEGER DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT NOW(),
|
|
UNIQUE(stock_code, dt)
|
|
)""")
|
|
await c.execute("CREATE INDEX IF NOT EXISTS idx_ss_code_dt ON stock_short_sale(stock_code, dt 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,
|
|
foreign_net_5d BIGINT DEFAULT 0,
|
|
short_weight 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("kiwoom.db.initialized")
|
|
|
|
# ── 수집 함수 ─────────────────────────────────────────────
|
|
|
|
async def fetch_basic_info(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore) -> Optional[dict]:
|
|
"""ka10001: 현재가 + 재무지표 (PER/PBR/ROE/EPS/BPS/외국인비중/시가총액)"""
|
|
async with sem:
|
|
try:
|
|
d = await kiwoom_post(client, "/api/dostk/stkinfo", "ka10001", {"stk_cd": code})
|
|
if d.get("return_code", -1) != 0:
|
|
return None
|
|
return {
|
|
"code": code,
|
|
"name": d.get("stk_nm", ""),
|
|
"price": abs(to_int(d.get("cur_prc", "0"))),
|
|
"change_amount": to_int(d.get("pred_pre", "0")),
|
|
"change_pct": to_float(d.get("flu_rt", "0")),
|
|
"volume": to_int(d.get("trde_qty", "0")),
|
|
"open_price": abs(to_int(d.get("open_pric", "0"))),
|
|
"high": abs(to_int(d.get("high_pric", "0"))),
|
|
"low": abs(to_int(d.get("low_pric", "0"))),
|
|
"market_cap": to_int(d.get("mac", "0")) * 100_000_000, # 억 → 원
|
|
"per": to_float(d.get("per", "0")),
|
|
"pbr": to_float(d.get("pbr", "0")),
|
|
"eps": to_float(d.get("eps", "0")),
|
|
"bps": to_float(d.get("bps", "0")),
|
|
"roe": to_float(d.get("roe", "0")),
|
|
"ev": to_float(d.get("ev", "0")),
|
|
"high_52w": abs(to_int(d.get("250hgst", "0"))),
|
|
"low_52w": abs(to_int(d.get("250lwst", "0"))),
|
|
"foreign_ratio": to_float(d.get("for_exh_rt", "0")),
|
|
"credit_ratio": to_float(d.get("crd_rt", "0")),
|
|
"float_shares": to_int(d.get("flo_stk", "0")),
|
|
"sale_amt": to_int(d.get("sale_amt", "0")),
|
|
"operating_profit": to_int(d.get("bus_pro", "0")),
|
|
"net_income": to_int(d.get("cup_nga", "0")),
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
except Exception as e:
|
|
logger.debug("ka10001.err", code=code, error=str(e))
|
|
return None
|
|
|
|
async def fetch_ohlcv(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore, days: int = 365) -> list:
|
|
"""ka10005: 일봉 OHLCV + 외국인·기관 순매수"""
|
|
async with sem:
|
|
try:
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
result = []
|
|
cont_yn, next_key = "N", ""
|
|
for _ in range(20): # 안전 상한(20p ≈ 365봉 충분)
|
|
d, hdr = await kiwoom_post(
|
|
client, "/api/dostk/mrkcond", "ka10005",
|
|
{"stk_cd": code, "dt": today, "req_cnt": days},
|
|
cont_yn=cont_yn, next_key=next_key, return_headers=True)
|
|
for r in d.get("stk_ddwkmm", []):
|
|
cp = abs(to_int(r.get("close_pric", "0")))
|
|
if cp <= 0:
|
|
continue
|
|
result.append({
|
|
"dt": r.get("date", ""),
|
|
"open": abs(to_int(r.get("open_pric", "0"))),
|
|
"high": abs(to_int(r.get("high_pric", "0"))),
|
|
"low": abs(to_int(r.get("low_pric", "0"))),
|
|
"close": cp,
|
|
"volume": to_int(r.get("trde_qty", "0")),
|
|
"trade_amount": to_int(r.get("trde_prica", "0")),
|
|
"foreign_ratio": to_float(r.get("for_poss", "0")),
|
|
"foreign_net": to_int(r.get("for_netprps", "0")),
|
|
"institution_net": to_int(r.get("orgn_netprps", "0")),
|
|
"individual_net": to_int(r.get("ind_netprps", "0")),
|
|
})
|
|
# 키움 연속조회: 응답 cont-yn=Y + next-key 있으면 다음 페이지
|
|
if (len(result) >= days or hdr.get("cont-yn") != "Y"
|
|
or not hdr.get("next-key")):
|
|
break
|
|
cont_yn, next_key = "Y", hdr.get("next-key", "")
|
|
return result
|
|
except Exception as e:
|
|
logger.debug("ka10005.err", code=code, error=str(e))
|
|
return []
|
|
|
|
async def fetch_foreign_flow(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore) -> list:
|
|
"""ka10008: 외국인 종목별 매매동향"""
|
|
async with sem:
|
|
try:
|
|
d = await kiwoom_post(client, "/api/dostk/frgnistt", "ka10008", {"stk_cd": code})
|
|
rows = d.get("stk_frgnr", [])
|
|
result = []
|
|
for r in rows:
|
|
result.append({
|
|
"dt": r.get("dt", ""),
|
|
"close_price": abs(to_int(r.get("close_pric", "0"))),
|
|
"change_qty": to_int(r.get("chg_qty", "0")),
|
|
"hold_qty": to_int(r.get("poss_stkcnt", "0")),
|
|
"hold_ratio": to_float(r.get("wght", "0")),
|
|
"limit_ratio": to_float(r.get("limit_exh_rt", "0")),
|
|
})
|
|
return result
|
|
except Exception as e:
|
|
logger.debug("ka10008.err", code=code, error=str(e))
|
|
return []
|
|
|
|
async def fetch_minute_chart(client: httpx.AsyncClient, code: str, tic_scope: str = "1") -> list:
|
|
"""ka10080: 주식분봉차트조회. tic_scope: 1, 3, 5, 10, 15, 30, 45, 60 분"""
|
|
try:
|
|
d = await kiwoom_post(client, "/api/dostk/chart", "ka10080",
|
|
{"stk_cd": code, "tic_scope": tic_scope, "upd_stkpc_tp": "1"})
|
|
if d.get("return_code", -1) != 0:
|
|
return []
|
|
rows = d.get("stk_min_pole_chart_qry", []) or d.get("stk_min_pole", []) or []
|
|
result = []
|
|
for r in rows:
|
|
result.append({
|
|
"dt": r.get("cntr_tm", r.get("dt", "")),
|
|
"open": abs(to_int(r.get("open_pric", "0"))),
|
|
"high": abs(to_int(r.get("high_pric", "0"))),
|
|
"low": abs(to_int(r.get("low_pric", "0"))),
|
|
"close": abs(to_int(r.get("cur_prc", r.get("close_pric", "0")))),
|
|
"volume": to_int(r.get("trde_qty", "0")),
|
|
})
|
|
return result
|
|
except Exception as e:
|
|
logger.debug("ka10080.err", code=code, error=str(e))
|
|
return []
|
|
|
|
async def fetch_orderbook(client: httpx.AsyncClient, code: str) -> dict:
|
|
"""ka10004: 호가잔량 (10단계 매수/매도). path=/api/dostk/mrkcond"""
|
|
try:
|
|
d = await kiwoom_post(client, "/api/dostk/mrkcond", "ka10004", {"stk_cd": code})
|
|
if d.get("return_code", -1) != 0:
|
|
return {}
|
|
bid, ask = [], []
|
|
# 매도 호가 1~10 (sel_1th_pre_bid = 가격, sel_1th_pre_req = 잔량)
|
|
for i in range(1, 11):
|
|
p = abs(to_int(d.get(f"sel_{i}th_pre_bid", "0")))
|
|
q = to_int(d.get(f"sel_{i}th_pre_req", "0"))
|
|
if p > 0:
|
|
ask.append({"price": p, "qty": q})
|
|
# 매수 호가 1~10
|
|
for i in range(1, 11):
|
|
p = abs(to_int(d.get(f"buy_{i}th_pre_bid", "0")))
|
|
q = to_int(d.get(f"buy_{i}th_pre_req", "0"))
|
|
if p > 0:
|
|
bid.append({"price": p, "qty": q})
|
|
return {
|
|
"code": code,
|
|
"ask": ask, # 매도 (가격 낮은 것부터)
|
|
"bid": bid, # 매수 (가격 높은 것부터)
|
|
"ask_total": to_int(d.get("tot_sel_req", "0")),
|
|
"bid_total": to_int(d.get("tot_buy_req", "0")),
|
|
"base_time": d.get("bid_req_base_tm", ""),
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
except Exception as e:
|
|
logger.debug("ka10004.err", code=code, error=str(e))
|
|
return {}
|
|
|
|
async def fetch_volume_surge() -> list:
|
|
"""ka10023: 거래량급증 종목. path=/api/dostk/rkinfo"""
|
|
try:
|
|
async with httpx.AsyncClient() as c:
|
|
d = await kiwoom_post(c, "/api/dostk/rkinfo", "ka10023",
|
|
{"mrkt_tp": "000", # 000=전체, 001=코스피, 101=코스닥
|
|
"sort_tp": "1", # 1=급증량, 2=급증률
|
|
"tm_tp": "2", # 1=분, 2=전일
|
|
"trde_qty_tp": "5", # 1=5천주, 2=1만주, 5=5만주
|
|
"tm": "",
|
|
"stk_cnd": "0", # 0=전체
|
|
"pric_tp": "0", # 0=전체
|
|
"stex_tp": "3"}) # 3=통합
|
|
if d.get("return_code", -1) != 0:
|
|
return []
|
|
rows = d.get("trde_qty_sdnin", []) or []
|
|
result = []
|
|
for r in rows[:50]:
|
|
raw_code = r.get("stk_cd", "")
|
|
# NXT 등 접미사 제거 ('_AL', '_NX' 등)
|
|
clean_code = raw_code.split("_")[0] if raw_code else ""
|
|
result.append({
|
|
"code": clean_code,
|
|
"name": r.get("stk_nm", ""),
|
|
"price": abs(to_int(r.get("cur_prc", "0"))),
|
|
"change_pct": to_float(r.get("flu_rt", "0")),
|
|
"volume": to_int(r.get("now_trde_qty", r.get("trde_qty", "0"))),
|
|
"prev_volume":to_int(r.get("pred_trde_qty", "0")),
|
|
"surge_rate": to_float(r.get("sdnin_rt", r.get("sdnin_qty", "0"))),
|
|
})
|
|
return result
|
|
except Exception as e:
|
|
logger.debug("ka10023.err", error=str(e))
|
|
return []
|
|
|
|
async def fetch_short_sale(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore) -> list:
|
|
"""ka10014: 공매도 추이"""
|
|
async with sem:
|
|
try:
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
month_ago = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d")
|
|
d = await kiwoom_post(client, "/api/dostk/shsa", "ka10014",
|
|
{"stk_cd": code, "strt_dt": month_ago, "end_dt": today})
|
|
rows = d.get("shrts_trnsn", [])
|
|
result = []
|
|
for r in rows:
|
|
result.append({
|
|
"dt": r.get("dt", ""),
|
|
"close_price": abs(to_int(r.get("close_pric", "0"))),
|
|
"short_qty": to_int(r.get("shrts_qty", "0")),
|
|
"short_balance_qty": to_int(r.get("ovr_shrts_qty", "0")),
|
|
"trade_weight": to_float(r.get("trde_wght", "0")),
|
|
"short_avg_price": to_int(r.get("shrts_avg_pric", "0")),
|
|
})
|
|
return result
|
|
except Exception as e:
|
|
logger.debug("ka10014.err", code=code, error=str(e))
|
|
return []
|
|
|
|
# ── 종목 코드 로드 ────────────────────────────────────────
|
|
|
|
async def get_stock_codes(limit: int = 0) -> list:
|
|
"""is_active 종목 전체 (limit=0이면 제한 없음)"""
|
|
if pg_pool:
|
|
try:
|
|
if limit > 0:
|
|
rows = await pg_pool.fetch(
|
|
"SELECT stock_code FROM dart_corps WHERE is_active=TRUE ORDER BY stock_code LIMIT $1", limit)
|
|
else:
|
|
rows = await pg_pool.fetch(
|
|
"SELECT stock_code FROM dart_corps WHERE is_active=TRUE ORDER BY stock_code")
|
|
codes = [r["stock_code"] for r in rows if r["stock_code"]]
|
|
if len(codes) >= 50:
|
|
return codes
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
async def get_priority_codes(min_mcap: int = 30_000_000_000) -> list:
|
|
"""5분 수집 우선군 — 시총 min_mcap(기본 300억) 이상 활성종목 (잡주 제외).
|
|
조회 실패·표본 부족 시 전체 종목으로 폴백."""
|
|
if pg_pool:
|
|
try:
|
|
rows = await pg_pool.fetch("""
|
|
SELECT d.stock_code FROM dart_corps d
|
|
JOIN LATERAL (
|
|
SELECT market_cap FROM stock_prices
|
|
WHERE stock_code=d.stock_code AND market_cap>0
|
|
ORDER BY collected_at DESC LIMIT 1) p ON true
|
|
WHERE d.is_active=TRUE AND p.market_cap >= $1
|
|
ORDER BY d.stock_code
|
|
""", min_mcap)
|
|
codes = [r["stock_code"] for r in rows if r["stock_code"]]
|
|
if len(codes) >= 50:
|
|
return codes
|
|
except Exception:
|
|
pass
|
|
return await get_stock_codes(0)
|
|
|
|
# ── 저장 함수 ─────────────────────────────────────────────
|
|
|
|
async def save_price(info: dict):
|
|
if redis_cl:
|
|
redis_data = {
|
|
"code": info["code"], "name": info["name"],
|
|
"price": info["price"], "change_pct": info["change_pct"],
|
|
"change_amount": info["change_amount"],
|
|
"volume": info["volume"], "high": info["high"], "low": info["low"],
|
|
"open_price": info["open_price"],
|
|
"market_cap": info["market_cap"],
|
|
"per": info["per"], "pbr": info["pbr"],
|
|
"eps": info["eps"], "bps": info["bps"],
|
|
"roe": info["roe"], "ev": info["ev"],
|
|
"high_52w": info["high_52w"], "low_52w": info["low_52w"],
|
|
"foreign_ratio": info["foreign_ratio"],
|
|
"credit_ratio": info["credit_ratio"],
|
|
"timestamp": info["timestamp"],
|
|
}
|
|
await redis_cl.set(f"price:{info['code']}", json.dumps(redis_data, ensure_ascii=False), ex=600)
|
|
|
|
if pg_pool:
|
|
try:
|
|
async with pg_pool.acquire() as c:
|
|
await c.execute("""
|
|
INSERT INTO stock_prices (
|
|
stock_code, stock_name, price, change_pct, change_amount,
|
|
volume, high, low, open_price, market_cap,
|
|
per, pbr, eps, bps, roe, ev,
|
|
high_52w, low_52w, foreign_ratio, credit_ratio, float_shares, collected_at)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22)
|
|
""", info["code"], info["name"], info["price"], info["change_pct"], info["change_amount"],
|
|
info["volume"], info["high"], info["low"], info["open_price"], info["market_cap"],
|
|
info["per"], info["pbr"], info["eps"], info["bps"], info["roe"], info["ev"],
|
|
info["high_52w"], info["low_52w"], info["foreign_ratio"], info["credit_ratio"],
|
|
info["float_shares"], datetime.now())
|
|
except Exception as e:
|
|
logger.debug("save_price.err", code=info["code"], error=str(e))
|
|
|
|
async def save_ohlcv(code: str, rows: list):
|
|
if not rows or not pg_pool:
|
|
return
|
|
# Redis에 최근 60일 일봉 저장 (ta-engine 보조)
|
|
if redis_cl:
|
|
await redis_cl.set(f"ohlcv:{code}", json.dumps(rows[:60], ensure_ascii=False), ex=86400)
|
|
async with pg_pool.acquire() as c:
|
|
for r in rows:
|
|
try:
|
|
dt = datetime.strptime(r["dt"], "%Y%m%d").date()
|
|
await c.execute("""
|
|
INSERT INTO stock_ohlcv (
|
|
stock_code, dt, open_price, high_price, low_price, close_price,
|
|
volume, trade_amount, foreign_ratio, foreign_net, institution_net, individual_net)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
|
|
ON CONFLICT (stock_code, dt) DO UPDATE SET
|
|
close_price=$6, volume=$7, foreign_ratio=$9,
|
|
foreign_net=$10, institution_net=$11, individual_net=$12
|
|
""", code, dt, r["open"], r["high"], r["low"], r["close"],
|
|
r["volume"], r["trade_amount"], r["foreign_ratio"],
|
|
r["foreign_net"], r["institution_net"], r["individual_net"])
|
|
except Exception:
|
|
pass
|
|
|
|
async def save_foreign_flow(code: str, rows: list):
|
|
if not rows or not pg_pool:
|
|
return
|
|
if redis_cl and rows:
|
|
await redis_cl.set(f"foreign:{code}", json.dumps(rows[:20], ensure_ascii=False), ex=86400)
|
|
async with pg_pool.acquire() as c:
|
|
for r in rows:
|
|
try:
|
|
dt = datetime.strptime(r["dt"], "%Y%m%d").date()
|
|
await c.execute("""
|
|
INSERT INTO stock_foreign_flow (
|
|
stock_code, dt, close_price, change_qty, hold_qty, hold_ratio, limit_ratio)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7)
|
|
ON CONFLICT (stock_code, dt) DO UPDATE SET
|
|
change_qty=$4, hold_qty=$5, hold_ratio=$6, limit_ratio=$7
|
|
""", code, dt, r["close_price"], r["change_qty"],
|
|
r["hold_qty"], r["hold_ratio"], r["limit_ratio"])
|
|
except Exception:
|
|
pass
|
|
|
|
async def save_short_sale(code: str, rows: list):
|
|
if not rows or not pg_pool:
|
|
return
|
|
if redis_cl and rows:
|
|
await redis_cl.set(f"short:{code}", json.dumps(rows[:20], ensure_ascii=False), ex=86400)
|
|
async with pg_pool.acquire() as c:
|
|
for r in rows:
|
|
try:
|
|
dt = datetime.strptime(r["dt"], "%Y%m%d").date()
|
|
await c.execute("""
|
|
INSERT INTO stock_short_sale (
|
|
stock_code, dt, close_price, short_qty, short_balance_qty,
|
|
trade_weight, short_avg_price)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7)
|
|
ON CONFLICT (stock_code, dt) DO UPDATE SET
|
|
short_qty=$4, short_balance_qty=$5, trade_weight=$6, short_avg_price=$7
|
|
""", code, dt, r["close_price"], r["short_qty"],
|
|
r["short_balance_qty"], r["trade_weight"], r["short_avg_price"])
|
|
except Exception:
|
|
pass
|
|
|
|
# ── 수집 작업 ─────────────────────────────────────────────
|
|
|
|
class Stats:
|
|
collected = 0; errors = 0; last_run = ""; last_full_run = ""
|
|
|
|
stats = Stats()
|
|
|
|
_job_price_running = False
|
|
|
|
async def job_price():
|
|
"""평일 9~15시: 시총 300억+ 우선군 현재가 수집 (ka10001).
|
|
잡주(시총<300억)는 후순위 — job_full(17시)에서 일봉으로 커버.
|
|
글로벌 6/s 페이싱으로 한 사이클 수 분 소요 → 중복 실행 가드."""
|
|
global _job_price_running
|
|
if _job_price_running:
|
|
logger.info("job_price.skip_running")
|
|
return
|
|
_job_price_running = True
|
|
try:
|
|
codes = await get_priority_codes()
|
|
if not codes:
|
|
logger.warning("job_price.no_codes")
|
|
return
|
|
sem = asyncio.Semaphore(8) # 동시 in-flight 한도 (실제 속도는 kiwoom_rate가 제어)
|
|
ok = 0
|
|
async with httpx.AsyncClient() as client:
|
|
tasks = [fetch_basic_info(client, code, sem) for code in codes]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
for info in results:
|
|
if isinstance(info, dict) and info.get("price", 0) > 0:
|
|
await save_price(info)
|
|
ok += 1
|
|
else:
|
|
stats.errors += 1
|
|
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())
|
|
try:
|
|
await check_price_alerts()
|
|
except Exception as e:
|
|
logger.warning("alert.check_err", error=str(e))
|
|
logger.info("job_price.done", ok=ok, total=len(codes))
|
|
finally:
|
|
_job_price_running = False
|
|
|
|
_job_priority_running = False
|
|
|
|
async def job_price_priority():
|
|
"""30초마다: 보유 + 강력매수 + 매수관심 종목만 빠른 수집 (자동매매·알림 latency 단축).
|
|
키움 rate limit 안전 — 우선순위 종목은 보통 30~50건이라 6/s × 10초 이내 완료."""
|
|
global _job_priority_running
|
|
if _job_priority_running:
|
|
return
|
|
_job_priority_running = True
|
|
try:
|
|
async with pg_pool.acquire() as conn:
|
|
rows = await conn.fetch("""
|
|
SELECT DISTINCT stock_code FROM (
|
|
SELECT stock_code FROM user_portfolio WHERE active=true
|
|
UNION
|
|
SELECT stock_code FROM user_alerts WHERE active=true
|
|
UNION
|
|
SELECT stock_code FROM stock_scores
|
|
WHERE score_date=(SELECT MAX(score_date) FROM stock_scores)
|
|
AND recommendation IN ('강력매수','매수관심')
|
|
) AS t LIMIT 60
|
|
""")
|
|
codes = [r["stock_code"] for r in rows]
|
|
if not codes:
|
|
return
|
|
sem = asyncio.Semaphore(8)
|
|
ok = 0
|
|
async with httpx.AsyncClient() as client:
|
|
tasks = [fetch_basic_info(client, c, sem) for c in codes]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
for info in results:
|
|
if isinstance(info, dict) and info.get("price", 0) > 0:
|
|
await save_price(info)
|
|
ok += 1
|
|
try:
|
|
await check_price_alerts() # 알림 즉시 평가
|
|
except Exception as e:
|
|
logger.warning("priority.alert_err", error=str(e))
|
|
logger.info("job_price_priority.done", ok=ok, total=len(codes))
|
|
finally:
|
|
_job_priority_running = False
|
|
|
|
|
|
async def job_full(days: int = 10):
|
|
"""평일 17:00: 일봉·외국인·공매도 전체 수집 (ka10005·ka10008·ka10014).
|
|
일별은 days=10(경량, ON CONFLICT 갱신). 백필은 /collect/full?days=400."""
|
|
codes = await get_stock_codes(0)
|
|
if not codes:
|
|
return
|
|
sem = asyncio.Semaphore(5) # 야간 수집 느리게
|
|
ok = 0
|
|
async with httpx.AsyncClient() as client:
|
|
for code in codes:
|
|
try:
|
|
ohlcv, foreign, short = await asyncio.gather(
|
|
fetch_ohlcv(client, code, sem, days),
|
|
fetch_foreign_flow(client, code, sem),
|
|
fetch_short_sale(client, code, sem),
|
|
return_exceptions=True
|
|
)
|
|
if isinstance(ohlcv, list):
|
|
await save_ohlcv(code, ohlcv)
|
|
if isinstance(foreign, list):
|
|
await save_foreign_flow(code, foreign)
|
|
if isinstance(short, list):
|
|
await save_short_sale(code, short)
|
|
ok += 1
|
|
except Exception as e:
|
|
logger.debug("job_full.err", code=code, error=str(e))
|
|
stats.last_full_run = datetime.now().isoformat()
|
|
logger.info("job_full.done", ok=ok, total=len(codes))
|
|
|
|
# ── 시그널 생성 ───────────────────────────────────────────
|
|
|
|
def calc_signal(info: dict, news_sc: float, dart_sc: float,
|
|
foreign_net_5d: int, short_weight: float) -> Optional[dict]:
|
|
price = info["price"]
|
|
if price <= 0:
|
|
return None
|
|
h52 = info.get("high_52w", 0)
|
|
l52 = info.get("low_52w", 0)
|
|
cpct = info.get("change_pct", 0)
|
|
foreign_ratio = info.get("foreign_ratio", 0)
|
|
|
|
# 외국인 수급 보너스 (5일 누적 순매수 양수면 +10, 음수면 -10)
|
|
foreign_bonus = min(10, max(-10, foreign_net_5d / 500_000)) if foreign_net_5d else 0
|
|
# 공매도 패널티 (비중 5% 이상이면 -5)
|
|
short_penalty = -5 if short_weight >= 5 else 0
|
|
|
|
total = (news_sc * 0.35 + dart_sc * 0.25
|
|
+ (cpct * 10) * 0.25 + foreign_bonus * 0.10 + short_penalty * 0.05)
|
|
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.10)
|
|
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.90)
|
|
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),
|
|
"foreign_net_5d": foreign_net_5d, "short_weight": short_weight,
|
|
}
|
|
|
|
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 not redis_cl:
|
|
continue
|
|
cached = await redis_cl.get(f"price:{code}")
|
|
if not cached:
|
|
continue
|
|
info = json.loads(cached)
|
|
|
|
# 외국인 5일 순매수 합산
|
|
foreign_net_5d = 0
|
|
f_cached = await redis_cl.get(f"foreign:{code}")
|
|
if f_cached:
|
|
f_rows = json.loads(f_cached)
|
|
foreign_net_5d = sum(r.get("change_qty", 0) for r in f_rows[:5])
|
|
|
|
# 공매도 최근 비중
|
|
short_weight = 0.0
|
|
s_cached = await redis_cl.get(f"short:{code}")
|
|
if s_cached:
|
|
s_rows = json.loads(s_cached)
|
|
if s_rows:
|
|
short_weight = s_rows[0].get("trade_weight", 0.0)
|
|
|
|
sig = calc_signal(info, row["news_score"], row["dart_score"],
|
|
foreign_net_5d, short_weight)
|
|
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,
|
|
foreign_net_5d, short_weight)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
|
|
ON CONFLICT (stock_code, (created_at::date)) DO UPDATE SET
|
|
signal_type=EXCLUDED.signal_type, current_price=EXCLUDED.current_price,
|
|
target_price=EXCLUDED.target_price, stop_loss=EXCLUDED.stop_loss,
|
|
expected_return_pct=EXCLUDED.expected_return_pct,
|
|
risk_reward_ratio=EXCLUDED.risk_reward_ratio,
|
|
confidence=EXCLUDED.confidence, reason=EXCLUDED.reason,
|
|
news_score=EXCLUDED.news_score, dart_score=EXCLUDED.dart_score,
|
|
price_momentum=EXCLUDED.price_momentum,
|
|
foreign_net_5d=EXCLUDED.foreign_net_5d, short_weight=EXCLUDED.short_weight
|
|
""", 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"],
|
|
sig["foreign_net_5d"], sig["short_weight"])
|
|
n += 1
|
|
logger.info("signals.done", count=n)
|
|
|
|
# ── FastAPI ───────────────────────────────────────────────
|
|
|
|
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()
|
|
# 현재가: 평일 9~16시 2분마다 (전체)
|
|
scheduler.add_job(job_price, "cron", day_of_week="mon-fri",
|
|
hour="9-15", minute="*/2", id="price", replace_existing=True)
|
|
# Fast track: 보유 + 강력매수 + 매수관심만 30초마다 (자동매매 latency)
|
|
scheduler.add_job(job_price_priority, "cron", day_of_week="mon-fri",
|
|
hour="9-15", second="*/30", id="price_priority", replace_existing=True)
|
|
# 장 마감 후 전체 수집: 평일 17:00
|
|
scheduler.add_job(job_full, "cron", day_of_week="mon-fri",
|
|
hour="17", minute="0", id="full", replace_existing=True)
|
|
scheduler.start()
|
|
logger.info("kiwoom-api.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, "errors": stats.errors,
|
|
"last_run": stats.last_run, "last_full_run": stats.last_full_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))
|
|
# 실시간 단건 조회
|
|
sem = asyncio.Semaphore(1)
|
|
async with httpx.AsyncClient() as client:
|
|
info = await fetch_basic_info(client, code, sem)
|
|
if info:
|
|
await save_price(info)
|
|
return JSONResponse(content=info)
|
|
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:*")
|
|
result = {}
|
|
for k in keys[:limit]:
|
|
c = await redis_cl.get(k)
|
|
if c:
|
|
d = json.loads(c)
|
|
result[d["code"]] = d
|
|
return JSONResponse(content={"count": len(result), "data": result})
|
|
|
|
@app.get("/foreign/{code}")
|
|
async def foreign(code: str):
|
|
if redis_cl:
|
|
c = await redis_cl.get(f"foreign:{code}")
|
|
if c:
|
|
return JSONResponse(content={"code": code, "data": json.loads(c)})
|
|
sem = asyncio.Semaphore(1)
|
|
async with httpx.AsyncClient() as client:
|
|
rows = await fetch_foreign_flow(client, code, sem)
|
|
await save_foreign_flow(code, rows)
|
|
return JSONResponse(content={"code": code, "data": rows})
|
|
|
|
@app.get("/short/{code}")
|
|
async def short_sale(code: str):
|
|
if redis_cl:
|
|
c = await redis_cl.get(f"short:{code}")
|
|
if c:
|
|
return JSONResponse(content={"code": code, "data": json.loads(c)})
|
|
sem = asyncio.Semaphore(1)
|
|
async with httpx.AsyncClient() as client:
|
|
rows = await fetch_short_sale(client, code, sem)
|
|
await save_short_sale(code, rows)
|
|
return JSONResponse(content={"code": code, "data": rows})
|
|
|
|
@app.get("/ohlcv/{code}")
|
|
async def ohlcv(code: str, days: int = Query(default=60)):
|
|
if redis_cl:
|
|
c = await redis_cl.get(f"ohlcv:{code}")
|
|
if c:
|
|
data = json.loads(c)
|
|
return JSONResponse(content={"code": code, "data": data[:days]})
|
|
sem = asyncio.Semaphore(1)
|
|
async with httpx.AsyncClient() as client:
|
|
rows = await fetch_ohlcv(client, code, sem, days)
|
|
await save_ohlcv(code, rows)
|
|
return JSONResponse(content={"code": code, "data": rows})
|
|
|
|
@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 50
|
|
""" % 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.get("/summary/{code}")
|
|
async def summary(code: str):
|
|
"""종목 종합 요약 (현재가 + 외국인 + 공매도 + 최근 신호)"""
|
|
result = {"code": code}
|
|
if redis_cl:
|
|
for key in [f"price:{code}", f"foreign:{code}", f"short:{code}", f"ohlcv:{code}"]:
|
|
c = await redis_cl.get(key)
|
|
if c:
|
|
field = key.split(":")[0]
|
|
result[field] = json.loads(c) if field != "price" else json.loads(c)
|
|
return JSONResponse(content=result)
|
|
|
|
# ── 가격 알림 (user_alerts 트리거 + 5가지 물타기 진단) ────────
|
|
|
|
async def send_telegram(msg: str):
|
|
if not TG_TOKEN or not TG_CHAT_ID:
|
|
return
|
|
try:
|
|
async with httpx.AsyncClient() as c:
|
|
await c.post(
|
|
f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage",
|
|
json={"chat_id": TG_CHAT_ID, "text": msg, "parse_mode": "HTML"},
|
|
timeout=10)
|
|
except Exception as e:
|
|
logger.warning("telegram.err", error=str(e))
|
|
|
|
|
|
async def diagnose_avg_down(conn, code: str, current_price: float) -> tuple[int, list[str]]:
|
|
"""물타기 진단 5가지 조건. (충족 개수, 표시줄 리스트) 반환."""
|
|
score = 0
|
|
lines = []
|
|
|
|
# 1. RSI ≤ 30 (과매도)
|
|
rsi = await conn.fetchval("""
|
|
SELECT rsi FROM stock_technical WHERE stock_code=$1
|
|
""", code)
|
|
rsi_disp = f"{rsi:.0f}" if rsi is not None else "N/A"
|
|
if rsi is not None and rsi <= 30:
|
|
score += 1
|
|
lines.append(f"✅ RSI {rsi_disp} (과매도)")
|
|
elif rsi is not None and rsi <= 40:
|
|
lines.append(f"⚠️ RSI {rsi_disp} (약한 과매도)")
|
|
else:
|
|
lines.append(f"❌ RSI {rsi_disp} (과매도 아님)")
|
|
|
|
# 2. 거래량 비율 ≤ 1.5 (패닉 매도 끝)
|
|
vol_ratio = await conn.fetchval("""
|
|
SELECT vol_ratio FROM stock_technical WHERE stock_code=$1
|
|
""", code)
|
|
vol_disp = f"{vol_ratio:.2f}" if vol_ratio is not None else "N/A"
|
|
if vol_ratio is not None and vol_ratio <= 1.5:
|
|
score += 1
|
|
lines.append(f"✅ 거래량비율 {vol_disp} (안정)")
|
|
else:
|
|
lines.append(f"❌ 거래량비율 {vol_disp} (패닉 진행)")
|
|
|
|
# 3. 공매도 잔고 감소 추세 (최근 5일)
|
|
short_rows = await conn.fetch("""
|
|
SELECT short_balance_qty FROM stock_short_sale
|
|
WHERE stock_code=$1 ORDER BY dt DESC LIMIT 5
|
|
""", code)
|
|
if len(short_rows) >= 2:
|
|
latest = short_rows[0]["short_balance_qty"] or 0
|
|
older = short_rows[-1]["short_balance_qty"] or 0
|
|
if older > 0 and latest < older:
|
|
chg = (latest - older) / older * 100
|
|
score += 1
|
|
lines.append(f"✅ 공매도잔고 {chg:+.1f}% (감소)")
|
|
else:
|
|
lines.append(f"❌ 공매도잔고 증가/유지")
|
|
else:
|
|
lines.append("⚠️ 공매도 데이터 부족")
|
|
|
|
# 4. 최근 24시간 악재(intensity≥3) 없음
|
|
bad_news = await conn.fetchval("""
|
|
SELECT COUNT(*) FROM news_analysis
|
|
WHERE primary_stock=$1
|
|
AND sentiment='악재' AND intensity >= 3
|
|
AND analyzed_at >= NOW() - INTERVAL '24 hours'
|
|
""", code)
|
|
if bad_news == 0:
|
|
score += 1
|
|
lines.append("✅ 최근 악재 없음")
|
|
else:
|
|
lines.append(f"❌ 최근 악재 {bad_news}건")
|
|
|
|
# 5. 시장 레짐 = 강세/중립 (약세장 회피)
|
|
regime = await conn.fetchval("""
|
|
SELECT regime FROM market_regime ORDER BY dt DESC LIMIT 1
|
|
""")
|
|
if regime in ("강세", "중립"):
|
|
score += 1
|
|
lines.append(f"✅ 시장레짐 {regime}")
|
|
else:
|
|
lines.append(f"❌ 시장레짐 {regime} (약세장 회피)")
|
|
|
|
return score, lines
|
|
|
|
|
|
async def check_price_alerts():
|
|
"""수집 후 호출: active=true 알림 중 임계값 도달한 것 처리."""
|
|
if not pg_pool:
|
|
return
|
|
async with pg_pool.acquire() as conn:
|
|
alerts = await conn.fetch("""
|
|
SELECT a.id, a.stock_code, a.alert_type, a.threshold,
|
|
d.corp_name
|
|
FROM user_alerts a
|
|
LEFT JOIN dart_corps d ON d.stock_code=a.stock_code
|
|
WHERE a.active = true
|
|
""")
|
|
for a in alerts:
|
|
code = a["stock_code"]
|
|
price = 0.0
|
|
# 1차: Redis price:{code} (평일 수집 직후)
|
|
cached = await redis_cl.get(f"price:{code}") if redis_cl else None
|
|
if cached:
|
|
try:
|
|
price = float(json.loads(cached).get("price", 0))
|
|
except Exception:
|
|
price = 0.0
|
|
# 2차 폴백: stock_ohlcv 최신 종가 (장 닫힘·주말·실패 대비)
|
|
if price <= 0:
|
|
p = await conn.fetchval("""
|
|
SELECT close_price FROM stock_ohlcv
|
|
WHERE stock_code=$1 ORDER BY dt DESC LIMIT 1
|
|
""", code)
|
|
price = float(p) if p else 0.0
|
|
if price <= 0:
|
|
continue
|
|
hit = False
|
|
if a["alert_type"] == "price_below" and price <= a["threshold"]:
|
|
hit = True
|
|
elif a["alert_type"] == "price_above" and price >= a["threshold"]:
|
|
hit = True
|
|
if not hit:
|
|
continue
|
|
|
|
score, lines = await diagnose_avg_down(conn, code, price)
|
|
verdict = "🟢 물타기 OK" if score >= 4 else ("🟡 신중" if score >= 3 else "🔴 NO")
|
|
name = a["corp_name"] or code
|
|
|
|
# LLM hybrid 분석 동봉 (score-engine /deep-analysis 호출)
|
|
llm_block = ""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=120) as c:
|
|
r = await c.get(
|
|
f"http://score-engine:8686/deep-analysis/{code}",
|
|
params={"refresh": "true", "model": "hybrid", "notify": "false"})
|
|
if r.status_code == 200:
|
|
d = r.json()
|
|
if not d.get("error"):
|
|
agr = d.get("agreement")
|
|
agr_tag = "✅두AI일치" if agr is True else ("⚠️AI의견갈림" if agr is False else "")
|
|
llm_block = (
|
|
f"\n\n🤖 <b>AI 하이브리드 분석 {agr_tag}</b>\n"
|
|
f"판단: {d.get('recommendation','-')} (확신도 {d.get('conviction',0)}/5)\n"
|
|
f"논거: {d.get('thesis','')[:250]}\n"
|
|
f"🎯 목표 {int(d.get('target_price') or 0):,}원 / "
|
|
f"손절 {int(d.get('stop_loss') or 0):,}원"
|
|
)
|
|
if agr is False and d.get("exaone"):
|
|
e = d["exaone"]
|
|
llm_block += (
|
|
f"\n🔄 EXAONE 의견: {e.get('recommendation')} "
|
|
f"{e.get('conviction',0)}/5 — {e.get('thesis','')[:120]}"
|
|
)
|
|
except Exception as e:
|
|
logger.warning("alert.llm_err", code=code, error=str(e))
|
|
|
|
msg = (
|
|
f"🔔 <b>가격 알림: {name} ({code})</b>\n"
|
|
f"현재가: {int(price):,}원\n"
|
|
f"임계값: {int(a['threshold']):,}원 ({a['alert_type']})\n\n"
|
|
f"📊 <b>물타기 진단 ({score}/5)</b>\n"
|
|
+ "\n".join(lines)
|
|
+ f"\n\n<b>종합: {verdict}</b>"
|
|
+ llm_block
|
|
)
|
|
await send_telegram(msg)
|
|
await conn.execute("""
|
|
UPDATE user_alerts
|
|
SET active=false, last_triggered=NOW()
|
|
WHERE id=$1
|
|
""", a["id"])
|
|
logger.info("alert.fired", code=code, price=price,
|
|
threshold=a["threshold"], verdict=verdict)
|
|
|
|
|
|
@app.post("/alerts/register")
|
|
async def register_alert(code: str = Query(...),
|
|
alert_type: str = Query(..., regex="^(price_below|price_above)$"),
|
|
threshold: float = Query(..., gt=0),
|
|
user_id: int = Query(default=1)):
|
|
async with pg_pool.acquire() as c:
|
|
row = await c.fetchrow("""
|
|
INSERT INTO user_alerts (user_id, stock_code, alert_type, threshold, active)
|
|
VALUES ($1, $2, $3, $4, true)
|
|
RETURNING id
|
|
""", user_id, code, alert_type, threshold)
|
|
return {"id": row["id"], "code": code, "alert_type": alert_type, "threshold": threshold}
|
|
|
|
|
|
@app.get("/alerts")
|
|
async def list_alerts(active_only: bool = Query(default=True)):
|
|
async with pg_pool.acquire() as c:
|
|
if active_only:
|
|
rows = await c.fetch("""
|
|
SELECT a.*, d.corp_name FROM user_alerts a
|
|
LEFT JOIN dart_corps d ON d.stock_code=a.stock_code
|
|
WHERE a.active=true ORDER BY a.created_at DESC
|
|
""")
|
|
else:
|
|
rows = await c.fetch("""
|
|
SELECT a.*, d.corp_name FROM user_alerts a
|
|
LEFT JOIN dart_corps d ON d.stock_code=a.stock_code
|
|
ORDER BY a.created_at DESC LIMIT 50
|
|
""")
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
@app.post("/alerts/check")
|
|
async def trigger_alert_check():
|
|
"""수동 알림 체크 — 주말·장마감·테스트 시 사용 (job_price 내 자동 호출 외 추가 경로)."""
|
|
await check_price_alerts()
|
|
return {"status": "checked"}
|
|
|
|
|
|
@app.delete("/alerts/{alert_id}")
|
|
async def delete_alert(alert_id: int):
|
|
async with pg_pool.acquire() as c:
|
|
await c.execute("UPDATE user_alerts SET active=false WHERE id=$1", alert_id)
|
|
return {"status": "deactivated", "id": alert_id}
|
|
|
|
|
|
@app.post("/collect/price")
|
|
async def collect_price():
|
|
asyncio.create_task(job_price())
|
|
return {"status": "started", "job": "price"}
|
|
|
|
@app.post("/collect/full")
|
|
async def collect_full(days: int = Query(default=10, ge=1, le=600)):
|
|
asyncio.create_task(job_full(days))
|
|
return {"status": "started", "job": "full", "days": days}
|
|
|
|
# ── 추가: 분봉 / 호가 / 거래량급증 ─────────────────────────
|
|
|
|
@app.get("/minute/{code}")
|
|
async def minute_chart(code: str, scope: str = Query(default="5", description="분봉 단위: 1,3,5,10,15,30,60")):
|
|
"""ka10080: 분봉차트 (실시간 호출)"""
|
|
async with httpx.AsyncClient() as c:
|
|
data = await fetch_minute_chart(c, code, scope)
|
|
return JSONResponse(content={"code": code, "scope": scope, "data": data})
|
|
|
|
@app.get("/orderbook/{code}")
|
|
async def orderbook(code: str):
|
|
"""ka10004: 10단계 호가 잔량"""
|
|
async with httpx.AsyncClient() as c:
|
|
data = await fetch_orderbook(c, code)
|
|
return JSONResponse(content=data or {"code": code, "ask": [], "bid": []})
|
|
|
|
@app.get("/volume-surge")
|
|
async def volume_surge():
|
|
"""ka10023: 거래량 급증 종목 TOP50 (Redis 5분 캐시)"""
|
|
cached = None
|
|
if redis_cl:
|
|
try:
|
|
v = await redis_cl.get("vol_surge:list")
|
|
if v: cached = json.loads(v)
|
|
except: pass
|
|
if cached:
|
|
return JSONResponse(content={"data": cached, "cached": True})
|
|
data = await fetch_volume_surge()
|
|
if redis_cl and data:
|
|
try: await redis_cl.setex("vol_surge:list", 300, json.dumps(data))
|
|
except: pass
|
|
return JSONResponse(content={"data": data, "cached": False})
|