总数: {total} 只
+成功: {succeeded} 只
+失败: {failed} 只
+耗时: {elapsed_time:.2f} 秒
+ """ + + # 添加监测同步结果 + if sync_result: + html_body += f""" +新增监测: {sync_result.get('added', 0)} 只
+更新监测: {sync_result.get('updated', 0)} 只
+同步失败: {sync_result.get('failed', 0)} 只
+ """ + + html_body += """ +进场区间: {entry_range}
+止盈位: {take_profit} | 止损位: {stop_loss}
+错误: {error}
+...还有 {len(results) - 10} 只股票未显示
" + + html_body += """ + + + """ + + # 构建纯文本版本 + text_body = f""" +持仓定时分析完成 + +分析概况: +- 总数: {total} 只 +- 成功: {succeeded} 只 +- 失败: {failed} 只 +- 耗时: {elapsed_time:.2f} 秒 +""" + + if sync_result: + text_body += f""" +监测同步结果: +- 新增监测: {sync_result.get('added', 0)} 只 +- 更新监测: {sync_result.get('updated', 0)} 只 +- 同步失败: {sync_result.get('failed', 0)} 只 +""" + + text_body += "\n分析结果详情:\n" + for item in results[:10]: + code = item.get("code", "") + result = item.get("result", {}) + + if result.get("success"): + final_decision = result.get("final_decision", {}) + stock_info = result.get("stock_info", {}) + # 使用正确的字段名 + rating = final_decision.get("rating", "未知") + text_body += f"- {code} {stock_info.get('name', '')}: {rating}\n" + else: + error = result.get("error", "未知错误") + text_body += f"- {code}: 分析失败 ({error})\n" + + success = False + + # 发送邮件 + if self.config['email_enabled']: + email_success = self._send_custom_email(subject, html_body, text_body) + if email_success: + success = True + print("[OK] 邮件通知发送成功") + + # 发送Webhook + if self.config['webhook_enabled']: + webhook_success = self._send_portfolio_webhook(analysis_results, sync_result) + if webhook_success: + success = True + print("[OK] Webhook通知发送成功") + + return success + + except Exception as e: + print(f"[ERROR] 发送持仓分析通知失败: {str(e)}") + import traceback + traceback.print_exc() + return False + + def _send_custom_email(self, subject: str, html_body: str, text_body: str) -> bool: + """发送自定义邮件""" + try: + msg = MIMEMultipart('alternative') + msg['From'] = self.config['email_from'] + msg['To'] = self.config['email_to'] + msg['Subject'] = subject + + part1 = MIMEText(text_body, 'plain', 'utf-8') + part2 = MIMEText(html_body, 'html', 'utf-8') + + msg.attach(part1) + msg.attach(part2) + + with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server: + server.starttls() + server.login(self.config['email_from'], self.config['email_password']) + server.send_message(msg) + + return True + + except Exception as e: + print(f"[ERROR] 邮件发送失败: {str(e)}") + return False + + def _send_portfolio_webhook(self, analysis_results: dict, sync_result: dict = None) -> bool: + """发送持仓分析Webhook通知""" + try: + import requests + + total = analysis_results.get("total", 0) + succeeded = len([r for r in analysis_results.get("results", []) if r.get("result", {}).get("success")]) + failed = total - succeeded + elapsed_time = analysis_results.get("elapsed_time", 0) + + # 构建Markdown消息 + content = f"### 持仓定时分析完成\\n\\n" + content += f"**分析概况**\\n" + content += f"- 总数: {total} 只\\n" + content += f"- 成功: {succeeded} 只\\n" + content += f"- 失败: {failed} 只\\n" + content += f"- 耗时: {elapsed_time:.2f} 秒\\n\\n" + + if sync_result: + content += f"**监测同步**\\n" + content += f"- 新增: {sync_result.get('added', 0)} 只\\n" + content += f"- 更新: {sync_result.get('updated', 0)} 只\\n\\n" + + # 根据webhook类型构建请求 + if self.config['webhook_type'] == 'dingtalk': + data = { + "msgtype": "markdown", + "markdown": { + "title": f"{self.config['webhook_keyword']}", + "text": f"{self.config['webhook_keyword']}\\n\\n{content}" + } + } + else: # feishu + data = { + "msg_type": "text", + "content": { + "text": content + } + } + + response = requests.post(self.config['webhook_url'], json=data, timeout=10) + return response.status_code == 200 + + except Exception as e: + print(f"[ERROR] Webhook发送失败: {str(e)}") + return False # 全局通知服务实例 notification_service = NotificationService() \ No newline at end of file diff --git a/portfolio_db.py b/portfolio_db.py new file mode 100644 index 0000000..6b0f595 --- /dev/null +++ b/portfolio_db.py @@ -0,0 +1,625 @@ +""" +持仓股票数据库管理模块 + +提供持仓股票和分析历史的数据库操作接口 +""" + +import sqlite3 +from datetime import datetime +from typing import List, Dict, Optional, Tuple +import os + +# 数据库文件路径 +DB_PATH = "portfolio_stocks.db" + + +class PortfolioDB: + """持仓股票数据库管理类""" + + def __init__(self, db_path: str = DB_PATH): + """ + 初始化数据库连接 + + Args: + db_path: 数据库文件路径 + """ + self.db_path = db_path + self._init_database() + + def _get_connection(self) -> sqlite3.Connection: + """获取数据库连接""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row # 使查询结果可以通过列名访问 + return conn + + def _init_database(self): + """初始化数据库表结构""" + conn = self._get_connection() + cursor = conn.cursor() + + try: + # 创建持仓股票表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS portfolio_stocks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + cost_price REAL, + quantity INTEGER, + note TEXT, + auto_monitor BOOLEAN DEFAULT 1, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + ''') + + # 创建持仓分析历史表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS portfolio_analysis_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + portfolio_stock_id INTEGER NOT NULL, + analysis_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + rating TEXT, + confidence REAL, + current_price REAL, + target_price REAL, + entry_min REAL, + entry_max REAL, + take_profit REAL, + stop_loss REAL, + summary TEXT, + FOREIGN KEY (portfolio_stock_id) REFERENCES portfolio_stocks(id) ON DELETE CASCADE + ) + ''') + + # 创建索引以提升查询性能 + cursor.execute(''' + CREATE INDEX IF NOT EXISTS idx_portfolio_analysis_stock_id + ON portfolio_analysis_history(portfolio_stock_id) + ''') + + cursor.execute(''' + CREATE INDEX IF NOT EXISTS idx_portfolio_analysis_time + ON portfolio_analysis_history(analysis_time DESC) + ''') + + conn.commit() + print(f"[OK] 数据库初始化成功: {self.db_path}") + + except Exception as e: + print(f"[ERROR] 数据库初始化失败: {e}") + conn.rollback() + raise + finally: + conn.close() + + # ==================== 持仓股票CRUD操作 ==================== + + def add_stock(self, code: str, name: str, cost_price: Optional[float] = None, + quantity: Optional[int] = None, note: str = "", + auto_monitor: bool = True) -> int: + """ + 添加持仓股票 + + Args: + code: 股票代码 + name: 股票名称 + cost_price: 持仓成本价（可选） + quantity: 持仓数量（可选） + note: 备注信息 + auto_monitor: 是否自动同步到监测列表 + + Returns: + 新增股票的ID + + Raises: + sqlite3.IntegrityError: 如果股票代码已存在 + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute(''' + INSERT INTO portfolio_stocks + (code, name, cost_price, quantity, note, auto_monitor, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ''', (code, name, cost_price, quantity, note, auto_monitor, + datetime.now(), datetime.now())) + + conn.commit() + stock_id = cursor.lastrowid + print(f"[OK] 添加持仓股票成功: {code} {name} (ID: {stock_id})") + return stock_id + + except sqlite3.IntegrityError as e: + print(f"[ERROR] 股票代码已存在: {code}") + raise ValueError(f"股票代码 {code} 已存在") from e + except Exception as e: + print(f"[ERROR] 添加持仓股票失败: {e}") + conn.rollback() + raise + finally: + conn.close() + + def update_stock(self, stock_id: int, **kwargs) -> bool: + """ + 更新持仓股票信息 + + Args: + stock_id: 股票ID + **kwargs: 要更新的字段（code, name, cost_price, quantity, note, auto_monitor） + + Returns: + 是否更新成功 + """ + conn = self._get_connection() + cursor = conn.cursor() + + # 允许更新的字段 + allowed_fields = ['code', 'name', 'cost_price', 'quantity', 'note', 'auto_monitor'] + update_fields = {k: v for k, v in kwargs.items() if k in allowed_fields} + + if not update_fields: + print("[WARN] 没有需要更新的字段") + return False + + # 添加更新时间 + update_fields['updated_at'] = datetime.now() + + # 构建SQL语句 + set_clause = ', '.join([f"{field} = ?" for field in update_fields.keys()]) + values = list(update_fields.values()) + [stock_id] + + try: + cursor.execute(f''' + UPDATE portfolio_stocks + SET {set_clause} + WHERE id = ? + ''', values) + + conn.commit() + + if cursor.rowcount > 0: + print(f"[OK] 更新持仓股票成功: ID {stock_id}") + return True + else: + print(f"[WARN] 未找到股票: ID {stock_id}") + return False + + except Exception as e: + print(f"[ERROR] 更新持仓股票失败: {e}") + conn.rollback() + raise + finally: + conn.close() + + def delete_stock(self, stock_id: int) -> bool: + """ + 删除持仓股票（级联删除其所有分析历史） + + Args: + stock_id: 股票ID + + Returns: + 是否删除成功 + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + # 由于设置了ON DELETE CASCADE，删除股票会自动删除其分析历史 + cursor.execute('DELETE FROM portfolio_stocks WHERE id = ?', (stock_id,)) + conn.commit() + + if cursor.rowcount > 0: + print(f"[OK] 删除持仓股票成功: ID {stock_id}") + return True + else: + print(f"[WARN] 未找到股票: ID {stock_id}") + return False + + except Exception as e: + print(f"[ERROR] 删除持仓股票失败: {e}") + conn.rollback() + raise + finally: + conn.close() + + def get_stock(self, stock_id: int) -> Optional[Dict]: + """ + 获取单只持仓股票信息 + + Args: + stock_id: 股票ID + + Returns: + 股票信息字典，不存在则返回None + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute('SELECT * FROM portfolio_stocks WHERE id = ?', (stock_id,)) + row = cursor.fetchone() + + if row: + return dict(row) + return None + + finally: + conn.close() + + def get_stock_by_code(self, code: str) -> Optional[Dict]: + """ + 根据股票代码获取持仓股票信息 + + Args: + code: 股票代码 + + Returns: + 股票信息字典，不存在则返回None + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute('SELECT * FROM portfolio_stocks WHERE code = ?', (code,)) + row = cursor.fetchone() + + if row: + return dict(row) + return None + + finally: + conn.close() + + def get_all_stocks(self, auto_monitor_only: bool = False) -> List[Dict]: + """ + 获取所有持仓股票列表 + + Args: + auto_monitor_only: 是否只返回启用自动监测的股票 + + Returns: + 股票信息字典列表 + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + if auto_monitor_only: + cursor.execute(''' + SELECT * FROM portfolio_stocks + WHERE auto_monitor = 1 + ORDER BY created_at DESC + ''') + else: + cursor.execute('SELECT * FROM portfolio_stocks ORDER BY created_at DESC') + + rows = cursor.fetchall() + return [dict(row) for row in rows] + + finally: + conn.close() + + def search_stocks(self, keyword: str) -> List[Dict]: + """ + 搜索持仓股票（按代码或名称） + + Args: + keyword: 搜索关键词 + + Returns: + 匹配的股票信息字典列表 + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + keyword_pattern = f"%{keyword}%" + cursor.execute(''' + SELECT * FROM portfolio_stocks + WHERE code LIKE ? OR name LIKE ? + ORDER BY created_at DESC + ''', (keyword_pattern, keyword_pattern)) + + rows = cursor.fetchall() + return [dict(row) for row in rows] + + finally: + conn.close() + + def get_stock_count(self) -> int: + """ + 获取持仓股票总数 + + Returns: + 股票数量 + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute('SELECT COUNT(*) as count FROM portfolio_stocks') + result = cursor.fetchone() + return result['count'] + + finally: + conn.close() + + # ==================== 分析历史记录操作 ==================== + + def save_analysis(self, stock_id: int, rating: str, confidence: float, + current_price: float, target_price: Optional[float] = None, + entry_min: Optional[float] = None, entry_max: Optional[float] = None, + take_profit: Optional[float] = None, stop_loss: Optional[float] = None, + summary: str = "") -> int: + """ + 保存分析历史记录 + + Args: + stock_id: 持仓股票ID + rating: 投资评级（买入/持有/卖出） + confidence: 信心度（0-10） + current_price: 当前价格 + target_price: 目标价位 + entry_min: 进场区间最小值 + entry_max: 进场区间最大值 + take_profit: 止盈位 + stop_loss: 止损位 + summary: 分析摘要 + + Returns: + 新增分析记录的ID + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute(''' + INSERT INTO portfolio_analysis_history + (portfolio_stock_id, analysis_time, rating, confidence, current_price, + target_price, entry_min, entry_max, take_profit, stop_loss, summary) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ''', (stock_id, datetime.now(), rating, confidence, current_price, + target_price, entry_min, entry_max, take_profit, stop_loss, summary)) + + conn.commit() + analysis_id = cursor.lastrowid + print(f"[OK] 保存分析历史成功: 股票ID {stock_id}, 评级 {rating}") + return analysis_id + + except Exception as e: + print(f"[ERROR] 保存分析历史失败: {e}") + conn.rollback() + raise + finally: + conn.close() + + def get_analysis_history(self, stock_id: int, limit: int = 10) -> List[Dict]: + """ + 获取股票的分析历史记录 + + Args: + stock_id: 持仓股票ID + limit: 返回记录数量限制 + + Returns: + 分析历史记录列表（按时间倒序） + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute(''' + SELECT * FROM portfolio_analysis_history + WHERE portfolio_stock_id = ? + ORDER BY analysis_time DESC + LIMIT ? + ''', (stock_id, limit)) + + rows = cursor.fetchall() + return [dict(row) for row in rows] + + finally: + conn.close() + + def get_latest_analysis_history(self, stock_id: int, limit: int = 10) -> List[Dict]: + """ + 获取股票的最新分析历史记录（按时间倒序） + + 这是 get_analysis_history 的别名方法，用于保持代码兼容性 + + Args: + stock_id: 持仓股票ID + limit: 返回记录数量限制 + + Returns: + 分析历史记录列表（按时间倒序） + """ + return self.get_analysis_history(stock_id, limit) + + def get_latest_analysis(self, stock_id: int) -> Optional[Dict]: + """ + 获取股票的最新一次分析记录 + + Args: + stock_id: 持仓股票ID + + Returns: + 最新分析记录字典，不存在则返回None + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute(''' + SELECT * FROM portfolio_analysis_history + WHERE portfolio_stock_id = ? + ORDER BY analysis_time DESC + LIMIT 1 + ''', (stock_id,)) + + row = cursor.fetchone() + if row: + return dict(row) + return None + + finally: + conn.close() + + def get_rating_changes(self, stock_id: int, days: int = 30) -> List[Tuple[str, str, str]]: + """ + 获取股票在指定天数内的评级变化 + + Args: + stock_id: 持仓股票ID + days: 查询天数 + + Returns: + 评级变化列表 [(时间, 旧评级, 新评级), ...] + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute(''' + SELECT analysis_time, rating + FROM portfolio_analysis_history + WHERE portfolio_stock_id = ? + AND analysis_time >= datetime('now', '-' || ? || ' days') + ORDER BY analysis_time ASC + ''', (stock_id, days)) + + rows = cursor.fetchall() + + changes = [] + for i in range(1, len(rows)): + prev_rating = rows[i-1]['rating'] + curr_rating = rows[i]['rating'] + if prev_rating != curr_rating: + changes.append(( + rows[i]['analysis_time'], + prev_rating, + curr_rating + )) + + return changes + + finally: + conn.close() + + def delete_old_analysis(self, days: int = 90) -> int: + """ + 删除超过指定天数的分析历史记录 + + Args: + days: 保留天数 + + Returns: + 删除的记录数量 + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute(''' + DELETE FROM portfolio_analysis_history + WHERE analysis_time < datetime('now', '-' || ? || ' days') + ''', (days,)) + + conn.commit() + deleted_count = cursor.rowcount + print(f"[OK] 清理历史分析记录: 删除 {deleted_count} 条记录") + return deleted_count + + except Exception as e: + print(f"[ERROR] 清理历史分析记录失败: {e}") + conn.rollback() + raise + finally: + conn.close() + + def get_all_latest_analysis(self) -> List[Dict]: + """ + 获取所有持仓股票的最新分析记录 + + Returns: + 包含股票信息和最新分析的字典列表 + """ + conn = self._get_connection() + cursor = conn.cursor() + + try: + cursor.execute(''' + SELECT + s.*, + h.rating, h.confidence, h.current_price, h.target_price, + h.entry_min, h.entry_max, h.take_profit, h.stop_loss, + h.analysis_time + FROM portfolio_stocks s + LEFT JOIN ( + SELECT h1.* + FROM portfolio_analysis_history h1 + INNER JOIN ( + SELECT portfolio_stock_id, MAX(analysis_time) as max_time + FROM portfolio_analysis_history + GROUP BY portfolio_stock_id + ) h2 + ON h1.portfolio_stock_id = h2.portfolio_stock_id + AND h1.analysis_time = h2.max_time + ) h ON s.id = h.portfolio_stock_id + ORDER BY s.created_at DESC + ''') + + rows = cursor.fetchall() + return [dict(row) for row in rows] + + finally: + conn.close() + + +# 创建全局数据库实例 +portfolio_db = PortfolioDB() + + +if __name__ == "__main__": + # 测试代码 + print("=" * 50) + print("持仓股票数据库测试") + print("=" * 50) + + # 初始化数据库 + db = PortfolioDB("test_portfolio.db") + + # 测试添加股票 + try: + stock_id = db.add_stock("600519", "贵州茅台", 1650.5, 100, "长期持有") + print(f"\n添加股票ID: {stock_id}") + except ValueError as e: + print(f"\n{e}") + + # 测试查询所有股票 + print("\n所有持仓股票:") + stocks = db.get_all_stocks() + for stock in stocks: + print(f" {stock['code']} {stock['name']}") + + # 测试保存分析历史 + if stocks: + stock_id = stocks[0]['id'] + analysis_id = db.save_analysis( + stock_id, "买入", 8.5, 1700.0, 1850.0, + 1600.0, 1650.0, 1900.0, 1500.0, + "技术面和基本面均良好" + ) + print(f"\n保存分析记录ID: {analysis_id}") + + # 查询分析历史 + print(f"\n股票 {stocks[0]['code']} 的分析历史:") + history = db.get_analysis_history(stock_id) + for h in history: + print(f" {h['analysis_time']}: {h['rating']} (信心度: {h['confidence']})") + + print("\n[OK] 数据库测试完成") + diff --git a/portfolio_manager.py b/portfolio_manager.py new file mode 100644 index 0000000..879ab84 --- /dev/null +++ b/portfolio_manager.py @@ -0,0 +1,526 @@ +""" +持仓管理器模块 + +提供持仓股票管理和批量分析功能 +""" + +import time +from typing import List, Dict, Optional, Tuple +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime + +# 导入必要的模块 +from portfolio_db import portfolio_db + + +class PortfolioManager: + """持仓管理器类""" + + def __init__(self, model="deepseek-chat"): + """ + 初始化持仓管理器 + + Args: + model: AI模型（deepseek-chat 或 deepseek-reasoner） + """ + self.model = model + self.db = portfolio_db + + # ==================== 持仓股票管理 ==================== + + def add_stock(self, code: str, name: str, cost_price: Optional[float] = None, + quantity: Optional[int] = None, note: str = "", + auto_monitor: bool = True) -> Tuple[bool, str, Optional[int]]: + """ + 添加持仓股票 + + Args: + code: 股票代码 + name: 股票名称 + cost_price: 持仓成本价 + quantity: 持仓数量 + note: 备注 + auto_monitor: 是否自动同步到监测 + + Returns: + (成功标志, 消息, 股票ID) + """ + try: + # 验证股票代码格式 + code = code.strip().upper() + if not code: + return False, "股票代码不能为空", None + + # 检查股票代码是否已存在 + existing = self.db.get_stock_by_code(code) + if existing: + return False, f"股票代码 {code} 已存在", None + + # 添加到数据库 + stock_id = self.db.add_stock(code, name, cost_price, quantity, note, auto_monitor) + return True, f"添加持仓股票成功: {code} {name}", stock_id + + except Exception as e: + return False, f"添加失败: {str(e)}", None + + def update_stock(self, stock_id: int, **kwargs) -> Tuple[bool, str]: + """ + 更新持仓股票信息 + + Args: + stock_id: 股票ID + **kwargs: 要更新的字段 + + Returns: + (成功标志, 消息) + """ + try: + success = self.db.update_stock(stock_id, **kwargs) + if success: + return True, "更新成功" + else: + return False, f"未找到股票ID: {stock_id}" + except Exception as e: + return False, f"更新失败: {str(e)}" + + def delete_stock(self, stock_id: int) -> Tuple[bool, str]: + """ + 删除持仓股票（级联删除分析历史） + + Args: + stock_id: 股票ID + + Returns: + (成功标志, 消息) + """ + try: + success = self.db.delete_stock(stock_id) + if success: + return True, "删除成功" + else: + return False, f"未找到股票ID: {stock_id}" + except Exception as e: + return False, f"删除失败: {str(e)}" + + def get_stock(self, stock_id: int) -> Optional[Dict]: + """获取单只持仓股票信息""" + return self.db.get_stock(stock_id) + + def get_all_stocks(self, auto_monitor_only: bool = False) -> List[Dict]: + """获取所有持仓股票列表""" + return self.db.get_all_stocks(auto_monitor_only) + + def search_stocks(self, keyword: str) -> List[Dict]: + """搜索持仓股票""" + return self.db.search_stocks(keyword) + + def get_stock_count(self) -> int: + """获取持仓股票总数""" + return self.db.get_stock_count() + + # ==================== 单只股票分析 ==================== + + def analyze_single_stock(self, stock_code: str, period="1y", + selected_agents: List[str] = None) -> Dict: + """ + 分析单只股票（复用app.py中的分析逻辑） + + Args: + stock_code: 股票代码 + period: 数据周期 + selected_agents: 选中的分析师列表 + + Returns: + 分析结果字典 + """ + print(f"\n{'='*60}") + print(f"开始分析股票: {stock_code}") + print(f"{'='*60}\n") + + try: + # 导入app.py中的分析函数 + from app import analyze_single_stock_for_batch + + # 构建分析师配置 + if selected_agents is None: + enabled_analysts_config = { + 'technical': True, + 'fundamental': True, + 'fund_flow': True, + 'risk': True, + 'sentiment': False, + 'news': False + } + else: + enabled_analysts_config = { + 'technical': 'technical' in selected_agents, + 'fundamental': 'fundamental' in selected_agents, + 'fund_flow': 'fund_flow' in selected_agents, + 'risk': 'risk' in selected_agents, + 'sentiment': 'sentiment' in selected_agents, + 'news': 'news' in selected_agents + } + + # 调用首页的分析函数 + result = analyze_single_stock_for_batch( + symbol=stock_code, + period=period, + enabled_analysts_config=enabled_analysts_config, + selected_model=self.model + ) + + # 检查结果 + if not result.get("success", False): + error_msg = result.get("error", "未知错误") + print(f"\n[ERROR] 分析失败: {error_msg}") + return {"success": False, "error": error_msg} + + print(f"\n{'='*60}") + print(f"分析完成！") + print(f"{'='*60}\n") + + return result + + except Exception as e: + print(f"\n[ERROR] 分析失败: {str(e)}") + import traceback + traceback.print_exc() + return {"success": False, "error": str(e)} + + # ==================== 批量分析 ==================== + + def batch_analyze_sequential(self, stock_codes: List[str], period="1y", + selected_agents: List[str] = None, + progress_callback=None) -> Dict: + """ + 顺序批量分析（逐只分析） + + Args: + stock_codes: 股票代码列表 + period: 数据周期 + selected_agents: 选中的分析师列表 + progress_callback: 进度回调函数 callback(current, total, code, status) + + Returns: + 批量分析结果字典 + """ + print(f"\n{'='*60}") + print(f"开始批量分析 (顺序模式): {len(stock_codes)}只股票") + print(f"{'='*60}\n") + + start_time = time.time() + results = [] + failed = [] + + for i, code in enumerate(stock_codes, 1): + print(f"\n--- 分析进度: {i}/{len(stock_codes)} ---") + + if progress_callback: + progress_callback(i, len(stock_codes), code, "analyzing") + + try: + result = self.analyze_single_stock(code, period, selected_agents) + + if result.get("success"): + results.append({ + "code": code, + "result": result + }) + if progress_callback: + progress_callback(i, len(stock_codes), code, "success") + else: + failed.append({ + "code": code, + "error": result.get("error", "未知错误") + }) + if progress_callback: + progress_callback(i, len(stock_codes), code, "failed") + + except Exception as e: + print(f"[ERROR] 股票 {code} 分析失败: {str(e)}") + failed.append({ + "code": code, + "error": str(e) + }) + if progress_callback: + progress_callback(i, len(stock_codes), code, "error") + + elapsed_time = time.time() - start_time + + print(f"\n{'='*60}") + print(f"批量分析完成！") + print(f"成功: {len(results)}只, 失败: {len(failed)}只, 耗时: {elapsed_time:.1f}秒") + print(f"{'='*60}\n") + + return { + "success": True, + "mode": "sequential", + "total": len(stock_codes), + "succeeded": len(results), + "failed": len(failed), + "results": results, + "failed_stocks": failed, + "elapsed_time": elapsed_time + } + + def batch_analyze_parallel(self, stock_codes: List[str], period="1y", + selected_agents: List[str] = None, + max_workers: int = 3, + progress_callback=None) -> Dict: + """ + 并行批量分析（多线程） + + Args: + stock_codes: 股票代码列表 + period: 数据周期 + selected_agents: 选中的分析师列表 + max_workers: 最大并发数（默认3） + progress_callback: 进度回调函数 + + Returns: + 批量分析结果字典 + """ + print(f"\n{'='*60}") + print(f"开始批量分析 (并行模式): {len(stock_codes)}只股票, 并发数: {max_workers}") + print(f"{'='*60}\n") + + start_time = time.time() + results = [] + failed = [] + completed = 0 + + with ThreadPoolExecutor(max_workers=max_workers) as executor: + # 提交所有任务 + future_to_code = { + executor.submit(self.analyze_single_stock, code, period, selected_agents): code + for code in stock_codes + } + + # 处理完成的任务 + for future in as_completed(future_to_code): + code = future_to_code[future] + completed += 1 + + try: + result = future.result() + + if result.get("success"): + results.append({ + "code": code, + "result": result + }) + print(f"\n[{completed}/{len(stock_codes)}] {code} 分析完成") + if progress_callback: + progress_callback(completed, len(stock_codes), code, "success") + else: + failed.append({ + "code": code, + "error": result.get("error", "未知错误") + }) + print(f"\n[{completed}/{len(stock_codes)}] {code} 分析失败: {result.get('error')}") + if progress_callback: + progress_callback(completed, len(stock_codes), code, "failed") + + except Exception as e: + failed.append({ + "code": code, + "error": str(e) + }) + print(f"\n[{completed}/{len(stock_codes)}] {code} 分析异常: {str(e)}") + if progress_callback: + progress_callback(completed, len(stock_codes), code, "error") + + elapsed_time = time.time() - start_time + + print(f"\n{'='*60}") + print(f"批量分析完成！") + print(f"成功: {len(results)}只, 失败: {len(failed)}只, 耗时: {elapsed_time:.1f}秒") + print(f"{'='*60}\n") + + return { + "success": True, + "mode": "parallel", + "total": len(stock_codes), + "succeeded": len(results), + "failed": len(failed), + "results": results, + "failed_stocks": failed, + "elapsed_time": elapsed_time + } + + def batch_analyze_portfolio(self, mode="sequential", period="1y", + selected_agents: List[str] = None, + max_workers: int = 3, + progress_callback=None) -> Dict: + """ + 批量分析所有持仓股票 + + Args: + mode: 分析模式 ("sequential" 或 "parallel") + period: 数据周期 + selected_agents: 选中的分析师列表 + max_workers: 并行模式下的最大并发数（默认3） + progress_callback: 进度回调函数 + + Returns: + 批量分析结果字典 + """ + # 获取所有持仓股票 + stocks = self.get_all_stocks() + + if not stocks: + return { + "success": False, + "error": "没有持仓股票" + } + + stock_codes = [stock['code'] for stock in stocks] + + # 根据模式选择分析方法 + if mode == "parallel": + return self.batch_analyze_parallel(stock_codes, period, selected_agents, max_workers, progress_callback) + else: + return self.batch_analyze_sequential(stock_codes, period, selected_agents, progress_callback) + + # ==================== 分析结果保存 ==================== + + def save_analysis_results(self, analysis_results: Dict) -> List[int]: + """ + 保存批量分析结果到数据库 + + Args: + analysis_results: 批量分析结果字典 + + Returns: + 保存的分析记录ID列表 + """ + saved_ids = [] + + if not analysis_results.get("success"): + print("[WARN] 分析未成功，跳过保存") + return saved_ids + + for item in analysis_results.get("results", []): + code = item.get("code") + result = item.get("result", {}) + + # 获取持仓股票ID + stock = self.db.get_stock_by_code(code) + if not stock: + print(f"[WARN] 未找到持仓股票: {code}，跳过保存") + continue + + stock_id = stock['id'] + + # 提取分析结果关键信息 + final_decision = result.get("final_decision", {}) + stock_info = result.get("stock_info", {}) + + # 使用正确的字段名 + rating = final_decision.get("rating", "持有") + confidence = final_decision.get("confidence_level", 5.0) + current_price = stock_info.get("current_price", 0.0) + target_price_str = final_decision.get("target_price", "") + entry_range = final_decision.get("entry_range", "") + take_profit_str = final_decision.get("take_profit", "") + stop_loss_str = final_decision.get("stop_loss", "") + + # 解析目标价格 + import re + target_price = None + if target_price_str: + try: + numbers = re.findall(r'\d+\.?\d*', str(target_price_str)) + if numbers: + target_price = float(numbers[0]) + except: + pass + + # 解析进场区间 + entry_min, entry_max = None, None + if entry_range and isinstance(entry_range, str) and "-" in entry_range: + try: + parts = entry_range.split("-") + entry_min = float(parts[0].strip()) + entry_max = float(parts[1].strip()) + except: + pass + + # 解析止盈止损 + take_profit, stop_loss = None, None + if take_profit_str: + try: + numbers = re.findall(r'\d+\.?\d*', str(take_profit_str)) + if numbers: + take_profit = float(numbers[0]) + except: + pass + + if stop_loss_str: + try: + numbers = re.findall(r'\d+\.?\d*', str(stop_loss_str)) + if numbers: + stop_loss = float(numbers[0]) + except: + pass + + # 生成摘要（使用advice或summary字段） + summary = final_decision.get("advice", final_decision.get("summary", ""))[:500] # 限制长度 + + try: + # 保存到数据库 + analysis_id = self.db.save_analysis( + stock_id, rating, confidence, current_price, target_price, + entry_min, entry_max, take_profit, stop_loss, summary + ) + saved_ids.append(analysis_id) + + except Exception as e: + print(f"[ERROR] 保存分析结果失败 ({code}): {str(e)}") + + print(f"\n[OK] 保存分析结果: {len(saved_ids)}条记录") + return saved_ids + + # ==================== 分析历史查询 ==================== + + def get_analysis_history(self, stock_id: int, limit: int = 10) -> List[Dict]: + """获取股票分析历史""" + return self.db.get_analysis_history(stock_id, limit) + + def get_latest_analysis(self, stock_id: int) -> Optional[Dict]: + """获取最新一次分析""" + return self.db.get_latest_analysis(stock_id) + + def get_all_latest_analysis(self) -> List[Dict]: + """获取所有持仓股票的最新分析""" + return self.db.get_all_latest_analysis() + + def get_rating_changes(self, stock_id: int, days: int = 30) -> List[Tuple]: + """获取评级变化""" + return self.db.get_rating_changes(stock_id, days) + + +# 创建全局实例 +portfolio_manager = PortfolioManager() + + +if __name__ == "__main__": + # 测试代码 + print("="*60) + print("持仓管理器测试") + print("="*60) + + manager = PortfolioManager() + + # 测试添加持仓 + success, msg, stock_id = manager.add_stock("000001", "平安银行", 12.5, 1000, "测试持仓") + print(f"\n添加持仓: {msg}") + + # 测试获取所有持仓 + stocks = manager.get_all_stocks() + print(f"\n持仓数量: {len(stocks)}") + for stock in stocks: + print(f" {stock['code']} {stock['name']} - 成本:{stock['cost_price']}, 数量:{stock['quantity']}") + + print("\n[OK] 持仓管理器测试完成") + diff --git a/portfolio_scheduler.py b/portfolio_scheduler.py new file mode 100644 index 0000000..af07b15 --- /dev/null +++ b/portfolio_scheduler.py @@ -0,0 +1,715 @@ +""" +持仓定时分析调度器模块 + +提供定时任务调度功能，在设定时间自动执行持仓批量分析 +""" + +import schedule +import threading +import time +from datetime import datetime +from typing import Optional, Callable +import traceback + +from portfolio_manager import portfolio_manager +from notification_service import NotificationService + + +class PortfolioScheduler: + """持仓分析定时调度器""" + + def __init__(self): + """初始化调度器""" + self.schedule_times = ["09:30"] # 支持多个定时时间点 + self.analysis_mode = "sequential" # 默认顺序分析 + self._is_running = False # 使用私有属性 + self.thread = None + self.last_run_time = None + self.next_run_time = None + self.auto_monitor_sync = True # 默认启用自动监测同步 + self.notification_enabled = True # 默认启用通知 + self.selected_agents = None # None表示全部分析师 + self.notification_service = NotificationService() + self.max_workers = 3 # 并行模式的线程数 + + # 兼容旧代码的属性 + @property + def schedule_time(self) -> str: + """获取第一个定时时间（向后兼容）""" + return self.schedule_times[0] if self.schedule_times else "09:30" + + def is_running(self) -> bool: + """ + 检查调度器是否正在运行 + + Returns: + bool: True表示运行中，False表示已停止 + """ + return self._is_running + + def set_schedule_time(self, time_str: str): + """ + 设置定时分析时间（向后兼容，设置为单个时间） + + Args: + time_str: 时间字符串，格式"HH:MM"（如"08:00"） + """ + try: + # 验证时间格式 + datetime.strptime(time_str, "%H:%M") + self.schedule_times = [time_str] + print(f"[OK] 设置定时分析时间: {time_str}") + + # 如果调度器正在运行，重新调度 + if self._is_running: + self._reschedule() + + except ValueError: + print(f"[ERROR] 无效的时间格式: {time_str}，应为 HH:MM") + + def add_schedule_time(self, time_str: str) -> bool: + """ + 添加一个定时分析时间点 + + Args: + time_str: 时间字符串，格式"HH:MM" + + Returns: + 是否添加成功 + """ + try: + # 验证时间格式 + datetime.strptime(time_str, "%H:%M") + + # 检查是否已存在 + if time_str in self.schedule_times: + print(f"[WARN] 定时时间 {time_str} 已存在") + return False + + self.schedule_times.append(time_str) + self.schedule_times.sort() # 保持时间顺序 + print(f"[OK] 添加定时时间: {time_str}") + + # 如果调度器正在运行，重新调度 + if self._is_running: + self._reschedule() + + return True + + except ValueError: + print(f"[ERROR] 无效的时间格式: {time_str}，应为 HH:MM") + return False + + def remove_schedule_time(self, time_str: str) -> bool: + """ + 删除一个定时分析时间点 + + Args: + time_str: 时间字符串 + + Returns: + 是否删除成功 + """ + if time_str in self.schedule_times: + self.schedule_times.remove(time_str) + print(f"[OK] 删除定时时间: {time_str}") + + # 如果调度器正在运行，重新调度 + if self._is_running: + self._reschedule() + + return True + else: + print(f"[WARN] 定时时间 {time_str} 不存在") + return False + + def get_schedule_times(self) -> list: + """ + 获取所有定时分析时间点 + + Returns: + 时间列表 + """ + return self.schedule_times.copy() + + def set_schedule_times(self, times: list): + """ + 批量设置定时分析时间点 + + Args: + times: 时间字符串列表 + """ + valid_times = [] + for time_str in times: + try: + datetime.strptime(time_str, "%H:%M") + valid_times.append(time_str) + except ValueError: + print(f"[WARN] 跳过无效时间: {time_str}") + + if valid_times: + self.schedule_times = sorted(valid_times) + print(f"[OK] 设置定时时间: {', '.join(self.schedule_times)}") + + # 如果调度器正在运行，重新调度 + if self._is_running: + self._reschedule() + else: + print(f"[ERROR] 没有有效的时间配置") + + def set_analysis_mode(self, mode: str): + """ + 设置分析模式 + + Args: + mode: "sequential" 或 "parallel" + """ + if mode in ["sequential", "parallel"]: + self.analysis_mode = mode + print(f"[OK] 设置分析模式: {mode}") + else: + print(f"[ERROR] 无效的分析模式: {mode}") + + def set_auto_monitor_sync(self, enabled: bool): + """设置是否启用自动监测同步""" + self.auto_monitor_sync = enabled + print(f"[OK] 自动监测同步: {'启用' if enabled else '禁用'}") + + def set_notification_enabled(self, enabled: bool): + """设置是否启用通知""" + self.notification_enabled = enabled + print(f"[OK] 通知推送: {'启用' if enabled else '禁用'}") + + def set_selected_agents(self, agents: Optional[list]): + """设置参与分析的AI分析师""" + self.selected_agents = agents + if agents: + print(f"[OK] 选择分析师: {', '.join(agents)}") + else: + print("[OK] 选择分析师: 全部") + + def _scheduled_job(self): + """定时任务执行的作业""" + print("\n" + "="*60) + print(f"定时分析开始: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print("="*60 + "\n") + + try: + # 1. 执行批量分析 + print("[1/4] 执行持仓批量分析...") + analysis_results = portfolio_manager.batch_analyze_portfolio( + mode=self.analysis_mode, + max_workers=self.max_workers, + selected_agents=self.selected_agents + ) + + if not analysis_results.get("success"): + error_msg = analysis_results.get("error", "未知错误") + print(f"[ERROR] 批量分析失败: {error_msg}") + + # 发送错误通知 + if self.notification_enabled: + self._send_error_notification(error_msg) + + self.last_run_time = datetime.now() + return + + # 2. 保存分析结果 + print("\n[2/4] 保存分析结果...") + saved_ids = portfolio_manager.save_analysis_results(analysis_results) + print(f"[OK] 保存 {len(saved_ids)} 条分析记录") + + # 3. 自动监测同步 + sync_result = None + if self.auto_monitor_sync: + print("\n[3/4] 自动同步到监测列表...") + sync_result = self._sync_to_monitor(analysis_results) + else: + print("\n[3/4] 跳过监测同步（已禁用）") + + # 4. 发送通知 + if self.notification_enabled: + print("\n[4/4] 发送通知...") + self._send_notification(analysis_results, sync_result) + else: + print("\n[4/4] 跳过通知发送（已禁用）") + + # 更新运行时间 + self.last_run_time = datetime.now() + + print("\n" + "="*60) + print(f"定时分析完成: {self.last_run_time.strftime('%Y-%m-%d %H:%M:%S')}") + print("="*60 + "\n") + + except Exception as e: + print(f"\n[ERROR] 定时任务执行异常: {str(e)}") + traceback.print_exc() + + # 发送错误通知 + if self.notification_enabled: + self._send_error_notification(str(e)) + + self.last_run_time = datetime.now() + + def _sync_to_monitor(self, analysis_results: dict) -> dict: + """ + 同步分析结果到监测列表 + + Args: + analysis_results: 批量分析结果 + + Returns: + 同步结果统计 + """ + try: + from monitor_db import monitor_db + + # 准备批量监测数据 + monitors_data = [] + failed_count = 0 + + for item in analysis_results.get("results", []): + code = item.get("code") + result = item.get("result", {}) + + # 检查分析是否成功 + if not result.get("success"): + continue + + final_decision = result.get("final_decision", {}) + stock_info = result.get("stock_info", {}) + + # 检查是否启用自动监测 + stock = portfolio_manager.db.get_stock_by_code(code) + if not stock or not stock.get("auto_monitor"): + continue + + # 从final_decision中提取数据（使用正确的字段名） + rating = final_decision.get("rating", "持有") + entry_range = final_decision.get("entry_range", "") + take_profit_str = final_decision.get("take_profit", "") + stop_loss_str = final_decision.get("stop_loss", "") + + # 解析进场区间（格式如"10.5-12.3"） + entry_min, entry_max = None, None + if entry_range and isinstance(entry_range, str) and "-" in entry_range: + try: + parts = entry_range.split("-") + entry_min = float(parts[0].strip()) + entry_max = float(parts[1].strip()) + except: + pass + + # 解析止盈止损（提取数字） + import re + take_profit, stop_loss = None, None + if take_profit_str: + try: + numbers = re.findall(r'\d+\.?\d*', str(take_profit_str)) + if numbers: + take_profit = float(numbers[0]) + except: + pass + + if stop_loss_str: + try: + numbers = re.findall(r'\d+\.?\d*', str(stop_loss_str)) + if numbers: + stop_loss = float(numbers[0]) + except: + pass + + # 检查参数有效性 + if not all([entry_min, entry_max, take_profit, stop_loss]): + print(f"[WARN] {code} 参数不完整，跳过同步") + failed_count += 1 + continue + + # 构建监测数据 + monitor_data = { + "code": code, + "name": stock_info.get("name", stock.get("name", code)), + "rating": rating, + "entry_min": entry_min, + "entry_max": entry_max, + "take_profit": take_profit, + "stop_loss": stop_loss, + "check_interval": 60, + "notification_enabled": True + } + + monitors_data.append(monitor_data) + + # 使用批量API同步 + if monitors_data: + result = monitor_db.batch_add_or_update_monitors(monitors_data) + return result + else: + print("[WARN] 没有需要同步的监测数据") + return {"added": 0, "updated": 0, "failed": 0, "total": 0} + + except Exception as e: + print(f"[ERROR] 监测同步异常: {str(e)}") + import traceback + traceback.print_exc() + return {"added": 0, "updated": 0, "failed": 0, "total": 0} + + def _send_notification(self, analysis_results: dict, sync_result: Optional[dict]): + """ + 发送分析完成通知（使用新的notification_service方法） + + Args: + analysis_results: 批量分析结果 + sync_result: 监测同步结果 + """ + try: + from notification_service import notification_service + + # 使用新的专用通知方法 + success = notification_service.send_portfolio_analysis_notification( + analysis_results, sync_result + ) + + if success: + print("[OK] 持仓分析通知发送成功") + else: + print("[WARN] 持仓分析通知发送失败（可能未配置通知服务）") + + except Exception as e: + print(f"[ERROR] 发送通知失败: {str(e)}") + import traceback + traceback.print_exc() + + def _send_error_notification(self, error_msg: str): + """发送错误通知""" + try: + content = f""" +持仓定时分析执行失败 + +错误信息： +{error_msg} + +时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} + +请检查系统日志或手动运行分析。 +""" + + if self.notification_service.email_enabled: + self.notification_service.send_email("【持仓定时分析】执行失败", content) + + if self.notification_service.webhook_enabled: + self.notification_service.send_webhook("【持仓定时分析】执行失败", content) + + except Exception as e: + print(f"[ERROR] 发送错误通知失败: {str(e)}") + + def _generate_notification_content(self, analysis_results: dict, + sync_result: Optional[dict]) -> str: + """ + 生成通知内容 + + Args: + analysis_results: 批量分析结果 + sync_result: 监测同步结果 + + Returns: + 通知内容文本 + """ + total = analysis_results.get("total", 0) + succeeded = analysis_results.get("succeeded", 0) + failed = analysis_results.get("failed", 0) + mode = analysis_results.get("mode", "sequential") + elapsed_time = analysis_results.get("elapsed_time", 0) + + # 统计评级分布 + rating_stats = {"买入": 0, "持有": 0, "卖出": 0} + rating_changes = [] + + for item in analysis_results.get("results", []): + code = item.get("code") + result = item.get("result", {}) + final_decision = result.get("final_decision", {}) + rating = final_decision.get("investment_rating", "持有") + + rating_stats[rating] = rating_stats.get(rating, 0) + 1 + + # 检查评级变化 + stock = portfolio_manager.db.get_stock_by_code(code) + if stock: + history = portfolio_manager.db.get_analysis_history(stock['id'], limit=2) + if len(history) >= 2: + old_rating = history[1]['rating'] + new_rating = history[0]['rating'] + if old_rating != new_rating: + stock_info = result.get("stock_info", {}) + name = stock_info.get("name", stock.get("name", code)) + rating_changes.append(f"{code} {name}: {old_rating} → {new_rating}") + + # 构建通知内容 + content = f""" +持仓定时分析报告 - {datetime.now().strftime('%Y-%m-%d %H:%M')} + +📊 分析完成：{total}只持仓股票 +✅ 成功：{succeeded}只 +❌ 失败：{failed}只 +⏱ 耗时：{elapsed_time:.1f}秒 +🔄 模式：{'顺序分析' if mode == 'sequential' else '并行分析'} + +📈 投资评级分布： +• 买入：{rating_stats.get('买入', 0)}只 +• 持有：{rating_stats.get('持有', 0)}只 +• 卖出：{rating_stats.get('卖出', 0)}只 +""" + + # 添加评级变化 + if rating_changes: + content += "\n🔔 评级变化：\n" + for change in rating_changes[:5]: # 最多显示5个 + content += f"• {change}\n" + + # 添加监测同步结果 + if sync_result: + content += f""" +🎯 监测同步： +• 新增：{sync_result.get('added', 0)}只 +• 更新：{sync_result.get('updated', 0)}只 +• 失败：{sync_result.get('failed', 0)}只 +""" + + # 添加失败股票 + if failed > 0: + failed_stocks = analysis_results.get("failed_stocks", []) + content += "\n⚠️ 失败股票：\n" + for stock in failed_stocks[:3]: # 最多显示3个 + content += f"• {stock.get('code')}: {stock.get('error')}\n" + + content += "\n详细报告请登录系统查看。" + + return content + + def _reschedule(self): + """重新调度任务（支持多个时间点）""" + schedule.clear() + for time_str in self.schedule_times: + schedule.every().day.at(time_str).do(self._scheduled_job) + self._update_next_run_time() + print(f"[OK] 重新调度任务: 每天 {', '.join(self.schedule_times)}") + + def _update_next_run_time(self): + """更新下次运行时间""" + jobs = schedule.jobs + if jobs: + self.next_run_time = jobs[0].next_run + else: + self.next_run_time = None + + def _run_schedule_loop(self): + """调度循环（在后台线程中运行）""" + print("[OK] 定时调度器线程启动") + + while self._is_running: + schedule.run_pending() + self._update_next_run_time() + time.sleep(1) + + print("[OK] 定时调度器线程停止") + + def start(self) -> bool: + """ + 启动定时任务 + + Returns: + 是否启动成功 + """ + if self._is_running: + print("[WARN] 定时任务已在运行中") + return False + + # 检查持仓数量 + stock_count = portfolio_manager.get_stock_count() + if stock_count == 0: + print("[ERROR] 没有持仓股票，无法启动定时任务") + return False + + # 检查时间配置 + if not self.schedule_times: + print("[ERROR] 没有配置定时时间") + return False + + # 调度任务（为每个时间点创建任务） + schedule.clear() + for time_str in self.schedule_times: + schedule.every().day.at(time_str).do(self._scheduled_job) + print(f"[OK] 添加调度任务: 每天 {time_str}") + + self._update_next_run_time() + + # 启动后台线程 + self._is_running = True + self.thread = threading.Thread(target=self._run_schedule_loop, daemon=True) + self.thread.start() + + print(f"\n[OK] 定时任务已启动") + print(f" 调度时间: {', '.join(self.schedule_times)}") + print(f" 分析模式: {self.analysis_mode}") + print(f" 持仓数量: {stock_count}只") + if self.next_run_time: + print(f" 下次运行: {self.next_run_time.strftime('%Y-%m-%d %H:%M:%S')}") + + return True + + def stop(self) -> bool: + """ + 停止定时任务 + + Returns: + 是否停止成功 + """ + if not self._is_running: + print("[WARN] 定时任务未运行") + return False + + self._is_running = False + schedule.clear() + + # 等待线程结束（最多等待2秒） + if self.thread and self.thread.is_alive(): + self.thread.join(timeout=2) + + self.thread = None + self.next_run_time = None + + print("[OK] 定时任务已停止") + return True + + def run_once(self) -> bool: + """ + 立即执行一次分析（不影响定时计划） + + Returns: + 是否执行成功 + """ + # 检查持仓数量 + stock_count = portfolio_manager.get_stock_count() + if stock_count == 0: + print("[ERROR] 没有持仓股票") + return False + + print("[OK] 立即执行持仓分析...") + self._scheduled_job() + return True + + def get_status(self) -> dict: + """ + 获取定时任务状态 + + Returns: + 状态字典 + """ + return { + "is_running": self._is_running, + "schedule_time": self.schedule_time, + "analysis_mode": self.analysis_mode, + "auto_monitor_sync": self.auto_monitor_sync, + "notification_enabled": self.notification_enabled, + "last_run_time": self.last_run_time.strftime("%Y-%m-%d %H:%M:%S") if self.last_run_time else None, + "next_run_time": self.next_run_time.strftime("%Y-%m-%d %H:%M:%S") if self.next_run_time else None, + "portfolio_count": portfolio_manager.get_stock_count() + } + + def get_next_run_time(self) -> Optional[str]: + """ + 获取下次运行时间 + + Returns: + 下次运行时间字符串，格式"HH:MM"，如果未设置则返回None + """ + if self.next_run_time: + return self.next_run_time.strftime("%H:%M") + return None + + def update_config(self, schedule_time: str = None, analysis_mode: str = None, + max_workers: int = None, auto_sync_monitor: bool = None, + send_notification: bool = None): + """ + 更新调度器配置 + + Args: + schedule_time: 定时分析时间（格式"HH:MM"，可选，用于向后兼容） + analysis_mode: 分析模式（"sequential"或"parallel"） + max_workers: 并行线程数（仅在parallel模式下有效） + auto_sync_monitor: 是否自动同步到监测 + send_notification: 是否发送通知 + """ + if schedule_time is not None: + self.set_schedule_time(schedule_time) + + if analysis_mode is not None: + self.set_analysis_mode(analysis_mode) + + if max_workers is not None: + self.max_workers = max_workers + print(f"[OK] 设置并行线程数: {max_workers}") + + if auto_sync_monitor is not None: + self.set_auto_monitor_sync(auto_sync_monitor) + + if send_notification is not None: + self.set_notification_enabled(send_notification) + + print("[OK] 配置已更新") + + def start_scheduler(self) -> bool: + """ + 启动调度器（UI友好方法名） + + Returns: + 是否启动成功 + """ + return self.start() + + def stop_scheduler(self) -> bool: + """ + 停止调度器（UI友好方法名） + + Returns: + 是否停止成功 + """ + return self.stop() + + def run_analysis_now(self) -> bool: + """ + 立即执行一次分析（UI友好方法名） + + Returns: + 是否执行成功 + """ + return self.run_once() + + +# 创建全局实例 +portfolio_scheduler = PortfolioScheduler() + + +if __name__ == "__main__": + # 测试代码 + print("="*60) + print("持仓定时调度器测试") + print("="*60) + + scheduler = PortfolioScheduler() + + # 设置配置 + scheduler.set_schedule_time("09:00") + scheduler.set_analysis_mode("sequential") + scheduler.set_auto_monitor_sync(True) + scheduler.set_notification_enabled(False) # 测试时禁用通知 + + # 获取状态 + status = scheduler.get_status() + print("\n调度器状态:") + for key, value in status.items(): + print(f" {key}: {value}") + + print("\n[OK] 调度器测试完成") + diff --git a/portfolio_stocks.db b/portfolio_stocks.db new file mode 100644 index 0000000..e7995ca Binary files /dev/null and b/portfolio_stocks.db differ diff --git a/portfolio_ui.py b/portfolio_ui.py new file mode 100644 index 0000000..29eb2ac --- /dev/null +++ b/portfolio_ui.py @@ -0,0 +1,771 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +持仓管理UI模块 +提供持仓股票的增删改查、批量分析、定时任务管理界面 +""" + +import streamlit as st +import pandas as pd +from datetime import datetime +from typing import List, Dict +import time + +from portfolio_manager import portfolio_manager +from portfolio_scheduler import portfolio_scheduler + + +def display_portfolio_manager(): + """显示持仓管理主界面""" + + st.markdown("## 📊 持仓定时分析") + st.markdown("---") + + # 创建标签页 + tab1, tab2, tab3, tab4 = st.tabs([ + "📝 持仓管理", + "🔄 批量分析", + "⏰ 定时任务", + "📈 分析历史" + ]) + + with tab1: + display_portfolio_stocks() + + with tab2: + display_batch_analysis() + + with tab3: + display_scheduler_management() + + with tab4: + display_analysis_history() + + +def display_portfolio_stocks(): + """显示持仓股票列表和管理""" + + st.markdown("### 📝 持仓股票管理") + + # 添加新股票表单 + with st.expander("➕ 添加持仓股票", expanded=False): + display_add_stock_form() + + # 获取所有持仓股票 + stocks = portfolio_manager.get_all_stocks() + + if not stocks: + st.info("暂无持仓股票，请添加股票代码开始管理。") + return + + # 显示统计 + col1, col2, col3 = st.columns(3) + with col1: + st.metric("持仓股票数", len(stocks)) + with col2: + auto_monitor_count = sum(1 for s in stocks if s.get("auto_monitor")) + st.metric("启用自动监测", auto_monitor_count) + with col3: + total_cost = sum( + s.get("cost_price", 0) * s.get("quantity", 0) + for s in stocks + if s.get("cost_price") and s.get("quantity") + ) + st.metric("总持仓成本", f"¥{total_cost:,.2f}") + + st.markdown("---") + + # 显示股票列表（卡片式布局） + for stock in stocks: + display_stock_card(stock) + + +def display_stock_card(stock: Dict): + """显示单个股票卡片""" + + code = stock.get("code", "") + name = stock.get("name", "") + cost_price = stock.get("cost_price") + quantity = stock.get("quantity") + note = stock.get("note", "") + auto_monitor = stock.get("auto_monitor", True) + created_at = stock.get("created_at", "") + + # 创建卡片 + with st.container(): + col1, col2, col3, col4 = st.columns([3, 2, 2, 2]) + + with col1: + st.markdown(f"**{code}** {name}") + if note: + st.caption(f"备注: {note}") + + with col2: + if cost_price and quantity: + st.write(f"成本: ¥{cost_price:.2f}") + st.caption(f"数量: {quantity}股") + else: + st.caption("未设置持仓") + + with col3: + if auto_monitor: + st.success("🔔 自动监测") + else: + st.info("🔕 不监测") + + with col4: + col_edit, col_del = st.columns(2) + with col_edit: + if st.button("✏️", key=f"edit_{code}", help="编辑"): + st.session_state[f"editing_{code}"] = True + st.rerun() + with col_del: + if st.button("🗑️", key=f"del_{code}", help="删除"): + portfolio_manager.delete_stock(code) + st.success(f"已删除 {code}") + time.sleep(0.5) + st.rerun() + + # 编辑表单（如果处于编辑状态） + if st.session_state.get(f"editing_{code}"): + with st.form(key=f"edit_form_{code}"): + st.markdown(f"#### 编辑 {code}") + + col_a, col_b = st.columns(2) + with col_a: + new_cost = st.number_input( + "成本价", + value=cost_price if cost_price else 0.0, + min_value=0.0, + step=0.01 + ) + new_quantity = st.number_input( + "持仓数量", + value=quantity if quantity else 0, + min_value=0, + step=100 + ) + + with col_b: + new_note = st.text_area("备注", value=note, height=80) + new_auto_monitor = st.checkbox("自动同步到监测", value=auto_monitor) + + col_submit, col_cancel = st.columns(2) + with col_submit: + if st.form_submit_button("保存", type="primary"): + portfolio_manager.update_stock( + code, + cost_price=new_cost if new_cost > 0 else None, + quantity=new_quantity if new_quantity > 0 else None, + note=new_note, + auto_monitor=new_auto_monitor + ) + del st.session_state[f"editing_{code}"] + st.success("更新成功！") + time.sleep(0.5) + st.rerun() + + with col_cancel: + if st.form_submit_button("取消"): + del st.session_state[f"editing_{code}"] + st.rerun() + + st.markdown("---") + + +def display_add_stock_form(): + """显示添加股票表单""" + + with st.form(key="add_stock_form"): + col1, col2 = st.columns(2) + + with col1: + code = st.text_input( + "股票代码*", + placeholder="例如: 600519.SH 或 000001.SZ", + help="必填，格式：代码.市场（SH/SZ/HK/US）" + ) + name = st.text_input( + "股票名称", + placeholder="例如: 贵州茅台", + help="可选，留空将自动获取" + ) + + with col2: + cost_price = st.number_input( + "成本价", + min_value=0.0, + step=0.01, + help="可选，用于计算收益" + ) + quantity = st.number_input( + "持仓数量", + min_value=0, + step=100, + help="可选，单位：股" + ) + + note = st.text_area("备注", height=80, placeholder="可选，记录买入理由等信息") + auto_monitor = st.checkbox("分析后自动同步到监测", value=True) + + if st.form_submit_button("➕ 添加股票", type="primary"): + if not code: + st.error("请输入股票代码") + else: + try: + portfolio_manager.add_stock( + code=code.strip().upper(), + name=name.strip() if name else None, + cost_price=cost_price if cost_price > 0 else None, + quantity=quantity if quantity > 0 else None, + note=note.strip() if note else None, + auto_monitor=auto_monitor + ) + st.success(f"✅ 已添加 {code} 到持仓列表") + time.sleep(0.5) + st.rerun() + except Exception as e: + st.error(f"添加失败: {str(e)}") + + +def display_batch_analysis(): + """显示批量分析功能""" + + st.markdown("### 🔄 批量分析持仓股票") + + stocks = portfolio_manager.get_all_stocks() + + if not stocks: + st.warning("暂无持仓股票，请先添加股票。") + return + + # 分析选项 + col1, col2, col3 = st.columns(3) + + with col1: + st.metric("持仓股票数", len(stocks)) + + with col2: + analysis_mode = st.selectbox( + "分析模式", + options=["sequential", "parallel"], + format_func=lambda x: "顺序分析" if x == "sequential" else "并行分析", + help="顺序分析较慢但稳定，并行分析更快但消耗更多资源" + ) + + with col3: + if analysis_mode == "parallel": + max_workers = st.number_input( + "并行线程数", + min_value=2, + max_value=10, + value=3, + help="同时分析的股票数量" + ) + else: + max_workers = 1 + + st.markdown("---") + + # 同步和通知选项 + col_a, col_b = st.columns(2) + + with col_a: + auto_sync = st.checkbox( + "自动同步到监测", + value=True, + help="分析完成后自动将评级结果同步到实时监测列表" + ) + + with col_b: + send_notification = st.checkbox( + "发送完成通知", + value=True, + help="通过邮件或Webhook发送分析完成通知" + ) + + # 立即分析按钮 + if st.button("🚀 立即开始分析", type="primary", use_container_width=True): + with st.spinner("正在批量分析持仓股票..."): + # 显示进度 + progress_bar = st.progress(0) + status_text = st.empty() + + # 执行批量分析 + try: + # 定义进度回调函数 + def update_progress(current, total, code, status): + progress_bar.progress(current / total) + status_map = { + "analyzing": "正在分析", + "success": "✅ 完成", + "failed": "❌ 失败", + "error": "⚠️ 错误" + } + status_text.text(f"{status_map.get(status, '处理中')} {code} ({current}/{total})") + + result = portfolio_manager.batch_analyze_portfolio( + mode=analysis_mode, + max_workers=max_workers, + progress_callback=update_progress + ) + + # 清除进度显示 + progress_bar.empty() + status_text.empty() + + # 显示结果 + st.success(f"✅ 批量分析完成！") + + col_r1, col_r2, col_r3, col_r4 = st.columns(4) + with col_r1: + st.metric("总计", result.get("total", 0)) + with col_r2: + st.metric("成功", result.get("succeeded", 0)) + with col_r3: + st.metric("失败", result.get("failed", 0)) + with col_r4: + st.metric("耗时", f"{result.get('elapsed_time', 0):.1f}秒") + + # 保存分析结果到数据库 + saved_ids = portfolio_manager.save_analysis_results(result) + st.info(f"💾 已保存 {len(saved_ids)} 条分析记录到数据库") + + # 同步到监测 + sync_result = None # 初始化同步结果 + if auto_sync: + with st.spinner("正在同步到监测列表..."): + from monitor_db import monitor_db + + # 准备同步数据 + monitors_to_sync = [] + for item in result.get("results", []): + # 检查分析是否成功 + if not item.get("result", {}).get("success"): + continue + + code = item["code"] + stock = portfolio_manager.db.get_stock_by_code(code) + + # 只同步启用了自动监测的股票 + if not stock or not stock.get("auto_monitor"): + continue + + analysis_result = item["result"] + stock_info = analysis_result.get("stock_info", {}) + final_decision = analysis_result.get("final_decision", {}) + + # 从final_decision中提取数据 + rating = final_decision.get("rating", "持有") + entry_range = final_decision.get("entry_range", "") + take_profit_str = final_decision.get("take_profit", "") + stop_loss_str = final_decision.get("stop_loss", "") + + # 解析进场区间（格式如"10.5-12.3"） + entry_min, entry_max = None, None + if entry_range and isinstance(entry_range, str) and "-" in entry_range: + try: + parts = entry_range.split("-") + entry_min = float(parts[0].strip()) + entry_max = float(parts[1].strip()) + except: + pass + + # 解析止盈止损（提取数字） + import re + take_profit, stop_loss = None, None + if take_profit_str: + try: + numbers = re.findall(r'\d+\.?\d*', str(take_profit_str)) + if numbers: + take_profit = float(numbers[0]) + except: + pass + + if stop_loss_str: + try: + numbers = re.findall(r'\d+\.?\d*', str(stop_loss_str)) + if numbers: + stop_loss = float(numbers[0]) + except: + pass + + # 只有当所有必需字段都有效时才添加 + if entry_min and entry_max and take_profit and stop_loss: + monitors_to_sync.append({ + "code": code, + "name": stock_info.get("name", stock.get("name", "")), + "rating": rating, + "entry_min": entry_min, + "entry_max": entry_max, + "take_profit": take_profit, + "stop_loss": stop_loss + }) + + if monitors_to_sync: + sync_result = monitor_db.batch_add_or_update_monitors(monitors_to_sync) + st.info(f"📊 监测同步: 新增 {sync_result.get('added', 0)} 只, 更新 {sync_result.get('updated', 0)} 只") + else: + sync_result = {"added": 0, "updated": 0, "failed": 0, "total": 0} + st.info("📊 无需同步监测列表（无启用自动监测的股票）") + + # 发送通知 + if send_notification: + from notification_service import notification_service + notification_service.send_portfolio_analysis_notification( + result, + sync_result if auto_sync else None + ) + st.info("✉️ 已发送完成通知") + + # 显示详细结果 + st.markdown("### 分析结果详情") + for item in result.get("results", []): + display_analysis_result_card(item) + + except Exception as e: + st.error(f"批量分析失败: {str(e)}") + import traceback + st.code(traceback.format_exc()) + + +def display_analysis_result_card(item: Dict): + """显示单个分析结果卡片""" + + code = item.get("code", "") + result = item.get("result", {}) + + # 检查分析是否成功 + if result.get("success"): + final_decision = result.get("final_decision", {}) + stock_info = result.get("stock_info", {}) + + # 使用正确的字段名 + rating = final_decision.get("rating", "未知") + confidence = final_decision.get("confidence_level", "N/A") + target_price = final_decision.get("target_price", "N/A") + entry_range = final_decision.get("entry_range", "N/A") + take_profit = final_decision.get("take_profit", "N/A") + stop_loss = final_decision.get("stop_loss", "N/A") + + # 评级颜色 + if "强烈买入" in rating or "买入" in rating: + rating_color = "🟢" + elif "卖出" in rating: + rating_color = "🔴" + else: + rating_color = "🟡" + + with st.expander(f"{rating_color} {code} {stock_info.get('name', '')} - {rating} (信心度: {confidence})"): + col1, col2 = st.columns(2) + + with col1: + st.markdown("**进出场位置**") + st.write(f"进场区间: {entry_range}") + st.write(f"目标价: {target_price}") + + with col2: + st.markdown("**风控位置**") + st.write(f"止盈位: {take_profit}") + st.write(f"止损位: {stop_loss}") + + # 投资建议 + advice = final_decision.get("advice", "") + if advice: + st.markdown("**投资建议**") + st.info(advice) + + else: + # 分析失败 + error = result.get("error", "未知错误") + with st.expander(f"🔴 {code} - 分析失败"): + st.error(f"错误: {error}") + + +def display_scheduler_management(): + """显示定时任务管理""" + + st.markdown("### ⏰ 定时任务管理") + + # 调度器状态 + is_running = portfolio_scheduler.is_running() + schedule_times = portfolio_scheduler.get_schedule_times() + + col1, col2, col3 = st.columns(3) + + with col1: + if is_running: + st.success("🟢 调度器运行中") + else: + st.error("🔴 调度器已停止") + + with col2: + st.info(f"⏰ 定时数量: {len(schedule_times)}个") + + with col3: + next_run = portfolio_scheduler.get_next_run_time() + if next_run: + st.info(f"⏭️ 下次运行: {next_run}") + else: + st.info("⏭️ 下次运行: 未设置") + + st.markdown("---") + + # 显示所有定时时间点 + st.markdown("#### 📋 已配置的定时时间") + + if schedule_times: + cols_per_row = 4 + for i in range(0, len(schedule_times), cols_per_row): + cols = st.columns(cols_per_row) + for j, col in enumerate(cols): + idx = i + j + if idx < len(schedule_times): + time_str = schedule_times[idx] + with col: + col_time, col_del = st.columns([3, 1]) + with col_time: + st.info(f"⏰ {time_str}") + with col_del: + if st.button("🗑️", key=f"del_time_{idx}", help="删除"): + if len(schedule_times) > 1: + portfolio_scheduler.remove_schedule_time(time_str) + st.success(f"已删除 {time_str}") + time.sleep(0.3) + st.rerun() + else: + st.error("至少保留一个定时时间") + else: + st.warning("暂无定时配置") + + # 添加新的定时时间 + with st.expander("➕ 添加定时时间", expanded=False): + col_input, col_add = st.columns([3, 1]) + with col_input: + new_time = st.time_input( + "选择时间", + value=datetime.strptime("15:05", "%H:%M").time(), + help="添加新的每日分析时间" + ) + with col_add: + st.write("") # 占位，对齐按钮 + st.write("") + if st.button("➕ 添加", type="primary", use_container_width=True): + time_str = new_time.strftime("%H:%M") + if portfolio_scheduler.add_schedule_time(time_str): + st.success(f"已添加 {time_str}") + time.sleep(0.3) + st.rerun() + else: + st.warning(f"{time_str} 已存在") + + st.markdown("---") + + # 任务配置 + with st.form(key="scheduler_config_form"): + st.markdown("#### 分析配置") + + col_a, col_b = st.columns(2) + + with col_a: + analysis_mode = st.selectbox( + "分析模式", + options=["sequential", "parallel"], + format_func=lambda x: "顺序分析" if x == "sequential" else "并行分析", + index=0 if portfolio_scheduler.analysis_mode == "sequential" else 1 + ) + + with col_b: + max_workers = st.number_input( + "并行线程数", + min_value=2, + max_value=10, + value=portfolio_scheduler.max_workers, + disabled=(analysis_mode == "sequential"), + help="仅在并行模式下生效" + ) + + auto_sync_monitor = st.checkbox( + "自动同步到监测", + value=portfolio_scheduler.auto_monitor_sync, + help="分析完成后自动将结果同步到实时监测列表" + ) + send_notification = st.checkbox( + "发送完成通知", + value=portfolio_scheduler.notification_enabled, + help="通过邮件或Webhook发送分析结果" + ) + + col_update, col_reset = st.columns(2) + + with col_update: + if st.form_submit_button("💾 更新配置", type="primary"): + portfolio_scheduler.update_config( + analysis_mode=analysis_mode, + max_workers=max_workers if analysis_mode == "parallel" else 1, + auto_sync_monitor=auto_sync_monitor, + send_notification=send_notification + ) + st.success("配置已更新！") + time.sleep(0.5) + st.rerun() + + with col_reset: + if st.form_submit_button("🔄 恢复默认"): + portfolio_scheduler.set_schedule_times(["09:30"]) + portfolio_scheduler.update_config( + analysis_mode="sequential", + max_workers=1, + auto_sync_monitor=True, + send_notification=True + ) + st.success("已恢复默认配置！") + time.sleep(0.5) + st.rerun() + + st.markdown("---") + + # 控制按钮 + col_btn1, col_btn2, col_btn3 = st.columns(3) + + with col_btn1: + if is_running: + if st.button("⏹️ 停止调度器", type="secondary", use_container_width=True): + portfolio_scheduler.stop_scheduler() + st.success("调度器已停止") + time.sleep(0.5) + st.rerun() + else: + if st.button("▶️ 启动调度器", type="primary", use_container_width=True): + portfolio_scheduler.start_scheduler() + st.success("调度器已启动") + time.sleep(0.5) + st.rerun() + + with col_btn2: + if st.button("🚀 立即执行一次", type="primary", use_container_width=True): + with st.spinner("正在执行持仓分析..."): + try: + portfolio_scheduler.run_analysis_now() + st.success("执行完成！请查看分析历史。") + except Exception as e: + st.error(f"执行失败: {str(e)}") + + with col_btn3: + if st.button("🔄 刷新状态", use_container_width=True): + st.rerun() + + +def display_analysis_history(): + """显示分析历史""" + + st.markdown("### 📈 分析历史记录") + + stocks = portfolio_manager.get_all_stocks() + + if not stocks: + st.info("暂无持仓股票") + return + + # 选择股票 + stock_codes = [s["code"] for s in stocks] + selected_code = st.selectbox( + "选择股票", + options=["全部"] + stock_codes, + help="查看特定股票的分析历史" + ) + + # 获取历史记录 + if selected_code == "全部": + # 获取所有股票的最新历史 + all_history = [] + for stock in stocks: + stock_id = stock["id"] + history = portfolio_manager.db.get_latest_analysis_history(stock_id, limit=5) + for h in history: + h["code"] = stock["code"] + h["name"] = stock["name"] + all_history.extend(history) + + # 按时间排序 + all_history.sort(key=lambda x: x.get("analysis_time", ""), reverse=True) + history_list = all_history[:20] # 只显示最近20条 + else: + # 获取指定股票的历史 + stock = next((s for s in stocks if s["code"] == selected_code), None) + if stock: + history_list = portfolio_manager.db.get_latest_analysis_history( + stock["id"], limit=20 + ) + for h in history_list: + h["code"] = stock["code"] + h["name"] = stock["name"] + else: + history_list = [] + + if not history_list: + st.info(f"暂无分析历史记录") + return + + # 显示历史记录 + st.markdown(f"共 {len(history_list)} 条记录") + + for record in history_list: + display_history_record(record) + + +def display_history_record(record: Dict): + """显示单条历史记录""" + + code = record.get("code", "") + name = record.get("name", "") + analysis_time = record.get("analysis_time", "") + rating = record.get("rating", "未知") + confidence = record.get("confidence", 0) + current_price = record.get("current_price") + target_price = record.get("target_price") + entry_min = record.get("entry_min") + entry_max = record.get("entry_max") + take_profit = record.get("take_profit") + stop_loss = record.get("stop_loss") + summary = record.get("summary", "") + + # 评级颜色 + if "强烈买入" in rating or "买入" in rating: + rating_icon = "🟢" + elif "卖出" in rating: + rating_icon = "🔴" + else: + rating_icon = "🟡" + + with st.expander( + f"{rating_icon} {code} {name} - {rating} | {analysis_time}", + expanded=False + ): + col1, col2, col3 = st.columns(3) + + with col1: + st.markdown("**价格信息**") + if current_price: + st.write(f"当时价格: ¥{current_price:.2f}") + if target_price: + st.write(f"目标价: ¥{target_price:.2f}") + + with col2: + st.markdown("**进场区间**") + if entry_min and entry_max: + st.write(f"¥{entry_min:.2f} ~ ¥{entry_max:.2f}") + + with col3: + st.markdown("**风控位置**") + if take_profit: + st.write(f"止盈: ¥{take_profit:.2f}") + if stop_loss: + st.write(f"止损: ¥{stop_loss:.2f}") + + if summary: + st.markdown("**分析摘要**") + st.info(summary) + + st.caption(f"置信度: {confidence}%") + diff --git a/stock_analysis.db b/stock_analysis.db index f2f0f27..c29f662 100644 Binary files a/stock_analysis.db and b/stock_analysis.db differ diff --git a/stock_monitor.db b/stock_monitor.db index ace9f81..5439c65 100644 Binary files a/stock_monitor.db and b/stock_monitor.db differ