mirror of
https://github.com/ZhuLinsen/daily_stock_analysis
synced 2026-09-20 10:53:33 +08:00
1074 lines
41 KiB
Python
1074 lines
41 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
===================================
|
||
A股自选股智能分析系统 - 主调度程序
|
||
===================================
|
||
|
||
职责：
|
||
1. 协调各模块完成股票分析流程
|
||
2. 实现低并发的线程池调度
|
||
3. 全局异常处理，确保单股失败不影响整体
|
||
4. 提供命令行入口
|
||
|
||
使用方式：
|
||
python main.py # 正常运行
|
||
python main.py --debug # 调试模式
|
||
python main.py --dry-run # 仅获取数据不分析
|
||
|
||
交易理念（已融入分析）：
|
||
- 严进策略：不追高，乖离率 > 5% 不买入
|
||
- 趋势交易：只做 MA5>MA10>MA20 多头排列
|
||
- 效率优先：关注筹码集中度好的股票
|
||
- 买点偏好：缩量回踩 MA5/MA10 支撑
|
||
"""
|
||
import os
|
||
|
||
# 代理配置 - 仅在本地环境使用，GitHub Actions 不需要
|
||
if os.getenv("GITHUB_ACTIONS") != "true":
|
||
# 本地开发环境，如需代理请取消注释或修改端口
|
||
# os.environ["http_proxy"] = "http://127.0.0.1:10809"
|
||
# os.environ["https_proxy"] = "http://127.0.0.1:10809"
|
||
pass
|
||
|
||
import argparse
|
||
import logging
|
||
import sys
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from datetime import datetime, date, timezone, timedelta
|
||
from logging.handlers import RotatingFileHandler
|
||
from pathlib import Path
|
||
from typing import List, Dict, Any, Optional, Tuple
|
||
from feishu_doc import FeishuDocManager
|
||
|
||
from config import get_config, Config
|
||
from storage import get_db, DatabaseManager
|
||
from data_provider import DataFetcherManager
|
||
from data_provider.akshare_fetcher import AkshareFetcher, RealtimeQuote, ChipDistribution
|
||
from analyzer import GeminiAnalyzer, AnalysisResult, STOCK_NAME_MAP
|
||
from notification import NotificationService, NotificationChannel, send_daily_report
|
||
from bot.models import BotMessage
|
||
from search_service import SearchService, SearchResponse
|
||
from enums import ReportType
|
||
from stock_analyzer import StockTrendAnalyzer, TrendAnalysisResult
|
||
from market_analyzer import MarketAnalyzer
|
||
|
||
# 配置日志格式
|
||
LOG_FORMAT = '%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s'
|
||
LOG_DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
|
||
|
||
|
||
def setup_logging(debug: bool = False, log_dir: str = "./logs") -> None:
|
||
"""
|
||
配置日志系统（同时输出到控制台和文件）
|
||
|
||
Args:
|
||
debug: 是否启用调试模式
|
||
log_dir: 日志文件目录
|
||
"""
|
||
level = logging.DEBUG if debug else logging.INFO
|
||
|
||
# 创建日志目录
|
||
log_path = Path(log_dir)
|
||
log_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 日志文件路径（按日期分文件）
|
||
today_str = datetime.now().strftime('%Y%m%d')
|
||
log_file = log_path / f"stock_analysis_{today_str}.log"
|
||
debug_log_file = log_path / f"stock_analysis_debug_{today_str}.log"
|
||
|
||
# 创建根 logger
|
||
root_logger = logging.getLogger()
|
||
root_logger.setLevel(logging.DEBUG) # 根 logger 设为 DEBUG，由 handler 控制输出级别
|
||
|
||
# Handler 1: 控制台输出
|
||
console_handler = logging.StreamHandler(sys.stdout)
|
||
console_handler.setLevel(level)
|
||
console_handler.setFormatter(logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT))
|
||
root_logger.addHandler(console_handler)
|
||
|
||
# Handler 2: 常规日志文件（INFO 级别，10MB 轮转）
|
||
file_handler = RotatingFileHandler(
|
||
log_file,
|
||
maxBytes=10 * 1024 * 1024, # 10MB
|
||
backupCount=5,
|
||
encoding='utf-8'
|
||
)
|
||
file_handler.setLevel(logging.INFO)
|
||
file_handler.setFormatter(logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT))
|
||
root_logger.addHandler(file_handler)
|
||
|
||
# Handler 3: 调试日志文件（DEBUG 级别，包含所有详细信息）
|
||
debug_handler = RotatingFileHandler(
|
||
debug_log_file,
|
||
maxBytes=50 * 1024 * 1024, # 50MB
|
||
backupCount=3,
|
||
encoding='utf-8'
|
||
)
|
||
debug_handler.setLevel(logging.DEBUG)
|
||
debug_handler.setFormatter(logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT))
|
||
root_logger.addHandler(debug_handler)
|
||
|
||
# 降低第三方库的日志级别
|
||
logging.getLogger('urllib3').setLevel(logging.WARNING)
|
||
logging.getLogger('sqlalchemy').setLevel(logging.WARNING)
|
||
logging.getLogger('google').setLevel(logging.WARNING)
|
||
logging.getLogger('httpx').setLevel(logging.WARNING)
|
||
|
||
logging.info(f"日志系统初始化完成，日志目录: {log_path.absolute()}")
|
||
logging.info(f"常规日志: {log_file}")
|
||
logging.info(f"调试日志: {debug_log_file}")
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class StockAnalysisPipeline:
|
||
"""
|
||
股票分析主流程调度器
|
||
|
||
职责：
|
||
1. 管理整个分析流程
|
||
2. 协调数据获取、存储、搜索、分析、通知等模块
|
||
3. 实现并发控制和异常处理
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
config: Optional[Config] = None,
|
||
max_workers: Optional[int] = None,
|
||
source_message: Optional[BotMessage] = None
|
||
):
|
||
"""
|
||
初始化调度器
|
||
|
||
Args:
|
||
config: 配置对象（可选，默认使用全局配置）
|
||
max_workers: 最大并发线程数（可选，默认从配置读取）
|
||
"""
|
||
self.config = config or get_config()
|
||
self.max_workers = max_workers or self.config.max_workers
|
||
self.source_message = source_message
|
||
|
||
# 初始化各模块
|
||
self.db = get_db()
|
||
self.fetcher_manager = DataFetcherManager()
|
||
self.akshare_fetcher = AkshareFetcher() # 用于获取增强数据（量比、筹码等）
|
||
self.trend_analyzer = StockTrendAnalyzer() # 趋势分析器
|
||
self.analyzer = GeminiAnalyzer()
|
||
self.notifier = NotificationService(source_message=source_message)
|
||
|
||
# 初始化搜索服务
|
||
self.search_service = SearchService(
|
||
bocha_keys=self.config.bocha_api_keys,
|
||
tavily_keys=self.config.tavily_api_keys,
|
||
serpapi_keys=self.config.serpapi_keys,
|
||
)
|
||
|
||
logger.info(f"调度器初始化完成，最大并发数: {self.max_workers}")
|
||
logger.info("已启用趋势分析器 (MA5>MA10>MA20 多头判断)")
|
||
if self.search_service.is_available:
|
||
logger.info("搜索服务已启用 (Tavily/SerpAPI)")
|
||
else:
|
||
logger.warning("搜索服务未启用（未配置 API Key）")
|
||
|
||
def fetch_and_save_stock_data(
|
||
self,
|
||
code: str,
|
||
force_refresh: bool = False
|
||
) -> Tuple[bool, Optional[str]]:
|
||
"""
|
||
获取并保存单只股票数据
|
||
|
||
断点续传逻辑：
|
||
1. 检查数据库是否已有今日数据
|
||
2. 如果有且不强制刷新，则跳过网络请求
|
||
3. 否则从数据源获取并保存
|
||
|
||
Args:
|
||
code: 股票代码
|
||
force_refresh: 是否强制刷新（忽略本地缓存）
|
||
|
||
Returns:
|
||
Tuple[是否成功, 错误信息]
|
||
"""
|
||
try:
|
||
today = date.today()
|
||
|
||
# 断点续传检查：如果今日数据已存在，跳过
|
||
if not force_refresh and self.db.has_today_data(code, today):
|
||
logger.info(f"[{code}] 今日数据已存在，跳过获取（断点续传）")
|
||
return True, None
|
||
|
||
# 从数据源获取数据
|
||
logger.info(f"[{code}] 开始从数据源获取数据...")
|
||
df, source_name = self.fetcher_manager.get_daily_data(code, days=30)
|
||
|
||
if df is None or df.empty:
|
||
return False, "获取数据为空"
|
||
|
||
# 保存到数据库
|
||
saved_count = self.db.save_daily_data(df, code, source_name)
|
||
logger.info(f"[{code}] 数据保存成功（来源: {source_name}，新增 {saved_count} 条）")
|
||
|
||
return True, None
|
||
|
||
except Exception as e:
|
||
error_msg = f"获取/保存数据失败: {str(e)}"
|
||
logger.error(f"[{code}] {error_msg}")
|
||
return False, error_msg
|
||
|
||
def analyze_stock(self, code: str) -> Optional[AnalysisResult]:
|
||
"""
|
||
分析单只股票（增强版：含量比、换手率、筹码分析、多维度情报）
|
||
|
||
流程：
|
||
1. 获取实时行情（量比、换手率）
|
||
2. 获取筹码分布
|
||
3. 进行趋势分析（基于交易理念）
|
||
4. 多维度情报搜索（最新消息+风险排查+业绩预期）
|
||
5. 从数据库获取分析上下文
|
||
6. 调用 AI 进行综合分析
|
||
|
||
Args:
|
||
code: 股票代码
|
||
|
||
Returns:
|
||
AnalysisResult 或 None（如果分析失败）
|
||
"""
|
||
try:
|
||
# 获取股票名称（优先从实时行情获取真实名称）
|
||
stock_name = STOCK_NAME_MAP.get(code, '')
|
||
|
||
# Step 1: 获取实时行情（量比、换手率等）
|
||
realtime_quote: Optional[RealtimeQuote] = None
|
||
try:
|
||
realtime_quote = self.akshare_fetcher.get_realtime_quote(code)
|
||
if realtime_quote:
|
||
# 使用实时行情返回的真实股票名称
|
||
if realtime_quote.name:
|
||
stock_name = realtime_quote.name
|
||
logger.info(f"[{code}] {stock_name} 实时行情: 价格={realtime_quote.price}, "
|
||
f"量比={realtime_quote.volume_ratio}, 换手率={realtime_quote.turnover_rate}%")
|
||
except Exception as e:
|
||
logger.warning(f"[{code}] 获取实时行情失败: {e}")
|
||
|
||
# 如果还是没有名称，使用代码作为名称
|
||
if not stock_name:
|
||
stock_name = f'股票{code}'
|
||
|
||
# Step 2: 获取筹码分布
|
||
chip_data: Optional[ChipDistribution] = None
|
||
try:
|
||
chip_data = self.akshare_fetcher.get_chip_distribution(code)
|
||
if chip_data:
|
||
logger.info(f"[{code}] 筹码分布: 获利比例={chip_data.profit_ratio:.1%}, "
|
||
f"90%集中度={chip_data.concentration_90:.2%}")
|
||
except Exception as e:
|
||
logger.warning(f"[{code}] 获取筹码分布失败: {e}")
|
||
|
||
# Step 3: 趋势分析（基于交易理念）
|
||
trend_result: Optional[TrendAnalysisResult] = None
|
||
try:
|
||
# 获取历史数据进行趋势分析
|
||
context = self.db.get_analysis_context(code)
|
||
if context and 'raw_data' in context:
|
||
import pandas as pd
|
||
raw_data = context['raw_data']
|
||
if isinstance(raw_data, list) and len(raw_data) > 0:
|
||
df = pd.DataFrame(raw_data)
|
||
trend_result = self.trend_analyzer.analyze(df, code)
|
||
logger.info(f"[{code}] 趋势分析: {trend_result.trend_status.value}, "
|
||
f"买入信号={trend_result.buy_signal.value}, 评分={trend_result.signal_score}")
|
||
except Exception as e:
|
||
logger.warning(f"[{code}] 趋势分析失败: {e}")
|
||
|
||
# Step 4: 多维度情报搜索（最新消息+风险排查+业绩预期）
|
||
news_context = None
|
||
if self.search_service.is_available:
|
||
logger.info(f"[{code}] 开始多维度情报搜索...")
|
||
|
||
# 使用多维度搜索（最多3次搜索）
|
||
intel_results = self.search_service.search_comprehensive_intel(
|
||
stock_code=code,
|
||
stock_name=stock_name,
|
||
max_searches=3
|
||
)
|
||
|
||
# 格式化情报报告
|
||
if intel_results:
|
||
news_context = self.search_service.format_intel_report(intel_results, stock_name)
|
||
total_results = sum(
|
||
len(r.results) for r in intel_results.values() if r.success
|
||
)
|
||
logger.info(f"[{code}] 情报搜索完成: 共 {total_results} 条结果")
|
||
logger.debug(f"[{code}] 情报搜索结果:\n{news_context}")
|
||
else:
|
||
logger.info(f"[{code}] 搜索服务不可用，跳过情报搜索")
|
||
|
||
# Step 5: 获取分析上下文（技术面数据）
|
||
context = self.db.get_analysis_context(code)
|
||
|
||
if context is None:
|
||
logger.warning(f"[{code}] 无法获取分析上下文，跳过分析")
|
||
return None
|
||
|
||
# Step 6: 增强上下文数据（添加实时行情、筹码、趋势分析结果、股票名称）
|
||
enhanced_context = self._enhance_context(
|
||
context,
|
||
realtime_quote,
|
||
chip_data,
|
||
trend_result,
|
||
stock_name # 传入股票名称
|
||
)
|
||
|
||
# Step 7: 调用 AI 分析（传入增强的上下文和新闻）
|
||
result = self.analyzer.analyze(enhanced_context, news_context=news_context)
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
logger.error(f"[{code}] 分析失败: {e}")
|
||
logger.exception(f"[{code}] 详细错误信息:")
|
||
return None
|
||
|
||
def _enhance_context(
|
||
self,
|
||
context: Dict[str, Any],
|
||
realtime_quote: Optional[RealtimeQuote],
|
||
chip_data: Optional[ChipDistribution],
|
||
trend_result: Optional[TrendAnalysisResult],
|
||
stock_name: str = ""
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
增强分析上下文
|
||
|
||
将实时行情、筹码分布、趋势分析结果、股票名称添加到上下文中
|
||
|
||
Args:
|
||
context: 原始上下文
|
||
realtime_quote: 实时行情数据
|
||
chip_data: 筹码分布数据
|
||
trend_result: 趋势分析结果
|
||
stock_name: 股票名称
|
||
|
||
Returns:
|
||
增强后的上下文
|
||
"""
|
||
enhanced = context.copy()
|
||
|
||
# 添加股票名称
|
||
if stock_name:
|
||
enhanced['stock_name'] = stock_name
|
||
elif realtime_quote and realtime_quote.name:
|
||
enhanced['stock_name'] = realtime_quote.name
|
||
|
||
# 添加实时行情
|
||
if realtime_quote:
|
||
enhanced['realtime'] = {
|
||
'name': realtime_quote.name, # 股票名称
|
||
'price': realtime_quote.price,
|
||
'volume_ratio': realtime_quote.volume_ratio,
|
||
'volume_ratio_desc': self._describe_volume_ratio(realtime_quote.volume_ratio),
|
||
'turnover_rate': realtime_quote.turnover_rate,
|
||
'pe_ratio': realtime_quote.pe_ratio,
|
||
'pb_ratio': realtime_quote.pb_ratio,
|
||
'total_mv': realtime_quote.total_mv,
|
||
'circ_mv': realtime_quote.circ_mv,
|
||
'change_60d': realtime_quote.change_60d,
|
||
}
|
||
|
||
# 添加筹码分布
|
||
if chip_data:
|
||
current_price = realtime_quote.price if realtime_quote else 0
|
||
enhanced['chip'] = {
|
||
'profit_ratio': chip_data.profit_ratio,
|
||
'avg_cost': chip_data.avg_cost,
|
||
'concentration_90': chip_data.concentration_90,
|
||
'concentration_70': chip_data.concentration_70,
|
||
'chip_status': chip_data.get_chip_status(current_price),
|
||
}
|
||
|
||
# 添加趋势分析结果
|
||
if trend_result:
|
||
enhanced['trend_analysis'] = {
|
||
'trend_status': trend_result.trend_status.value,
|
||
'ma_alignment': trend_result.ma_alignment,
|
||
'trend_strength': trend_result.trend_strength,
|
||
'bias_ma5': trend_result.bias_ma5,
|
||
'bias_ma10': trend_result.bias_ma10,
|
||
'volume_status': trend_result.volume_status.value,
|
||
'volume_trend': trend_result.volume_trend,
|
||
'buy_signal': trend_result.buy_signal.value,
|
||
'signal_score': trend_result.signal_score,
|
||
'signal_reasons': trend_result.signal_reasons,
|
||
'risk_factors': trend_result.risk_factors,
|
||
}
|
||
|
||
return enhanced
|
||
|
||
def _describe_volume_ratio(self, volume_ratio: float) -> str:
|
||
"""
|
||
量比描述
|
||
|
||
量比 = 当前成交量 / 过去5日平均成交量
|
||
"""
|
||
if volume_ratio < 0.5:
|
||
return "极度萎缩"
|
||
elif volume_ratio < 0.8:
|
||
return "明显萎缩"
|
||
elif volume_ratio < 1.2:
|
||
return "正常"
|
||
elif volume_ratio < 2.0:
|
||
return "温和放量"
|
||
elif volume_ratio < 3.0:
|
||
return "明显放量"
|
||
else:
|
||
return "巨量"
|
||
|
||
def process_single_stock(
|
||
self,
|
||
code: str,
|
||
skip_analysis: bool = False,
|
||
single_stock_notify: bool = False,
|
||
report_type: ReportType = ReportType.SIMPLE
|
||
) -> Optional[AnalysisResult]:
|
||
"""
|
||
处理单只股票的完整流程
|
||
|
||
包括：
|
||
1. 获取数据
|
||
2. 保存数据
|
||
3. AI 分析
|
||
4. 单股推送（可选，#55）
|
||
|
||
此方法会被线程池调用，需要处理好异常
|
||
|
||
Args:
|
||
code: 股票代码
|
||
skip_analysis: 是否跳过 AI 分析
|
||
single_stock_notify: 是否启用单股推送模式（每分析完一只立即推送）
|
||
report_type: 报告类型枚举（从配置读取，Issue #119）
|
||
|
||
Returns:
|
||
AnalysisResult 或 None
|
||
"""
|
||
logger.info(f"========== 开始处理 {code} ==========")
|
||
|
||
try:
|
||
# Step 1: 获取并保存数据
|
||
success, error = self.fetch_and_save_stock_data(code)
|
||
|
||
if not success:
|
||
logger.warning(f"[{code}] 数据获取失败: {error}")
|
||
# 即使获取失败，也尝试用已有数据分析
|
||
|
||
# Step 2: AI 分析
|
||
if skip_analysis:
|
||
logger.info(f"[{code}] 跳过 AI 分析（dry-run 模式）")
|
||
return None
|
||
|
||
result = self.analyze_stock(code)
|
||
|
||
if result:
|
||
logger.info(
|
||
f"[{code}] 分析完成: {result.operation_advice}, "
|
||
f"评分 {result.sentiment_score}"
|
||
)
|
||
|
||
# 单股推送模式（#55）：每分析完一只股票立即推送
|
||
if single_stock_notify and self.notifier.is_available():
|
||
try:
|
||
# 根据报告类型选择生成方法
|
||
if report_type == ReportType.FULL:
|
||
# 完整报告：使用决策仪表盘格式
|
||
report_content = self.notifier.generate_dashboard_report([result])
|
||
logger.info(f"[{code}] 使用完整报告格式")
|
||
else:
|
||
# 精简报告：使用单股报告格式（默认）
|
||
report_content = self.notifier.generate_single_stock_report(result)
|
||
logger.info(f"[{code}] 使用精简报告格式")
|
||
|
||
if self.notifier.send(report_content):
|
||
logger.info(f"[{code}] 单股推送成功")
|
||
else:
|
||
logger.warning(f"[{code}] 单股推送失败")
|
||
except Exception as e:
|
||
logger.error(f"[{code}] 单股推送异常: {e}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
# 捕获所有异常，确保单股失败不影响整体
|
||
logger.exception(f"[{code}] 处理过程发生未知异常: {e}")
|
||
return None
|
||
|
||
def run(
|
||
self,
|
||
stock_codes: Optional[List[str]] = None,
|
||
dry_run: bool = False,
|
||
send_notification: bool = True
|
||
) -> List[AnalysisResult]:
|
||
"""
|
||
运行完整的分析流程
|
||
|
||
流程：
|
||
1. 获取待分析的股票列表
|
||
2. 使用线程池并发处理
|
||
3. 收集分析结果
|
||
4. 发送通知
|
||
|
||
Args:
|
||
stock_codes: 股票代码列表（可选，默认使用配置中的自选股）
|
||
dry_run: 是否仅获取数据不分析
|
||
send_notification: 是否发送推送通知
|
||
|
||
Returns:
|
||
分析结果列表
|
||
"""
|
||
start_time = time.time()
|
||
|
||
# 使用配置中的股票列表
|
||
if stock_codes is None:
|
||
self.config.refresh_stock_list()
|
||
stock_codes = self.config.stock_list
|
||
|
||
if not stock_codes:
|
||
logger.error("未配置自选股列表，请在 .env 文件中设置 STOCK_LIST")
|
||
return []
|
||
|
||
logger.info(f"===== 开始分析 {len(stock_codes)} 只股票 =====")
|
||
logger.info(f"股票列表: {', '.join(stock_codes)}")
|
||
logger.info(f"并发数: {self.max_workers}, 模式: {'仅获取数据' if dry_run else '完整分析'}")
|
||
|
||
# 单股推送模式（#55）：从配置读取
|
||
single_stock_notify = getattr(self.config, 'single_stock_notify', False)
|
||
# Issue #119: 从配置读取报告类型
|
||
report_type_str = getattr(self.config, 'report_type', 'simple').lower()
|
||
report_type = ReportType.FULL if report_type_str == 'full' else ReportType.SIMPLE
|
||
# Issue #128: 从配置读取分析间隔
|
||
analysis_delay = getattr(self.config, 'analysis_delay', 0)
|
||
|
||
if single_stock_notify:
|
||
logger.info(f"已启用单股推送模式：每分析完一只股票立即推送（报告类型: {report_type_str}）")
|
||
|
||
results: List[AnalysisResult] = []
|
||
|
||
# 使用线程池并发处理
|
||
# 注意：max_workers 设置较低（默认3）以避免触发反爬
|
||
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
|
||
# 提交任务
|
||
future_to_code = {
|
||
executor.submit(
|
||
self.process_single_stock,
|
||
code,
|
||
skip_analysis=dry_run,
|
||
single_stock_notify=single_stock_notify and send_notification,
|
||
report_type=report_type # Issue #119: 传递报告类型
|
||
): code
|
||
for code in stock_codes
|
||
}
|
||
|
||
# 收集结果
|
||
for idx, future in enumerate(as_completed(future_to_code)):
|
||
code = future_to_code[future]
|
||
try:
|
||
result = future.result()
|
||
if result:
|
||
results.append(result)
|
||
|
||
# Issue #128: 个股之间添加延迟，避免API限流
|
||
# 在非最后一只股票完成后添加延迟
|
||
if idx < len(stock_codes) - 1 and analysis_delay > 0:
|
||
logger.debug(f"等待 {analysis_delay} 秒后继续下一只股票...")
|
||
time.sleep(analysis_delay)
|
||
|
||
except Exception as e:
|
||
logger.error(f"[{code}] 任务执行失败: {e}")
|
||
|
||
# 统计
|
||
elapsed_time = time.time() - start_time
|
||
|
||
# dry-run 模式下，数据获取成功即视为成功
|
||
if dry_run:
|
||
# 检查哪些股票的数据今天已存在
|
||
success_count = sum(1 for code in stock_codes if self.db.has_today_data(code))
|
||
fail_count = len(stock_codes) - success_count
|
||
else:
|
||
success_count = len(results)
|
||
fail_count = len(stock_codes) - success_count
|
||
|
||
logger.info(f"===== 分析完成 =====")
|
||
logger.info(f"成功: {success_count}, 失败: {fail_count}, 耗时: {elapsed_time:.2f} 秒")
|
||
|
||
# 发送通知（单股推送模式下跳过汇总推送，避免重复）
|
||
if results and send_notification and not dry_run:
|
||
if single_stock_notify:
|
||
# 单股推送模式：只保存汇总报告，不再重复推送
|
||
logger.info("单股推送模式：跳过汇总推送，仅保存报告到本地")
|
||
self._send_notifications(results, skip_push=True)
|
||
else:
|
||
self._send_notifications(results)
|
||
|
||
return results
|
||
|
||
def _send_notifications(self, results: List[AnalysisResult], skip_push: bool = False) -> None:
|
||
"""
|
||
发送分析结果通知
|
||
|
||
生成决策仪表盘格式的报告
|
||
|
||
Args:
|
||
results: 分析结果列表
|
||
skip_push: 是否跳过推送（仅保存到本地，用于单股推送模式）
|
||
"""
|
||
try:
|
||
logger.info("生成决策仪表盘日报...")
|
||
|
||
# 生成决策仪表盘格式的详细日报
|
||
report = self.notifier.generate_dashboard_report(results)
|
||
|
||
# 保存到本地
|
||
filepath = self.notifier.save_report_to_file(report)
|
||
logger.info(f"决策仪表盘日报已保存: {filepath}")
|
||
|
||
# 跳过推送（单股推送模式）
|
||
if skip_push:
|
||
return
|
||
|
||
# 推送通知
|
||
if self.notifier.is_available():
|
||
channels = self.notifier.get_available_channels()
|
||
context_success = self.notifier.send_to_context(report)
|
||
|
||
# 企业微信：只发精简版（平台限制）
|
||
wechat_success = False
|
||
if NotificationChannel.WECHAT in channels:
|
||
dashboard_content = self.notifier.generate_wechat_dashboard(results)
|
||
logger.info(f"企业微信仪表盘长度: {len(dashboard_content)} 字符")
|
||
logger.debug(f"企业微信推送内容:\n{dashboard_content}")
|
||
wechat_success = self.notifier.send_to_wechat(dashboard_content)
|
||
|
||
# 其他渠道：发完整报告（避免自定义 Webhook 被 wechat 截断逻辑污染）
|
||
non_wechat_success = False
|
||
for channel in channels:
|
||
if channel == NotificationChannel.WECHAT:
|
||
continue
|
||
if channel == NotificationChannel.FEISHU:
|
||
non_wechat_success = self.notifier.send_to_feishu(report) or non_wechat_success
|
||
elif channel == NotificationChannel.TELEGRAM:
|
||
non_wechat_success = self.notifier.send_to_telegram(report) or non_wechat_success
|
||
elif channel == NotificationChannel.EMAIL:
|
||
non_wechat_success = self.notifier.send_to_email(report) or non_wechat_success
|
||
elif channel == NotificationChannel.CUSTOM:
|
||
non_wechat_success = self.notifier.send_to_custom(report) or non_wechat_success
|
||
else:
|
||
logger.warning(f"未知通知渠道: {channel}")
|
||
|
||
success = wechat_success or non_wechat_success or context_success
|
||
if success:
|
||
logger.info("决策仪表盘推送成功")
|
||
else:
|
||
logger.warning("决策仪表盘推送失败")
|
||
else:
|
||
logger.info("通知渠道未配置，跳过推送")
|
||
|
||
except Exception as e:
|
||
logger.error(f"发送通知失败: {e}")
|
||
|
||
|
||
def parse_arguments() -> argparse.Namespace:
|
||
"""解析命令行参数"""
|
||
parser = argparse.ArgumentParser(
|
||
description='A股自选股智能分析系统',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog='''
|
||
示例:
|
||
python main.py # 正常运行
|
||
python main.py --debug # 调试模式
|
||
python main.py --dry-run # 仅获取数据，不进行 AI 分析
|
||
python main.py --stocks 600519,000001 # 指定分析特定股票
|
||
python main.py --no-notify # 不发送推送通知
|
||
python main.py --single-notify # 启用单股推送模式（每分析完一只立即推送）
|
||
python main.py --schedule # 启用定时任务模式
|
||
python main.py --market-review # 仅运行大盘复盘
|
||
'''
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--debug',
|
||
action='store_true',
|
||
help='启用调试模式，输出详细日志'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--dry-run',
|
||
action='store_true',
|
||
help='仅获取数据，不进行 AI 分析'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--stocks',
|
||
type=str,
|
||
help='指定要分析的股票代码，逗号分隔（覆盖配置文件）'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--no-notify',
|
||
action='store_true',
|
||
help='不发送推送通知'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--single-notify',
|
||
action='store_true',
|
||
help='启用单股推送模式：每分析完一只股票立即推送，而不是汇总推送'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--workers',
|
||
type=int,
|
||
default=None,
|
||
help='并发线程数（默认使用配置值）'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--schedule',
|
||
action='store_true',
|
||
help='启用定时任务模式，每日定时执行'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--market-review',
|
||
action='store_true',
|
||
help='仅运行大盘复盘分析'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--no-market-review',
|
||
action='store_true',
|
||
help='跳过大盘复盘分析'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--webui',
|
||
action='store_true',
|
||
help='启动本地配置 WebUI'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--webui-only',
|
||
action='store_true',
|
||
help='仅启动 WebUI 服务，不自动执行分析（通过 /analysis API 手动触发）'
|
||
)
|
||
|
||
return parser.parse_args()
|
||
|
||
|
||
def run_market_review(notifier: NotificationService, analyzer=None, search_service=None) -> Optional[str]:
|
||
"""
|
||
执行大盘复盘分析
|
||
|
||
Args:
|
||
notifier: 通知服务
|
||
analyzer: AI分析器（可选）
|
||
search_service: 搜索服务（可选）
|
||
|
||
Returns:
|
||
复盘报告文本
|
||
"""
|
||
logger.info("开始执行大盘复盘分析...")
|
||
|
||
try:
|
||
market_analyzer = MarketAnalyzer(
|
||
search_service=search_service,
|
||
analyzer=analyzer
|
||
)
|
||
|
||
# 执行复盘
|
||
review_report = market_analyzer.run_daily_review()
|
||
|
||
if review_report:
|
||
# 保存报告到文件
|
||
date_str = datetime.now().strftime('%Y%m%d')
|
||
report_filename = f"market_review_{date_str}.md"
|
||
filepath = notifier.save_report_to_file(
|
||
f"# 🎯 大盘复盘\n\n{review_report}",
|
||
report_filename
|
||
)
|
||
logger.info(f"大盘复盘报告已保存: {filepath}")
|
||
|
||
# 推送通知
|
||
if notifier.is_available():
|
||
# 添加标题
|
||
report_content = f"🎯 大盘复盘\n\n{review_report}"
|
||
|
||
success = notifier.send(report_content)
|
||
if success:
|
||
logger.info("大盘复盘推送成功")
|
||
else:
|
||
logger.warning("大盘复盘推送失败")
|
||
|
||
return review_report
|
||
|
||
except Exception as e:
|
||
logger.error(f"大盘复盘分析失败: {e}")
|
||
|
||
return None
|
||
|
||
|
||
def run_full_analysis(
|
||
config: Config,
|
||
args: argparse.Namespace,
|
||
stock_codes: Optional[List[str]] = None
|
||
):
|
||
"""
|
||
执行完整的分析流程（个股 + 大盘复盘）
|
||
|
||
这是定时任务调用的主函数
|
||
"""
|
||
try:
|
||
# 命令行参数 --single-notify 覆盖配置（#55）
|
||
if getattr(args, 'single_notify', False):
|
||
config.single_stock_notify = True
|
||
|
||
# 创建调度器
|
||
pipeline = StockAnalysisPipeline(
|
||
config=config,
|
||
max_workers=args.workers
|
||
)
|
||
|
||
# 1. 运行个股分析
|
||
results = pipeline.run(
|
||
stock_codes=stock_codes,
|
||
dry_run=args.dry_run,
|
||
send_notification=not args.no_notify
|
||
)
|
||
|
||
# Issue #128: 分析间隔 - 在个股分析和大盘分析之间添加延迟
|
||
analysis_delay = getattr(config, 'analysis_delay', 0)
|
||
if analysis_delay > 0 and config.market_review_enabled and not args.no_market_review:
|
||
logger.info(f"等待 {analysis_delay} 秒后执行大盘复盘（避免API限流）...")
|
||
time.sleep(analysis_delay)
|
||
|
||
# 2. 运行大盘复盘（如果启用且不是仅个股模式）
|
||
market_report = ""
|
||
if config.market_review_enabled and not args.no_market_review:
|
||
# 只调用一次，并获取结果
|
||
review_result = run_market_review(
|
||
notifier=pipeline.notifier,
|
||
analyzer=pipeline.analyzer,
|
||
search_service=pipeline.search_service
|
||
)
|
||
# 如果有结果，赋值给 market_report 用于后续飞书文档生成
|
||
if review_result:
|
||
market_report = review_result
|
||
|
||
# 输出摘要
|
||
if results:
|
||
logger.info("\n===== 分析结果摘要 =====")
|
||
for r in sorted(results, key=lambda x: x.sentiment_score, reverse=True):
|
||
emoji = r.get_emoji()
|
||
logger.info(
|
||
f"{emoji} {r.name}({r.code}): {r.operation_advice} | "
|
||
f"评分 {r.sentiment_score} | {r.trend_prediction}"
|
||
)
|
||
|
||
logger.info("\n任务执行完成")
|
||
|
||
# === 新增：生成飞书云文档 ===
|
||
try:
|
||
feishu_doc = FeishuDocManager()
|
||
if feishu_doc.is_configured() and (results or market_report):
|
||
logger.info("正在创建飞书云文档...")
|
||
|
||
# 1. 准备标题 "01-01 13:01大盘复盘"
|
||
tz_cn = timezone(timedelta(hours=8))
|
||
now = datetime.now(tz_cn)
|
||
doc_title = f"{now.strftime('%Y-%m-%d %H:%M')} 大盘复盘"
|
||
|
||
# 2. 准备内容 (拼接个股分析和大盘复盘)
|
||
full_content = ""
|
||
|
||
# 添加大盘复盘内容（如果有）
|
||
if market_report:
|
||
full_content += f"# 📈 大盘复盘\n\n{market_report}\n\n---\n\n"
|
||
|
||
# 添加个股决策仪表盘（使用 NotificationService 生成）
|
||
if results:
|
||
dashboard_content = pipeline.notifier.generate_dashboard_report(results)
|
||
full_content += f"# 🚀 个股决策仪表盘\n\n{dashboard_content}"
|
||
|
||
# 3. 创建文档
|
||
doc_url = feishu_doc.create_daily_doc(doc_title, full_content)
|
||
if doc_url:
|
||
logger.info(f"飞书云文档创建成功: {doc_url}")
|
||
# 可选：将文档链接也推送到群里
|
||
pipeline.notifier.send(f"[{now.strftime('%Y-%m-%d %H:%M')}] 复盘文档创建成功: {doc_url}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"飞书文档生成失败: {e}")
|
||
|
||
except Exception as e:
|
||
logger.exception(f"分析流程执行失败: {e}")
|
||
|
||
|
||
def start_bot_stream_clients(config: Config) -> None:
|
||
"""Start bot stream clients when enabled in config."""
|
||
# 启动钉钉 Stream 客户端
|
||
if config.dingtalk_stream_enabled:
|
||
try:
|
||
from bot.platforms import start_dingtalk_stream_background, DINGTALK_STREAM_AVAILABLE
|
||
if DINGTALK_STREAM_AVAILABLE:
|
||
if start_dingtalk_stream_background():
|
||
logger.info("[Main] Dingtalk Stream client started in background.")
|
||
else:
|
||
logger.warning("[Main] Dingtalk Stream client failed to start.")
|
||
else:
|
||
logger.warning("[Main] Dingtalk Stream enabled but SDK is missing.")
|
||
logger.warning("[Main] Run: pip install dingtalk-stream")
|
||
except Exception as exc:
|
||
logger.error(f"[Main] Failed to start Dingtalk Stream client: {exc}")
|
||
|
||
# 启动飞书 Stream 客户端
|
||
if getattr(config, 'feishu_stream_enabled', False):
|
||
try:
|
||
from bot.platforms import start_feishu_stream_background, FEISHU_SDK_AVAILABLE
|
||
if FEISHU_SDK_AVAILABLE:
|
||
if start_feishu_stream_background():
|
||
logger.info("[Main] Feishu Stream client started in background.")
|
||
else:
|
||
logger.warning("[Main] Feishu Stream client failed to start.")
|
||
else:
|
||
logger.warning("[Main] Feishu Stream enabled but SDK is missing.")
|
||
logger.warning("[Main] Run: pip install lark-oapi")
|
||
except Exception as exc:
|
||
logger.error(f"[Main] Failed to start Feishu Stream client: {exc}")
|
||
|
||
|
||
def main() -> int:
|
||
"""
|
||
主入口函数
|
||
|
||
Returns:
|
||
退出码（0 表示成功）
|
||
"""
|
||
# 解析命令行参数
|
||
args = parse_arguments()
|
||
|
||
# 加载配置（在设置日志前加载，以获取日志目录）
|
||
config = get_config()
|
||
|
||
# 配置日志（输出到控制台和文件）
|
||
setup_logging(debug=args.debug, log_dir=config.log_dir)
|
||
|
||
logger.info("=" * 60)
|
||
logger.info("A股自选股智能分析系统 启动")
|
||
logger.info(f"运行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
logger.info("=" * 60)
|
||
|
||
# 验证配置
|
||
warnings = config.validate()
|
||
for warning in warnings:
|
||
logger.warning(warning)
|
||
|
||
# 解析股票列表
|
||
stock_codes = None
|
||
if args.stocks:
|
||
stock_codes = [code.strip() for code in args.stocks.split(',') if code.strip()]
|
||
logger.info(f"使用命令行指定的股票列表: {stock_codes}")
|
||
|
||
# === 启动 WebUI (如果启用) ===
|
||
# 优先级: 命令行参数 > 配置文件
|
||
start_webui = (args.webui or args.webui_only or config.webui_enabled) and os.getenv("GITHUB_ACTIONS") != "true"
|
||
|
||
if start_webui:
|
||
try:
|
||
from webui import run_server_in_thread
|
||
run_server_in_thread(host=config.webui_host, port=config.webui_port)
|
||
start_bot_stream_clients(config)
|
||
except Exception as e:
|
||
logger.error(f"启动 WebUI 失败: {e}")
|
||
|
||
# === 仅 WebUI 模式：不自动执行分析 ===
|
||
if args.webui_only:
|
||
logger.info("模式: 仅 WebUI 服务")
|
||
logger.info(f"WebUI 运行中: http://{config.webui_host}:{config.webui_port}")
|
||
logger.info("通过 /analysis?code=xxx 接口手动触发分析")
|
||
logger.info("按 Ctrl+C 退出...")
|
||
try:
|
||
while True:
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
logger.info("\n用户中断，程序退出")
|
||
return 0
|
||
|
||
try:
|
||
# 模式1: 仅大盘复盘
|
||
if args.market_review:
|
||
logger.info("模式: 仅大盘复盘")
|
||
notifier = NotificationService()
|
||
|
||
# 初始化搜索服务和分析器（如果有配置）
|
||
search_service = None
|
||
analyzer = None
|
||
|
||
if config.bocha_api_keys or config.tavily_api_keys or config.serpapi_keys:
|
||
search_service = SearchService(
|
||
bocha_keys=config.bocha_api_keys,
|
||
tavily_keys=config.tavily_api_keys,
|
||
serpapi_keys=config.serpapi_keys
|
||
)
|
||
|
||
if config.gemini_api_key:
|
||
analyzer = GeminiAnalyzer(api_key=config.gemini_api_key)
|
||
|
||
run_market_review(notifier, analyzer, search_service)
|
||
return 0
|
||
|
||
# 模式2: 定时任务模式
|
||
if args.schedule or config.schedule_enabled:
|
||
logger.info("模式: 定时任务")
|
||
logger.info(f"每日执行时间: {config.schedule_time}")
|
||
|
||
from scheduler import run_with_schedule
|
||
|
||
def scheduled_task():
|
||
run_full_analysis(config, args, stock_codes)
|
||
|
||
run_with_schedule(
|
||
task=scheduled_task,
|
||
schedule_time=config.schedule_time,
|
||
run_immediately=True # 启动时先执行一次
|
||
)
|
||
return 0
|
||
|
||
# 模式3: 正常单次运行
|
||
run_full_analysis(config, args, stock_codes)
|
||
|
||
logger.info("\n程序执行完成")
|
||
|
||
# 如果启用了 WebUI 且是非定时任务模式，保持程序运行以便访问 WebUI
|
||
if start_webui and not (args.schedule or config.schedule_enabled):
|
||
logger.info("WebUI 运行中 (按 Ctrl+C 退出)...")
|
||
try:
|
||
# 简单的保持活跃循环
|
||
while True:
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
pass
|
||
|
||
return 0
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info("\n用户中断，程序退出")
|
||
return 130
|
||
|
||
except Exception as e:
|
||
logger.exception(f"程序执行失败: {e}")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|