量化之tushare日线数据入库Mysql例子

import tushare as ts
import pandas as pd
from sqlalchemy import create_engine
import schedule
import time
from datetime import datetime, timedelta
import akshare as ak
# 初始化Tushare Pro
ts.set_token('XXXXX')  # 替换为实际Token
pro = ts.pro_api()
# 数据库配置 
engine = create_engine('mysql://root:12345678@localhost/db_stock?charset=utf8')
def get_trade_dates(start_date, end_date):
    """
    使用AkShare获取指定日期范围内的交易日历
    参数格式：YYYYMMDD (如：20200101)
    """
    # 获取完整的交易日历(为啥不用tushare，积分不够)
    trade_date_df = ak.tool_trade_date_hist_sina()
    # 格式转换：将日期列转换为YYYYMMDD格式
    trade_date_df['trade_date'] = pd.to_datetime(trade_date_df['trade_date']).dt.strftime('%Y%m%d')
    # 过滤指定日期范围
    mask = (trade_date_df['trade_date'] >= start_date) & (trade_date_df['trade_date'] <= end_date)
    trade_dates = trade_date_df.loc[mask, 'trade_date'].tolist()
    return trade_dates
def init_historical_data():
    """
    初始化历史数据（20200102 - 20250719）
    按交易日获取全市场数据，避免个股循环
    """
    # 获取交易日历
    trade_dates = get_trade_dates('20200101', '20250719')
    print(f"📅 共获取 {len(trade_dates)} 个交易日数据")
    for i, date in enumerate(trade_dates):
        # 频率控制：每分钟不超过500次
        if i % 400 == 0 and i > 0:
            print(f"⏳ 已完成 {i} 个交易日，暂停60秒避免超频...")
            time.sleep(60)
        try:
            # 单次获取当日全市场数据（约4800条）[6](@ref)
            df = pro.daily(trade_date=date)
            # 数据存储
            df.to_sql('stock_daily', engine, if_exists='append', index=False)
            print(f"✅ {date} 写入 {len(df)} 条")
        except Exception as e:
            print(f"❌ {date} 获取失败: {e}")
            # 失败重试逻辑可扩展
def daily_update():
    """每日16:00执行的更新任务"""
    today = datetime.now().strftime("%Y%m%d")
    # 检查今日数据是否存在
    if pd.read_sql(f"SELECT 1 FROM stock_daily WHERE trade_date='{today}'", engine).empty:
        try:
            df = pro.daily(trade_date=today)
            if not df.empty:
                df.to_sql('stock_daily', engine, if_exists='append', index=False)
                print(f"🔄 更新{today}数据: {len(df)}条")
            else:
                print(f"⚠️ 今日{today}无交易数据（可能为节假日）")
        except Exception as e:
            print(f"🔥 更新失败: {e}")
    else:
        print(f"⏩ {today}数据已存在，跳过更新")
# 主程序逻辑
if __name__ == '__main__':
    print("=" * 50)
    print("📊 股票数据初始化系统启动")
    print("=" * 50)
    # 首次初始化历史数据
    start_time = time.time()
    init_historical_data()
    duration = time.time() - start_time
    print(f"⏱️ 历史数据初始化完成，耗时: {duration // 60:.0f}分{duration % 60:.0f}秒")
    # 设置每日定时任务
    schedule.every().day.at("16:00").do(daily_update)
    print("\n🚀 定时任务守护进程已启动 (每日16:00执行)")
    print("=" * 50)
    while True:
        schedule.run_pending()
        time.sleep(60)