如何在Python中最优实现market、sector、group、stock的股市分类体系？

现有扁平文件解析 #

你当前的星号标记层级的文本可以快速解析为嵌套字典结构，比列表嵌套更适合做节点查找，Python实现代码如下：

def parse_stock_hierarchy(file_path: str) -> dict:
    hierarchy = {}
    # 栈存储各级节点，索引对应层级深度：0=market、1=sector、2=group、3=stock
    stack = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            star_count = line.count('*')
            content = line.lstrip('*')
            depth = star_count
            # 弹出栈中超出当前层级的节点
            while len(stack) > depth:
                stack.pop()
            current_node = {"name": content, "children": {}}
            if depth == 0:
                hierarchy = current_node
            else:
                stack[-1]["children"][content] = current_node
            stack.append(current_node)
    return hierarchy

存储格式选择 #

优先选JSON格式，完全适配嵌套结构，Python原生支持，冗余度远低于XML，读写成本极低：

• 序列化存储代码：

import json

hierarchy = parse_stock_hierarchy("your_raw_data.txt")
with open("stock_hierarchy.json", "w", encoding="utf-8") as f:
    json.dump(hierarchy, f, ensure_ascii=False, indent=2)

• 读取直接调用json.load()即可，1500条数据的读写耗时可以忽略。如果后续需要做复杂统计、多条件筛选，也可以转存SQLite，建4张带外键关联的表分别对应4个分类层级即可。

工具类封装 #

为了符合Pythonic的封装规范，建议把常用操作封装成独立类，避免裸操作嵌套字典，示例实现如下：

class StockClassifier:
    def __init__(self, json_path: str):
        with open(json_path, 'r', encoding='utf-8') as f:
            self.data = json.load(f)
    
    def get_stock_belong(self, stock_name: str) -> dict:
        """查询单支股票所属的全部分类层级"""
        for sector_name, sector in self.data["children"].items():
            for group_name, group in sector["children"].items():
                if stock_name in group["children"]:
                    return {
                        "market": self.data["name"],
                        "sector": sector_name,
                        "group": group_name,
                        "stock": stock_name
                    }
        return {}
    
    def list_sector_stocks(self, sector_name: str) -> list:
        """获取指定行业下的所有股票列表"""
        stocks = []
        sector = self.data["children"].get(sector_name)
        if not sector:
            return stocks
        for group in sector["children"].values():
            stocks.extend(group["children"].keys())
        return stocks

使用时直接实例化调用对应方法即可，后续新增分类修改、批量导出等功能都可以直接在类中扩展。

可选优化 #

如果后续数据高频更新、需要对外提供查询接口，可以把各级映射关系存到Redis中，用哈希结构存储，查询响应速度更快。

内容的提问来源于stack exchange，提问作者John