from fastapi import APIRouter, HTTPException, Query, Depends from typing import List, Optional from datetime import datetime, date, timedelta import asyncio import logging import httpx from collections import namedtuple from app.core.database import get_db, AsyncSessionLocal from app.models.models import StockAnalysis from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.core.config import settings from app.services.market_analyzer import ( check_market_regime, get_all_tradable_stocks, analyze_stock, RecommendationCriteria, StockRecommendation, KST ) from app.services.trade_executor import execute_auto_trade_background, execute_auto_trade from app.services.mean_reversion_strategy import MarketRegime router = APIRouter() logger = logging.getLogger(__name__) # 매매 규칙 namedtuple 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 변환 헬퍼 (중복 제거).""" 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 @router.get("/recommendations") async def get_recommendations( use_macd: bool = Query(False, description="MACD 지표 사용"), use_rsi: bool = Query(False, description="RSI 지표 사용"), use_bollinger: bool = Query(False, description="볼린저밴드 사용"), use_moving_average: bool = Query(False, description="이동평균선 사용"), use_stochastic: bool = Query(False, description="스토캐스틱 사용"), use_adx: bool = Query(False, description="ADX 사용"), use_obv: bool = Query(False, description="거래량(OBV) 사용"), use_volatility_breakout: bool = Query(False, description="변동성 돌파 전략 사용"), min_volume_ratio: float = Query(1.0, description="최소 거래량 비율 (평균 대비)"), max_disparity: float = Query(200.0, description="최대 이격도 (20일선 대비 %)"), max_stocks: int = Query(100, description="분석할 최대 종목 수 (기본 100개)"), ): market_regime = await check_market_regime() logger.info(f"📊 시장 상태: {market_regime['message']}") criteria = RecommendationCriteria( use_macd=use_macd, use_rsi=use_rsi, use_bollinger=use_bollinger, use_moving_average=use_moving_average, use_stochastic=use_stochastic, use_adx=use_adx, use_obv=use_obv, use_volatility_breakout=use_volatility_breakout, min_volume_ratio=min_volume_ratio, max_disparity=max_disparity ) if not any([use_macd, use_rsi, use_bollinger, use_moving_average, use_stochastic, use_adx, use_obv, use_volatility_breakout]): criteria.use_rsi = True criteria.use_moving_average = True criteria.use_volatility_breakout = True all_stocks = await get_all_tradable_stocks() if not all_stocks: return { "total": 0, "analyzed_stocks": 0, "criteria": criteria.model_dump(), "recommendations": [], "message": "거래 가능한 종목을 조회할 수 없습니다" } stocks_to_analyze = all_stocks[:max_stocks] logger.info(f"Analyzing {len(stocks_to_analyze)} stocks with criteria: {criteria.model_dump()}") batch_size = 5 delay_between_batches = 1.2 all_results = [] for i in range(0, len(stocks_to_analyze), batch_size): batch = stocks_to_analyze[i:i + batch_size] tasks = [analyze_stock(code, name, criteria) for code, name in batch] batch_results = await asyncio.gather(*tasks, return_exceptions=True) all_results.extend(batch_results) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) regime_min_score = market_regime.get("min_score", 1) recommendations = [ r for r in all_results if r is not None and not isinstance(r, Exception) and r.current_price >= 3000 and r.score >= regime_min_score ] recommendations.sort(key=lambda x: x.score, reverse=True) return { "total": len(recommendations), "analyzed_stocks": len(stocks_to_analyze), "criteria": criteria.model_dump(), "market_regime": market_regime, "recommendations": recommendations[:20] } @router.get("/rule-based") async def get_rule_based_recommendations( realtime: bool = Query(False, description="실시간 분석 여부 (느림, 30초)"), auto_trade: bool = Query(True, description="자동매매 실행 여부"), db: AsyncSession = Depends(get_db) ): if realtime: result = await get_rule_based_recommendations_realtime() else: result = await get_rule_based_recommendations_from_db(db) if auto_trade and result.get("recommendations_by_rule"): rec_snapshot = result async def _bg_trade(): async with AsyncSessionLocal() as bg_db: await execute_auto_trade_background(rec_snapshot, bg_db) asyncio.create_task(_bg_trade()) return result async def _load_active_rules_from_db(db: AsyncSession) -> list: 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: 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}") 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: 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}") return [ Rule( id=1, name="RSI+볼린저 평균회귀", description="기본 평균회귀 규칙", 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_val = float(stock.rsi) if stock.rsi else None close = int(stock.current_price) if stock.current_price else 0 bb_lower = float(stock.bollinger_lower) if stock.bollinger_lower else None bb_upper = float(stock.bollinger_upper) if stock.bollinger_upper else None bb_middle = float(stock.bollinger_middle) if stock.bollinger_middle else None if not all([rsi_val, close, bb_lower, bb_upper, bb_middle]): return False if close < settings.MIN_PRICE_KRW: return False regime = MarketRegime.BULL if market_regime_str == "bull" else MarketRegime.BEAR adx_val = float(stock.adx_14) if getattr(stock, "adx_14", None) else None adx_strong_trend = adx_val is not None and adx_val > 30 if regime == MarketRegime.BULL: rsi_threshold = conditions.get("rsi_bull_max", 25) bb_lower_pct = conditions.get("bb_lower_pct", 3.0) / 100 bb_upper_pct = conditions.get("bb_upper_pct", 1.5) / 100 in_bb_zone = bb_lower * (1 - bb_lower_pct) <= close <= bb_lower * (1 + bb_upper_pct) if adx_strong_trend: return False return rsi_val <= rsi_threshold and in_bb_zone else: rsi_threshold = conditions.get("rsi_bear_max", 35) if adx_strong_trend: return False return rsi_val <= rsi_threshold and close <= bb_lower async def get_rule_based_recommendations_from_db(db: AsyncSession): active_rules = await _load_active_rules_from_db(db) today_kst = datetime.now(KST).date() since_date = today_kst - timedelta(days=1) result = await db.execute( select(StockAnalysis) .where(StockAnalysis.analysis_date >= since_date) .order_by(StockAnalysis.analysis_date.desc(), StockAnalysis.score.desc()) ) raw_stocks = result.scalars().all() seen = set() all_analyzed_stocks = [] for s in raw_stocks: if s.stock_code not in seen: seen.add(s.stock_code) all_analyzed_stocks.append(s) if not all_analyzed_stocks or len(all_analyzed_stocks) < 1: return { "total": 0, "active_rules": len(active_rules), "analyzed_stocks": 0, "analyzed_stock_list": [], "recommendations": [], "recommendations_by_rule": [], "rules_applied": [], "message": "⚠️ 배치 분석 데이터 없음. 'make batch-test' 명령어로 분석 실행 필요." } rule_matches = [] recommendations_by_rule = [] stock_to_rules = {} market_regime_info = await check_market_regime() current_market_regime = market_regime_info.get("regime_200d", "bear") for rule in active_rules: conditions = rule.conditions actions = rule.actions rule_matched_stocks = [] for stock in all_analyzed_stocks: matches = True if rule.rule_type == "MEAN_REVERSION": matches = is_mean_reversion_candidate(stock, conditions, current_market_regime) else: matches = False if not matches: continue rule_matched_stocks.append(stock) rule_matched_stocks.sort(key=lambda x: x.score or 0, reverse=True) rule_stocks = [] for stock in rule_matched_stocks[:10]: stock_dict = _stock_analysis_to_dict(stock, rule=rule, actions=actions) rule_stocks.append(stock_dict) rule_matches.append(stock_dict) if stock.stock_code not in stock_to_rules: stock_to_rules[stock.stock_code] = [] rule_exists = any(r["rule_id"] == rule.id for r in stock_to_rules[stock.stock_code]) if not rule_exists: stock_to_rules[stock.stock_code].append({ "rule_id": rule.id, "rule_name": rule.name, "priority": rule.priority, }) if rule_stocks: recommendations_by_rule.append({ "rule_id": rule.id, "rule_name": rule.name, "rule_description": rule.description, "rule_type": rule.rule_type, "priority": rule.priority, "matched_count": len(rule_matched_stocks), "stocks": rule_stocks, }) if not rule_matches and all_analyzed_stocks: already_added = set() fallback_stocks = [] for stock in all_analyzed_stocks[:20]: if stock.stock_code in already_added: continue already_added.add(stock.stock_code) stock_dict = _stock_analysis_to_dict(stock) fallback_stocks.append(stock_dict) rule_matches.append(stock_dict) if fallback_stocks: recommendations_by_rule.append({ "rule_id": None, "rule_name": "고점수 분석 종목", "rule_description": "현재 매매 규칙 조건 미충족 - 분석 점수 기준 상위 종목", "rule_type": "ANALYSIS", "priority": 99, "matched_count": len(fallback_stocks), "stocks": fallback_stocks, }) return { "total": len(rule_matches), "active_rules": len(active_rules), "analyzed_stocks": len(all_analyzed_stocks), "analyzed_stock_list": [ { "stock_code": s.stock_code, "stock_name": s.stock_name, "current_price": int(s.current_price) if s.current_price else 0, "change_rate": round(float(s.change_rate), 2) if s.change_rate else 0, "rsi": round(float(s.rsi), 2) if s.rsi else None, "stochastic_k": round(float(s.stochastic_k), 2) if s.stochastic_k else None, "signals": s.signals or [], "score": s.score or 0, } for s in all_analyzed_stocks ], "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], "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(): async with AsyncSessionLocal() as _db: active_rules = await _load_active_rules_from_db(_db) if not active_rules: criteria = RecommendationCriteria( use_rsi=True, use_moving_average=True ) all_stocks = await get_all_tradable_stocks() stocks_to_analyze = all_stocks[:20] batch_size = 5 delay_between_batches = 0.3 batch_results = [] for i in range(0, len(stocks_to_analyze), batch_size): batch = stocks_to_analyze[i:i + batch_size] tasks = [analyze_stock(code, name, criteria) for code, name in batch] results = await asyncio.gather(*tasks, return_exceptions=True) batch_results.extend(results) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) valid_recommendations = [ r for r in batch_results if r is not None and not isinstance(r, Exception) ] valid_recommendations.sort(key=lambda x: x.score, reverse=True) default_stocks = [] for rec in valid_recommendations[:10]: stock_dict = { "stock_code": rec.stock_code, "stock_name": rec.stock_name, "current_price": rec.current_price, "change_rate": rec.change_rate, "signals": rec.signals, "score": rec.score, "price_date": rec.price_date, "rule_id": None, "rule_name": "기본 추천", "rule_description": "RSI + 이동평균 기반 추천", "rule_priority": 1, "position_ratio": 0.15, "take_profit": 5.0, "stop_loss": 3.0 } default_stocks.append(stock_dict) return { "total": len(default_stocks), "active_rules": 0, "analyzed_stocks": len(stocks_to_analyze), "recommendations": default_stocks, "recommendations_by_rule": [{ "rule_id": None, "rule_name": "기본 추천 (RSI + 이동평균)", "rule_description": "RSI 40 이하 과매도 + 이동평균 정배열 종목", "rule_type": "BUY", "priority": 1, "matched_count": len(valid_recommendations), "stocks": default_stocks }] if default_stocks else [], "rules_applied": [], "message": "기본 지표로 분석되었습니다 (규칙 미설정)" } all_stocks = await get_all_tradable_stocks() stocks_to_analyze = all_stocks[:settings.ANALYSIS_STOCK_COUNT] try: from app.models.models import Portfolio as _PortfolioModel async with AsyncSessionLocal() as _pf_db: _pf_result = await _pf_db.execute( select(_PortfolioModel).where(_PortfolioModel.quantity > 0) ) _pf_stocks = _pf_result.scalars().all() existing_codes = {code for code, _ in stocks_to_analyze} for _pf in _pf_stocks: if _pf.stock_code not in existing_codes: stocks_to_analyze = list(stocks_to_analyze) stocks_to_analyze.append((_pf.stock_code, _pf.stock_name or _pf.stock_code)) existing_codes.add(_pf.stock_code) except Exception as _e: logger.warning(f"포트폴리오 종목 추가 실패 (non-critical): {_e}") criteria = RecommendationCriteria( use_rsi=True, use_moving_average=True, use_macd=True, use_bollinger=True, use_stochastic=True, for_batch=True, ) batch_size = 1 delay_between_batches = 1.5 all_analyzed_stocks = [] for i in range(0, len(stocks_to_analyze), batch_size): batch = stocks_to_analyze[i:i + batch_size] tasks = [analyze_stock(code, name, criteria) for code, name in batch] results = await asyncio.gather(*tasks, return_exceptions=True) all_analyzed_stocks.extend(results) if i + batch_size < len(stocks_to_analyze): await asyncio.sleep(delay_between_batches) valid_analyzed_stocks = [ r for r in all_analyzed_stocks if r is not None and not isinstance(r, Exception) ] 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: conditions = rule.conditions actions = rule.actions rule_matched_stocks = [] for stock in valid_analyzed_stocks: etf_keywords = ["ETF", "etf", "레버리지", "인버스", "KODEX", "선물"] if any(k in stock.stock_name for k in etf_keywords): continue 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 else: 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 or 0, reverse=True) rule_stocks = [] for rec in rule_matched_stocks[:10]: stock_with_rule = { **rec.model_dump(), "rule_id": rule.id, "rule_name": rule.name, "rule_description": rule.description, "rule_priority": rule.priority, "position_ratio": actions.get("position_sizing", {}).get("target_ratio", 0.10), "take_profit": actions.get("take_profit", 3.0), "stop_loss": actions.get("stop_loss", -1.5), "max_position_pct": actions.get("max_position_pct", 0.12), "max_add_buys": actions.get("max_add_buys", 3), "add_buy_cooldown_min": actions.get("add_buy_cooldown_min", 15) } rule_stocks.append(stock_with_rule) rule_matches.append(stock_with_rule) if rule_stocks: recommendations_by_rule.append({ "rule_id": rule.id, "rule_name": rule.name, "rule_description": rule.description, "rule_type": rule.rule_type, "priority": rule.priority, "matched_count": len(rule_matched_stocks), "stocks": rule_stocks }) rule_matches.sort(key=lambda x: (x["rule_priority"], -x["score"])) from app.services.korea_investment_api import korea_investment_api as kis_api top_recommendations = rule_matches[:20] for rec in top_recommendations: try: await asyncio.sleep(0.2) price_data = await kis_api.get_current_price(rec["stock_code"]) if price_data and price_data.get("output"): output = price_data["output"] rec["current_price"] = int(output.get("stck_prpr", rec["current_price"])) rec["change_rate"] = float(output.get("prdy_ctrt", rec.get("change_rate", 0))) except Exception as e: logger.warning(f"Failed to get real-time price for {rec['stock_code']}: {e}") for rule_group in recommendations_by_rule: for stock in rule_group["stocks"]: updated = next((r for r in top_recommendations if r["stock_code"] == stock["stock_code"]), None) if updated: stock["current_price"] = updated["current_price"] stock["change_rate"] = updated["change_rate"] from app.services.analysis_db_service import bulk_upsert_analysis as _bulk_upsert from datetime import date as _date _today = _date.today() _items = [(rec.stock_code, rec.stock_name, rec) for rec in valid_analyzed_stocks] async with AsyncSessionLocal() as _db: saved = await _bulk_upsert(_db, _items, _today) await _db.commit() return { "total": len(rule_matches), "active_rules": len(active_rules), "analyzed_stocks": len(stocks_to_analyze), "recommendations": top_recommendations, "recommendations_by_rule": recommendations_by_rule, "rules_applied": [{"id": r.id, "name": r.name, "priority": r.priority} for r in active_rules] } @router.post("/auto-trade") async def execute_auto_trading(db: AsyncSession = Depends(get_db)): try: recommendations_response = await get_rule_based_recommendations(realtime=True) if not recommendations_response.get("recommendations_by_rule"): return {"message": "추천 종목이 없습니다", "trades": []} trade_results = [] total_processed = 0 max_trades = 5 for rule_group in recommendations_response["recommendations_by_rule"]: if total_processed >= max_trades: break for stock in rule_group["stocks"][:settings.AUTO_TRADE_TOP_N_PER_RULE]: if total_processed >= max_trades: break try: result = await execute_auto_trade(stock, db) total_processed += 1 if result["action"] in ["BUY", "SELL"]: trade_results.append(result) await asyncio.sleep(0.5) except Exception as e: logger.error(f"Error executing auto-trade for {stock.get('stock_code')}: {e}") continue return { "message": f"자동 매매 완료: {len(trade_results)}건 실행 / {total_processed}개 분석", "trades": trade_results, "total_analyzed": total_processed } except Exception as e: logger.error(f"❌ 자동 매매 실행 오류: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get("/ai/{stock_code}") async def get_ai_recommendation(stock_code: str): from app.services.korea_investment_api import korea_investment_api try: price_data = {} try: price_data = await korea_investment_api.get_current_price(stock_code) except Exception as e: logger.warning(f"Failed to fetch price data locally: {e}") news_data = [] try: async with httpx.AsyncClient() as client: mcp_url = f"{settings.MCP_SERVER_URL}/tools/call" payload = { "tool_name": "get_news", "parameters": {"stock_code": stock_code, "limit": 3} } response = await client.post(mcp_url, json=payload, timeout=10.0) if response.status_code == 200: result = response.json() if result.get("status") == "success": data = result.get("result", {}) if "news" in data: news_data = data["news"] else: news_data = data except Exception as e: logger.warning(f"Failed to fetch news via MCP: {e}") analysis_result = "AI 분석을 수행할 수 없습니다." try: async with httpx.AsyncClient() as client: analyze_url = f"{settings.MCP_SERVER_URL}/analyze" market_summary = {} if "output" in price_data: output = price_data["output"] market_summary = { "current_price": output.get("stck_prpr"), "change_rate": output.get("prdy_ctrt"), "volume": output.get("acml_vol"), "high": output.get("stck_hgpr"), "low": output.get("stck_lwpr") } payload = { "analysis_type": "MARKET_ANALYSIS", "data": { "stock_code": stock_code, "market_data": market_summary, "news": news_data } } response = await client.post(analyze_url, json=payload, timeout=45.0) if response.status_code == 200: analysis_res = response.json() if "result" in analysis_res: analysis_result = analysis_res["result"] except Exception as e: logger.error(f"Failed to fetch AI analysis: {e}") return { "stock_code": stock_code, "analysis": analysis_result, "news": news_data, "market_data": price_data.get("output", {}) } except Exception as e: logger.error(f"AI Recommendation error: {e}") raise HTTPException(status_code=500, detail=str(e))