用Python+akshare打造智能选股系统：从数据采集到策略落地的完整指南
在信息爆炸的时代，股票投资者面临的最大挑战不是数据不足，而是如何从海量数据中快速准确地识别出符合自己投资策略的优质标的。传统的手工筛选方式不仅效率低下，还容易因人为因素错过最佳投资时机。本文将带你用Python和akshare库构建一个自动化选股系统，实现从数据获取、策略编写到结果通知的全流程自动化。
1. 环境准备与基础配置
1.1 安装必要工具
构建自动化选股系统需要以下基础环境：
- Python 3.7或更高版本
- akshare库（最新版）
- pandas用于数据处理
- smtplib用于邮件通知
安装命令如下：
pip install akshare pandas numpy提示：建议使用虚拟环境管理项目依赖，避免与其他项目产生冲突
1.2 初始化项目结构
合理的项目结构能提高代码可维护性：
/stock_selection │── config.py # 存放配置信息 │── strategy.py # 选股策略实现 │── data_fetcher.py # 数据获取模块 │── notifier.py # 通知模块 │── main.py # 主程序入口2. 数据获取与处理
2.1 使用akshare获取基础数据
akshare提供了丰富的金融数据接口，以下代码展示如何获取A股实时行情：
import akshare as ak def fetch_stock_list(): """获取A股实时行情数据""" stock_df = ak.stock_zh_a_spot() # 保留关键字段 return stock_df[['代码', '名称', '最新价', '涨跌幅', '成交量', '成交额']]2.2 财务数据获取与分析
价值投资需要分析财务指标，以下代码获取单只股票的财务数据：
def fetch_financial_data(stock_code): """获取个股财务指标""" # 去除股票代码前缀（如'sh'或'sz'） pure_code = stock_code[2:] if len(stock_code) > 6 else stock_code return ak.stock_financial_analysis_indicator(pure_code)2.3 数据清洗与预处理
原始数据往往包含缺失值和异常值，需要清洗：
def clean_financial_data(df): """清洗财务数据""" # 设置日期为索引 df = df.set_index('日期') # 处理特殊字符 df = df.replace('--', np.nan) # 转换数据类型 for col in df.columns: df[col] = pd.to_numeric(df[col], errors='ignore') return df3. 选股策略设计与实现
3.1 价值投资策略构建
基于巴菲特的价值投资理念，我们可以设计以下筛选条件：
- 盈利能力：过去5年平均ROE>15%
- 估值水平：市盈率(PE)<30且>0
- 现金流状况：经营现金流为正
- 成长性：最新季度净利润同比增长>10%
策略实现代码：
def value_strategy(stock_code): """价值投资策略""" try: # 获取财务数据 df = fetch_financial_data(stock_code) df = clean_financial_data(df) # 指标1: ROE筛选 roe_series = df['净资产收益率(%)'].last('5Y') avg_roe = roe_series.mean() if not roe_series.empty else 0 # 指标2: PE筛选 pe_data = ak.stock_a_lg_indicator(stock_code) current_pe = pe_data['pe'].iloc[-1] # 指标3: 现金流筛选 cash_flow = df['每股经营性现金流(元)'].iloc[-1] # 指标4: 净利润增长 net_profit = df['扣除非经常性损益后的净利润(元)'] profit_growth = (net_profit.iloc[-1] - net_profit.iloc[-5]) / abs(net_profit.iloc[-5]) return { 'code': stock_code, 'name': get_stock_name(stock_code), 'avg_roe': avg_roe, 'current_pe': current_pe, 'cash_flow': cash_flow, 'profit_growth': profit_growth, 'pass_all': (avg_roe > 15) and (0 < current_pe < 30) and (cash_flow > 0) and (profit_growth > 0.1) } except Exception as e: print(f"处理{stock_code}时出错: {str(e)}") return None3.2 技术指标策略实现
对于偏好技术分析的投资者，可以加入以下技术指标：
def technical_strategy(stock_code): """技术分析策略""" try: # 获取历史K线数据 kline = ak.stock_zh_a_daily(symbol=stock_code, adjust="hfq") # 计算20日均线 kline['ma20'] = kline['close'].rolling(20).mean() # 计算MACD exp12 = kline['close'].ewm(span=12, adjust=False).mean() exp26 = kline['close'].ewm(span=26, adjust=False).mean() macd = exp12 - exp26 signal = macd.ewm(span=9, adjust=False).mean() # 策略条件 last_close = kline['close'].iloc[-1] condition1 = last_close > kline['ma20'].iloc[-1] # 价格在20日均线上方 condition2 = macd.iloc[-1] > signal.iloc[-1] # MACD金叉 return { 'code': stock_code, 'name': get_stock_name(stock_code), 'close': last_close, 'ma20': kline['ma20'].iloc[-1], 'macd': macd.iloc[-1], 'signal': signal.iloc[-1], 'pass_all': condition1 and condition2 } except Exception as e: print(f"技术分析处理{stock_code}出错: {str(e)}") return None4. 结果通知与系统集成
4.1 邮件通知实现
筛选结果可以通过邮件自动发送：
import smtplib from email.mime.text import MIMEText from email.header import Header def send_email(results, strategy_name): """发送选股结果邮件""" # 邮件内容构建 content = f"<h2>{strategy_name}选股结果</h2><table border='1'>" content += "<tr><th>代码</th><th>名称</th><th>最新价</th><th>条件详情</th></tr>" for r in results: content += f"<tr><td>{r['code']}</td><td>{r['name']}</td><td>{r.get('close','N/A')}</td><td>{str(r)}</td></tr>" content += "</table>" # 邮件配置 msg = MIMEText(content, 'html', 'utf-8') msg['From'] = 'your_email@example.com' msg['To'] = 'receiver@example.com' msg['Subject'] = Header(f'{strategy_name}选股结果', 'utf-8') # 发送邮件 try: smtp_obj = smtplib.SMTP('smtp.example.com', 587) smtp_obj.login('your_email@example.com', 'your_password') smtp_obj.sendmail('your_email@example.com', ['receiver@example.com'], msg.as_string()) print("邮件发送成功") except Exception as e: print(f"邮件发送失败: {str(e)}")4.2 定时任务设置
使用APScheduler实现定时运行：
from apscheduler.schedulers.blocking import BlockingScheduler def run_strategy(): """执行选股策略""" stock_list = fetch_stock_list() value_results = [] tech_results = [] for _, row in stock_list.iterrows(): stock_code = row['代码'] value_result = value_strategy(stock_code) if value_result and value_result['pass_all']: value_results.append(value_result) tech_result = technical_strategy(stock_code) if tech_result and tech_result['pass_all']: tech_results.append(tech_result) if value_results: send_email(value_results, "价值投资策略") if tech_results: send_email(tech_results, "技术分析策略") # 设置每天15:30运行（收盘后） scheduler = BlockingScheduler() scheduler.add_job(run_strategy, 'cron', hour=15, minute=30) scheduler.start()5. 策略优化与扩展
5.1 多因子评分系统
简单的通过/不通过筛选可能过于严格，可以改为评分制：
def score_strategy(stock_code): """多因子评分策略""" try: factors = {} # 获取数据 financial_data = clean_financial_data(fetch_financial_data(stock_code)) kline_data = ak.stock_zh_a_daily(symbol=stock_code, adjust="hfq") # 因子1: ROE评分 (0-30分) roe = financial_data['净资产收益率(%)'].last('3Y').mean() factors['roe_score'] = min(30, max(0, (roe - 5) / 20 * 30)) # 因子2: PE评分 (0-20分) pe = ak.stock_a_lg_indicator(stock_code)['pe'].iloc[-1] factors['pe_score'] = 20 if 0 < pe < 15 else (10 if 15 <= pe < 25 else 0) # 因子3: 营收增长评分 (0-20分) revenue = financial_data['营业收入(元)'] rev_growth = (revenue.iloc[-1] - revenue.iloc[-4]) / abs(revenue.iloc[-4]) factors['growth_score'] = min(20, max(0, rev_growth * 100)) # 因子4: 技术面评分 (0-30分) close = kline_data['close'] ma20 = close.rolling(20).mean() factors['tech_score'] = 30 if close.iloc[-1] > ma20.iloc[-1] else 0 # 总分 total_score = sum(factors.values()) factors['total_score'] = total_score factors['code'] = stock_code factors['name'] = get_stock_name(stock_code) return factors except Exception as e: print(f"评分策略处理{stock_code}出错: {str(e)}") return None5.2 数据可视化分析
使用matplotlib对选股结果进行可视化：
import matplotlib.pyplot as plt def visualize_results(results): """可视化选股结果""" if not results: return df = pd.DataFrame(results) # 创建图表 fig, axes = plt.subplots(2, 2, figsize=(12, 10)) # ROE分布 axes[0,0].hist(df['avg_roe'], bins=20, color='skyblue') axes[0,0].set_title('ROE分布') axes[0,0].axvline(x=15, color='red', linestyle='--') # PE分布 axes[0,1].hist(df['current_pe'], bins=20, color='lightgreen') axes[0,1].set_title('PE分布') axes[0,1].axvline(x=30, color='red', linestyle='--') # 现金流分布 axes[1,0].hist(df['cash_flow'], bins=20, color='salmon') axes[1,0].set_title('每股现金流分布') axes[1,0].axvline(x=0, color='red', linestyle='--') # 净利润增长分布 axes[1,1].hist(df['profit_growth'], bins=20, color='gold') axes[1,1].set_title('净利润增长分布') axes[1,1].axvline(x=0.1, color='red', linestyle='--') plt.tight_layout() plt.savefig('selection_results.png') plt.close()5.3 异常处理与日志记录
健壮的系统需要完善的错误处理：
import logging from datetime import datetime def setup_logger(): """配置日志记录""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f'stock_selection_{datetime.now().strftime("%Y%m%d")}.log'), logging.StreamHandler() ] ) return logging.getLogger(__name__) logger = setup_logger() def safe_run_strategy(): """带错误处理的策略运行""" try: logger.info("开始执行选股策略") run_strategy() logger.info("策略执行完成") except Exception as e: logger.error(f"策略执行出错: {str(e)}", exc_info=True) # 发送错误通知 send_email([{'error': str(e)}], "选股系统错误报告")
