[키움API]파이썬 주식 종가정보 불러오기4: 데이터 DB저장(mysql)

MySQL연결하기 

[키움API]파이썬 주식 종가정보 불러오기3: 전체종목 기간별 종가조회에서 작업해 본 것처럼 종목당 600개 정도의 시계열 데이터가 나오고 전체 종목 수는 약 2,000 개 정도니까 전체 데이터를 전부 수집하면 약 1,200,000 개 정도 된다. 이런 방대한 데이터는 엑셀로 관리하기는 힘들고 MySQL을 통해서 DB로 관리하면 편하다.(심지어 무료다.)

파이썬에서 mysql을 사용하려면 먼저 mysql을 설치한 뒤(MYSQL 설치(Windows)를 참조) 아나콘다 프롬프트에서 내가 현재 작업 중인 가상환경을 활성화 한다. 그리고 pip install pymysql, pip install sqlalchemy를 통해서 파이썬과 mysql을 연결해 주는 라이브러리를 설치해 준다. 설치가 완료되면 아래 그림 1과 같은 라이브러리 목록이 완성된다.

그림1: sql 관련 라이브러리 추가설치

DB작업 클래스 생성

DB에 연결, 데이터 입수, 데이터 저장 등의 기능은 양이 방대해서 새롭게 모듈과 클래스를 만들어서 처리해주는 것이 낫다. 그래서 키움 API를 작업하는 경로에 sqlhandler.py라는 파일을 만들고 DBWork라는 클래스를 아래와 같이 만들어서 작업할 계획이다.

import pymysql
import sqlalchemy as db
import pandas as pd
pymysql.install_as_MySQLdb()

class DBWork:
    def __init__(self):
        self.user = "MySQL에 설정한 사용자명"
        self.passwd = "서버 비밀번호"
        self.host = "127.0.0.1" # 내 컴퓨터의 localhost 접속이므로 기본 ip인 127.0.0.1 입력
        self.db_port = "포트번호" # 로컬이면 3306 
        self.db_name = "kiwoom"
        self.engine = self.createengine()
    
    def createengine(self):
        db_connection_str = "mysql+mysqldb://" + self.user + ":" +  \
            self.passwd + "@" + self.host + ":" + self.db_port +"/"+ \
            self.db_name
        db_connection = db.create_engine(db_connection_str)
        return db_connection

 

모듈 import 이후 pymysql.install_as_MySQLdb() 을 한 줄 추가해줘야 하는데 "ModuleNotFoundError: No module named 'MySQLdb'" 문제 발생 시 해결해 주는 코드이다. 그리고 __init__에서 클래스가 시작할 때 내가 사용하는 sql 서버의 사용자명, 비밀번호 등을 설정해 준다.  마지막으로 def createengine이라는 엔진객체를 만드는 함수를 정의해 놓은 뒤에 self.engine이라는 엔진 객체를 클래스가 시작할 때 만들도록 설정해 놓는다. 이렇게 해놓으면 추가되는 메서드부터는 self.engine.connect()만 사용해도 sql 서버와 붙을 수 있어서 편하다.

DB에 데이터 저장하기

테이블 존재여부 확인

MySQL은 스키마라는 큰 통을 먼저 만들고 그 안에 다양한 테이블을 생성, 삭제, 수정 하면서 돌아가는 프로그램이다. 따라서 어떤 데이터를 저장하기 위해서는 내가 작업하려는 스키마 안에 해당 테이블이 있는지부터 확인을 해야 한다. 이 기능은 sqlalchemy 라이브러리의 inspect라는 메서드에 구현되어 있다. inspect 메서드를 사용해서 테이블이 존재하는지 확인하는 함수인 checktableexists라는 함수를 아래와 같이 만들 수 있다. 

import sqlalchemy as db
...(중략)...

def checktableexists(self, engine, tablename: str):
	# sqlalchemy 를 db 로 불러왔으니까 db 객체의 inspect 매서드에 변수로 받은 engine을 넣은 값을
    # inspector로 지정
    inspector = db.inspect(engine)
    # inspector 에서 has_table 메서드를 사용, 변수로 받은 테이블명이 있으면 return 값이 있을 것
    return inspector.has_table(tablename)

 

데이터 존재여부 확인

