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

上一节，我们通过下载相关的 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 timebs.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+1if i%100==0 :gc.collect()print('已完成',i)time.sleep(2)   
bs.logout()

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

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

日常执行

# 日常执行
import baostock as bs
import pandas as pd
import gc
import time
import datetimetodate=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()

同样，将首次执行中将日期修改当天日期，写入数据方式，由原来的“repalce”修改为了“append”，以完成追加。

总结

至此，我们用两种方式将数据本地化，有了数据我们就可以进行相关的操作。从下一讲开始介绍数据的相关处理。