current position:Home>Python quantitative data warehouse construction 3: data drop library code encapsulation

Python quantitative data warehouse construction 3: data drop library code encapsulation

2022-01-31 00:55:07 Hang Seng light cloud community

Python Quantitative data warehouse construction 3: Data drop library code encapsulation

 This series of tutorials is for quantitative developers , Provide a tutorial on the construction of local quantitative financial data warehouse and a full set of source code . We count with constancy (UDATA) The financial data community is the data source , Drop the basic financial data into the local database . The tutorial provides a full set of source code , Including historical data download and incremental data update , Data update task deployment and daily monitoring .
 Copy code 

In the previous section , We encapsulate Python operation MySQL Custom class of database , Save as MySQLOperation.py file ; This section contains the operation database section , Will call MySQLOperation The method in , as well as pandas.to_sql and pandas.read_sql The operation of .

One 、 Constant number (UDATA) Operation introduction

1、 obtain Token

A、 On the official website (udata.hs.net) Register and log in , On the subscription page , Order a free experience package ;

B、 In the upper right corner , Avatar drop-down menu , Enter the overview page , Copy Token;

C、 On the data page , View the data interface documentation , Get interface name 、 Request parameters 、 Return parameters and Python Code example ;

2、 install hs_udata

pip install hs_udata
 Copy code 

An example is as follows :

import hs_udata as hs
#  Set up Token
hs.set_token(token = 'xxxxxxxxxxxxxxx')  #  Get personal information from the overview page Token
#  Take the minute line as an example , obtain 000001.SZ stay 2021-05-01 to 2021-06-01 Minute line data during 
#  For the interface documentation, see :https://udata.hs.net/datas/342/
df = hs.stock_quote_minutes(en_prod_code="000001.SZ",begin_date="20210501",end_date="20210601")
#  Add stock code column 
df['hs_code']='000001.SZ'
df.head()
 Copy code 

The use process of other interfaces is similar ;

Two 、 Data drop database example

List of stocks (stock_list) For example , Explain how to build tables 、 Drop storage 、 Query and so on ; See Chapter 3 of this article for the complete code ;

1、 preparation

(1) stay MySQL In the database , Create database udata, See Lecture 1 for the creation process 《Python Quantitative data warehouse building series 1: Database installation and operation 》;

(2) stay MySQL In the database , Create data update record table udata.tb_update_records, The table structure is as follows :

Build table SQL as follows :

CREATE TABLE udata.tb_update_records (
				table_name CHAR(40),
				data_date CHAR(20),
				update_type CHAR(20),
				data_number INT(20),
				elapsed_time INT(20),
				updatetime CHAR(20)
				)
 Copy code 

(3) take Token Write configuration file with database parameters DB_MySQL.config, The contents of the document are as follows :

[udata]
token=' Yours Token'
host='127.0.0.1'
port=3306
user='root'
passwd=' password '
db='udata'
 Copy code 

(4) Read the parameters in the configuration file

import configparser
#  Read the configuration file , Constant number and database parameters 
configFilePath = 'DB_MySQL.config'
section = 'udata'
config = configparser.ConfigParser()
config.read(configFilePath)
#  Read   Constant number (UData)  Of  token
token = eval(config.get(section=section, option='token'))
# MySQL Connection parameter reading 
host = eval(config.get(section=section, option='host'))
port = int(config.get(section=section, option='port'))
db = eval(config.get(section=section, option='db'))
user = eval(config.get(section=section, option='user'))
passwd = eval(config.get(section=section, option='passwd'))
 Copy code 

2、 Build table

On the community data page , View the parameter fields returned from the data table , List of stocks (stock_list,udata.hs.net/datas/202/)…

Table to be created 、 Delete table 、 Clear data SQL, Write to configuration file DB_Table.config, The contents of the document are as follows :

[tb_stock_list]
DROP_TABLE='DROP TABLE IF EXISTS tb_stock_list'
CREATE_TABLE='''CREATE TABLE tb_stock_list (
				secu_code CHAR(20),
				hs_code CHAR(20),
				secu_abbr CHAR(20),
				chi_name CHAR(40),
				secu_market CHAR(20), 
				listed_state CHAR(20),
				listed_sector CHAR(20),
				updatetime CHAR(20)
				)'''
DELETE_DATA = 'truncate table tb_stock_list'
 Copy code 

