最近在做一个量化策略回测项目时，需要快速筛选出符合特定财务和技术条件的股票。手动翻看几千只股票的数据，效率低不说，还容易看花眼。于是，我花了三天时间，用 Python 写了一个自动化选股工具。当第一次跑出结果，看到那些被筛选出来的股票组合在历史回测中的表现时，确实有点“惊了”——原来通过一些简单的规则组合，真的能初步过滤掉大量“噪音”，找到潜在的机会池。

本文就将这个工具的完整开发思路、代码实现以及使用心得分享出来。无论你是对量化投资感兴趣的初学者，还是想为自己的策略增加一个自动化筛选环节的开发者，都能从零开始，跟着本文一步步搭建起属于自己的 Python 选股工具。文末会提供完整的、可运行的源码。

1. 选股工具的核心思路与准备工作

在开始敲代码之前，我们需要明确工具要做什么，以及需要哪些“原材料”。

1.1 工具要解决什么问题？

传统选股可能依赖于经验或感觉，而我们的工具旨在实现：

1. 自动化 ：代替人工，快速遍历全市场股票。
2. 规则化 ：将模糊的选股标准（如“业绩好”、“趋势强”）转化为具体的、可量化的规则（如“市盈率小于30”、“收盘价在20日均线之上”）。
3. 可回溯 ：基于历史数据运行，可以验证规则在过去是否有效。
4. 可扩展 ：规则可以方便地增加、修改或组合。

1.2 技术栈与数据源选择

• 编程语言 ：Python。因其在数据分析（pandas, numpy）和可视化（matplotlib）领域的强大生态而成为不二之选。
• 核心库 ：
  • pandas & numpy : 数据处理与分析基石。
  • akshare : 一个免费、开源的财经数据接口库，可以获取到A股行情、基本面等数据，非常适合个人开发者和小型项目。 （注意：数据接口有变更风险，本文代码基于某个稳定版本，使用时请关注akshare官方文档）
  • matplotlib / seaborn : 用于简单的数据可视化，例如绘制选中股票的股价走势。
• 开发环境 ：任何你熟悉的Python环境即可，如PyCharm、VSCode、Jupyter Notebook。建议使用Python 3.8及以上版本。
• 数据源 ：本文示例使用 akshare 。你需要理解，任何免费数据源都可能存在延迟、误差或接口变动，用于学习和小规模回测没问题，但若用于实盘，务必寻求更稳定、更实时的数据源并进行充分验证。

1.3 环境搭建与依赖安装

首先，确保你的Python环境已就绪。然后，通过pip安装必要的库。建议创建一个新的虚拟环境。

# 创建并激活虚拟环境（以venv为例）
python -m venv stock_env
# Windows:
stock_env\Scripts\activate
# Linux/Mac:
source stock_env/bin/activate

# 安装核心依赖
pip install pandas numpy akshare matplotlib

如果安装 akshare 较慢或遇到问题，可以使用国内镜像源：

pip install pandas numpy akshare matplotlib -i https://pypi.tuna.tsinghua.edu.cn/simple

2. 数据获取：构建选股的基础数据库

没有数据，一切分析都是空谈。我们的第一步是学会如何获取并组织股票数据。

2.1 获取股票基础信息列表

我们需要知道市场上有哪些股票。 akshare 提供了接口获取A股列表。

import akshare as ak

# 获取沪深京A股列表
stock_info_a_code_name_df = ak.stock_info_a_code_name()
print(f“获取到A股数量：{len(stock_info_a_code_name_df)}”)
print(stock_info_a_code_name_df.head())

# 保存到本地CSV，避免每次运行都重新下载（可选）
stock_info_a_code_name_df.to_csv(“stock_basic_list.csv”, index=False, encoding=‘utf-8-sig’)

运行后，你会得到一个包含股票代码和股票名称的DataFrame。股票代码是后续获取详细数据的钥匙。

