feat: MCP AI 분석을 score >= MCP_ANALYSIS_MIN_SCORE(70) 종목만 호출하도록 최적화

This commit is contained in:
2026-02-26 11:41:57 +09:00
parent 8f2262580f
commit de8645b3bf

View File

@@ -7,6 +7,10 @@
3. 분석 결과를 watchlist_history에 시계열로 누적 저장 3. 분석 결과를 watchlist_history에 시계열로 누적 저장
4. score 기준으로 분석종목 진입(≥60) / 제거(<20) 자동 관리 4. score 기준으로 분석종목 진입(≥60) / 제거(<20) 자동 관리
[최적화]
- MCP AI 분석은 DB 점수가 MCP_ANALYSIS_MIN_SCORE(기본 70) 이상인 종목만 호출
- 저점수 종목은 루트림 지표 계산만 수행하고 수동 score 유지
실행: python batch_intraday_analysis.py 실행: python batch_intraday_analysis.py
Cron: */10 9-15 * * 1-5 docker exec autostock-backend python /app/batch_intraday_analysis.py >> /logs/batch_intraday.log 2>&1 Cron: */10 9-15 * * 1-5 docker exec autostock-backend python /app/batch_intraday_analysis.py >> /logs/batch_intraday.log 2>&1
""" """
@@ -48,6 +52,9 @@ BATCH_CONCURRENCY = 5
# 워치리스트 외부 후보 재분석 상한 (10분 배치 내에 처리 가능한 수준) # 워치리스트 외부 후보 재분석 상한 (10분 배치 내에 처리 가능한 수준)
OUTSIDE_RESCORE_LIMIT = 30 OUTSIDE_RESCORE_LIMIT = 30
# MCP AI 분석 호출 최소 점수 (settings에서 읽음)
MCP_ANALYSIS_MIN_SCORE: int = settings.MCP_ANALYSIS_MIN_SCORE
async def get_active_watchlist() -> list[tuple[str, str]]: async def get_active_watchlist() -> list[tuple[str, str]]:
"""현재 활성 분석종목 목록 조회.""" """현재 활성 분석종목 목록 조회."""
@@ -68,6 +75,24 @@ async def get_active_watchlist() -> list[tuple[str, str]]:
return [(row.stock_code, row.stock_name) for row in rows] 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 분리: 분석 / 저장 / 관리 # analyze_update_and_manage 분리: 분석 / 저장 / 관리
# ------------------------------------------------------------------- # -------------------------------------------------------------------
@@ -139,10 +164,33 @@ async def analyze_update_and_manage(
) -> tuple[bool, int | None, bool]: ) -> tuple[bool, int | None, bool]:
"""종목 재분석 → stock_analysis 갱신 → watchlist_history 저장 → watchlist 관리. """종목 재분석 → stock_analysis 갱신 → watchlist_history 저장 → watchlist 관리.
[최적화] DB score < MCP_ANALYSIS_MIN_SCORE이면 AI 분석 스킵.
- AI 호출 없이 기존 score 유지, watchlist 적스 체크만 수행
- score 개선 시코어는 다음 주기에 반영됨
Returns: (success, score, was_removed) Returns: (success, score, was_removed)
was_removed: _manage_watchlist_membership이 실제로 비활성화 처리했으면 True was_removed: _manage_watchlist_membership이 실제로 비활성화 처리했으면 True
""" """
try: 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,
)
# watchlist 제거 여부만 체크 (WATCHLIST_EXIT_SCORE 미만이면 제거)
now = datetime.now(KST)
async with AsyncSessionLocal() as session:
was_removed = await _manage_watchlist_membership(
session, stock_code, stock_name, current_score, now
)
await session.commit()
return True, current_score, was_removed
# --- [3단계] score 기준 이상 → AI 분석 실행 ---
result = await analyze_stock_for_batch(stock_code, stock_name) result = await analyze_stock_for_batch(stock_code, stock_name)
if not result: if not result:
logger.info("No result for %s (chart/price issue)", stock_code) logger.info("No result for %s (chart/price issue)", stock_code)
@@ -193,6 +241,8 @@ async def rescore_outside_candidates(
) -> int: ) -> int:
"""워치리스트 외부 후보 종목의 score를 실시간 재분석으로 갱신. """워치리스트 외부 후보 종목의 score를 실시간 재분석으로 갱신.
[최적화] DB score < MCP_ANALYSIS_MIN_SCORE인 외부 후보는 AI 호출 없이 건너뜀.
아침 배치에서 score 상위권이었던 종목이 장중 거래량·뉴스 급등으로 아침 배치에서 score 상위권이었던 종목이 장중 거래량·뉴스 급등으로
score가 올라가도 DB에는 오전 값이 그대로 남아 있다. score가 올라가도 DB에는 오전 값이 그대로 남아 있다.
이를 포착하기 위해 워치리스트 밖 상위 OUTSIDE_RESCORE_LIMIT개를 이를 포착하기 위해 워치리스트 밖 상위 OUTSIDE_RESCORE_LIMIT개를
@@ -205,7 +255,7 @@ async def rescore_outside_candidates(
async with AsyncSessionLocal() as session: async with AsyncSessionLocal() as session:
exclude = list(current_codes) if current_codes else ["__none__"] exclude = list(current_codes) if current_codes else ["__none__"]
result = await session.execute( result = await session.execute(
select(StockAnalysis.stock_code, StockAnalysis.stock_name) select(StockAnalysis.stock_code, StockAnalysis.stock_name, StockAnalysis.score)
.where( .where(
StockAnalysis.analysis_date == analysis_date, StockAnalysis.analysis_date == analysis_date,
StockAnalysis.stock_code.notin_(exclude), StockAnalysis.stock_code.notin_(exclude),
@@ -224,8 +274,15 @@ async def rescore_outside_candidates(
semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) semaphore = asyncio.Semaphore(BATCH_CONCURRENCY)
async def rescore_one(i: int, code: str, name: str) -> bool: async def rescore_one(i: int, code: str, name: str, db_score: int | None) -> bool:
async with semaphore: async with semaphore:
# DB score < MCP_ANALYSIS_MIN_SCORE 이면 AI 호출 스킵
if db_score is not None and db_score < MCP_ANALYSIS_MIN_SCORE:
logger.debug(
" [%d/%d] ⏭️ AI 스킵 %s: score=%s < %s",
i, total, code, db_score, MCP_ANALYSIS_MIN_SCORE,
)
return False
try: try:
res = await analyze_stock_for_batch(code, name) res = await analyze_stock_for_batch(code, name)
if not res: if not res:
@@ -246,12 +303,13 @@ async def rescore_outside_candidates(
return False return False
tasks = [ tasks = [
rescore_one(i, code, name) rescore_one(i, code, name, score)
for i, (code, name) in enumerate(candidates, 1) for i, (code, name, score) in enumerate(candidates, 1)
] ]
results = await asyncio.gather(*tasks, return_exceptions=True) results = await asyncio.gather(*tasks, return_exceptions=True)
success = sum(1 for r in results if r is True) success = sum(1 for r in results if r is True)
logger.info("✅ 외부 후보 재분석 완료: %d/%d개 갱신", success, total) skipped = sum(1 for r in results if r is False)
logger.info("✅ 외부 후보 재분석 완료: %d/%d개 갱신, %d개 AI 스킵", success, total, skipped)
return success return success
@@ -328,8 +386,8 @@ async def main():
start_time = datetime.now(KST) start_time = datetime.now(KST)
analysis_date = datetime.now(KST).date() analysis_date = datetime.now(KST).date()
logger.info( logger.info(
"⚡ [Intraday] %s 시작 — 분석종목 재분석 (진입≥%d, 제거<%d)", "⚡ [Intraday] %s 시작 — 분석종목 재분석 (진입≥%d, 제거<%d, AI호출≥%d점)",
start_time.strftime("%H:%M:%S"), WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, start_time.strftime("%H:%M:%S"), WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, MCP_ANALYSIS_MIN_SCORE,
) )
# 활성 분석종목 조회 # 활성 분석종목 조회
@@ -362,6 +420,11 @@ async def main():
) )
fail = sum(1 for r in results if isinstance(r, tuple) and r[0] 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)) error = sum(1 for r in results if isinstance(r, Exception))
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
)
# 워치리스트 외부 후보 재분석 — score DB 최신화 (refresh_watchlist 전에 실행) # 워치리스트 외부 후보 재분석 — score DB 최신화 (refresh_watchlist 전에 실행)
await rescore_outside_candidates(analysis_date, current_codes) await rescore_outside_candidates(analysis_date, current_codes)
@@ -416,8 +479,9 @@ async def main():
duration = (datetime.now(KST) - start_time).total_seconds() duration = (datetime.now(KST) - start_time).total_seconds()
logger.info( logger.info(
"✅ [Intraday] 완료: %d/%d 유지, %d 제거, +%d 신규(score), +%d 신규(스캐너), " "✅ [Intraday] 완료: %d/%d 유지, %d 제거, +%d 신규(score), +%d 신규(스캐너), "
"%d 실패, %d 오류 — %.1f초", "%d AI스킵(score<%d), %d 실패, %d 오류 — %.1f초",
retained, total, removed, added, added_scanner, fail, error, duration, retained, total, removed, added, added_scanner,
ai_skipped, MCP_ANALYSIS_MIN_SCORE, fail, error, duration,
) )