162 lines
5.1 KiB
Python
Executable File
162 lines
5.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
美股分析脚本（修复版）
|
||
|
||
用法:
|
||
python3 scripts/test_stock.py AAPL
|
||
python3 scripts/test_stock.py AAPL TSLA NVDA
|
||
"""
|
||
import sys
|
||
import os
|
||
|
||
# 确保路径正确
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
project_root = os.path.dirname(script_dir)
|
||
backend_dir = os.path.join(project_root, 'backend')
|
||
sys.path.insert(0, backend_dir)
|
||
|
||
import asyncio
|
||
from app.services.yfinance_service import get_yfinance_service
|
||
from app.services.feishu_service import get_feishu_service
|
||
from app.services.telegram_service import get_telegram_service
|
||
from app.crypto_agent.llm_signal_analyzer import LLMSignalAnalyzer
|
||
from app.utils.logger import logger
|
||
|
||
|
||
async def analyze(symbol: str, send_notification: bool = True):
|
||
"""分析单只股票
|
||
|
||
Args:
|
||
symbol: 股票代码
|
||
send_notification: 是否发送通知（默认True）
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"📊 分析 {symbol}")
|
||
print(f"{'='*60}")
|
||
|
||
try:
|
||
# 获取服务
|
||
yf_service = get_yfinance_service()
|
||
llm = LLMSignalAnalyzer()
|
||
feishu = get_feishu_service()
|
||
telegram = get_telegram_service()
|
||
|
||
# 获取行情
|
||
print(f"获取行情...")
|
||
ticker = yf_service.get_ticker(symbol)
|
||
if not ticker:
|
||
print(f"❌ 无法获取 {symbol} 行情")
|
||
return
|
||
|
||
price = ticker['lastPrice']
|
||
change = ticker['priceChangePercent']
|
||
print(f"价格: ${price:,.2f} ({change:+.2f}%)")
|
||
print(f"成交量: {ticker['volume']:,}")
|
||
|
||
# 获取K线
|
||
print(f"获取K线数据...")
|
||
data = yf_service.get_multi_timeframe_data(symbol)
|
||
|
||
if not data:
|
||
print(f"❌ 无法获取K线数据")
|
||
return
|
||
|
||
print(f"时间周期: {', '.join(data.keys())}")
|
||
|
||
# LLM分析
|
||
print(f"\n🤖 LLM分析中...\n")
|
||
result = await llm.analyze(symbol, data, symbols=[symbol], position_info=None)
|
||
|
||
# 输出结果
|
||
summary = result.get('analysis_summary', '')
|
||
signals = result.get('signals', [])
|
||
|
||
print(f"市场状态: {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:
|
||
# 限制理由长度
|
||
short_reason = reason[:80] + "..." if len(reason) > 80 else reason
|
||
print(f" 理由: {short_reason}")
|
||
|
||
# 发送通知（仅发送置信度 >= 60% 的信号）
|
||
if send_notification:
|
||
best_signal = None
|
||
for sig in signals:
|
||
if sig.get('confidence', 0) >= 60 and sig.get('grade', 'D') != 'D':
|
||
best_signal = sig
|
||
break
|
||
|
||
if best_signal:
|
||
# 使用正确的方法格式化通知
|
||
card = llm.format_feishu_card(best_signal, symbol)
|
||
title = card['title']
|
||
content = card['content']
|
||
|
||
# 发送通知
|
||
await feishu.send_markdown(title, content)
|
||
await telegram.send_message(llm.format_signal_message(best_signal, symbol))
|
||
print(f"\n📬 通知已发送：{title}")
|
||
else:
|
||
print(f"\n⏸️ 置信度不足，不发送通知")
|
||
else:
|
||
print(f"\n⏸️ 无交易信号")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
async def main():
|
||
# 从命令行参数获取股票代码
|
||
if len(sys.argv) < 2:
|
||
print("用法: python3 scripts/test_stock.py AAPL [TSLA] [NVDA] ...")
|
||
print("\n示例:")
|
||
print(" python3 scripts/test_stock.py AAPL")
|
||
print(" python3 scripts/test_stock.py AAPL TSLA NVDA")
|
||
sys.exit(1)
|
||
|
||
symbols = sys.argv[1:]
|
||
|
||
print("="*60)
|
||
print("🤖 美股分析脚本")
|
||
print("="*60)
|
||
print(f"股票: {', '.join(symbols)}")
|
||
print(f"时间: {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
for symbol in symbols:
|
||
await analyze(symbol.upper())
|
||
|
||
print("\n" + "="*60)
|
||
print("✅ 分析完成")
|
||
print("="*60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
asyncio.run(main())
|
||
except KeyboardInterrupt:
|
||
print("\n\n⚠️ 用户中断")
|