import os import json import numpy as np from modules.services.ollama import OllamaManager from modules.analysis.technical import TechnicalAnalyzer from modules.analysis.deep_learning import PricePredictor # [최적화] 워커 프로세스별 전역 변수 (LSTM 모델 캐싱) _lstm_predictor = None def get_predictor(): """워커 프로세스 내에서 PricePredictor 인스턴스를 싱글톤으로 관리""" global _lstm_predictor if _lstm_predictor is None: print(f"🧩 [Worker {os.getpid()}] Initializing LSTM Predictor...") _lstm_predictor = PricePredictor() return _lstm_predictor def analyze_stock_process(ticker, prices, news_items, investor_trend=None): """ [CPU Intensive] 기술적 분석 및 AI 판단을 수행하는 함수 (ProcessPoolExecutor에서 실행됨) """ try: print(f"⚙️ [Bot Process] Analyzing {ticker} ({len(prices)} candles)...") # 1. 기술적 지표 계산 current_price = prices[-1] if prices else 0 # [수정] 변동성, 거래량 비율, MA 정보 반환 tech_score, rsi, volatility, vol_ratio, ma_info = TechnicalAnalyzer.get_technical_score(current_price, prices, volume_history=None) # 2. LSTM 주가 예측 # [최적화] 전역 캐시된 Predictor 사용 lstm_predictor = get_predictor() if lstm_predictor: lstm_predictor.training_status['current_ticker'] = ticker pred_result = lstm_predictor.train_and_predict(prices) lstm_score = 0.5 # 중립 ai_confidence = 0.5 ai_loss = 1.0 if pred_result: ai_confidence = pred_result.get('confidence', 0.5) ai_loss = pred_result.get('loss', 1.0) # 상승/하락 예측에 따라 점수 조정 (신뢰도 반영) # 최대 5% 변동폭까지 반영 change_magnitude = min(abs(pred_result['change_rate']), 5.0) / 5.0 if pred_result['trend'] == 'UP': # 상승 예측 시: 기본 0.5 + (강도 * 신뢰도 * 0.4) -> 최대 0.9 lstm_score = 0.5 + (change_magnitude * ai_confidence * 0.4) else: # 하락 예측 시: 기본 0.5 - (강도 * 신뢰도 * 0.4) -> 최소 0.1 lstm_score = 0.5 - (change_magnitude * ai_confidence * 0.4) lstm_score = max(0.0, min(1.0, lstm_score)) # [신규] 수급 분석 (외인/기관) investor_score = 0.0 frgn_net_buy = 0 orgn_net_buy = 0 consecutive_frgn_buy = 0 if investor_trend: # 최근 5일 합산 for day in investor_trend: frgn_net_buy += day['foreigner'] orgn_net_buy += day['institutional'] if day['foreigner'] > 0: consecutive_frgn_buy += 1 # 외인 수급 점수 (단순화) if frgn_net_buy > 0: investor_score += 0.05 if consecutive_frgn_buy >= 3: investor_score += 0.05 if investor_score > 0: print(f" 💰 [Investor] Foreign Buy Detected (Net: {frgn_net_buy})") # 3. AI 뉴스 분석 # pred_result가 None일 경우 기본값 사용 if pred_result: pred_price = pred_result.get('predicted', 0) pred_change = pred_result.get('change_rate', 0) else: pred_price = current_price pred_change = 0.0 ollama = OllamaManager() prompt = f""" [System Instruction] 1. Role: You are a Expert Quant Trader with 20 years of experience. 2. Market Data: - Technical Score: {tech_score:.2f} (RSI: {rsi:.2f}) - Moving Average: {ma_info['trend']} (Price is {ma_info['position']}) - AI Prediction: {pred_price:.0f} KRW ({pred_change}%) - AI Confidence: {ai_confidence:.2f} (Loss: {ai_loss:.4f}) - Investor Trend (5 Days): Foreigner Net Buy {frgn_net_buy}, Institutional Net Buy {orgn_net_buy} 3. Strategy: - If Foreigners are buying AND Trend is UP -> Strong BUY. - If AI Confidence > 0.8 and Trend is UP -> Strong BUY. - If MA is Bullish (Golden Alignment) -> Positive Signal. - If Price is above MA20 -> Support Uptrend. - If Trend is DOWN -> SELL/AVOID. 4. Task: Analyze the news and combine with market data to decide sentiment. News Data: {json.dumps(news_items, ensure_ascii=False)} Response (JSON): {{ "sentiment_score": 0.8, "reason": "Foreigners buying and Golden Cross detected." }} """ ai_resp = ollama.request_inference(prompt) sentiment_score = 0.5 try: data = json.loads(ai_resp) sentiment_score = float(data.get("sentiment_score", 0.5)) except: pass # 4. 통합 점수 (동적 가중치) # AI 신뢰도가 높으면 AI 비중을 대폭 상향 if ai_confidence >= 0.85: w_tech, w_news, w_ai = 0.2, 0.2, 0.6 print(f" 🤖 [High Confidence] AI Weight Boosted to 60%") else: w_tech, w_news, w_ai = 0.4, 0.3, 0.3 total_score = (w_tech * tech_score) + (w_news * sentiment_score) + (w_ai * lstm_score) # [수신] 수급 가산점 추가 (최대 +0.1) total_score += investor_score total_score = min(total_score, 1.0) decision = "HOLD" # [신규] 강한 단일 신호 매수 로직 (기준 강화) strong_signal = False strong_reason = "" if tech_score >= 0.80: strong_signal = True strong_reason = "Super Strong Technical" elif lstm_score >= 0.80 and ai_confidence >= 0.8: strong_signal = True strong_reason = f"High Confidence AI Buy (Conf: {ai_confidence})" elif sentiment_score >= 0.85: strong_signal = True strong_reason = "Strong News Sentiment" elif investor_score >= 0.1 and total_score >= 0.6: # 외인 수급이 좋고 전체 점수 양호 strong_signal = True strong_reason = "Strong Foreigner Buying" if strong_signal: decision = "BUY" print(f" 🎯 [{strong_reason}] Overriding to BUY!") elif total_score >= 0.60: # (0.5 -> 0.6 상향 조정으로 보수적 접근) decision = "BUY" elif total_score <= 0.30: decision = "SELL" print(f" └─ Scores: Tech={tech_score:.2f} News={sentiment_score:.2f} LSTM={lstm_score:.2f} → Total={total_score:.2f} [{decision}]") return { "ticker": ticker, "score": total_score, "tech": tech_score, "sentiment": sentiment_score, "lstm_score": lstm_score, "volatility": volatility, "volume_ratio": vol_ratio, "prediction": pred_result, "decision": decision, "current_price": current_price, "ma_info": ma_info } except Exception as e: print(f"❌ [Worker Error] Failed to analyze {ticker}: {e}") import traceback traceback.print_exc() # 기본 실패 응답 반환 (프로세스 크래시 방지) return { "ticker": ticker, "score": 0.0, "decision": "HOLD", "current_price": 0, "error": str(e) }