From 883f87753ce54aa831376eff309d92ef8a620cbc Mon Sep 17 00:00:00 2001 From: shuim Date: Tue, 24 Feb 2026 11:38:35 +0900 Subject: [PATCH] =?UTF-8?q?=EA=B5=AC=EC=A1=B0=EC=A0=81=EC=9D=B8=20?= =?UTF-8?q?=EB=B6=80=EB=B6=84=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/endpoints/day_trading.py | 80 ++- backend/app/api/endpoints/journals.py | 13 +- backend/app/api/endpoints/market_data.py | 5 +- backend/app/api/endpoints/mcp.py | 2 +- backend/app/api/endpoints/recommendations.py | 461 +++++++----------- backend/app/core/config.py | 2 + backend/app/main.py | 32 +- backend/app/models/models.py | 7 +- backend/app/services/analysis_db_service.py | 196 ++++++++ backend/batch_daily_analysis.py | 285 ++++------- backend/batch_intraday_analysis.py | 265 +++++----- backend/migrations/fix_analysis_date_type.sql | 66 +++ docker-compose.yml | 2 +- docker/init-db.sql | 4 + frontend/src/pages/Dashboard.tsx | 41 +- 15 files changed, 789 insertions(+), 672 deletions(-) create mode 100644 backend/app/services/analysis_db_service.py create mode 100644 backend/migrations/fix_analysis_date_type.sql diff --git a/backend/app/api/endpoints/day_trading.py b/backend/app/api/endpoints/day_trading.py index f1dba70..fdbe2cb 100644 --- a/backend/app/api/endpoints/day_trading.py +++ b/backend/app/api/endpoints/day_trading.py @@ -84,6 +84,12 @@ _orb_cache: Dict[str, OpeningRange] = {} # 일일 실현 손익 (원) _daily_pnl: float = 0.0 +# 마지막으로 조회된 총자산 (is_daily_loss_exceeded 계산용 캐시) +_last_total_assets: float = 0.0 + +# 당일 청산 종목 (재진입 방지 — 청산 후 당일 동일 종목 재매수 차단) +_closed_today: set = set() + # 마지막 초기화 날짜 _last_reset_date: Optional[date] = None @@ -93,11 +99,13 @@ _trade_lock = asyncio.Lock() def _reset_if_new_day() -> None: """날짜가 바뀌면 일일 상태 초기화""" - global _daily_pnl, _orb_cache, _last_reset_date + global _daily_pnl, _orb_cache, _last_reset_date, _last_total_assets, _closed_today today = _today_kst() if _last_reset_date != today: _daily_pnl = 0.0 _orb_cache = {} + _last_total_assets = 0.0 + _closed_today = set() _last_reset_date = today logger.info(f"📅 데이트레이딩 상태 일일 초기화: {today}") @@ -169,7 +177,7 @@ async def _place_buy_order( async def _place_sell_order( stock_code: str, quantity: int, - price: int = 0, # 0 = 시장가 + price: int = 0, # 0 = 시장가 (None으로 변환) ) -> bool: """매도 주문 실행""" try: @@ -177,7 +185,7 @@ async def _place_sell_order( stock_code=stock_code, order_type="SELL", quantity=quantity, - price=price, + price=price if price > 0 else None, # 0이면 None → 시장가(01) ) return result.get("rt_cd") == "0" except Exception as e: @@ -194,18 +202,24 @@ async def _record_trade( quantity: int, reason: str, profit_loss: float = 0.0, + order_price: Optional[float] = None, # 주문가 (슬리피지 분석용) ) -> None: """TradeHistory DB 기록""" try: + total = price * quantity + slippage = (price - order_price) if order_price is not None else None trade = TradeHistory( stock_code=stock_code, stock_name=stock_name, order_type=order_type, + order_price=order_price, + order_quantity=quantity, executed_price=price, executed_quantity=quantity, - total_amount=price * quantity, - profit_loss=profit_loss, - profit_loss_rate=(profit_loss / (price * quantity) * 100) if price * quantity > 0 else 0, + total_amount=total, + net_profit_loss=profit_loss, # ← net_profit_loss (profit_loss 컬럼 없음) + profit_loss_rate=(profit_loss / total * 100) if total > 0 else 0, + slippage=slippage, status="COMPLETED", entry_reason=reason if order_type == "BUY" else None, exit_reason=reason if order_type == "SELL" else None, @@ -257,8 +271,11 @@ async def run_day_trading_cycle() -> Dict[str, Any]: result["orb_count"] = len(_orb_cache) return result - # ── 3. 매매 가능 구간 (09:30~15:00) ── - if not is_day_trade_entry_window(): + # ── 3. 매매 가능 구간 ── + in_entry_window = is_day_trade_entry_window() # 09:30~15:00 + + # 포지션이 있으면 입장 창 밖(15:00~15:20)에서도 청산 체크 수행 + if not in_entry_window and not _positions: result["phase"] = "outside_window" return result @@ -271,11 +288,18 @@ async def run_day_trading_cycle() -> Dict[str, Any]: exits = await _check_exits(candle_cache=_cycle_candle_cache) result["exits"] = exits - # ── 5. 신규 진입 신호 체크 ── + # ── 5. 신규 진입 신호 체크 (진입 창 안에서만) ── + if not in_entry_window: + result["phase"] = "exit_only" + return result + if len(_positions) < MAX_POSITIONS and settings.AUTO_TRADE_ENABLED: portfolio = await _get_portfolio_summary() total_assets = portfolio["total_assets"] available_cash = portfolio["available_cash"] + global _last_total_assets + if total_assets > 0: + _last_total_assets = total_assets # is_daily_loss_exceeded 계산용 캐시 갱신 if not is_daily_loss_exceeded(total_assets, _daily_pnl): current_day_trade_amount = sum( @@ -379,8 +403,10 @@ async def _check_exits( ok = await _place_sell_order(code, quantity, price=0) # 시장가 if ok: pnl = (current_price - pos["entry_price"]) * quantity - _daily_pnl += pnl - del _positions[code] + async with _trade_lock: + _daily_pnl += pnl + _positions.pop(code, None) + _closed_today.add(code) # 당일 재진입 차단 async with AsyncSessionLocal() as db: await _record_trade( @@ -432,6 +458,8 @@ async def _scan_signals( for code, name in list(_watchlist.items()): if code in _positions: continue + if code in _closed_today: # 당일 청산 종목 재진입 차단 + continue try: if candle_cache is not None and code in candle_cache: candles_1m = candle_cache[code] @@ -494,6 +522,9 @@ async def _execute_entry(signal: DayTradeSignal, quantity: int) -> Optional[Dict return None if signal.stock_code in _positions: return None + if signal.stock_code in _closed_today: # 당일 청산 종목 재진입 최종 방어 + logger.info(f"⛔ 당일 청산 종목 재진입 차단: {signal.stock_code}") + return None order_no = await _place_buy_order( signal.stock_code, quantity, int(signal.entry_price) @@ -517,6 +548,7 @@ async def _execute_entry(signal: DayTradeSignal, quantity: int) -> Optional[Dict await _record_trade( db, signal.stock_code, signal.stock_name, "BUY", signal.entry_price, quantity, signal.reason, + order_price=float(int(signal.entry_price)), # 주문가 = 지정가 ) # ActiveTrade 저장 — 재시작 시 포지션 복구를 위해 active = ActiveTrade( @@ -563,8 +595,10 @@ async def _execute_eod_flat() -> None: ok = await _place_sell_order(code, pos["quantity"], price=0) if ok: pnl = (current_price - pos["entry_price"]) * pos["quantity"] - _daily_pnl += pnl - del _positions[code] + async with _trade_lock: + _daily_pnl += pnl + _positions.pop(code, None) + _closed_today.add(code) # 당일 재진입 차단 logger.info(f"🔔 EOD 강제 청산 [{code}]: PNL={pnl:+,.0f}원") async with AsyncSessionLocal() as db: @@ -595,10 +629,15 @@ async def restore_positions_from_db() -> None: strategy_name='day_trade', status='active'인 당일 ActiveTrade 레코드로 인메모리 _positions를 재구성한다. + + created_at은 UTC naive datetime으로 저장되므로 날짜 비교는 UTC 기준으로 수행. + 한국 장중(09:00~15:30 KST = 00:00~06:30 UTC)은 UTC/KST 날짜가 항상 일치하므로 + UTC 기준으로 비교해도 동일한 결과를 얻는다. """ if _positions: return # 이미 포지션 있으면 스킵 - today = _today_kst() + from datetime import timezone + today_utc = datetime.now(timezone.utc).date() # UTC 기준 오늘 날짜 (created_at과 동일 기준) try: async with AsyncSessionLocal() as db: result = await db.execute( @@ -609,9 +648,16 @@ async def restore_positions_from_db() -> None: ) rows = result.scalars().all() for row in rows: - if row.created_at and row.created_at.date() != today: + # created_at은 UTC 저장 → UTC 날짜로 비교 + if row.created_at and row.created_at.date() != today_utc: continue entry_price = float(row.entry_price) + # entry_time 표시: UTC → KST 변환 (UTC+9) + if row.created_at: + created_kst = row.created_at.replace(tzinfo=pytz.utc).astimezone(KST) + entry_time_str = created_kst.strftime("%H:%M:%S") + else: + entry_time_str = "??:??:??" _positions[row.stock_code] = { "stock_name": row.stock_name or row.stock_code, "strategy": row.entry_reason or "unknown", @@ -620,7 +666,7 @@ async def restore_positions_from_db() -> None: "target_price": float(row.take_profit) if row.take_profit else entry_price * 1.03, "quantity": row.quantity, "highest_price": float(row.highest_since_entry or row.entry_price), - "entry_time": row.created_at.strftime("%H:%M:%S") if row.created_at else "??:??:??", + "entry_time": entry_time_str, "reason": row.entry_reason or "", } if _positions: @@ -793,7 +839,7 @@ async def get_positions(): "count": len(result), "daily_pnl": round(_daily_pnl, 0), "is_daily_loss_exceeded": is_daily_loss_exceeded( - sum(p["entry_price"] * p["quantity"] for p in _positions.values()), + _last_total_assets, # 사이클 캐시 사용 (포지션 가치가 아닌 총자산) _daily_pnl, ), } diff --git a/backend/app/api/endpoints/journals.py b/backend/app/api/endpoints/journals.py index 8187420..c8a7b55 100644 --- a/backend/app/api/endpoints/journals.py +++ b/backend/app/api/endpoints/journals.py @@ -8,9 +8,16 @@ from fastapi import APIRouter, Depends, Query from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, desc from typing import Optional -from datetime import datetime +from datetime import datetime, timezone import logging + +def _utc_iso(dt: datetime | None) -> str | None: + """UTC naive datetime → ISO 8601 with +00:00 (JS가 UTC로 올바르게 파싱)""" + if dt is None: + return None + return dt.replace(tzinfo=timezone.utc).isoformat() + from app.core.database import get_db from app.models.models import TradeJournal @@ -42,12 +49,12 @@ async def list_journals( "trade_id": j.trade_id, "stock_code": j.stock_code, "stock_name": j.stock_name, - "journal_date": j.journal_date.isoformat() if j.journal_date else None, + "journal_date": _utc_iso(j.journal_date) if j.journal_date else None, "content": j.content, "strategy_name": j.strategy_name, "mood": j.mood, "tags": j.tags, - "created_at": j.created_at.isoformat() if j.created_at else None, + "created_at": _utc_iso(j.created_at), } for j in journals ] diff --git a/backend/app/api/endpoints/market_data.py b/backend/app/api/endpoints/market_data.py index d72c370..1df3f0d 100644 --- a/backend/app/api/endpoints/market_data.py +++ b/backend/app/api/endpoints/market_data.py @@ -10,6 +10,9 @@ from datetime import datetime, timedelta import random import logging +import pytz + +_KST = pytz.timezone("Asia/Seoul") from fastapi import APIRouter, HTTPException, Query, Depends from typing import List, Dict @@ -693,7 +696,7 @@ async def get_ai_analysis(stock_code: str, db: AsyncSession = Depends(get_db)): "recommendation": recommendation, "confidence": confidence, "trend": trend, - "analyzed_at": datetime.now().isoformat(), + "analyzed_at": datetime.now(_KST).isoformat(), # KST +09:00 포함 } except HTTPException: diff --git a/backend/app/api/endpoints/mcp.py b/backend/app/api/endpoints/mcp.py index da36eb3..4a107e8 100644 --- a/backend/app/api/endpoints/mcp.py +++ b/backend/app/api/endpoints/mcp.py @@ -22,7 +22,7 @@ async def get_mcp_status(): """MCP 서버 상태 확인""" try: async with httpx.AsyncClient() as client: - response = await client.get(f"{settings.MCP_SERVER_URL}/health", timeout=5.0) + response = await client.get(f"{settings.MCP_SERVER_URL}/health", timeout=2.0) if response.status_code == 200: health_data = response.json() diff --git a/backend/app/api/endpoints/recommendations.py b/backend/app/api/endpoints/recommendations.py index 9bd997b..a5b6b18 100644 --- a/backend/app/api/endpoints/recommendations.py +++ b/backend/app/api/endpoints/recommendations.py @@ -7,6 +7,8 @@ import asyncio import logging import httpx import pytz +import time +from collections import namedtuple from app.services.korea_investment_api import korea_investment_api from app.core.database import get_db, AsyncSessionLocal @@ -27,6 +29,58 @@ from app.services.loss_tracker import loss_tracker router = APIRouter() logger = logging.getLogger(__name__) +# 매매 규칙 namedtuple (get_rule_based_recommendations_from_db/realtime 공용) +Rule = namedtuple('Rule', ['id', 'name', 'description', 'rule_type', 'priority', 'conditions', 'actions', 'is_active']) + + +def _stock_analysis_to_dict(stock, rule=None, actions=None) -> dict: + """StockAnalysis ORM 행 → stock_dict 변환 헬퍼 (중복 제거). + + rule=None 이면 '고점수 분석 종목' 폴백 값을 채운다. + """ + base = { + "stock_code": stock.stock_code, + "stock_name": stock.stock_name, + "current_price": int(stock.current_price) if stock.current_price else 0, + "change_rate": round(float(stock.change_rate), 2) if stock.change_rate else 0, + "volume": stock.volume, + "rsi": round(float(stock.rsi), 2) if stock.rsi else None, + "adx_14": round(float(stock.adx_14), 2) if stock.adx_14 else None, + "ma5": int(stock.ma5) if stock.ma5 else None, + "ma20": int(stock.ma20) if stock.ma20 else None, + "stochastic_k": round(float(stock.stochastic_k), 2) if stock.stochastic_k else None, + "signals": stock.signals or [], + "score": stock.score, + "ml_score": stock.ml_score, + "ml_confidence": round(float(stock.ml_confidence), 4) if stock.ml_confidence else None, + "ml_signal": stock.ml_signal, + "price_date": stock.analysis_date.strftime("%Y%m%d"), + } + if rule is not None: + _a = actions or {} + base.update({ + "rule_id": rule.id, + "rule_name": rule.name, + "rule_description": rule.description, + "rule_priority": rule.priority, + "take_profit": _a.get("take_profit", 3.0), + "stop_loss": _a.get("stop_loss", -1.5), + "max_add_buys": _a.get("max_add_buys", 3), + "add_buy_cooldown_min": _a.get("add_buy_cooldown_min", 15), + }) + else: + base.update({ + "rule_id": None, + "rule_name": "고점수 분석 종목", + "rule_description": "매매 규칙 조건 미충족 - 분석 점수 기준 상위 종목", + "rule_priority": 99, + "take_profit": 3.0, + "stop_loss": -1.5, + "max_add_buys": 2, + "add_buy_cooldown_min": 30, + }) + return base + # 한국 시간대 KST = pytz.timezone('Asia/Seoul') @@ -170,7 +224,7 @@ def analyze_custom_signals(chart_data): close = closes[-1] # 🔥 이격도 계산 (추격 매수 방지) - 더 엄격하게 - disparity = (close / ma20) * 100 + disparity = (close / ma20) * 100 if ma20 else 100 # 🚫 급등 종목 필터: 최근 3일 상승률 체크 if len(closes) >= 4: @@ -428,6 +482,11 @@ def analyze_volatility_breakout(chart_data: list, k: float = 0.5) -> dict: return {"signal": None, "description": "변동성 돌파 미달"} +# check_market_regime() 인메모리 캐시 (KIS API 반복 호출 방지) +_market_regime_cache: dict = {"result": None, "ts": 0.0} +_MARKET_REGIME_CACHE_TTL = 120 # 2분 + + async def check_market_regime() -> dict: """시장 레짐 판별 (CLAUDE.md 섹션 7.1: 200일 MA + 기울기) @@ -444,7 +503,14 @@ async def check_market_regime() -> dict: if not is_market_open(): return {"regime": "neutral", "regime_200d": "bear", "message": "장외 시간 - 시장 레짐 판별 스킵", "position_multiplier": 0.5} + # 캐시 히트: TTL 내 이전 결과 반환 (KIS API 반복 호출 방지) + now_ts = time.monotonic() + if _market_regime_cache["result"] is not None and (now_ts - _market_regime_cache["ts"]) < _MARKET_REGIME_CACHE_TTL: + logger.debug("check_market_regime: cache hit") + return _market_regime_cache["result"] + # KODEX 200 (069500) = KOSPI 200 추종 ETF 사용 (지수코드 0001은 daily-price API 미지원) + logger.info("check_market_regime: KIS API 호출 (캐시 미스)") kospi_chart = await korea_investment_api.get_daily_chart("069500", "D", limit=240) if "output2" not in kospi_chart or not kospi_chart["output2"]: @@ -480,9 +546,14 @@ async def check_market_regime() -> dict: drop_from_ma20 = ((current_price - ma20) / ma20) * 100 + def _cache_and_return(result: dict) -> dict: + _market_regime_cache["result"] = result + _market_regime_cache["ts"] = time.monotonic() + return result + # 강한 하락장: 코스피 < MA60 또는 5일 연속 하락 if current_price < ma60 or consecutive_down >= 5: - return { + return _cache_and_return({ "regime": "strong_bearish", "regime_200d": "bear", "message": f"강한 하락장 (코스피: {current_price:.0f}, MA200: {ma200:.0f}, {consecutive_down}일 연속하락)", @@ -491,11 +562,11 @@ async def check_market_regime() -> dict: "max_trades": 1, "consecutive_down": consecutive_down, "drop_from_ma20": round(drop_from_ma20, 1), - } + }) # 약세장: 코스피 < MA20 if current_price < ma20: - return { + return _cache_and_return({ "regime": "bearish", "regime_200d": "bear", "message": f"하락장 (코스피: {current_price:.0f} < MA20: {ma20:.0f}, MA200: {ma200:.0f})", @@ -504,10 +575,10 @@ async def check_market_regime() -> dict: "max_trades": settings.BEAR_MARKET_MAX_TRADES, "consecutive_down": consecutive_down, "drop_from_ma20": round(drop_from_ma20, 1), - } + }) # 상승장: 200일 MA 기반 - return { + return _cache_and_return({ "regime": "bullish", "regime_200d": "bull" if is_bull_200 else "bear", "message": f"상승장 (코스피: {current_price:.0f}, MA200: {ma200:.0f}, Bull={is_bull_200})", @@ -516,7 +587,7 @@ async def check_market_regime() -> dict: "max_trades": settings.MAX_DAILY_TRADES, "consecutive_down": consecutive_down, "drop_from_ma20": round(drop_from_ma20, 1), - } + }) except Exception as e: logger.error(f"시장 지수 확인 실패: {e}") return {"regime": "neutral", "message": "시장 지수 확인 실패", "position_multiplier": 0.5} @@ -1136,6 +1207,65 @@ async def get_rule_based_recommendations( return result +async def _load_active_rules_from_db(db: AsyncSession) -> list: + """trading_rules 테이블에서 활성 규칙 로드. + DB가 비어있으면 JSON 파일 → 하드코딩 폴백 순으로 시도한다. + """ + from app.models.models import TradingRule as TradingRuleModel + try: + result = await db.execute( + select(TradingRuleModel) + .where(TradingRuleModel.is_active == True) + .order_by(TradingRuleModel.priority.desc()) + ) + db_rules = result.scalars().all() + if db_rules: + logger.info(f"trading_rules DB에서 {len(db_rules)}개 활성 규칙 로드") + return [ + Rule(id=r.id, name=r.name, description=r.description or "", + rule_type=r.rule_type, priority=r.priority, + conditions=r.conditions, actions=r.actions, is_active=True) + for r in db_rules + ] + except Exception as e: + logger.warning(f"trading_rules DB 조회 실패: {e}") + + # 폴백 1: JSON 파일 + try: + import json as _json + from pathlib import Path as _Path + config_path = _Path(__file__).parent.parent.parent.parent / "trading_rules_config.json" + with open(config_path, 'r', encoding='utf-8') as f: + config = _json.load(f) + rules_data = [r for r in config.get("rules", []) if r.get("is_active")] + if rules_data: + logger.info(f"trading_rules DB 비어있음 — JSON 파일에서 {len(rules_data)}개 규칙 로드") + return [ + Rule(id=i + 1, name=r["name"], description=r.get("description", ""), + rule_type=r["rule_type"].upper(), priority=r.get("priority", i + 1), + conditions=r.get("conditions", {}), actions=r.get("actions", {}), + is_active=True) + for i, r in enumerate(rules_data) + ] + except Exception as e: + logger.warning(f"JSON 규칙 파일 로드 실패: {e}") + + # 폴백 2: 하드코딩 기본 규칙 + logger.warning("규칙 소스 없음 — 하드코딩 기본 규칙 사용") + return [ + Rule( + id=1, name="RSI+볼린저 평균회귀", description="기본 평균회귀 규칙 (CLAUDE.md 섹션 6)", + rule_type="MEAN_REVERSION", priority=1, + conditions={"rsi_bull_max": 25, "rsi_bear_max": 35, + "bb_lower_pct": 3.0, "bb_upper_pct": 1.5}, + actions={"take_profit": 3.0, "stop_loss": -1.5, + "position_sizing": {"target_ratio": 0.10}, + "max_add_buys": 2, "add_buy_cooldown_min": 30}, + is_active=True + ) + ] + + def is_mean_reversion_candidate(stock, conditions: dict, market_regime_str: str) -> bool: """RSI+볼린저 평균회귀 매수 조건 확인 (CLAUDE.md 섹션 7.2) @@ -1181,65 +1311,17 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): """DB에서 사전 분석된 데이터 조회 (빠름 - 0.1초)""" logger.info("Rule-based recommendations from DB") - - # JSON 파일에서 규칙 로드 - import json - from pathlib import Path - from collections import namedtuple - - Rule = namedtuple('Rule', ['id', 'name', 'description', 'rule_type', 'priority', 'conditions', 'actions', 'is_active']) - - try: - config_path = Path(__file__).parent.parent.parent.parent / "trading_rules_config.json" - with open(config_path, 'r', encoding='utf-8') as f: - config = json.load(f) - - rules_data = config.get("rules", []) - active_rules = [] - - for idx, rule_data in enumerate(rules_data, start=1): - if rule_data.get("is_active", False): - active_rules.append(Rule( - id=idx, - name=rule_data["name"], - description=rule_data["description"], - rule_type=rule_data["rule_type"].upper(), - priority=rule_data.get("priority", idx), - conditions=rule_data.get("conditions", {}), - actions=rule_data.get("actions", {}), - is_active=True - )) - - logger.info(f"Loaded {len(active_rules)} active trading rules from config file") - except Exception as e: - logger.error(f"Failed to load trading rules config: {e}") - # 폴백: 기본 규칙 사용 - active_rules = [ - Rule( - id=1, - name="RSI 과매도 반등", - description="RSI 30 이하 과매도 종목", - rule_type="BUY", - priority=1, - conditions={"rsi_threshold": 30}, - actions={ - "take_profit": 5.0, - "stop_loss": 3.0, - "position_sizing": {"target_ratio": 0.15} - }, - is_active=True - ) - ] - logger.warning(f"Using fallback rule: {len(active_rules)} rule(s)") - + + active_rules = await _load_active_rules_from_db(db) logger.info(f"Found {len(active_rules)} active trading rules") - # 오늘 날짜의 분석 데이터 조회 (중복 종목은 score 높은 것 우선) - today = date.today() + # 오늘·어제 분석 데이터 조회 (KST 날짜 기준, 배치 미실행 시 전날 데이터 사용) + today_kst = datetime.now(KST).date() + since_date = today_kst - timedelta(days=1) # 어제~오늘 포함 result = await db.execute( select(StockAnalysis) - .where(StockAnalysis.analysis_date >= today) - .order_by(StockAnalysis.score.desc()) + .where(StockAnalysis.analysis_date >= since_date) + .order_by(StockAnalysis.analysis_date.desc(), StockAnalysis.score.desc()) ) raw_stocks = result.scalars().all() # 종목코드 기준 중복 제거 (score 내림차순 정렬 후 첫 번째만) @@ -1264,7 +1346,7 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): "message": "⚠️ 배치 분석 데이터 없음. 'make batch-test' 명령어로 분석 실행 필요." } - logger.info(f"Found {len(all_analyzed_stocks)} stocks in DB for {today}") + logger.info(f"Found {len(all_analyzed_stocks)} stocks in DB for {today_kst}") # 규칙별로 매칭되는 종목 찾기 rule_matches = [] @@ -1313,38 +1395,13 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): # 여기까지 왔으면 rule_matched_stocks에 추가 rule_matched_stocks.append(stock) - # 점수순 정렬 - rule_matched_stocks.sort(key=lambda x: x.score, reverse=True) + # 점수순 정렬 (score=None 안전 처리) + rule_matched_stocks.sort(key=lambda x: x.score or 0, reverse=True) # 이하 로직은 그대로 (Top N, rule_matches, stock_to_rules, 실시간 가격 업데이트 등) rule_stocks = [] for stock in rule_matched_stocks[:10]: - stock_dict = { - "stock_code": stock.stock_code, - "stock_name": stock.stock_name, - "current_price": int(stock.current_price) if stock.current_price else 0, - "change_rate": round(float(stock.change_rate), 2) if stock.change_rate else 0, - "volume": stock.volume, - "rsi": round(float(stock.rsi), 2) if stock.rsi else None, - "adx_14": round(float(stock.adx_14), 2) if stock.adx_14 else None, - "ma5": int(stock.ma5) if stock.ma5 else None, - "ma20": int(stock.ma20) if stock.ma20 else None, - "stochastic_k": round(float(stock.stochastic_k), 2) if stock.stochastic_k else None, - "signals": stock.signals or [], - "score": stock.score, - "ml_score": stock.ml_score, - "ml_confidence": round(float(stock.ml_confidence), 4) if stock.ml_confidence else None, - "ml_signal": stock.ml_signal, - "price_date": stock.analysis_date.strftime("%Y%m%d"), - "rule_id": rule.id, - "rule_name": rule.name, - "rule_description": rule.description, - "rule_priority": rule.priority, - "take_profit": actions.get("take_profit", 3.0), - "stop_loss": actions.get("stop_loss", -1.5), - "max_add_buys": actions.get("max_add_buys", 3), - "add_buy_cooldown_min": actions.get("add_buy_cooldown_min", 15), - } + stock_dict = _stock_analysis_to_dict(stock, rule=rule, actions=actions) rule_stocks.append(stock_dict) rule_matches.append(stock_dict) @@ -1381,32 +1438,7 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): if stock.stock_code in already_added: continue already_added.add(stock.stock_code) - stock_dict = { - "stock_code": stock.stock_code, - "stock_name": stock.stock_name, - "current_price": int(stock.current_price) if stock.current_price else 0, - "change_rate": round(float(stock.change_rate), 2) if stock.change_rate else 0, - "volume": stock.volume, - "rsi": round(float(stock.rsi), 2) if stock.rsi else None, - "adx_14": round(float(stock.adx_14), 2) if stock.adx_14 else None, - "ma5": int(stock.ma5) if stock.ma5 else None, - "ma20": int(stock.ma20) if stock.ma20 else None, - "stochastic_k": round(float(stock.stochastic_k), 2) if stock.stochastic_k else None, - "signals": stock.signals or [], - "score": stock.score, - "ml_score": stock.ml_score, - "ml_confidence": round(float(stock.ml_confidence), 4) if stock.ml_confidence else None, - "ml_signal": stock.ml_signal, - "price_date": stock.analysis_date.strftime("%Y%m%d"), - "rule_id": None, - "rule_name": "고점수 분석 종목", - "rule_description": "매매 규칙 조건 미충족 - 분석 점수 기준 상위 종목", - "rule_priority": 99, - "take_profit": 3.0, - "stop_loss": -1.5, - "max_add_buys": 2, - "add_buy_cooldown_min": 30, - } + stock_dict = _stock_analysis_to_dict(stock) fallback_stocks.append(stock_dict) rule_matches.append(stock_dict) @@ -1442,82 +1474,23 @@ async def get_rule_based_recommendations_from_db(db: AsyncSession): "recommendations": rule_matches, "recommendations_by_rule": recommendations_by_rule, "rules_applied": [{"id": r.id, "name": r.name, "priority": r.priority} for r in active_rules], - # DB에 저장된 실제 배치 분석 시각 (API 요청 시각이 아님) - "batch_time": max( - (s.analysis_date for s in all_analyzed_stocks), - default=datetime.now() + # DB에 저장된 실제 배치 분석 날짜 (TIMESTAMP/DATE 모두 YYYY-MM-DD 반환) + "batch_time": (lambda d: d.date() if isinstance(d, datetime) else d)( + max((s.analysis_date for s in all_analyzed_stocks), default=date.today()) ).isoformat(), } async def get_rule_based_recommendations_realtime(): """ - 실시간 KIS API 분석 (느림 - 30초) - - 기존 방식 유지 + 실시간 KIS API 분석 (느림 - 배치 주기에 호출됨) + 규칙은 trading_rules DB → JSON 파일 → 하드코딩 순으로 로드한다. """ - # TODO: 데이터베이스 연결 후 실제 규칙 조회 - # 현재는 샘플 규칙으로 동작 - logger.info("Rule-based recommendations requested (realtime mode)") - - # 샘플 규칙 생성 (실제로는 DB에서 조회) - from collections import namedtuple - Rule = namedtuple('Rule', ['id', 'name', 'description', 'rule_type', 'priority', 'conditions', 'actions', 'is_active']) - - active_rules = [ - Rule( - id=1, - name="RSI 과매도 반등", - description="RSI 40 이하 과매도 종목", - rule_type="BUY", - priority=1, - conditions={"rsi_threshold": 40}, - actions={ - "take_profit": 3.0, - "stop_loss": -1.5, - "position_sizing": {"target_ratio": 0.10}, - "max_position_pct": 0.12, - "max_add_buys": 3, - "add_buy_cooldown_min": 15 - }, - is_active=True - ), - Rule( - id=2, - name="골든크로스 추세", - description="5일선이 20일선 돌파", - rule_type="BUY", - priority=2, - conditions={"sma_short": 5, "sma_long": 20}, - actions={ - "take_profit": 3.0, - "stop_loss": -1.5, - "position_sizing": {"target_ratio": 0.10}, - "max_position_pct": 0.12, - "max_add_buys": 3, - "add_buy_cooldown_min": 15 - }, - is_active=True - ), - Rule( - id=3, - name="볼린저밴드 하단", - description="볼린저밴드 하단 근처 반등", - rule_type="BUY", - priority=3, - conditions={"bb_threshold": 0.2}, - actions={ - "take_profit": 3.0, - "stop_loss": -1.5, - "position_sizing": {"target_ratio": 0.10}, - "max_position_pct": 0.12, - "max_add_buys": 3, - "add_buy_cooldown_min": 15 - }, - is_active=True - ), - ] + + async with AsyncSessionLocal() as _db: + active_rules = await _load_active_rules_from_db(_db) + logger.info(f"Found {len(active_rules)} active trading rules (realtime)") # active_rules가 비어있으면 기본 지표로 분석 if not active_rules: @@ -1597,9 +1570,9 @@ async def get_rule_based_recommendations_realtime(): logger.info(f"Found {len(active_rules)} active trading rules") - # 분석할 종목 리스트 (타임아웃 방지를 위해 10개로 제한) + # 분석할 종목 리스트 all_stocks = await get_all_tradable_stocks() - stocks_to_analyze = all_stocks[:10] # 상위 10개 종목으로 제한 + stocks_to_analyze = all_stocks[:settings.ANALYSIS_STOCK_COUNT] # 보유 포트폴리오 종목 추가 (보유 종목은 거래대금 순위 무관하게 항상 분석) try: @@ -1653,54 +1626,44 @@ async def get_rule_based_recommendations_realtime(): logger.info(f"Successfully analyzed {len(valid_analyzed_stocks)} stocks") + # 시장 레짐 판별 (DB 모드와 동일한 로직) + market_regime_info = await check_market_regime() + current_market_regime_str = market_regime_info.get("regime_200d", "bear") + # 규칙별로 매칭되는 종목 찾기 (이미 분석된 데이터 재사용) rule_matches = [] recommendations_by_rule = [] - + for rule in active_rules: logger.info(f"Applying rule: {rule.name}") - - # 규칙 조건 파싱 + conditions = rule.conditions actions = rule.actions - - # 이미 분석된 종목들 중에서 규칙에 맞는 종목 필터링 + rule_matched_stocks = [] - + for stock in valid_analyzed_stocks: # ETF/레버리지/인버스/KODEX/선물 종목 제외 etf_keywords = ["ETF", "etf", "레버리지", "인버스", "KODEX", "선물"] if any(k in stock.stock_name for k in etf_keywords): continue - # 규칙 조건 체크 - matches = True - - # RSI 조건 체크 - if "rsi_threshold" in conditions: - rsi_threshold = conditions["rsi_threshold"] - if not any(f"RSI" in signal for signal in stock.signals): - matches = False - - # Trend filter: only allow long entries in uptrend - if TREND_FILTER_ENABLED: - try: - t_ok = False - if stock.ma20 is not None and stock.ma60 is not None: - t_ok = (stock.ma20 > stock.ma60) and (stock.current_price > stock.ma20) - else: - # Fallback compute from price history is skipped here (already done in analyze_stock) - t_ok = False - if not t_ok: - matches = False - except Exception: - matches = False + + # === 통합 규칙 조건 매칭 (DB 모드와 동일한 로직) === + if rule.rule_type == "MEAN_REVERSION": + matches = is_mean_reversion_candidate(stock, conditions, current_market_regime_str) + elif rule.rule_type in ("TREND", "BREAKOUT", "MOMENTUM"): + matches = False # 미구현 (DB 모드와 동일) + else: + # BUY 타입: RSI 임계값 체크 + rsi_threshold = conditions.get("rsi_threshold", 40) + matches = (stock.rsi is not None and stock.rsi <= rsi_threshold) if matches: rule_matched_stocks.append(stock) - # 점수순 정렬 - rule_matched_stocks.sort(key=lambda x: x.score, reverse=True) - + # 점수순 정렬 (score=None 안전 처리) + rule_matched_stocks.sort(key=lambda x: x.score or 0, reverse=True) + # 규칙별로 종목 저장 rule_stocks = [] for rec in rule_matched_stocks[:10]: # 규칙당 상위 10개 @@ -1764,65 +1727,15 @@ async def get_rule_based_recommendations_realtime(): stock["change_rate"] = updated["change_rate"] # 실시간 분석 결과를 DB에 저장 (규칙 매칭 여부와 무관하게 분석된 전 종목 저장) - from app.models.models import StockAnalysis + # N+1 SELECT 루프 → 단일 bulk UPSERT로 대체 (성능 개선) + from app.services.analysis_db_service import bulk_upsert_analysis as _bulk_upsert from datetime import date as _date _today = _date.today() - _now = datetime.now() + _items = [(rec.stock_code, rec.stock_name, rec) for rec in valid_analyzed_stocks] async with AsyncSessionLocal() as _db: - # valid_analyzed_stocks = StockRecommendation Pydantic 모델 리스트 - # top_recommendations에만 저장하면 급등과열/필터로 모두 차단될 때 아무것도 저장 안 됨 - for rec in valid_analyzed_stocks: - _existing = (await _db.execute( - select(StockAnalysis).where( - StockAnalysis.stock_code == rec.stock_code, - StockAnalysis.analysis_date >= _today, - ) - )).scalar_one_or_none() - if _existing: - # 기존 레코드 업데이트 (지표 갱신) - _existing.analysis_date = _now # 배치 실행 시각으로 갱신 - _existing.current_price = rec.current_price - _existing.change_rate = rec.change_rate or 0 - _existing.volume = rec.volume or 0 - _existing.rsi = rec.rsi - _existing.ma5 = rec.ma5 - _existing.ma20 = rec.ma20 - _existing.ma60 = rec.ma60 - _existing.stochastic_k = rec.stochastic_k - _existing.stochastic_d = rec.stochastic_d - _existing.bollinger_upper = rec.bollinger_upper - _existing.bollinger_lower = rec.bollinger_lower - _existing.bollinger_middle = rec.bollinger_middle - _existing.signals = rec.signals or [] - _existing.score = rec.score - _existing.ml_score = rec.ml_score - _existing.ml_confidence = rec.ml_confidence - _existing.ml_signal = rec.ml_signal - else: - _db.add(StockAnalysis( - stock_code=rec.stock_code, - stock_name=rec.stock_name, - analysis_date=_now, - current_price=rec.current_price, - change_rate=rec.change_rate or 0, - volume=rec.volume or 0, - rsi=rec.rsi, - ma5=rec.ma5, - ma20=rec.ma20, - ma60=rec.ma60, - stochastic_k=rec.stochastic_k, - stochastic_d=rec.stochastic_d, - bollinger_upper=rec.bollinger_upper, - bollinger_lower=rec.bollinger_lower, - bollinger_middle=rec.bollinger_middle, - signals=rec.signals or [], - score=rec.score, - ml_score=rec.ml_score, - ml_confidence=rec.ml_confidence, - ml_signal=rec.ml_signal, - )) + saved = await _bulk_upsert(_db, _items, _today) await _db.commit() - logger.info(f"✅ [Batch] DB upsert 완료: {len(valid_analyzed_stocks)}종목 저장") + logger.info(f"✅ [Batch] DB upsert 완료: {saved}종목 저장") return { "total": len(rule_matches), "active_rules": len(active_rules), @@ -2614,7 +2527,7 @@ async def execute_auto_trading(db: AsyncSession = Depends(get_db)): if total_processed >= max_trades: break - for stock in rule_group["stocks"][:2]: # 규칙당 상위 2개만 + for stock in rule_group["stocks"][:settings.AUTO_TRADE_TOP_N_PER_RULE]: if total_processed >= max_trades: break diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 3b09212..e754006 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -119,6 +119,8 @@ class Settings(BaseSettings): STOP_LOSS_PERCENT: float = -6.0 # 저변동 종목 손절 -6% (섹션 6.2) TAKE_PROFIT_PERCENT: float = 5.0 # 익절 기본값 AUTO_TRADE_ENABLED: bool = True + ANALYSIS_STOCK_COUNT: int = 150 # 배치 분석 종목 수 (.env로 조정 가능) + AUTO_TRADE_TOP_N_PER_RULE: int = 2 # 규칙당 자동매매 실행 최대 종목 수 # 포지션 규칙 (섹션 6.1) MAX_HOLDINGS_COUNT: int = 20 # 동시 보유 최대 20종목 diff --git a/backend/app/main.py b/backend/app/main.py index 80b4fae..c8b8d43 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -24,9 +24,23 @@ from app.core.market_schedule import is_market_open from sqlalchemy import text # 로깅 설정 +import os +import logging.handlers +_log_level = getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO) +_log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' +_log_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "logs") +os.makedirs(_log_dir, exist_ok=True) +_log_file = os.path.join(_log_dir, "backend.log") + logging.basicConfig( - level=getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO), - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' + level=_log_level, + format=_log_format, + handlers=[ + logging.StreamHandler(), + logging.handlers.RotatingFileHandler( + _log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8" + ), + ] ) logger = logging.getLogger(__name__) @@ -191,6 +205,20 @@ async def lifespan(app: FastAPI): ADD COLUMN IF NOT EXISTS ml_confidence NUMERIC(5, 4), ADD COLUMN IF NOT EXISTS ml_signal VARCHAR(10); """)) + # UPSERT를 위한 unique constraint — init-db.sql에 없을 경우 자동 생성 + await conn.execute(text(""" + DO $$ BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'uq_stock_analysis_code_date' + AND conrelid = 'stock_analysis'::regclass + ) THEN + ALTER TABLE stock_analysis + ADD CONSTRAINT uq_stock_analysis_code_date + UNIQUE (stock_code, analysis_date); + END IF; + END $$; + """)) # active_trades에 분할 청산 컬럼 추가 (CLAUDE.md 섹션 7.3) await conn.execute(text(""" ALTER TABLE active_trades diff --git a/backend/app/models/models.py b/backend/app/models/models.py index eee8b42..904b19c 100644 --- a/backend/app/models/models.py +++ b/backend/app/models/models.py @@ -178,11 +178,14 @@ class MCPToolCall(Base): class StockAnalysis(Base): """매일 배치로 저장되는 종목 분석 데이터""" __tablename__ = "stock_analysis" - + __table_args__ = ( + UniqueConstraint("stock_code", "analysis_date", name="uq_stock_analysis_code_date"), + ) + id = Column(Integer, primary_key=True, index=True) stock_code = Column(String(20), nullable=False, index=True) stock_name = Column(String(100), nullable=False) - analysis_date = Column(DateTime, nullable=False, index=True) + analysis_date = Column(Date, nullable=False, index=True) # 가격 정보 current_price = Column(Integer, nullable=False) diff --git a/backend/app/services/analysis_db_service.py b/backend/app/services/analysis_db_service.py new file mode 100644 index 0000000..0c21cc9 --- /dev/null +++ b/backend/app/services/analysis_db_service.py @@ -0,0 +1,196 @@ +"""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을 원자적으로 해결. +""" +from __future__ import annotations + +import logging +from datetime import date, datetime, timezone +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.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절). + + True → 통과 (거래 가능), False → 필터됨 (5,000원 미만). + """ + 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 딕셔너리 변환. + + 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": datetime.now(timezone.utc).replace(tzinfo=None), + } + + +async def upsert_analysis( + session: AsyncSession, + stock_code: str, + stock_name: str, + result: "StockRecommendation", + analysis_date: date, +) -> None: + """단건 UPSERT — 동일 (stock_code, analysis_date)가 있으면 UPDATE, 없으면 INSERT. + + DELETE+INSERT 대비: + - 원자적 연산 (race condition 없음) + - 단일 DB 왕복 (네트워크 절반) + """ + row = to_analysis_dict(stock_code, stock_name, result, analysis_date) + stmt = pg_insert(StockAnalysis).values(**row) + 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) + + +async def bulk_upsert_analysis( + session: AsyncSession, + items: list[tuple[str, str, "StockRecommendation"]], + analysis_date: date, +) -> int: + """다건 UPSERT — 단일 INSERT ... ON CONFLICT 배치 실행. + + items: [(stock_code, stock_name, StockRecommendation), ...] + Returns: 처리된 건수 + """ + if not items: + return 0 + + rows = [to_analysis_dict(code, name, r, analysis_date) for code, name, r in items] + 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) + return len(rows) + + +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 (지나치게 강한 추세는 평균회귀 부적합) + - MA5 ≥ MA20 × 0.95 (ADX 방향 proxy — 강한 하락 추세 종목 제외) + + 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 + + # 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 + + # ADX 방향 proxy: MA5 < MA20 × 0.95 → 강한 하락 추세 → 평균회귀 부적합 + 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() diff --git a/backend/batch_daily_analysis.py b/backend/batch_daily_analysis.py index b1c746b..5bee7b9 100644 --- a/backend/batch_daily_analysis.py +++ b/backend/batch_daily_analysis.py @@ -4,14 +4,15 @@ 전체 거래 종목을 분석하여 DB에 저장 1. 시장 레짐 판별 (KOSPI200 기준 BULL/BEAR/SIDEWAYS) 2. 차트 데이터 조회 + 기술적 지표 계산 - 3. 매매 신호 생성 + DB 일괄 저장 (BULK INSERT) + 3. 매매 신호 생성 + DB Streaming 저장 (분석 즉시 UPSERT) 4. 배치 완료 후 DB 스캐너 실행 (추가 API 호출 없이) 5. 워치리스트 초기화 — 스캐너 신호 종목 우선 배치 성능 개선 (이전 대비): - BATCH_CONCURRENCY: 5 → 10 (API 세마포어 허용 범위 내) - - DB 트랜잭션: 종목당 2개(DELETE+INSERT) → BULK_SAVE_CHUNK(20개)당 1개 - - 500종목 기준: 1,000 트랜잭션 → 25 트랜잭션 (40× 감소) + - 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 @@ -28,13 +29,10 @@ KST = pytz.timezone('Asia/Seoul') 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.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 @@ -44,32 +42,11 @@ logging.basicConfig( ) 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) +BATCH_CONCURRENCY = 10 # API 세마포어 허용 범위 내 동시 처리 수 +BULK_SAVE_CHUNK = 20 # Streaming 저장 단위 (큐에 20개 쌓이면 즉시 UPSERT) # ------------------------------------------------------------------- @@ -92,13 +69,13 @@ async def detect_and_log_market_regime() -> str: # 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)) + 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", + 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() @@ -140,17 +117,18 @@ async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500) 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, "시가총액"])) + 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_stocks = [(code, name) for code, name, _ in stocks_with_cap[:limit]] + top_with_mkt = stocks_with_cap[:limit] + top_stocks = [(code, name) for code, name, _, _mkt in top_with_mkt] - 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 + 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) return top_stocks @@ -169,7 +147,7 @@ async def analyze_one_stock( stock_code: str, stock_name: str, ) -> tuple[str, Optional[StockRecommendation]]: - """종목 분석만 수행하고 결과 반환. DB 저장은 bulk_save_results에서 일괄 처리. + """종목 분석만 수행하고 결과 반환. DB 저장은 Streaming 저장기에서 처리. Returns: (status, result) status: "ok" | "no_chart" | "price_filter" | "error" @@ -182,8 +160,8 @@ async def analyze_one_stock( 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: + # 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 @@ -197,153 +175,55 @@ async def analyze_one_stock( # ------------------------------------------------------------------- -# 4. StockRecommendation → StockAnalysis ORM 변환 헬퍼 +# 4. Streaming 저장기 — 분석 완료 즉시 BULK_SAVE_CHUNK 단위 UPSERT # ------------------------------------------------------------------- -def _to_analysis_model( - stock_code: str, - stock_name: str, - result: StockRecommendation, +async def _streaming_saver( + queue: asyncio.Queue, 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, - ) +) -> dict: + """큐에서 분석 결과를 꺼내 BULK_SAVE_CHUNK(20)개마다 즉시 DB UPSERT. + 전체 메모리 적재 없이 분석 완료 즉시 저장하여 메모리 사용 최소화. -# ------------------------------------------------------------------- -# 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: 저장된 건수 + Returns: stats dict {"ok", "no_chart", "price_filter", "error", "saved"} """ - if not items: - return 0 + buffer: list[tuple[str, str, StockRecommendation]] = [] + stats = {"ok": 0, "no_chart": 0, "price_filter": 0, "error": 0, "saved": 0} - codes = [code for code, _, _ in items] + while True: + item = await queue.get() + if item is None: # sentinel: 분석 완료 + break - 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() + status, code, name, result = item + stats[status] = stats.get(status, 0) + 1 - return len(objs) + 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() -# ------------------------------------------------------------------- -# 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 (지나치게 강한 추세는 평균회귀 부적합) - - MA5 ≥ MA20 × 0.95 (ADX 방향 proxy — 강한 하락 추세 종목 제외) - - Returns: 신호 종목 코드 set - """ - try: + # 나머지 잔여분 저장 + if buffer: 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() + saved = await bulk_upsert_analysis(session, buffer, analysis_date) + await session.commit() + stats["saved"] += saved + logger.info("💾 DB 저장(잔여): %d건 누적", stats["saved"]) - 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 - - # ADX 방향 proxy: MA5 < MA20 × 0.95 → 강한 하락 추세 → 평균회귀 부적합 - # DI+/DI- 컬럼이 없으므로 MA 방향으로 추세 하락 여부 판단 - 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() + return stats # ------------------------------------------------------------------- -# 7. 메인 배치 +# 6. 메인 배치 # ------------------------------------------------------------------- async def batch_analyze_all_stocks(limit: int = None): @@ -353,13 +233,14 @@ async def batch_analyze_all_stocks(limit: int = None): 1. 시장 레짐 판별 (KOSPI 지수 기준) 2. 시가총액 상위 500개 종목 조회 3. 세마포어로 BATCH_CONCURRENCY(10)개씩 병렬 분석 (API 호출만) - 4. 분석 완료 후 BULK_SAVE_CHUNK(20)개씩 일괄 DB 저장 + 4. Streaming: 분석 완료 즉시 BULK_SAVE_CHUNK(20)개 단위 DB UPSERT 5. DB 스캐너 실행 (추가 API 호출 없음) 6. 워치리스트 초기화 — 스캐너 신호 종목 우선 성능 개선 (기존 대비): - - DB 트랜잭션: 1,000회 → 50회 (20× 감소) - - 동시 처리: 5 → 10 종목 + - DB 트랜잭션: DELETE+INSERT race-condition → 원자적 UPSERT + - 메모리: 전체 결과 적재 → Streaming(분석 즉시 저장) + - DB 엔진: 스크립트 자체 생성 → 공통 모듈 공유 """ start_time = datetime.now(KST) analysis_date = datetime.now(KST).date() @@ -371,45 +252,47 @@ async def batch_analyze_all_stocks(limit: int = None): # 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) + # 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: 병렬 분석 — 세마포어로 BATCH_CONCURRENCY 제한 - # asyncio 단일 스레드이므로 results 리스트 append는 thread-safe + # Step 3 + 4: 병렬 분석 & Streaming 저장 + # 분석 태스크 → queue → _streaming_saver (동시 실행) + queue: asyncio.Queue = asyncio.Queue(maxsize=BULK_SAVE_CHUNK * 3) 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)) + 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)) + # 저장기를 백그라운드로 먼저 시작 + saver_task = asyncio.create_task(_streaming_saver(queue, analysis_date)) + + # 분석 태스크 전체 실행 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)) + # sentinel 전송 → 저장기 종료 + await queue.put(None) + stats = await saver_task - # 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) + 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 6: DB 스캐너 (배치 완료 후 추가 API 호출 없이) + # Step 5: DB 스캐너 (배치 완료 후 추가 API 호출 없이) scanner_tickers = await run_db_scanner(analysis_date) - # Step 7: 워치리스트 초기화 — 스캐너 신호 종목 우선 + # Step 6: 워치리스트 초기화 — 스캐너 신호 종목 우선 try: from app.services.watchlist import initialize_watchlist async with AsyncSessionLocal() as session: @@ -423,7 +306,7 @@ async def batch_analyze_all_stocks(limit: int = None): # 결과 요약 duration = (datetime.now(KST) - start_time).total_seconds() - stats = korea_investment_api.get_monitor_stats() + api_stats = korea_investment_api.get_monitor_stats() logger.info("=" * 60) logger.info("✅ Batch analysis completed!") @@ -435,8 +318,8 @@ async def batch_analyze_all_stocks(limit: int = None): 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']) + api_stats['total_calls'], api_stats['total_errors'], + api_stats['error_rate'] * 100, api_stats['avg_duration']) logger.info("=" * 60) diff --git a/backend/batch_intraday_analysis.py b/backend/batch_intraday_analysis.py index 4ff1fc7..54724f5 100644 --- a/backend/batch_intraday_analysis.py +++ b/backend/batch_intraday_analysis.py @@ -21,21 +21,20 @@ KST = pytz.timezone('Asia/Seoul') sys.path.insert(0, str(Path(__file__).parent)) -from sqlalchemy import select, delete, update, func -from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession -from sqlalchemy.orm import sessionmaker -import logging +from sqlalchemy import select, update, func from app.core.config import settings +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.watchlist import ( WATCHLIST_MAX_SIZE, WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, ) -from batch_daily_analysis import run_db_scanner +import logging logging.basicConfig( level=logging.INFO, @@ -49,25 +48,6 @@ BATCH_CONCURRENCY = 5 # 워치리스트 외부 후보 재분석 상한 (10분 배치 내에 처리 가능한 수준) OUTSIDE_RESCORE_LIMIT = 30 -# 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=5, - max_overflow=10, -) - -AsyncSessionLocal = sessionmaker( - engine, - class_=AsyncSession, - expire_on_commit=False, -) - async def get_active_watchlist() -> list[tuple[str, str]]: """현재 활성 분석종목 목록 조회.""" @@ -84,28 +64,93 @@ async def get_active_watchlist() -> list[tuple[str, str]]: "아침 배치(batch_daily_analysis.py)가 먼저 실행되었는지 확인하세요." ) else: - logger.info(f"📋 분석종목 {len(rows)}개 대상") + logger.info("📋 분석종목 %d개 대상", len(rows)) return [(row.stock_code, row.stock_name) for row in rows] +# ------------------------------------------------------------------- +# analyze_update_and_manage 분리: 분석 / 저장 / 관리 +# ------------------------------------------------------------------- + +async def _save_analysis_and_history( + session, + stock_code: str, + stock_name: str, + result, + analysis_date: date, + now: datetime, +) -> None: + """stock_analysis UPSERT + watchlist_history 시계열 기록. + + DELETE+INSERT race condition 없이 원자적으로 갱신. + """ + # 1. stock_analysis UPSERT (race-condition-free) + await upsert_analysis(session, stock_code, stock_name, result, analysis_date) + + # 2. watchlist_history 시계열 기록 + session.add(WatchlistHistory( + stock_code=stock_code, + stock_name=stock_name, + recorded_at=now, + score=result.score, + current_price=result.current_price, + change_rate=result.change_rate, + rsi=result.rsi, + signals=result.signals, + )) + + +async def _manage_watchlist_membership( + session, + stock_code: str, + stock_name: str, + score: int | None, + now: datetime, +) -> bool: + """score 기준 미달 시 watchlist에서 제거. + + Returns: True if removed, False if retained. + """ + if score is not None and score < WATCHLIST_EXIT_SCORE: + await session.execute( + update(WatchlistStock) + .where( + WatchlistStock.stock_code == stock_code, + WatchlistStock.is_active == True, + ) + .values( + is_active=False, + removed_at=now, + removed_reason="score_drop", + ) + ) + logger.info( + "🚫 %s (%s) 분석종목 제거: score=%s < %s", + stock_code, stock_name, score, WATCHLIST_EXIT_SCORE, + ) + return True + return False + + async def analyze_update_and_manage( stock_code: str, stock_name: str, analysis_date: date, -) -> tuple[bool, int | None]: +) -> tuple[bool, int | None, bool]: """종목 재분석 → stock_analysis 갱신 → watchlist_history 저장 → watchlist 관리. - Returns: (success, score) + Returns: (success, score, was_removed) + was_removed: _manage_watchlist_membership이 실제로 비활성화 처리했으면 True """ try: result = await analyze_stock_for_batch(stock_code, stock_name) if not result: - logger.info(f"No result for {stock_code} (chart/price issue)") - return False, None + logger.info("No result for %s (chart/price issue)", stock_code) + return False, None, False # 5,000원 이하 저가주 → watchlist에서도 제거 - if result.current_price and result.current_price < 5000: - logger.debug(f"Skip {stock_code}: price={result.current_price:.0f}원 < 5,000원") + if not passes_price_filter(result): + logger.debug("Skip %s: price=%.0f원 < 5,000원", stock_code, result.current_price) async with AsyncSessionLocal() as session: await session.execute( update(WatchlistStock) @@ -120,83 +165,26 @@ async def analyze_update_and_manage( ) ) await session.commit() - return False, None + return False, None, True # 가격필터 제거도 was_removed=True now = datetime.now(KST) score = result.score async with AsyncSessionLocal() as session: - # 1. stock_analysis 갱신 (DELETE + INSERT) - await session.execute( - delete(StockAnalysis).where( - StockAnalysis.stock_code == stock_code, - StockAnalysis.analysis_date == analysis_date, - ) + await _save_analysis_and_history( + session, stock_code, stock_name, result, analysis_date, now + ) + was_removed = await _manage_watchlist_membership( + session, stock_code, stock_name, score, now ) - session.add(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=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, - )) - - # 2. watchlist_history 시계열 기록 - session.add(WatchlistHistory( - stock_code=stock_code, - stock_name=stock_name, - recorded_at=now, - score=score, - current_price=result.current_price, - change_rate=result.change_rate, - rsi=result.rsi, - signals=result.signals, - )) - - # 3. score가 기준 이하면 분석종목에서 제거 - if score is not None and score < WATCHLIST_EXIT_SCORE: - await session.execute( - update(WatchlistStock) - .where( - WatchlistStock.stock_code == stock_code, - WatchlistStock.is_active == True, - ) - .values( - is_active=False, - removed_at=now, - removed_reason="score_drop", - ) - ) - logger.info( - f"🚫 {stock_code} ({stock_name}) 분석종목 제거: " - f"score={score} < {WATCHLIST_EXIT_SCORE}" - ) - await session.commit() - logger.info(f"✅ {stock_code} ({stock_name}): score={score}") - return True, score + logger.info("✅ %s (%s): score=%s", stock_code, stock_name, score) + return True, score, was_removed except Exception as e: - logger.error(f"❌ Failed {stock_code}: {e}") - return False, None + logger.error("❌ Failed %s: %s", stock_code, e) + return False, None, False async def rescore_outside_candidates( @@ -232,7 +220,7 @@ async def rescore_outside_candidates( return 0 total = len(candidates) - logger.info(f"🔍 워치리스트 외부 후보 {total}개 재분석 시작") + logger.info("🔍 워치리스트 외부 후보 %d개 재분석 시작", total) semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) @@ -242,48 +230,19 @@ async def rescore_outside_candidates( res = await analyze_stock_for_batch(code, name) if not res: return False - # 저가주 필터 (CLAUDE.md 섹션 6.4) - if res.current_price and res.current_price < 5000: - logger.debug(f"Skip {code}: price={res.current_price:.0f}원 < 5,000원") + # 저가주 필터 (CLAUDE.md 5절 가격 필터) + 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 session.execute( - delete(StockAnalysis).where( - StockAnalysis.stock_code == code, - StockAnalysis.analysis_date == analysis_date, - ) - ) - session.add(StockAnalysis( - stock_code=code, - stock_name=name, - analysis_date=analysis_date, - current_price=res.current_price, - change_rate=res.change_rate, - volume=res.volume, - rsi=res.rsi, - ma5=res.ma5, - ma20=res.ma20, - ma60=res.ma60, - stochastic_k=res.stochastic_k, - stochastic_d=res.stochastic_d, - bollinger_upper=res.bollinger_upper, - bollinger_lower=res.bollinger_lower, - bollinger_middle=res.bollinger_middle, - signals=res.signals, - score=res.score, - ml_score=res.ml_score, - ml_confidence=res.ml_confidence, - ml_signal=res.ml_signal, - atr_14=res.atr_14, - adx_14=res.adx_14, - )) + await upsert_analysis(session, code, name, res, analysis_date) await session.commit() - logger.debug(f" [{i}/{total}] {code} ({name}): score={res.score}") + logger.debug(" [%d/%d] %s (%s): score=%s", i, total, code, name, res.score) return True except Exception as e: - logger.error(f"❌ 외부 후보 재분석 실패 {code}: {e}") + logger.error("❌ 외부 후보 재분석 실패 %s: %s", code, e) return False tasks = [ @@ -292,7 +251,7 @@ async def rescore_outside_candidates( ] results = await asyncio.gather(*tasks, return_exceptions=True) success = sum(1 for r in results if r is True) - logger.info(f"✅ 외부 후보 재분석 완료: {success}/{total}개 갱신") + logger.info("✅ 외부 후보 재분석 완료: %d/%d개 갱신", success, total) return success @@ -316,7 +275,7 @@ async def refresh_watchlist(analysis_date: date, current_codes: set[str]) -> int slots = WATCHLIST_MAX_SIZE - active_count if slots <= 0: - logger.debug(f"Watchlist 최대치 도달 ({active_count}/{WATCHLIST_MAX_SIZE})") + logger.debug("Watchlist 최대치 도달 (%d/%d)", active_count, WATCHLIST_MAX_SIZE) return 0 # stock_analysis에서 고점수 종목 중 현재 watchlist에 없는 것 조회 @@ -353,8 +312,8 @@ async def refresh_watchlist(analysis_date: date, current_codes: set[str]) -> int await session.commit() logger.info( - f"➕ 분석종목 {len(candidates)}개 신규 추가 " - f"(score ≥ {WATCHLIST_ENTRY_SCORE}, 현재 {active_count + len(candidates)}개)" + "➕ 분석종목 %d개 신규 추가 (score ≥ %d, 현재 %d개)", + len(candidates), WATCHLIST_ENTRY_SCORE, active_count + len(candidates), ) return len(candidates) @@ -369,8 +328,8 @@ async def main(): start_time = datetime.now(KST) analysis_date = datetime.now(KST).date() logger.info( - f"⚡ [Intraday] {start_time.strftime('%H:%M:%S')} 시작 — " - f"분석종목 재분석 (진입≥{WATCHLIST_ENTRY_SCORE}, 제거<{WATCHLIST_EXIT_SCORE})" + "⚡ [Intraday] %s 시작 — 분석종목 재분석 (진입≥%d, 제거<%d)", + start_time.strftime("%H:%M:%S"), WATCHLIST_ENTRY_SCORE, WATCHLIST_EXIT_SCORE, ) # 활성 분석종목 조회 @@ -385,9 +344,9 @@ async def main(): # 병렬 재분석 semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) - async def analyze_one(i: int, code: str, name: str) -> tuple[bool, int | None]: + async def analyze_one(i: int, code: str, name: str) -> tuple[bool, int | None, bool]: async with semaphore: - logger.info(f"[{i}/{total}] {code} ({name})") + logger.info("[%d/%d] %s (%s)", i, total, code, name) return await analyze_update_and_manage(code, name, analysis_date) tasks = [ @@ -396,15 +355,11 @@ async def main(): ] results = await asyncio.gather(*tasks, return_exceptions=True) - # removed 카운터 수정: analyze_update_and_manage는 제거 시에도 True 반환하므로 - # success(=분석 성공)에 removed가 포함됨 → retained = success - removed 로 표시 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 r[0] is True - and r[1] is not None and r[1] < WATCHLIST_EXIT_SCORE + 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 ) - retained = analyzed_ok - removed # 워치리스트에 실제 유지된 종목 수 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)) @@ -419,9 +374,8 @@ async def main(): now_kst = datetime.now(KST) async with AsyncSessionLocal() as session: # 현재 활성 종목 수 확인 - from sqlalchemy import func as sa_func count_result = await session.execute( - select(sa_func.count()) + select(func.count()) .select_from(WatchlistStock) .where(WatchlistStock.is_active == True) ) @@ -453,7 +407,7 @@ async def main(): await session.commit() if added_scanner: - logger.info(f"📡 장중 스캐너 신규 추가: {added_scanner}개 (RSI≤30+BB하단)") + logger.info("📡 장중 스캐너 신규 추가: %d개 (RSI≤30+BB하단)", added_scanner) current_codes |= scanner_new # refresh_watchlist 중복 방지용 갱신 # 고점수 신규 종목 추가 (스캐너 추가분 제외) @@ -461,12 +415,11 @@ async def main(): duration = (datetime.now(KST) - start_time).total_seconds() logger.info( - f"✅ [Intraday] 완료: {retained}/{total} 유지, " - f"{removed} 제거, +{added} 신규(score), +{added_scanner} 신규(스캐너), " - f"{fail} 실패, {error} 오류 — {duration:.1f}초" + "✅ [Intraday] 완료: %d/%d 유지, %d 제거, +%d 신규(score), +%d 신규(스캐너), " + "%d 실패, %d 오류 — %.1f초", + retained, total, removed, added, added_scanner, fail, error, duration, ) - if __name__ == "__main__": asyncio.run(main()) diff --git a/backend/migrations/fix_analysis_date_type.sql b/backend/migrations/fix_analysis_date_type.sql new file mode 100644 index 0000000..8e78328 --- /dev/null +++ b/backend/migrations/fix_analysis_date_type.sql @@ -0,0 +1,66 @@ +-- Fix: stock_analysis.analysis_date 컬럼 타입 TIMESTAMP → DATE +-- 배경: +-- analysis_date가 TIMESTAMP이면 datetime.now() 저장 시 매 실행마다 서로 다른 타임스탬프가 +-- 생성되어 unique constraint가 무의미해지고 중복 행이 무한 누적된다. +-- DATE 타입으로 변경하면 동일 (stock_code, 날짜) 쌍의 ON CONFLICT가 올바르게 동작한다. +-- +-- 실행: psql -d autostock_db -f migrations/fix_analysis_date_type.sql +-- 주의: 실행 전 DB 백업 권장 +-- 배치 스크립트 중단 후 실행할 것 + +DO $$ +DECLARE + col_type TEXT; +BEGIN + -- 현재 analysis_date 컬럼 타입 확인 + SELECT data_type INTO col_type + FROM information_schema.columns + WHERE table_name = 'stock_analysis' + AND column_name = 'analysis_date'; + + IF col_type IS NULL THEN + RAISE EXCEPTION 'stock_analysis 테이블 또는 analysis_date 컬럼이 존재하지 않습니다.'; + END IF; + + IF col_type = 'date' THEN + RAISE NOTICE 'analysis_date 컬럼이 이미 DATE 타입입니다. 마이그레이션 불필요.'; + RETURN; + END IF; + + IF col_type NOT IN ('timestamp without time zone', 'timestamp with time zone') THEN + RAISE EXCEPTION '예상치 못한 컬럼 타입: %. 수동 확인이 필요합니다.', col_type; + END IF; + + RAISE NOTICE '현재 타입: %. TIMESTAMP → DATE 변환을 시작합니다.', col_type; + + -- Step 1: 기존 unique constraint 제거 (컬럼 타입 변경 전 반드시 삭제) + ALTER TABLE stock_analysis DROP CONSTRAINT IF EXISTS uq_stock_analysis_code_date; + RAISE NOTICE 'unique constraint uq_stock_analysis_code_date 삭제 완료'; + + -- Step 2: 동일 (stock_code, analysis_date::date) 쌍의 중복 행 제거 + -- TIMESTAMP 시절에 datetime.now()로 저장된 경우 같은 날짜에 여러 행이 존재할 수 있음 + -- → 가장 최근 id만 남기고 나머지 삭제 + DELETE FROM stock_analysis + WHERE id NOT IN ( + SELECT DISTINCT ON (stock_code, analysis_date::date) id + FROM stock_analysis + ORDER BY stock_code, analysis_date::date, id DESC + ); + RAISE NOTICE '중복 행 정리 완료 (남은 행 수: %)', (SELECT COUNT(*) FROM stock_analysis); + + -- Step 3: 컬럼 타입 변환 (TIMESTAMP → DATE) + ALTER TABLE stock_analysis + ALTER COLUMN analysis_date TYPE DATE + USING analysis_date::date; + RAISE NOTICE 'analysis_date 컬럼 타입 변환 완료: % → DATE', col_type; + + -- Step 4: unique constraint 재생성 + ALTER TABLE stock_analysis + ADD CONSTRAINT uq_stock_analysis_code_date + UNIQUE (stock_code, analysis_date); + RAISE NOTICE 'unique constraint uq_stock_analysis_code_date 재생성 완료'; + + -- Step 5: 인덱스 확인 (기존 인덱스는 타입 변경 시 자동 갱신됨) + RAISE NOTICE '마이그레이션 완료: analysis_date TIMESTAMP → DATE'; +END +$$; diff --git a/docker-compose.yml b/docker-compose.yml index 4a4f478..2733263 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -133,7 +133,7 @@ services: - POSTGRES_DB=${POSTGRES_DB} - POSTGRES_USER=${POSTGRES_USER} - POSTGRES_PASSWORD=${POSTGRES_PASSWORD} - - SCHEDULE=0 17 * * * # UTC 17:00 = KST 02:00 + - SCHEDULE=0 8,17 * * * - BACKUP_KEEP_DAYS=7 - BACKUP_KEEP_WEEKS=4 - BACKUP_KEEP_MONTHS=3 diff --git a/docker/init-db.sql b/docker/init-db.sql index d5712a1..371f627 100644 --- a/docker/init-db.sql +++ b/docker/init-db.sql @@ -180,5 +180,9 @@ CREATE INDEX IF NOT EXISTS idx_stock_analysis_stock_code ON stock_analysis(stock CREATE INDEX IF NOT EXISTS idx_stock_analysis_analysis_date ON stock_analysis(analysis_date); CREATE INDEX IF NOT EXISTS idx_stock_analysis_score_desc ON stock_analysis(score DESC); +-- UPSERT (ON CONFLICT) 동작을 위한 unique constraint (batch_daily/intraday UPSERT 필수) +ALTER TABLE stock_analysis + ADD CONSTRAINT uq_stock_analysis_code_date UNIQUE (stock_code, analysis_date); + CREATE TRIGGER update_stock_analysis_updated_at BEFORE UPDATE ON stock_analysis FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); diff --git a/frontend/src/pages/Dashboard.tsx b/frontend/src/pages/Dashboard.tsx index 6e0ec91..c69d83e 100644 --- a/frontend/src/pages/Dashboard.tsx +++ b/frontend/src/pages/Dashboard.tsx @@ -10,7 +10,7 @@ import './Dashboard.css' const formatTime = (isoStr?: string) => { if (!isoStr) return null; const d = new Date(isoStr); - return d.toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit', second: '2-digit', hour12: false }); + return d.toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit', second: '2-digit', hour12: false, timeZone: 'Asia/Seoul' }); }; // 마켓 운영 시간 체크 (한국장 + 미국장) @@ -70,10 +70,11 @@ const Dashboard = () => { const { data: mcpStatus } = useQuery({ queryKey: ['mcp-status'], queryFn: async () => (await api.get('/mcp/status')).data, - refetchInterval: isMarketOpen ? 5000 : false, + refetchInterval: isMarketOpen ? 30000 : false, // 5초 → 30초 (MCP 상태는 빈번한 폴링 불필요) + staleTime: 20000, }) - const { data: ruleBasedRecommendations } = useQuery({ + const { data: ruleBasedRecommendations, refetch: refetchRecommendations, isError: isRecommendationsError } = useQuery({ queryKey: ['rule-based-recommendations'], queryFn: async () => { const response = await api.get('/recommendations/rule-based', { @@ -95,8 +96,9 @@ const Dashboard = () => { return data; }, - refetchInterval: isMarketOpen ? 600000 : false, // 10분마다 배치 실행 - // 장외에도 초기 로드는 실행 (DB 조회는 빠름) + refetchInterval: isMarketOpen ? 120000 : 300000, // 장중 2분, 장외 5분 + staleTime: isMarketOpen ? 110000 : 290000, // refetchInterval보다 약간 짧게 → 재진입 시 즉시 표시 + refetchOnWindowFocus: false, // 탭 전환 시 불필요한 재요청 방지 (자동매매 중복 트리거 차단) }) const { data: chartData, isLoading: chartLoading } = useQuery({ @@ -201,13 +203,11 @@ const Dashboard = () => { } }, [wsMessages, queryClient]) - // 보유종목 + 분석종목 전체 코드 수집 (현재가 일괄 조회용) + // 보유종목 코드만 수집 (현재가 일괄 실시간 조회용) + // 분석종목은 stock_analysis.current_price (DB값)를 폴백으로 사용하므로 실시간 조회 불필요 const allStockCodes = useMemo(() => { - const codes = new Set() - ;(portfolio || []).forEach((s: any) => codes.add(s.stock_code)) - ;(ruleBasedRecommendations?.analyzed_stock_list || []).forEach((s: any) => codes.add(s.stock_code)) - return Array.from(codes) - }, [portfolio, ruleBasedRecommendations]) + return (portfolio || []).map((s: any) => s.stock_code) + }, [portfolio]) // 현재가 일괄 실시간 조회 (보유종목·분석종목 동기화) const { data: livePrices } = useQuery({ @@ -698,10 +698,23 @@ const Dashboard = () => { {ruleBasedRecommendations?.analyzed_stocks || 0} +
- {ruleBasedRecommendations === undefined ? ( + {isRecommendationsError ? ( +
+

⚠️ API 오류

+

백엔드 로그 확인 필요

+ +
+ ) : ruleBasedRecommendations === undefined ? (

분석 종목 로딩 중...

@@ -790,8 +803,8 @@ const Dashboard = () => { {journals && journals.length > 0 ? ( journals.map((journal: any) => { const journalDate = new Date(journal.journal_date || journal.created_at); - const dateStr = journalDate.toLocaleDateString('ko-KR', { month: '2-digit', day: '2-digit' }); - const timeStr = journalDate.toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit', hour12: false }); + const dateStr = journalDate.toLocaleDateString('ko-KR', { month: '2-digit', day: '2-digit', timeZone: 'Asia/Seoul' }); + const timeStr = journalDate.toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit', hour12: false, timeZone: 'Asia/Seoul' }); const isGoodMood = journal.mood === '좋음' || journal.mood === 'good' || journal.mood === 'positive'; const isBadMood = journal.mood === '나쁨' || journal.mood === 'bad' || journal.mood === 'negative'; const tags: string[] = Array.isArray(journal.tags) ? journal.tags : [];