Programming

< 3. 변동성 돌파 전략과 노이즈 상관 분석 - 평균 노이즈 계산 >

1. 개요

2. 노이즈 계산

해당 포스트를 읽기 전에 위 포스트를 읽길 추천한다.

이번 포스트에서는 이전 포스트에서 계산한 노이즈 값을 토대로 평균 노이즈 값을 계산한다. 평균 노이즈 값을 계산해 이를 가지고 변동성 돌파 전략의 수익율과 상관 분석을 할 것이다. 평균 노이즈를 주식의 이동 평균선과 같은 원리라고 생각하면 된다. 5/10/20/40/60/120일 각각을 구해서 데이터 베이스에 저장할 것이다. 테이블 이름은 "kospi_noise_일자", "kosdaq_noise_일자"로 한다. 테이블 구조는 "kospi_noise"와 같다. 

노이즈 평균은 현재 주식 날짜의 이전 날부터 n개의 노이즈 평균으로 한다. 즉, 5일 노이즈 평균을 구한다고 가정하자. 그러면 2018년 1월 25일 주식은 18, 19, 22, 23, 24일 주식의 노이즈 값의 평균으로 한다. 즉, 25일 노이즈를 포함시키지 않는다. 25일 노이즈는 25일 주식 시장 개장 기간 기준으로 미래이기 때문이다. 또한 노이즈 평균은 각 코드의 노이즈 평균으로 구한다. 현재 상장된 모든 주식(etf, 우선주 제외)을 대상으로 하기에 계산할 데이터 양이 많을 수 밖에 없다. 그래서 이번 포스트에서 비동기 프로그래밍과 멀티 프로세싱 방식을 사용한다. 

먼저 DB 관련 코드를 보자. 

import aiomysql as aio
import logging as log
import pandas as pd

class StockDB():
    async def init_pool(self,loop):
        log.info("Connection to Connection Pool")
        try:
            self.__pool = await aio.create_pool(host='127.0.0.1', port=3306, user='root', password='qhdks12#$', db='stock',loop=loop, maxsize=50)
        except:
            log.warning("Connecting Pool Error {}".format(repr(0)))
            raise

    async def req_noise_code_list(self, market):
        log.info("Selecting {} noise code list".format(market))

        sql = "select distinct Code from "+market+"_noise"

        try:
            async with self.__pool.acquire() as conn:
                async with conn.cursor() as cur:
                    await cur.execute(sql)
                    rows = await cur.fetchall()
        except aio.Error as e:
            log.warning("Selecting noise data Aiomysql Error : {}".format(repr(e)))
            raise
        except Exception as e:
            log.warning("Selecting noise data Error : {}".format(repr(e)))
            raise
        return list(rows)

    async def req_noise_data(self, market, code):
        log.info("Selecting noise data")

        sql = "select Date, Noise from "+market+"_noise where Code='"+code+"';"

        try:
            async with self.__pool.acquire() as conn:
                async with conn.cursor() as cur:
                    await cur.execute(sql)
                    rows = await cur.fetchall()
                    result = pd.DataFrame.from_records(list(rows))
                    result.columns = ['Date', 'Noise']
                    result = result.set_index('Date')

        except aio.Error as e:
            log.warning("Selecting noise data Aiomysql Error : {}".format(repr(e)))
            raise
        except Exception as e:
            log.warning("Selecting noise data Error : {}".format(repr(e)))
            raise

        return result

    async def insert_profit(self,market,type, data):
        log.info("Inserting {}_noise_{} data".format(market,type))

        sql = "insert into "+market+"_noise_"+type+" (Date, Noise , Code) values (%s,%s,%s)"

        try:
            async with self.__pool.acquire() as conn:
                async with conn.cursor() as cur:
                    await cur.executemany(sql,  data.reset_index().values.tolist())
                    await conn.commit()
        except aio.Error as e:
            log.warning("Inserting {}_noise_{} data Aiomysql Error : {}".format(market, repr(e),type))
            raise
        except Exception as e:
            log.warning("Inserting {}_noise_{} data Error".format(market, repr(e),type))
            raise

import logging as log
import pandas as pd
import asyncio as asy
import sys
import StockDB as db
from multiprocessing import Pool

log.basicConfig(stream=sys.stdout, level=log.DEBUG)


def mean_noise(data, window):
    log.info("Calculating Noise Mean - Size : {} ".format(window))

    try:
        result = data.rolling(window=window).mean()
        result['Noise'] = result['Noise'].shift()
        result = result.dropna()
    except Exception as e:
        log.warning("Calculating Noise Mean Error : {} , Size : {}".format(repr(e), window))
        raise

    return result

