5. 실 거래 후 보완된 ETF 변동성 돌파 전략(파이썬)

< 5. 실 거래 후 보완된 ETF 변동성 돌파 전략 - 노이즈 계산>

1. 개요

2. 데이터 수집

3. 개별 종목 수익율 계산

4. 최적 Scope 계산

이번 포스트는 모든 종목의 일봉데이터에 대한 노이즈와 이에 대한 일간 평균 노이즈를 계산한다. 1일, 10일, 20일, 40일, 60일, 120일 평균 노이즈를 계산한다. 노이즈 계산은 이전에 많이 다뤘기에 코드 설명은 하지 않는다. 노이즈 데이터는 아래 사진과 같이 명시된 테이블에 저장된다. 

DB.py

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

class StockDB():

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

        try:
            self.engine = create_engine("mysql+mysqldb://root:"+password+"@localhost/etf1", encoding='utf-8')
            self.conn = pymysql.connect(host='localhost', user='root', password=password, db='etf1', 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):
        log.info("Selecting market all data....")

        try:
            data = pd.read_sql('select Date, Code, Open, High, Low, Close from etf_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, data):
        log.info("Inserting into noise")

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

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

        try:
            data = pd.read_sql('select * from noise', self.conn, index_col='Date')
        except Exception as e:
            log.info("Selecting market all data Error : {}".format(repr(e)))

        return data


    def insert_noise_mean_data(self, data, mean):
        log.info("Inserting into noise_{}".format(mean))

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

CalNoise.py : 각 일봉에 대해 노이즈를 계산한다.

import DB 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()
noise_data = cal_noise(data)
dB.insert_noise_data(noise_data)

MeanNoise.py : 평균 노이즈(1일, 10일, 20일, 40일, 60일, 120일)를 계산한다.

import DB as db
import pandas as pd
import logging as log
import sys


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

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

    try:
        mean = data.groupby('Code')['Noise'].rolling(window=window).mean()
        mean = mean.reset_index(level='Code')
        mean['Noise'] = mean.groupby('Code')['Noise'].shift()
        mean = mean.dropna()

    except Exception as e:
        log.warning("Calculating Noise Mean - Size : {} Error ".format(window,repr(e)))

    return mean

dB = db.StockDB("qhdks12#$")
data = dB.select_noise_all_data()
for window in [1,5,10,20,40,60,120]:
    mean = mean_noise(data, window)
    dB.insert_noise_mean_data(mean, window)

    log.info("Finish window = {}".format(window))