2.2 获取个股历史行情数据

行情数据是技术分析的基础。我们写一个函数来获取单只股票的历史日K线数据。

import pandas as pd

def get_stock_daily_data(stock_code, start_date=“20230101”, end_date=“20231231”):
    “””
    获取单只股票的日线行情数据
    :param stock_code: 股票代码，带交易所前缀，例如 ‘sh600000’ 或 ‘sz000001’
    :param start_date: 开始日期，格式 ‘YYYYMMDD’
    :param end_date: 结束日期，格式 ‘YYYYMMDD’
    :return: pandas DataFrame
    “””
    try:
        # 注意：ak.stock_zh_a_hist 需要标准代码，如 ‘600000’， 不需要前缀
        # 但我们需要区分市场，这里假设传入的是 ‘600000’ 或 ‘000001’
        # 实际使用中，需要根据代码前几位判断市场，这里做简单处理
        if stock_code.startswith(‘6’):
            symbol = stock_code + ‘.SH’
        elif stock_code.startswith(‘0’) or stock_code.startswith(‘3’):
            symbol = stock_code + ‘.SZ’
        else:
            print(f“无法识别股票代码市场：{stock_code}”)
            return None

        # 使用 ak.stock_zh_a_hist 获取数据，注意参数名可能随版本变化
        df = ak.stock_zh_a_hist(symbol=symbol, period=“daily”, start_date=start_date, end_date=end_date, adjust=“qfq”)
        # adjust=“qfq” 表示前复权，保证价格可比性

        if df is not None and not df.empty:
            # 重命名列，使其更易读
            df.columns = [‘日期’， ‘开盘’， ‘收盘’， ‘最高’， ‘最低’， ‘成交量’， ‘成交额’， ‘振幅’， ‘涨跌幅’， ‘涨跌额’， ‘换手率’]
            df[‘日期’] = pd.to_datetime(df[‘日期’])
            df.set_index(‘日期’， inplace=True)
            df.sort_index(inplace=True) # 按日期升序排列
            return df
        else:
            print(f“未获取到股票 {stock_code} 的数据”)
            return None
    except Exception as e:
        print(f“获取股票 {stock_code} 数据时出错：{e}”)
        return None

# 测试获取贵州茅台的数据
df_mt = get_stock_daily_data(‘600519’， ‘20240101’， ‘20240601’)
if df_mt is not None:
    print(df_mt.head())
    print(f“数据形状：{df_mt.shape}”)

关键点说明 ：

1. 复权 ： adjust=“qfq” （前复权）至关重要。它消除了分红、送股等事件对股价造成的断层，使得历史价格序列连续可比，是进行任何技术分析的前提。
2. 错误处理 ：网络请求可能失败，接口可能变化，所以用 try-except 包裹，并检查返回数据是否为空。
3. 数据缓存 ：频繁请求大量股票数据可能被限制或速度慢。在实际开发中，应考虑将数据缓存到本地数据库（如SQLite）或文件中，下次直接读取。

2.3 获取基本面数据（以市盈率为例）

基本面数据更新频率较低（季度或年度），我们可以一次性获取全市场的最新数据。

