增加定时分析功能

This commit is contained in:
oficcejo
2025-10-20 09:18:44 +08:00
parent f7d81a1469
commit 1bddf393e0
13 changed files with 3132 additions and 1 deletions
+3
View File
@@ -7,3 +7,6 @@
/__pycache__
# 环境变量文件
.env
/.cursor
/openspec
/docs
+18
View File
@@ -0,0 +1,18 @@
<!-- OPENSPEC:START -->
# OpenSpec Instructions
These instructions are for AI assistants working in this project.
Always open `@/openspec/AGENTS.md` when the request:
- Mentions planning or proposals (words like proposal, spec, change, plan)
- Introduces new capabilities, breaking changes, architecture shifts, or big performance/security work
- Sounds ambiguous and you need the authoritative spec before coding
Use `@/openspec/AGENTS.md` to learn:
- How to create and apply change proposals
- Spec format and conventions
- Project structure and guidelines
Keep this managed block so 'openspec update' can refresh the instructions.
<!-- OPENSPEC:END -->
+69
View File
@@ -12,6 +12,75 @@
<img width="1910" height="923" alt="image" src="https://github.com/user-attachments/assets/ff80584a-0599-4891-9b8d-47485fa3678a" />
## ✨最新更新 - 持仓定时分析功能 ⭐️
### 📊 持仓定时分析系统全新上线
专为长期投资者和组合管理打造的智能持仓管理与自动化分析系统
#### 核心功能
- **📝 持仓管理**:可视化管理持仓股票清单
- 添加/编辑/删除持仓股票
- 记录成本价、持仓数量、买入备注
- 支持A股、港股、美股全市场
- **🔄 批量分析**:一键分析全部持仓
- 顺序分析模式:稳定可靠,逐个深度分析
- 并行分析模式:速度更快,支持3-10线程
- 实时进度显示,详细结果展示
- **⏰ 定时任务**:全自动定时分析(支持多时间点)⭐️
- **多时间点支持**:可设置多个每日分析时间(如09:25+15:05
- 灵活添加/删除时间点,自动排序
- 后台自动执行,无需人工干预
- 支持立即执行和手动触发
- 调度器状态实时监控
- **🎯 自动监测同步**:智能联动实时监测
- 分析完成自动将评级结果同步到监测列表
- 自动设置进场区间、止盈位、止损位
- 覆盖同名股票,始终保持最新结果
- **📬 通知推送**:及时获取分析报告
- 邮件通知:HTML精美格式,详细结果展示
- Webhook通知:钉钉/飞书群消息推送
- 包含分析概况、同步结果、个股详情
- **📈 分析历史**:完整的历史记录追踪
- 每只股票的历史分析记录
- 评级变化趋势可视化
- 验证预测准确性
- 优化投资决策
#### 使用场景
-**长期投资者**:每日跟踪持仓,及时调整策略
-**组合管理**:批量分析多只股票,优化资产配置
-**自动化交易**:联动监测和量化交易,实现全自动化
-**策略回测**:历史记录对比实际走势,验证有效性
#### 技术实现
- **数据库**SQLite持久化存储(portfolio_stocks, portfolio_analysis_history
- **批量分析**:复用AI智能体团队,支持顺序/并行模式
- **定时调度**:schedule库实现多时间点调度管理
- **监测集成**:批量API无缝对接实时监测系统
- **通知扩展**:专用通知模板,邮件/Webhook双通道
- **统一规范**:强制使用统一的分析函数和字段名,确保代码复用和一致性
#### 快速开始
```bash
# 安装依赖
pip install schedule
# 启动应用
streamlit run app.py
# 点击侧边栏"📊 持仓分析"进入功能
```
#### 详细文档
请查看 [`docs/PORTFOLIO_USAGE.md`](docs/PORTFOLIO_USAGE.md) 获取完整使用指南
---
## ✨1019 更新说明
### 🎯 风险管理师功能增强 ⭐️
+35
View File
@@ -296,6 +296,8 @@ def main():
del st.session_state.show_monitor
if 'show_longhubang' in st.session_state:
del st.session_state.show_longhubang
if 'show_portfolio' in st.session_state:
del st.session_state.show_portfolio
if st.button("📊 实时监测", use_container_width=True, key="nav_monitor"):
st.session_state.show_monitor = True
@@ -305,6 +307,8 @@ def main():
del st.session_state.show_main_force
if 'show_longhubang' in st.session_state:
del st.session_state.show_longhubang
if 'show_portfolio' in st.session_state:
del st.session_state.show_portfolio
if st.button("🎯 主力选股", use_container_width=True, key="nav_main_force"):
st.session_state.show_main_force = True
@@ -318,6 +322,8 @@ def main():
del st.session_state.show_sector_strategy
if 'show_longhubang' in st.session_state:
del st.session_state.show_longhubang
if 'show_portfolio' in st.session_state:
del st.session_state.show_portfolio
if st.button("🎯 智策板块", use_container_width=True, key="nav_sector_strategy"):
st.session_state.show_sector_strategy = True
@@ -331,6 +337,8 @@ def main():
del st.session_state.show_main_force
if 'show_longhubang' in st.session_state:
del st.session_state.show_longhubang
if 'show_portfolio' in st.session_state:
del st.session_state.show_portfolio
if st.button("🎯 智瞰龙虎", use_container_width=True, key="nav_longhubang"):
st.session_state.show_longhubang = True
@@ -344,6 +352,23 @@ def main():
del st.session_state.show_main_force
if 'show_sector_strategy' in st.session_state:
del st.session_state.show_sector_strategy
if 'show_portfolio' in st.session_state:
del st.session_state.show_portfolio
if st.button("📊 持仓分析", use_container_width=True, key="nav_portfolio"):
st.session_state.show_portfolio = True
if 'show_history' in st.session_state:
del st.session_state.show_history
if 'show_monitor' in st.session_state:
del st.session_state.show_monitor
if 'show_config' in st.session_state:
del st.session_state.show_config
if 'show_main_force' in st.session_state:
del st.session_state.show_main_force
if 'show_sector_strategy' in st.session_state:
del st.session_state.show_sector_strategy
if 'show_longhubang' in st.session_state:
del st.session_state.show_longhubang
if st.button("🏠 返回首页", use_container_width=True, key="nav_home"):
if 'show_history' in st.session_state:
@@ -358,6 +383,8 @@ def main():
del st.session_state.show_sector_strategy
if 'show_longhubang' in st.session_state:
del st.session_state.show_longhubang
if 'show_portfolio' in st.session_state:
del st.session_state.show_portfolio
if st.button("⚙️ 环境配置", use_container_width=True, key="nav_config"):
st.session_state.show_config = True
@@ -371,6 +398,8 @@ def main():
del st.session_state.show_sector_strategy
if 'show_longhubang' in st.session_state:
del st.session_state.show_longhubang
if 'show_portfolio' in st.session_state:
del st.session_state.show_portfolio
st.markdown("---")
@@ -471,6 +500,12 @@ def main():
display_longhubang()
return
# 检查是否显示持仓分析
if 'show_portfolio' in st.session_state and st.session_state.show_portfolio:
from portfolio_ui import display_portfolio_manager
display_portfolio_manager()
return
# 检查是否显示环境配置
if 'show_config' in st.session_state and st.session_state.show_config:
display_config_manager()
+131
View File
@@ -391,5 +391,136 @@ class StockMonitorDatabase:
}
return None
def get_monitor_by_code(self, symbol: str) -> Optional[Dict]:
"""
根据股票代码获取监测信息
Args:
symbol: 股票代码
Returns:
监测股票信息字典,不存在则返回None
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT * FROM monitored_stocks WHERE symbol = ?
''', (symbol,))
row = cursor.fetchone()
conn.close()
if row:
entry_range = json.loads(row[4])
quant_config = json.loads(row[12]) if row[12] else None
return {
'id': row[0],
'symbol': row[1],
'name': row[2],
'rating': row[3],
'entry_range': entry_range,
'take_profit': row[5],
'stop_loss': row[6],
'current_price': row[7],
'last_checked': row[8],
'check_interval': row[9],
'notification_enabled': row[10],
'quant_enabled': row[11],
'quant_config': quant_config
}
return None
def batch_add_or_update_monitors(self, monitors_data: List[Dict]) -> Dict[str, int]:
"""
批量添加或更新监测股票
Args:
monitors_data: 监测股票数据列表,每个字典包含:
- code/symbol: 股票代码
- name: 股票名称
- rating: 投资评级
- entry_min, entry_max: 进场区间
- take_profit: 止盈位
- stop_loss: 止损位
- check_interval: 检查间隔(可选,默认60秒)
- notification_enabled: 是否启用通知(可选,默认True)
Returns:
统计字典 {"added": X, "updated": Y, "failed": Z, "total": N}
"""
added = 0
updated = 0
failed = 0
for data in monitors_data:
try:
# 兼容code和symbol两种字段名
symbol = data.get('code') or data.get('symbol')
name = data.get('name', symbol)
rating = data.get('rating', '持有')
entry_min = data.get('entry_min')
entry_max = data.get('entry_max')
take_profit = data.get('take_profit')
stop_loss = data.get('stop_loss')
check_interval = data.get('check_interval', 60)
notification_enabled = data.get('notification_enabled', True)
# 验证必需字段
if not symbol or not all([entry_min, entry_max, take_profit, stop_loss]):
print(f"[WARN] {symbol} 参数不完整,跳过")
failed += 1
continue
# 构建entry_range
entry_range = {"min": entry_min, "max": entry_max}
# 检查是否已存在
existing = self.get_monitor_by_code(symbol)
if existing:
# 更新现有监测
self.update_monitored_stock(
existing['id'],
rating=rating,
entry_range=entry_range,
take_profit=take_profit,
stop_loss=stop_loss,
check_interval=check_interval,
notification_enabled=notification_enabled
)
updated += 1
print(f"[OK] 更新监测: {symbol}")
else:
# 添加新监测
self.add_monitored_stock(
symbol=symbol,
name=name,
rating=rating,
entry_range=entry_range,
take_profit=take_profit,
stop_loss=stop_loss,
check_interval=check_interval,
notification_enabled=notification_enabled
)
added += 1
print(f"[OK] 添加监测: {symbol}")
except Exception as e:
symbol_str = data.get('code') or data.get('symbol', 'Unknown')
print(f"[ERROR] 处理监测失败 ({symbol_str}): {str(e)}")
failed += 1
result = {
"added": added,
"updated": updated,
"failed": failed,
"total": added + updated + failed
}
print(f"\n[OK] 批量同步完成: 新增{added}只, 更新{updated}只, 失败{failed}")
return result
# 全局数据库实例
monitor_db = StockMonitorDatabase()
+238
View File
@@ -517,5 +517,243 @@ _此消息由AI股票分析系统自动发送_"""
'configured': bool(self.config['webhook_url'])
}
def send_portfolio_analysis_notification(self, analysis_results: dict, sync_result: dict = None) -> bool:
"""
发送持仓分析完成通知
Args:
analysis_results: 批量分析结果
sync_result: 监测同步结果(可选)
Returns:
是否发送成功
"""
try:
# 构建通知内容
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)
results = analysis_results.get("results", [])
# 邮件主题
subject = f"持仓定时分析完成 - 共{total}只股票"
# 构建邮件正文(HTML格式)
html_body = f"""
<html>
<head>
<style>
body {{ font-family: Arial, sans-serif; }}
.summary {{ background-color: #f0f8ff; padding: 15px; border-radius: 5px; margin-bottom: 20px; }}
.stock {{ border: 1px solid #ddd; padding: 10px; margin-bottom: 10px; border-radius: 5px; }}
.success {{ color: green; }}
.failed {{ color: red; }}
.rating-buy {{ color: #28a745; font-weight: bold; }}
.rating-hold {{ color: #ffc107; font-weight: bold; }}
.rating-sell {{ color: #dc3545; font-weight: bold; }}
</style>
</head>
<body>
<h2>持仓定时分析完成</h2>
<div class="summary">
<h3>分析概况</h3>
<p>总数: {total} 只</p>
<p class="success">成功: {succeeded} 只</p>
<p class="failed">失败: {failed} 只</p>
<p>耗时: {elapsed_time:.2f} 秒</p>
"""
# 添加监测同步结果
if sync_result:
html_body += f"""
<h3>监测同步结果</h3>
<p>新增监测: {sync_result.get('added', 0)} 只</p>
<p>更新监测: {sync_result.get('updated', 0)} 只</p>
<p>同步失败: {sync_result.get('failed', 0)} 只</p>
"""
html_body += """
</div>
<h3>分析结果详情</h3>
"""
# 添加每只股票的详细结果
for item in results[:10]: # 只显示前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", "未知")
confidence = final_decision.get("confidence_level", "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")
# 评级颜色
rating_class = "rating-hold"
if "强烈买入" in rating or "买入" in rating:
rating_class = "rating-buy"
elif "卖出" in rating:
rating_class = "rating-sell"
html_body += f"""
<div class="stock">
<h4>{code} {stock_info.get('name', '')} - <span class="{rating_class}">{rating}</span> (信心度: {confidence})</h4>
<p>进场区间: {entry_range}</p>
<p>止盈位: {take_profit} | 止损位: {stop_loss}</p>
</div>
"""
else:
error = result.get("error", "未知错误")
html_body += f"""
<div class="stock">
<h4 class="failed">{code} - 分析失败</h4>
<p>错误: {error}</p>
</div>
"""
if len(results) > 10:
html_body += f"<p>...还有 {len(results) - 10} 只股票未显示</p>"
html_body += """
</body>
</html>
"""
# 构建纯文本版本
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()
+625
View File
@@ -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] 数据库测试完成")
+526
View File
@@ -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] 持仓管理器测试完成")
+715
View File
@@ -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] 调度器测试完成")
Binary file not shown.
+771
View File
@@ -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}%")
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.