智能股票分析系统
连板天梯复盘系统是一款专为短线交易者设计的智能股票分析平台，集成了多数据源和AI分析引擎。该系统采用PyQt5框架开发，包含18个功能模块，覆盖连板分析、AI智能诊断、市场情绪量化等功能。核心技术包括：1）三层架构设计（UI层、业务逻辑层、数据层）；2）多AI服务商集成（腾讯、豆包、DeepSeek）；3）智能缓存策略；4）异步加载机制。系统支持实时数据更新、盘中监控和风险预警，通过连板天梯算法自
> 作者: 机器懒得学习
> 日期: 2025年2月6日
> 版本: v38.0
> 关键词: PyQt5, 股票分析, AI智能, 量化交易, 连板天梯
目录
- [项目概述](项目概述)
- [技术架构](技术架构)
- [核心功能模块](核心功能模块)
- [AI智能分析引擎](ai智能分析引擎)
- [数据处理系统](数据处理系统)
- [用户界面设计](用户界面设计)
- [性能优化策略](性能优化策略)
- [部署与配置](部署与配置)
- [未来发展规划](未来发展规划)
项目概述
连板天梯复盘系统是一款专为短线交易者设计的智能股票分析平台，通过集成多个数据源、AI分析引擎和专业的量化指标，为投资者提供全方位的市场分析和决策支持。系统历经38个版本的迭代优化，已成为功能完善、性能稳定的专业级交易辅助工具。
主要特色
- ✅ 18个功能模块 - 覆盖股票分析的各个维度
- ✅ 多AI引擎集成 - 腾讯、豆包、DeepSeek等多家AI服务
- ✅ 实时数据更新 - 支持盘中实时监控
- ✅ 智能预警系统 - 基于AI的风险识别
- ✅ 专业连板分析 - 核心连板天梯算法
- ✅ 情绪量化指标 - 市场情绪数字化呈现
技术架构
系统架构图
```
┌─────────────────────────────────────────────────────────────┐
│ 连板天梯复盘系统 v38.0 │
├─────────────────────────────────────────────────────────────┤
│ 用户界面层 (PyQt5) │
│ ├── 18个功能页面 (QStackedWidget) │
│ ├── 自定义控件 (表格/图表/Web引擎) │
│ └── 主题切换 (深色/浅色) │
├─────────────────────────────────────────────────────────────┤
│ 业务逻辑层 │
│ ├── 数据加载器 (DataLoadWorker) │
│ ├── 筛选处理器 (FilterWorker) │
│ ├── AI分析器 (AIAnalysisThread) │
│ └── 更新管理器 (UpdateThread) │
├─────────────────────────────────────────────────────────────┤
│ 数据访问层 │
│ ├── 本地数据库 (SQLite) │
│ ├── 远程数据接口 (HTTP API) │
│ ├── 问财数据适配器 (pywencai) │
│ └── 通达信集成 (TDX Link) │
├─────────────────────────────────────────────────────────────┤
│ 外部数据源 │
│ ├── 同花顺问财 │
│ ├── 新浪财经 │
│ ├── 通达信行情 │
│ └── 腾讯/豆包/DeepSeek AI │
└─────────────────────────────────────────────────────────────┘
核心技术栈
|
技术类别 |
具体技术 |
用途说明 |
|
--------- |
--------- |
---------- |
|
GUI框架 |
PyQt5 + QtWebEngine |
跨平台桌面应用界面 |
|
数据处理 |
pandas + numpy |
股票数据分析和处理 |
|
图表展示 |
matplotlib + QtWebEngine |
数据可视化展示 |
|
AI集成 |
腾讯AI + 豆包 + DeepSeek |
智能分析和决策支持 |
|
数据存储 |
SQLite + 远程数据库 |
本地缓存和云端同步 |
|
网络请求 |
requests + aiohttp |
数据接口调用 |
|
打包部署 |
PyInstaller |
应用程序打包 |
核心功能模块详解
1. 连板天梯模块 - 系统核心
连板天梯是系统的核心功能，通过智能算法将涨停股票按连板天数进行分组展示，帮助用户快速识别市场龙头股。
```python
class LadderWindow(QMainWindow):
def create_group_widgets(self, df):
"""创建连板分组组件"""
按连板天数分组
grouped = df.groupby('连板天数')
for days, group in grouped:
每组按涨停时间排序
group = group.sort_values('涨停时间')
创建分组卡片
group_widget = GroupWidget(
title=f"{days}连板 ({len(group)}只)",
data=group,
color=self.get_level_color(days)
)
self.ladder_layout.addWidget(group_widget)
- *核心算法逻辑:
- 数据分组 - 按连板天数分类（1板、2板、3板...）
- 时间排序 - 组内按涨停时间先后排序
- 指标计算 - 计算封单比、换手率、成交额等关键指标
- 状态标记 - 识别一字板、T字板、换手板等状态
2. AI智能分析引擎
系统集成多个AI服务商，提供全方位的股票分析：
```python
class AIAnalysisThread(QThread):
"""AI分析工作线程"""
def run(self):
try:
根据不同AI服务商调用相应接口
if self.ai_provider == "tencent":
result = self.tencent_analysis()
elif self.ai_provider == "doubao":
result = self.doubao_analysis()
elif self.ai_provider == "deepseek":
result = self.deepseek_analysis()
self.result_signal.emit(result)
except Exception as e:
self.error_signal.emit(str(e))
- *AI分析维度:
- 基本面分析 - 财务数据、行业地位、成长性
- 技术面分析 - K线形态、支撑压力位、技术指标
- 情绪面分析 - 市场关注度、概念热度、资金流向
- 风险分析 - 业绩风险、技术风险、系统性风险
- 投资建议 - 基于AI模型的买卖建议
3. 实时数据监控系统
```python
class MarketSentimentThread(QThread):
"""市场情绪监控线程"""
def run(self):
while self.is_running:
try:
获取实时市场数据
market_data = self.fetch_market_data()
计算情绪指标
sentiment_score = self.calculate_sentiment(market_data)
发送更新信号
self.data_update.emit({
'up_count': market_data['up_count'],
'down_count': market_data['down_count'],
'limit_up': market_data['limit_up'],
'limit_down': market_data['limit_down'],
'sentiment': sentiment_score
})
每30秒更新一次
time.sleep(30)
except Exception as e:
self.error.emit(str(e))
- *监控指标:
- 涨跌家数比 - 市场总体强弱
- 涨停跌停比 - 市场赚钱效应
- 连板高度 - 投机情绪强度
- 炸板率 - 市场分歧程度
- 热点持续性 - 板块轮动速度
AI智能分析引擎深度解析
多AI服务商集成架构
```python
class AIAnalyzer:
"""AI分析器 - 集成多个AI服务商"""
def __init__(self):
self.providers = {
'tencent': TencentAI(),
'doubao': DoubaoAI(),
'deepseek': DeepSeekAI()
}
def analyze_stock(self, stock_code, analysis_type='comprehensive'):
"""股票分析入口"""
results = {}
并行调用多个AI分析
with ThreadPoolExecutor(max_workers=3) as executor:
futures = {}
for name, provider in self.providers.items():
future = executor.submit(
provider.analyze,
stock_code,
analysis_type
)
futures[name] = future
收集结果
for name, future in futures.items():
try:
results[name] = future.result(timeout=30)
except Exception as e:
results[name] = {'error': str(e)}
return self.combine_results(results)
AI分析流程
```
┌─────────────────────────────────────────────────────────────┐
│ AI股票分析流程 │
├─────────────────────────────────────────────────────────────┤
│ 1. 数据准备 │
│ ├── 获取股票基础数据 (价格、成交量、财务数据) │
│ ├── 提取技术指标 (MACD、KDJ、RSI等) │
│ ├── 获取市场情绪数据 (关注度、资金流向) │
│ └── 收集相关新闻和公告 │
│ │
│ 2. AI分析处理 │
│ ├── 腾讯AI - 深度基本面分析 │
│ ├── 豆包AI - 实时情绪分析 │
│ └── DeepSeek - 技术面综合分析 │
│ │
│ 3. 结果整合 │
│ ├── 多模型结果对比验证 │
│ ├── 置信度评分 │
│ ├── 风险等级评估 │
│ └── 生成投资建议 │
└─────────────────────────────────────────────────────────────┘
数据处理系统架构
数据流程图
```
┌─────────────────────────────────────────────────────────────┐
│ 数据处理流程 │
├─────────────────────────────────────────────────────────────┤
│ 数据源层 │
│ ├── 同花顺问财 ←──┐ │
│ ├── 新浪财经 ←──┼──┐ │
│ ├── 通达信行情 ←─┼──┼──┐ │
│ └── 其他数据源 ←─┼──┼──┼──┐ │
│ │ │ │ │ │
├───────────────────┼──┼──┼──┤ │
│ 数据采集层 │ │ │ │ │
│ ├── API调用管理器 │ │ │ │ │
│ ├── 数据验证器 │ │ │ │ │
│ ├── 异常处理器 │ │ │ │ │
│ └── 重试机制 │ │ │ │ │
│ │ │ │ │ │
├────────────────────┼──┼──┼──┤ │
│ 数据处理层 │ │ │ │ │
│ ├── 数据清洗 │ │ │ │ │
│ ├── 格式标准化 │ │ │ │ │
│ ├── 指标计算 │ │ │ │ │
│ └── 质量检查 │ │ │ │ │
│ ↓ ↓ ↓ ↓ │
├─────────────────────┴──┴──┴──┤ │
│ 数据存储层 │ │
│ ├── 本地SQLite缓存 │ │
│ ├── 远程数据库 │ │
│ └── 文件缓存 │ │
│ │ │
├────────────────────────────────┼─────────────────────────────┤
│ 应用接口层 │ │
│ ├── 统一数据接口 │ │
│ ├── 缓存管理器 │ │
│ └── 数据同步器 │ │
│ ↓ │
└────────────────────────────────┴─── 业务逻辑层 (UI展示/AI分析) ─┘
数据缓存策略
```python
class DataManager:
"""数据管理器 - 智能缓存策略"""
def __init__(self):
self.local_cache = {} 内存缓存
self.cache_timeout = 300 5分钟超时
self.db_manager = DBManager()
def get_stock_data(self, stock_code, date_str, force_remote=False):
"""获取股票数据 - 智能缓存策略"""
cache_key = f"{stock_code}_{date_str}"
1. 检查内存缓存
if not force_remote and cache_key in self.local_cache:
cached_data = self.local_cache[cache_key]
if time.time() - cached_data['timestamp'] < self.cache_timeout:
return cached_data['data']
2. 检查本地数据库
db_data = self.db_manager.get_stock_data(stock_code, date_str)
if db_data and not force_remote:
更新内存缓存
self.local_cache[cache_key] = {
'data': db_data,
'timestamp': time.time()
}
return db_data
3. 从远程获取
try:
remote_data = self.fetch_remote_data(stock_code, date_str)
保存到数据库
self.db_manager.save_stock_data(stock_code, date_str, remote_data)
更新内存缓存
self.local_cache[cache_key] = {
'data': remote_data,
'timestamp': time.time()
}
return remote_data
except Exception as e:
如果远程获取失败，返回数据库中的历史数据
if db_data:
return db_data
else:
raise Exception(f"无法获取股票数据: {e}")
用户界面设计详解
界面布局架构
```python
class LadderWindow(QMainWindow):
def init_ui(self):
"""初始化用户界面"""
主布局 - 垂直布局
main_layout = QVBoxLayout()
1. 顶部统计栏
stats_bar = self.create_stats_bar()
main_layout.addWidget(stats_bar)
2. 主内容区域 - 水平分割
splitter = QSplitter(Qt.Horizontal)
左侧导航面板
left_panel = self.create_left_panel()
splitter.addWidget(left_panel)
右侧内容区域
right_panel = self.create_right_panel()
splitter.addWidget(right_panel)
设置分割比例
splitter.setStretchFactor(0, 1) 左侧
splitter.setStretchFactor(1, 4) 右侧
main_layout.addWidget(splitter)
3. 底部状态栏
status_bar = self.create_status_bar()
main_layout.addWidget(status_bar)
界面截图展示
主界面布局
```
┌─────────────────────────────────────────────────────────────┐
│ [Logo] 连板天梯复盘系统 v38.0 [最小化][最大化][关闭] │
├─────────────────────────────────────────────────────────────┤
│ 涨停: 45 跌停: 8 上涨: 2185 下跌: 1895 大盘: +0.85% │
│ 涨停连板: 12 跌停连板: 3 核心题材: 人工智能、新能源车 │
├─────────────────────┬───────────────────────────────────────┤
│ □ 今日涨停 │ │
│ □ 连板天梯 │ 主要功能展示区域 │
│ □ AI智能复盘 │ │
│ □ 批量AI诊断 │ (QStackedWidget管理18个页面) │
│ □ 今日跌停 │ │
│ □ 上涨股票 │ │
│ □ 下跌股票 │ │
│ □ 炸板分析 │ │
│ □ 回封分析 │ │
│ □ 新题材 │ │
│ □ 龙头股 │ │
│ □ N字反包 │ │
│ □ 赚钱效应 │ │
│ □ 涨停分析 │ │
│ □ 跌停分析 │ │
│ □ 短线情绪 │ │
│ □ 投资日历 │ │
│ □ 板块轮动 │ │
├─────────────────────┼───────────────────────────────────────┤
│ [设置] [关于] [帮助] │ 状态栏: 数据更新状态、版本信息等 │
└─────────────────────┴───────────────────────────────────────┘
连板天梯展示效果
```
┌─── 1连板 (28只) ───────────────────────────────────────────┐
│ ┌──┬──────────┬──────┬──────┬──────┬──────┬──────┬──────┐ │
│ │序号│ 股票名称 │ 代码 │ 涨停时间│ 封单比│ 换手率│ 成交额│ 概念 │ │
│ ├──┼──────────┼──────┼──────┼──────┼──────┼──────┼──────┤ │
│ │ 1 │ 平安银行 │000001│ 09:25 │ 2.5% │ 1.2% │ 12.5亿│ 银行 │ │
│ │ 2 │ 比亚迪 │002594│ 09:30 │ 1.8% │ 2.1% │ 45.2亿│ 新能源│ │
│ │ 3 │ 宁德时代│300750│ 09:32 │ 3.2% │ 1.8% │ 38.6亿│ 锂电池│ │
│ └──┴──────────┴──────┴──────┴──────┴──────┴──────┴──────┘ │
└─────────────────────────────────────────────────────────────┘
┌─── 2连板 (8只) ────────────────────────────────────────────┐
│ ┌──┬──────────┬──────┬──────┬──────┬──────┬──────┬──────┐ │
│ │序号│ 股票名称 │ 代码 │ 涨停时间│ 封单比│ 换手率│ 成交额│ 概念 │ │
│ ├──┼──────────┼──────┼──────┼──────┼──────┼──────┼──────┤ │
│ │ 1 │ 科大讯飞│002230│ 09:25 │ 5.8% │ 0.9% │ 18.3亿│ AI │ │
│ │ 2 │ 中科曙光│603019│ 09:27 │ 4.2% │ 1.5% │ 22.1亿│ 算力 │ │
│ │ 3 │ 浪潮信息│000977│ 09:35 │ 3.8% │ 2.3% │ 31.2亿│ 服务器│ │
│ └──┴──────────┴──────┴──────┴──────┴──────┴──────┴──────┘ │
└─────────────────────────────────────────────────────────────┘
性能优化策略
1. 异步加载机制
```python
class DataLoadWorker(QThread):
"""数据加载工作线程 - 避免界面卡顿"""
data_loaded = pyqtSignal(dict)
error_occurred = pyqtSignal(str)
def __init__(self, params):
super().__init__()
self.params = params
self.is_running = True
def run(self):
try:
显示加载进度
self.progress_updated.emit(0, "开始加载数据...")
步骤1: 获取数据
self.progress_updated.emit(20, "获取股票数据...")
raw_data = self.fetch_stock_data()
if not self.is_running:
return
步骤2: 数据处理
self.progress_updated.emit(50, "处理数据...")
processed_data = self.process_data(raw_data)
if not self.is_running:
return
步骤3: 计算指标
self.progress_updated.emit(80, "计算技术指标...")
result = self.calculate_indicators(processed_data)
步骤4: 完成
self.progress_updated.emit(100, "数据加载完成")
self.data_loaded.emit(result)
except Exception as e:
self.error_occurred.emit(str(e))
2. 内存管理优化
```python
class MemoryManager:
"""内存管理器 - 优化大数据处理"""
def __init__(self, max_memory_mb=500):
self.max_memory = max_memory_mb * 1024 * 1024 转换为字节
self.current_memory = 0
self.data_cache = {}
def cache_data(self, key, data):
"""智能缓存数据"""
data_size = sys.getsizeof(data)
检查内存限制
if self.current_memory + data_size > self.max_memory:
清理最久未使用的数据
self.cleanup_cache()
self.data_cache[key] = {
'data': data,
'size': data_size,
'timestamp': time.time(),
'access_count': 1
}
self.current_memory += data_size
def cleanup_cache(self):
"""清理缓存 - LRU算法"""
if not self.data_cache:
return
按访问时间排序
sorted_items = sorted(
self.data_cache.items(),
key=lambda x: (x[1]['access_count'], -x[1]['timestamp'])
)
移除最久未使用且访问次数最少的数据
remove_count = max(1, len(sorted_items) // 4)
for i in range(remove_count):
key, item = sorted_items[i]
self.current_memory -= item['size']
del self.data_cache[key]
3. 数据库查询优化
```python
class DatabaseOptimizer:
"""数据库查询优化器"""
def __init__(self):
self.query_cache = {}
self.indexes_created = False
def create_indexes(self):
"""创建数据库索引 - 提高查询性能"""
if self.indexes_created:
return
indexes = [
"CREATE INDEX IF NOT EXISTS idx_stock_code ON stock_data(code)",
"CREATE INDEX IF NOT EXISTS idx_stock_date ON stock_data(date)",
"CREATE INDEX IF NOT EXISTS idx_stock_code_date ON stock_data(code, date)",
"CREATE INDEX IF NOT EXISTS idx_limit_up ON stock_data(limit_up)",
"CREATE INDEX IF NOT EXISTS idx_continuous_days ON stock_data(continuous_days)"
]
for index_sql in indexes:
self.execute(index_sql)
self.indexes_created = True
def get_stock_data_optimized(self, stock_codes, date_range):
"""优化的股票数据查询"""
cache_key = f"{','.join(stock_codes)}_{date_range[0]}_{date_range[1]}"
检查查询缓存
if cache_key in self.query_cache:
cached_result = self.query_cache[cache_key]
if time.time() - cached_result['timestamp'] < 300: 5分钟缓存
return cached_result['data']
使用参数化查询避免SQL注入
placeholders = ','.join(['?' for _ in stock_codes])
query = f"""
SELECT code, name, date, open, high, low, close, volume, amount,
limit_up, limit_down, continuous_days, limit_time
FROM stock_data
WHERE code IN ({placeholders})
AND date BETWEEN ? AND ?
ORDER BY code, date
"""
params = stock_codes + [date_range[0], date_range[1]]
result = self.execute(query, params).fetchall()
缓存查询结果
self.query_cache[cache_key] = {
'data': result,
'timestamp': time.time()
}
return result
未来发展规划
短期目标 (v39.0-v40.0)
- ✅ 移动端适配 - 开发移动端APP
- ✅ 云端同步 - 用户数据云端同步
- ✅ AI模型优化 - 提高分析准确率
- ✅ 更多数据源 - 接入Level2行情数据
中期目标 (v41.0-v43.0)
- 量化回测 - 添加策略回测功能
- 组合管理 - 投资组合跟踪分析
- 智能预警 - 基于AI的实时预警
- 小程序 - 微信小程序版本
长期愿景 (v44.0+)
深度学习 - 自建深度学习模型
- 社区功能 - 用户交流社区
- 机构版 - 面向专业机构版本
- 国际化 - 支持海外市场
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
27
0- 0
已为社区贡献2条内容
所有评论(0)