def get_basic_financial_data():
    “””
    获取A股最新基本面数据（示例：市盈率、市净率）
    :return: pandas DataFrame
    “””
    try:
        # 获取沪深京A股实时行情数据，其中包含一些基本面字段
        # 注意：不同接口返回的字段名可能不同，这里是一个示例
        df = ak.stock_zh_a_spot_em()
        # 筛选我们关心的列：代码、名称、市盈率（动态）、市净率
        # 需要根据 ak.stock_zh_a_spot_em 返回的实际列名调整
        if ‘代码’ in df.columns and ‘名称’ in df.columns:
            # 假设列名是 ‘市盈率-动态’ 和 ‘市净率’， 请根据打印出的列名确认
            print(“可用列：”， df.columns.tolist())
            # 例如，选择以下列
            selected_cols = [‘代码’， ‘名称’， ‘最新价’， ‘市盈率-动态’， ‘市净率’]
            if all(col in df.columns for col in selected_cols):
                basic_df = df[selected_cols].copy()
                # 重命名
                basic_df.columns = [‘code’， ‘name’， ‘price’， ‘pe_ratio’， ‘pb_ratio’]
                # 将字符串类型的数值转换为浮点数，处理‘-’等无效值
                basic_df[‘pe_ratio’] = pd.to_numeric(basic_df[‘pe_ratio’]， errors=‘coerce’)
                basic_df[‘pb_ratio’] = pd.to_numeric(basic_df[‘pb_ratio’]， errors=‘coerce’)
                return basic_df
        return None
    except Exception as e:
        print(f“获取基本面数据出错：{e}”)
        return None

# 测试
basic_df = get_basic_financial_data()
if basic_df is not None:
    print(basic_df.head())
    print(f“获取到 {len(basic_df)} 条基本面数据”)

注意 ：基本面数据接口和字段名变化相对频繁。运行上述代码后，请先查看打印出的 df.columns ，确认正确的市盈率、市净率等字段名，然后修改 selected_cols 列表。 pd.to_numeric(..., errors=‘coerce’) 会将无效字符串（如‘-’）转换为 NaN ，便于后续过滤。

3. 定义并实现选股规则（策略）

这是工具的核心。我们将选股逻辑封装成一个个独立的“过滤器”（Filter），每个过滤器负责检查股票是否满足单一条件。最后可以灵活组合这些过滤器。

3.1 规则一：市盈率过滤器（价值型）

筛选市盈率在合理范围内的股票，排除亏损（PE为负）和估值过高的股票。

def filter_by_pe(basic_df, pe_max=30, pe_min=0):
    “””
    根据市盈率过滤股票
    :param basic_df: 包含 ‘pe_ratio’ 列的基本面DataFrame
    :param pe_max: 市盈率上限
    :param pe_min: 市盈率下限（通常为0，排除负值）
    :return: 满足条件的股票代码列表
    “””
    if basic_df is None or ‘pe_ratio’ not in basic_df.columns:
        print(“基本面数据缺失或没有市盈率字段”)
        return []
    # 过滤：PE在 [pe_min, pe_max] 区间内，且不是NaN
    condition = (basic_df[‘pe_ratio’] >= pe_min) & (basic_df[‘pe_ratio’] <= pe_max) & (basic_df[‘pe_ratio’].notna())
    filtered_codes = basic_df.loc[condition， ‘code’].tolist()
    print(f“市盈率过滤器：从 {len(basic_df)} 只股票中筛选出 {len(filtered_codes)} 只。”)
    return filtered_codes

3.2 规则二：价格均线过滤器（趋势型）

筛选当前价格位于某条均线之上的股票，例如“收盘价高于20日均线”，这通常被认为是短期趋势向上的信号。

def filter_by_ma(stock_code, daily_df, ma_days=20):
    “””
    判断单只股票是否满足均线条件
    :param stock_code: 股票代码
    :param daily_df: 该股票的日线DataFrame
    :param ma_days: 均线周期
    :return: True 如果满足条件（最新收盘价 > MA），否则 False
    “””
    if daily_df is None or daily_df.empty:
        return False
    # 计算移动平均线
    daily_df[‘ma’] = daily_df[‘收盘’].rolling(window=ma_days).mean()
    # 获取最新的数据点
    latest_data = daily_df.iloc[-1]
    # 检查最新收盘价是否高于均线，并且均线值不是NaN（确保有足够数据计算）
    if pd.notna(latest_data[‘ma’]) and latest_data[‘收盘’] > latest_data[‘ma’]:
        return True
    else:
        return False