이미 존재하는 테이블에 데이터를 저장할 때 가장 고려해야 할 문제는 데이터의 중복이다. 같은 의미를 가진 데이터가 여러 개 있을 경우 데이터를 사용하거나 처리할 때 에러가 발생할 수 있기 때문이다. [키움API]파이썬 주식 종가정보 불러오기3: 전체종목 기간별 종가조회에서는 일자별, 종목별 주가 데이터를 입수했기 때문에 일자와 종목이 같은 데이터가 있는지 없는지 판단해 주는 checkdataexists라는 함수를 아래와 같이 만들 수 있다. 

    def checkdataexists(self, tablename: str, condition: list):
    	# self.engine 이라는 엔진 객체에 connect() 메서드로 연결
        with self.engine.connect() as connection:
        	# 조건을 list로 받아서 list 첫번재 조건인 종목코드, 두번째 조건인 일자가 같은 데이터를 입수
            query = db.text(f"SELECT * FROM {tablename} WHERE 종목코드 = '{condition[0]}' AND 일자 = '{condition[1]}'")
            result = connection.execute(query)
            # 데이터가 없으면 fetchall 했을 때 return 값이 없을 것
            return result.fetchall()

 

위 함수는 먼저 리스트 형식으로 [종목코드, 일자]를 condition 이라는 변수로 받아서 종목코드와 일자가 일치하는 데이터를 뽑아내는 쿼리를 작성한다. 그리고 그 쿼리의 결과를 fetchall() 메서드로 return 하는 함수이다. 

일자별, 종목별 종가 저장하는 함수

지금까지 만든 함수를 바탕으로 savedata 라는 일자별, 종목별 종가정보를 저장하는 함수를 아래와 같은 구조로 만들 수 있다. 

그림2: savedata(종목별 종가 정보 저장하는 함수) 함수 구조

변수로 입수한 데이터프레임에 데이터가 있으면, 먼저 checktableexists로 테이블이 존재하는지 확인한다. 테이블이 존재하면 데이터프레임의 각각의 행을 row 변수에 넣고 checkdataexists 함수를 이용해서 종목코드, 날짜가 동일한 데이터가 있는지 확인한다. 데이터가 있으면 넘어가고, 아닐 경우 valueslist에 저장해 둔 뒤 맨 마지막에 valueslist라는 데이터 덩어리를 INSERT 해 준다. 테이블이 존재하지 않으면 mysql 서버에 테이블을 만들고 거기에 데이터를 저장하는 기능을 가진 메서드인 to_sql 메서드를 사용해서 데이터 전체를 일괄 저장한다. DBWork 클래스의 전체 코드는 아래와 같이 작성할 수 있다.

import pymysql
import sqlalchemy as db
import pandas as pd
pymysql.install_as_MySQLdb()

class DBWork:
    def __init__(self):
        self.user = "MySQL에 설정한 사용자명"
        self.passwd = "서버 비밀번호"
        self.host = "127.0.0.1" # 내 컴퓨터의 localhost 접속이므로 기본 ip인 127.0.0.1 입력
        self.db_port = "포트번호" # 로컬이면 3306 
        self.db_name = "kiwoom"
        self.engine = self.createengine()
    
    def createengine(self):
        db_connection_str = "mysql+mysqldb://" + self.user + ":" +  \
            self.passwd + "@" + self.host + ":" + self.db_port +"/"+ \
            self.db_name
        db_connection = db.create_engine(db_connection_str)
        return db_connection
    
    def checktableexists(self, engine, tablename: str):
        inspector = db.inspect(engine)
        return inspector.has_table(tablename)

    def checkdataexists(self, tablename: str, condition: list):
        with self.engine.connect() as connection:
            query = db.text(f"SELECT * FROM {tablename} WHERE 종목코드 = '{condition[0]}' AND 일자 = '{condition[1]}'")
            result = connection.execute(query)
            return result.fetchall()
    
    def getcolumns(self, tablename: str):
        with self.engine.connect() as connection:
            columns = db.inspect(connection).get_columns(tablename)
            columns = [column['name'] for column in columns]
            return columns

    def savedata(self, tablename: str, df: pd.DataFrame):
        if len(df)>0:
            conn = self.engine.connect()
            if self.checktableexists(self.engine, tablename):
                columns = self.getcolumns(tablename)
                columnnames = ", ".join(columns)
                valueslist = []
                for row in df.itertuples():
                    conditions = [row[1], row[2]]
                    if self.checkdataexists(tablename, conditions):
                        pass
                    else:
                        values = [f"'{value}'" for value in row[1:]]
                        values = ", ".join(values)
                        valueslist.append(f"({values})")
                if len(valueslist) > 0:
                    valueslist = ", ".join(valueslist)
                    query = db.text(f"INSERT INTO {tablename} ({columnnames}) VALUES {valueslist}")
                    try:
                        conn.execute(query)
                        conn.commit()
                    except Exception as e:
                        print(f"Error executing query: {e}")
                else:
                    pass   
                conn.close()     
            else:
                df.to_sql(name=tablename, con=conn, index=False)
                conn.close()
        else:
            pass