The table code is as follows :

# MySQL Operation instantiation ,pymysql connections setting up , May open 、 close 、 perform sql、 perform sql Reading data 
MySQL = MySQLOperation(host, port, db, user, passwd)
#  Read the configuration file , Constant number and database parameters 
configFilePath = 'DB_Table.config'
section = 'tb_stock_list'
config = configparser.ConfigParser()
config.read(configFilePath)
DROP_TABLE = eval(config.get(section=section, option='DROP_TABLE'))
CREATE_TABLE = eval(config.get(section=section, option='CREATE_TABLE'))
DELETE_DATA = eval(config.get(section=section, option='DELETE_DATA'))
#  Delete table structure 
MySQL.Execute_Code(DROP_TABLE)
#  Create a table structure 
MySQL.Execute_Code(CREATE_TABLE)
 Copy code 

3、 Data in the database

import hs_udata as hs
from sqlalchemy import create_engine
from datetime import datetime
#  Set up token
hs.set_token(token)
#  obtain   Stock list   data 
df = hs.stock_list()
#  Add the system timestamp in the last column 
dt = datetime.now()
df['updatetime'] = dt.strftime('%Y-%m-%d %H:%M:%S')
#  As the stock list data is fully updated , Before data insertion , First clear the data in the table 
MySQL.Execute_Code(DELETE_DATA)
# sqlalchemy  connections setting up , Can be used for pandas.read_sql、pandas.to_sql
engine = create_engine('mysql://{0}:{1}@{2}:{3}/{4}?charset=utf8'.format(user,passwd,host,port,db))
#  Writes data to MySQL Data table in 
df.to_sql(name='tb_stock_list', con=engine, index=False, if_exists='append')
 Copy code 

Check the results in the database as follows :

4、 Reading data

import pandas as pd
result = pd.read_sql('''SELECT * FROM tb_stock_list ''', con=engine)
print(result.head())
 Copy code 

3、 ... and 、 Drop library code encapsulation

Take the above steps , Put it all together , Define and call python in class Class properties and methods . The main technical points involved in the code are as follows :

(1) Use pymysql、pandas.to_sql and pandas.read_sql operation MySQL database ;

(2) Use class Class method , Integrated table creation 、 Insert data and query data ;

(3) How to use a profile , From local files , Read database parameters and table operations SQL Code ;

(4) Use try Fault tolerance mechanism , Combined with log function , Print the execution log to the local DB_MySQL_LOG.txt file ;

import pandas as pd
import hs_udata as hs
from MySQLOperation import *
from sqlalchemy import create_engine
from datetime import datetime
import time
import configparser
import logging
import traceback
import warnings
warnings.filterwarnings("ignore")

