同花顺概念行业板块分析及成分股展示

pip install pyside6 pandas requests pywencai

pywencai需要依赖node， 建议node18， 不知道怎么操作的可以翻一翻我之前的文章。

import sys
import json
import pandas as pd
import requests
import pywencai
from functools import lru_cache
from PySide6.QtWidgets import (
    QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
    QTabWidget, QListWidget, QListWidgetItem, QTableView, QSplitter,
    QMessageBox, QStatusBar, QProgressBar, QHeaderView, QLabel
)
from PySide6.QtCore import Qt, QThread, Signal, QObject, QTimer, QAbstractTableModel, QModelIndex, QSize
from PySide6.QtGui import QFont, QColor, QBrush
# -------------------- 数据模型 --------------------
class StockTableModel(QAbstractTableModel):
    """成分股表格模型，涨跌幅列按正负显示颜色"""
    def __init__(self):
        super().__init__()
        self._data = pd.DataFrame(columns=['股票代码', '股票名称', '最新价', '涨跌幅'])
        self._headers = ['股票代码', '股票名称', '最新价', '涨跌幅']
    def update_data(self, df: pd.DataFrame):
        self.beginResetModel()
        self._data = df
        self.endResetModel()
    def rowCount(self, parent=QModelIndex()):
        return self._data.shape[0]
    def columnCount(self, parent=QModelIndex()):
        return self._data.shape[1]
    def data(self, index, role=Qt.DisplayRole):
        if not index.isValid():
            return None
        row, col = index.row(), index.column()
        value = self._data.iloc[row, col]
        if role == Qt.DisplayRole:
            if col == 3 and isinstance(value, (int, float)):
                return f"{value:.2f}%"
            return str(value)
        elif role == Qt.ForegroundRole:
            if col == 3:
                try:
                    if isinstance(value, str):
                        val_str = value.replace('%', '').strip()
                        val = float(val_str) if val_str else 0.0
                    else:
                        val = float(value)
                    if val > 0:
                        return QBrush(QColor(255, 80, 80))   # 红色
                    elif val < 0:
                        return QBrush(QColor(0, 150, 0))     # 绿色
                    else:
                        return QBrush(Qt.black)
                except:
                    return QBrush(Qt.gray)
            return QBrush(Qt.black)
        return None
    def headerData(self, section, orientation, role=Qt.DisplayRole):
        if orientation == Qt.Horizontal and role == Qt.DisplayRole:
            return self._headers[section]
        return None
# -------------------- 工作线程 --------------------
class DataLoader(QObject):
    concept_blocks_ready = Signal(pd.DataFrame)
    industry_blocks_ready = Signal(pd.DataFrame)
    stock_list_ready = Signal(pd.DataFrame, str)
    error_occurred = Signal(str)
    def __init__(self):
        super().__init__()
        self._cache_concept = None
        self._cache_industry = None
    @lru_cache(maxsize=128)
    def _fetch_concept_blocks(self):
        return pywencai.get(query="概念板块,涨跌幅排序", query_type="zhishu", sort_order='desc', loop=True)
    @lru_cache(maxsize=128)
    def _fetch_industry_blocks(self):
        return pywencai.get(query="行业板块,涨跌幅排序", query_type="zhishu", sort_order='desc', loop=True)
    def load_concept_blocks(self):
        try:
            df = self._fetch_concept_blocks()
            self.concept_blocks_ready.emit(df)
        except Exception as e:
            self.error_occurred.emit(f"获取概念板块失败: {str(e)}")
    def load_industry_blocks(self):
        try:
            df = self._fetch_industry_blocks()
            self.industry_blocks_ready.emit(df)
        except Exception as e:
            self.error_occurred.emit(f"获取行业板块失败: {str(e)}")
    def load_stock_list(self, block_code: str):
        try:
            url = f"https://d.10jqka.com.cn/v2/blockrank/{block_code}/199112/d1000.js"
            headers = {
                'Referer': 'http://q.10jqka.com.cn/',
                'User-Agent': ('Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                               'AppleWebKit/537.36 (KHTML, like Gecko) '
                               'Chrome/119.0.0.0 Safari/537.36')
            }
            resp = requests.get(url, headers=headers, timeout=10)
            if resp.status_code != 200:
                self.error_occurred.emit(f"请求失败，状态码：{resp.status_code}")
                return
            json_str = resp.text.split('(', 1)[1].rsplit(')', 1)[0]
            data = json.loads(json_str)
            items = data.get('items', [])
            if not items:
                self.stock_list_ready.emit(pd.DataFrame(), block_code)
                return
            rows = []
            for s in items:
                rows.append([
                    s.get('5', '').zfill(6),
                    s.get('55', ''),
                    s.get('8', ''),
                    s.get('199112', 0)
                ])
            df = pd.DataFrame(rows, columns=['股票代码', '股票名称', '最新价', '涨跌幅'])
            self.stock_list_ready.emit(df, block_code)
        except Exception as e:
            self.error_occurred.emit(f"获取成分股失败: {str(e)}")
