Investment Agent 規格文檔
概述
Investment Agent 是 Cortex 的第一個應用實例，提供股票技術分析能力。
核心功能: - 單一股票分析（台股全部 + 美股七巨頭） - 28 種技術指標計算與解讀（62 個欄位） - Meta-Agent 動態生成分析策略 - VRR 驗證分析可靠性 - 對話式互動 + 圖表視覺化
資料範圍: - 台灣股票: ~1,700 支（全部上市上櫃） - 美國股票: 7 支（AAPL, MSFT, GOOGL, AMZN, NVDA, META, TSLA） - 歷史資料: 2020-01-01 至今 - 更新頻率: 每日 06:00 AM
系統架構
模組劃分
investment-agent/
├── backend/
│ ├── api/ # FastAPI 端點
│ ├── agents/ # 領域 Agents
│ ├── data_pipeline/ # 資料服務
│ ├── database/ # 資料庫層
│ ├── knowledge/ # 投資領域知識
│ └── config/ # 配置檔
│
└── streamlit_ui/ # Streamlit 前端
1. Agent 層 (backend/agents/)
1.1 StockAnalyzer (主要 Agent)
檔案: backend/agents/analyzer.py
職責: 協調整個分析流程，使用 Meta-Agent 生成策略
類別定義:
from core.agent_framework import BaseAgent, MetaAgent, VRREngine
from core.llm import GeminiLLM
class StockAnalyzer(BaseAgent):
"""股票分析 Agent"""
def __init__(
self,
llm: GeminiLLM,
meta_agent: MetaAgent,
vrr_engine: VRREngine,
rag_service: RAGService,
db_session: AsyncSession
):
"""
初始化股票分析 Agent
Args:
llm: Gemini LLM 實例
meta_agent: Meta-Agent 實例
vrr_engine: VRR Engine 實例
rag_service: RAG 服務（投資知識庫）
db_session: 資料庫 session
"""
super().__init__(name="stock_analyzer", llm=llm, config={})
self.meta_agent = meta_agent
self.vrr_engine = vrr_engine
self.rag = rag_service
self.db = db_session
async def analyze(
self,
ticker: str,
analysis_type: str = "comprehensive",
timeframe: str = "3_months"
) -> AnalysisResult:
"""
分析股票
Args:
ticker: 股票代碼 (e.g., "2330.TW", "AAPL")
analysis_type: 分析類型
- "comprehensive": 全面分析
- "technical": 純技術面
- "trend": 趨勢分析
timeframe: 時間範圍
- "1_month", "3_months", "6_months", "1_year"
Returns:
AnalysisResult:
- summary: 分析摘要
- trend: 趨勢判斷 (bullish/bearish/neutral)
- key_levels: 關鍵價位
- indicators: 技術指標解讀
- risks: 風險提示
- confidence: 分析信心分數
- chart_data: 圖表資料
Flow:
1. Meta-Agent 生成分析策略
2. 從 DB 讀取 OHLC 和技術指標
3. 從 RAG 獲取相關投資知識
4. LLM 生成分析
5. VRR Engine 驗證與改進
6. 返回結果
"""
# Step 1: 生成策略
strategy = await self.meta_agent.generate_strategy(
task_description=f"分析 {ticker} 的{analysis_type}",
domain="investment",
context={
"ticker": ticker,
"timeframe": timeframe,
"analysis_type": analysis_type
}
)
# Step 2: 獲取資料
stock_data = await self._fetch_stock_data(ticker, timeframe)
# Step 3: 獲取領域知識
knowledge = await self._fetch_domain_knowledge(ticker, analysis_type)
# Step 4: 生成分析
analysis = await self._generate_analysis(
ticker, stock_data, knowledge, analysis_type
)
# Step 5: VRR 驗證與改進
vrr_result = await self.vrr_engine.execute(
initial_result=analysis,
strategy=strategy,
task_context={
"ticker": ticker,
"data": stock_data
}
)
# Step 6: 結構化輸出
return self._parse_analysis_result(vrr_result)
async def _fetch_stock_data(
self,
ticker: str,
timeframe: str
) -> StockData:
"""從資料庫獲取股票資料"""
# 詳細實現在 DATABASE_SCHEMA.md
pass
async def _fetch_domain_knowledge(
self,
ticker: str,
analysis_type: str
) -> List[str]:
"""從 RAG 獲取相關投資知識"""
queries = [
f"{ticker} 的行業特性",
f"{analysis_type} 分析方法",
"技術指標解讀原則"
]
knowledge_chunks = []
for query in queries:
results = await self.rag.query(query, k=3)
knowledge_chunks.extend(results.sources)
return knowledge_chunks
async def _generate_analysis(
self,
ticker: str,
stock_data: StockData,
knowledge: List[str],
analysis_type: str
) -> str:
"""使用 LLM 生成分析"""
# 建構 prompt
prompt = self._build_analysis_prompt(
ticker, stock_data, knowledge, analysis_type
)
# 使用 Gemini Pro 進行深度分析
analysis = await self.llm.generate(
prompt=prompt,
system_prompt=STOCK_ANALYST_SYSTEM_PROMPT,
temperature=0.3, # 較低溫度確保一致性
max_tokens=3000
)
return analysis
def _build_analysis_prompt(
self,
ticker: str,
stock_data: StockData,
knowledge: List[str],
analysis_type: str
) -> str:
"""建構分析 prompt"""
# 使用 PromptManager
pass
Prompt 模板 (prompts/stock_analysis.txt):
你是一位專業的股票分析師，專精於技術分析。
股票資訊:
- 代碼: {{ ticker }}
- 名稱: {{ stock_name }}
- 產業: {{ industry }}
- 當前價格: {{ current_price }}
歷史資料 ({{ timeframe }}):
{{ ohlc_summary }}
技術指標:
{% for indicator, value in indicators.items() %}
- {{ indicator }}: {{ value }}
{% if indicator in ['RSI', 'KD'] %}
(超買: >70, 超賣: <30)
{% elif indicator == 'MACD' %}
(交叉訊號: {{ macd_signal }})
{% endif %}
{% endfor %}
相關知識:
{{ domain_knowledge }}
請提供以下分析:
1. **趨勢判斷** (bullish/bearish/neutral)
- 短期趨勢 (1個月)
- 中期趨勢 (3個月)
- 理由說明
2. **關鍵價位**
- 支撐位:
- 壓力位:
- 依據:
3. **技術指標解讀**
- 各指標當前含義
- 指標之間的互相印證
- 注意事項
4. **風險提示**
- 技術面風險
- 可能的轉折訊號
5. **建議**
- 觀察要點
- 進出場參考 (僅供參考，非投資建議)
**重要提醒**:
- 基於事實數據分析
- 避免過度樂觀或悲觀
- 明確指出不確定性
- 強調「僅供參考，非投資建議」
1.2 TechnicalCalculator
檔案: backend/agents/calculator.py
職責: 計算技術指標（用於實時計算或驗證預算指標）
類別定義:
class TechnicalCalculator(BaseAgent):
"""技術指標計算 Agent"""
async def calculate_indicators(
self,
ticker: str,
ohlc_data: pd.DataFrame,
indicators: List[str]
) -> Dict[str, Any]:
"""
計算技術指標
Args:
ticker: 股票代碼
ohlc_data: OHLC DataFrame
indicators: 要計算的指標列表
["MA", "RSI", "MACD", "KD", "BB", etc.]
Returns:
指標字典 {indicator_name: values}
"""
results = {}
for indicator in indicators:
if indicator == "MA":
results["MA"] = self._calculate_ma(ohlc_data)
elif indicator == "RSI":
results["RSI"] = self._calculate_rsi(ohlc_data)
elif indicator == "MACD":
results["MACD"] = self._calculate_macd(ohlc_data)
# ... 其他指標
return results
def _calculate_ma(
self,
df: pd.DataFrame,
periods: List[int] = [5, 10, 20, 50, 100, 200]
) -> Dict[int, float]:
"""計算移動平均線"""
pass
def _calculate_rsi(
self,
df: pd.DataFrame,
period: int = 14
) -> float:
"""計算 RSI"""
pass
def _calculate_macd(
self,
df: pd.DataFrame,
fast: int = 12,
slow: int = 26,
signal: int = 9
) -> Dict[str, float]:
"""計算 MACD"""
pass
# 其他指標方法...
技術指標清單 (Option B - 完整指標):
INDICATORS_CONFIG = {
"MA": {
"periods": [5, 10, 20, 50, 100, 200],
"library": "pandas"
},
"EMA": {
"periods": [5, 10, 20, 50, 100, 200],
"library": "pandas"
},
"RSI": {
"period": 14,
"library": "pandas_ta"
},
"MACD": {
"fast": 12,
"slow": 26,
"signal": 9,
"library": "pandas_ta"
},
"KD": {
"k_period": 9,
"d_period": 3,
"smooth": 3,
"library": "pandas_ta"
},
"BB": { # 布林通道
"period": 20,
"std": 2,
"library": "pandas_ta"
},
"ATR": { # 真實波動幅度
"period": 14,
"library": "pandas_ta"
},
"ADX": { # 趨勢強度
"period": 14,
"library": "pandas_ta"
},
"OBV": { # 能量潮
"library": "pandas_ta"
},
"William_R": {
"period": 14,
"library": "pandas_ta"
},
"Volume_MA": {
"periods": [5, 20],
"library": "pandas"
},
"VWAP": { # 成交量加權均價
"library": "pandas_ta"
},
"Ichimoku": { # 一目均衡表
"tenkan": 9,
"kijun": 26,
"senkou": 52,
"library": "pandas_ta"
},
"CCI": { # 商品通道指標
"period": 20,
"library": "pandas_ta"
},
"MFI": { # 資金流量指標
"period": 14,
"library": "pandas_ta"
}
}
1.3 VerifierAgent
檔案: backend/agents/verifier.py
職責: 驗證分析結果的合理性
類別定義:
class VerifierAgent(BaseAgent):
"""驗證 Agent"""
async def verify(
self,
analysis: str,
criteria: List[str],
stock_data: StockData
) -> VerificationResult:
"""
驗證分析結果
Args:
analysis: 分析結果文字
criteria: 驗證標準（由 Meta-Agent 生成）
stock_data: 原始股票資料
Returns:
VerificationResult
"""
failed_criteria = []
details = {}
for criterion in criteria:
passed, detail = await self._check_criterion(
analysis, criterion, stock_data
)
if not passed:
failed_criteria.append(criterion)
details[criterion] = detail
return VerificationResult(
passed=(len(failed_criteria) == 0),
failed_criteria=failed_criteria,
details=details
)
async def _check_criterion(
self,
analysis: str,
criterion: str,
stock_data: StockData
) -> Tuple[bool, str]:
"""檢查單一驗證標準"""
# 使用 LLM 進行語義驗證
prompt = f"""
驗證以下分析是否符合標準:
標準: {criterion}
分析內容:
{analysis}
參考資料:
{stock_data.summary()}
回答格式:
{{
"passed": true/false,
"reason": "原因說明"
}}
"""
result = await self.llm.generate_structured(
prompt=prompt,
schema={
"type": "object",
"properties": {
"passed": {"type": "boolean"},
"reason": {"type": "string"}
}
}
)
return result["passed"], result["reason"]
2. LangGraph 工作流程
2.1 StockAnalysisWorkflow
檔案: backend/agents/coordinator.py
用途: 使用 LangGraph 編排完整分析流程
狀態定義:
from typing import TypedDict
from langgraph.graph import StateGraph, END
class AnalysisState(TypedDict):
"""分析工作流程狀態"""
# 輸入
ticker: str
analysis_type: str
timeframe: str
# 中間狀態
strategy: Optional[VRRStrategy]
stock_data: Optional[StockData]
knowledge: Optional[List[str]]
initial_analysis: Optional[str]
# VRR 狀態
verification_result: Optional[VerificationResult]
reflection_result: Optional[ReflectionResult]
iteration: int
# 輸出
final_analysis: Optional[AnalysisResult]
error: Optional[str]
工作流程圖:
class StockAnalysisWorkflow:
"""股票分析 LangGraph 工作流程"""
def __init__(
self,
meta_agent: MetaAgent,
analyzer: StockAnalyzer,
calculator: TechnicalCalculator,
verifier: VerifierAgent,
db_session: AsyncSession,
rag_service: RAGService
):
self.meta_agent = meta_agent
self.analyzer = analyzer
self.calculator = calculator
self.verifier = verifier
self.db = db_session
self.rag = rag_service
self.graph = self._build_graph()
def _build_graph(self) -> StateGraph:
"""建立工作流程圖"""
workflow = StateGraph(AnalysisState)
# 節點
workflow.add_node("generate_strategy", self._generate_strategy)
workflow.add_node("fetch_data", self._fetch_data)
workflow.add_node("fetch_knowledge", self._fetch_knowledge)
workflow.add_node("calculate_indicators", self._calculate_indicators)
workflow.add_node("generate_analysis", self._generate_analysis)
workflow.add_node("verify", self._verify)
workflow.add_node("reflect", self._reflect)
workflow.add_node("refine", self._refine)
# 邊
workflow.set_entry_point("generate_strategy")
workflow.add_edge("generate_strategy", "fetch_data")
workflow.add_edge("fetch_data", "fetch_knowledge")
workflow.add_edge("fetch_knowledge", "calculate_indicators")
workflow.add_edge("calculate_indicators", "generate_analysis")
workflow.add_edge("generate_analysis", "verify")
# 條件邊: 驗證通過 -> END, 失敗 -> 反思
workflow.add_conditional_edges(
"verify",
self._should_refine,
{
"continue": "reflect",
"end": END
}
)
workflow.add_edge("reflect", "refine")
workflow.add_edge("refine", "verify")
return workflow.compile()
async def _generate_strategy(self, state: AnalysisState) -> AnalysisState:
"""節點: 生成策略"""
strategy = await self.meta_agent.generate_strategy(
task_description=f"分析 {state['ticker']}",
domain="investment",
context={
"ticker": state["ticker"],
"analysis_type": state["analysis_type"]
}
)
state["strategy"] = strategy
return state
async def _fetch_data(self, state: AnalysisState) -> AnalysisState:
"""節點: 獲取股票資料"""
# 從資料庫讀取
stock_data = await self._query_stock_data(
state["ticker"],
state["timeframe"]
)
state["stock_data"] = stock_data
return state
async def _fetch_knowledge(self, state: AnalysisState) -> AnalysisState:
"""節點: 獲取領域知識"""
knowledge = await self.rag.query(
question=f"{state['ticker']} 技術分析知識",
k=5
)
state["knowledge"] = knowledge.sources
return state
async def _calculate_indicators(self, state: AnalysisState) -> AnalysisState:
"""節點: 計算技術指標"""
# 如果資料庫已有預算指標，直接讀取
# 否則實時計算
indicators = await self.calculator.calculate_indicators(
ticker=state["ticker"],
ohlc_data=state["stock_data"].ohlc,
indicators=list(INDICATORS_CONFIG.keys())
)
state["stock_data"].indicators = indicators
return state
async def _generate_analysis(self, state: AnalysisState) -> AnalysisState:
"""節點: 生成分析"""
analysis = await self.analyzer._generate_analysis(
ticker=state["ticker"],
stock_data=state["stock_data"],
knowledge=state["knowledge"],
analysis_type=state["analysis_type"]
)
state["initial_analysis"] = analysis
state["iteration"] = 0
return state
async def _verify(self, state: AnalysisState) -> AnalysisState:
"""節點: 驗證分析"""
# 使用當前分析（initial 或 refined）
current_analysis = (
state.get("refined_analysis") or state["initial_analysis"]
)
verification = await self.verifier.verify(
analysis=current_analysis,
criteria=state["strategy"].verify_criteria,
stock_data=state["stock_data"]
)
state["verification_result"] = verification
return state
async def _reflect(self, state: AnalysisState) -> AnalysisState:
"""節點: 反思問題"""
# 使用 LLM 分析失敗原因
reflection_prompt = f"""
分析結果驗證失敗，請反思問題:
失敗標準:
{state["verification_result"].failed_criteria}
原始分析:
{state.get("refined_analysis") or state["initial_analysis"]}
反思重點:
{state["strategy"].reflect_focus}
請指出:
1. 問題所在
2. 根本原因
3. 改進建議
"""
reflection_text = await self.analyzer.llm.generate(reflection_prompt)
state["reflection_result"] = ReflectionResult(
identified_issues=[], # 從 reflection_text 解析
root_causes=[],
improvement_suggestions=[]
)
return state
async def _refine(self, state: AnalysisState) -> AnalysisState:
"""節點: 改進分析"""
refine_prompt = f"""
基於反思結果改進分析:
原始分析:
{state.get("refined_analysis") or state["initial_analysis"]}
反思結果:
{state["reflection_result"]}
改進優先順序:
{state["strategy"].refine_priority}
請提供改進後的分析。
"""
refined = await self.analyzer.llm.generate(refine_prompt)
state["refined_analysis"] = refined
state["iteration"] += 1
return state
def _should_refine(self, state: AnalysisState) -> str:
"""決策函數: 是否需要改進"""
if state["verification_result"].passed:
# 驗證通過，結束
state["final_analysis"] = self._build_final_result(state)
return "end"
if state["iteration"] >= 3:
# 達到最大迭代次數，強制結束
state["final_analysis"] = self._build_final_result(state)
state["error"] = "達到最大迭代次數，分析可能不完整"
return "end"
# 繼續改進
return "continue"
def _build_final_result(self, state: AnalysisState) -> AnalysisResult:
"""建構最終結果"""
final_text = state.get("refined_analysis") or state["initial_analysis"]
# 解析 LLM 輸出為結構化結果
return AnalysisResult(
ticker=state["ticker"],
summary=final_text,
# ... 解析其他欄位
verification_passed=state["verification_result"].passed,
iterations=state["iteration"]
)
async def run(
self,
ticker: str,
analysis_type: str = "comprehensive",
timeframe: str = "3_months"
) -> AnalysisResult:
"""執行完整工作流程"""
initial_state: AnalysisState = {
"ticker": ticker,
"analysis_type": analysis_type,
"timeframe": timeframe,
"iteration": 0
}
final_state = await self.graph.ainvoke(initial_state)
return final_state["final_analysis"]
3. 資料模型
3.1 StockData
檔案: backend/agents/models.py
from pydantic import BaseModel
from datetime import datetime
import pandas as pd
class StockData(BaseModel):
"""股票資料模型"""
ticker: str
name: str
industry: Optional[str] = None
# OHLC 資料
ohlc: pd.DataFrame # columns: [date, open, high, low, close, volume]
# 技術指標
indicators: Dict[str, Any] = {}
# 統計資訊
current_price: float
price_change_pct: float
volume_avg: float
# 時間範圍
start_date: datetime
end_date: datetime
class Config:
arbitrary_types_allowed = True # 允許 pd.DataFrame
def summary(self) -> str:
"""生成資料摘要"""
return f"""
股票: {self.ticker} ({self.name})
當前價格: {self.current_price}
漲跌幅: {self.price_change_pct:.2f}%
資料範圍: {self.start_date} ~ {self.end_date}
指標數量: {len(self.indicators)}
"""
class AnalysisResult(BaseModel):
"""分析結果模型"""
ticker: str
timestamp: datetime = Field(default_factory=datetime.now)
# 分析內容
summary: str
trend: Literal["bullish", "bearish", "neutral"]
trend_short: Optional[str] = None # 短期趨勢
trend_medium: Optional[str] = None # 中期趨勢
# 關鍵價位
support_levels: List[float] = []
resistance_levels: List[float] = []
# 技術指標解讀
indicators_interpretation: Dict[str, str] = {}
# 風險提示
risks: List[str] = []
# 建議
recommendations: List[str] = []
# 元資料
confidence: float # 0-1
verification_passed: bool
iterations: int
# 圖表資料
chart_data: Optional[Dict[str, Any]] = None
4. 投資領域知識
4.1 InvestmentKnowledge
檔案: backend/knowledge/stock_knowledge.py
用途: 管理投資分析相關的領域知識
類別定義:
from core.knowledge import DomainKnowledge
class InvestmentKnowledge(DomainKnowledge):
"""投資領域知識管理"""
async def initialize(self) -> None:
"""初始化知識庫"""
# 載入知識文件
knowledge_docs = [
self._load_technical_analysis_basics(),
self._load_indicator_interpretations(),
self._load_market_concepts(),
self._load_risk_management()
]
# 加入向量資料庫
await self.rag_service.add_knowledge(
documents=knowledge_docs,
metadata={"domain": "investment"}
)
def _load_technical_analysis_basics(self) -> List[Document]:
"""載入技術分析基礎知識"""
# 從 markdown 檔案讀取
# docs/knowledge/technical_analysis.md
pass
def _load_indicator_interpretations(self) -> List[Document]:
"""載入指標解讀知識"""
# docs/knowledge/indicators/*.md
# - rsi.md
# - macd.md
# - ma.md
# - etc.
pass
def get_concepts(self) -> List[str]:
"""獲取核心概念"""
return [
"趨勢線",
"支撐壓力",
"技術指標",
"K線型態",
"成交量分析",
"風險管理"
]
知識文件範例 (docs/knowledge/indicators/rsi.md):
# RSI (Relative Strength Index) 相對強弱指標
## 定義
RSI 是衡量價格變動速度和幅度的動量指標，範圍 0-100。
## 計算方法
RSI = 100 - (100 / (1 + RS))
其中 RS = 平均上漲幅度 / 平均下跌幅度（通常 14 日）
## 解讀
- **RSI > 70**: 超買區，可能面臨回調
- **RSI < 30**: 超賣區，可能出現反彈
- **RSI 50**: 中性區，多空均勢
## 注意事項
- 強勢行情中，RSI 可能長期停留在超買區
- 應結合其他指標綜合判斷
- 背離訊號（價格創新高但 RSI 未創新高）值得關注
## 常見錯誤
- 僅依賴 RSI 單一指標做決策
- 忽略市場大趨勢
- 在震盪市中過度交易
5. 配置檔案
5.1 markets.yaml
檔案: backend/config/markets.yaml
markets:
taiwan:
enabled: true
source: yfinance
market_type: full # full: 全市場, custom: 自訂清單
suffix: .TW
# 全市場模式：自動爬取台股清單
auto_fetch_tickers: true
fetch_url: "https://isin.twse.com.tw/isin/C_public.jsp?strMode=2"
# 或使用靜態清單
tickers_file: null
# 市場資訊
timezone: Asia/Taipei
trading_hours:
start: "09:00"
end: "13:30"
us:
enabled: true
source: yfinance
market_type: custom
# 七巨頭清單
tickers:
- AAPL # Apple
- MSFT # Microsoft
- GOOGL # Alphabet
- AMZN # Amazon
- NVDA # NVIDIA
- META # Meta
- TSLA # Tesla
timezone: America/New_York
trading_hours:
start: "09:30"
end: "16:00"
japan:
enabled: false # Phase 2
data_settings:
start_date: "2020-01-01"
end_date: null # null = 至今
# 更新排程
update_schedule:
cron: "0 6 * * *" # 每天早上 6:00 (台北時間)
timezone: Asia/Taipei
# 重試設定
retry:
max_attempts: 3
backoff_factor: 2
timeout_seconds: 30
5.2 indicators.yaml
檔案: backend/config/indicators.yaml
indicators:
MA:
enabled: true
periods: [5, 10, 20, 50, 100, 200]
library: pandas
EMA:
enabled: true
periods: [5, 10, 20, 50, 100, 200]
library: pandas
RSI:
enabled: true
period: 14
library: pandas_ta
MACD:
enabled: true
fast: 12
slow: 26
signal: 9
library: pandas_ta
KD:
enabled: true
k_period: 9
d_period: 3
smooth: 3
library: pandas_ta
BB: # Bollinger Bands
enabled: true
period: 20
std: 2
library: pandas_ta
ATR:
enabled: true
period: 14
library: pandas_ta
ADX:
enabled: true
period: 14
library: pandas_ta
OBV:
enabled: true
library: pandas_ta
William_R:
enabled: true
period: 14
library: pandas_ta
Volume_MA:
enabled: true
periods: [5, 20]
library: pandas
VWAP:
enabled: true
library: pandas_ta
Ichimoku:
enabled: true
tenkan: 9
kijun: 26
senkou: 52
library: pandas_ta
CCI:
enabled: true
period: 20
library: pandas_ta
MFI:
enabled: true
period: 14
library: pandas_ta
# 預算設定
precompute:
enabled: true # 是否預先計算並存入資料庫
batch_size: 100 # 批次處理大小
6. 系統 Prompts
檔案: backend/prompts/system_prompts.py
STOCK_ANALYST_SYSTEM_PROMPT = """
你是一位專業的股票技術分析師，具有以下特質:
1. **客觀中立**: 基於數據和技術指標進行分析，避免情緒化判斷
2. **風險意識**: 總是提醒投資風險，明確指出不確定性
3. **專業術語**: 使用正確的技術分析術語
4. **結構化表達**: 條理清晰，邏輯嚴謹
5. **教育導向**: 解釋技術指標的含義，幫助用戶理解
**重要原則**:
- 所有分析「僅供參考，非投資建議」
- 不做價格預測或買賣推薦
- 承認分析的局限性
- 鼓勵用戶多方面研究
**語言**: 繁體中文（台灣用戶）或英文（美股用戶）
"""
META_AGENT_SYSTEM_PROMPT = """
你是一個 Meta-Agent，負責為股票分析任務生成 Verify-Reflect-Refine 策略。
你的目標是:
1. 分析任務需求
2. 識別關鍵驗證點
3. 預見可能的失敗模式
4. 設計有效的驗證標準
輸出 JSON 格式的策略，包含:
- verify_criteria: 具體可檢查的驗證點
- reflect_focus: 失敗時應反思的面向
- refine_priority: 改進的優先順序
"""
文檔版本: 1.0
最後更新: 2025-02-01
（繼續閱讀 DATA_PIPELINE_SPEC.md 了解資料管道實現）