# 批量应用均线过滤器
def batch_filter_by_ma(code_list, start_date, end_date, ma_days=20):
    “””
    批量过滤满足均线条件的股票
    :param code_list: 待筛选的股票代码列表
    :param ma_days: 均线周期
    :return: 满足条件的股票代码列表
    “””
    passed_codes = []
    total = len(code_list)
    for i, code in enumerate(code_list):
        # 添加简单进度提示
        if i % 100 == 0:
            print(f“均线过滤进度：{i}/{total}”)
        df = get_stock_daily_data(code, start_date, end_date)
        if filter_by_ma(code, df, ma_days):
            passed_codes.append(code)
    print(f“均线过滤器：从 {total} 只股票中筛选出 {len(passed_codes)} 只。”)
    return passed_codes

注意 ： batch_filter_by_ma 函数会为每只股票请求历史数据，如果股票数量很多，运行会非常慢。 这是第一个性能瓶颈 。生产环境中需要优化，例如使用本地缓存的数据，或者采用异步请求。

3.3 规则三：量价齐升过滤器（动能型）

筛选“近期成交量显著放大且股价上涨”的股票。一个简单的实现：今日成交量大于过去N日平均成交量的M倍，且今日股价上涨。

def filter_by_volume_price(stock_code, daily_df, avg_days=5, volume_ratio=1.5):
    “””
    量价齐升过滤器
    :param stock_code: 股票代码
    :param daily_df: 日线数据
    :param avg_days: 计算平均成交量的周期
    :param volume_ratio: 今日成交量需要超过均量的倍数
    :return: True 如果满足条件
    “””
    if daily_df is None or len(daily_df) < avg_days + 1:
        return False
    # 计算过去 avg_days 日的平均成交量（不包括今日）
    avg_volume = daily_df[‘成交量’].iloc[-(avg_days+1):-1].mean()
    latest_data = daily_df.iloc[-1]
    # 条件：今日成交量 > 平均成交量 * volume_ratio，且今日股价上涨（涨跌幅>0）
    condition = (latest_data[‘成交量’] > avg_volume * volume_ratio) and (latest_data[‘涨跌幅’] > 0)
    return condition

3.4 规则组合器

我们可以将多个过滤器串联（“与”逻辑）或并联（“或”逻辑）。这里实现一个简单的串联组合：必须通过所有指定的过滤器。

def composite_filter(code_list, basic_df, start_date, end_date, filters_config):
    “””
    组合过滤器
    :param code_list: 初始股票池
    :param basic_df: 基本面数据
    :param start_date: 行情开始日期
    :param end_date: 行情结束日期
    :param filters_config: 过滤器配置字典
        例如：{‘pe’: {‘pe_max’: 30, ‘pe_min’: 5},
               ‘ma’: {‘ma_days’: 20},
               ‘volume’: {‘avg_days’: 5, ‘volume_ratio’: 1.8}}
    :return: 通过所有过滤器的股票代码列表
    “””
    current_pool = code_list.copy()
    print(f“初始股票池数量：{len(current_pool)}”)

    # 1. 应用基本面过滤器（无需行情数据，速度快）
    if ‘pe’ in filters_config:
        pe_codes = filter_by_pe(basic_df, **filters_config[‘pe’])
        # 取交集：当前股票池中且满足PE条件的
        current_pool = list(set(current_pool) & set(pe_codes))
        print(f“应用PE过滤器后剩余：{len(current_pool)}”)

    # 2. 应用需要行情数据的过滤器
    # 为了效率，先获取这些股票的历史数据（可考虑并行或缓存优化）
    passed_codes = []
    total = len(current_pool)
    for i, code in enumerate(current_pool):
        if i % 50 == 0:
            print(f“行情数据过滤进度：{i}/{total}”)
        daily_df = get_stock_daily_data(code, start_date, end_date)
        if daily_df is None:
            continue

        passed = True
        # 检查均线条件
        if ‘ma’ in filters_config and passed:
            if not filter_by_ma(code, daily_df, **filters_config[‘ma’]):
                passed = False
        # 检查量价条件
        if ‘volume’ in filters_config and passed:
            if not filter_by_volume_price(code, daily_df, **filters_config[‘volume’]):
                passed = False
        # 可以继续添加其他过滤器...

        if passed:
            passed_codes.append(code)

    print(f“组合过滤完成。最终选出 {len(passed_codes)} 只股票。”)
    return passed_codes

