【Python技术】量化之tushare股票列表数据入mysql

import tushare as tsimport pandas as pdfrom sqlalchemy import create_engineimport scheduleimport timeimport logging# 配置日志logging.basicConfig(    level=logging.INFO,    format='%(asctime)s - %(levelname)s - %(message)s',    handlers=[logging.StreamHandler()])logger = logging.getLogger()# 初始化Tushare Prots.set_token('XXXXXX')  # 替换为实际Tokenpro = ts.pro_api()# 数据库配置DB_CONFIG = {    'user': 'root',    'password': '12345678',    'host': 'localhost',    'database': 'db_stock',    'charset': 'utf8mb4',}engine = create_engine(    f"mysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}/{DB_CONFIG['database']}?charset={DB_CONFIG['charset']}")def get_stock_basic_data():    """获取股票基础数据并进行类型处理"""    try:        df = pro.stock_basic()        return df    except Exception as e:        logger.error(f"❌ 获取股票数据失败: {str(e)}")        return pd.DataFrame()def save_to_database(df, table_name):    """保存数据到数据库，使用批处理并优化性能"""    if df.empty:        logger.warning("⚠️ 无有效数据需要保存")        return False    try:        # 分块写入防止内存溢出        df.to_sql(            table_name,            engine,            if_exists='replace',  


    
# 替换模式防止重复数据            index=False,            chunksize=1000        )        logger.info(f"💾 成功保存 {len(df)} 条数据到表 {table_name}")        return True    except Exception as e:        logger.error(f"❌ 数据库写入失败: {str(e)}")        return Falsedef full_update_stock_basic():    """全量更新股票基础数据"""    start_time = time.time()    logger.info("🔄 开始全量更新股票基础数据...")    df = get_stock_basic_data()    if not df.empty:        success = save_to_database(df, 'stock_basic')        if success:            duration = time.time() - start_time            logger.info(f"✅ 更新完成! 耗时: {duration:.2f}秒")    else:        logger.warning("⚠️ 未获取到有效数据，更新终止")# 主程序逻辑if __name__ == '__main__':    print("=" * 50)    print("📊 股票数据管理系统启动")    print("=" * 50)    # 立即执行一次全量更新    full_update_stock_basic()    # 设置每日定时任务    schedule.every().day.at("20:00").do(        lambda: logger.info("⏰ 触发定时任务...") or full_update_stock_basic()    )    print("\n🚀 定时任务守护进程已启动 (每日20:00执行)")    print("=" * 50)    print("📋 运行日志:")    try:        while True:            schedule.run_pending()            time.sleep(60)    except KeyboardInterrupt:        print("\n🔚 程序已手动终止")