649 lines
25 KiB
Python
Executable File
649 lines
25 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.services.fundamental_service import get_fundamental_service
|
||
from app.services.news_service import get_news_service
|
||
from app.stock_agent.market_signal_analyzer import StockMarketSignalAnalyzer
|
||
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:
|
||
分析结果字典
|
||
"""
|
||
# 获取服务
|
||
yf_service = get_yfinance_service()
|
||
market_analyzer = StockMarketSignalAnalyzer() # 使用新的市场信号分析器
|
||
fundamental = get_fundamental_service() # 基本面服务
|
||
news = get_news_service() # 新闻服务
|
||
feishu = get_feishu_service()
|
||
telegram = get_telegram_service()
|
||
|
||
result = {
|
||
'symbol': symbol,
|
||
'stock_name': '', # 从基本面数据获取
|
||
'price': 0,
|
||
'signals': [],
|
||
'notified': False
|
||
}
|
||
|
||
# 获取配置
|
||
settings = get_settings()
|
||
threshold = settings.stock_llm_threshold * 100 # 转换为百分比
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"📊 分析 {symbol}")
|
||
print(f"{'='*60}")
|
||
|
||
try:
|
||
# 获取行情
|
||
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())}")
|
||
|
||
# 获取基本面数据（包含公司名称）
|
||
print(f"\n📈 基本面分析中...")
|
||
fundamental_data = None
|
||
fundamental_summary = ""
|
||
stock_name = ""
|
||
try:
|
||
fundamental_data = fundamental.get_fundamental_data(symbol)
|
||
if fundamental_data:
|
||
# 传递已获取的数据，避免重复调用
|
||
fundamental_summary = fundamental.get_fundamental_summary(symbol, fundamental_data)
|
||
# 获取公司名称
|
||
stock_name = fundamental_data.get('company_name', '')
|
||
result['stock_name'] = stock_name
|
||
# 输出基本面详细信息
|
||
score = fundamental_data.get('score', {})
|
||
print(f" ✓ 基本面数据获取成功")
|
||
|
||
# 公司信息
|
||
company = fundamental_data.get('company_name', 'N/A')
|
||
sector = fundamental_data.get('sector', 'N/A')
|
||
print(f" 【公司】{company} | {sector}")
|
||
|
||
# 更新显示
|
||
symbol_display = f"{stock_name}({symbol})" if stock_name else symbol
|
||
print(f"\n{'='*60}")
|
||
print(f"📊 分析 {symbol_display} @ ${price:,.2f}")
|
||
print(f"{'='*60}")
|
||
|
||
# 评分
|
||
print(f" 【评分】总分: {score.get('total', 0):.0f}/100 ({score.get('rating', 'N/A')}级) | "
|
||
f"估值:{score.get('valuation', 0)} 盈利:{score.get('profitability', 0)} "
|
||
f"成长:{score.get('growth', 0)} 财务:{score.get('financial_health', 0)}")
|
||
|
||
# 估值指标
|
||
val = fundamental_data.get('valuation', {})
|
||
if val.get('pe_ratio'):
|
||
pe = val['pe_ratio']
|
||
pb = val.get('pb_ratio')
|
||
ps = val.get('ps_ratio')
|
||
peg = val.get('peg_ratio')
|
||
pb_str = f"{pb:.2f}" if pb is not None else "N/A"
|
||
ps_str = f"{ps:.2f}" if ps is not None else "N/A"
|
||
peg_str = f"{peg:.2f}" if peg is not None else "N/A"
|
||
print(f" 【估值】PE:{pe:.2f} | PB:{pb_str} | PS:{ps_str} | PEG:{peg_str}")
|
||
|
||
# 盈利能力
|
||
prof = fundamental_data.get('profitability', {})
|
||
if prof.get('return_on_equity'):
|
||
roe = prof['return_on_equity']
|
||
pm = prof.get('profit_margin')
|
||
gm = prof.get('gross_margin')
|
||
pm_str = f"{pm:.1f}" if pm is not None else "N/A"
|
||
gm_str = f"{gm:.1f}" if gm is not None else "N/A"
|
||
print(f" 【盈利】ROE:{roe:.2f}% | 净利率:{pm_str}% | 毛利率:{gm_str}%")
|
||
|
||
# 成长性
|
||
growth = fundamental_data.get('growth', {})
|
||
rg = growth.get('revenue_growth')
|
||
eg = growth.get('earnings_growth')
|
||
if rg is not None or eg is not None:
|
||
rg_str = f"{rg:.1f}" if rg is not None else "N/A"
|
||
eg_str = f"{eg:.1f}" if eg is not None else "N/A"
|
||
print(f" 【成长】营收增长:{rg_str}% | 盈利增长:{eg_str}%")
|
||
|
||
# 财务健康
|
||
fin = fundamental_data.get('financial_health', {})
|
||
if fin.get('debt_to_equity'):
|
||
de = fin['debt_to_equity']
|
||
cr = fin.get('current_ratio')
|
||
cr_str = f"{cr:.2f}" if cr is not None else "N/A"
|
||
print(f" 【财务】债务股本比:{de:.2f} | 流动比率:{cr_str}")
|
||
|
||
# 分析师建议
|
||
analyst = fundamental_data.get('analyst', {})
|
||
tp = analyst.get('target_price')
|
||
if tp:
|
||
rec = analyst.get('recommendation', 'N/A')
|
||
print(f" 【分析师】目标价:${tp:.2f} | 评级:{rec}")
|
||
|
||
else:
|
||
print(f" ⚠️ 无法获取基本面数据")
|
||
except Exception as e:
|
||
print(f" ⚠️ 获取基本面数据失败: {e}")
|
||
|
||
# 获取新闻数据
|
||
print(f"\n📰 新闻分析...")
|
||
news_data = None
|
||
try:
|
||
news_data = await news.search_stock_news(symbol, stock_name, max_results=5)
|
||
if news_data:
|
||
print(f" 获取到 {len(news_data)} 条相关新闻")
|
||
else:
|
||
print(f" 暂无相关新闻")
|
||
except Exception as e:
|
||
print(f" ⚠️ 获取新闻数据失败: {e}")
|
||
|
||
# 市场信号分析（使用新架构 - 技术面 + 基本面 + 新闻）
|
||
print(f"\n🤖 市场信号分析中...\n")
|
||
market_signal = await market_analyzer.analyze(
|
||
symbol, data,
|
||
symbols=[symbol],
|
||
fundamental_data=fundamental_data,
|
||
news_data=news_data
|
||
)
|
||
|
||
# 输出结果
|
||
summary = market_signal.get('analysis_summary', '')
|
||
signals = market_signal.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:
|
||
# 使用格式化工具格式化通知
|
||
from app.utils.signal_formatter import get_signal_formatter
|
||
formatter = get_signal_formatter()
|
||
|
||
card = formatter.format_feishu_card(best_signal, symbol, agent_type='stock')
|
||
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(formatter.format_signal_message(best_signal, symbol, agent_type='stock'))
|
||
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
|
||
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'] = r.get('stock_name', '')
|
||
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
|
||
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'] = r.get('stock_name', '')
|
||
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⚠️ 用户中断")
|