async  def noise(code, type):
    log.info("Call the noise function - code : {}".format(code))

    global  dB
    global  market

    try:
        data = await dB.req_noise_data(market, code)
        result = mean_noise(data, int(type))
        if result.empty is True:
            return
        result['Code'] = code
        await  dB.insert_mean_profit(market, type, result)
    except Exception as e:
        log.warning("Call the noise function Error - code : {} ... {}".format(code),repr(e))
        raise

async def main_function(loop,index):

    global  dB
    global  market

    market = "kospi"
    type = "5"

    try:
        dB = db.StockDB()
        await dB.init_pool(loop)

        code_list = await dB.req_noise_code_list(market)
        divide = int(len(code_list)/4)+1
        code_list = code_list[index * divide:(index + 1) *divide]

        futures = [asy.ensure_future(noise(code[0], type)) for code in code_list]
        await asy.gather(*futures)
    except Exception as e:
        log.warning("Error : {}".format(repr(e)))

def main(index):

    loop = asy.get_event_loop()
    loop.run_until_complete(main_function(loop,index))


if __name__ == '__main__':
    ranges = [0,1,2,3]
    pool = Pool(processes=4)
    pool.map(main,ranges)

필자의 CPU가 쿼드코어이기에 프로세스를 4개 생성한다. 각 프로세스는 거의 동일한 개수의 주식 Code 리스트를 할당받아 Code 마다 평균 Noise 값을 계산해 DB에 저장한다. 주식 Code 리스트를 프로세스마다 동일하게 나누기 위해 "ranges" 변수를 사용한다. "dB.req_noise_code_list()"로 코스피 또는 코스닥에 상장된 모든 주식 Code 리스트를 받으면 이를 ranges 변수를 이용해 4등분해 각 프로세스에 할당하고 "noise()" 메서드를 비동기적으로 호출한다. "noise()" 메서드는 인자로 받은 code의 평균 Noise 를 계산해 DB에 저장한다. 

"market" 변수는 "kospi", "kosdaq" 값이 들어간다. type은 몇 일 평균으로 할 지 결정하는 변수이다. 사용자가 5,10,20 등 으로 설정한다. 물론 튜플로 만들어 한 번 실행해서 끝낼 수 있다. 하지만 시간이 너무 오래 걸려 사용자가 일일이 바꾸는 걸로 했다. 그러면 나눠서 처리할 수 있다. 

"noise()" 변수는 평균 noise 값을 계산에 DB에 저장한다. "mean_noise()" 메서드는 평균 노이즈 값을 계산 후 리턴한다. 다만 리턴 값이 empty 일 수도 있다. 만약 10일 전에 상장한 주식이 있을 경우 120일 평균 노이즈 값을 계산할 수 없기 때문이다. 그래서 empty인지 확인해야 한다. 

"mean_noise()" 메서드는 평균 노이즈 값을 계산한다. 계산을 위해 rolling() 메서드를 사용한다. 'shift()'  메서드 사용 이유는 평균 노이즈 값 계산에 당일날 노이즈를 포함시키지 않아야 하기 때문이다. 그냥 rolling().mean()을 사용하면 평균 노이즈 값에 당일날 노이즈값이 포함된다. 그래서 shift를 시켜 전날부터 n일의 평균 노이즈 값이 현재 노이즈 값이 되게 해야 한다.

여기까지가 평균 노이즈 값을 계산해봤다. 다음 포스트에서는 평균 노이즈 값과 변동성 돌파 전략 수익율 사이의 상관 관계를 계산하고 차트로 나타내는 방법에 대해 알아볼 예정이다. 

참고로 필자는 컴퓨터 공학과를 재학 중인 대학생입니다. 따라서 코드가 완벽할 수 없습니다. 알고리즘이나 코드가 비효율적이거나 오류가 있다면 댓글 달아주세요..

< 2. 변동성 돌파 전략과 노이즈 상관 분석 - 노이즈 계산 >

1. 개요

해당 포스트를 읽기 전에 위 포스트를 읽길 추천한다.

이번 포스트에서는 2007년 1월 1일부터 2018년 1월 10일까지의 주식 일봉 데이터를 가지고 노이즈를 계산한 후 데이터 베이스에 저장한다. 일봉 데이터는 "kospi_market"과 "kosdaq_market" 테이블에 있다.(개요를 보면 알 수 있다.) DB에서 해당 테이블에 있는 모든 일봉 데이터를 가져온 후 노이즈 계산을 하고 노이즈 데이터를 저장할 테이블에 저장한다. 

노이즈 테이블에 저장할 테이블 이름은 "kospi_noise", "kosdaq_noise"라고 한다. 테이블 구조는 아래와 같다. 

