告别付费数据源：用Python+efinance构建免费A股数据采集系统
引言
在量化投资和股票分析领域，数据是决策的基础。传统金融机构通常依赖Wind、iFinD等付费数据服务，但对于个人投资者、量化交易初学者或学生群体来说，这些专业服务的费用往往令人望而却步。有没有一种方法，既能获取可靠的A股历史行情数据，又不需要承担高昂的成本？
本文将介绍如何利用Python和开源库efinance，构建一个完全免费的A股历史数据采集系统。不同于简单的API调用教程，我们将从实际应用场景出发，深入探讨数据获取、存储、更新和维护的全流程解决方案，帮助读者建立自己的本地股票数据库。
1. 为什么选择efinance？
在众多免费金融数据源中，efinance凭借其简单易用的接口和相对稳定的数据质量脱颖而出。与其他替代方案相比，efinance具有以下优势：
- 完全免费：无需注册账号或申请API密钥
- 数据全面：覆盖A股市场所有上市公司的历史行情
- 接口简洁：几行代码即可获取所需数据
- 更新及时：数据更新频率满足一般分析需求
注意：虽然efinance提供了便捷的数据获取方式，但在商业用途中仍需注意数据使用的合规性。
2. 环境准备与基础配置
2.1 安装必要的Python库
首先确保你的Python环境已安装以下库：
pip install efinance pandas tqdm loguru pymongoefinance: 核心数据获取库pandas: 数据处理和分析tqdm: 进度条显示loguru: 日志记录pymongo: MongoDB数据库连接（可选）
2.2 数据存储方案选择
根据数据量和使用场景，可以选择不同的存储方案：
| 存储方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| CSV文件 | 简单易用，无需额外服务 | 查询效率低，不适合大数据量 | 小规模数据，临时分析 |
| SQLite | 轻量级，单文件数据库 | 并发性能有限 | 个人使用，中等数据量 |
| MySQL | 性能好，支持复杂查询 | 需要单独安装服务 | 团队协作，较大数据量 |
| MongoDB | 灵活的模式，适合非结构化数据 | 内存占用较高 | 快速迭代开发，复杂数据结构 |
3. 核心数据获取实现
3.1 单只股票历史数据获取
最基本的操作是获取单只股票的历史行情数据：
import efinance as ef # 获取贵州茅台(600519)2020年全年的日线数据 df = ef.stock.get_quote_history("600519", beg="20200101", end="20201231") print(df.head())返回的数据包含以下字段：
- 日期(date)
- 开盘价(open)
- 收盘价(close)
- 最高价(high)
- 最低价(low)
- 成交量(volume)
- 成交额(turnover)
3.2 批量获取全市场股票数据
对于构建本地数据库的需求，我们需要批量获取所有股票的历史数据。以下是完整的实现方案：
import efinance as ef import pandas as pd from tqdm import tqdm from loguru import logger import time def get_all_stock_codes(): """获取全市场股票代码列表""" return ef.stock.get_realtime_quotes()['股票代码'].tolist() def process_stock_data(stock_code, start_date, end_date): """获取并处理单只股票历史数据""" try: df = ef.stock.get_quote_history(stock_code, beg=start_date, end=end_date) if df.empty: return None # 数据清洗和标准化 df = df.iloc[:, :9] df.columns = ['name', 'code', 'date', 'open', 'close', 'high', 'low', 'volume', 'turnover'] df.index = pd.to_datetime(df['date']) df.drop(['name', 'code', 'date'], axis=1, inplace=True) return df except Exception as e: logger.error(f"Error processing {stock_code}: {str(e)}") return None def save_to_csv(df, stock_code, output_dir="data"): """保存数据到CSV文件""" import os os.makedirs(output_dir, exist_ok=True) df.to_csv(f"{output_dir}/{stock_code}.csv") def batch_download(start_date="20150101", end_date="20221231", delay=3): """批量下载全市场股票数据""" stock_codes = get_all_stock_codes() for code in tqdm(stock_codes, desc="Downloading stock data"): data = process_stock_data(code, start_date, end_date) if data is not None: save_to_csv(data, code) time.sleep(delay) # 避免请求过于频繁 if __name__ == "__main__": batch_download()4. 高级功能与优化
4.1 增量更新机制
维护本地数据库时，我们不需要每次都重新下载全部历史数据，而是可以采用增量更新策略：
def incremental_update(stock_code, last_date, output_dir="data"): """增量更新单只股票数据""" import os from datetime import datetime, timedelta # 计算开始日期（最后日期+1天） start_date = (datetime.strptime(last_date, "%Y%m%d") + timedelta(days=1)).strftime("%Y%m%d") end_date = datetime.now().strftime("%Y%m%d") # 获取新增数据 new_data = process_stock_data(stock_code, start_date, end_date) if new_data is None or new_data.empty: return # 合并数据 file_path = f"{output_dir}/{stock_code}.csv" if os.path.exists(file_path): old_data = pd.read_csv(file_path, index_col=0, parse_dates=True) combined_data = pd.concat([old_data, new_data]) else: combined_data = new_data # 保存更新后的数据 combined_data.to_csv(file_path)4.2 数据质量检查
免费数据源可能存在数据质量问题，建议实施以下检查：
- 缺失值检查：识别并处理缺失的数据点
- 异常值检测：找出明显不合理的数据（如价格为0或极高/极低）
- 连续性验证：确保交易日数据没有不合理的间隔
def validate_data(df): """数据质量验证""" issues = [] # 检查缺失值 if df.isnull().any().any(): issues.append("存在缺失值") # 检查异常价格 if (df['close'] <= 0).any(): issues.append("存在零或负价格") # 检查交易量 if (df['volume'] < 0).any(): issues.append("存在负交易量") return issues if issues else "数据质量良好"5. 实际应用与注意事项
5.1 回测系统集成
获取的历史数据可以直接用于量化回测。以下是一个简单的移动平均策略示例：
def moving_average_strategy(data, short_window=5, long_window=20): """双均线策略""" signals = pd.DataFrame(index=data.index) signals['price'] = data['close'] signals['short_ma'] = data['close'].rolling(window=short_window).mean() signals['long_ma'] = data['close'].rolling(window=long_window).mean() signals['signal'] = 0 signals['signal'][short_window:] = np.where( signals['short_ma'][short_window:] > signals['long_ma'][short_window:], 1, 0) signals['positions'] = signals['signal'].diff() return signals5.2 合规使用建议
虽然efinance是免费的数据源，但在使用时仍需注意：
- 非商业用途：确保你的使用方式符合efinance的服务条款
- 合理请求频率：避免过于频繁的请求，建议单次请求间隔不低于3秒
- 数据缓存：对获取的数据进行本地存储，减少重复请求
- 免责声明：在研究成果中注明数据来源，并说明数据可能存在的不准确性
5.3 性能优化技巧
当处理全市场数据时，性能可能成为瓶颈。以下优化措施值得考虑：
- 多线程/异步请求：使用
concurrent.futures或asyncio提高下载效率 - 数据压缩存储：使用
parquet格式替代CSV，节省存储空间 - 内存管理：分批处理数据，避免一次性加载过多数据到内存
from concurrent.futures import ThreadPoolExecutor, as_completed def parallel_download(stock_codes, max_workers=5): """并行下载股票数据""" with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = { executor.submit(process_stock_data, code, "20150101", "20221231"): code for code in stock_codes } for future in as_completed(futures): code = futures[future] try: data = future.result() if data is not None: save_to_csv(data, code) except Exception as e: logger.error(f"Error downloading {code}: {str(e)}")
看Tensor视图与内存布局)
