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

import tushare as tsimport pandas as pdfrom sqlalchemy import create_engineimport scheduleimport timefrom datetime import datetime, timedeltaimport akshare as ak# 初始化Tushare Prots.set_token('XXXXX')  # 替换为实际Tokenpro = 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_datesdef 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)