fix: batch_intraday_analysis 리뷰 이슈 수정
- [🔴] logging.basicConfig를 __main__ 블록으로 이동 - [🟡] analyze_update_and_manage: get_current_score 세션 통합 (세션 인스턴스 감소) - [🟡] rescore_outside_candidates: bulk_upsert_analysis로 일괄 저장 - [🟠] retained 통계: AI 스킵 종목 명확히 분리
This commit is contained in:
@@ -7,7 +7,7 @@
|
||||
3. 분석 결과를 watchlist_history에 시계열로 누적 저장
|
||||
4. score 기준으로 분석종목 진입(≥60) / 제거(<20) 자동 관리
|
||||
|
||||
[최적화]
|
||||
[전략]
|
||||
- MCP AI 분석은 DB 점수가 MCP_ANALYSIS_MIN_SCORE(기본 70) 이상인 종목만 호출
|
||||
- 저점수 종목은 루트림 지표 계산만 수행하고 수동 score 유지
|
||||
|
||||
@@ -32,7 +32,12 @@ from app.core.database import AsyncSessionLocal # 🔧 공통 DB 모듈
|
||||
from app.core.market_schedule import is_market_open
|
||||
from app.models.models import StockAnalysis, WatchlistStock, WatchlistHistory
|
||||
from app.api.endpoints.recommendations import analyze_stock_for_batch
|
||||
from app.services.analysis_db_service import upsert_analysis, passes_price_filter, run_db_scanner # 🔧 단일화
|
||||
from app.services.analysis_db_service import (
|
||||
upsert_analysis,
|
||||
bulk_upsert_analysis,
|
||||
passes_price_filter,
|
||||
run_db_scanner,
|
||||
) # 🔧 단일화
|
||||
from app.services.watchlist import (
|
||||
WATCHLIST_MAX_SIZE,
|
||||
WATCHLIST_ENTRY_SCORE,
|
||||
@@ -40,10 +45,7 @@ from app.services.watchlist import (
|
||||
)
|
||||
import logging
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
# 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 동시 처리 종목 수 (KIS API 세마포어가 추가로 속도 제어)
|
||||
@@ -75,24 +77,6 @@ async def get_active_watchlist() -> list[tuple[str, str]]:
|
||||
return [(row.stock_code, row.stock_name) for row in rows]
|
||||
|
||||
|
||||
async def get_current_score(stock_code: str, analysis_date: date) -> int | None:
|
||||
"""최신 DB score 조회.
|
||||
|
||||
MCP AI 분석 전 score를 확인하여 MCP_ANALYSIS_MIN_SCORE 미만은 AI 호출을 스킵하기 위해 사용.
|
||||
"""
|
||||
async with AsyncSessionLocal() as session:
|
||||
result = await session.execute(
|
||||
select(StockAnalysis.score)
|
||||
.where(
|
||||
StockAnalysis.stock_code == stock_code,
|
||||
StockAnalysis.analysis_date == analysis_date,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
row = result.first()
|
||||
return row.score if row else None
|
||||
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# analyze_update_and_manage 분리: 분석 / 저장 / 관리
|
||||
# -------------------------------------------------------------------
|
||||
@@ -164,33 +148,47 @@ async def analyze_update_and_manage(
|
||||
) -> tuple[bool, int | None, bool]:
|
||||
"""종목 재분석 → stock_analysis 갱신 → watchlist_history 저장 → watchlist 관리.
|
||||
|
||||
[최적화] DB score < MCP_ANALYSIS_MIN_SCORE이면 AI 분석 스킵.
|
||||
- AI 호출 없이 기존 score 유지, watchlist 적스 체크만 수행
|
||||
[전략] DB score < MCP_ANALYSIS_MIN_SCORE이면 AI 분석 스킵.
|
||||
- AI 호출 없이 기존 score 유지, watchlist 진입 체크만 수행
|
||||
- score 개선 시코어는 다음 주기에 반영됨
|
||||
|
||||
[포인트] 세션을 한 번만 열어 score 조회 + 저장 + 관리를 한 트랜잭션에서 처리
|
||||
|
||||
Returns: (success, score, was_removed)
|
||||
was_removed: _manage_watchlist_membership이 실제로 비활성화 처리했으면 True
|
||||
"""
|
||||
try:
|
||||
# --- [1단계] DB에서 현재 score 조회 ---
|
||||
current_score = await get_current_score(stock_code, analysis_date)
|
||||
|
||||
# --- [2단계] score 미달 시 AI 분석 스킵 ---
|
||||
if current_score is not None and current_score < MCP_ANALYSIS_MIN_SCORE:
|
||||
logger.debug(
|
||||
"⏭️ AI 스킵 %s (%s): score=%s < %s",
|
||||
stock_code, stock_name, current_score, MCP_ANALYSIS_MIN_SCORE,
|
||||
# 🔧 [fix] score 조회 + 저장 + watchlist 관리를 단일 세션으로 통합
|
||||
# 기존에는 get_current_score에서 세션 1번 + 저장에서 세션 1번으로 최소 2번 열렸으나
|
||||
# 이제 score 조회부터 저장까지 단일 세션에서 모두 처리
|
||||
now = datetime.now(KST)
|
||||
async with AsyncSessionLocal() as session:
|
||||
# --- [1단계] DB에서 현재 score 조회 ---
|
||||
score_result = await session.execute(
|
||||
select(StockAnalysis.score)
|
||||
.where(
|
||||
StockAnalysis.stock_code == stock_code,
|
||||
StockAnalysis.analysis_date == analysis_date,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
# watchlist 제거 여부만 체크 (WATCHLIST_EXIT_SCORE 미만이면 제거)
|
||||
now = datetime.now(KST)
|
||||
async with AsyncSessionLocal() as session:
|
||||
score_row = score_result.first()
|
||||
current_score = score_row.score if score_row else None
|
||||
|
||||
# --- [2단계] score 미달 시 AI 분석 스킵 ---
|
||||
if current_score is not None and current_score < MCP_ANALYSIS_MIN_SCORE:
|
||||
logger.debug(
|
||||
"⏭️ AI 스킵 %s (%s): score=%s < %s",
|
||||
stock_code, stock_name, current_score, MCP_ANALYSIS_MIN_SCORE,
|
||||
)
|
||||
# watchlist 제거 여부만 체크 (WATCHLIST_EXIT_SCORE 미만이면 제거)
|
||||
was_removed = await _manage_watchlist_membership(
|
||||
session, stock_code, stock_name, current_score, now
|
||||
)
|
||||
await session.commit()
|
||||
return True, current_score, was_removed
|
||||
return True, current_score, was_removed
|
||||
|
||||
# --- [3단계] score 기준 이상 → AI 분석 실행 ---
|
||||
# --- [3단계] score 기준 이상 → AI 분석 실행 (세션 밖에서 실행) ---
|
||||
result = await analyze_stock_for_batch(stock_code, stock_name)
|
||||
if not result:
|
||||
logger.info("No result for %s (chart/price issue)", stock_code)
|
||||
@@ -208,14 +206,13 @@ async def analyze_update_and_manage(
|
||||
)
|
||||
.values(
|
||||
is_active=False,
|
||||
removed_at=datetime.now(KST),
|
||||
removed_at=now,
|
||||
removed_reason="price_filter",
|
||||
)
|
||||
)
|
||||
await session.commit()
|
||||
return False, None, True # 가격필터 제거도 was_removed=True
|
||||
|
||||
now = datetime.now(KST)
|
||||
score = result.score
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
@@ -241,7 +238,8 @@ async def rescore_outside_candidates(
|
||||
) -> int:
|
||||
"""워치리스트 외부 후보 종목의 score를 실시간 재분석으로 갱신.
|
||||
|
||||
[최적화] DB score < MCP_ANALYSIS_MIN_SCORE인 외부 후보는 AI 호출 없이 건너뜀.
|
||||
[전략] DB score < MCP_ANALYSIS_MIN_SCORE인 외부 후보는 AI 호출 없이 건너뛰.
|
||||
[최적화] 분석 결과를 모아서 bulk_upsert_analysis로 일괄 저장 (세션 최소화)
|
||||
|
||||
아침 배치에서 score 상위권이었던 종목이 장중 거래량·뉴스 급등으로
|
||||
score가 올라가도 DB에는 오전 값이 그대로 남아 있다.
|
||||
@@ -266,7 +264,7 @@ async def rescore_outside_candidates(
|
||||
candidates = result.all()
|
||||
|
||||
if not candidates:
|
||||
logger.info("외부 후보 종목 없음 — 재분석 건너뜀")
|
||||
logger.info("외부 후보 종목 없음 — 재분석 건너뚁")
|
||||
return 0
|
||||
|
||||
total = len(candidates)
|
||||
@@ -274,7 +272,14 @@ async def rescore_outside_candidates(
|
||||
|
||||
semaphore = asyncio.Semaphore(BATCH_CONCURRENCY)
|
||||
|
||||
async def rescore_one(i: int, code: str, name: str, db_score: int | None) -> bool:
|
||||
# 🔧 [fix] 분석 결과를 수집하여 bulk_upsert_analysis로 일괄 저장
|
||||
# 기존: rescore_one마다 세션 1개 open/close (30개 대상 시 30번)
|
||||
# 개선: 모아서 한 번에 bulk upsert (1번)
|
||||
|
||||
async def rescore_one(
|
||||
i: int, code: str, name: str, db_score: int | None
|
||||
) -> tuple[str, str, object] | None:
|
||||
"""Returns (code, name, result) on success, None otherwise."""
|
||||
async with semaphore:
|
||||
# DB score < MCP_ANALYSIS_MIN_SCORE 이면 AI 호출 스킵
|
||||
if db_score is not None and db_score < MCP_ANALYSIS_MIN_SCORE:
|
||||
@@ -282,33 +287,39 @@ async def rescore_outside_candidates(
|
||||
" [%d/%d] ⏭️ AI 스킵 %s: score=%s < %s",
|
||||
i, total, code, db_score, MCP_ANALYSIS_MIN_SCORE,
|
||||
)
|
||||
return False
|
||||
return None
|
||||
try:
|
||||
res = await analyze_stock_for_batch(code, name)
|
||||
if not res:
|
||||
return False
|
||||
# 저가주 필터 (CLAUDE.md 5절 가격 필터)
|
||||
return None
|
||||
if not passes_price_filter(res):
|
||||
logger.debug("Skip %s: price=%.0f원 < 5,000원", code, res.current_price)
|
||||
return False
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
await upsert_analysis(session, code, name, res, analysis_date)
|
||||
await session.commit()
|
||||
|
||||
return None
|
||||
logger.debug(" [%d/%d] %s (%s): score=%s", i, total, code, name, res.score)
|
||||
return True
|
||||
return (code, name, res)
|
||||
except Exception as e:
|
||||
logger.error("❌ 외부 후보 재분석 실패 %s: %s", code, e)
|
||||
return False
|
||||
return None
|
||||
|
||||
tasks = [
|
||||
rescore_one(i, code, name, score)
|
||||
for i, (code, name, score) in enumerate(candidates, 1)
|
||||
]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
success = sum(1 for r in results if r is True)
|
||||
skipped = sum(1 for r in results if r is False)
|
||||
raw_results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
# 유효한 결과만 필터막한 후 일괄 저장
|
||||
valid_items = [
|
||||
r for r in raw_results
|
||||
if isinstance(r, tuple) and len(r) == 3
|
||||
]
|
||||
success = len(valid_items)
|
||||
skipped = len(candidates) - success
|
||||
|
||||
if valid_items:
|
||||
async with AsyncSessionLocal() as session:
|
||||
await bulk_upsert_analysis(session, valid_items, analysis_date)
|
||||
await session.commit()
|
||||
|
||||
logger.info("✅ 외부 후보 재분석 완료: %d/%d개 갱신, %d개 AI 스킵", success, total, skipped)
|
||||
return success
|
||||
|
||||
@@ -413,18 +424,23 @@ async def main():
|
||||
]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
analyzed_ok = sum(1 for r in results if isinstance(r, tuple) and r[0] is True)
|
||||
removed = sum(1 for r in results if isinstance(r, tuple) and len(r) == 3 and r[2] is True)
|
||||
retained = analyzed_ok - sum(
|
||||
1 for r in results if isinstance(r, tuple) and len(r) == 3 and r[0] is True and r[2] is True
|
||||
analyzed_ok = sum(1 for r in results if isinstance(r, tuple) and r[0] is True)
|
||||
removed = sum(1 for r in results if isinstance(r, tuple) and len(r) == 3 and r[2] is True)
|
||||
|
||||
# 🔧 [fix] retained: AI스킵이지만 성공한 종목은 유지(제거 X)로 계산
|
||||
# was_removed=True인 네 가지 코드(success=True, removed=True)를 제외만 한 것
|
||||
retained = sum(
|
||||
1 for r in results
|
||||
if isinstance(r, tuple) and len(r) == 3 and r[0] is True and r[2] is False
|
||||
)
|
||||
fail = sum(1 for r in results if isinstance(r, tuple) and r[0] is False)
|
||||
error = sum(1 for r in results if isinstance(r, Exception))
|
||||
# AI 스킵 종목 수: success=True이고 score < MCP_ANALYSIS_MIN_SCORE인 것
|
||||
ai_skipped = sum(
|
||||
1 for r in results
|
||||
if isinstance(r, tuple) and r[0] is True
|
||||
and r[1] is not None and r[1] < MCP_ANALYSIS_MIN_SCORE
|
||||
)
|
||||
fail = sum(1 for r in results if isinstance(r, tuple) and r[0] is False)
|
||||
error = sum(1 for r in results if isinstance(r, Exception))
|
||||
|
||||
# 워치리스트 외부 후보 재분석 — score DB 최신화 (refresh_watchlist 전에 실행)
|
||||
await rescore_outside_candidates(analysis_date, current_codes)
|
||||
@@ -486,4 +502,9 @@ async def main():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 🔧 [fix] basicConfig를 __main__ 블록으로 이동
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
asyncio.run(main())
|
||||
|
||||
Reference in New Issue
Block a user