4. 完整实战：运行选股并分析结果

现在，我们将所有模块组合起来，进行一次完整的选股演练。

4.1 主程序流程

创建一个 main.py 文件，整合所有步骤。

# main.py
import pandas as pd
import akshare as ak
from datetime import datetime, timedelta
import time

# 导入我们上面写的函数（假设它们在一个叫 stock_filters.py 的文件里）
# 为了演示，我们把所有函数放在同一个文件里运行
# 这里直接复制了上面的函数定义，实际项目中应分模块

def main():
    print(“=== Python选股工具开始运行 ===”)
    start_time = time.time()

    # 步骤1：设置参数
    end_date = datetime.now().strftime(“%Y%m%d”)
    start_date = (datetime.now() - timedelta(days=180)).strftime(“%Y%m%d”) # 最近180天数据
    print(f“分析周期：{start_date} 至 {end_date}”)

    # 步骤2：获取基础数据
    print(“\n1. 获取A股基础列表...”)
    stock_list_df = ak.stock_info_a_code_name()
    initial_codes = stock_list_df[‘code’].tolist()
    print(f“获取到初始股票池：{len(initial_codes)} 只股票”)

    print(“\n2. 获取基本面数据（市盈率、市净率）...”)
    basic_df = get_basic_financial_data() # 调用之前定义的函数
    if basic_df is None:
        print(“获取基本面数据失败，退出。”)
        return

    # 步骤3：定义选股策略（过滤器组合）
    my_strategy = {
        ‘pe’: {‘pe_max’: 40, ‘pe_min’: 5},      # 市盈率介于5到40倍
        ‘ma’: {‘ma_days’: 20},                  # 股价在20日均线之上
        ‘volume’: {‘avg_days’: 10, ‘volume_ratio’: 1.5} # 量能是10日均量的1.5倍以上且上涨
    }

    # 步骤4：执行组合过滤
    print(“\n3. 开始执行组合选股策略...”)
    selected_codes = composite_filter(
        code_list=initial_codes,
        basic_df=basic_df,
        start_date=start_date,
        end_date=end_date,
        filters_config=my_strategy
    )

    # 步骤5：输出和保存结果
    print(“\n4. 选股结果：”)
    if selected_codes:
        selected_stocks = stock_list_df[stock_list_df[‘code’].isin(selected_codes)]
        print(selected_stocks[[‘code’， ‘name’]].to_string(index=False))
        # 保存到文件
        result_file = f“selected_stocks_{end_date}.csv”
        selected_stocks.to_csv(result_file, index=False, encoding=‘utf-8-sig’)
        print(f“\n结果已保存到文件：{result_file}”)

        # 步骤6：（可选）简单分析选中股票
        print(“\n5. 对选中股票进行简要分析...”)
        # 例如：计算平均市盈率
        if ‘pe_ratio’ in basic_df.columns:
            selected_pe = basic_df[basic_df[‘code’].isin(selected_codes)][‘pe_ratio’]
            print(f“选中股票的平均市盈率：{selected_pe.mean():.2f}”)
            print(f“选中股票的市盈率中位数：{selected_pe.median():.2f}”)

    else:
        print(“没有股票满足所有选股条件。”)

    elapsed_time = time.time() - start_time
    print(f“\n=== 选股完成，总耗时：{elapsed_time:.2f} 秒 ===”)

if __name__ == “__main__”:
    main()

4.2 运行结果与初步分析

