shuim
a264561576
- 스윙 체결가 확인 전 executed_price NULL + fill_sweep, 진입가 resolve(ActiveTrade↔Portfolio), 0체결/부분체결 매수 장부 원복 - DT 고아 감지(근거 한정)·0체결 매도 복원·stale 점검 확대, 시간손절 -0.3% 원복 - 폴백 손절 ATR 캐시, 거래량부족-차단 신호 매수 차단, 현금 5% 유보 재적용, MDD peak/reset 수정 - KIS 서킷브레이커 청산 우회·경보, db_worker 재큐잉 순서, heartbeat·health-watch 사이드카 - 실시간 WS 재구독·41건 우선순위·다건 프레임 분할, 차트 샘플 제거·당일봉 합성, WS 서브프로토콜 인증 - 프론트: 미관리 그룹·확인 모달·차트 실시간 반영·예수금 표시 정정 - 전문가 에이전트 6종 추가(.claude/agents) - 유닛 스위트 2667 passed (소켓 차단 하), tsc OK Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y83Za98XsNw9zuA4FTUytN
243 lines
10 KiB
Python
243 lines
10 KiB
Python
"""StockAnalysis DB 서비스 — 단일 진실 공급원 (Single Source of Truth)
|
||
|
||
StockAnalysis ORM 저장 로직을 중앙화하여 3곳의 중복 제거:
|
||
- batch_daily_analysis.py (bulk)
|
||
- batch_intraday_analysis.py (per-stock, rescore)
|
||
|
||
PostgreSQL UPSERT(ON CONFLICT DO UPDATE)를 사용하여
|
||
DELETE → INSERT 사이의 race condition 을 원자적으로 해결.
|
||
|
||
워커 패턴 (CLAUDE.md 3.2):
|
||
- session 을 직접 받는 `upsert_analysis` / `bulk_upsert_analysis` 는
|
||
scheduler / batch 에서 기존처럼 사용 가능 (시그니처 불변).
|
||
- 실시간 콜백·인트라데이 분석에서는 `queue_upsert_analysis` /
|
||
`queue_bulk_upsert_analysis` 를 사용하여 DB 락 경합을 방지.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import date
|
||
from typing import TYPE_CHECKING
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.database import AsyncSessionLocal
|
||
from app.core.timeutil import now_kst
|
||
from app.models.models import StockAnalysis
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
if TYPE_CHECKING:
|
||
from app.api.endpoints.recommendations import StockRecommendation
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 헬퍼
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
def passes_price_filter(result: "StockRecommendation") -> bool:
|
||
"""5,000원 이하 저가주 필터 (CLAUDE.md 5.4).
|
||
|
||
True → 통과, False → 필터됨.
|
||
"""
|
||
return not (result.current_price and result.current_price < 5000)
|
||
|
||
|
||
# UPSERT 시 갱신할 컬럼 집합 (PK·unique key·최초등록 시각 제외)
|
||
_UPSERT_UPDATE_COLS = frozenset({
|
||
"stock_name", "current_price", "change_rate", "volume",
|
||
"rsi", "ma5", "ma20", "ma60",
|
||
"stochastic_k", "stochastic_d",
|
||
"bollinger_upper", "bollinger_lower", "bollinger_middle",
|
||
"signals", "score",
|
||
"ml_score", "ml_confidence", "ml_signal",
|
||
"atr_14", "adx_14",
|
||
"updated_at",
|
||
})
|
||
|
||
|
||
def to_analysis_dict(
|
||
stock_code: str,
|
||
stock_name: str,
|
||
result: "StockRecommendation",
|
||
analysis_date: date,
|
||
) -> dict:
|
||
"""StockRecommendation → StockAnalysis INSERT dict 변환.
|
||
|
||
ORM 객체 대신 dict 를 반환하여 단건 UPSERT 와 bulk INSERT 에서 모두 재사용.
|
||
"""
|
||
return {
|
||
"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,
|
||
"updated_at": now_kst(),
|
||
}
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 내부 실행 함수 (db_worker.py 에서도 호출)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
async def _execute_upsert(session: AsyncSession, rows: list[dict]) -> None:
|
||
"""실제 PostgreSQL UPSERT 실행 — session 은 호출 측이 관리."""
|
||
if not rows:
|
||
return
|
||
stmt = pg_insert(StockAnalysis).values(rows)
|
||
stmt = stmt.on_conflict_do_update(
|
||
constraint="uq_stock_analysis_code_date",
|
||
set_={col: stmt.excluded[col] for col in _UPSERT_UPDATE_COLS},
|
||
)
|
||
await session.execute(stmt)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 기존 session 기반 함수 — 시그니처 불변 (CLAUDE.md 4.1)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
async def upsert_analysis(
|
||
session: AsyncSession,
|
||
stock_code: str,
|
||
stock_name: str,
|
||
result: "StockRecommendation",
|
||
analysis_date: date,
|
||
) -> None:
|
||
"""단건 UPSERT — session 직접 사용. 배치 스케줄러에서 그대로 호출 가능."""
|
||
row = to_analysis_dict(stock_code, stock_name, result, analysis_date)
|
||
await _execute_upsert(session, [row])
|
||
|
||
|
||
async def bulk_upsert_analysis(
|
||
session: AsyncSession,
|
||
items: list[tuple[str, str, "StockRecommendation"]],
|
||
analysis_date: date,
|
||
) -> int:
|
||
"""다건 UPSERT — session 직접 사용. 일봉 배치에서 그대로 호출 가능."""
|
||
if not items:
|
||
return 0
|
||
rows = [to_analysis_dict(code, name, r, analysis_date) for code, name, r in items]
|
||
await _execute_upsert(session, rows)
|
||
return len(rows)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# NEW: 워커 큐 기반 함수 — 실시간 콜백·인트라데이에서 사용 (CLAUDE.md 3.2)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
async def queue_upsert_analysis(
|
||
stock_code: str,
|
||
stock_name: str,
|
||
result: "StockRecommendation",
|
||
analysis_date: date,
|
||
) -> None:
|
||
"""단건 UPSERT 를 db_worker 큐에 위임.
|
||
|
||
실시간 스레드 / 인트라데이 분석에서 사용.
|
||
DB 세션을 직접 열지 않으므로 락 경합 없음.
|
||
"""
|
||
from app.services.db_worker import enqueue_upsert_analysis
|
||
row = to_analysis_dict(stock_code, stock_name, result, analysis_date)
|
||
await enqueue_upsert_analysis(row)
|
||
|
||
|
||
async def queue_bulk_upsert_analysis(
|
||
items: list[tuple[str, str, "StockRecommendation"]],
|
||
analysis_date: date,
|
||
) -> int:
|
||
"""다건 UPSERT 를 db_worker 큐에 위임.
|
||
|
||
배치 분석 결과를 비동기로 저장할 때 사용.
|
||
API 응답 지연 없이 큐에 넣고 즉시 반환.
|
||
Returns: 큐에 등록된 row 수
|
||
"""
|
||
if not items:
|
||
return 0
|
||
from app.services.db_worker import enqueue_bulk_upsert_analysis
|
||
rows = [to_analysis_dict(code, name, r, analysis_date) for code, name, r in items]
|
||
await enqueue_bulk_upsert_analysis(rows)
|
||
return len(rows)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# DB 스캐너 — 저장된 분석 결과로 신호 종목 필터링 (API 호출 없음)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
async def run_db_scanner(analysis_date: date) -> set[str]:
|
||
"""배치 DB 기반 코스닥·코스피 스캐너 (추가 API 호출 없음).
|
||
|
||
조건:
|
||
- RSI(14) ≤ 30 (과매도)
|
||
- 종가 ≤ BB 하단 × 1.015
|
||
- 시세 ≥ 5,000원
|
||
- ADX ≤ 30 (강한 추세 = 평균회귀 부적합)
|
||
- MA5 ≥ MA20 × 0.95 (강한 하락 추세 제외)
|
||
|
||
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,
|
||
StockAnalysis.ma5,
|
||
StockAnalysis.ma20,
|
||
).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
|
||
|
||
pct = (price - bb_lower) / bb_lower * 100
|
||
if pct > 1.5:
|
||
continue
|
||
|
||
adx = row.adx_14
|
||
if adx is not None and adx > 30:
|
||
continue
|
||
|
||
ma5, ma20 = row.ma5, row.ma20
|
||
if ma5 is not None and ma20 is not None and ma20 > 0:
|
||
if float(ma5) < float(ma20) * 0.95:
|
||
continue
|
||
|
||
signal_tickers.add(row.stock_code)
|
||
|
||
logger.info(
|
||
"🔍 DB 스캐너: %d개 신호 종목 (RSI≤30 + BB하단 + ADX≤30 + MA방향)",
|
||
len(signal_tickers),
|
||
)
|
||
return signal_tickers
|
||
|
||
except Exception as exc:
|
||
logger.error("❌ DB 스캐너 실패: %s", exc, exc_info=True)
|
||
return set()
|