YoungWooSeo
015c88eb25
- BATCH_CONCURRENCY 5 → 10 (동시 분석 종목 수) - 개별 commit(종목당 2 트랜잭션) → bulk_save_results(20개당 1 트랜잭션) 500종목 기준 DB 트랜잭션: 1,000회 → 25회 (40× 감소) - analyze_and_save_stock 분리: analyze_one_stock(분석) + bulk_save_results(저장) - _to_analysis_model 헬퍼 함수 추가 (StockRecommendation → ORM 변환)
452 lines
18 KiB
Python
452 lines
18 KiB
Python
"""
|
||
매일 아침 장 시작 전(9시) 실행되는 배치 분석 스크립트
|
||
|
||
전체 거래 종목을 분석하여 DB에 저장
|
||
1. 시장 레짐 판별 (KOSPI200 기준 BULL/BEAR/SIDEWAYS)
|
||
2. 차트 데이터 조회 + 기술적 지표 계산
|
||
3. 매매 신호 생성 + DB 일괄 저장 (BULK INSERT)
|
||
4. 배치 완료 후 DB 스캐너 실행 (추가 API 호출 없이)
|
||
5. 워치리스트 초기화 — 스캐너 신호 종목 우선 배치
|
||
|
||
성능 개선 (이전 대비):
|
||
- BATCH_CONCURRENCY: 5 → 10 (API 세마포어 허용 범위 내)
|
||
- DB 트랜잭션: 종목당 2개(DELETE+INSERT) → BULK_SAVE_CHUNK(20개)당 1개
|
||
- 500종목 기준: 1,000 트랜잭션 → 25 트랜잭션 (40× 감소)
|
||
|
||
실행: 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
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
|
||
from sqlalchemy import delete, select
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
from app.core.config import settings
|
||
from app.models.models import StockAnalysis
|
||
from app.services.korea_investment_api import KoreaInvestmentAPI, korea_investment_api
|
||
from app.api.endpoints.recommendations import analyze_stock_for_batch, StockRecommendation
|
||
import logging
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# -------------------------------------------------------------------
|
||
# DB 연결
|
||
# -------------------------------------------------------------------
|
||
database_url = settings.DATABASE_URL
|
||
if database_url.startswith("postgresql://"):
|
||
database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
||
|
||
engine = create_async_engine(
|
||
database_url,
|
||
echo=False,
|
||
pool_pre_ping=True,
|
||
pool_size=10,
|
||
max_overflow=20
|
||
)
|
||
|
||
AsyncSessionLocal = sessionmaker(
|
||
engine,
|
||
class_=AsyncSession,
|
||
expire_on_commit=False
|
||
)
|
||
|
||
# -------------------------------------------------------------------
|
||
# 성능 설정
|
||
# -------------------------------------------------------------------
|
||
BATCH_CONCURRENCY = 10 # ↑ 5→10: API 세마포어 허용 범위 내 동시 처리 수
|
||
BULK_SAVE_CHUNK = 20 # DB 일괄 저장 단위 (트랜잭션 수 = 총종목 / BULK_SAVE_CHUNK)
|
||
|
||
|
||
# -------------------------------------------------------------------
|
||
# 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().strftime("%Y%m%d")
|
||
start = (datetime.now() - timedelta(days=400)).strftime("%Y%m%d")
|
||
|
||
# KOSPI 지수 코드 1001
|
||
df = pykrx_stock.get_index_ohlcv_by_date(start, today, "1001")
|
||
if df is None or len(df) < 205:
|
||
logger.warning("⚠️ KOSPI 데이터 부족(열 %d) — 레짐 판별 스킵", 0 if df is None else len(df))
|
||
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개 반환.
|
||
"""
|
||
logger.info("Fetching top %d stocks by market cap from pykrx...", limit)
|
||
try:
|
||
from pykrx import stock as pykrx_stock
|
||
|
||
today = datetime.now().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, "시가총액"]))
|
||
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_stocks = [(code, name) for code, name, _ in stocks_with_cap[:limit]]
|
||
|
||
kospi_n = sum(1 for c, _ in top_stocks if c.startswith(('0','1','2','3','4','5')))
|
||
kosdaq_n = len(top_stocks) - kospi_n
|
||
logger.info("✅ 선정 %d개 (KOSPI %d / KOSDAQ %d)", len(top_stocks), kospi_n, kosdaq_n)
|
||
return top_stocks
|
||
|
||
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 저장은 bulk_save_results에서 일괄 처리.
|
||
|
||
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 6.4)
|
||
if result.current_price and result.current_price < 5000:
|
||
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. StockRecommendation → StockAnalysis ORM 변환 헬퍼
|
||
# -------------------------------------------------------------------
|
||
|
||
def _to_analysis_model(
|
||
stock_code: str,
|
||
stock_name: str,
|
||
result: StockRecommendation,
|
||
analysis_date: date,
|
||
) -> StockAnalysis:
|
||
"""StockRecommendation Pydantic 모델 → StockAnalysis ORM 객체 변환"""
|
||
return StockAnalysis(
|
||
stock_code=stock_code,
|
||
stock_name=stock_name,
|
||
analysis_date=analysis_date,
|
||
current_price=result.current_price,
|
||
change_rate=result.change_rate,
|
||
volume=result.volume,
|
||
rsi=result.rsi,
|
||
ma5=result.ma5,
|
||
ma20=result.ma20,
|
||
ma60=result.ma60,
|
||
stochastic_k=result.stochastic_k,
|
||
stochastic_d=result.stochastic_d,
|
||
bollinger_upper=result.bollinger_upper,
|
||
bollinger_lower=result.bollinger_lower,
|
||
bollinger_middle=result.bollinger_middle,
|
||
signals=result.signals,
|
||
score=result.score,
|
||
ml_score=result.ml_score,
|
||
ml_confidence=result.ml_confidence,
|
||
ml_signal=result.ml_signal,
|
||
atr_14=result.atr_14,
|
||
adx_14=result.adx_14,
|
||
)
|
||
|
||
|
||
# -------------------------------------------------------------------
|
||
# 5. Bulk DB 저장 (BULK_SAVE_CHUNK 개씩 한 트랜잭션)
|
||
# -------------------------------------------------------------------
|
||
|
||
async def bulk_save_results(
|
||
items: list[tuple[str, str, StockRecommendation]],
|
||
analysis_date: date,
|
||
) -> int:
|
||
"""분석 결과를 한 트랜잭션으로 일괄 저장.
|
||
|
||
기존 코드: 종목당 DELETE + INSERT → 2 트랜잭션 × 500종목 = 1,000 트랜잭션
|
||
개선 코드: IN 쿼리 DELETE 1회 + add_all INSERT 1회 = 2 트랜잭션 × 25청크 = 50 트랜잭션
|
||
|
||
items: [(stock_code, stock_name, result), ...]
|
||
Returns: 저장된 건수
|
||
"""
|
||
if not items:
|
||
return 0
|
||
|
||
codes = [code for code, _, _ in items]
|
||
|
||
async with AsyncSessionLocal() as session:
|
||
# 기존 당일 데이터 일괄 삭제 (IN 쿼리 1회)
|
||
await session.execute(
|
||
delete(StockAnalysis).where(
|
||
StockAnalysis.stock_code.in_(codes),
|
||
StockAnalysis.analysis_date == analysis_date,
|
||
)
|
||
)
|
||
# 일괄 삽입 (add_all → 단일 INSERT 배치)
|
||
objs = [_to_analysis_model(code, name, r, analysis_date) for code, name, r in items]
|
||
session.add_all(objs)
|
||
await session.commit()
|
||
|
||
return len(objs)
|
||
|
||
|
||
# -------------------------------------------------------------------
|
||
# 6. DB 스캐너 — 배치 완료 후 추가 API 호출 없이 신호 종목 추출
|
||
# -------------------------------------------------------------------
|
||
|
||
async def run_db_scanner(analysis_date: date) -> set[str]:
|
||
"""배치 DB 기반 스캐너 — KosdaqScanner와 동일한 기준을 단일일 데이터로 적용.
|
||
|
||
저장된 RSI / 볼린저 / 거래량 데이터만 사용하므로 추가 API 호출 없음.
|
||
|
||
조건 (성능 중심 필터):
|
||
- RSI(14) ≤ 30 (과매도)
|
||
- 종가 ≤ BB 하단 × 1.015 (BB 하단 근처)
|
||
- 시세 ≥ 5,000원
|
||
- ADX ≤ 30 (지나치게 강한 추세는 평균회귀 부적합)
|
||
|
||
Returns: 신호 종목 코드 set
|
||
"""
|
||
try:
|
||
async with AsyncSessionLocal() as session:
|
||
result = await session.execute(
|
||
select(
|
||
StockAnalysis.stock_code,
|
||
StockAnalysis.current_price,
|
||
StockAnalysis.rsi,
|
||
StockAnalysis.bollinger_lower,
|
||
StockAnalysis.adx_14,
|
||
).where(
|
||
StockAnalysis.analysis_date == analysis_date,
|
||
StockAnalysis.current_price >= 5000,
|
||
StockAnalysis.rsi.isnot(None),
|
||
StockAnalysis.rsi <= 30,
|
||
StockAnalysis.bollinger_lower.isnot(None),
|
||
)
|
||
)
|
||
rows = result.all()
|
||
|
||
signal_tickers: set[str] = set()
|
||
for row in rows:
|
||
price, bb_lower = row.current_price, row.bollinger_lower
|
||
if not bb_lower or bb_lower == 0:
|
||
continue
|
||
|
||
# BB 하단선 근처 확인 (-inf ~ +1.5%)
|
||
pct = (price - bb_lower) / bb_lower * 100
|
||
if pct > 1.5:
|
||
continue
|
||
|
||
# 구조적 추세장(ADX > 30)는 평균회귀에 부적합
|
||
adx = row.adx_14
|
||
if adx is not None and adx > 30:
|
||
continue
|
||
|
||
signal_tickers.add(row.stock_code)
|
||
|
||
logger.info("🔍 DB 스캐너: %d개 신호 종목 (RSI≤30 + BB하단 + ADX≤30)", len(signal_tickers))
|
||
return signal_tickers
|
||
|
||
except Exception as exc:
|
||
logger.error("❌ DB 스캐너 실패: %s", exc, exc_info=True)
|
||
return set()
|
||
|
||
|
||
# -------------------------------------------------------------------
|
||
# 7. 메인 배치
|
||
# -------------------------------------------------------------------
|
||
|
||
async def batch_analyze_all_stocks(limit: int = None):
|
||
"""전체 종목 배치 분석
|
||
|
||
실행 순서:
|
||
1. 시장 레짐 판별 (KOSPI 지수 기준)
|
||
2. 시가총액 상위 500개 종목 조회
|
||
3. 세마포어로 BATCH_CONCURRENCY(10)개씩 병렬 분석 (API 호출만)
|
||
4. 분석 완료 후 BULK_SAVE_CHUNK(20)개씩 일괄 DB 저장
|
||
5. DB 스캐너 실행 (추가 API 호출 없음)
|
||
6. 워치리스트 초기화 — 스캐너 신호 종목 우선
|
||
|
||
성능 개선 (기존 대비):
|
||
- DB 트랜잭션: 1,000회 → 50회 (20× 감소)
|
||
- 동시 처리: 5 → 10 종목
|
||
"""
|
||
start_time = datetime.now()
|
||
analysis_date = date.today()
|
||
|
||
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: 종목 목록 조회
|
||
kis_api = KoreaInvestmentAPI()
|
||
all_stocks = await get_all_tradable_stocks(kis_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: 병렬 분석 — 세마포어로 BATCH_CONCURRENCY 제한
|
||
# asyncio 단일 스레드이므로 results 리스트 append는 thread-safe
|
||
semaphore = asyncio.Semaphore(BATCH_CONCURRENCY)
|
||
all_results: list[tuple[str, str, str, Optional[StockRecommendation]]] = []
|
||
|
||
async def analyze_one(i: int, code: str, name: str):
|
||
async with semaphore:
|
||
status, result = await analyze_one_stock(i, total, code, name)
|
||
all_results.append((status, code, name, result))
|
||
|
||
tasks = [analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)]
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
# Step 4: 통계 집계
|
||
ok_items = [(code, name, r) for status, code, name, r in all_results if status == "ok" and r]
|
||
ok_count = len(ok_items)
|
||
no_chart = sum(1 for s, _, _, _ in all_results if s == "no_chart")
|
||
price_flt = sum(1 for s, _, _, _ in all_results if s == "price_filter")
|
||
err_count = sum(1 for s, _, _, _ in all_results if s in ("error",) or isinstance(s, Exception))
|
||
|
||
# Step 5: BULK_SAVE_CHUNK 단위 일괄 DB 저장
|
||
save_count = 0
|
||
for chunk_start in range(0, len(ok_items), BULK_SAVE_CHUNK):
|
||
chunk = ok_items[chunk_start: chunk_start + BULK_SAVE_CHUNK]
|
||
saved = await bulk_save_results(chunk, analysis_date)
|
||
save_count += saved
|
||
logger.info("💾 DB 저장: %d/%d건 완료", save_count, ok_count)
|
||
|
||
# Step 6: DB 스캐너 (배치 완료 후 추가 API 호출 없이)
|
||
scanner_tickers = await run_db_scanner(analysis_date)
|
||
|
||
# Step 7: 워치리스트 초기화 — 스캐너 신호 종목 우선
|
||
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() - start_time).total_seconds()
|
||
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초",
|
||
stats['total_calls'], stats['total_errors'],
|
||
stats['error_rate'] * 100, 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__":
|
||
asyncio.run(main())
|