3. 변동성 돌파 전략과 노이즈 상관 분석 프로그래밍(파이썬)

< 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

"init_pool" 메서드는 DB와 커낵션 풀을 생성한다. "req_noise_code_list()"는 "kospi_noise" 또는 "kosdaq_noise" 테이블에 있는전체 Code를 가져온다. "req_noise_data"는 code 종목의 전체 noise 데이터를 가져온다. "insert_mean_noise"는 계산한 평균 노이즈 데이터를 테이블에 저장한다. 

다음은 실제 평균 노이즈를 계산하는 부분이다. 

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일의 평균 노이즈 값이 현재 노이즈 값이 되게 해야 한다.

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

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