当前位置：首页 > news >正文

Python + MySQL + Tushare 实现股票数据采集与K线分析系统

news 2026/8/11 2:05:56

下面给你一套Python + MySQL + Tushare Pro 实现股票数据采集与 K 线分析系统的可落地方案，覆盖：环境准备 → 数据库设计 → Tushare 采集（历史+增量）→ K 线指标计算 → 查询分析接口。代码基于 Tushare Pro 最新pro.daily接口与 MySQL 8.x 生产可用写法。

一、系统架构

Tushare Pro (daily) │ pandas DataFrame ▼ collector.py ──▶ MySQL (kline_daily) │ │ │ ▼ │ analyzer.py (MA/MACD/KDJ/BOLL) ▼ │ scheduler.py query_api.py (SQL 查询 / 回测准备)

核心原则：

按交易日维度采集（推荐trade_date=拉全市场，比按股票循环快 10 倍）
MySQL 用唯一索引去重，ON DUPLICATE KEY UPDATE幂等写入
复权类型分开存表（未复权 / 后复权 / 前复权）

二、依赖与配置

pip install tushare pandas sqlalchemy pymysql mysql-connector-python

config.py

import os class Config: TUSHARE_TOKEN = os.getenv("TUSHARE_TOKEN", "你的token") MYSQL_USER = "root" MYSQL_PASSWORD = "pwd" MYSQL_HOST = "127.0.0.1" MYSQL_PORT = 3306 MYSQL_DB = "stock_db" SQLALCHEMY_URI = ( f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}" f"@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}?charset=utf8mb4" ) # 采集节流（基础积分 500次/分钟） API_SLEEP_PER_100 = 12 # 每100次请求睡12秒

三、MySQL 表结构（重点）

CREATE DATABASE stock_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE stock_db; CREATE TABLE kline_daily ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ts_code VARCHAR(20) NOT NULL COMMENT '000001.SZ', trade_date DATE NOT NULL, `open` DECIMAL(12,4) NOT NULL, high DECIMAL(12,4) NOT NULL, low DECIMAL(12,4) NOT NULL, close DECIMAL(12,4) NOT NULL, pre_close DECIMAL(12,4), pct_chg DECIMAL(8,4), vol BIGINT COMMENT '成交量(手)', amount DECIMAL(20,4) COMMENT '成交额(千元)', adj_type TINYINT NOT NULL DEFAULT 0 COMMENT '0未复权 1后复权 2前复权', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_code_date_adj (ts_code, trade_date, adj_type), KEY idx_date (trade_date), KEY idx_code (ts_code) ) ENGINE=InnoDB COMMENT='A股日K线';

⚠️open/high/low/close必须用反引号，否则和 MySQL 关键字冲突。

四、Tushare 采集层（历史 + 增量）

collector.py

import time import pandas as pd from sqlalchemy import create_engine import tushare as ts from config import Config ts.set_token(Config.TUSHARE_TOKEN) pro = ts.pro_api() engine = create_engine(Config.SQLALCHEMY_URI) FIELDS = "ts_code,trade_date,open,high,low,close,pre_close,pct_chg,vol,amount" def collect_by_date(trade_date: str, adj_type: int = 0): """按交易日拉全市场日线（推荐方式）""" df = pro.daily(trade_date=trade_date, fields=FIELDS) if df is None or df.empty: return 0 df["trade_date"] = pd.to_datetime(df["trade_date"]) df["adj_type"] = adj_type df = df.rename(columns={"vol": "vol"}) df.to_sql( "kline_daily", engine, if_exists="append", index=False, method="multi", chunksize=2000 ) return len(df) def collect_history(start: str, end: str): """遍历交易日历（用 trade_date 循环，不按股票循环）""" cal = pro.trade_cal(exchange="SSE", start_date=start, end_date=end, is_open="1") dates = cal["cal_date"].tolist() for i, d in enumerate(dates): try: n = collect_by_date(d, adj_type=0) print(f"{d} ✅ {n} rows") except Exception as e: print(f"{d} ❌ {e}") if i % 100 == 0 and i: time.sleep(Config.API_SLEEP_PER_100) def incremental_update(): """每日 16:00 后跑：补昨天""" yesterday = pd.Timestamp.now().normalize() - pd.Timedelta(days=1) d = yesterday.strftime("%Y%m%d") collect_by_date(d, adj_type=0)

Tushare 官方建议：“建议提供循环日期来提取全市场数据，不要通过循环 ts_code 来拉取历史”。

五、K 线指标分析层

analyzer.py

import pandas as pd from sqlalchemy import create_engine from config import Config engine = create_engine(Config.SQLALCHEMY_URI) def load_kline(ts_code: str, start: str, end: str) -> pd.DataFrame: sql = f""" SELECT trade_date, open, high, low, close, vol FROM kline_daily WHERE ts_code='{ts_code}' AND adj_type=0 AND trade_date BETWEEN '{start}' AND '{end}' ORDER BY trade_date """ df = pd.read_sql(sql, engine, parse_dates=["trade_date"]) return df.set_index("trade_date") def add_indicators(df: pd.DataFrame) -> pd.DataFrame: # MA df["ma5"] = df["close"].rolling(5).mean() df["ma20"] = df["close"].rolling(20).mean() df["ma60"] = df["close"].rolling(60).mean() # EMA df["ema12"] = df["close"].ewm(span=12, adjust=False).mean() df["ema26"] = df["close"].ewm(span=26, adjust=False).mean() # MACD df["dif"] = df["ema12"] - df["ema26"] df["dea"] = df["dif"].ewm(span=9, adjust=False).mean() df["macd"] = (df["dif"] - df["dea"]) * 2 # BOLL df["boll_mid"] = df["close"].rolling(20).mean() df["boll_std"] = df["close"].rolling(20).std() df["boll_up"] = df["boll_mid"] + 2 * df["boll_std"] df["boll_dn"] = df["boll_mid"] - 2 * df["boll_std"] # KDJ low_9 = df["low"].rolling(9).min() high_9 = df["high"].rolling(9).max() df["rsv"] = (df["close"] - low_9) / (high_9 - low_9) * 100 df["k"] = df["rsv"].ewm(alpha=1/3, adjust=False).mean() df["d"] = df["k"].ewm(alpha=1/3, adjust=False).mean() df["j"] = 3 * df["k"] - 2 * df["d"] return df if __name__ == "__main__": df = load_kline("600519.SH", "20240101", "20240801") df = add_indicators(df) print(df.tail(10))

六、定时调度（每天收盘后增量）

# scheduler.py from apscheduler.schedulers.blocking import BlockingScheduler from collector import incremental_update sched = BlockingScheduler(tz="Asia/Shanghai") sched.add_job(incremental_update, "cron", hour=16, minute=10) sched.start()

七、常见坑（生产必看）

限流：基础积分 500 次/分钟，全市场约 5000 只股票/天，按日期拉最快。
重复数据：必须UNIQUE(ts_code, trade_date, adj_type)+to_sql(append)，否则重跑会炸。
复权：pro.daily是未复权；前/后复权用pro.daily(fields=..., adj='hfq/qfq')或pro.daily_qfq。
vol 单位：Tushare 返回“手”，MySQL 存“手”，分析时注意和成交额单位不一致。
停牌：daily停牌不返回行，不要误以为采集失败。