shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

추천로직

Browse Source
This commit is contained in:
shuim
2026-03-06 10:01:13 +09:00
parent d6841a06f7
commit c93b35ba04
11 changed files with 1051 additions and 586 deletions
Download Patch File Download Diff File Expand all files Collapse all files

317
backend/batch_daily_analysis.py
View File

@@ -31,9 +31,11 @@ sys.path.insert(0, str(Path(__file__).parent))
from app.core.config import settings
from app.core.database import AsyncSessionLocal # 🔧 공통 DB 모듈 사용
from app.services.korea_investment_api import KoreaInvestmentAPI, korea_investment_api
import aiohttp
from sqlalchemy import select as sa_select
from app.models.models import StockDailyPrice
from app.services.analysis_db_service import bulk_upsert_analysis, passes_price_filter, run_db_scanner # 🔧 단일화
from app.api.endpoints.recommendations import analyze_stock_for_batch, StockRecommendation
from app.services.market_analyzer import analyze_stock_for_batch_db, StockRecommendation
import logging
# 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출
@@ -55,33 +57,57 @@ MIN_STOCK_COUNT = 50
# 1. 시장 레짐 판별 (KOSPI 기준)
# -------------------------------------------------------------------
async def _fetch_db_chart(stock_code: str, limit: int = 240) -> list:
"""stock_daily_prices DB에서 KIS 형식 차트 데이터 조회 (newest first).
KIS API 호출 없이 collect_krx_data.py가 사전 수집한 데이터를 사용.
"""
async with AsyncSessionLocal() as db:
result = await db.execute(
sa_select(StockDailyPrice)
.where(StockDailyPrice.stock_code == stock_code)
.order_by(StockDailyPrice.trade_date.desc())
.limit(limit)
)
rows = result.scalars().all()
return [
{
"stck_bsop_date": row.trade_date.strftime("%Y%m%d"),
"stck_clpr": str(row.close_price or 0),
"stck_oprc": str(row.open_price or 0),
"stck_hgpr": str(row.high_price or 0),
"stck_lwpr": str(row.low_price or 0),
"acml_vol": str(row.volume or 0),
"hts_kor_isnm": row.stock_name or "",
}
for row in rows
]
async def detect_and_log_market_regime() -> str:
"""배치 시작 시 KOSPI 200일 데이터로 시장 레짐 판별.
"""배치 시작 시 KOSPI200 레짐 판별 — DB에서 KODEX200 데이터 조회.
BULL / BEAR / SIDEWAYS 중 하나를 리턴하며 배치 전체에 로깅된다.
BULL / BEAR 중 하나를 리턴하며 배치 전체에 로깅된다.
실패 시 UNKNOWN을 리턴하여 배치를 차단하지 않는다.
pykrx 대신 KIS API 사용 — KODEX 200(069500)으로 KOSPI 레짐 판별.
KODEX 200은 KOSPI 200을 추종하므로 지수 대용으로 적합하다.
"""
try:
from app.services.mean_reversion_strategy import detect_market_regime
# KODEX 200(069500) 일봉 240개 = 약 1년치 데이터
data = await korea_investment_api.get_daily_chart("069500", period="D", limit=240)
output = data.get("output2") or []
if not isinstance(output, list) or len(output) < 205:
logger.warning("⚠️ KODEX200 데이터 부족(%d행) — 레짐 판별 스킵", len(output) if output else 0)
chart_data = await _fetch_db_chart("069500", limit=240)
if not chart_data:
logger.warning("⚠️ KOSPI200(KODEX200) DB 데이터 없음 — 레짐 판별 스킵")
return "UNKNOWN"
# KIS API는 최신순 반환 → 오래된 순(시계열 순)으로 역정렬
closes = [float(item["stck_clpr"]) for item in reversed(output) if item.get("stck_clpr")]
regime = detect_market_regime(closes)
# DB 데이터는 newest first → oldest first 변환 후 detect_market_regime 호출
closes = list(reversed([float(item["stck_clpr"]) for item in chart_data]))
regime_emoji = {"상승장": "📈 BULL", "하락장": "📉 BEAR",
"bull": "📈 BULL", "bear": "📉 BEAR"}.get(regime.value, "➡️ " + regime.value.upper())
logger.info("🌏 시장 레짐: %s (KODEX200 %d일 기준)", regime_emoji, len(closes))
if len(closes) < 205:
logger.warning("⚠️ KOSPI200 데이터 부족(%d행) — 레짐 판별 스킵", len(closes))
return "UNKNOWN"
regime = detect_market_regime(closes)
regime_emoji = {"bull": "📈 BULL", "bear": "📉 BEAR"}.get(regime.value, "➡️ " + regime.value.upper())
logger.info("🌏 시장 레짐: %s (KODEX200 %d일 기준, DB)", regime_emoji, len(closes))
return regime.value.upper()
except Exception as exc:
@@ -93,87 +119,146 @@ async def detect_and_log_market_regime() -> str:
# 2. 종목 목록 조회
# -------------------------------------------------------------------
async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500) -> list[tuple[str, str]]:
"""거래량 상위 종목 조회 (KIS API 사용)
async def get_all_tradable_stocks(limit: int = 500) -> list[tuple[str, str]]:
"""거래 가능 종목 조회 (네이버 금융 sise 사용)
pykrx 의존성 제거 — KIS 거래량 순위 API(FHPST01710000) 사용.
시가총액 기준 → 거래량 기준으로 변경 (유동성 높은 종목 우선).
네이버 금융 전체 시세 페이지에서 KOSPI+KOSDAQ 종목 수집.
(data.krx.co.kr IP 차단 우회 — 네이버 서버는 별도)
ETF / 레버리지 / 스팩 등 특수종목 자동 제외.
KOSPI + KOSDAQ 합산 거래량 상위 limit개 반환.
Raises:
RuntimeError: 조회 결과가 MIN_STOCK_COUNT(50) 미만인 경우 배치 중단
"""
logger.info("Fetching top %d stocks by volume from KIS API...", limit)
from bs4 import BeautifulSoup
SKIP_KEYWORDS = [
"ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG",
"KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX",
"KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "RISE",
"선물", "스팩", "리츠", "SPAC",
]
HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; autostock-batch/1.0)"}
# 모의투자 모드에서 실계좌 시세 API를 사용할 수 있으면 준비
real_market_api: Optional[KoreaInvestmentAPI] = None
if settings.KIS_MODE == "virtual" and settings.KIS_REAL_APP_KEY:
try:
real_market_api = KoreaInvestmentAPI.for_real_market_data()
logger.info("📡 KOSDAQ 조회: 실계좌 시세 API 사용 (모의투자 서버 제한 우회)")
except Exception as exc:
logger.warning("⚠️ 실계좌 시세 API 초기화 실패, 모의투자 API 사용: %s", exc)
logger.info("📡 네이버 금융 종목 조회 (KOSPI+KOSDAQ)")
stocks_with_vol = []
for market_code, market_name in [("J", "KOSPI"), ("Q", "KOSDAQ")]:
# KOSDAQ은 실계좌 시세 API 우선 사용 (모의투자 서버는 KOSDAQ 미지원)
api_to_use = real_market_api if (market_code == "Q" and real_market_api) else kis_api
try:
result = await api_to_use.get_volume_rank(market_code)
items = result.get("output") or []
async def _fetch_market_naver(
session: aiohttp.ClientSession, sosok: int, market_name: str
) -> list[tuple[str, str, str]]:
"""sosok=0: KOSPI, sosok=1: KOSDAQ"""
base_url = "https://finance.naver.com/sise/sise_market_sum.naver"
results: list[tuple[str, str, str]] = []
if not isinstance(items, list):
logger.warning(" %s: output이 리스트 아님 — 건너뜀", market_name)
continue
# 첫 페이지로 총 페이지 수 파악
async with session.get(
base_url, params={"sosok": sosok, "page": 1}, headers=HEADERS,
timeout=aiohttp.ClientTimeout(total=30)
) as resp:
if resp.status != 200:
logger.warning(" 네이버 %s HTTP %s", market_name, resp.status)
return []
html = await resp.text(encoding="euc-kr", errors="replace")
count = 0
for item in items:
ticker = item.get("mksc_shrn_iscd") or item.get("stck_shrn_iscd") or ""
name = item.get("hts_kor_isnm") or item.get("stck_kor_isnm") or ""
volume = int(item.get("acml_vol") or 0)
soup = BeautifulSoup(html, "html.parser")
# 총 페이지 수 추출
pager = soup.select("td.pgRR a")
if pager:
import re
m = re.search(r"page=(\d+)", pager[-1].get("href", ""))
total_pages = int(m.group(1)) if m else 1
else:
total_pages = 1
total_pages = min(total_pages, 30) # 최대 30페이지 (약 1,500종목)
if not ticker or not name:
# 첫 페이지 파싱
def _parse_page(s: BeautifulSoup) -> list[tuple[str, str, str]]:
rows = s.select("table.type_2 tr")
items = []
for row in rows:
a_tag = row.select_one("a.tltle")
if not a_tag:
continue
if any(k in name for k in SKIP_KEYWORDS):
name = a_tag.get_text(strip=True)
href = a_tag.get("href", "")
import re as _re
m2 = _re.search(r"code=(\d+)", href)
if not m2:
continue
code = m2.group(1)
if any(kw in name for kw in SKIP_KEYWORDS):
continue
items.append((code, name, market_name))
return items
stocks_with_vol.append((ticker, name, volume, market_name))
count += 1
results.extend(_parse_page(soup))
logger.info(" %s: %d개 종목 조회", market_name, count)
# 나머지 페이지 병렬 수집
sem = asyncio.Semaphore(5)
except Exception as exc:
logger.error("❌ %s 종목 조회 실패: %s", market_name, exc)
async def _fetch_page(page: int):
async with sem:
try:
async with session.get(
base_url, params={"sosok": sosok, "page": page},
headers=HEADERS, timeout=aiohttp.ClientTimeout(total=30)
) as r:
if r.status != 200:
return []
h = await r.text(encoding="euc-kr", errors="replace")
return _parse_page(BeautifulSoup(h, "html.parser"))
except Exception:
return []
stocks_with_vol.sort(key=lambda x: x[2], reverse=True)
top_with_mkt = stocks_with_vol[:limit]
top_stocks = [(code, name) for code, name, _, _mkt in top_with_mkt]
tasks = [_fetch_page(p) for p in range(2, total_pages + 1)]
pages_data = await asyncio.gather(*tasks)
for page_items in pages_data:
results.extend(page_items)
# 중복 제거
seen: set[str] = set()
unique = []
for item in results:
if item[0] not in seen:
seen.add(item[0])
unique.append(item)
logger.info(" %s: %d개 선정 (총 %d페이지)", market_name, len(unique), total_pages)
return unique
kospi_stocks: list[tuple[str, str, str]] = []
kosdaq_stocks: list[tuple[str, str, str]] = []
async with aiohttp.ClientSession() as session:
kospi_stocks = await _fetch_market_naver(session, 0, "KOSPI")
kosdaq_stocks = await _fetch_market_naver(session, 1, "KOSDAQ")
# 네이버 스크래핑 실패 시 KIS API로 폴백
naver_total = len(kospi_stocks) + len(kosdaq_stocks)
if naver_total < MIN_STOCK_COUNT:
logger.warning(
"⚠️ 네이버 금융 조회 부족(%d개) — KIS API로 폴백", naver_total
)
try:
from app.services.market_analyzer import get_all_tradable_stocks as _kis_stocks
kis_result = await _kis_stocks()
if len(kis_result) >= MIN_STOCK_COUNT:
logger.info("✅ KIS API 폴백 성공: %d개", len(kis_result))
return kis_result[:limit]
except Exception as _exc:
logger.error("❌ KIS API 폴백도 실패: %s", _exc)
raise RuntimeError(
f"종목 조회 실패: 네이버({naver_total}개) + KIS API 모두 불가. "
f"네트워크 및 API 설정을 확인하세요."
)
# KOSPI 60% / KOSDAQ 40% 비율로 합산
kospi_n = int(limit * 0.6)
kosdaq_n = limit - kospi_n
all_stocks = kospi_stocks[:kospi_n] + kosdaq_stocks[:kosdaq_n]
top_with_mkt = all_stocks[:limit]
top_stocks = [(code, name) for code, name, _ in top_with_mkt]
kospi_n = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSPI")
kosdaq_n = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSDAQ")
logger.info("✅ 선정 %d개 (KOSPI %d / KOSDAQ %d)", len(top_stocks), kospi_n, kosdaq_n)
if len(top_stocks) < MIN_STOCK_COUNT:
if settings.KIS_MODE == "virtual":
# 모의투자 서버는 KOSDAQ 거래량 순위 미지원 — 조회된 종목으로 진행
logger.warning(
"⚠️ 모의투자 서버 종목 수 제한: %d개 (기준 %d개). "
"KOSDAQ 거래량 순위 미지원으로 인한 정상 동작 — 배치 계속 진행합니다.",
len(top_stocks), MIN_STOCK_COUNT
)
else:
raise RuntimeError(
f"종목 조회 결과 너무 적음: {len(top_stocks)}개 (최소 {MIN_STOCK_COUNT}개 필요). "
"KIS API 오류 또는 장 휴장 여부를 확인하세요."
)
return top_stocks
@@ -189,12 +274,20 @@ async def analyze_one_stock(
) -> tuple[str, Optional[StockRecommendation]]:
"""종목 분석만 수행하고 결과 반환. DB 저장은 Streaming 저장기에서 처리.
DB(stock_daily_prices)에서 차트 데이터를 조회 — KIS API 호출 없음.
Returns: (status, result)
status: "ok" | "no_chart" | "price_filter" | "error"
"""
try:
logger.info("[%d/%d] 분석 중: %s (%s)", idx, total, stock_code, stock_name)
result = await analyze_stock_for_batch(stock_code, stock_name)
chart_data = await _fetch_db_chart(stock_code)
if not chart_data:
logger.info("샘 차트없음(DB) %s (%s)", stock_code, stock_name)
return "no_chart", None
result = await analyze_stock_for_batch_db(chart_data, stock_code, stock_name)
if not result:
logger.info("샘 차트없음 %s (%s)", stock_code, stock_name)
@@ -300,11 +393,35 @@ async def batch_analyze_all_stocks(limit: int = None):
logger.info("⚙️ 동시 처리: %d종목 | 벌크저장 단위: %d개 | API 세마포어: %dreq",
BATCH_CONCURRENCY, BULK_SAVE_CHUNK, settings.API_MAX_CONCURRENT_REQUESTS)
# Step 0: DB 데이터 사전 확인 (collect_krx_data.py 미실행 감지)
async with AsyncSessionLocal() as _db:
_db_check = await _db.execute(
sa_select(StockDailyPrice.trade_date)
.order_by(StockDailyPrice.trade_date.desc())
.limit(1)
)
_last_row = _db_check.scalar()
if _last_row is None:
logger.error(
"❌ stock_daily_prices 테이블이 비어있습니다. "
"먼저 collect_krx_data.py를 실행하여 데이터를 수집하세요."
)
raise RuntimeError("DB에 차트 데이터 없음 — collect_krx_data.py 선행 실행 필요")
from datetime import timedelta as _td
_days_old = (date.today() - _last_row).days
if _days_old > 3:
logger.warning(
"⚠️ DB 마지막 데이터: %s (%d일 전) — collect_krx_data.py --update 실행 권장",
_last_row, _days_old
)
else:
logger.info("✅ DB 최신 데이터: %s", _last_row)
# Step 1: 시장 레짐 판별
regime = await detect_and_log_market_regime()
# Step 2: 종목 목록 조회 (모듈 레벨 싱글톤 재사용, 불필요한 새 인스턴스 생성 금지)
all_stocks = await get_all_tradable_stocks(korea_investment_api, limit=500)
# Step 2: 종목 목록 조회 (KRX OpenAPI)
all_stocks = await get_all_tradable_stocks(limit=500)
filtered_stocks = all_stocks[:limit] if limit else all_stocks
total = len(filtered_stocks)
logger.info("📊 분석 대상: %d개 (시장 레짐: %s)", total, regime)
@@ -314,43 +431,41 @@ async def batch_analyze_all_stocks(limit: int = None):
queue: asyncio.Queue = asyncio.Queue(maxsize=BULK_SAVE_CHUNK * 3)
semaphore = asyncio.Semaphore(BATCH_CONCURRENCY)
# saver_task를 먼저 시작
saver_task = asyncio.create_task(_streaming_saver(queue, analysis_date))
async def analyze_one(i: int, code: str, name: str):
try:
async with semaphore:
status, result = await analyze_one_stock(i, total, code, name)
await queue.put((status, code, name, result))
# saver가 이미 종료(예외 등)됐으면 put 블로킹 방지
if not saver_task.done():
await queue.put((status, code, name, result))
else:
logger.warning("⚠️ saver 종료됨 — %s 저장 건너뜀", code)
except Exception as exc:
logger.error("❌ analyze_one unexpected error %s: %s", code, exc)
await queue.put(("error", code, name, None))
if not saver_task.done():
await queue.put(("error", code, name, None))
# 🔧 [fix] saver_task 예외 감지: asyncio.wait로 saver_task 비정상 종료 시 조기 감지
saver_task = asyncio.create_task(_streaming_saver(queue, analysis_date))
# 분석 태스크 전체 실행
analysis_gather = asyncio.gather(
# 분석 태스크 전체 실행 (완료까지 대기)
await asyncio.gather(
*[analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)],
return_exceptions=True,
)
# saver_task가 예외로 죽으면 analysis_gather도 조기 취소
done, pending = await asyncio.wait(
[asyncio.ensure_future(analysis_gather), saver_task],
return_when=asyncio.FIRST_EXCEPTION,
)
# saver_task 예외 발생 시 분석 태스크 취소 처리
for task in pending:
task.cancel()
try:
await task
except (asyncio.CancelledError, Exception):
pass
# sentinel 전송 → 저장기 종료 (정상 흐름에서도 항상 실행)
# 모든 분석 완료 후 sentinel 전송 → saver 정상 종료
# 기존 코드의 deadlock 원인: sentinel을 asyncio.wait() 이후에 보냈기 때문에
# FIRST_EXCEPTION 모드에서 예외 없으면 ALL_COMPLETED 동작 → saver가 sentinel을
# 기다리며 무한 대기 (deadlock)
if not saver_task.done():
await queue.put(None)
stats = await saver_task if not saver_task.cancelled() else {}
stats = {}
try:
stats = await saver_task
except Exception as exc:
logger.error("❌ saver_task 최종 오류: %s", exc)
ok_count = stats.get("ok", 0)
no_chart = stats.get("no_chart", 0)
@@ -375,7 +490,6 @@ async def batch_analyze_all_stocks(limit: int = None):
# 결과 요약
duration = (datetime.now(KST) - start_time).total_seconds()
api_stats = korea_investment_api.get_monitor_stats()
logger.info("=" * 60)
logger.info("✅ Batch analysis completed!")
@@ -386,9 +500,6 @@ async def batch_analyze_all_stocks(limit: int = None):
logger.info(" ├─ 오류: %d건", err_count)
logger.info("🔍 DB 스캐너 신호: %d개", len(scanner_tickers))
logger.info("⏱️ %.1f초 (%.1f분)", duration, duration / 60)
logger.info("📡 API 통계 — 호출:%d 오류:%d(%.1f%%) 평균:%.2f초",
api_stats['total_calls'], api_stats['total_errors'],
api_stats['error_rate'] * 100, api_stats['avg_duration'])
logger.info("=" * 60)