""" LLM Investment Advisor Service 提供投資策略的AI分析服務，包含： - 投資建議生成 - 風險評估 - 市場洞察 - Prompt工程管理 """ import os import json import time import logging import hashlib from typing import Dict, Any, Optional, Tuple from openai import OpenAI try: from config_openai import OPENAI_CONFIG, RATE_LIMITS except Exception: OPENAI_CONFIG = { 'api_key': os.environ.get('OPENAI_API_KEY') or os.environ.get('OPENROUTER_API_KEY', ''), 'model': os.environ.get('OPENAI_MODEL', os.environ.get('OPENROUTER_MODEL', 'gpt-4')), 'timeout': int(os.environ.get('LLM_TIMEOUT', '60')), 'max_tokens': int(os.environ.get('LLM_MAX_TOKENS', '2000')), 'temperature': float(os.environ.get('LLM_TEMPERATURE', '0.7')) } RATE_LIMITS = { 'max_retries': int(os.environ.get('LLM_MAX_RETRIES', '3')), 'retry_delay': int(os.environ.get('LLM_RETRY_DELAY', '2')) } # 設定日誌 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class PromptManager: """管理不同的Prompt模板""" def __init__(self): self.system_prompt = self._get_system_prompt() def _get_system_prompt(self) -> str: """系統提示詞 - 定義基金經理人專家角色""" return """你是一位頂尖的基金經理人與投資策略師，擁有超過20年橫跨牛熊市的實戰經驗。你的專長是將複雜的金融數據轉化為清晰、易於理解的語言，為大眾投資者提供專業見解。 你的溝通風格： - **權威且親切**：你的語氣充滿自信與專業，但同時讓非專業人士感到安心與信賴。 - **教育家精神**：你會用生動的比喻來解釋關鍵指標，例如將「夏普比率」比喻為投資的「性價比」，或將「最大回落」形容為「最顛簸的一段路」。 - **客觀中立**：你總是基於數據進行分析，同時點出潛在的盲點與風險。 你的任務是根據接下來提供的策略回測數據，撰寫一份專業的投資策略分析報告。""" def build_strategy_context(self, strategy_data: Dict[str, Any]) -> str: """將策略資料轉換為結構化context，並附帶指標提示""" return f""" 策略基本資訊： - 策略編號：{strategy_data.get('id', 'N/A')} - 策略名稱：{strategy_data.get('name', 'N/A')} - 投資目標：{strategy_data.get('role', 'N/A')} - 建立時間：{strategy_data.get('date', 'N/A')} - 建立者：{strategy_data.get('username', 'N/A')} 核心績效指標： - 年化報酬率：{strategy_data.get('annual_ret', 0):.2%} (衡量平均每年賺取多少利潤) - 年化波動率：{strategy_data.get('vol', 0):.2%} (衡量資產價值的波動風險，越高代表起伏越大) - 年化夏普比率：{strategy_data.get('annual_sr', 0):.2f} (衡量每一單位風險能換來多少報酬，可視為「投資CP值」) - 最大回落（MDD）：{strategy_data.get('mdd', 0):.2%} (衡量策略從最高點到最低點可能出現的最大虧損幅度) 進階參考指標： - Alpha值：{strategy_data.get('alpha', 0):.4f} (相對於市場基準的超額報酬能力) - Beta值：{strategy_data.get('beta', 0):.4f} (與市場波動的關聯性，大於1代表比市場更敏感) - VaR (10天, 95%信心)：{strategy_data.get('var10', 0):.2%} (預估在未來10天內，有95%的機率虧損不會超過此比例) - R-squared：{strategy_data.get('r2', 0):.4f} (策略表現有多大比例可由市場表現來解釋) 投資組合配置： - 包含資產：{', '.join(strategy_data.get('assets', []))} - 市場類型：{'台灣市場' if strategy_data.get('tw', True) else '美國市場'} """ def get_investment_advice_prompt(self, strategy_data: Dict[str, Any]) -> str: """生成結構化投資建議報告的完整Prompt - 使用外部模板""" try: # 導入 prompts 模組 from prompts.investment_advice import get_comprehensive_analysis_prompt return get_comprehensive_analysis_prompt(strategy_data) except ImportError: # 回退到內建模板 logger.warning("Could not import prompts.investment_advice, using built-in template") return self._get_builtin_prompt(strategy_data) def _get_builtin_prompt(self, strategy_data: Dict[str, Any]) -> str: """內建模板作為回退方案""" context = self.build_strategy_context(strategy_data) return f"""{self.system_prompt} {context} 請嚴格遵循以下結構，為這份投資策略撰寫一份專業分析報告： --- ### **【投資策略總評：給您的執行摘要】** *在這部分，請用2-3句話總結這個策略的核心特點與績效等級。直接點出它適合哪一種類型的投資者。* ### **【績效深度解析：白話解讀關鍵指標】** *在這部分，請選擇2-3個最重要的指標（例如：夏普比率、最大回落），並用生動的比喻解釋它們在此策略中的意義。* - **指標1**：[指標名稱] - [用比喻解釋其表現] - **指標2**：[指標名稱] - [用比喻解釋其表現] ### **【策略的亮點與潛在風險】** *客觀分析此策略的優缺點。* - **👍 亮點 (Strengths)**：[至少列出2點，例如：在特定市場環境下表現優異、風險控制得當等] - **🤔 潛在風險 (Weaknesses/Risks)**：[至少列出2點，例如：資產過於集中、對利率變化敏感等] ### **【給您的具體投資建議】** *提供清晰、可執行的建議。* 1. **核心觀點**：[明確指出「繼續持有」、「考慮調整」或「尋找替代方案」] 2. **優化建議**：[提出1-2項具體優化方向，例如：「可考慮納入 OOO 類型的資產以分散風險」或「建議將再平衡頻率調整為 X 個月一次」] 3. **風險管理**：[提醒投資者應該注意的市場訊號或事件] ### **【未來展望與提醒】** *提供一個前瞻性的總結，並附上免責聲明。* --- 請用繁體中文回答，確保報告結構完整、語氣專業且易於理解。""" class LLMInvestmentAdvisor: """LLM投資顧問主類""" def __init__(self, api_key: Optional[str] = None): """初始化LLM服務""" provider = os.environ.get('LLM_PROVIDER', 'openai').lower() base_url = None default_headers = None if provider == 'openrouter': self.api_key = api_key or os.environ.get('OPENROUTER_API_KEY') or OPENAI_CONFIG.get('api_key') base_url = os.environ.get('OPENROUTER_BASE_URL', 'https://openrouter.ai/api/v1') self.model = os.environ.get('OPENROUTER_MODEL', OPENAI_CONFIG.get('model', 'openrouter/auto')) default_headers = { 'HTTP-Referer': os.environ.get('OPENROUTER_REFERER', ''), 'X-Title': os.environ.get('OPENROUTER_TITLE', 'TPM') } else: self.api_key = api_key or os.environ.get('OPENAI_API_KEY') or OPENAI_CONFIG.get('api_key') self.model = os.environ.get('OPENAI_MODEL', OPENAI_CONFIG.get('model', 'gpt-4')) if not self.api_key or self.api_key == 'your-api-key-here': raise ValueError("LLM API key is required. Please set OPENAI_API_KEY or OPENROUTER_API_KEY.") self.client = OpenAI( api_key=self.api_key, base_url=base_url, timeout=OPENAI_CONFIG.get('timeout', 30), default_headers=default_headers ) self.prompt_manager = PromptManager() self.max_tokens = OPENAI_CONFIG.get('max_tokens', 2000) self.temperature = OPENAI_CONFIG.get('temperature', 0.7) # 快取設定 self.cache: Dict[str, Tuple[str, float]] = {} self.cache_timeout = 3600 # 1小時快取 self.mock_mode = os.environ.get('MOCK_LLM', 'false').lower() in ('1', 'true', 'yes') def _is_cache_valid(self, cache_time: float) -> bool: """檢查快取是否有效""" return time.time() - cache_time < self.cache_timeout def generate_advice(self, strategy_id: str, strategy_data: Dict[str, Any]) -> str: """生成投資建議 Args: strategy_id: 策略ID strategy_data: 策略資料字典 Returns: str: 投資建議文本 """ stable_payload = json.dumps(strategy_data, sort_keys=True).encode('utf-8') cache_digest = hashlib.sha256(stable_payload).hexdigest() cache_key = f"advice_{strategy_id}_{cache_digest}" # 檢查快取 if cache_key in self.cache: advice, cache_time = self.cache[cache_key] if self._is_cache_valid(cache_time): logger.info(f"Returning cached advice for strategy {strategy_id}") return advice try: # 構建prompt prompt = self.prompt_manager.get_investment_advice_prompt(strategy_data) # Mock 模式：不呼叫外部API if self.mock_mode: logger.info("MOCK_LLM enabled, returning fallback advice without external API call") response = self._get_fallback_advice(strategy_data) else: # 調用LLM API response = self._call_openai_with_retry(prompt) # 快取結果 self.cache[cache_key] = (response, time.time()) logger.info(f"Generated new advice for strategy {strategy_id}") return response except Exception as e: logger.error(f"Error generating advice for strategy {strategy_id}: {str(e)}") return self._get_fallback_advice(strategy_data) def clear_cache(self): """清除快取""" self.cache.clear() logger.info("LLM advice cache cleared") def _call_openai_with_retry(self, prompt: str) -> str: """調用OpenAI API，包含重試機制 Args: prompt: 完整的prompt Returns: str: API回應內容 """ max_retries = RATE_LIMITS.get('max_retries', 3) retry_delay = RATE_LIMITS.get('retry_delay', 2) for attempt in range(max_retries): try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": "你是一位專業的投資顧問。"}, {"role": "user", "content": prompt} ], max_tokens=self.max_tokens, temperature=self.temperature, top_p=0.9 ) return response.choices[0].message.content.strip() except Exception as e: status_code = getattr(e, 'status_code', None) is_rate_limited = status_code == 429 or 'rate limit' in str(e).lower() if attempt < max_retries - 1 and (is_rate_limited or True): wait_time = retry_delay ** attempt logger.warning(f"LLM API error, retrying in {wait_time}s... (attempt {attempt + 1}/{max_retries}): {str(e)}") time.sleep(wait_time) continue logger.error(f"LLM API error after {max_retries} attempts: {str(e)}") raise e def _get_fallback_advice(self, strategy_data: Dict[str, Any]) -> str: """獲取fallback投資建議""" annual_ret = strategy_data.get('annual_ret', 0) vol = strategy_data.get('vol', 0) sharpe = strategy_data.get('annual_sr', 0) return f""" 基於您的投資策略數據，我提供以下初步分析： 📊 **表現評估**： - 年化報酬率：{annual_ret:.2%} - {'表現良好' if annual_ret > 0.1 else '有改進空間'} - 年化波動率：{vol:.2%} - {'風險適中' if vol < 0.2 else '風險較高'} - 夏普比率：{sharpe:.2f} - {'風險調整後報酬優良' if sharpe > 1 else '有改進空間'} 💡 **初步建議**： 1. 持續監控市場變化 2. 定期檢視投資組合配置 3. 考慮分散投資降低風險 *注意：此為預設建議，如需詳細分析請稍後再試。* """ # Duplicate clear_cache removed # 創建全域實例 llm_advisor = None def get_llm_advisor() -> LLMInvestmentAdvisor: """獲取LLM顧問實例""" global llm_advisor if llm_advisor is None: llm_advisor = LLMInvestmentAdvisor() return llm_advisor