当你运行这个脚本后，控制台会输出类似以下信息：

=== Python选股工具开始运行 ===
分析周期：20231001 至 20240601
1. 获取A股基础列表...
获取到初始股票池：5123 只股票
2. 获取基本面数据（市盈率、市净率）...
可用列： [‘代码’， ‘名称’， ‘最新价’， ‘涨跌幅’， ‘涨跌额’， ‘成交量’， ‘成交额’， ‘振幅’， ‘最高’， ‘最低’， ‘今开’， ‘昨收’， ‘量比’， ‘换手率’， ‘市盈率-动态’， ‘市净率’， ‘总市值’， ‘流通市值’， ‘涨速’， ‘5分钟涨跌’， ‘60日涨跌幅’， ‘年初至今涨跌幅’]
获取到 5123 条基本面数据
3. 开始执行组合选股策略...
初始股票池数量：5123
应用PE过滤器后剩余：1890
行情数据过滤进度：0/1890
行情数据过滤进度：50/1890
...
组合过滤完成。最终选出 47 只股票。
4. 选股结果：
   code  name
0 000001   平安银行
1 000002   万科A
...
结果已保存到文件：selected_stocks_20240601.csv
5. 对选中股票进行简要分析...
选中股票的平均市盈率：18.34
选中股票的市盈率中位数：16.72
=== 选股完成，总耗时：328.47 秒 ===

“惊了”的时刻 ：当你第一次看到从5000多只股票中，通过几条简单的量化规则，筛选出几十只股票，并且它们的平均市盈率、技术形态都符合你预设的“审美”时，你会感受到规则和自动化的力量。你可以立即对这些股票进行更深入的分析，或者将其作为进一步策略研究的股票池，效率提升巨大。

4.3 可视化验证（可选）

我们可以随机挑几只选中的股票，绘制其股价与均线走势，直观感受筛选结果。

import matplotlib.pyplot as plt

def plot_stock_with_ma(stock_code, start_date, end_date, ma_days=20):
    “””绘制股票价格及移动平均线”””
    df = get_stock_daily_data(stock_code, start_date, end_date)
    if df is None:
        return
    df[‘ma’] = df[‘收盘’].rolling(window=ma_days).mean()

    plt.figure(figsize=(12, 6))
    plt.plot(df.index, df[‘收盘’]， label=‘Close Price’， linewidth=1.5)
    plt.plot(df.index, df[‘ma’]， label=f‘MA{ma_days}’， linestyle=‘--’， linewidth=1.5)
    plt.title(f‘{stock_code} - Price & MA{ma_days}’)
    plt.xlabel(‘Date’)
    plt.ylabel(‘Price’)
    plt.legend()
    plt.grid(True, linestyle=‘:’， alpha=0.7)
    plt.show()

# 假设 selected_codes 是之前选出的股票列表
if selected_codes:
    # 画前3只股票的图
    for code in selected_codes[:3]:
        plot_stock_with_ma(code, start_date, end_date, ma_days=20)
        time.sleep(1) # 避免请求过快

5. 常见问题与优化方案（避坑指南）

在开发和运行过程中，你肯定会遇到一些问题。以下是一些典型问题及解决思路。

5.1 数据获取失败或速度慢

问题现象	可能原因	解决思路
ak.stock_zh_a_hist 报错或返回空	1. 接口已更新，参数名或URL变化。 2. 股票代码格式不对。 3. 网络问题。	1. 查看akshare最新官方文档或GitHub Issues。 2. 打印请求的URL和参数，手动在浏览器测试。 3. 使用 try-except 捕获异常，并跳过该股票。
获取全市场数据耗时极长	1. 循环请求，单线程串行。 2. 网络延迟。	1. 使用本地缓存 ：第一次获取后存入SQLite或 .pkl 文件，后续增量更新。 2. 使用并发 ：使用 concurrent.futures.ThreadPoolExecutor 进行多线程请求（注意目标网站的反爬策略）。 3. 减少请求频率 ：在循环中添加 time.sleep(0.1) 。
基本面数据字段名对不上	akshare接口字段名可能随版本更新。	打印 df.columns ，根据实际输出调整代码中的列名。