class TB_Stock_List:

    def __init__(self,MySQL_Config,BD_Name,Table_Config,Table_Name):
        #  Create a log 
        self.logging = logging
        self.logging.basicConfig(filename='DB_MySQL_LOG.txt', level=self.logging.DEBUG
                                 , format='%(asctime)s - %(levelname)s - %(message)s')
        #  Read the configuration file , Constant number and database parameters 
        configFilePath = MySQL_Config
        self.section1 = BD_Name
        config = configparser.ConfigParser()
        config.read(configFilePath)
        #  Read   Constant number (UData)  Of  token
        self.token = eval(config.get(section=self.section1, option='token'))
        # MySQL Connection parameter reading 
        self.host = eval(config.get(section=self.section1, option='host'))
        self.port = int(config.get(section=self.section1, option='port'))
        self.db = eval(config.get(section=self.section1, option='db'))
        self.user = eval(config.get(section=self.section1, option='user'))
        self.passwd = eval(config.get(section=self.section1, option='passwd'))
        # pymysql connections setting up , May open 、 close 、 perform sql、 perform sql Reading data 
        self.MySQL = MySQLOperation(self.host, self.port, self.db, self.user, self.passwd)
        # sqlalchemy  connections setting up , Can be used for pandas.read_sql、pandas.to_sql
        self.engine = create_engine('mysql://{0}:{1}@{2}:{3}/{4}?charset=utf8'.format(self.user
                                                                                      , self.passwd
                                                                                      , self.host
                                                                                      , self.port
                                                                                      , self.db))
        #  Read the configuration file , Constant number and database parameters 
        configFilePath = Table_Config
        self.section2 = Table_Name
        config = configparser.ConfigParser()
        config.read(configFilePath)
        self.DROP_TABLE_SQL = eval(config.get(section=self.section2, option='DROP_TABLE'))
        self.CREATE_TABLE_SQL = eval(config.get(section=self.section2, option='CREATE_TABLE'))
        self.DELETE_DATA_SQL = eval(config.get(section=self.section2, option='DELETE_DATA'))

        self.logging.info('*********************{0}.{1}*********************'.format(self.section1, self.section2))

    def CREATE_TABLE(self):
        try:
            #  Delete table structure 
            self.MySQL.Execute_Code('SET FOREIGN_KEY_CHECKS = 0')
            self.MySQL.Execute_Code(self.DROP_TABLE_SQL)
            #  Create a table structure 
            self.MySQL.Execute_Code(self.CREATE_TABLE_SQL)
            self.logging.info(' surface {0}.{1}, Table created successfully '.format(self.section1,self.section2))
        except:
            self.logging.info(' surface {0}.{1}, Table creation failed '.format(self.section1,self.section2))
            self.logging.debug(traceback.format_exc())

    def UPDATE_DATA(self):
        try:
            #  Set up token
            hs.set_token(self.token)
            time_start = time.time()  #  timing 
            #  obtain   Stock list   data 
            df = hs.stock_list()
            #  Add the system timestamp in the last column 
            dt = datetime.now()
            df['updatetime'] = dt.strftime('%Y-%m-%d %H:%M:%S')
            #  As the stock list data is fully updated , Before data insertion , First clear the data in the table 
            self.MySQL.Execute_Code(self.DELETE_DATA_SQL)
            #  Writes data to MySQL Data table in 
            df.to_sql(name='tb_stock_list', con=self.engine, index=False, if_exists='append')
            time_end = time.time()  #  timing 
            elapsed_time = round(time_end-time_start,2)
            #  towards mysql Record a data update record in the library : Table name , Data date , Update the way , Number of updates , Updating takes time , system time 
            self.RECORDS_SQL = '''INSERT INTO udata.tb_update_records VALUES ('{0}','{1}',' Total quantity ',{2},{3}, SYSDATE())'''.format(self.section2
                                                                   ,dt.strftime('%Y-%m-%d'),len(df),elapsed_time)
            self.MySQL.Execute_Code(self.RECORDS_SQL)
            self.logging.info(' surface {0}.{1}, Data updated successfully '.format(self.section1,self.section2))
        except:
            self.logging.info(' surface {0}.{1}, Data update failed '.format(self.section1,self.section2))
            self.logging.debug(traceback.format_exc())

    def READ_DATA(self, WhereCondition=''):
        try:
            result = pd.read_sql('''SELECT * FROM {0}.{1} '''.format(self.section1,self.section2)
                                 + WhereCondition, con=self.engine)
            self.logging.info(' surface {0}.{1}, Data read successful '.format(self.section1,self.section2))
            return result
        except:
            self.logging.info(' surface {0}.{1}, Data read failed '.format(self.section1,self.section2))
            self.logging.debug(traceback.format_exc())
            return 0
        
if __name__ == '__main__':
    MySQL_Config = 'DB_MySQL.config'
    BD_Name = 'udata'
    Table_Config = 'DB_Table.config'
    Table_Name = 'tb_stock_list'
    #  Instantiation 
    TB_Stock_List_Main = TB_Stock_List(MySQL_Config,BD_Name,Table_Config,Table_Name)
    #  Create a table structure 
    TB_Stock_List_Main.CREATE_TABLE()
    #  Update data 
    TB_Stock_List_Main.UPDATE_DATA()
    #  Reading data 
    data = TB_Stock_List_Main.READ_DATA()
 Copy code 

Four 、 Code and configuration file download

(1) Attachment file directory :

DB_MySQL.config

DB_Table.config

MySQLOperation.py

TB_Stock_List.py

(2) After downloading the attachment file , Set profile DB_MySQL.config Medium Token And database parameters , Modify to your own parameters ; Put the code in the same directory and execute .

Next section 《Python Quantitative data warehouse building series 4: Stock data is stored in the database 》

copyright notice
author[Hang Seng light cloud community],Please bring the original link to reprint, thank you.
https://en.pythonmana.com/2022/01/202201310055060304.html

Random recommended