1. 为什么你需要一个本地股票数据库做量化交易的朋友都知道数据是策略的基石。每次跑策略都要重新从网上拉数据不仅效率低还容易受到网络波动影响。我自己刚开始做量化时就经常遇到这种情况策略跑一半突然断网或者API调用次数超限一整天的工作就白费了。Tushare Pro作为国内知名的金融数据接口提供了丰富的股票数据。但直接调用接口有几个痛点一是免费版有调用频率限制二是历史数据查询需要反复请求三是网络延迟影响回测效率。把这些数据落地到本地数据库这些问题就都迎刃而解了。我去年用这个方法搭建了自己的数据仓库现在回测速度提升了10倍不止。更重要的是数据都在自己手里想怎么分析就怎么分析再也不用担心API服务不稳定了。2. 环境准备从零搭建Python数据工程环境2.1 基础软件安装首先确保你的电脑上已经安装了Python 3.7版本。我强烈推荐使用Anaconda来管理Python环境它能自动解决很多依赖问题。安装完Anaconda后创建一个新的虚拟环境conda create -n stock python3.8 conda activate stock接下来安装核心依赖包。除了Tushare我们还需要以下几个工具包pandas数据处理神器sqlalchemyPython的ORM工具pymysqlMySQL的Python驱动tqdm进度条显示一键安装命令pip install tushare pandas sqlalchemy pymysql tqdm2.2 MySQL数据库配置本地数据库我推荐使用MySQL 8.0。安装完成后需要创建一个专用数据库CREATE DATABASE stock_data CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;然后创建一个有权限的用户CREATE USER stock_userlocalhost IDENTIFIED BY your_password; GRANT ALL PRIVILEGES ON stock_data.* TO stock_userlocalhost; FLUSH PRIVILEGES;注意生产环境请使用更复杂的密码这里仅为演示3. Tushare Pro接入指南3.1 获取和配置API Token首先到Tushare官网注册账号并申请Pro版接口权限。免费版有调用限制Pro版数据更全且稳定性更好。拿到Token后我们可以这样配置import tushare as ts ts.set_token(你的Token) # 替换成你的实际Token pro ts.pro_api()建议把Token保存在环境变量中不要硬编码在脚本里。我吃过亏有一次不小心把带Token的代码传到GitHub结果被人恶意刷接口...3.2 接口调用测试先做个简单测试获取茅台(600519.SH)的基本信息df pro.stock_basic(ts_code600519.SH) print(df)如果能看到返回的DataFrame说明接口配置成功。Pro接口返回的都是结构化的pandas DataFrame这给我们后续处理带来了很大便利。4. 数据库设计与实现4.1 数据表结构设计经过多次迭代我总结出这个比较通用的表结构设计CREATE TABLE daily_quotes ( id int(11) NOT NULL AUTO_INCREMENT, ts_code varchar(20) NOT NULL COMMENT 股票代码, trade_date date NOT NULL COMMENT 交易日期, open decimal(12,3) DEFAULT NULL COMMENT 开盘价, high decimal(12,3) DEFAULT NULL COMMENT 最高价, low decimal(12,3) DEFAULT NULL COMMENT 最低价, close decimal(12,3) DEFAULT NULL COMMENT 收盘价, pre_close decimal(12,3) DEFAULT NULL COMMENT 昨收价, change decimal(12,3) DEFAULT NULL COMMENT 涨跌额, pct_chg decimal(12,3) DEFAULT NULL COMMENT 涨跌幅(%), vol decimal(16,3) DEFAULT NULL COMMENT 成交量(手), amount decimal(20,3) DEFAULT NULL COMMENT 成交额(千元), update_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY idx_code_date (ts_code,trade_date), KEY idx_date (trade_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT股票日线行情;这个设计有几个关键点使用复合唯一索引防止数据重复为交易日期建立了单独索引便于时间范围查询字段精度足够大避免数据截断4.2 自动化建表工具我写了个自动建表的Python函数from sqlalchemy import create_engine def init_db(): engine create_engine(mysqlpymysql://stock_user:your_passwordlocalhost/stock_data) with engine.connect() as conn: conn.execute( CREATE TABLE IF NOT EXISTS daily_quotes ( -- 同上表结构 ) ) return engine5. 数据采集全流程实现5.1 单只股票历史数据获取先实现获取单只股票全部历史数据的函数def get_daily_data(ts_code, start_date19901219): try: df pro.daily(ts_codets_code, start_datestart_date) df[ts_code] ts_code # 确保股票代码字段存在 return df except Exception as e: print(f获取{ts_code}数据失败: {str(e)}) return None5.2 批量获取股票列表获取沪深两市所有股票代码def get_all_stocks(): df pro.stock_basic(exchange, list_statusL) return df[ts_code].tolist()5.3 数据存储实现数据存储的核心函数from tqdm import tqdm def save_to_db(engine, df, table_name): try: df.to_sql(table_name, engine, if_existsappend, indexFalse) return True except Exception as e: if Duplicate entry in str(e): print(重复数据跳过) return True print(f存储失败: {str(e)}) return False def batch_save_stocks(): engine init_db() stocks get_all_stocks() for code in tqdm(stocks): data get_daily_data(code) if data is not None: save_to_db(engine, data, daily_quotes)5.4 增量更新策略为了后续更新数据我设计了增量更新逻辑def get_last_trade_date(engine, ts_code): query fSELECT MAX(trade_date) FROM daily_quotes WHERE ts_code{ts_code} last_date pd.read_sql(query, engine).iloc[0,0] return last_date.strftime(%Y%m%d) if last_date else 19901219 def update_daily_data(): engine init_db() stocks get_all_stocks() for code in tqdm(stocks): last_date get_last_trade_date(engine, code) start_date (pd.to_datetime(last_date) pd.Timedelta(days1)).strftime(%Y%m%d) if start_date pd.Timestamp.today().strftime(%Y%m%d): continue data get_daily_data(code, start_date) if data is not None and not data.empty: save_to_db(engine, data, daily_quotes)6. 异常处理与性能优化6.1 常见问题处理在实际运行中我遇到过这些问题和解决方案API限流Tushare Pro有每分钟500次的限制解决方法import time for code in stocks: data get_daily_data(code) time.sleep(0.12) # 控制请求频率网络异常添加重试机制from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def safe_get_daily_data(ts_code, start_date): return pro.daily(ts_codets_code, start_datestart_date)6.2 性能优化技巧批量提交攒够100条记录再写入数据库多线程采集使用concurrent.futures提高效率from concurrent.futures import ThreadPoolExecutor def batch_save_parallel(workers4): stocks get_all_stocks() with ThreadPoolExecutor(max_workersworkers) as executor: list(tqdm(executor.map(process_single_stock, stocks), totallen(stocks)))内存优化处理大数据时使用chunksizefor chunk in pd.read_sql_query(SELECT * FROM big_table, engine, chunksize10000): process(chunk)7. 数据验证与质量检查数据库建好后我通常会做这些检查完整性检查query SELECT ts_code, COUNT(*) as cnt, MIN(trade_date) as first_date, MAX(trade_date) as last_date FROM daily_quotes GROUP BY ts_code ORDER BY cnt DESC pd.read_sql(query, engine)异常值检测query SELECT * FROM daily_quotes WHERE close 0 OR vol 0 OR amount 0 pd.read_sql(query, engine)数据一致性检查query SELECT a.ts_code, a.trade_date FROM daily_quotes a LEFT JOIN daily_quotes b ON a.ts_code b.ts_code AND b.trade_date DATE_ADD(a.trade_date, INTERVAL 1 DAY) WHERE a.pct_chg IS NOT NULL AND b.pct_chg IS NOT NULL AND ABS((a.close - b.pre_close)/b.pre_close*100 - a.pct_chg) 0.1 这套本地股票数据库我已经稳定运行一年多支撑了多个量化策略的回测和实盘。最大的感受就是自己掌控数据的感觉真好再也不用担心第三方API的种种限制了。建议你也动手搭建一个过程中遇到的任何问题都可以参考我的GitHub仓库地址私信获取。