【量化交易笔记】3.实现数据库保存数据

news2025/1/23 14:47:39

上一节，我们通过下载相关的 pandas 数据保存为本地csv文件，这一节将上节的数据以数据库方式保存。

数据库保存

采集数据部分前一节已做说明，这里就直接用采用前面的内容。这里着重说明的事数据库连接。对与 python 相连接的数据库有很多，作为开放操作性及性能首选 mysql 。（MYSQL的安装这里不做说明），在使用之前，需要安装一个pymysql库，如果没有安装过，用以下命令进行安装，另外一个库sqlalchemy ，一般是默认安装好的。

库安装

pip install pymysql

定义一个数据库连接函数，返回连接对象，以下并非原创，感觉挺好用的，就引用来的。

数据库的连接

def conn():
    # 引擎参数信息
    host = 'localhost'
    user = 'root'
    passwd = 'root'
    port = '3306'
    db = 'quant'

    # 创建数据库引擎对象
    mysql_engine = sqlalchemy.create_engine(
        'mysql+pymysql://{0}:{1}@{2}:{3}'.format(user, passwd, host, port),
        poolclass=sqlalchemy.pool.NullPool
    )

    # 如果不存在数据库db_quant则创建
    mysql_engine.execute("CREATE DATABASE IF NOT EXISTS {0} ".format(db))

    # 创建连接数据库db_quant的引擎对象
    db_engine = sqlalchemy.create_engine(
        'mysql+pymysql://{0}:{1}@{2}:{3}/{4}?charset=utf8'.format(user, passwd, host, port, db),
        pool_size=80, max_overflow=80, pool_timeout=50
    )
    # 返回引擎对象
    return db_engine

上述代码，已很清晰的表述，返回的是数据库连接对象。
而数据表的有两种形式，一种将所有数据股票数据放在一张表里，好处是读写操作方便，缺点表的记录太大了，读取和写入效率非常低。因此将采取另一种方式，每一支股票一张表，那样的话，数据表大概有4000多，读取效率会很快，数据表名即股票名，例如“600001_sh”。通过to_sql()函数写入数据库。

# 写入数据库
table_name = '{}_{}'.format(code[3:], code[:2])
out_df.to_sql(name=table_name, con=engine, if_exists='replace', index=True, index_label='id')

首次执行

完整代码如下

# 第一次执行
import baostock as bs
import pandas as pd
import gc
import time

bs.login()

stock_df = bs.query_all_stock().get_data()
# 筛选股票数据，上证和深证股票代码在sh.600000与sz.39900之间
stock_df = stock_df[(stock_df['code'] >= 'sh.600000') & (stock_df['code'] < 'sz.399000')]
bs.logout()
stocks=stock_df['code'].to_list()

lg = bs.login()
i=0
#数据库连接
engine = conn()
for code in stocks:    
    rs = bs.query_history_k_data_plus(code,
    "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,isST",
    start_date='2020-01-01', end_date='2023-3-1', #实际应用开始时间选2000-1-1 或更早
    frequency="d", adjustflag="1")
    df=rs.get_data()


    # 剔除停盘数据
    if df.shape[0]:
        df = df[(df['volume'] != '0') & (df['volume'] != '')]

    # 如果数据为空，则不创建
    if not df.shape[0]:
        continue

    # 删除重复数据
    df.drop_duplicates(['date'], inplace=True)

    # 日线数据少于250，则不创建
    if df.shape[0] < 250:
        continue

    # 将数值数据转为float型，便于后续处理
    convert_list = ['open', 'high', 'low', 'close', 'preclose', 'volume', 'amount', 'turn', 'pctChg']
    df[convert_list] = df[convert_list].astype(float)

       
    #df.to_csv("./data/daily/{0}.csv".format(code), index=False)
    
    # 写入数据库
    table_name = '{}_{}'.format(code[3:], code[:2])
    df.to_sql(name=table_name, con=engine, if_exists='replace', index=True, index_label='id')

    i=i+1
    if i%100==0 :
        gc.collect()
        print('已完成',i)
        time.sleep(2)   
bs.logout()

与上节的代码的区别，增加了数据库连接，将写csv文件修改为写入数据库。

执行完毕，打开数据库查看如下图。
stockdb

日常执行

# 日常执行
import baostock as bs
import pandas as pd
import gc
import time
import datetime

todate=datetime.date.today().strftime('%Y-%m-%d')


bs.login()
stock_df =  bs.query_sz50_stocks().get_data() # bs.query_all_stock().get_data()

# 筛选股票数据，上证和深证股票代码在sh.600000与sz.39900之间
stock_df = stock_df[(stock_df['code'] >= 'sh.600000') & (stock_df['code'] < 'sz.399000')]
bs.logout()
stock=stock_df['code'].to_list()

lg = bs.login()
i=0

#数据库连接
engine = conn()

for code in stocks:    
    rs = bs.query_history_k_data_plus(code,
    "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,isST",
    start_date=todate, end_date=todate, #选择当天
    frequency="d", adjustflag="1")
    df=rs.get_data()

    # 剔除停盘数据
    if df.shape[0]:
        df = df[(df['volume'] != '0') & (df['volume'] != '')]

    # 如果数据为空，则不创建
    if not df.shape[0]:
        continue

    # 将数值数据转为float型，便于后续处理
    convert_list = ['open', 'high', 'low', 'close', 'preclose', 'volume', 'amount', 'turn', 'pctChg']
    df[convert_list] = df[convert_list].astype(float)

       
    #df.to_csv("./data/daily/{0}.csv".format(code), mode='a', index=False, header=False) 
    # 写入数据库
    table_name = '{}_{}'.format(code[3:], code[:2])
    df.to_sql(name=table_name, con=engine, if_exists='append', index=True, index_label='id')

    i=i+1    
    if i%500==0 :
        gc.collect()
        print('已完成',i)
        time.sleep(2)   
bs.logout()