代码示例：简单的数据缓存

import os
import pickle

def get_daily_data_with_cache(stock_code, start_date, end_date, cache_dir=‘./data_cache’):
    “””带缓存的数据获取函数”””
    os.makedirs(cache_dir, exist_ok=True)
    cache_file = os.path.join(cache_dir, f“{stock_code}_{start_date}_{end_date}.pkl”)

    # 如果缓存存在且未过期（例如，判断文件存在且修改时间在当天）
    if os.path.exists(cache_file):
        file_mtime = datetime.fromtimestamp(os.path.getmtime(cache_file))
        if file_mtime.date() == datetime.now().date(): # 当天缓存有效
            with open(cache_file, ‘rb’) as f:
                print(f“从缓存加载 {stock_code} 数据”)
                return pickle.load(f)

    # 缓存不存在或已过期，重新请求
    print(f“请求 {stock_code} 数据”)
    df = get_stock_daily_data(stock_code, start_date, end_date)
    if df is not None:
        with open(cache_file, ‘wb’) as f:
            pickle.dump(df, f)
    return df

5.2 选股逻辑的常见陷阱

1. 未来函数 ：这是量化策略中最致命的错误。例如，在 2024年1月1日 ，使用了 2024年1月2日 才能知道的数据（如当日收盘价）来做决策。我们的代码中， get_stock_daily_data 获取的是历史数据，在回测时，必须确保在每一个模拟的“交易日”，只能使用该日期及之前的数据。本文示例在批量过滤时，使用了统一的 end_date ，这相当于站在 end_date 这个时点回头看， 是符合回测逻辑的 。但如果你要模拟每日滚动选股，则需要重构代码，按时间点逐日计算。
2. 幸存者偏差 ：我们使用的 stock_info_a_code_name() 获取的是 当前 上市的股票列表。这意味着已经退市的股票不在其中。用这个列表回测过去，会天然地排除掉那些后来退市的“差生”，导致回测结果过于乐观。更严谨的做法是获取历史某一天的全体股票列表。
3. 过拟合 ：如果你不断调整 pe_max 、 ma_days 等参数，直到它们在历史数据上表现完美，那么这个策略很可能已经“过拟合”了历史噪音，在未来失效。应对方法是：将历史数据分为训练集和测试集，用训练集开发策略，用测试集验证；或者进行 交叉验证 。

5.3 性能优化建议

当股票数量多、历史周期长时，循环获取数据并计算会成为瓶颈。

1. 向量化计算 ：尽可能使用pandas/numpy的向量化操作，避免Python层面的循环。例如，计算所有股票的均线条件，可以尝试先构建一个包含所有股票历史数据的大DataFrame（多级索引），然后一次性计算。
2. 并行处理 ：使用 concurrent.futures 或 multiprocessing 模块并行处理股票数据获取和指标计算。
3. 使用专业回测框架 ：对于更复杂的策略，建议迁移到 backtrader 、 zipline 或 qstrader 等专业框架，它们提供了更高效的数据处理和事件驱动回测引擎。

6. 工程化与最佳实践

将这个脚本升级为一个可维护、可扩展的工具，你需要考虑以下几点：

6.1 项目结构

stock_screener/
├── data/               # 数据缓存目录
├── src/                # 源代码
│   ├── __init__.py
│   ├── data_fetcher.py # 数据获取模块
│   ├── filters.py      # 选股过滤器模块
│   ├── screener.py     # 核心筛选引擎
│   └── utils.py        # 工具函数（缓存、日志等）
├── config.yaml         # 配置文件（策略参数、数据源等）
├── main.py             # 主程序入口
├── requirements.txt    # 项目依赖
└── README.md           # 项目说明

