YoungWooSeo
62fccd6d6f
- [🔴] saver_task 예외 시 deadlock 방지: asyncio.wait로 saver_task 예외 감지 - [🔴] logging.basicConfig를 __main__ 블록으로 이동 (인프로세스 실행 시 충돌 방지) - [🟡] get_all_tradable_stocks: 최소 종목 수(50개) 미만 시 RuntimeError 발생 - [🟠] detect_and_log_market_regime: df=None일 때 len(df) 호출 안전하게 수정
400 lines
17 KiB
Python
400 lines
17 KiB
Python
"""
|
|
매일 아침 장 시작 전(9시) 실행되는 배치 분석 스크립트
|
|
|
|
전체 거래 종목을 분석하여 DB에 저장
|
|
1. 시장 레짐 판별 (KOSPI200 기준 BULL/BEAR/SIDEWAYS)
|
|
2. 차트 데이터 조회 + 기술적 지표 계산
|
|
3. 매매 신호 생성 + DB Streaming 저장 (분석 즉시 UPSERT)
|
|
4. 배치 완료 후 DB 스캐너 실행 (추가 API 호출 없이)
|
|
5. 워치리스트 초기화 — 스캐너 신호 종목 우선 배치
|
|
|
|
성능 개선 (이전 대비):
|
|
- BATCH_CONCURRENCY: 5 → 10 (API 세마포어 허용 범위 내)
|
|
- DB 트랜잭션: 종목당 2개(DELETE+INSERT) → UPSERT (원자적, race-condition 없음)
|
|
- Streaming: 전체 메모리 적재 → 분석 완료 즉시 BULK_SAVE_CHUNK(20)개 단위 저장
|
|
- DB 엔진: 스크립트 자체 생성 → app.core.database 공통 모듈 사용
|
|
|
|
실행: python batch_daily_analysis.py
|
|
Cron: 0 9 * * 1-5 cd /path/to/auto-stock/backend && python batch_daily_analysis.py
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from datetime import datetime, date, timedelta
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import pytz
|
|
KST = pytz.timezone('Asia/Seoul')
|
|
|
|
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
|
|
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
|
|
import logging
|
|
|
|
# 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출
|
|
# 인프로세스 스케줄러(main.py)로 실행될 때 로그 설정 충돌 방지
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# -------------------------------------------------------------------
|
|
# 성능 설정
|
|
# -------------------------------------------------------------------
|
|
BATCH_CONCURRENCY = 10 # API 세마포어 허용 범위 내 동시 처리 수
|
|
BULK_SAVE_CHUNK = 20 # Streaming 저장 단위 (큐에 20개 쌓이면 즉시 UPSERT)
|
|
|
|
# 종목 목록 조회 실패 시 배치를 중단할 최소 종목 수
|
|
MIN_STOCK_COUNT = 50
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# 1. 시장 레짐 판별 (KOSPI 기준)
|
|
# -------------------------------------------------------------------
|
|
|
|
async def detect_and_log_market_regime() -> str:
|
|
"""배치 시작 시 KOSPI 200일 데이터로 시장 레짐 판별.
|
|
|
|
BULL / BEAR / SIDEWAYS 중 하나를 리턴하며 배치 전체에 로깅된다.
|
|
실패 시 UNKNOWN을 리턴하여 배치를 차단하지 않는다.
|
|
"""
|
|
try:
|
|
from pykrx import stock as pykrx_stock
|
|
from app.services.mean_reversion_strategy import detect_market_regime
|
|
|
|
today = datetime.now(KST).strftime("%Y%m%d")
|
|
start = (datetime.now(KST) - timedelta(days=400)).strftime("%Y%m%d")
|
|
|
|
# KOSPI 지수 코드 1001
|
|
df = pykrx_stock.get_index_ohlcv_by_date(start, today, "1001")
|
|
|
|
# 🔧 [fix] df=None일 때 len(df) 직접 호출 방지
|
|
df_len = len(df) if df is not None else 0
|
|
if df_len < 205:
|
|
logger.warning("⚠️ KOSPI 데이터 부족(행 %d) — 레짐 판별 스킵", df_len)
|
|
return "UNKNOWN"
|
|
|
|
kospi_closes = df["종가"].tolist()
|
|
regime = detect_market_regime(kospi_closes)
|
|
|
|
regime_emoji = {"상승장": "📈 BULL", "하락장": "📉 BEAR",
|
|
"bull": "📈 BULL", "bear": "📉 BEAR"}.get(regime.value, "➡️ " + regime.value.upper())
|
|
logger.info("🌏 시장 레짐: %s (KOSPI %s일 기준)", regime_emoji, today)
|
|
return regime.value.upper()
|
|
|
|
except Exception as exc:
|
|
logger.error("❌ 레짐 판별 실패: %s", exc, exc_info=True)
|
|
return "UNKNOWN"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# 2. 종목 목록 조회
|
|
# -------------------------------------------------------------------
|
|
|
|
async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500) -> list[tuple[str, str]]:
|
|
"""시가총액 상위 종목 조회 (pykrx 사용)
|
|
|
|
ETF / 레버리지 / 스팭 등 특수종목 자동 제외.
|
|
KOSPI + KOSDAQ 합산 시가총액 상위 limit개 반환.
|
|
|
|
Raises:
|
|
RuntimeError: 조회 결과가 MIN_STOCK_COUNT(50) 미만인 경우 배치 중단
|
|
"""
|
|
logger.info("Fetching top %d stocks by market cap from pykrx...", limit)
|
|
try:
|
|
from pykrx import stock as pykrx_stock
|
|
|
|
today = datetime.now(KST).strftime("%Y%m%d")
|
|
|
|
SKIP_KEYWORDS = [
|
|
"ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG",
|
|
"KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX",
|
|
"선물", "스팭", "리츠", "SPAC",
|
|
]
|
|
|
|
stocks_with_cap = []
|
|
for market in ["KOSPI", "KOSDAQ"]:
|
|
try:
|
|
cap_df = pykrx_stock.get_market_cap_by_ticker(today, market=market)
|
|
logger.info(" %s: %d개 종목 시가총액 조회", market, len(cap_df))
|
|
for ticker in cap_df.index:
|
|
try:
|
|
name = pykrx_stock.get_market_ticker_name(ticker)
|
|
if any(k in name for k in SKIP_KEYWORDS):
|
|
continue
|
|
stocks_with_cap.append((ticker, name, cap_df.loc[ticker, "시가총액"], market))
|
|
except Exception:
|
|
continue
|
|
except Exception as exc:
|
|
logger.error("❌ %s 종목 조회 실패: %s", market, exc)
|
|
|
|
stocks_with_cap.sort(key=lambda x: x[2], reverse=True)
|
|
top_with_mkt = stocks_with_cap[:limit]
|
|
top_stocks = [(code, name) for code, name, _, _mkt 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)
|
|
|
|
# 🔧 [fix] 최소 종목 수 검증 — 너무 적으면 잘못된 데이터로 배치 실행 방지
|
|
if len(top_stocks) < MIN_STOCK_COUNT:
|
|
raise RuntimeError(
|
|
f"종목 조회 결과 너무 적음: {len(top_stocks)}개 (최소 {MIN_STOCK_COUNT}개 필요). "
|
|
"pykrx API 오류 또는 장 휴장 여부를 확인하세요."
|
|
)
|
|
|
|
return top_stocks
|
|
|
|
except RuntimeError:
|
|
raise # 최소 종목 수 미달 — 상위로 전파하여 배치 중단
|
|
except Exception as exc:
|
|
logger.error("❌ 종목 목록 조회 실패: %s", exc, exc_info=True)
|
|
return []
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# 3. 단일 종목 분석 (순수 분석만 — DB 저장 없음)
|
|
# -------------------------------------------------------------------
|
|
|
|
async def analyze_one_stock(
|
|
idx: int,
|
|
total: int,
|
|
stock_code: str,
|
|
stock_name: str,
|
|
) -> tuple[str, Optional[StockRecommendation]]:
|
|
"""종목 분석만 수행하고 결과 반환. DB 저장은 Streaming 저장기에서 처리.
|
|
|
|
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)
|
|
|
|
if not result:
|
|
logger.info("샘 차트없음 %s (%s)", stock_code, stock_name)
|
|
return "no_chart", None
|
|
|
|
# 5,000원 이하 저가주 제외 (CLAUDE.md 5절 가격 필터)
|
|
if not passes_price_filter(result):
|
|
logger.info("샘 가격필터 %s: %.0f원 < 5,000원", stock_code, result.current_price)
|
|
return "price_filter", None
|
|
|
|
ml_info = f", ml_score={result.ml_score}" if result.ml_score is not None else ""
|
|
logger.info("✅ 분석완료 %s (%s): score=%s%s", stock_code, stock_name, result.score, ml_info)
|
|
return "ok", result
|
|
|
|
except Exception as exc:
|
|
logger.error("❌ Failed %s: %s", stock_code, exc)
|
|
return "error", None
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# 4. Streaming 저장기 — 분석 완료 즉시 BULK_SAVE_CHUNK 단위 UPSERT
|
|
# -------------------------------------------------------------------
|
|
|
|
async def _streaming_saver(
|
|
queue: asyncio.Queue,
|
|
analysis_date: date,
|
|
) -> dict:
|
|
"""큐에서 분석 결과를 꺼내 BULK_SAVE_CHUNK(20)개마다 즉시 DB UPSERT.
|
|
|
|
전체 메모리 적재 없이 분석 완료 즉시 저장하여 메모리 사용 최소화.
|
|
|
|
Returns: stats dict {"ok", "no_chart", "price_filter", "error", "saved"}
|
|
"""
|
|
buffer: list[tuple[str, str, StockRecommendation]] = []
|
|
stats = {"ok": 0, "no_chart": 0, "price_filter": 0, "error": 0, "saved": 0}
|
|
|
|
try:
|
|
while True:
|
|
item = await queue.get()
|
|
if item is None: # sentinel: 분석 완료
|
|
break
|
|
|
|
status, code, name, result = item
|
|
stats[status] = stats.get(status, 0) + 1
|
|
|
|
if status == "ok" and result:
|
|
buffer.append((code, name, result))
|
|
|
|
# 버퍼가 청크 크기에 도달하면 즉시 저장
|
|
if len(buffer) >= BULK_SAVE_CHUNK:
|
|
async with AsyncSessionLocal() as session:
|
|
saved = await bulk_upsert_analysis(session, buffer, analysis_date)
|
|
await session.commit()
|
|
stats["saved"] += saved
|
|
logger.info("💾 DB 저장: %d건 누적", stats["saved"])
|
|
buffer.clear()
|
|
|
|
# 나머지 잔여분 저장
|
|
if buffer:
|
|
async with AsyncSessionLocal() as session:
|
|
saved = await bulk_upsert_analysis(session, buffer, analysis_date)
|
|
await session.commit()
|
|
stats["saved"] += saved
|
|
logger.info("💾 DB 저장(잔여): %d건 누적", stats["saved"])
|
|
|
|
except Exception as exc:
|
|
# 🔧 [fix] saver_task 예외 시 큐를 드레인하여 분석 태스크 deadlock 방지
|
|
logger.error("❌ _streaming_saver 예외 발생 — 큐 드레인 후 종료: %s", exc, exc_info=True)
|
|
# 큐에 남은 항목 모두 소비하여 analyze_one의 queue.put() 블로킹 해제
|
|
while not queue.empty():
|
|
try:
|
|
queue.get_nowait()
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
|
|
return stats
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# 6. 메인 배치
|
|
# -------------------------------------------------------------------
|
|
|
|
async def batch_analyze_all_stocks(limit: int = None):
|
|
"""전체 종목 배치 분석
|
|
|
|
실행 순서:
|
|
1. 시장 레짐 판별 (KOSPI 지수 기준)
|
|
2. 시가총액 상위 500개 종목 조회
|
|
3. 세마포어로 BATCH_CONCURRENCY(10)개씩 병렬 분석 (API 호출만)
|
|
4. Streaming: 분석 완료 즉시 BULK_SAVE_CHUNK(20)개 단위 DB UPSERT
|
|
5. DB 스캐너 실행 (추가 API 호출 없음)
|
|
6. 워치리스트 초기화 — 스캐너 신호 종목 우선
|
|
|
|
성능 개선 (기존 대비):
|
|
- DB 트랜잭션: DELETE+INSERT race-condition → 원자적 UPSERT
|
|
- 메모리: 전체 결과 적재 → Streaming(분석 즉시 저장)
|
|
- DB 엔진: 스크립트 자체 생성 → 공통 모듈 공유
|
|
"""
|
|
start_time = datetime.now(KST)
|
|
analysis_date = datetime.now(KST).date()
|
|
|
|
logger.info("🚀 Starting daily batch analysis for %s", analysis_date)
|
|
logger.info("⚙️ 동시 처리: %d종목 | 벌크저장 단위: %d개 | API 세마포어: %dreq",
|
|
BATCH_CONCURRENCY, BULK_SAVE_CHUNK, settings.API_MAX_CONCURRENT_REQUESTS)
|
|
|
|
# Step 1: 시장 레짐 판별
|
|
regime = await detect_and_log_market_regime()
|
|
|
|
# Step 2: 종목 목록 조회 (모듈 레벨 싱글톤 재사용, 불필요한 새 인스턴스 생성 금지)
|
|
all_stocks = await get_all_tradable_stocks(korea_investment_api, limit=500)
|
|
filtered_stocks = all_stocks[:limit] if limit else all_stocks
|
|
total = len(filtered_stocks)
|
|
logger.info("📊 분석 대상: %d개 (시장 레짐: %s)", total, regime)
|
|
|
|
# Step 3 + 4: 병렬 분석 & Streaming 저장
|
|
# 분석 태스크 → queue → _streaming_saver (동시 실행)
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=BULK_SAVE_CHUNK * 3)
|
|
semaphore = asyncio.Semaphore(BATCH_CONCURRENCY)
|
|
|
|
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))
|
|
except Exception as exc:
|
|
logger.error("❌ analyze_one unexpected error %s: %s", code, exc)
|
|
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(
|
|
*[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 전송 → 저장기 종료 (정상 흐름에서도 항상 실행)
|
|
if not saver_task.done():
|
|
await queue.put(None)
|
|
|
|
stats = await saver_task if not saver_task.cancelled() else {}
|
|
|
|
ok_count = stats.get("ok", 0)
|
|
no_chart = stats.get("no_chart", 0)
|
|
price_flt = stats.get("price_filter", 0)
|
|
err_count = stats.get("error", 0)
|
|
save_count = stats.get("saved", 0)
|
|
|
|
# Step 5: DB 스캐너 (배치 완료 후 추가 API 호출 없이)
|
|
scanner_tickers = await run_db_scanner(analysis_date)
|
|
|
|
# Step 6: 워치리스트 초기화 — 스캐너 신호 종목 우선
|
|
try:
|
|
from app.services.watchlist import initialize_watchlist
|
|
async with AsyncSessionLocal() as session:
|
|
count = await initialize_watchlist(
|
|
session, analysis_date, scanner_tickers=scanner_tickers
|
|
)
|
|
logger.info("📋 워치리스트 초기화 완료: %d개 (스캐너 신호 %d개 포함)",
|
|
count, len(scanner_tickers))
|
|
except Exception as exc:
|
|
logger.warning("⚠️ 워치리스트 초기화 실패 (non-critical): %s", exc)
|
|
|
|
# 결과 요약
|
|
duration = (datetime.now(KST) - start_time).total_seconds()
|
|
api_stats = korea_investment_api.get_monitor_stats()
|
|
|
|
logger.info("=" * 60)
|
|
logger.info("✅ Batch analysis completed!")
|
|
logger.info("🌏 시장 레짐: %s", regime)
|
|
logger.info("📊 성공: %d / 전체: %d | DB저장: %d건", ok_count, total, save_count)
|
|
logger.info(" ├─ 차트없음: %d건", no_chart)
|
|
logger.info(" ├─ 가격필터: %d건", price_flt)
|
|
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)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Entry Point
|
|
# -------------------------------------------------------------------
|
|
|
|
async def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description='Daily Stock Analysis Batch')
|
|
parser.add_argument('--limit', type=int, help='Limit number of stocks (for testing)')
|
|
parser.add_argument('--test', action='store_true', help='Test mode (10 stocks)')
|
|
args = parser.parse_args()
|
|
|
|
if args.test:
|
|
logger.info("🧪 TEST MODE: 10개 종목")
|
|
await batch_analyze_all_stocks(limit=10)
|
|
elif args.limit:
|
|
await batch_analyze_all_stocks(limit=args.limit)
|
|
else:
|
|
await batch_analyze_all_stocks()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 🔧 [fix] basicConfig를 __main__ 블록으로 이동
|
|
# 인프로세스 스케줄러(main.py lifespan)에서 import될 때 로그 설정 충돌 방지
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
asyncio.run(main())
|