549 lines
20 KiB
Python
Executable File
549 lines
20 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.config import get_settings
|
||
from app.utils.logger import logger
|
||
|
||
|
||
async def analyze(symbol: str, send_notification: bool = True):
|
||
"""分析单只股票
|
||
|
||
Args:
|
||
symbol: 股票代码
|
||
send_notification: 是否发送通知（默认True）
|
||
|
||
Returns:
|
||
分析结果字典
|
||
"""
|
||
# 导入股票名称映射
|
||
from app.stock_agent.stock_agent import STOCK_NAMES
|
||
|
||
result = {
|
||
'symbol': symbol,
|
||
'price': 0,
|
||
'signals': [],
|
||
'notified': False
|
||
}
|
||
|
||
# 获取配置
|
||
settings = get_settings()
|
||
threshold = settings.stock_llm_threshold * 100 # 转换为百分比
|
||
|
||
# 获取股票中文名称
|
||
stock_name = STOCK_NAMES.get(symbol, '')
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"📊 分析 {symbol_display}")
|
||
print(f"{'='*60}")
|
||
|
||
try:
|
||
# 获取服务
|
||
yf_service = get_yfinance_service()
|
||
llm = LLMSignalAnalyzer(agent_type="stock") # 指定使用 stock 模型配置
|
||
feishu = get_feishu_service()
|
||
telegram = get_telegram_service()
|
||
|
||
# 获取行情
|
||
print(f"获取行情...")
|
||
ticker = yf_service.get_ticker(symbol)
|
||
if not ticker:
|
||
print(f"❌ 无法获取 {symbol} 行情")
|
||
return result
|
||
|
||
price = ticker['lastPrice']
|
||
change = ticker['priceChangePercent']
|
||
result['price'] = price
|
||
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 result
|
||
|
||
print(f"时间周期: {', '.join(data.keys())}")
|
||
|
||
# LLM分析
|
||
print(f"\n🤖 LLM分析中...\n")
|
||
analysis = await llm.analyze(symbol, data, symbols=[symbol], position_info=None)
|
||
|
||
# 输出结果
|
||
summary = analysis.get('analysis_summary', '')
|
||
signals = analysis.get('signals', [])
|
||
result['signals'] = 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}")
|
||
|
||
# 发送通知（仅发送置信度 >= 阈值的信号）
|
||
if send_notification:
|
||
best_signal = None
|
||
for sig in signals:
|
||
if sig.get('confidence', 0) >= threshold 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']
|
||
|
||
# 发送通知 - 使用 send_card 方法
|
||
# 根据信号方向选择颜色
|
||
color = "green" if best_signal.get('action') == 'buy' else "red"
|
||
await feishu.send_card(title, content, color)
|
||
await telegram.send_message(llm.format_signal_message(best_signal, symbol))
|
||
print(f"\n📬 通知已发送：{title}")
|
||
result['notified'] = True
|
||
else:
|
||
print(f"\n⏸️ 置信度不足，不发送通知（阈值: {threshold}%）")
|
||
else:
|
||
print(f"\n⏸️ 无交易信号")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f"❌ 错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return result
|
||
|
||
|
||
def print_summary_report(results: list, send_notification: bool = True):
|
||
"""打印汇总报告并发送通知
|
||
|
||
Args:
|
||
results: 分析结果列表
|
||
send_notification: 是否发送通知（默认True）
|
||
"""
|
||
from app.config import get_settings
|
||
from app.stock_agent.stock_agent import STOCK_NAMES
|
||
settings = get_settings()
|
||
threshold = settings.stock_llm_threshold * 100 # 获取阈值
|
||
|
||
total = len(results)
|
||
with_signals = [r for r in results if r.get('signals')]
|
||
notified = [r for r in results if r.get('notified')]
|
||
|
||
# 区分美股和港股
|
||
us_results = [r for r in results if not r['symbol'].endswith('.HK')]
|
||
hk_results = [r for r in results if r['symbol'].endswith('.HK')]
|
||
|
||
# 统计信号
|
||
buy_count = 0
|
||
sell_count = 0
|
||
high_quality_signals = [] # A/B级信号且达到阈值
|
||
all_signals = [] # 所有信号
|
||
|
||
for r in with_signals:
|
||
for sig in r.get('signals', []):
|
||
sig['symbol'] = r['symbol']
|
||
sig['current_price'] = r.get('price', 0)
|
||
sig['is_hk'] = r['symbol'].endswith('.HK')
|
||
sig['stock_name'] = STOCK_NAMES.get(r['symbol'], '')
|
||
all_signals.append(sig)
|
||
|
||
if sig.get('action') == 'buy':
|
||
buy_count += 1
|
||
elif sig.get('action') == 'sell':
|
||
sell_count += 1
|
||
|
||
# 只统计达到阈值的A/B级信号
|
||
if sig.get('grade') in ['A', 'B'] and sig.get('confidence', 0) >= threshold:
|
||
high_quality_signals.append(sig)
|
||
|
||
# 按置信度排序
|
||
high_quality_signals.sort(key=lambda x: x.get('confidence', 0), reverse=True)
|
||
all_signals.sort(key=lambda x: x.get('confidence', 0), reverse=True)
|
||
|
||
# 打印汇总
|
||
print("\n" + "="*80)
|
||
print("📊 股票分析汇总报告")
|
||
print("="*80)
|
||
print(f"分析总数: {total} 只 (美股: {len(us_results)}, 港股: {len(hk_results)})")
|
||
print(f"有信号: {len(with_signals)} 只")
|
||
print(f"已通知: {len(notified)} 只")
|
||
print(f"通知阈值: {threshold}%")
|
||
print("")
|
||
|
||
# 显示高等级信号（达到阈值的）
|
||
if high_quality_signals:
|
||
print(f"⭐ 高等级信号达到阈值 (A/B级 >= {threshold}%): {len(high_quality_signals)} 个")
|
||
for sig in high_quality_signals[:10]:
|
||
symbol = sig['symbol']
|
||
stock_name = sig.get('stock_name', '')
|
||
market_tag = '[港股]' if sig.get('is_hk') else '[美股]'
|
||
action = '🟢 做多' if sig.get('action') == 'buy' else '🔴 做空'
|
||
grade = sig.get('grade', 'D')
|
||
confidence = sig.get('confidence', 0)
|
||
price = sig.get('current_price', 0)
|
||
entry = sig.get('entry_price', 0)
|
||
|
||
# 构建带名称的股票显示
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
|
||
print(f" {market_tag} {symbol_display} {action} [{grade}级] {confidence}% @ ${price:,.2f}")
|
||
if entry > 0:
|
||
print(f" 入场: ${entry:,.2f}")
|
||
print("")
|
||
|
||
# 显示未达到阈值但质量不错的信号
|
||
below_threshold = [s for s in all_signals
|
||
if s.get('grade') in ['A', 'B'] and s.get('confidence', 0) < threshold]
|
||
if below_threshold:
|
||
print(f"⚠️ 以下信号未达到通知阈值 ({threshold}%):")
|
||
for sig in below_threshold[:10]:
|
||
symbol = sig['symbol']
|
||
stock_name = sig.get('stock_name', '')
|
||
market_tag = '[港股]' if sig.get('is_hk') else '[美股]'
|
||
action = '🟢 做多' if sig.get('action') == 'buy' else '🔴 做空'
|
||
grade = sig.get('grade', 'D')
|
||
confidence = sig.get('confidence', 0)
|
||
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
print(f" {market_tag} {symbol_display} {action} {grade}级 {confidence}%")
|
||
print("")
|
||
|
||
# 统计汇总
|
||
print(f"📈 做多信号: {buy_count} 个")
|
||
print(f"📉 做空信号: {sell_count} 个")
|
||
print("="*80)
|
||
|
||
# 发送汇总通知
|
||
if send_notification:
|
||
asyncio.run(send_summary_notification(
|
||
results, total, with_signals, notified,
|
||
buy_count, sell_count, high_quality_signals, all_signals, threshold,
|
||
len(us_results), len(hk_results)
|
||
))
|
||
|
||
|
||
async def send_summary_notification(
|
||
results: list,
|
||
total: int,
|
||
with_signals: list,
|
||
notified: list,
|
||
buy_count: int,
|
||
sell_count: int,
|
||
high_quality_signals: list,
|
||
all_signals: list,
|
||
threshold: float,
|
||
us_count: int = 0,
|
||
hk_count: int = 0
|
||
):
|
||
"""发送汇总报告到飞书和Telegram
|
||
|
||
Args:
|
||
results: 分析结果列表
|
||
total: 总数
|
||
with_signals: 有信号的股票列表
|
||
notified: 已通知的股票列表
|
||
buy_count: 做多信号数量
|
||
sell_count: 做空信号数量
|
||
high_quality_signals: 达到阈值的高等级信号列表
|
||
all_signals: 所有信号列表
|
||
threshold: 通知阈值
|
||
us_count: 美股数量
|
||
hk_count: 港股数量
|
||
"""
|
||
try:
|
||
from datetime import datetime
|
||
feishu = get_feishu_service()
|
||
telegram = get_telegram_service()
|
||
|
||
now = datetime.now()
|
||
|
||
# 构建飞书汇总内容
|
||
content_parts = [
|
||
f"**📊 股票分析汇总报告**",
|
||
f"",
|
||
f"⏰ 时间: {now.strftime('%Y-%m-%d %H:%M')}",
|
||
f"",
|
||
f"📊 **分析概况**",
|
||
f"• 美股: {us_count} 只 | 港股: {hk_count} 只",
|
||
f"• 发现信号: {len(with_signals)} 只",
|
||
f"• 已发通知: {len(notified)} 只",
|
||
f"• 通知阈值: {threshold:.0f}%",
|
||
f"",
|
||
]
|
||
|
||
# 高等级信号（达到阈值的）
|
||
if high_quality_signals:
|
||
content_parts.append(f"⭐ **高等级信号 (A/B级 ≥ {threshold:.0f}%)**")
|
||
for sig in high_quality_signals[:5]:
|
||
symbol = sig['symbol']
|
||
stock_name = sig.get('stock_name', '')
|
||
market_tag = '[港股]' if sig.get('is_hk') else '[美股]'
|
||
action = '🟢 做多' if sig.get('action') == 'buy' else '🔴 做空'
|
||
grade = sig.get('grade', 'D')
|
||
confidence = sig.get('confidence', 0)
|
||
|
||
# 构建带名称的股票显示
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
|
||
content_parts.append(f"• {market_tag} {symbol_display} {action} {grade}级 {confidence}%")
|
||
content_parts.append(f"")
|
||
|
||
# 信号统计
|
||
content_parts.extend([
|
||
f"📈 做多信号: {buy_count} 个",
|
||
f"📉 做空信号: {sell_count} 个",
|
||
f"",
|
||
f"*⚠️ 仅供参考，不构成投资建议*"
|
||
])
|
||
|
||
content = "\n".join(content_parts)
|
||
|
||
# 发送飞书
|
||
title = f"📊 股票分析汇总 ({now.strftime('%H:%M')})"
|
||
color = "blue"
|
||
|
||
await feishu.send_card(title, content, color)
|
||
|
||
# 发送 Telegram
|
||
telegram_msg = f"📊 *股票分析汇总*\n\n"
|
||
telegram_msg += f"时间: {now.strftime('%H:%M')}\n"
|
||
telegram_msg += f"美股: {us_count}只 | 港股: {hk_count}只\n"
|
||
telegram_msg += f"信号: {len(with_signals)}只 | 通知: {len(notified)}只\n"
|
||
telegram_msg += f"阈值: {threshold:.0f}%\n\n"
|
||
|
||
if high_quality_signals:
|
||
telegram_msg += f"⭐ *高等级信号 (≥{threshold:.0f}%)*\n"
|
||
for sig in high_quality_signals[:5]:
|
||
symbol = sig['symbol']
|
||
stock_name = sig.get('stock_name', '')
|
||
market_tag = '[港股]' if sig.get('is_hk') else '[美股]'
|
||
action = '🟢 做多' if sig.get('action') == 'buy' else '🔴 做空'
|
||
grade = sig.get('grade', 'D')
|
||
confidence = sig.get('confidence', 0)
|
||
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
telegram_msg += f"{market_tag} {symbol_display} {action} {grade}级 {confidence}%\n"
|
||
telegram_msg += "\n"
|
||
|
||
telegram_msg += f"做多: {buy_count} | 做空: {sell_count}"
|
||
|
||
await telegram.send_message(telegram_msg)
|
||
|
||
print(f"\n📬 汇总报告已发送到飞书和Telegram")
|
||
|
||
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)
|
||
|
||
# 过滤掉无效的参数（如环境变量路径、配置项等）
|
||
# 只接受有效的股票代码（字母开头，1-5个字符，或包含数字的港股代码如0700.HK）
|
||
raw_symbols = sys.argv[1:]
|
||
symbols = []
|
||
for arg in raw_symbols:
|
||
# 跳过包含路径分隔符、等号、或明显不是股票代码的参数
|
||
if '/' in arg or '=' in arg or ':' in arg or len(arg) > 10:
|
||
logger.debug(f"跳过无效参数: {arg}")
|
||
continue
|
||
# 接受纯字母的参数（美股股票代码）
|
||
if arg.isalpha() and 1 <= len(arg) <= 5:
|
||
symbols.append(arg.upper())
|
||
# 接受包含数字和点号的参数（港股代码，如0700.HK）
|
||
elif '.HK' in arg.upper() and len(arg) <= 10:
|
||
symbols.append(arg.upper())
|
||
else:
|
||
logger.debug(f"跳过非股票代码参数: {arg}")
|
||
|
||
if not symbols:
|
||
print("❌ 未找到有效的股票代码")
|
||
print(f"接收到的参数: {raw_symbols}")
|
||
print("\n用法: python3 scripts/test_stock.py AAPL [TSLA] [NVDA] ...")
|
||
sys.exit(1)
|
||
|
||
print("="*60)
|
||
print("🤖 美股分析脚本")
|
||
print("="*60)
|
||
print(f"股票: {', '.join(symbols)}")
|
||
print(f"时间: {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
# 收集所有分析结果
|
||
results = []
|
||
for symbol in symbols:
|
||
result = await analyze(symbol.upper(), send_notification=True)
|
||
results.append(result)
|
||
|
||
# 生成汇总报告并发送通知
|
||
await send_summary_notification_async(results)
|
||
|
||
print("\n" + "="*60)
|
||
print("✅ 分析完成")
|
||
print("="*60)
|
||
|
||
|
||
async def send_summary_notification_async(results: list):
|
||
"""异步发送汇总通知"""
|
||
from app.config import get_settings
|
||
from app.stock_agent.stock_agent import STOCK_NAMES
|
||
settings = get_settings()
|
||
threshold = settings.stock_llm_threshold * 100 # 获取阈值
|
||
|
||
total = len(results)
|
||
with_signals = [r for r in results if r.get('signals')]
|
||
notified = [r for r in results if r.get('notified')]
|
||
|
||
# 区分美股和港股
|
||
us_results = [r for r in results if not r['symbol'].endswith('.HK')]
|
||
hk_results = [r for r in results if r['symbol'].endswith('.HK')]
|
||
|
||
# 统计信号
|
||
buy_count = 0
|
||
sell_count = 0
|
||
high_quality_signals = [] # A/B级信号且达到阈值
|
||
all_signals = [] # 所有信号
|
||
|
||
for r in with_signals:
|
||
for sig in r.get('signals', []):
|
||
sig['symbol'] = r['symbol']
|
||
sig['current_price'] = r.get('price', 0)
|
||
sig['is_hk'] = r['symbol'].endswith('.HK')
|
||
sig['stock_name'] = STOCK_NAMES.get(r['symbol'], '')
|
||
all_signals.append(sig)
|
||
|
||
if sig.get('action') == 'buy':
|
||
buy_count += 1
|
||
elif sig.get('action') == 'sell':
|
||
sell_count += 1
|
||
|
||
# 只统计达到阈值的A/B级信号
|
||
if sig.get('grade') in ['A', 'B'] and sig.get('confidence', 0) >= threshold:
|
||
high_quality_signals.append(sig)
|
||
|
||
# 按置信度排序
|
||
high_quality_signals.sort(key=lambda x: x.get('confidence', 0), reverse=True)
|
||
all_signals.sort(key=lambda x: x.get('confidence', 0), reverse=True)
|
||
|
||
# 打印汇总
|
||
print("\n" + "="*80)
|
||
print("📊 股票分析汇总报告")
|
||
print("="*80)
|
||
print(f"分析总数: {total} 只 (美股: {len(us_results)}, 港股: {len(hk_results)})")
|
||
print(f"有信号: {len(with_signals)} 只")
|
||
print(f"已通知: {len(notified)} 只")
|
||
print(f"通知阈值: {threshold}%")
|
||
print("")
|
||
|
||
# 显示高等级信号（达到阈值的）
|
||
if high_quality_signals:
|
||
print(f"⭐ 高等级信号达到阈值 (A/B级 >= {threshold}%): {len(high_quality_signals)} 个")
|
||
for sig in high_quality_signals[:10]:
|
||
symbol = sig['symbol']
|
||
stock_name = sig.get('stock_name', '')
|
||
market_tag = '[港股]' if sig.get('is_hk') else '[美股]'
|
||
action = '🟢 做多' if sig.get('action') == 'buy' else '🔴 做空'
|
||
grade = sig.get('grade', 'D')
|
||
confidence = sig.get('confidence', 0)
|
||
price = sig.get('current_price', 0)
|
||
entry = sig.get('entry_price', 0)
|
||
|
||
# 构建带名称的股票显示
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
|
||
print(f" {market_tag} {symbol_display} {action} [{grade}级] {confidence}% @ ${price:,.2f}")
|
||
if entry > 0:
|
||
print(f" 入场: ${entry:,.2f}")
|
||
print("")
|
||
|
||
# 显示未达到阈值但质量不错的信号
|
||
below_threshold = [s for s in all_signals
|
||
if s.get('grade') in ['A', 'B'] and s.get('confidence', 0) < threshold]
|
||
if below_threshold:
|
||
print(f"⚠️ 以下信号未达到通知阈值 ({threshold}%):")
|
||
for sig in below_threshold[:10]:
|
||
symbol = sig['symbol']
|
||
stock_name = sig.get('stock_name', '')
|
||
market_tag = '[港股]' if sig.get('is_hk') else '[美股]'
|
||
action = '🟢 做多' if sig.get('action') == 'buy' else '🔴 做空'
|
||
grade = sig.get('grade', 'D')
|
||
confidence = sig.get('confidence', 0)
|
||
|
||
# 构建带名称的股票显示
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
|
||
print(f" {market_tag} {symbol_display} {action} {grade}级 {confidence}%")
|
||
print("")
|
||
|
||
# 统计汇总
|
||
print(f"📈 做多信号: {buy_count} 个")
|
||
print(f"📉 做空信号: {sell_count} 个")
|
||
print("="*80)
|
||
|
||
# 发送汇总通知
|
||
await send_summary_notification(
|
||
results, total, with_signals, notified,
|
||
buy_count, sell_count, high_quality_signals, all_signals, threshold,
|
||
len(us_results), len(hk_results)
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
asyncio.run(main())
|
||
except KeyboardInterrupt:
|
||
print("\n\n⚠️ 用户中断")
|