101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
美股快速分析脚本（简化版）
|
|
|
|
用法:
|
|
cd backend && python3 ../scripts/analyze_stock_simple.py AAPL
|
|
cd backend && python3 ../scripts/analyze_stock_simple.py AAPL TSLA NVDA
|
|
"""
|
|
import sys
|
|
import os
|
|
import asyncio
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'backend'))
|
|
|
|
from app.services.yfinance_service import get_yfinance_service
|
|
from app.crypto_agent.llm_signal_analyzer import LLMSignalAnalyzer
|
|
from app.utils.logger import logger
|
|
|
|
|
|
async def analyze(symbol: str):
|
|
"""分析单只股票"""
|
|
print(f"\n{'='*60}")
|
|
print(f"📊 分析 {symbol}")
|
|
print(f"{'='*60}")
|
|
|
|
try:
|
|
# 获取服务和数据
|
|
yf_service = get_yfinance_service()
|
|
llm = LLMSignalAnalyzer()
|
|
|
|
# 获取行情
|
|
ticker = yf_service.get_ticker(symbol)
|
|
if not ticker:
|
|
print(f"❌ 无法获取行情")
|
|
return
|
|
|
|
price = ticker['lastPrice']
|
|
change = ticker['priceChangePercent']
|
|
print(f"价格: ${price:,.2f} ({change:+.2f}%)")
|
|
|
|
# 获取K线数据
|
|
data = yf_service.get_multi_timeframe_data(symbol)
|
|
print(f"K线: {list(data.keys())}")
|
|
|
|
# LLM分析
|
|
print(f"\n🤖 LLM分析中...")
|
|
result = await llm.analyze(symbol, data, symbols=[symbol], position_info=None)
|
|
|
|
# 输出结果
|
|
summary = result.get('analysis_summary', '')
|
|
signals = result.get('signals', [])
|
|
|
|
print(f"\n状态: {summary}")
|
|
|
|
if signals:
|
|
print(f"\n🎯 发现 {len(signals)} 个信号:")
|
|
for sig in signals:
|
|
action = sig.get('action', 'wait')
|
|
grade = sig.get('grade', 'D')
|
|
conf = sig.get('confidence', 0)
|
|
|
|
action_text = {'buy': '🟢 做多', 'sell': '🔴 做空'}.get(action, action)
|
|
grade_icon = {'A': '⭐⭐⭐', 'B': '⭐⭐', 'C': '⭐'}.get(grade, '')
|
|
|
|
print(f"\n {action_text} [{grade}{grade_icon}] {conf}%")
|
|
|
|
entry = sig.get('entry_price')
|
|
sl = sig.get('stop_loss')
|
|
tp = sig.get('take_profit')
|
|
|
|
if entry and sl and tp:
|
|
print(f" 入场: ${entry:,.2f}")
|
|
print(f" 止损: ${sl:,.2f}")
|
|
print(f" 止盈: ${tp:,.2f}")
|
|
|
|
reason = sig.get('reason', '')
|
|
if reason:
|
|
print(f" 理由: {reason[:100]}...")
|
|
else:
|
|
print(f"\n⏸️ 无交易信号")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 错误: {e}")
|
|
|
|
|
|
async def main():
|
|
symbols = sys.argv[1:] if len(sys.argv) > 1 else ['AAPL']
|
|
|
|
print(f"🤖 美股快速分析")
|
|
print(f"股票: {', '.join(symbols)}")
|
|
|
|
for symbol in symbols:
|
|
await analyze(symbol.upper())
|
|
|
|
print(f"\n✅ 完成")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|