코드를 살펴보자. 다음은 DB에 액세스하는 클래스 코드이다.

from sqlalchemy import create_engine
import pymysql
pymysql.install_as_MySQLdb()
import pandas as pd
import logging as log

class StockDB(object):
    # 싱글톤 패턴
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = object.__new__(cls)
            return cls._instance
        return cls._instance


    def __init__(self, password):
        log.info("Connecting database.....")

        try:
            self.engine = create_engine("mysql+mysqldb://root:"+password+"@localhost/stock", encoding='utf-8')
            self.conn = pymysql.connect(host='localhost', user='root', password=password, db='stock', charset='utf8')
            self.cursor = self.conn.cursor()
        except Exception as e:
            log.warning("Connecting database Error : {}".format(repr(e)))

    def select_market_all_data(self, market):
        log.info("Selecting market all data....")

        try:
            data = pd.read_sql('select Date, Code, Open, High, Low, Close from '+market+'_market', self.conn, index_col='Date')
        except Exception as e:
            log.info("Selecting market all data Error : {}".format(repr(e)))

        return data

    def insert_noise_data(self,market, data):
        log.info("Inserting into data in {} noise".format(market))

        try:
            data.to_sql(name=market+"_noise", con=self.engine, if_exists='append')
            self.conn.commit()
        except Exception as e:
            log.info("Inserting into data in {} noise Error : {}".format(market, repr(e)))

import StockDB as dB
import pandas as pd
import logging as log
import sys

log.basicConfig(stream=sys.stdout, level=log.DEBUG)


def cal_noise(data):
    log.info("Calculating noise")

    noise_data = pd.DataFrame()

    try:
        noise_data['Code'] = data['Code']
        noise_data['Noise'] = (1- abs(data['Open']-data['Close'])/(data['High']-data['Low']).replace(0,sys.maxsize)).astype(float)
        noise_data.set_index(data.index)
    except Exception as e:
        log.warning("Calculating noise Error {} : ".format(repr(e)))

    return noise_data



db = dB.StockDB("qhdks12#$")

data = db.select_market_all_data("kospi")
noise_data = cal_noise(data)
db.insert_noise_data("kospi",noise_data)


data = db.select_market_all_data("kosdaq")
noise_data = cal_noise(data)
db.insert_noise_data("kosdaq",noise_data)

noise 값 계산은 매우 간단하다. 전체적인 부분을 보면 db 객체를 초기화한 후 "db.select_market_all_data("kospi")"로 DB에 저장되어 있는 코스피 종목 모든 일자 데이터를 가져온다. "cal_noise()" 메서드는 인자로 받은 데이터를 토대로 노이즈 값을 계산하고 반환한다. 마지막으로 "insert_noise_data()" 메서드에 계산한 값을 전달해 실제 "kospi_noise" 테이블에 저장한다. 코스닥 종목에 대해서도 똑같이 한다.

"cal_noise" 메서드는 종목의 일봉 데이터를 토대로 노이즈를 계산한다. 노이즈 계산 공식은 아래와 같다.

Noise = 1 - | 종가 - 시가 | / ( 고가 - 저가 )

"cal_noise" 메서드는 위 공식을 그대로 코드로 표현했다. 다만 예외 사항이 한 가지 있다. ( 고가 - 저가 ) 가 0이 될 경우다. 이 경우가 의외로 많다. 거래가 몇 일간 정지됬거나 점 상한가 등의 경우가 이에 해당한다. ( 고가 - 저가 ) 가 0이 되면 0으로 나누려고 했기에 예외가 발생한다. 따라서 이를 처리해야 한다. ( 고가 - 저가 ) 가 0일 때는 0을 sys.maxsize로 int형 최댓값으로 바꿔 Noise를 1에 아주 가깝게 만든다. 그 이유는 점 상한가나 거래 정지가 안정성을 추구하는 거래라고 보지 않기 때문이다. 

나머지 코드는 쉽게 이해 가능할 거라 생각한다.

"cal_noise" 메서드 결과로 데이터 프레임이 반환된다. 인덱스 값은 Date가 되고 칼럼으로 Code와 Noise가 있다. 이를 그대로 insert_noise_data()에 전달하면 노이즈 값이 계산되 DB에 저장된다.     

이번 포스트에서는 변동성 돌파 전략과 노이즈 상관 분석을 위해서 노이즈 값을 계산해 DB에 저장했다. 다음 포스트에서는 5/10/20/40/60/120 평균 노이즈 값을 계산할 것이다.

참고로 필자는 컴퓨터 공학과를 재학 중인 대학생입니다. 따라서 코드가 완벽할 수 없습니다. 알고리즘이나 코드가 비효율적이거나 오류가 있다면 댓글 달아주세요..