194 lines
6.2 KiB
Python
194 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试不同 Ollama 模型的量化分析能力
|
|
评估标准：
|
|
1. 响应速度
|
|
2. JSON 格式准确性
|
|
3. 技术分析质量
|
|
4. 推理能力
|
|
"""
|
|
|
|
import httpx
|
|
import json
|
|
import time
|
|
import asyncio
|
|
|
|
# 测试用的技术指标数据
|
|
TEST_INDICATORS = {
|
|
"ma5": 270.5,
|
|
"ma20": 265.0,
|
|
"ma50": 255.0,
|
|
"rsi": 65.5,
|
|
"macd": 2.3,
|
|
"macd_signal": 1.8,
|
|
"macd_hist": 0.5,
|
|
"current_price": 272.0,
|
|
"volume": 50000000
|
|
}
|
|
|
|
# 量化分析 Prompt
|
|
QUANT_PROMPT = f"""
|
|
请根据以下技术指标分析 AAPL 的交易信号。
|
|
|
|
技术指标数据：
|
|
{json.dumps(TEST_INDICATORS, ensure_ascii=False, indent=2)}
|
|
|
|
请执行以下分析：
|
|
1. MA (移动平均线) 分析：当前价格与 MA5/MA20/MA50 的位置关系，趋势方向
|
|
2. RSI (相对强弱指标) 分析：超买/超卖信号，动量变化
|
|
3. MACD (指数平滑异同移动平均线) 分析：快线/慢线交叉，直方图趋势
|
|
4. 综合判断：BUY/SELL/HOLD 信号
|
|
5. 置信度评估：0.0-1.0 之间的置信度分数
|
|
6. 风险等级：基于指标波动性和不确定性的 LOW/MEDIUM/HIGH
|
|
|
|
请输出 JSON 格式（不需要额外说明，直接返回 JSON）：
|
|
{{
|
|
"signal": "BUY|SELL|HOLD",
|
|
"confidence": <0-1 的数字>,
|
|
"reason": "综合分析理由，100-150 字以内",
|
|
"ma_signal": "MA 分析结论",
|
|
"rsi_signal": "RSI 分析结论",
|
|
"macd_signal": "MACD 分析结论",
|
|
"risk_level": "LOW|MEDIUM|HIGH",
|
|
"entry_price": <建议入场价或 null>,
|
|
"stop_loss": <止损价或 null>,
|
|
"take_profit": <止盈价或 null>
|
|
}}
|
|
"""
|
|
|
|
# 候选模型列表
|
|
CANDIDATE_MODELS = [
|
|
"deepseek-r1:70b", # 70B DeepSeek 推理模型
|
|
"qwen2.5-coder:32b-instruct", # 32B Qwen 代码模型
|
|
"deepseek-coder:33b", # 33B DeepSeek 代码模型
|
|
"qwen3-coder:30b-a3b-q8_0", # 30B Qwen 高精度模型
|
|
"llama3.3:latest", # 70B Llama 3.3
|
|
]
|
|
|
|
async def test_model(model_name: str, timeout: int = 120) -> dict:
|
|
"""测试单个模型"""
|
|
print(f"\n{'='*60}")
|
|
print(f"Testing: {model_name}")
|
|
print(f"{'='*60}")
|
|
|
|
start_time = time.time()
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
response = await client.post(
|
|
"http://host.docker.internal:11434/api/generate",
|
|
json={
|
|
"model": model_name,
|
|
"prompt": QUANT_PROMPT,
|
|
"stream": False,
|
|
"temperature": 0.3,
|
|
"top_p": 0.9
|
|
}
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
return {
|
|
"model": model_name,
|
|
"success": False,
|
|
"error": f"HTTP {response.status_code}",
|
|
"time": 0
|
|
}
|
|
|
|
elapsed = time.time() - start_time
|
|
result = response.json()
|
|
response_text = result.get("response", "")
|
|
|
|
# 尝试解析 JSON
|
|
json_data = None
|
|
try:
|
|
json_data = json.loads(response_text)
|
|
except:
|
|
# 尝试从 markdown 提取
|
|
if "```json" in response_text:
|
|
json_str = response_text.split("```json")[1].split("```")[0]
|
|
json_data = json.loads(json_str.strip())
|
|
elif "```" in response_text:
|
|
json_str = response_text.split("```")[1].split("```")[0]
|
|
json_data = json.loads(json_str.strip())
|
|
|
|
# 评分
|
|
score = 0
|
|
has_signal = json_data and "signal" in json_data
|
|
has_confidence = json_data and "confidence" in json_data
|
|
has_reason = json_data and "reason" in json_data and len(json_data.get("reason", "")) > 50
|
|
|
|
if has_signal: score += 25
|
|
if has_confidence: score += 25
|
|
if has_reason: score += 25
|
|
if elapsed < 30: score += 25
|
|
elif elapsed < 60: score += 15
|
|
elif elapsed < 90: score += 5
|
|
|
|
print(f"✅ Success!")
|
|
print(f"⏱️ Time: {elapsed:.2f}s")
|
|
print(f"📊 Score: {score}/100")
|
|
if json_data:
|
|
print(f"Signal: {json_data.get('signal', 'N/A')}")
|
|
print(f"Confidence: {json_data.get('confidence', 'N/A')}")
|
|
print(f"Reason: {json_data.get('reason', 'N/A')[:100]}...")
|
|
|
|
return {
|
|
"model": model_name,
|
|
"success": True,
|
|
"time": elapsed,
|
|
"score": score,
|
|
"has_valid_json": json_data is not None,
|
|
"response_preview": response_text[:200]
|
|
}
|
|
|
|
except Exception as e:
|
|
elapsed = time.time() - start_time
|
|
print(f"❌ Failed: {str(e)}")
|
|
return {
|
|
"model": model_name,
|
|
"success": False,
|
|
"error": str(e),
|
|
"time": elapsed,
|
|
"score": 0
|
|
}
|
|
|
|
async def main():
|
|
"""主测试函数"""
|
|
print("🚀 Starting Ollama Model Quantitative Analysis Test")
|
|
print(f"Testing {len(CANDIDATE_MODELS)} models...")
|
|
|
|
results = []
|
|
for model in CANDIDATE_MODELS:
|
|
result = await test_model(model)
|
|
results.append(result)
|
|
await asyncio.sleep(2) # 避免过载
|
|
|
|
# 排序并显示结果
|
|
print("\n" + "="*60)
|
|
print("📊 FINAL RESULTS (Ranked by Score)")
|
|
print("="*60)
|
|
|
|
successful = [r for r in results if r["success"]]
|
|
successful.sort(key=lambda x: x["score"], reverse=True)
|
|
|
|
for i, result in enumerate(successful, 1):
|
|
print(f"\n{i}. {result['model']}")
|
|
print(f" Score: {result['score']}/100")
|
|
print(f" Time: {result['time']:.2f}s")
|
|
print(f" Valid JSON: {'✅' if result['has_valid_json'] else '❌'}")
|
|
|
|
if successful:
|
|
best = successful[0]
|
|
print(f"\n🏆 RECOMMENDED MODEL: {best['model']}")
|
|
print(f" Score: {best['score']}/100, Time: {best['time']:.2f}s")
|
|
|
|
# 失败的模型
|
|
failed = [r for r in results if not r["success"]]
|
|
if failed:
|
|
print("\n❌ Failed Models:")
|
|
for result in failed:
|
|
print(f" - {result['model']}: {result['error']}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|