# -------------------- 主窗口 --------------------
class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("同花顺板块分析")
        self.setMinimumSize(1100, 650)
        self._setup_ui()
        self._setup_loader()
        self._load_blocks()
    def _setup_ui(self):
        central = QWidget()
        self.setCentralWidget(central)
        main_layout = QHBoxLayout(central)
        main_layout.setContentsMargins(5, 5, 5, 5)
        splitter = QSplitter(Qt.Horizontal)
        main_layout.addWidget(splitter)
        left_widget = QWidget()
        left_layout = QVBoxLayout(left_widget)
        left_layout.setContentsMargins(0, 0, 0, 0)
        self.tab_widget = QTabWidget()
        left_layout.addWidget(self.tab_widget)
        # 概念板块列表
        self.concept_list = QListWidget()
        self.concept_list.setAlternatingRowColors(True)
        self.concept_list.itemClicked.connect(self._on_block_selected)
        self.tab_widget.addTab(self.concept_list, "概念板块")
        # 行业板块列表
        self.industry_list = QListWidget()
        self.industry_list.setAlternatingRowColors(True)
        self.industry_list.itemClicked.connect(self._on_block_selected)
        self.tab_widget.addTab(self.industry_list, "行业板块")
        splitter.addWidget(left_widget)
        # 右侧表格
        self.table_view = QTableView()
        self.table_model = StockTableModel()
        self.table_view.setModel(self.table_model)
        self.table_view.setAlternatingRowColors(True)
        self.table_view.horizontalHeader().setStretchLastSection(True)
        self.table_view.horizontalHeader().setSectionResizeMode(QHeaderView.Interactive)
        splitter.addWidget(self.table_view)
        splitter.setSizes([350, 750])
        self._progress = QProgressBar()
        self._progress.setVisible(False)
        self.statusBar().addPermanentWidget(self._progress)
        self.setStyleSheet("""
            QListWidget::item { border-bottom: 1px solid #e0e0e0; }
            QListWidget::item:selected { background-color: #c0e0ff; }
            QTabWidget::pane { border: 1px solid #cccccc; }
            QHeaderView::section { background-color: #f0f0f0; padding: 4px; }
        """)
    def _setup_loader(self):
        self._loader = DataLoader()
        self._thread = QThread()
        self._loader.moveToThread(self._thread)
        self._thread.start()
        self._loader.concept_blocks_ready.connect(self._on_concept_blocks_loaded)
        self._loader.industry_blocks_ready.connect(self._on_industry_blocks_loaded)
        self._loader.stock_list_ready.connect(self._on_stock_list_loaded)
        self._loader.error_occurred.connect(self._show_error)
    def _load_blocks(self):
        self._progress.setVisible(True)
        self._progress.setRange(0, 0)
        QTimer.singleShot(100, self._loader.load_concept_blocks)
        QTimer.singleShot(200, self._loader.load_industry_blocks)
    def _on_concept_blocks_loaded(self, df):
        self._populate_list_widget(self.concept_list, df)
        self._check_all_loaded()
    def _on_industry_blocks_loaded(self, df):
        self._populate_list_widget(self.industry_list, df)
        self._check_all_loaded()
    def _get_change_column(self, df: pd.DataFrame) -> str:
        possible_names = ['涨跌幅', '涨跌幅(%)', '涨幅', '涨幅%', '涨跌幅度']
        for col in df.columns:
            if col in possible_names:
                return col
        for col in df.columns:
            if '涨跌幅' in col or '涨幅' in col:
                return col
        return None
    def _format_change(self, value):
        if pd.isna(value):
            return '--'
        try:
            if isinstance(value, str):
                value = value.replace('%', '').strip()
            val = float(value)
            return f"{val:+.2f}%"
        except:
            return str(value)
    def _get_change_color_html(self, change_str: str) -> str:
        try:
            if change_str.endswith('%'):
                num_str = change_str[:-1].strip()
            else:
                num_str = change_str
            if num_str.startswith('+'):
                num_str = num_str[1:]
            val = float(num_str)
            if val > 0:
                return 'red'
            elif val < 0:
                return 'green'
            else:
                return 'black'
        except:
            return 'gray'
    def _populate_list_widget(self, list_widget: QListWidget, df: pd.DataFrame):
        """填充左侧列表，使用QLabel实现颜色区分，并增加项高度"""
        list_widget.clear()
        if '指数简称' not in df.columns or 'code' not in df.columns:
            self._show_error("数据格式错误：缺少指数简称或code列")
            return
        change_col = self._get_change_column(df)
        if change_col is None:
            self._show_error("未找到涨跌幅列，将显示默认值")
            changes = ['--'] * len(df)
        else:
            changes = df[change_col].values
        for idx, row in df.iterrows():
            name = row['指数简称']
            code = str(row['code'])
            change_val = changes[idx]
            change_str = self._format_change(change_val)
            color = self._get_change_color_html(change_str)
            # 创建列表项
            item = QListWidgetItem(list_widget)
            item.setData(Qt.UserRole, code)
            # 创建QLabel显示HTML，增加内边距以提高项高度
            label = QLabel()
            html = f"""
            <div style="display: flex; justify-content: space-between; font-family: Arial; width: 100%;">
                <span style="font-weight: normal;">{name} <span style="color: #666;">({code})</span></span>
                <span style="font-weight: bold; color: {color};">{change_str}</span>
            </div>
            """
            label.setText(html)
            label.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
            # 增加内边距，使项更高
            label.setStyleSheet("padding: 8px 4px; background-color: transparent;")
            label.setAutoFillBackground(False)
            # 让label根据内容和内边距调整大小
            label.adjustSize()
            # 设置项的大小提示，使其高度适应label
            item.setSizeHint(label.sizeHint())
            # 将label设置为item的widget
            list_widget.setItemWidget(item, label)
    def _check_all_loaded(self):
        if self.concept_list.count() > 0 and self.industry_list.count() > 0:
            self._progress.setVisible(False)
    def _on_block_selected(self, item: QListWidgetItem):
        block_code = item.data(Qt.UserRole)
        self.statusBar().showMessage("正在加载成分股...")
        self._progress.setVisible(True)
        self._progress.setRange(0, 0)
        QTimer.singleShot(0, lambda: self._loader.load_stock_list(block_code))
    def _on_stock_list_loaded(self, df: pd.DataFrame, block_code: str):
        self._progress.setVisible(False)
        self.statusBar().showMessage("就绪", 3000)
        if df.empty:
            QMessageBox.information(self, "提示", "该板块暂无成分股数据")
            self.table_model.update_data(pd.DataFrame())
        else:
            self.table_model.update_data(df)
            self.table_view.resizeColumnsToContents()
    def _show_error(self, msg: str):
        self._progress.setVisible(False)
        self.statusBar().showMessage("错误", 3000)
        QMessageBox.critical(self, "错误", msg)
    def closeEvent(self, event):
        self._thread.quit()
        self._thread.wait()
        event.accept()
# -------------------- 启动应用 --------------------
def main():
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app.exec())
if __name__ == "__main__":
    main()

如果我的分享对你有所帮助， 不吝啬给个点赞呗