6.2 配置化管理

将策略参数、数据源地址、缓存设置等写入配置文件（如 config.yaml ），避免硬编码。

# config.yaml
data:
  start_date: “20240101”
  end_date: “20240601”
  cache_enabled: true
  cache_dir: “./data”

strategy:
  filters:
    pe:
      enabled: true
      pe_max: 40
      pe_min: 5
    ma:
      enabled: true
      ma_days: 20
    volume:
      enabled: true
      avg_days: 10
      volume_ratio: 1.5

在主程序中用 yaml 库加载配置。

6.3 日志记录

使用Python内置的 logging 模块替代 print ，可以方便地控制日志级别、输出到文件等。

import logging

logging.basicConfig(
    level=logging.INFO,
    format=‘%(asctime)s - %(name)s - %(levelname)s - %(message)s’,
    handlers=[
        logging.FileHandler(‘screener.log’),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 使用 logger 代替 print
logger.info(f“获取到初始股票池：{len(initial_codes)} 只股票”)
logger.warning(“获取股票 %s 数据失败，跳过。”， stock_code)

6.4 策略的扩展性

本文展示了PE、均线、量价三个过滤器。你可以很容易地添加更多：

• 技术指标 ：MACD金叉、RSI超卖、布林带突破等。可以集成 TA-Lib 库来计算复杂指标。
• 基本面指标 ：营收增长率、净利润率、ROE、负债率等。需要从财报接口获取数据。
• 市场情绪 ：换手率、资金流向、龙虎榜数据等。
• 自定义规则 ：例如“所属行业为新能源”、“非ST股票”。

只需在 filters.py 中定义新的过滤函数，并在 composite_filter 函数中添加相应的判断逻辑即可。

6.5 关于实盘的严肃警告

本文构建的工具 仅适用于学习、研究和初步的策略探索 ，绝对不可直接用于实盘交易。原因包括：

1. 数据质量与延迟 ：免费数据可能存在错误、缺失或显著延迟。
2. 交易成本 ：策略未考虑佣金、印花税、滑点（实际成交价与预期价的偏差），这些会极大侵蚀利润。
3. 市场冲击 ：小资金模拟交易不影响市场，大资金实盘买卖本身就会影响价格。
4. 策略失效 ：任何历史有效的策略都可能在未来失效。
5. 系统风险 ：网络、程序BUG、交易所规则变化等。

如果你有实盘需求，必须在专业量化平台进行严格的回测和模拟盘交易，并充分理解所有风险。

7. 总结与后续学习方向

通过这三天的开发，我们完成了一个具备基本功能的Python选股工具。它实现了从数据获取、规则定义、组合筛选到结果输出的完整流程。这个工具的价值不在于提供一个“必胜策略”，而在于提供了一个 可验证、可迭代、可扩展的量化研究框架 。

你接下来可以做什么？

1. 丰富数据源 ：接入更多维度的数据，如财务报告、行业分类、宏观经济指标。
2. 开发更复杂的策略 ：学习经典量化策略（如双均线、海龟交易法、均值回归）并用代码实现。
3. 引入回测引擎 ：计算策略的历史收益、夏普比率、最大回撤等关键指标，科学评价策略好坏。
4. 优化性能 ：使用数据库存储数据，利用并行计算加速。
5. 增加可视化 ：用 Plotly 或 PyQt 制作图形化界面，让操作更直观。
6. 学习机器学习 ：尝试用线性回归、决策树等模型来预测股价或生成选股信号。

量化投资是一个交叉领域，结合了编程、金融和数学。这个选股工具是你踏入这个领域的第一块坚实的垫脚石。最重要的是，通过亲手搭建它，你不仅学会了如何使用Python处理金融数据，更关键的是建立了“逻辑化、规则化、可验证”的投资分析思维。