简介这是一套面向Python金融开发初学者与量化爱好者的基础自动选股系统源码帮助用户快速掌握基于数据驱动的股票筛选逻辑与工程实现。资源包含36个文件主体为29个Python脚本涵盖数据获取、指标计算、策略回测、实时行情、港股/A股接口及数据库存取等模块辅以3个Shell部署脚本、1个JSON配置文件和1个README说明文档整体仅29KB轻量易读结构清晰便于逐模块学习调试。已有1557人下载学习适合希望从零理解选股流程、复现基础量化策略并开展二次开发的实践者。读者可直接运行main.py启动核心流程通过models.py和strategy类文件掌握多因子筛选逻辑借助backtrader集成回测模块验证策略效果并利用requirements.txt快速搭建依赖环境。1. 这不是“选股软件安装包”而是一套可调试、可验证、可嵌入交易流程的Python自动选股逻辑骨架很多人下载“Python自动选股系统源码.zip”后第一反应是双击解压→运行main.py→期待弹出股票列表。结果报错ModuleNotFoundError: No module named akshare或卡在get_today_all()超时或跑出一堆ST股、退市预警股——这才意识到所谓“源码”不是开箱即用的黑盒工具而是一份带业务语义的策略工程脚手架。它默认依赖A股日频行情、基础财务数据和常见技术指标核心价值在于把“低市盈率放量突破行业轮动”这类模糊策略翻译成df.query(pe_ratio 20 and volume_ratio 1.5)这样的可执行逻辑。适合两类人一是想脱离同花顺/通达信公式编辑器、真正理解选股条件如何落地为数据管道的量化初学者二是需要快速验证新因子比如“近3日主力净流入占比”是否具备区分度的策略研究员。它不承诺收益但承诺每行代码都对应一个可解释的市场假设。2. 从解压到首次跑通四步构建最小可运行环境2.1 解压后必须检查的三个关键文件结构打开Python自动选股系统源码.zip典型目录结构如下不同作者略有差异但核心模块高度一致├── config/ │ ├── settings.yaml # 数据源配置、数据库连接参数 │ └── strategy_config.yaml # 各策略的阈值、周期、权重定义 ├── data/ │ └── cache/ # 本地缓存的CSV/Parquet行情文件首次为空 ├── strategies/ │ ├── value_filter.py # 估值类策略PE/PB/ROE筛选 │ ├── momentum_filter.py # 动量类策略N日涨幅、均线多头排列 │ └── volatility_filter.py # 波动率类策略ATR、标准差过滤 ├── utils/ │ ├── data_loader.py # 统一数据获取接口封装akshare/tushare │ └── signal_generator.py # 信号生成基类含回测框架钩子 ├── main.py # 策略调度入口 └── requirements.txt提示若目录中缺失strategies/或utils/说明该压缩包极可能是教学演示片段而非完整系统需谨慎评估可用性。真正的生产级源码必然包含策略分层与数据抽象。2.2 用conda创建隔离环境并安装核心依赖不要直接pip install -r requirements.txt——多数此类源码的requirements.txt包含过时版本或冲突依赖。推荐用conda创建干净环境# 创建Python 3.9环境兼容绝大多数金融库 conda create -n stock_picker python3.9 conda activate stock_picker # 安装基础科学计算栈避免pandas版本冲突 conda install pandas numpy scikit-learn matplotlib -c conda-forge # 安装金融数据获取库优先选akshare免token且覆盖全 pip install akshare1.10.82 # 固定版本防API变更 # 安装轻量级数据库替代SQLite支持并发读写 pip install duckdb0.10.2注意akshare1.10.82是当前2024年中最稳定的版本新版1.11.x已移除stock_zh_a_hist等关键函数。若main.py调用akshare.stock_zh_a_hist报错立即降级至此版本。2.3 修改config/settings.yaml适配本地数据路径原始配置常硬编码网络路径或旧版数据库地址。必须修改以下三项# config/settings.yaml data: source: akshare # 可选: akshare, tushare, csv cache_dir: ./data/cache/ # 确保此目录存在且有写权限 db_path: ./data/stock.db # DuckDB文件路径非SQLite akshare: timeout: 30 # 防止网络波动导致卡死 retry_times: 3 # 失败重试次数执行前创建缓存目录mkdir -p ./data/cache2.4 运行main.py前的三行验证命令在main.py同级目录执行确认环境无误# 1. 验证akshare能否获取单只股票日线测试网络与基础API python -c import akshare as ak; df ak.stock_zh_a_hist(symbol000001, perioddaily, start_date20240101, end_date20240601); print(df.shape) # 2. 验证DuckDB能否写入测试数据库配置 python -c import duckdb; conn duckdb.connect(./data/stock.db); conn.execute(CREATE TABLE test(x INT)); conn.close() # 3. 检查策略模块导入测试代码结构 python -c from strategies.value_filter import ValueFilter; print(Import OK)若三行均无报错且输出预期结果如(102, 15)、无输出、Import OK则python main.py可安全执行。3. 理解策略代码的三层结构数据层→逻辑层→信号层3.1 数据层utils/data_loader.py决定策略的“食材新鲜度”该文件封装了所有数据获取逻辑是策略稳定性的基石。典型实现包含# utils/data_loader.py import akshare as ak import pandas as pd from pathlib import Path def load_stock_daily(symbol: str, start_date: str, end_date: str) - pd.DataFrame: 从akshare加载A股日线自动处理停牌、复权 try: # 关键使用复权后价格避免除权缺口干扰技术指标 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq # 前复权确保MA计算连续 ) df.columns [date, open, close, high, low, volume, turnover] df[date] pd.to_datetime(df[date]) return df.set_index(date).sort_index() except Exception as e: print(fFailed to load {symbol}: {e}) return pd.DataFrame() # 返回空DataFrame避免中断整个批处理 def load_industry_info() - pd.DataFrame: 加载申万一级行业分类用于行业轮动策略 # akshare提供实时行业映射表 return ak.stock_individual_fund_flow_rank(indicator个股) # 示例实际需调整字段逻辑说明adjustqfq是核心参数——若用hfq后复权会导致历史价格虚高MA线严重偏移qfq保证技术指标计算基于真实交易价格。load_industry_info()返回的行业标签是后续groupby(industry).apply(lambda x: x.sort_values(score, ascendingFalse).head(5))的基础。3.2 逻辑层strategies/value_filter.py定义“选股裁判规则”以经典价值策略为例其核心是将财务指标转化为布尔掩码# strategies/value_filter.py import pandas as pd import numpy as np class ValueFilter: def __init__(self, pe_threshold20, pb_threshold1.5, roe_threshold10): self.pe_threshold pe_threshold self.pb_threshold pb_threshold self.roe_threshold roe_threshold def apply(self, df: pd.DataFrame) - pd.Series: 输入单只股票的完整日线DataFrame含pe/pb/roe列 输出布尔SeriesTrue表示通过筛选 # 关键使用滚动窗口计算避免静态截面错误 # 假设df已通过data_loader注入pe/pb/roe列需提前计算 mask ( (df[pe_ratio].rolling(window5).mean() self.pe_threshold) (df[pb_ratio].rolling(window5).mean() self.pb_threshold) (df[roe].rolling(window3).mean() self.roe_threshold) (df[volume_ratio] 1.2) # 近5日均量比大于1.2排除流动性陷阱 ) return mask # 使用示例 # vf ValueFilter(pe_threshold15) # signals vf.apply(stock_df) # 返回每日是否触发信号参数说明rolling(window5).mean()是关键设计——单日PE可能因财报发布剧烈波动取5日均值平滑噪声volume_ratio需在数据层预先计算当日成交量 / 近20日均量否则此处会报错。若源码未提供该列需在data_loader.py中补充df[volume_ratio] df[volume] / df[volume].rolling(20).mean()3.3 信号层main.py中的调度逻辑决定“何时发号施令”main.py本质是策略编排器其核心循环结构如下# main.py 关键片段 from utils.data_loader import load_stock_daily from strategies.value_filter import ValueFilter import pandas as pd def run_strategy(): # 1. 获取全市场股票代码列表示例取沪深300成分股 symbols pd.read_csv(data/hs300.csv)[symbol].tolist() # 需提前准备 # 2. 并行加载数据提升效率 from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers10) as executor: futures { executor.submit(load_stock_daily, sym, 20240101, 20240601): sym for sym in symbols } stock_data {} for future in futures: try: df future.result() if not df.empty: stock_data[futures[future]] df except Exception as e: print(fLoad failed for {futures[future]}: {e}) # 3. 应用策略并聚合结果 vf ValueFilter(pe_threshold18, pb_threshold1.2) results [] for symbol, df in stock_data.items(): try: signals vf.apply(df) # 取最新交易日信号True表示今日入选 if signals.iloc[-1]: results.append({ symbol: symbol, date: df.index[-1].strftime(%Y-%m-%d), pe: df[pe_ratio].iloc[-1], pb: df[pb_ratio].iloc[-1] }) except Exception as e: continue # 跳过异常股票不影响整体 # 4. 输出结果 result_df pd.DataFrame(results) result_df.to_csv(output/selected_stocks_20240601.csv, indexFalse) print(fSelected {len(result_df)} stocks) if __name__ __main__: run_strategy()逻辑说明ThreadPoolExecutor解决I/O瓶颈——akshare单线程加载300只股票需10分钟以上并行后压缩至2分钟内signals.iloc[-1]确保只取最新信号避免历史信号污染决策result_df.to_csv生成可直接导入Excel分析的结构化结果而非控制台打印。4. 三类高频报错的定位与修复方案4.1 “akshare超时/连接拒绝”网络层问题的标准化应对当load_stock_daily频繁报ReadTimeout或ConnectionError不要盲目增加timeout值。应分层诊断现象根本原因修复命令所有股票均超时akshare服务器限流或DNS解析失败pip install --upgrade akshare 清理DNS缓存ipconfig /flushdnsWindows或sudo dscacheutil -flushcachemacOS单只股票超时如600519该股票代码格式错误或已退市在load_stock_daily中添加校验if not symbol.isdigit() or len(symbol) ! 6: return pd.DataFrame()随机部分超时本地网络不稳定在data_loader.py中为ak.stock_zh_a_hist添加指数退避import time; time.sleep(0.5 * (2 ** retry_count))提示永久性解决方案是启用本地缓存。修改load_stock_daily函数在try块开头添加cache_file Path(f./data/cache/{symbol}_{start_date}_{end_date}.parquet) if cache_file.exists(): return pd.read_parquet(cache_file) # ...原有获取逻辑... df.to_parquet(cache_file) # 成功后缓存4.2 “KeyError: pe_ratio”数据字段缺失的根因追溯此错误表明策略代码期望的财务指标列不存在。需按顺序排查确认数据源是否提供该字段akshare的stock_zh_a_hist仅返回价格量数据不包含PE/PB/ROE。这些字段必须由用户自行计算或从其他接口获取。检查data_loader.py是否注入了衍生字段正确做法是在加载日线后调用财务数据接口补全# 在load_stock_daily末尾追加 from akshare import stock_a_indicator_lgb fin_data stock_a_indicator_lgb(symbolsymbol) # 获取PE/PB/ROE等 if not fin_data.empty: # 将财务数据按日期合并到日线DF df df.merge(fin_data[[date, pe, pb, roe]], ondate, howleft) df.rename(columns{pe:pe_ratio, pb:pb_ratio}, inplaceTrue)验证字段名一致性不同数据源字段名不同如pe_ratiovspevsPE_TTM需统一重命名。在策略类apply()方法开头添加# 强制标准化列名 required_cols [pe_ratio, pb_ratio, roe] for col in required_cols: if col not in df.columns: raise KeyError(fMissing required column: {col}. Available: {list(df.columns)})4.3 “DuckDB read-only error”数据库写入权限的静默陷阱当main.py报错IOError: Cannot open file... permission denied实际是DuckDB试图写入只读文件系统如某些云盘挂载点。验证与修复步骤# 1. 检查文件系统权限 ls -ld ./data/ # 输出应为 drwxr-xr-x若含 r-x 无 w 则需修复 # 2. 临时修复Linux/macOS chmod 755 ./data chmod 644 ./data/stock.db # 确保文件可写 # 3. 彻底规避改用内存数据库仅调试用 # 在main.py中替换数据库初始化 # conn duckdb.connect(./data/stock.db) → conn duckdb.connect(:memory:)注意内存数据库重启即丢失数据仅用于验证逻辑。生产环境必须使用文件数据库并确保./data/目录位于本地SSD而非网络共享盘。5. 将源码转化为可复用策略模块的四个实操技巧5.1 抽离策略为独立pip包让value_filter.py变成pip install stock-strategy将strategies/目录重构为标准Python包支持跨项目复用# 目录结构改造 stock-strategy/ ├── setup.py ├── stock_strategy/ │ ├── __init__.py │ ├── filters/ │ │ ├── __init__.py │ │ ├── value.py # 原value_filter.py │ │ └── momentum.py # 原momentum_filter.py │ └── utils/ │ ├── __init__.py │ └── data_loader.py # 原utils/data_loader.pysetup.py内容from setuptools import setup, find_packages setup( namestock-strategy, version0.1.0, packagesfind_packages(), install_requires[akshare1.10.82, pandas1.5.0], python_requires3.8, )安装后即可在任意项目中调用from stock_strategy.filters.value import ValueFilter vf ValueFilter(pe_threshold15)5.2 用Pydantic定义策略配置告别YAML语法错误将strategy_config.yaml替换为类型安全的Python配置# config/strategy_schema.py from pydantic import BaseModel, Field from typing import List, Optional class ValueStrategyConfig(BaseModel): pe_threshold: float Field(ge0, le100, default20) pb_threshold: float Field(ge0, le10, default1.5) min_roe: float Field(ge0, le50, default10) volume_ratio_min: float Field(ge0.5, le5.0, default1.2) class StrategyConfig(BaseModel): value: ValueStrategyConfig ValueStrategyConfig() momentum: dict {} # 可扩展 # 加载配置 config StrategyConfig.parse_file(config/strategy_config.json) print(config.value.pe_threshold) # IDE可自动补全编辑器实时校验5.3 添加单元测试用真实数据验证策略逻辑在tests/test_value_filter.py中编写可回溯的测试import pandas as pd import pytest from stock_strategy.filters.value import ValueFilter def test_value_filter_edge_cases(): # 构造模拟数据确保覆盖边界条件 df pd.DataFrame({ pe_ratio: [15, 25, 18, 30], # 第1、3日应通过PE筛选 pb_ratio: [1.0, 2.0, 1.3, 0.8], roe: [12, 8, 15, 5], volume_ratio: [1.5, 0.9, 2.0, 1.1] }, indexpd.date_range(2024-01-01, periods4, freqD)) vf ValueFilter(pe_threshold20, pb_threshold1.5, roe_threshold10) result vf.apply(df) # 断言只有第1日索引0满足全部条件 assert result.iloc[0] True assert result.iloc[1] False assert result.iloc[2] False # PB1.31.5但ROE1510但volume_ratio2.01.2需检查逻辑 # 发现BUG原逻辑是AND关系第2日应为True → 触发策略逻辑修正提示运行pytest tests/ -v可自动发现策略逻辑漏洞。真实项目中应使用akshare下载2023年某月全市场数据作为测试基准集。5.4 集成Telegram通知策略结果实时推送手机在main.py末尾添加消息推送无需额外服务import requests import json def send_telegram_alert(results: list): # 从环境变量读取Token和Chat ID避免硬编码 token os.getenv(TELEGRAM_TOKEN) chat_id os.getenv(TELEGRAM_CHAT_ID) if not token or not chat_id: return message f 自动选股结果{len(results)}只\n for r in results[:5]: # 仅推送前5只避免刷屏 message f{r[symbol]} {r[pe]:.1f}PE {r[pb]:.1f}PB\n url fhttps://api.telegram.org/bot{token}/sendMessage payload { chat_id: chat_id, text: message, parse_mode: HTML } requests.post(url, jsonpayload, timeout10) # 在run_strategy()末尾调用 send_telegram_alert(results)配置环境变量后即可生效export TELEGRAM_TOKENyour_bot_token_here export TELEGRAM_CHAT_IDyour_chat_id_here python main.py本文还有配套的精品资源点击获取